| |
| """Standalone release probe: Python stdlib + age/age-keygen; no app imports. |
| |
| Only VM_BACKUP_GITHUB_TOKEN is read for authentication. Existing releases and |
| checkpoints are never changed. Each run creates at most one new test release. |
| """ |
| import argparse |
| from datetime import datetime, timezone |
| import hashlib |
| import json |
| import os |
| from pathlib import Path |
| import re |
| import shutil |
| import ssl |
| import subprocess |
| import tempfile |
| from urllib.error import HTTPError, URLError |
| from urllib.parse import urlencode, urlsplit |
| from urllib.request import HTTPRedirectHandler, HTTPSHandler, Request, build_opener |
| import uuid |
|
|
| LIMIT = 1024 * 1024 |
|
|
|
|
| class Failure(Exception): |
| pass |
|
|
|
|
| class NoRedirect(HTTPRedirectHandler): |
| def redirect_request(self, req, fp, code, msg, headers, newurl): |
| return None |
|
|
|
|
| class Probe: |
| def __init__(self, repo, token, report, save): |
| self.repo, self.token, self.report, self.save = repo, token, report, save |
| self.base = 'https://api.github.com/repos/' + repo |
| |
| context = ssl.create_default_context(cafile=os.environ.get('SSL_CERT_FILE') or None) |
| self.opener = build_opener(NoRedirect(), HTTPSHandler(context=context)) |
|
|
| def request(self, method, url, data=None, authenticated=True, redirect=False): |
| parsed = urlsplit(url) |
| hosts = ('api.github.com', 'uploads.github.com') if authenticated else ( |
| 'release-assets.githubusercontent.com', 'objects.githubusercontent.com') |
| if (parsed.scheme != 'https' or parsed.hostname not in hosts |
| or parsed.username or parsed.password or parsed.port not in (None, 443) |
| or parsed.fragment): |
| raise Failure('destination_denied') |
| headers = {'Accept': 'application/vnd.github+json', 'User-Agent': 'vm-release-credential-probe', |
| 'X-GitHub-Api-Version': '2022-11-28'} |
| if authenticated: |
| headers['Authorization'] = 'Bearer ' + self.token |
| if isinstance(data, dict): |
| data = json.dumps(data).encode() |
| headers['Content-Type'] = 'application/json' |
| elif data is not None: |
| headers['Content-Type'] = 'application/octet-stream' |
| if '/releases/assets/' in parsed.path or not authenticated: |
| headers['Accept'] = 'application/octet-stream' |
| try: |
| response = self.opener.open(Request(url, data=data, headers=headers, method=method), timeout=30) |
| except HTTPError as error: |
| response = error |
| except (URLError, OSError): |
| raise Failure('network_or_tls_failure') from None |
| with response: |
| status = response.code |
| event = {'host': parsed.hostname, 'method': method, 'status': status, |
| 'request_id': response.headers.get('X-GitHub-Request-Id'), |
| 'authenticated': authenticated, |
| 'accepted_permissions': response.headers.get('X-Accepted-GitHub-Permissions')} |
| self.report['http_events'].append(event) |
| self.save() |
| if redirect and status in (301, 302, 303, 307, 308): |
| return status, b'', response.headers.get('Location', '') |
| if not 200 <= status < 300: |
| |
| raise Failure('authentication_rejected' if status == 401 else 'http_rejected_' + str(status)) |
| body = response.read(LIMIT + 1) |
| if len(body) > LIMIT: |
| raise Failure('response_too_large') |
| return status, body, None |
|
|
| def json_request(self, method, url, data=None): |
| status, body, _ = self.request(method, url, data) |
| try: |
| return status, json.loads(body) |
| except (ValueError, UnicodeError): |
| raise Failure('invalid_json_response') from None |
|
|
| def run(self, ciphertext, run_id): |
| self.report.update(ciphertext_bytes=len(ciphertext), |
| expected_sha256=hashlib.sha256(ciphertext).hexdigest()) |
| self.save() |
| _, repo = self.json_request('GET', self.base) |
| if repo.get('private') is not True: |
| raise Failure('repository_not_private') |
| self.report['private_repository_verified'] = True |
| push = repo.get('permissions', {}).get('push') |
| self.report['repository_push_permission_reported'] = push |
| if push is False: |
| raise Failure('repository_write_permission_denied') |
| tag = 'vm-credential-test-' + run_id |
| status, release = self.json_request('POST', self.base + '/releases', { |
| 'tag_name': tag, 'name': tag, 'prerelease': True, 'draft': False, |
| 'body': 'Disposable synthetic encrypted credential probe. No application data.'}) |
| if status != 201: |
| raise Failure('release_creation_not_confirmed') |
| release_id = release.get('id') |
| if not isinstance(release_id, int) or isinstance(release_id, bool) or release_id <= 0: |
| raise Failure('invalid_release_id') |
| self.report.update(release_id=release_id, release_tag=tag, contents_write_verified=True) |
| self.save() |
| upload = release.get('upload_url', '').split('{', 1)[0] |
| parsed = urlsplit(upload) |
| expected = '/repos/' + self.repo + '/releases/' + str(release_id) + '/assets' |
| if (parsed.scheme != 'https' or parsed.hostname != 'uploads.github.com' |
| or parsed.path != expected or parsed.query or parsed.fragment |
| or parsed.username or parsed.password or parsed.port not in (None, 443)): |
| raise Failure('upload_url_denied') |
| status, asset = self.json_request('POST', upload + '?' + urlencode({'name': 'synthetic.age'}), ciphertext) |
| if status != 201: |
| raise Failure('asset_creation_not_confirmed') |
| asset_id = asset.get('id') |
| if not isinstance(asset_id, int) or isinstance(asset_id, bool) or asset_id <= 0: |
| raise Failure('invalid_asset_id') |
| self.report['asset_id'] = asset_id |
| self.save() |
| status, downloaded, location = self.request('GET', self.base + '/releases/assets/' + str(asset_id), redirect=True) |
| if location is not None: |
| _, downloaded, _ = self.request('GET', location, authenticated=False) |
| expected_hash = hashlib.sha256(ciphertext).hexdigest() |
| actual_hash = hashlib.sha256(downloaded).hexdigest() |
| self.report.update(ciphertext_bytes=len(ciphertext), expected_sha256=expected_hash, |
| downloaded_sha256=actual_hash) |
| if actual_hash != expected_hash or len(downloaded) != len(ciphertext): |
| raise Failure('asset_readback_mismatch') |
| self.report.update(asset_readback_verified=True, upload_decision='GO', |
| staging_decision='NO-GO: complete recovery and private HF qualification still required', |
| status='passed') |
|
|
|
|
| def encrypt_fixture(age, keygen): |
| with tempfile.TemporaryDirectory(prefix='vm-release-probe-') as scratch: |
| root = Path(scratch) |
| identity, clear, encrypted = root / 'identity.agekey', root / 'synthetic.txt', root / 'synthetic.age' |
| def command(args): |
| try: |
| result = subprocess.run(args, capture_output=True, timeout=30) |
| except (OSError, subprocess.TimeoutExpired): |
| raise Failure('age_tool_failed') from None |
| if result.returncode: |
| raise Failure('age_tool_failed') |
| return result.stdout |
| command([keygen, '-o', str(identity)]) |
| identity.chmod(0o600) |
| recipient = command([keygen, '-y', str(identity)]).decode().strip() |
| payload = b'Visual Mentor synthetic release credential qualification; no user data.\n' |
| clear.write_bytes(payload) |
| command([age, '-r', recipient, '-o', str(encrypted), str(clear)]) |
| if command([age, '-d', '-i', str(identity), str(encrypted)]) != payload: |
| raise Failure('local_age_roundtrip_failed') |
| ciphertext = encrypted.read_bytes() |
| if not ciphertext.startswith(b'age-encryption.org/v1') or len(ciphertext) >= LIMIT: |
| raise Failure('invalid_encrypted_fixture') |
| return ciphertext |
|
|
|
|
| def main(): |
| parser = argparse.ArgumentParser(description=__doc__) |
| parser.add_argument('--repo', default='JsonLord/visual-mentor-backups') |
| parser.add_argument('--output', type=Path, help='New evidence file; existing files are refused') |
| parser.add_argument('--age', default='age', help='age executable path') |
| parser.add_argument('--age-keygen', default='age-keygen', help='age-keygen executable path') |
| args = parser.parse_args() |
| if not re.fullmatch(r'[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+', args.repo): |
| parser.error('invalid owner/repository') |
| run_id = datetime.now(timezone.utc).strftime('%Y%m%dT%H%M%SZ') + '-' + uuid.uuid4().hex |
| output = args.output or Path('github-release-qualification-' + run_id + '.json') |
| report = {'run_id': run_id, 'repository': args.repo, 'credential_source': 'VM_BACKUP_GITHUB_TOKEN', |
| 'status': 'started', 'upload_decision': 'NO-GO', 'staging_decision': 'NO-GO', |
| 'contents_write_verified': False, 'asset_readback_verified': False, 'http_events': [], |
| 'recovery_run': False, 'production_or_workflows_changed': False} |
| |
| try: |
| with output.open('x', encoding='utf-8') as stream: |
| stream.write(json.dumps(report, indent=2) + '\n') |
| output.chmod(0o600) |
| except OSError: |
| print('NO-GO: evidence path unavailable or already exists; no requests made.') |
| return 2 |
| def save(): |
| output.write_text(json.dumps(report, indent=2) + '\n', encoding='utf-8') |
| try: |
| token = os.environ.get('VM_BACKUP_GITHUB_TOKEN') |
| if not token: |
| raise Failure('dedicated_credential_missing') |
| age, keygen = shutil.which(args.age), shutil.which(args.age_keygen) |
| if not age or not keygen: |
| raise Failure('install_age_and_age_keygen') |
| ciphertext = encrypt_fixture(age, keygen) |
| report['local_age_roundtrip_verified'] = True |
| Probe(args.repo, token, report, save).run(ciphertext, run_id) |
| except Failure as error: |
| report.update(status='failed', failure_code=str(error)) |
| except Exception: |
| report.update(status='failed', failure_code='unexpected_local_failure') |
| save() |
| print(json.dumps({'status': report['status'], 'upload_decision': report['upload_decision'], |
| 'evidence': str(output), 'failure_code': report.get('failure_code'), |
| 'asset_id': report.get('asset_id')})) |
| return 0 if report['status'] == 'passed' else 2 |
|
|
|
|
| if __name__ == '__main__': |
| raise SystemExit(main()) |
|
|
