Prepare restricted seven-source reuse reader and bounded recovery

Assistant: codex
Assistant-Model: gpt-6.1-sol
Assistant-Session: 01a137ca-5c77-7a71-b9ae-23ad9b0b5684

scripts/revoke-reuse-surface-forgejo-source.py

@@ -0,0 +1,109 @@
+#!/usr/bin/env python3
+"""Silent separately approved exact-PAT revocation; preserve KV and delivery.
+
+Forgejo11.0.3 DELETE token requires Basic auth. Reset only the dedicated user
+password to process-memory random bytes, revoke the reviewed token and reconcile
+inventory absence. Never remint, rotate/write KV, or touch another identity.
+"""
+import argparse
+import base64
+import hashlib
+import importlib.util
+import json
+import os
+from pathlib import Path
+import secrets
+import subprocess
+import sys
+
+ROOT = Path(__file__).resolve().parent.parent
+SCRIPT = 'scripts/revoke-reuse-surface-forgejo-source.py'
+spec = importlib.util.spec_from_file_location('reuse_reader_bootstrap', ROOT / 'scripts/provision-reuse-surface-forgejo-source.py')
+p = importlib.util.module_from_spec(spec)
+spec.loader.exec_module(p)
+
+
+def approved_target(path):
+    try:
+        record = json.loads(Path(path).read_text())
+        expected_files = {SCRIPT, p.SCRIPT_FILE, p.CCR_FILE, p.PLAN_FILE}
+        if record['ccr_id'] != p.CCR or record['decision'] != 'approved' or not record['supervisor'] or not record['decision_ref'] or record['action'] != 'revoke_exact_pat' or set(record['approved_artifacts']) != expected_files:
+            raise p.Refused('recovery_approval_invalid')
+        for rel in expected_files:
+            if hashlib.sha256((ROOT / rel).read_bytes()).hexdigest() != record['approved_artifacts'][rel]:
+                raise p.Refused('recovery_approved_revision_mismatch')
+        target = record['target']
+        if set(target) != {'id', 'name'} or type(target['id']) is not int or target['id'] <= 0 or not isinstance(target['name'], str) or not target['name'].startswith(p.PREFIX) or len(target['name']) > 100:
+            raise p.Refused('recovery_target_invalid')
+        return target
+    except p.Refused:
+        raise
+    except Exception:
+        raise p.Refused('recovery_approval_invalid') from None
+
+
+def recover(client, target):
+    _, user = client.request('GET', '/users/' + p.USER)
+    if not p.valid_user(user):
+        raise p.Refused('recovery_user_conflict')
+    tokens = client.pages('/users/' + p.USER + '/tokens')
+    if not tokens:
+        # Absence reconciles a prior ambiguous revocation without new writes.
+        client.journal.put('terminal', outcome='revoke_verified_absent_noop')
+        return
+    if len(tokens) != 1 or tokens[0].get('id') != target['id'] or tokens[0].get('name') != target['name'] or tokens[0].get('scopes') != p.SCOPES:
+        raise p.Refused('recovery_token_inventory_conflict')
+    password = secrets.token_urlsafe(48)
+    client.request('PATCH', '/admin/users/' + p.USER, {'password': password, 'must_change_password': False}, operation='recovery_password_randomize')
+    basic = 'Basic ' + base64.b64encode((p.USER + ':' + password).encode()).decode()
+    try:
+        client.request('DELETE', '/users/' + p.USER + '/tokens/' + str(target['id']), expected=(204, 404), basic=basic, operation='exact_pat_revoke')
+    except p.Refused:
+        # Observation failure does not mean DELETE did not apply. Read-only
+        # reconciliation follows; never issue another DELETE in this execution.
+        pass
+    remaining = client.pages('/users/' + p.USER + '/tokens')
+    if remaining:
+        raise p.Refused('revoke_unverified_stop')
+    client.journal.put('verified', operation='exact_pat_revoke', token_id=target['id'], status='inventory_absent')
+    client.journal.put('terminal', outcome='revoked_verified_kv_preserved')
+
+
+def main(argv=None):
+    parser = argparse.ArgumentParser()
+    parser.add_argument('--approval', required=True)
+    parser.add_argument('--receipt', required=True)
+    parser.add_argument('--inner', action='store_true', help=argparse.SUPPRESS)
+    args = parser.parse_args(argv)
+    journal, lock = None, None
+    try:
+        target = approved_target(args.approval)
+        if not args.inner:
+            result = subprocess.run(['warden', 'access', 'forgejo-admin-api-token', '--field', 'API_TOKEN', '--exec', '--', sys.executable, str(ROOT / SCRIPT), '--inner', '--approval', str(Path(args.approval).resolve()), '--receipt', str(Path(args.receipt).resolve())], capture_output=True, check=False, timeout=900, cwd=ROOT)
+            return 0 if result.returncode == 0 else 1
+        lock = os.open('/tmp/' + p.CCR + '-reuse-reader-' + str(os.getuid()) + '.lock', os.O_RDWR | os.O_CREAT | os.O_NOFOLLOW, 0o600)
+        if os.fstat(lock).st_uid != os.getuid() or os.fstat(lock).st_mode & 0o077:
+            raise p.Refused('lane_lock_owner_or_permissions')
+        p.fcntl.flock(lock, p.fcntl.LOCK_EX | p.fcntl.LOCK_NB)
+        journal = p.Journal(args.receipt)
+        admin = os.environ.get('API_TOKEN')
+        if not admin:
+            raise p.Refused('admin_route_missing')
+        recover(p.Client(admin, journal), target)
+        return 0
+    except Exception as error:
+        if journal:
+            journal.put('terminal', outcome='recovery_stopped_reconcile_before_retry', error_code=str(error) if isinstance(error, p.Refused) else 'internal_failure')
+        return 1
+    finally:
+        if journal:
+            journal.close()
+        if lock is not None:
+            os.close(lock)
+
+if __name__ == '__main__':
+    try:
+        result = main()
+    except Exception:
+        result = 1
+    raise SystemExit(result)