Repository navigation
bughunt pip probe: egg-info installs #7
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| 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)) |