feat(recovery): prepare owner-bound offline continuity preflight

Assistant: codex
Assistant-Model: gpt-6.1-sol
Assistant-Session: 01a1468b-63fa-7710-85dc-8f2e16b3ebfa

tools/check-keycape-recovery-selector.py

@@ -0,0 +1,64 @@
+#!/usr/bin/env python3
+"""Offline public selector preflight; no imports of credential/backend clients.
+
+Checks current consumer source constants and the producer's already-deployed
+public env/denial selector. This is source metadata, not runtime/owner assent.
+"""
+import argparse
+import ast
+import hashlib
+import json
+import os
+from pathlib import Path
+import re
+import stat
+
+ROOT=Path(__file__).resolve().parents[1]
+LIMIT=128*1024
+
+def read(path):
+    path=Path(os.path.abspath(path))
+    if any(p.is_symlink() for p in (path,*path.parents)):raise ValueError()
+    fd=os.open(path,os.O_RDONLY|os.O_NOFOLLOW|os.O_NONBLOCK)
+    with os.fdopen(fd,'rb') as source:
+        info=os.fstat(source.fileno())
+        if not stat.S_ISREG(info.st_mode) or not 0<info.st_size<=LIMIT:raise ValueError()
+        raw=source.read(LIMIT+1)
+    if len(raw)>LIMIT:raise ValueError()
+    return raw
+
+def unique(pairs):
+    result={}
+    for key,value in pairs:
+        if key in result:raise ValueError()
+        result[key]=value
+    return result
+
+def verify(selector,source,source_sha256):
+    if not re.fullmatch('sha256:[a-f0-9]{64}',source_sha256) or 'sha256:'+hashlib.sha256(source).hexdigest()!=source_sha256:raise ValueError()
+    needed={'CLIENT_ID','SUBJECT','TENANT','ROLE','SCOPE'};constants={}
+    for node in ast.parse(source).body:
+        if isinstance(node,ast.Assign):
+            for target in node.targets:
+                if isinstance(target,ast.Name) and target.id in needed:
+                    if target.id in constants or not isinstance(node.value,ast.Constant) or not isinstance(node.value.value,str):raise ValueError()
+                    constants[target.id]=node.value.value
+    if set(constants)!=needed:raise ValueError()
+    expected={'client_id':constants['CLIENT_ID'],'secret_env':'KEYCAPE_SECRETS_ENGINE_OPENBAO_CLIENT_SECRET',
+      'audience':constants['CLIENT_ID'],'scope':constants['SCOPE'],'subject':constants['SUBJECT'],
+      'tenant':constants['TENANT'],'roles':constants['ROLE'],'deny_scope':'openbao:admin'}
+    if selector!=expected:raise ValueError()
+    return {'schema':'secrets-engine.recovery-selector-preflight.v1','status':'source_selector_matches_not_native_assent',
+      'source_sha256':source_sha256,'consumer':expected,'credential_source_exercised':False,'network_contacted':False}
+
+def main():
+    p=argparse.ArgumentParser(description=__doc__);p.add_argument('--selector',required=True);p.add_argument('--source-sha256',required=True)
+    args=p.parse_args()
+    try:
+        proposal=json.loads(read(args.selector),object_pairs_hook=unique)
+        if not isinstance(proposal,dict) or not isinstance(proposal.get('consumer'),dict):raise ValueError()
+        result=verify(proposal['consumer'],read(ROOT/'src/secrets_engine/service_auth.py'),args.source_sha256)
+    except Exception:print('{"status":"source_selector_refused","private_diagnostics_suppressed":true}');return 1
+    print(json.dumps(result,indent=2));return 0
+
+if __name__=='__main__':raise SystemExit(main())