secrets-engine/tests/test_keycape_recovery_selector.py

import hashlib
import importlib.util
import json
from pathlib import Path
import pytest

spec=importlib.util.spec_from_file_location('check',Path(__file__).parents[1]/'tools/check-keycape-recovery-selector.py');m=importlib.util.module_from_spec(spec);spec.loader.exec_module(m)
SOURCE=b'CLIENT_ID="secrets-engine-openbao"\nSUBJECT="service:secrets-engine"\nTENANT="tenant:platform"\nROLE="secrets-engine"\nSCOPE="openbao:login"\nraise RuntimeError("source is parsed, never executed")\n'
SELECTOR={'client_id':'secrets-engine-openbao','secret_env':'KEYCAPE_SECRETS_ENGINE_OPENBAO_CLIENT_SECRET','audience':'secrets-engine-openbao','scope':'openbao:login','subject':'service:secrets-engine','tenant':'tenant:platform','roles':'secrets-engine','deny_scope':'openbao:admin'}
DIGEST='sha256:'+hashlib.sha256(SOURCE).hexdigest()
def test_source_parsed_not_executed():
    result=m.verify(SELECTOR,SOURCE,DIGEST)
    assert result['network_contacted'] is False
@pytest.mark.parametrize('field',list(SELECTOR))
def test_each_selector_binding_is_exact(field):
    bad={**SELECTOR,field:'different'}
    with pytest.raises(ValueError):m.verify(bad,SOURCE,DIGEST)
def test_source_drift_and_untyped_digest_refused():
    for source,digest in ((SOURCE+b'\n',DIGEST),(SOURCE,DIGEST.removeprefix('sha256:'))):
        with pytest.raises(ValueError):m.verify(SELECTOR,source,digest)
def test_duplicate_constants_refused():
    source=SOURCE+b'CLIENT_ID="other"\n'
    with pytest.raises(ValueError):m.verify(SELECTOR,source,'sha256:'+hashlib.sha256(source).hexdigest())
def test_duplicate_public_metadata_keys_refused():
    with pytest.raises(ValueError):json.loads('{"consumer":{},"consumer":{}}',object_pairs_hook=m.unique)
def test_oversize_public_input_refused(tmp_path):
    p=tmp_path/'selector.json';p.write_bytes(b'x'*(m.LIMIT+1))
    with pytest.raises(ValueError):m.read(p)
def test_symlink_public_input_refused(tmp_path):
    p=tmp_path/'selector.json';p.write_text('{}');link=tmp_path/'link';link.symlink_to(p)
    with pytest.raises(ValueError):m.read(link)

# Founder attempt admission (SECRETS-WP-0015-T05), gated-executor signature pattern.
from datetime import datetime, timezone
from secrets_engine.errors import DecisionError
from tests.signature_support import admit, FOUNDER
NOW=datetime(2026,10,21,12,0,0,tzinfo=timezone.utc)
ISSUER='https://kc.coulomb.social'
SELECTOR_RAW=json.dumps({'issuer':ISSUER,'consumer':SELECTOR}).encode()
ADMISSION={'schema':'secrets-engine.recovery-attempt-admission.v1','selector_sha256':'sha256:'+hashlib.sha256(SELECTOR_RAW).hexdigest(),
           'source_sha256':DIGEST,'issuer':ISSUER,'case':'secrets15-nk49-readiness',
           'not_before':'2026-10-21T11:00:00Z','not_after':'2026-10-21T13:00:00Z'}
def signed(tmp_path,payload=ADMISSION,namespace='secrets-engine-recovery-attempt'):
    p=tmp_path/'attempt.json';p.write_text(json.dumps(payload));admit(p,namespace=namespace)
    return p.read_bytes(),(tmp_path/'attempt.json.sig').read_bytes()
def attempt(data,sig,selector_raw=SELECTOR_RAW,now=NOW):
    return m.verify_attempt(data,sig,selector_raw=selector_raw,issuer=ISSUER,source_sha256=DIGEST,now=now)
def test_founder_admitted_attempt_binds_exact_inputs(tmp_path):
    result=attempt(*signed(tmp_path))
    assert result['principal']==FOUNDER and result['case']=='secrets15-nk49-readiness'
    assert result['namespace']=='secrets-engine-recovery-attempt'
@pytest.mark.parametrize('field,value',[('selector_sha256','sha256:'+'0'*64),('source_sha256','sha256:'+'0'*64),
    ('issuer','https://other'),('case','Bad Case'),('schema','other'),('not_after','2026-10-22T12:00:01Z'),
    ('not_before','2026-10-21T12:00:01Z'),('not_after','2026-10-21T12:00:00Z')])
def test_attempt_binding_mismatch_refused(tmp_path,field,value):
    with pytest.raises((ValueError,DecisionError)):attempt(*signed(tmp_path,{**ADMISSION,field:value}))
def test_attempt_signature_negatives_refused(tmp_path):
    data,sig=signed(tmp_path)
    for bad in [(data.replace(b'secrets15',b'secrets16'),sig),signed(tmp_path,namespace='secrets-engine-pdp-trust'),
                signed(tmp_path,namespace='gexec-approval'),(data,b'')]:
        with pytest.raises((ValueError,DecisionError)):attempt(*bad)
def test_attempt_refuses_changed_selector(tmp_path):
    with pytest.raises((ValueError,DecisionError)):attempt(*signed(tmp_path),selector_raw=SELECTOR_RAW+b' ')
def test_cli_admission_refusal_is_suppressed(tmp_path,capsys,monkeypatch):
    sel=tmp_path/'selector.json';sel.write_bytes(SELECTOR_RAW);(tmp_path/'a.json').write_text('{}')
    monkeypatch.setattr('sys.argv',['x','--selector',str(sel),'--source-sha256',DIGEST,'--admission',str(tmp_path/'a.json')])
    assert m.main()==1 and 'refused' in capsys.readouterr().out