security: prepare and rehearse contained Forgejo HTTP acceptance

docs/changes/CUST-WP-0080/accept_forge_routes.py

@@ -0,0 +1,316 @@
+#!/usr/bin/env python3
+"""Contained HTTP/API acceptance on one owned private disposable repository.
+
+Production entrypoint accepts API_TOKEN only from the protected child environment.
+No token, finding body, raw response or child diagnostic is written to receipts.
+The same Suite runs against the pinned isolated image before live admission.
+"""
+import argparse
+import base64
+import datetime
+import hashlib
+import importlib.util
+import json
+import os
+from pathlib import Path
+import re
+import subprocess
+import tempfile
+import time
+import urllib.error
+import urllib.request
+
+HERE = Path(__file__).parent
+ORIGIN = 'https://forgejo.coulomb.social'
+PROFILE = json.loads((HERE / 'receive-hook-proposal.json').read_text())
+
+
+def detected(body):
+    # Only parse the control's fixed status; never preserve response text.
+    return re.search(rb'"status"\s*:\s*"blocked"', body) is not None
+
+
+class Suite:
+    def __init__(self, backend, receipt, name):
+        if not re.fullmatch(r'custodian-routes-[a-f0-9]{24}', name):
+            raise ValueError('invalid-probe-name')
+        self.b, self.receipt, self.name = backend, receipt, name
+        self.path = '/api/v1/repos/' + backend.user + '/' + name
+        self.created_id = None
+        self.phase = 'preflight'
+        self.evidence = {'result': 'unverified', 'repository_name': name,
+                         'observed_at': datetime.datetime.now(datetime.timezone.utc).isoformat(),
+                         'checks': {}, 'scope': backend.scope}
+
+    def save(self):
+        self.evidence['phase'] = self.phase
+        self.evidence['executor_sha256'] = hashlib.sha256(Path(__file__).read_bytes()).hexdigest()
+        self.receipt.write_text(json.dumps(self.evidence, indent=2) + '\n')
+
+    def check(self, key, value):
+        self.evidence['checks'][key] = bool(value)
+        self.save()
+        if not value:
+            raise ValueError('acceptance-incomplete')
+
+    def api(self, method, suffix='', data=None, authenticated=True):
+        return self.b.api(method, self.path + suffix, data, authenticated)
+
+    def head(self):
+        code, body = self.api('GET', '/branches/main')
+        if code != 200:
+            raise ValueError('branch-read-refused')
+        oid = json.loads(body)['commit']['id']
+        if not re.fullmatch(r'[a-f0-9]{40}', oid):
+            raise ValueError('invalid-object-id')
+        return oid
+
+    def content(self, name, content, branch='main'):
+        return self.api('POST', '/contents/' + name,
+                        {'content': base64.b64encode(content).decode(),
+                         'message': 'Disposable route acceptance', 'branch': branch})
+
+    def run(self):
+        code, body = self.b.api('GET', '/api/v1/user')
+        self.check('expected_authenticated_caller', code == 200 and
+                   json.loads(body).get('login') == self.b.user)
+        self.check('target_does_not_preexist', self.api('GET')[0] == 404)
+        self.phase = 'create'
+        self.evidence['creation_attempted'] = True
+        self.save()  # An interrupted POST is ambiguous; never replay automatically.
+        code, body = self.b.api('POST', '/api/v1/user/repos',
+                               {'name': self.name, 'private': True, 'auto_init': True,
+                                'default_branch': 'main', 'description': self.name})
+        if code == 201:
+            data = json.loads(body)
+            if (type(data.get('id')) is int and data.get('name') == self.name and
+                    data.get('private') is True and data.get('owner', {}).get('login') == self.b.user):
+                self.created_id = data['id']
+                self.evidence['created_repository_id'] = self.created_id
+                self.save()
+        self.check('owned_private_repository_created', self.created_id is not None)
+        self.check('anonymous_private_read_denied', self.api('GET', authenticated=False)[0] in (401, 403, 404))
+        self.b.profile(self.name)
+        self.check('new_repository_inherits_exact_hook', True)
+        self.phase = 'http-git'
+        remote = self.b.origin + '/' + self.b.user + '/' + self.name + '.git'
+        self.b.git('init', '-q', '-b', 'main')
+        self.b.git('config', 'core.hooksPath', '/dev/null')
+        self.b.git('config', 'user.name', 'Disposable route probe')
+        self.b.git('config', 'user.email', 'probe@example.test')
+        self.b.git('fetch', '-q', remote, 'main')
+        self.b.git('reset', '--hard', 'FETCH_HEAD')
+        self.b.put('clean.txt', b'Ordinary disposable source.\n')
+        self.b.git('add', 'clean.txt'); self.b.git('commit', '-qm', 'Clean HTTP proof')
+        r = self.b.git('push', remote, 'HEAD:refs/heads/main', allow_failure=True)
+        self.check('http_clean_push_accepted', r.returncode == 0 and
+                   re.search(rb'"status"\s*:\s*"clean"', r.stdout + r.stderr) is not None)
+        clean_head = self.head()
+        self.check('http_clean_ref_matches', self.b.git('rev-parse', 'HEAD').stdout.decode().strip() == clean_head)
+        fixture = self.b.fixture()
+        self.b.put('synthetic.txt', fixture)
+        self.b.git('add', 'synthetic.txt'); self.b.git('commit', '-qm', 'Unbound synthetic detection fixture')
+        oid = self.b.git('rev-parse', 'HEAD').stdout.decode().strip()
+        r = self.b.git('push', remote, 'HEAD:refs/heads/refused', allow_failure=True)
+        self.check('http_secret_push_detected_and_refused', r.returncode != 0 and detected(r.stdout + r.stderr))
+        self.check('http_refused_ref_absent', self.api('GET', '/branches/refused')[0] == 404)
+        self.check('http_rejection_preserves_main', self.head() == clean_head)
+        self.b.absent_object(self.name, oid)
+        self.check('http_quarantine_object_absent', True)
+        self.phase = 'content-api'
+        self.check('api_clean_content_accepted', self.content('accepted.txt', b'Clean API source.\n')[0] == 201)
+        before = self.head()
+        code, body = self.content('refused.txt', fixture)
+        self.check('api_secret_content_refused', code == 500)
+        # API error bodies encode the hook message as JSON; decode privately.
+        message = json.loads(body).get('message', '').encode()
+        self.check('api_refusal_has_detector_evidence', detected(message))
+        self.check('api_refusal_preserves_main', self.head() == before)
+        self.check('api_refused_content_absent', self.api('GET', '/contents/refused.txt')[0] == 404)
+        self.phase = 'merge-api'
+        self.check('feature_branch_created', self.api('POST', '/branches',
+                   {'new_branch_name': 'feature', 'old_branch_name': 'main'})[0] == 201)
+        self.check('feature_content_accepted', self.content('feature.txt', b'Clean feature.\n', 'feature')[0] == 201)
+        code, body = self.api('POST', '/pulls', {'head': 'feature', 'base': 'main', 'title': 'Disposable merge proof'})
+        self.check('pull_request_created', code == 201)
+        number = json.loads(body)['number']
+        if type(number) is not int:
+            raise ValueError('invalid-pull-number')
+        suffix = '/pulls/' + str(number)
+        for _ in range(40):
+            code, body = self.api('GET', suffix)
+            if code == 200 and json.loads(body).get('mergeable') is True:
+                break
+            time.sleep(.5)
+        else:
+            raise ValueError('pull-not-mergeable')
+        before = self.head()
+        code, body = self.api('POST', suffix + '/merge', {'Do': 'merge',
+                             'MergeTitleField': 'Disposable merge', 'MergeMessageField': fixture.decode(),
+                             'force_merge': False})
+        self.check('merge_secret_message_detected_and_refused', code == 409 and
+                   detected(json.loads(body).get('message', '').encode()))
+        code, body = self.api('GET', suffix)
+        self.check('rejected_merge_not_applied', code == 200 and json.loads(body).get('merged') is False and self.head() == before)
+        self.check('clean_merge_accepted', self.api('POST', suffix + '/merge', {'Do': 'merge',
+                   'MergeTitleField': 'Clean merge', 'MergeMessageField': 'Clean proof.'})[0] in (200, 201))
+        code, body = self.api('GET', suffix)
+        self.check('clean_merge_applied', code == 200 and json.loads(body).get('merged') is True and self.head() != before)
+        self.evidence['acceptance_passed'] = True
+
+    def close(self):
+        self.phase = 'cleanup'
+        if self.created_id is not None:
+            code, body = self.api('GET')
+            data = json.loads(body) if code == 200 else {}
+            self.check('cleanup_identity_matches', code == 200 and data.get('id') == self.created_id and
+                       data.get('name') == self.name and data.get('private') is True and
+                       data.get('description') == self.name and data.get('owner', {}).get('login') == self.b.user)
+            self.check('owned_disposable_repository_deleted', self.api('DELETE')[0] == 204)
+            self.check('deleted_repository_absent', self.api('GET')[0] == 404)
+        elif self.evidence.get('creation_attempted'):
+            self.evidence['creation_outcome_requires_reconciliation'] = True
+        if self.evidence.get('acceptance_passed') and self.evidence['checks'].get('deleted_repository_absent'):
+            self.evidence['result'] = 'verified'
+        self.save()
+
+
+class NoRedirect(urllib.request.HTTPRedirectHandler):
+    def redirect_request(self, *args, **kwargs):
+        return None
+
+
+class Production:
+    user = 'tegwick'
+    origin = ORIGIN
+    scope = ('Deployed HTTPS Git, repository creation, content API and merge API on one owned private '
+             'disposable repository. No browser/SSO, CI, import/mirror or finding-response conformance inferred.')
+
+    def __init__(self, token, directory):
+        if not token or '\n' in token or '\r' in token:
+            raise ValueError('injected-token-absent-or-invalid')
+        self.token, self.directory = token, directory
+        spec = importlib.util.spec_from_file_location('template', HERE / 'activate_forge_template.py')
+        self.native = importlib.util.module_from_spec(spec); spec.loader.exec_module(self.native)
+        self.pod = self.native.target_pod(self.native.deployment(), PROFILE)
+        self.native.check_artifacts(self.pod)
+        self.opener = urllib.request.build_opener(urllib.request.ProxyHandler({}), NoRedirect())
+
+    def api(self, method, path, data=None, authenticated=True):
+        if not path.startswith('/api/v1/') or '://' in path:
+            raise ValueError('invalid-api-path')
+        headers = {'Accept': 'application/json', 'Content-Type': 'application/json'}
+        if authenticated:
+            headers['Authorization'] = 'token ' + self.token
+        request = urllib.request.Request(self.origin + path, method=method, headers=headers,
+                                         data=None if data is None else json.dumps(data).encode())
+        try:
+            response = self.opener.open(request, timeout=120)
+        except urllib.error.HTTPError as error:
+            response = error
+        with response:
+            body = response.read(1024 * 1024 + 1)
+            if len(body) > 1024 * 1024:
+                raise ValueError('response-too-large')
+            return response.code, body
+
+    def git(self, *args, allow_failure=False):
+        auth = base64.b64encode((self.user + ':' + self.token).encode()).decode()
+        settings = {'credential.helper': '', 'http.followRedirects': 'false', 'http.proxy': '',
+                    'http.sslVerify': 'true', 'http.' + self.origin + '/.extraHeader': 'Authorization: Basic ' + auth}
+        env = {'PATH': '/usr/bin:/bin', 'HOME': str(self.directory), 'LANG': 'C',
+               'GIT_CONFIG_NOSYSTEM': '1', 'GIT_CONFIG_GLOBAL': '/dev/null', 'GIT_TERMINAL_PROMPT': '0',
+               'GIT_ALLOW_PROTOCOL': 'https', 'GIT_CONFIG_COUNT': str(len(settings))}
+        for i, (key, value) in enumerate(settings.items()):
+            env['GIT_CONFIG_KEY_' + str(i)] = key; env['GIT_CONFIG_VALUE_' + str(i)] = value
+        result = subprocess.run(['git', '-C', str(self.directory), *args], env=env, capture_output=True, timeout=120)
+        if result.returncode and not allow_failure:
+            raise ValueError('git-command-refused')
+        return result
+
+    def put(self, name, content):
+        (self.directory / name).write_bytes(content)
+
+    def fixture(self):
+        result = subprocess.run(['ssh-keygen', '-q', '-t', 'rsa', '-b', '2048', '-N', '',
+                                 '-f', str(self.directory / 'unbound-key')], capture_output=True, timeout=30)
+        if result.returncode:
+            raise ValueError('fixture-generation-failed')
+        return (self.directory / 'unbound-key').read_bytes()
+
+    def profile(self, name):
+        root = '/data/git/gitea-repositories/' + self.user + '/' + name + '.git'
+        hook = root + '/hooks/pre-receive.d/' + PROFILE['hook_name']
+        self.native.native.execute(self.pod, 'sh', '-c',
+                                   'test -f "$1" && test ! -L "$1" && test -x "$1"', 'sh', hook)
+        observed = self.native.native.execute(self.pod, 'sha256sum', hook).decode().split()[0]
+        mode = self.native.native.execute(self.pod, 'stat', '-c', '%a %u %g', hook).decode().strip()
+        if observed != PROFILE['hook_sha256'] or mode not in ('555 1000 1000', '755 1000 1000'):
+            raise ValueError('inherited-profile-mismatch')
+
+    def absent_object(self, name, oid):
+        if not re.fullmatch(r'[a-f0-9]{40}', oid):
+            raise ValueError('invalid-object-id')
+        root = '/data/git/gitea-repositories/' + self.user + '/' + name + '.git'
+        raw = self.native.native.execute(self.pod, 'git', '--git-dir=' + root, 'cat-file',
+                                         '--batch-check', payload=(oid + '\n').encode()).decode().strip()
+        if raw != oid + ' missing':
+            raise ValueError('quarantine-retained')
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument('--receipt', type=Path, required=True)
+    parser.add_argument('--name', required=True)
+    parser.add_argument('--proposal', type=Path, required=True)
+    parser.add_argument('--cleanup-only', action='store_true')
+    args = parser.parse_args()
+    # Remove injection before invoking SSH or other unrelated child tools.
+    token = os.environ.pop('API_TOKEN', '')
+    if args.receipt.exists() and not args.cleanup_only:
+        return 2
+    suite = None
+    try:
+        proposal_bytes = args.proposal.read_bytes()
+        proposal = json.loads(proposal_bytes)
+        if (proposal['repository_name'] != args.name or proposal['origin'] != ORIGIN or
+                proposal['user'] != Production.user):
+            raise ValueError('proposal-target-mismatch')
+        for name, digest in proposal['source_sha256'].items():
+            if Path(name).name != name or hashlib.sha256((HERE / name).read_bytes()).hexdigest() != digest:
+                raise ValueError('proposal-source-mismatch')
+        if 'accept_forge_routes.py' not in proposal['source_sha256']:
+            raise ValueError('proposal-executor-unbound')
+        with tempfile.TemporaryDirectory(prefix='forge-routes-') as directory:
+            backend = Production(token, Path(directory))
+            suite = Suite(backend, args.receipt, args.name)
+            if args.cleanup_only:
+                previous = json.loads(args.receipt.read_text())
+                if (previous.get('repository_name') != args.name or
+                        previous.get('proposal_sha256') != hashlib.sha256(proposal_bytes).hexdigest() or
+                        type(previous.get('created_repository_id')) is not int):
+                    raise ValueError('cleanup-ownership-unverified')
+                suite.evidence = previous
+                suite.created_id = previous['created_repository_id']
+                suite.close()
+                return 0 if suite.evidence['checks'].get('deleted_repository_absent') else 1
+            suite.evidence['proposal_sha256'] = hashlib.sha256(proposal_bytes).hexdigest()
+            try:
+                suite.run()
+            except Exception:
+                suite.evidence['failure_phase'] = suite.phase
+            finally:
+                try:
+                    suite.close()
+                except Exception:
+                    suite.evidence['cleanup_unverified'] = True
+                    suite.save()
+        return 0 if suite.evidence['result'] == 'verified' else 1
+    except Exception:
+        if not args.receipt.exists():
+            args.receipt.write_text(json.dumps({'result': 'unverified', 'failure_phase': 'preflight'}) + '\n')
+        return 1
+
+
+if __name__ == '__main__':
+    raise SystemExit(main())

docs/changes/CUST-WP-0080/forge-routes-proposal-pre-rehearsal.json

@@ -0,0 +1,28 @@
+{
+  "proposal_id": "FORGE6-ROUTES-001",
+  "revision": 1,
+  "state": "prepared-awaiting-isolated-proof-and-attended-login",
+  "origin": "https://forgejo.coulomb.social",
+  "user": "tegwick",
+  "repository_name": "custodian-routes-f140c6b840da4bbb9de64a74",
+  "source_sha256": {
+    "accept_forge_routes.py": "f87e643f1d49de031c2bfc3f956280eca0c61d2126fb20fd04541aa6324ef565",
+    "run_forge_routes.py": "6258140d408e5fbed7373611cdcd65e64aee743a0b36ce4fe55b5dc94663120f",
+    "receive-hook-proposal.json": "9c8ea3b7fbe16a997359432af7bcf09f429881fd23cc567188c21129e80ccf8d",
+    "activate_forge_template.py": "ef2b38f655d5520d69dcd69cdb4997eaad4b50773f32c1d798d16331b204c789",
+    "activate_receive_hooks.py": "9b01ee383e16463981112226e19156b0de59af2d03dd71b9403e144e280a9fb3",
+    "verify_receive_native.py": "c9120122c6f6f0b7c41fe4fa975e5a2a340e776885d852b861b8c13b7664adcd"
+  },
+  "target": "One newly created private disposable repository; pinned deployed Forgejo and admitted hook. Existing repository content, hooks, accounts and policies remain untouched.",
+  "expected_result": "Authenticated native HTTPS clean acceptance and synthetic-secret refusal with detector evidence; API private creation inherits exact hook; clean/content-secret and clean/merge-message refusal with unchanged refs; quarantine commit absent; anonymous private access refused; owned test repository deleted and absent.",
+  "verification": "Exact shared Suite rehearsed against pinned isolated Forgejo; four boundary regression tests. Live receipt required; rehearsal is not deployed acceptance.",
+  "rollback": "Automatic cleanup checks numeric repository ID, owner, private flag, name and ownership description before deleting only the newly created test repository. On interrupted known creation, rerun run_forge_routes.py with the same proposal and receipt plus --cleanup-only. Ambiguous POST before numeric ID receipt requires reconciliation, never replay or delete by name.",
+  "abort_owner": "railiance-forge; custodian-codex prepares and verifies receipts",
+  "cost_estimate_eur": 0,
+  "cost_basis": "Existing service only; serialized small private probes, no added resources or spend commitments.",
+  "risk_estimate": "Bounded synthetic test writes to a unique private disposable repository. No real secret fixtures, account creation, access-policy change or external publication. Unexpected acceptance may retain an unbound synthetic private-key fixture until cleanup. API/network failure stops the trial; ambiguous creation and cleanup remain explicitly unverified.",
+  "authorization_ref": "canon/standards/build-mode-change-authorization_v0.1.md",
+  "lane": "orange: existing caller OIDC/MFA required; no access or secret-policy scope change",
+  "live_receipt": "docs/evidence/2026-10-13-netkingdom-security/forge-routes-live.json",
+  "remaining_scope": "Browser/SSO journey, external gitlink repository, CI, import/mirror, historical/LFS/log/artifact detection and finding response remain open."
+}

docs/changes/CUST-WP-0080/forge-routes-proposal.json

@@ -0,0 +1,30 @@
+{
+  "proposal_id": "FORGE6-ROUTES-001",
+  "revision": 2,
+  "state": "prepared-isolated-verified-awaiting-attended-login",
+  "origin": "https://forgejo.coulomb.social",
+  "user": "tegwick",
+  "repository_name": "custodian-routes-f140c6b840da4bbb9de64a74",
+  "source_sha256": {
+    "accept_forge_routes.py": "96608481ab62445d2688e20f48c76a35c3d7be26c37452be3c46202eec4e103c",
+    "run_forge_routes.py": "6258140d408e5fbed7373611cdcd65e64aee743a0b36ce4fe55b5dc94663120f",
+    "receive-hook-proposal.json": "9c8ea3b7fbe16a997359432af7bcf09f429881fd23cc567188c21129e80ccf8d",
+    "activate_forge_template.py": "ef2b38f655d5520d69dcd69cdb4997eaad4b50773f32c1d798d16331b204c789",
+    "activate_receive_hooks.py": "9b01ee383e16463981112226e19156b0de59af2d03dd71b9403e144e280a9fb3",
+    "verify_receive_native.py": "c9120122c6f6f0b7c41fe4fa975e5a2a340e776885d852b861b8c13b7664adcd"
+  },
+  "target": "One newly created private disposable repository; pinned deployed Forgejo and admitted hook. Existing repository content, hooks, accounts and policies remain untouched.",
+  "expected_result": "Authenticated native HTTPS clean acceptance and synthetic-secret refusal with detector evidence; API private creation inherits exact hook; clean/content-secret and clean/merge-message refusal with unchanged refs; quarantine commit absent; anonymous private access refused; owned test repository deleted and absent.",
+  "verification": "Exact shared Suite rehearsed against pinned isolated Forgejo; five boundary regression tests. Live receipt required; rehearsal is not deployed acceptance.",
+  "rollback": "Automatic cleanup checks numeric repository ID, owner, private flag, name and ownership description before deleting only the newly created test repository. On interrupted known creation, rerun run_forge_routes.py with the same proposal and receipt plus --cleanup-only. Ambiguous POST before numeric ID receipt requires reconciliation, never replay or delete by name.",
+  "abort_owner": "railiance-forge; custodian-codex prepares and verifies receipts",
+  "cost_estimate_eur": 0,
+  "cost_basis": "Existing service only; serialized small private probes, no added resources or spend commitments.",
+  "risk_estimate": "Bounded synthetic test writes to a unique private disposable repository. No real secret fixtures, account creation, access-policy change or external publication. Unexpected acceptance may retain an unbound synthetic private-key fixture until cleanup. API/network failure stops the trial; ambiguous creation and cleanup remain explicitly unverified.",
+  "authorization_ref": "canon/standards/build-mode-change-authorization_v0.1.md",
+  "lane": "orange: existing caller OIDC/MFA required; no access or secret-policy scope change",
+  "live_receipt": "docs/evidence/2026-10-13-netkingdom-security/forge-routes-live.json",
+  "remaining_scope": "Browser/SSO journey, external gitlink repository, CI, import/mirror, historical/LFS/log/artifact detection and finding response remain open.",
+  "original_revision_sha256": "a253fce3a565a2af94133da019bdc434dc57b21a8c61b06b4c27e036cf5a01d3",
+  "refinement": "Pre-execution rehearsal corrected inherited Git hook mode to accept exact hash/owner with 0555 or 0755; initial failed rehearsal and source-binding limitation are retained."
+}

docs/changes/CUST-WP-0080/run_forge_routes.py

@@ -0,0 +1,49 @@
+#!/usr/bin/env python3
+"""Silent attended child: protected PAT injection, then a value-free receipt."""
+import argparse
+import hashlib
+import json
+from pathlib import Path
+import subprocess
+import sys
+
+HERE = Path(__file__).resolve().parent
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument('--proposal', type=Path, required=True)
+    parser.add_argument('--receipt', type=Path, required=True)
+    parser.add_argument('--cleanup-only', action='store_true')
+    args = parser.parse_args()
+    proposal_path, receipt = args.proposal.resolve(), args.receipt.resolve()
+    transport = receipt.with_suffix('.cleanup-transport.json' if args.cleanup_only else '.transport.json')
+    if transport.exists() or (receipt.exists() and not args.cleanup_only):
+        return 2
+    try:
+        proposal = json.loads(proposal_path.read_text())
+        for name, digest in proposal['source_sha256'].items():
+            if Path(name).name != name or hashlib.sha256((HERE / name).read_bytes()).hexdigest() != digest:
+                return 2
+        if not {'run_forge_routes.py', 'accept_forge_routes.py'} <= set(proposal['source_sha256']):
+            return 2
+        child = [sys.executable, str(HERE / 'accept_forge_routes.py'), '--proposal', str(proposal_path),
+                 '--receipt', str(receipt), '--name', proposal['repository_name']]
+        if args.cleanup_only:
+            child.append('--cleanup-only')
+        # Capture/discard both streams, including warden proxy notices. The
+        # containing login lane requires silence and handles session revocation.
+        result = subprocess.run(['warden', 'access', 'forgejo-admin-api-token', '--field', 'API_TOKEN',
+                                 '--exec', '--', *child], capture_output=True)
+        transport.write_text(json.dumps({'route_exit_code': result.returncode,
+            'receipt_present': receipt.exists(), 'cleanup_only': args.cleanup_only,
+            'proposal_sha256': hashlib.sha256(proposal_path.read_bytes()).hexdigest(),
+            'raw_diagnostics_retained': False, 'credential_returned_to_agent': False}, indent=2) + '\n')
+        return result.returncode
+    except Exception:
+        transport.write_text(json.dumps({'result': 'handoff-unverified', 'raw_diagnostics_retained': False}) + '\n')
+        return 1
+
+
+if __name__ == '__main__':
+    raise SystemExit(main())

docs/changes/CUST-WP-0080/test_forge_routes_isolated.py

@@ -0,0 +1,121 @@
+#!/usr/bin/env python3
+"""Run the exact shared route suite on the pinned disposable Forgejo image."""
+import argparse
+import base64
+import hashlib
+import importlib.util
+import json
+from pathlib import Path
+import uuid
+
+HERE = Path(__file__).parent
+
+
+def load(name, filename):
+    spec = importlib.util.spec_from_file_location(name, HERE / filename)
+    module = importlib.util.module_from_spec(spec); spec.loader.exec_module(module)
+    return module
+
+
+base = load('lifecycle', 'verify_forge_lifecycle.py')
+routes = load('routes', 'accept_forge_routes.py')
+LOADED_ROUTES_SHA256 = hashlib.sha256((HERE / 'accept_forge_routes.py').read_bytes()).hexdigest()
+
+
+class FixtureReady(Exception):
+    pass
+
+
+class SetupProbe(base.Probe):
+    def check(self, name, condition):
+        super().check(name, condition)
+        if name == 'native_repository_created':
+            raise FixtureReady()
+
+
+class Backend:
+    user = 'probe'
+    origin = 'http://127.0.0.1:3000'
+    scope = 'Exact shared HTTP/API suite on pinned isolated Forgejo, SQLite, no external network; no live credential or deployed acceptance.'
+
+    def __init__(self, probe):
+        self.p = probe
+        self.directory = base.ROOT + '/routes-client'
+        probe.execute('mkdir', self.directory)
+
+    def api(self, method, path, data=None, authenticated=True):
+        return self.p.api(method, path, data, authenticated)
+
+    def git(self, *args, allow_failure=False):
+        auth = base64.b64encode(('probe:' + self.p.password).encode()).decode()
+        return self.p.execute('env', 'GIT_CONFIG_COUNT=1',
+                              'GIT_CONFIG_KEY_0=http.http://127.0.0.1:3000/.extraHeader',
+                              'GIT_CONFIG_VALUE_0=Authorization: Basic ' + auth,
+                              'git', '-C', self.directory, *args,
+                              allowed=(0, 1, 128) if allow_failure else (0,), timeout=120)
+
+    def put(self, name, content):
+        self.p.put(self.directory + '/' + name, content)
+
+    def fixture(self):
+        return self.p.execute('cat', base.ROOT + '/synthetic-key').stdout
+
+    def profile(self, name):
+        path = self.p.profile(name)
+        digest = self.p.execute('sha256sum', path).stdout.decode().split()[0]
+        mode = self.p.execute('stat', '-c', '%a %u %g', path).stdout.decode().strip()
+        self.p.evidence['inherited_hook_mode_owner'] = mode
+        if digest != routes.PROFILE['hook_sha256'] or mode not in ('555 1000 1000', '755 1000 1000'):
+            raise ValueError('inherited-profile-mismatch')
+
+    def absent_object(self, name, oid):
+        root = base.REPOS + '/' + name + '.git'
+        raw = self.p.execute('git', '--git-dir=' + root, 'cat-file', '--batch-check',
+                             payload=(oid + '\n').encode()).stdout.decode().strip()
+        if raw != oid + ' missing':
+            raise ValueError('quarantine-retained')
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument('--artifact-dir', type=Path, required=True)
+    parser.add_argument('--receipt', type=Path, required=True)
+    args = parser.parse_args()
+    if args.receipt.exists():
+        raise SystemExit('Retain prior attempt; use a new receipt.')
+    probe = SetupProbe(args.receipt.with_suffix('.baseline.json'))
+    suite = None
+    try:
+        try:
+            probe.run_probe(args.artifact_dir)
+        except FixtureReady:
+            pass
+        probe.evidence['scope'] = 'Pinned-image fixture setup only; shared suite has its own acceptance receipt.'
+        template = base.ROOT + '/template'
+        probe.execute('mkdir', '-p', template + '/hooks/pre-receive.d')
+        probe.put(template + '/hooks/pre-receive.d/10-custodian-secret-scan',
+                  (HERE / 'receive-profile-hook').read_bytes(), '555')
+        probe.execute('git', 'config', '--file', base.ROOT + '/git-home/.gitconfig', 'init.templateDir', template)
+        probe.execute('ssh-keygen', '-q', '-t', 'rsa', '-b', '2048', '-N', '', '-f', base.ROOT + '/synthetic-key')
+        probe.evidence['result'] = 'fixture-ready'
+        suite = routes.Suite(Backend(probe), args.receipt, 'custodian-routes-' + uuid.uuid4().hex[:24])
+        suite.evidence['loaded_executor_sha256'] = LOADED_ROUTES_SHA256
+        suite.run()
+        suite.check('executor_unchanged_during_rehearsal', LOADED_ROUTES_SHA256 ==
+                    hashlib.sha256((HERE / 'accept_forge_routes.py').read_bytes()).hexdigest())
+    except Exception:
+        if suite:
+            suite.evidence['failure_phase'] = suite.phase
+        raise SystemExit('Shared route acceptance incomplete; raw diagnostics suppressed.')
+    finally:
+        try:
+            if suite:
+                suite.close()
+        finally:
+            probe.close()
+    print(json.dumps({'result': suite.evidence['result'], 'checks': len(suite.evidence['checks']),
+                      'receipt': str(args.receipt)}))
+
+
+if __name__ == '__main__':
+    main()

docs/changes/CUST-WP-0080/test_forge_routes_safety.py

@@ -0,0 +1,111 @@
+"""Boundary regressions for receipt-scoped cleanup and credential forwarding."""
+import importlib.util
+import contextlib
+import hashlib
+import io
+import json
+from pathlib import Path
+import tempfile
+import threading
+import unittest
+from unittest import mock
+import urllib.request
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+
+spec = importlib.util.spec_from_file_location('routes', Path(__file__).with_name('accept_forge_routes.py'))
+m = importlib.util.module_from_spec(spec); spec.loader.exec_module(m)
+
+
+class CleanupTests(unittest.TestCase):
+    def test_replaced_or_changed_repository_is_not_deleted(self):
+        for change in ({'id': 18}, {'private': False}, {'description': 'changed'},
+                       {'owner': {'login': 'someone-else'}}):
+            with self.subTest(change=change), tempfile.TemporaryDirectory() as directory:
+                name = 'custodian-routes-' + 'a' * 24
+                record = dict(id=17, name=name, private=True, description=name, owner={'login': 'tegwick'})
+                record.update(change)
+                class Backend:
+                    user = 'tegwick'
+                    scope = 'test'
+                    def api(self, method, *args):
+                        self_outer.assertEqual(method, 'GET', 'must not delete changed ownership')
+                        return 200, json.dumps(record).encode()
+                self_outer = self
+                suite = m.Suite(Backend(), Path(directory) / 'receipt.json', name)
+                suite.created_id = 17
+                with self.assertRaises(ValueError):
+                    suite.close()
+
+    def test_ambiguous_creation_never_deletes_by_name(self):
+        class Backend:
+            user = 'tegwick'
+            scope = 'test'
+            def api(self, *args):
+                raise AssertionError('unowned creation must not issue API calls')
+        with tempfile.TemporaryDirectory() as directory:
+            suite = m.Suite(Backend(), Path(directory) / 'receipt.json', 'custodian-routes-' + 'a' * 24)
+            suite.evidence['creation_attempted'] = True
+            suite.close()
+            self.assertTrue(suite.evidence['creation_outcome_requires_reconciliation'])
+            self.assertEqual(suite.evidence['result'], 'unverified')
+
+    def test_generic_error_is_not_detection_evidence(self):
+        self.assertFalse(m.detected(b'{"message":"internal error"}'))
+        self.assertFalse(m.detected(b'{"status":"unverified"}'))
+
+
+class TransportTests(unittest.TestCase):
+    def test_http_redirect_does_not_forward_injected_credential(self):
+        hits = []
+        class Handler(BaseHTTPRequestHandler):
+            def do_GET(self):
+                hits.append(self.path)
+                self.send_response(302)
+                self.send_header('Location', '/receiver')
+                self.end_headers()
+            def log_message(self, *args):
+                pass
+        server = ThreadingHTTPServer(('127.0.0.1', 0), Handler)
+        worker = threading.Thread(target=server.serve_forever, daemon=True); worker.start()
+        try:
+            backend = object.__new__(m.Production)
+            backend.origin = 'http://127.0.0.1:' + str(server.server_port)
+            backend.token = 'unbound-test-value'
+            backend.opener = urllib.request.build_opener(urllib.request.ProxyHandler({}), m.NoRedirect())
+            code, _ = backend.api('GET', '/api/v1/user')
+            self.assertEqual(code, 302)
+            self.assertEqual(hits, ['/api/v1/user'])
+        finally:
+            server.shutdown(); server.server_close(); worker.join()
+
+    def test_handoff_is_silent_and_rejects_changed_source_before_fetch(self):
+        spec = importlib.util.spec_from_file_location('handoff', Path(__file__).with_name('run_forge_routes.py'))
+        wrapper = importlib.util.module_from_spec(spec); spec.loader.exec_module(wrapper)
+        for changed in (False, True):
+            with self.subTest(changed=changed), tempfile.TemporaryDirectory() as directory:
+                root = Path(directory)
+                proposal = {'repository_name': 'custodian-routes-' + 'a' * 24, 'source_sha256': {
+                    n: hashlib.sha256((wrapper.HERE / n).read_bytes()).hexdigest()
+                    for n in ('run_forge_routes.py', 'accept_forge_routes.py')}}
+                if changed:
+                    proposal['source_sha256']['accept_forge_routes.py'] = '0' * 64
+                (root / 'proposal.json').write_text(json.dumps(proposal))
+                out, err = io.StringIO(), io.StringIO()
+                with mock.patch('sys.argv', ['handoff', '--proposal', str(root / 'proposal.json'),
+                                '--receipt', str(root / 'result.json')]), mock.patch.object(
+                                    wrapper.subprocess, 'run', return_value=mock.Mock(returncode=5,
+                                    stdout=b'UNBOUND-CREDENTIAL-MARKER', stderr=b'UNBOUND-CREDENTIAL-MARKER')) as child:
+                    with contextlib.redirect_stdout(out), contextlib.redirect_stderr(err):
+                        result = wrapper.main()
+                self.assertEqual(out.getvalue() + err.getvalue(), '')
+                if changed:
+                    child.assert_not_called()
+                    self.assertEqual(result, 2)
+                else:
+                    self.assertEqual(result, 5)
+                    self.assertTrue(child.call_args.kwargs['capture_output'])
+                    self.assertNotIn('UNBOUND-CREDENTIAL-MARKER', (root / 'result.transport.json').read_text())
+
+
+if __name__ == '__main__':
+    unittest.main()