bughunt pip probe: egg-info installs

Author: claude

.github/workflows/bughunt-pip.yml

@@ -0,0 +1,102 @@
+name: bughunt pip egg-info probe
+on:
+  push:
+    branches: ['bughunt/pip/**']
+permissions:
+  contents: read
+jobs:
+  probe:
+    strategy:
+      fail-fast: false
+      matrix:
+        os: [ubuntu-latest, macos-latest, windows-latest]
+        python: ['3.8', '3.11']
+    runs-on: ${{ matrix.os }}
+    timeout-minutes: 45
+    steps:
+      - uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2
+      - uses: Swatinem/rust-cache@c19371144df3bb44fab255c43d04cbc2ab54d1c4 # v2.9.1
+      - run: cargo build --release -p socket-patch-cli
+      - uses: actions/setup-python@a26af69be951a213d495a4c3e4e4022e16d87065 # v5
+        with:
+          python-version: ${{ matrix.python }}
+      - name: probe
+        shell: python
+        run: |
+          import base64, hashlib, json, os, subprocess, sys, threading, zipfile, shutil, pathlib, re
+          from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+          WIN = os.name == 'nt'
+          SP = str(pathlib.Path('target/release/socket-patch' + ('.exe' if WIN else '')).resolve())
+          W = pathlib.Path(os.environ['RUNNER_TEMP']) / 'bh'; shutil.rmtree(W, ignore_errors=True); W.mkdir()
+          def run(cmd, cwd=None, env=None):
+              r = subprocess.run(cmd, cwd=cwd, env=env, capture_output=True, text=True)
+              return r.returncode, r.stdout, r.stderr
+          run([sys.executable, '-m', 'pip', 'download', '-q', '--no-deps', 'six==1.16.0', '--only-binary', ':all:', '-d', str(W / 'wh')])
+          whl = next((W / 'wh').glob('six-*.whl'))
+          orig = zipfile.ZipFile(whl).read('six.py')
+          patched = orig.replace(b'"""Utilities for writing code', b'# SOCKET-PATCH-MARKER\n"""Utilities for writing code', 1)
+          gh = lambda b: hashlib.sha256(b"blob %d\0" % len(b) + b).hexdigest()
+          pw = W / 'six-1.16.0-py2.py3-none-any.whl'
+          zi = zipfile.ZipFile(whl); zo = zipfile.ZipFile(pw, 'w', zipfile.ZIP_DEFLATED)
+          for i in zi.infolist():
+              zo.writestr(i, patched if i.filename == 'six.py' else zi.read(i.filename))
+          zo.close(); wb = pw.read_bytes()
+          UUID = '11111111-1111-4111-8111-111111111111'; PURL = 'pkg:pypi/six@1.16.0'
+          class H(BaseHTTPRequestHandler):
+              def log_message(self, *a): pass
+              def send(self, o, c=200):
+                  b = json.dumps(o).encode(); self.send_response(c); self.send_header('content-type', 'application/json'); self.send_header('content-length', str(len(b))); self.end_headers(); self.wfile.write(b)
+              def do_POST(self):
+                  body = self.rfile.read(int(self.headers.get('content-length', 0))).decode()
+                  if self.path.endswith('/patches/batch'):
+                      pk = [{"purl": PURL, "patches": [{"uuid": UUID, "purl": PURL, "tier": "free", "cveIds": ["CVE-2024-30004"], "ghsaIds": ["GHSA-aaaa-bbbb-cccc"], "severity": "high", "title": "six", "publishedAt": "2024-01-01T00:00:00Z"}]}] if 'pkg:pypi/six@1.16.0' in body else []
+                      return self.send({"packages": pk, "canAccessPaidPatches": False})
+                  if self.path.endswith('/patches/package'):
+                      url = f"http://127.0.0.1:{PORT}/patch/pypi/six/1.16.0/{UUID}/tok/six-1.16.0-py2.py3-none-any.whl"
+                      return self.send({"results": {UUID: {"status": "granted", "url": url, "purl": PURL, "artifacts": [{"kind": "tarball", "url": url, "integrity": {"sha256": hashlib.sha256(wb).hexdigest()}}], "registryOverride": None}}})
+                  self.send({}, 404)
+              def do_HEAD(self):
+                  self.send_response(200); self.end_headers()
+              def do_GET(self):
+                  if self.path.endswith('.whl'):
+                      self.send_response(200); self.send_header('content-length', str(len(wb))); self.end_headers(); self.wfile.write(wb); return
+                  if '/patches/by-package/' in self.path:
+                      return self.send({"patches": [{"uuid": UUID, "purl": PURL, "publishedAt": "2024-01-01T00:00:00Z", "description": "d", "license": "MIT", "tier": "free", "vulnerabilities": {}}], "canAccessPaidPatches": False})
+                  if '/patches/view/' in self.path:
+                      return self.send({"uuid": UUID, "purl": PURL, "publishedAt": "2024-01-01T00:00:00Z", "files": {"six.py": {"beforeHash": gh(orig), "afterHash": gh(patched), "blobContent": base64.b64encode(patched).decode()}}, "vulnerabilities": {"GHSA-aaaa-bbbb-cccc": {"cves": ["CVE-2024-30004"], "summary": "s", "severity": "high", "description": "d"}}, "description": "d", "license": "MIT", "tier": "free"})
+                  self.send({}, 404)
+          srv = ThreadingHTTPServer(('127.0.0.1', 0), H); PORT = srv.server_address[1]
+          threading.Thread(target=srv.serve_forever, daemon=True).start()
+          env = dict(os.environ, SOCKET_API_URL=f'http://127.0.0.1:{PORT}', SOCKET_API_TOKEN='fake', SOCKET_ORG_SLUG='org', SOCKET_NO_TELEMETRY='1')
+          env.pop('VIRTUAL_ENV', None)
+          rows = []
+          for pv in ['20.3.4', '22.3.1', '23.0.1', '23.1']:
+              res = {'pip': pv}
+              def mk(name):
+                  d = W / f'{name}-{pv}'; d.mkdir()
+                  run([sys.executable, '-m', 'venv', str(d / '.venv')])
+                  py = str(d / '.venv' / ('Scripts/python.exe' if WIN else 'bin/python'))
+                  run([py, '-m', 'pip', 'install', '-q', f'pip=={pv}'])
+                  run([py, '-m', 'pip', 'uninstall', '-y', '-q', 'wheel'])
+                  (d / 'requirements.txt').write_text('six==1.16.0\n')
+                  rc, o, e = run([py, '-m', 'pip', 'install', '-q', '--disable-pip-version-check', '--no-binary', 'six', '-r', 'requirements.txt'], cwd=d)
+                  if rc: print('install failed', pv, e[-800:])
+                  site = pathlib.Path(run([py, '-c', 'import six,os;print(os.path.dirname(six.__file__))'])[1].strip())
+                  return d, py, site
+              def marker(py):
+                  rc, o, e = run([py, '-c', "import six;print(open(six.__file__).read().count('SOCKET-PATCH-MARKER'))"]); return o.strip()
+              d, py, site = mk('agent')
+              res['meta'] = ','.join(sorted(p.name for p in site.glob('six-*info')))
+              rc, o, e = run([SP, 'scan', '--mode', 'agent', '-e', 'pypi', '--yes', '--json'], cwd=d, env=env)
+              try: ap = json.loads(o).get('apply', {}).get('patches', [])
+              except Exception: ap = o[-300:]
+              res['agent_rc'] = rc; res['agent'] = [(p.get('action'), p.get('errorCode')) for p in ap] if isinstance(ap, list) else ap; res['agent_marker'] = marker(py)
+              rc, o, e = run([SP, 'scan', '-g', '--global-prefix', str(site), '-e', 'pypi', '--json'], cwd=str(W), env=env)
+              res['global_report_has_six'] = 'pkg:pypi/six@1.16.0' in o
+              d, py, site = mk('hosted')
+              rc, o, e = run([SP, 'scan', '-e', 'pypi', '--yes', '--json'], cwd=d, env=env)
+              res['hosted_rc'] = rc; res['stale_warning'] = 'redirect_pypi_stale_install' in o
+              run([py, '-m', 'pip', 'install', '-q', '--disable-pip-version-check', '-r', 'requirements.txt'], cwd=d)
+              res['hosted_marker_after_install'] = marker(py)
+              rows.append(res); print(json.dumps(res), flush=True)
+          print('RESULTS', sys.version.split()[0], sys.platform, json.dumps(rows))