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

#!/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)