99fd21ee37
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a140c4-354a-7823-8f0d-8f717397f786
195 lines
11 KiB
Python
195 lines
11 KiB
Python
import importlib.util,json,hashlib,shutil,sys,uuid,os
|
|
from pathlib import Path
|
|
import pytest,yaml
|
|
SOURCE=Path(__file__).resolve().parents[1]/'scripts/task_execution_plan.py'
|
|
sys.path.insert(0,str(SOURCE.parent))
|
|
import task_execution_plan as m
|
|
PLATFORM=Path(os.environ.get('PLATFORM_PLAN_TEST_SOURCE_ROOT',str(SOURCE.parents[1])))
|
|
SECRETS=Path(os.environ.get('SECRETS_ENGINE_SOURCE_ROOT',str(PLATFORM.parent/'secrets-engine')))
|
|
@pytest.fixture
|
|
def compiler(tmp_path):
|
|
repo=tmp_path/'platform';(repo/'scripts').mkdir(parents=True);(repo/'credential-change-requests').mkdir();(repo/'openbao/policies').mkdir(parents=True)
|
|
shutil.copyfile(PLATFORM/'scripts/credential-change.py',repo/'scripts/credential-change.py')
|
|
ccr=yaml.safe_load((PLATFORM/'credential-change-requests/CCR-2026-0002-issue-core-ingestion-api-key.yaml').read_text())
|
|
ccr['status']='approved';ccr['access_frontdoor'].update(readiness='approved-pending-apply',resolvable=False)
|
|
ccrfile=repo/'credential-change-requests/CCR-2026-0002-fixture.yaml';ccrfile.write_text(yaml.safe_dump(ccr))
|
|
policy=repo/ccr['openbao']['policy_file'];shutil.copyfile(PLATFORM/ccr['openbao']['policy_file'],policy)
|
|
catalogdir=tmp_path/'catalog';catalogdir.mkdir();catalog=yaml.safe_load((SECRETS/'catalog/issue-core-ingestion-api-key.yaml').read_text())
|
|
catalog['delivery_auth']['management']='existing'
|
|
catalogfile=catalogdir/'issue-core-ingestion-api-key.yaml';catalogfile.write_text(yaml.safe_dump(catalog))
|
|
paths=[SOURCE,repo/'scripts/credential-change.py',ccrfile,policy,catalogfile,*[SECRETS/'src/secrets_engine'/name for name in ['catalog.py','__init__.py','errors.py','redact.py']]]
|
|
pins=tuple((str(p),m.tagged(p.read_bytes())) for p in paths)
|
|
return m.Compiler(m.OwnerConfig(repo,SECRETS,ccrfile.name,catalogfile.name,catalogdir,pins))
|
|
|
|
def request(c):
|
|
value=c.proposal();value.pop('catalog_target');value.update(operation_id=str(uuid.uuid4()),request_digest='sha256:'+'a'*64);return value
|
|
|
|
def test_compiles_real_owner_generated_bodies_and_fixed_six_calls(compiler):
|
|
req=request(compiler);plan=compiler.compile(req)
|
|
calls=[compiler.before_call(plan,i) for i in range(6)]
|
|
assert [x.method for x in calls]==['GET','PUT','GET','GET','PUT','GET']
|
|
assert len({x.path for x in calls})==2
|
|
policy=json.loads(calls[1].payload)['policy'];assert 'capabilities = ["read"]' in policy
|
|
role=json.loads(calls[4].payload);assert role['policies']=='workload-kv-read-issue-core-runtime'
|
|
assert role['bound_service_account_namespaces']==['external-secrets']
|
|
assert len(plan.content_digest)==64 and not plan.content_digest.startswith('sha256:')
|
|
assert all(x.payload_sha256.startswith('sha256:') and len(x.call_digest)==64 for x in calls)
|
|
|
|
@pytest.mark.parametrize('field,value',[('action','apply'),('kind','service-jwt-login'),('catalog_id','another'),('ccr_sha256','sha256:'+'b'*64),('catalog_sha256','sha256:'+'b'*64),('request_digest','a'*64)])
|
|
def test_altered_bindings_refuse(compiler,field,value):
|
|
req=request(compiler);req[field]=value
|
|
with pytest.raises(m.Refused):compiler.compile(req)
|
|
|
|
@pytest.mark.parametrize('ordinal',[False,0.0,'0',-1,6])
|
|
def test_fixed_ordinals_refuse(compiler,ordinal):
|
|
plan=compiler.compile(request(compiler))
|
|
with pytest.raises(m.Refused):compiler.before_call(plan,ordinal)
|
|
|
|
@pytest.mark.parametrize('field,value',[('method','DELETE'),('path','platform/data/workloads/keycape/x'),('payload_sha256','sha256:'+'c'*64)])
|
|
def test_caller_operation_cannot_change_target_body_or_method(compiler,field,value):
|
|
req=request(compiler);req['operations'][1][field]=value
|
|
with pytest.raises(m.Refused):compiler.compile(req)
|
|
|
|
@pytest.mark.parametrize('which',['ccr','catalog','policy','validator'])
|
|
def test_owner_source_drift_between_calls_refuses(compiler,which):
|
|
plan=compiler.compile(request(compiler));compiler.before_call(plan,0)
|
|
path={'ccr':compiler.ccr,'catalog':compiler.catalog,'policy':next(Path(p) for p in compiler.pins if p.endswith('.hcl')),'validator':compiler.validator}[which]
|
|
path.write_bytes(path.read_bytes()+b'\n# drift\n')
|
|
with pytest.raises(m.Refused):compiler.before_call(plan,1)
|
|
|
|
@pytest.mark.parametrize('change',['extra','budget','request'])
|
|
def test_direct_plan_constructor_cannot_bypass_recompilation(compiler,change):
|
|
plan=compiler.compile(request(compiler));data=plan.metadata()
|
|
if change=='extra':data['caller_hcl']='path "*" {}'
|
|
elif change=='budget':data['task_token_max_uses']=999
|
|
else:data['request']['operations'][3]['payload_sha256']='sha256:'+'d'*64
|
|
with pytest.raises(m.Refused):compiler.before_call(m.Plan(m.encoded(data)),3)
|
|
|
|
|
|
def repin(c,path):
|
|
pins=dict(c.config.source_pins);pins[str(path)]=m.tagged(path.read_bytes())
|
|
return m.Compiler(m.OwnerConfig(c.config.platform_repo,c.config.secrets_repo,c.config.ccr_filename,c.config.catalog_filename,c.config.catalog_directory,tuple(pins.items())))
|
|
|
|
@pytest.mark.parametrize('status',['in_flight','proposed','applied','verified','active','cancelled'])
|
|
def test_even_re_pinned_nonapproved_owner_record_refuses(compiler,status):
|
|
ccr=yaml.safe_load(compiler.ccr.read_text());ccr['status']=status;compiler.ccr.write_text(yaml.safe_dump(ccr))
|
|
with pytest.raises(m.Refused):repin(compiler,compiler.ccr).proposal()
|
|
|
|
|
|
def test_source_role_policy_escalation_refuses_even_when_re_pinned(compiler):
|
|
ccr=yaml.safe_load(compiler.ccr.read_text());ccr['openbao']['auth']['policies']=['root'];compiler.ccr.write_text(yaml.safe_dump(ccr))
|
|
with pytest.raises(m.Refused):repin(compiler,compiler.ccr).proposal()
|
|
|
|
|
|
def test_catalog_hash_agrees_with_actual_consumer_unicode_canonicalization(compiler):
|
|
catalog=yaml.safe_load(compiler.catalog.read_text());catalog['consumers'][0]['purpose']='café';compiler.catalog.write_text(yaml.safe_dump(catalog))
|
|
compiler=repin(compiler,compiler.catalog)
|
|
from secrets_engine.execution_contract import catalog_target_digest
|
|
proposed=compiler.proposal();assert proposed['catalog_sha256']==catalog_target_digest(proposed['catalog_target'])
|
|
|
|
|
|
def test_caller_file_path_and_policy_fields_are_refused(compiler):
|
|
req=request(compiler);req['file']='/etc/passwd';req['policy']='path "*" {}'
|
|
with pytest.raises(m.Refused):compiler.compile(req)
|
|
|
|
|
|
def test_actual_inflight_ccr42_is_not_an_admitted_target():
|
|
actual=yaml.safe_load((PLATFORM/'credential-change-requests/CCR-2026-0042-forgejo-netkingdom-oidc-consumer-reader.yaml').read_text())
|
|
assert actual['status']=='in_flight' and 'auth' not in actual['openbao']
|
|
|
|
@pytest.mark.parametrize('kind',['symlink','hardlink','writable'])
|
|
def test_owner_policy_file_indirection_or_permissions_refuse(compiler,tmp_path,kind):
|
|
req=request(compiler)
|
|
path=next(Path(p) for p in compiler.pins if p.endswith('.hcl'))
|
|
if kind=='symlink':
|
|
copy=tmp_path/'outside-policy';copy.write_bytes(path.read_bytes());path.unlink();path.symlink_to(copy)
|
|
elif kind=='hardlink':
|
|
import os
|
|
os.link(path,tmp_path/'policy-link')
|
|
else:path.chmod(0o666)
|
|
with pytest.raises(m.Refused):compiler.compile(req)
|
|
|
|
|
|
def test_re_pinned_policy_body_escalation_refuses(compiler):
|
|
path=next(Path(p) for p in compiler.pins if p.endswith('.hcl'))
|
|
path.write_text('path "*" { capabilities = ["sudo", "read", "update"] }\n')
|
|
with pytest.raises(m.Refused):repin(compiler,path).proposal()
|
|
|
|
|
|
def test_duplicate_owner_yaml_key_refuses_even_when_re_pinned(compiler):
|
|
compiler.ccr.write_bytes(compiler.ccr.read_bytes()+b'\nstatus: approved\n')
|
|
with pytest.raises(m.Refused):repin(compiler,compiler.ccr).proposal()
|
|
|
|
|
|
def test_unconfirmed_claim_binding_refuses_even_when_re_pinned(compiler):
|
|
ccr=yaml.safe_load(compiler.ccr.read_text());ccr['openbao']['auth']['bound_claims_confirmed']=False
|
|
compiler.ccr.write_text(yaml.safe_dump(ccr))
|
|
with pytest.raises(m.Refused):repin(compiler,compiler.ccr).proposal()
|
|
|
|
|
|
def test_catalogue_engine_managed_approle_is_not_silently_generic_apply(compiler):
|
|
data=yaml.safe_load(compiler.catalog.read_text());data['delivery_auth']['management']='engine'
|
|
compiler.catalog.write_text(yaml.safe_dump(data))
|
|
with pytest.raises(m.Refused):repin(compiler,compiler.catalog).proposal()
|
|
|
|
|
|
def test_oidc_role_body_reuses_owner_auth_payload_and_existing_mount(compiler):
|
|
ccr=yaml.safe_load(compiler.ccr.read_text())
|
|
source=yaml.safe_load((PLATFORM/'credential-change-requests/CCR-2026-0034.yaml').read_text())
|
|
ccr['openbao']['auth']=source['openbao']['auth'];ccr['openbao']['auth']['role']='synthetic-operator-workload-kv-read'
|
|
ccr['openbao']['auth']['policies']=[ccr['openbao']['policy_name']]
|
|
compiler.ccr.write_text(yaml.safe_dump(ccr));compiler=repin(compiler,compiler.ccr)
|
|
plan=compiler.compile(request(compiler));call=compiler.before_call(plan,4)
|
|
role=json.loads(call.payload)
|
|
assert call.path=='auth/netkingdom/role/synthetic-operator-workload-kv-read'
|
|
assert role['role_type']=='oidc' and role['policies']=='workload-kv-read-issue-core-runtime'
|
|
assert role['bound_claims']==source['openbao']['auth']['bound_claims']
|
|
|
|
|
|
def test_cached_catalogue_code_cannot_replace_verified_source(compiler, monkeypatch):
|
|
import secrets_engine.catalog as catalogue
|
|
expected = compiler.proposal()
|
|
|
|
def stale_validator(*args, **kwargs):
|
|
raise AssertionError('cached code differs from the pinned source')
|
|
|
|
monkeypatch.setattr(catalogue, 'validate_entry', stale_validator)
|
|
assert compiler.proposal() == expected
|
|
|
|
|
|
def test_cached_catalogue_dependency_cannot_replace_verified_source(compiler, monkeypatch):
|
|
import secrets_engine.catalog as catalogue
|
|
expected = compiler.proposal()
|
|
|
|
def stale_redactor(*args, **kwargs):
|
|
raise AssertionError('cached dependency differs from the pinned source')
|
|
|
|
monkeypatch.setattr(catalogue, 'looks_secret', stale_redactor)
|
|
assert compiler.proposal() == expected
|
|
|
|
|
|
def test_cached_validator_cannot_admit_invalid_pinned_catalogue(compiler, monkeypatch):
|
|
import secrets_engine.catalog as catalogue
|
|
accepted = catalogue.validate_entry(yaml.safe_load(compiler.catalog.read_text()), source='fixture')
|
|
data = yaml.safe_load(compiler.catalog.read_text())
|
|
del data['rotation']
|
|
compiler.catalog.write_text(yaml.safe_dump(data))
|
|
compiler = repin(compiler, compiler.catalog)
|
|
monkeypatch.setattr(catalogue, 'validate_entry', lambda *a, **kw: accepted)
|
|
with pytest.raises(m.Refused):
|
|
compiler.proposal()
|
|
|
|
|
|
@pytest.mark.parametrize('invalid', [False, True])
|
|
def test_snapshot_modules_removed_without_replacing_ambient_modules(compiler, invalid):
|
|
import secrets_engine.catalog
|
|
before = {k: v for k, v in sys.modules.items() if k.startswith('secrets_engine')}
|
|
if invalid:
|
|
compiler.catalog.write_text('id: incomplete\n')
|
|
compiler = repin(compiler, compiler.catalog)
|
|
with pytest.raises(m.Refused):
|
|
compiler.proposal()
|
|
else:
|
|
compiler.proposal()
|
|
assert {k: v for k, v in sys.modules.items() if k.startswith('secrets_engine')} == before
|
|
assert not any(k.startswith('_platform_catalogue_') for k in sys.modules)
|