"""Actual disposable OpenBao effects; all identities/approvals are synthetic. These tests do not contact estate OpenBao or claim production owner acceptance. """ import hashlib import http.client import json import shutil import socket import subprocess import time from datetime import datetime,timedelta,timezone from pathlib import Path import sys import pytest sys.path.insert(0,str(Path(__file__).resolve().parents[1]/'scripts')) from test_task_execution_plan import compiler,request,m from task_execution_journal import Binding,Cleanup,Journal,Refused from task_execution_cleanup import run_cleanup from task_execution_native_fixture import FixtureAdmission,NativeFixtureAdapter,PrivateSession @pytest.fixture def bao_dev(): executable=shutil.which('bao') if not executable: pytest.skip('OpenBao executable unavailable') with socket.socket() as sock: sock.bind(('127.0.0.1',0));port=sock.getsockname()[1] root='synthetic-disposable-fixture-root' process=subprocess.Popen([executable,'server','-dev','-dev-no-store-token', '-dev-root-token-id='+root,'-dev-listen-address=127.0.0.1:'+str(port)], stdout=subprocess.DEVNULL,stderr=subprocess.DEVNULL) def call(method,path,body=None,token=root): conn=http.client.HTTPConnection('127.0.0.1',port,timeout=2) try: conn.request(method,'/v1/'+path,body=json.dumps(body) if body is not None else None, headers={'X-Vault-Token':token,'Content-Type':'application/json'}) response=conn.getresponse();raw=response.read() return response.status,json.loads(raw) if raw else {} finally: conn.close() for _ in range(100): if process.poll() is not None: pytest.fail('disposable backend exited') try: if call('GET','sys/health')[0]==200: break except OSError: pass time.sleep(0.05) else: process.kill();process.wait();pytest.fail('disposable backend failed to start') try: yield 'http://127.0.0.1:'+str(port),root,call,process finally: process.terminate() try: process.wait(timeout=5) except subprocess.TimeoutExpired: process.kill();process.wait() @pytest.fixture def owner(compiler,bao_dev,tmp_path): origin,root,call,process=bao_dev assert call('POST','sys/auth/kubernetes',{'type':'kubernetes'})[0] in {200,204} hcl='path "auth/token/revoke-self" { capabilities = ["update"] }' assert call('PUT','sys/policies/acl/fixture-self-only',{'policy':hcl})[0] in {200,204} status,result=call('POST','auth/token/create',{'policies':['fixture-self-only'],'no_default_policy':True,'renewable':False,'ttl':'300s','num_uses':8}) assert status==200 identity=PrivateSession(result['auth']['client_token'],result['auth']['accessor']) req=request(compiler);plan=compiler.compile(req) descriptor={'contract_version':'SECRETS-EXEC-20261011-v1','operation_id':req['operation_id'], 'operation':'metadata-apply-ccr','execution_actor':'service:secrets-engine', 'ccr_id':compiler._derive()[0]['id'],'ccr_immutable_sha256':req['ccr_sha256'], 'execution_grant_ccr_id':'FIXTURE-GRANT-NOT-PRODUCTION','grant_immutable_sha256':'sha256:'+'f'*64, 'catalog_sha256':req['catalog_sha256'], 'source_bundle':{role:{'path':'/fixture/'+role,'sha256':'sha256:'+'d'*64} for role in ['engine_dispatcher','platform_mediator','typed_ccr','policy_auth']}, 'backend_origin':origin,'cluster_uid':'disposable-fixture-only', 'operations':req['operations'],'operation_deadline':(datetime.now(timezone.utc)+timedelta(seconds=240)).isoformat(), 'backend_budget':{'ttl_seconds':120,'uses':12},'expected_before':'sha256:'+'0'*64,'expected_after':'sha256:'+'1'*64} # Exercise the actual consumer closed descriptor parser as well; source pins, # grant/expected-state digests remain deliberately synthetic owner fixtures. from secrets_engine.execution_contract import ExecutionContract descriptor=ExecutionContract.parse(descriptor).as_dict() installed=m.encoded(descriptor) plan=compiler.bind_execution(plan,descriptor,installed) binding=Binding('SYNTHETIC-APPROVAL-NOT-PRODUCTION',req['request_digest'],req['operation_id'],plan.content_digest) state=tmp_path/'private-journal';state.mkdir(mode=0o700);journal=Journal(state) admission=FixtureAdmission(installed,root,identity,process) native=NativeFixtureAdapter(compiler,plan,journal,binding,admission) try: yield native,journal,binding,call,compiler,descriptor finally: journal.close() def admit(native,journal,binding): # Synthetic postconsume transition is fixture setup, not served approval proof. journal.reserve(binding,native.cleanup) journal.transition(binding,'reserved','consume-confirmed') journal.transition(binding,'consume-confirmed','authenticating') journal.transition(binding,'authenticating','grant-issuing') def test_native_fixed_grant_metadata_denials_and_confirmed_cleanup(owner): native,journal,binding,call,compiler,descriptor=owner with journal.locked(): admit(native,journal,binding) info=native.materialize_grant() assert info['issued'] and info['ttl_ceiling_seconds']==120 and info['use_ceiling']==12 journal.transition(binding,'grant-issued','applying') baseline=native.metadata_call(0);assert baseline[0]==404 assert native.metadata_call(1)[0] in {200,204} policy=native.metadata_call(2);assert policy[0]==200 assert policy[1]['data']['policy'].rstrip()==json.loads(compiler.before_call(native.plan,1).payload)['policy'].rstrip() assert native.metadata_call(3)[0]==404 assert native.metadata_call(4)[0] in {200,204} role=native.metadata_call(5);assert role[0]==200 # Actual native role metadata correspondence for fixture's known K8s body. expected=json.loads(compiler.before_call(native.plan,4).payload) assert role[1]['data']['bound_service_account_names']==expected['bound_service_account_names'] assert role[1]['data']['bound_service_account_namespaces']==expected['bound_service_account_namespaces'] assert role[1]['data']['token_policies']==expected['policies'].split(',') # Root probes native capabilities without spending the limited task token. for path in ['sys/policies/acl/sibling','auth/kubernetes/role/sibling','secret/data/sibling','auth/token/create','sys/mounts/secret','auth/approle/role/sibling/secret-id']: status,data=call('POST','sys/capabilities',{'token':native.task.token,'paths':[path]}) assert status==200 and data['capabilities']==['deny'] task_accessor=native.task.accessor;identity_accessor=native.admission.identity_session.accessor result=run_cleanup(journal,binding,native) # Assert native absence before summary so fixture failures identify protocol semantics. for accessor in [task_accessor,identity_accessor]: status,data=call('POST','auth/token/lookup-accessor',{'accessor':accessor}) assert status==400 and data.get('errors') in (['invalid accessor'],['1 error occurred:\n\t* invalid accessor\n\n']) assert result['confirmed'] assert call('GET','sys/policies/acl/'+native.policy_name)[0]==404 assert journal.observe(binding)['state']=='applying' # native success is not full owner verification with pytest.raises(Refused): native.metadata_call(6) with pytest.raises(Refused): run_cleanup(journal,binding,native) def test_no_consume_no_native_grant_and_changed_digest_no_cleanup(owner): native,journal,binding,call,*_=owner with journal.locked(): journal.reserve(binding,native.cleanup) with pytest.raises(Refused):native.materialize_grant() assert call('GET','sys/policies/acl/'+native.policy_name)[0]==404 with pytest.raises(Refused):native.delete_grant_policy('0'*64) def test_native_cleanup_after_deadline_and_pending_business_uncertainty(owner): native,journal,binding,call,*_=owner with journal.locked(): admit(native,journal,binding);native.materialize_grant() journal.transition(binding,'grant-issued','applying') journal.before_call(binding,native._ordinal,'a'*64) journal.transition(binding,'applying','unknown') native.monotonic_deadline=time.monotonic()-1 with pytest.raises(Refused):native.metadata_call(0) assert run_cleanup(journal,binding,native)['confirmed'] assert journal.observe(binding)['state']=='unknown' assert journal.observe(binding)['external_outcome_unknown'] def test_whole_descriptor_and_cleanup_commitment_are_bound(owner): native,journal,binding,call,compiler,descriptor=owner base=compiler.compile(native.meta['request']) changed=dict(descriptor,cluster_uid='different-cluster') with pytest.raises(m.Refused):compiler.bind_execution(base,changed,native.admission.descriptor) data=native.plan.metadata();data['cleanup'][0]['call_digest']='9'*64 with pytest.raises(m.Refused):compiler.before_call(m.Plan(m.encoded(data)),0) with pytest.raises(Refused):NativeFixtureAdapter(compiler,m.Plan(m.encoded(data)),journal,binding,native.admission) def test_source_drift_after_admission_prevents_mint_but_allows_native_cleanup(owner): native,journal,binding,call,compiler,*_=owner with journal.locked(): admit(native,journal,binding) compiler.ccr.write_bytes(compiler.ccr.read_bytes()+b'\n# changed owner source\n') with pytest.raises(m.Refused):native.materialize_grant() assert call('GET','sys/policies/acl/'+native.policy_name)[0]==404 assert run_cleanup(journal,binding,native)['confirmed'] def test_actual_native_write_then_lost_response_stays_unknown_and_cleans(owner): native,journal,binding,call,compiler,*_=owner with journal.locked(): admit(native,journal,binding);native.materialize_grant() journal.transition(binding,'grant-issued','applying') assert native.metadata_call(0)[0]==404 original=native._http target=compiler.before_call(native.plan,1).path def lost_response(method,path,token,body=b''): result=original(method,path,token,body) if method=='PUT' and path==target: raise Refused('synthetic lost response after real native write') return result native._http=lost_response with pytest.raises(Refused):native.metadata_call(1) assert call('GET',target)[0]==200 # actual side effect despite missing confirmation assert journal.observe(binding)['state']=='unknown' assert journal.observe(binding)['external_outcome_unknown'] assert run_cleanup(journal,binding,native)['confirmed'] assert journal.observe(binding)['state']=='unknown' with pytest.raises(Refused):native.metadata_call(1) assert call('GET',target)[0]==200 # cleanup never rolls back/reapplies business state def test_actual_native_policy_retirement_does_not_prove_failed_token_revoke(owner): native,journal,binding,call,*_=owner with journal.locked(): admit(native,journal,binding);native.materialize_grant() original=native._http def refuse_task_revoke(method,path,token,body=b''): if method=='POST' and path=='auth/token/revoke-self' and token==native.task.token: return 503,{'errors':['synthetic cleanup transport failure']} return original(method,path,token,body) native._http=refuse_task_revoke result=run_cleanup(journal,binding,native) assert not result['confirmed'] assert [item['outcome'] for item in result['obligations']]==['unconfirmed','confirmed','confirmed'] assert call('GET','sys/policies/acl/'+native.policy_name)[0]==404 # Actual owner lookup demonstrates token still exists despite deleted ACL. status,data=call('POST','auth/token/lookup-accessor',{'accessor':native.task.accessor}) assert status==200 and data['data']['policies']==[native.policy_name] with pytest.raises(Refused):run_cleanup(journal,binding,native) # Explicit synthetic owner recovery, not adapter retry/resumption. assert call('POST','auth/token/revoke-accessor',{'accessor':native.task.accessor})[0] in {200,204} def test_nonloopback_estate_origin_is_refused_before_native_transport(owner): native,journal,binding,call,compiler,descriptor=owner changed=dict(descriptor,backend_origin='https://estate.invalid:8200') installed=m.encoded(changed) plan=compiler.bind_execution(compiler.compile(native.meta['request']),changed,installed) changed_binding=Binding(binding.approval_id,binding.request_digest,binding.operation_id,plan.content_digest) admission=FixtureAdmission(installed,native.admission.owner_token,native.admission.identity_session,native.admission.server_process) with pytest.raises(Refused):NativeFixtureAdapter(compiler,plan,journal,changed_binding,admission)