secrets-engine/tools/check-keycape-recovery-selector.py

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

With --admission, additionally requires a founder SSH signature (gated-executor
pattern, namespace secrets-engine-recovery-attempt) over an attempt admission
that binds these exact selector bytes, source hash, issuer, case and a bounded
window. That is checkable founder authority for one attempt, still not native
acceptance; single-use consumption stays with the bounded NK49 execution.
"""
import argparse
import ast
from datetime import datetime, timezone
import hashlib
import json
import os
from pathlib import Path
import re
import stat

ROOT=Path(__file__).resolve().parents[1]
LIMIT=128*1024
ADMISSION_FIELDS={'schema','selector_sha256','source_sha256','issuer','case','not_before','not_after'}
MAX_WINDOW=86400

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 ts(value):
    if not isinstance(value,str) or not re.fullmatch(r'\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}Z',value):raise ValueError()
    return datetime.strptime(value,'%Y-%m-%dT%H:%M:%SZ').replace(tzinfo=timezone.utc)

def verify_attempt(admission,signature,*,selector_raw,issuer,source_sha256,now=None):
    """Founder-signed exact attempt binding; the signature covers the admission bytes."""
    import sys;sys.path.insert(0,str(ROOT/'src'))
    from secrets_engine.founder_admission import verify_admission
    payload=json.loads(admission,object_pairs_hook=unique)
    if not isinstance(payload,dict) or set(payload)!=ADMISSION_FIELDS:raise ValueError()
    if payload['schema']!='secrets-engine.recovery-attempt-admission.v1':raise ValueError()
    if payload['selector_sha256']!='sha256:'+hashlib.sha256(selector_raw).hexdigest():raise ValueError()
    if payload['source_sha256']!=source_sha256 or payload['issuer']!=issuer:raise ValueError()
    if not isinstance(payload['case'],str) or not re.fullmatch(r'[a-z0-9][a-z0-9.-]{0,127}',payload['case']):raise ValueError()
    start,end=ts(payload['not_before']),ts(payload['not_after']);now=now or datetime.now(timezone.utc)
    if not start<=now<end or (end-start).total_seconds()>MAX_WINDOW:raise ValueError()
    signer=verify_admission(admission,signature,namespace='secrets-engine-recovery-attempt')
    return {'case':payload['case'],'not_after':payload['not_after'],**signer}

def main():
    p=argparse.ArgumentParser(description=__doc__);p.add_argument('--selector',required=True);p.add_argument('--source-sha256',required=True)
    p.add_argument('--admission',help='founder-signed attempt admission JSON; signature at <admission>.sig')
    args=p.parse_args()
    try:
        selector_raw=read(args.selector)
        proposal=json.loads(selector_raw,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)
        if args.admission:
            result['attempt_admission']=verify_attempt(read(args.admission),read(args.admission+'.sig'),
                selector_raw=selector_raw,issuer=proposal.get('issuer'),source_sha256=args.source_sha256)
            result['status']='founder_admitted_attempt_not_native_acceptance'
    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())