feat: journal strict task consumption before native fixture effects

Assistant: codex
Assistant-Model: gpt-6-astra
Assistant-Session: 01a140c4-354a-7823-8f0d-8f717397f786

tests/task_execution_source_servers.py

@@ -0,0 +1,206 @@
+"""Real Approval/Access servers; every admission/key/identity is synthetic local.
+
+The positive fixture policy is separately named and exact-descriptor scoped. It
+never edits either existing unadmitted production task policy or their false gate.
+"""
+import base64
+from contextlib import contextmanager
+from dataclasses import replace
+from datetime import datetime,timedelta,timezone
+import hashlib
+from http.client import HTTPConnection
+from http.server import BaseHTTPRequestHandler,ThreadingHTTPServer
+import json
+import os
+from pathlib import Path
+import shutil
+import socket
+import subprocess
+import threading
+import time
+from types import SimpleNamespace
+from wsgiref.simple_server import make_server,WSGIRequestHandler
+
+from cryptography import x509
+from cryptography.hazmat.primitives import serialization,hashes
+from cryptography.hazmat.primitives.asymmetric import ed25519,rsa
+from cryptography.x509.oid import NameOID
+
+ACCESS=Path(os.environ.get('ACCESS_ENGINE_SOURCE_ROOT','/home/worsch/access-engine'))
+APPROVAL=Path(os.environ.get('APPROVAL_ENGINE_SOURCE_ROOT','/home/worsch/approval-engine'))
+SECRETS=Path(os.environ.get('SECRETS_ENGINE_SOURCE_ROOT','/home/worsch/secrets-engine'))
+FIXTURE_PACKAGE='platform.disposable-task-seam.fixture'
+FIXTURE_VERSION='v1'
+PRINCIPAL='system:serviceaccount:secrets-engine:secrets-engine'
+
+
+def private(path,text):path.write_text(text);path.chmod(0o600);return path
+
+def free_port():
+    with socket.socket() as sock:sock.bind(('127.0.0.1',0));return sock.getsockname()[1]
+
+
+def http(origin,method,path,body=None,token=None):
+    from urllib.parse import urlsplit
+    parsed=urlsplit(origin);conn=HTTPConnection(parsed.hostname,parsed.port,timeout=4)
+    try:
+        headers={'Content-Type':'application/json'}
+        if token is not None:headers['Authorization']='Bearer '+token
+        conn.request(method,path,body=json.dumps(body) if body is not None else None,headers=headers)
+        response=conn.getresponse();raw=response.read(1024*1024)
+        return response.status,json.loads(raw) if raw else {}
+    finally:conn.close()
+
+
+class QuietWSGI(WSGIRequestHandler):
+    def log_message(self,*args):pass
+
+
+@contextmanager
+def approval_server(directory):
+    from approval_engine.api import App
+    from approval_engine.auth import Identity,StaticTokenAuthenticator
+    from approval_engine.store import Engine
+    service='synthetic-source-service';human='synthetic-source-human';no_read='synthetic-source-no-read'
+    def identity(subject,kind,scopes):
+        return Identity(subject,'https://synthetic.invalid',('approval-engine',),kind,'tenant:platform',
+                        frozenset(),frozenset(scopes),{'level':'aal2','methods':['synthetic']},
+                        'fixture:synthetic-identity',tenant_source='directory' if kind=='human' else 'registration')
+    identities={service:identity('secrets-engine','service',['approval:create','approval:read','approval:consume']),
+                human:identity('synthetic-founder','human',['approval:approve','approval:read']),
+                no_read:identity('platform-fixture-confirmation','service',['approval:consume'])}
+    engine=Engine(str(directory/'approval.sqlite'));app=App(engine,StaticTokenAuthenticator(identities))
+    calls=[]
+    def logged(environ,start_response):
+        calls.append((environ['REQUEST_METHOD'],environ['PATH_INFO']))
+        return app(environ,start_response)
+    server=make_server('127.0.0.1',0,logged,handler_class=QuietWSGI)
+    thread=threading.Thread(target=server.serve_forever,daemon=True);thread.start()
+    try:
+        yield SimpleNamespace(origin='http://127.0.0.1:'+str(server.server_port),engine=engine,
+                              service_token=service,human_token=human,no_read_token=no_read,calls=calls)
+    finally:server.shutdown();server.server_close();thread.join(5);engine.close()
+
+
+@contextmanager
+def reviewer_server(directory):
+    # Actual Access TokenReview client consumes this independently admitted
+    # synthetic reviewer. This is not a Kubernetes authentication proof.
+    exp=int(time.time())+900
+    payload=base64.urlsafe_b64encode(json.dumps({'exp':exp}).encode()).decode().rstrip('=')
+    token='synthetic.'+payload+'.synthetic';reviewer='synthetic-reviewer';calls=[]
+    class Handler(BaseHTTPRequestHandler):
+        def log_message(self,*args):pass
+        def do_POST(self):
+            size=int(self.headers.get('Content-Length','0'));data=json.loads(self.rfile.read(size))
+            valid=(self.path=='/apis/authentication.k8s.io/v1/tokenreviews'
+                   and self.headers.get('Authorization')=='Bearer '+reviewer
+                   and data.get('spec',{}).get('token')==token
+                   and 'flex-auth' in data.get('spec',{}).get('audiences',[]))
+            calls.append(valid)
+            result={'apiVersion':'authentication.k8s.io/v1','kind':'TokenReview',
+                    'status':{'authenticated':valid,'audiences':['flex-auth'],
+                              'user':{'username':PRINCIPAL}}}
+            raw=json.dumps(result).encode();self.send_response(201)
+            self.send_header('Content-Type','application/json');self.send_header('Content-Length',str(len(raw)));self.end_headers();self.wfile.write(raw)
+    server=ThreadingHTTPServer(('127.0.0.1',0),Handler);thread=threading.Thread(target=server.serve_forever,daemon=True);thread.start()
+    # The actual reviewer constructor requires a valid CA even for its explicitly
+    # synthetic HTTP endpoint; this CA never trusts an estate endpoint.
+    key=rsa.generate_private_key(public_exponent=65537,key_size=2048)
+    name=x509.Name([x509.NameAttribute(NameOID.COMMON_NAME,'synthetic-local-reviewer')])
+    now=datetime.now(timezone.utc)
+    cert=x509.CertificateBuilder().subject_name(name).issuer_name(name).public_key(key.public_key()).serial_number(x509.random_serial_number()).not_valid_before(now-timedelta(minutes=1)).not_valid_after(now+timedelta(days=1)).sign(key,hashes.SHA256())
+    ca=directory/'fixture-ca.pem';ca.write_bytes(cert.public_bytes(serialization.Encoding.PEM))
+    reviewer_file=private(directory/'reviewer.token',reviewer)
+    try:yield SimpleNamespace(origin='http://127.0.0.1:'+str(server.server_port),caller_token=token,reviewer_file=reviewer_file,ca=ca,calls=calls)
+    finally:server.shutdown();server.server_close();thread.join(5)
+
+
+def fixture_policy(directory,descriptor,request):
+    policy=directory/'fixture-policy.md'
+    fixture=directory/'fixture-cases.json'
+    fixture.write_text(json.dumps([{'id':'synthetic-claim-free-deny','request':request,
+                                   'expect':{'effect':'deny','reason':'fixture_unapproved'}}]))
+    document='''---
+id: platform.disposable-task-seam.fixture
+name: Disposable exact task seam fixture
+namespace: secrets-engine:secret-catalog-lane
+version: v1
+status: proposed
+package: fixture.platform.task_seam
+allow_ttl: 5m
+actions: [execute-ccr]
+owner: team:synthetic-fixture
+fixtures: [fixture-cases.json]
+caring: {profile: caring-0.4.0-rc2, enforce: false}
+activation: {mode: local}
+---
+Synthetic local policy only. No production task policy is edited or enabled.
+
+```rego
+default decision := {"effect":"deny","reason":"fixture_unapproved"}
+decision := {"effect":"allow","reason":"synthetic_exact_task_approved"} {
+  input.action == "execute-ccr"
+  input.tenant == "tenant:platform"
+  input.subject.id == "secrets-engine"
+  input.subject.type == "service"
+  input.resource.system == "secrets-engine"
+  input.context.human_control == true
+  input.context.approval.valid_now == true
+  input.context.approval.consumed == false
+  input.context.approval.binding.human_control == true
+  input.context.execution_contract == DESCRIPTOR
+}
+```
+''' + '''
+```rego test
+import data.fixture.platform.task_seam
+test_missing_approval_denies { task_seam.decision.effect == "deny" with input as {} }
+```
+'''
+    document=document.replace('DESCRIPTOR',json.dumps(descriptor,separators=(',',':')))
+    policy.write_text(document)
+    return policy
+
+
+@contextmanager
+def access_server(directory,descriptor,request,reviewer,*,actual_policy=False):
+    binary=Path(os.environ.get('PLATFORM_ACCESS_FIXTURE_BINARY','/tmp/netkingdom-platform-wave4-access'))
+    verifier=Path(os.environ.get('PLATFORM_PDP_FIXTURE_VERIFIER','/tmp/netkingdom-platform-wave4-verify'))
+    if not binary.is_file() or not verifier.is_file():raise RuntimeError('build source fixture binaries first')
+    registry=directory/'registry.json';registry.write_bytes((ACCESS/'examples/secrets-engine/registry_snapshot.json').read_bytes())
+    policy=ACCESS/'examples/secrets-engine-service-task-execution/policy_package.md' if actual_policy else fixture_policy(directory,descriptor,request)
+    requestfile=directory/'claim-free-request.json';requestfile.write_text(json.dumps(request))
+    # Trusted owner-local source CLI derives the real package pin independently
+    # from the network responder. Its DENY is used only as digest provenance.
+    offline=subprocess.run([str(binary),'check','--registry',str(registry),'--policy',str(policy),'--request',str(requestfile)],capture_output=True,text=True,timeout=15)
+    if offline.returncode!=0:raise RuntimeError('source policy fixture invalid: '+offline.stderr[:300])
+    produced=json.loads(offline.stdout);provenance=produced['provenance']
+    key=ed25519.Ed25519PrivateKey.generate()
+    seed=key.private_bytes(serialization.Encoding.Raw,serialization.PrivateFormat.Raw,serialization.NoEncryption())
+    public=key.public_key().public_bytes(serialization.Encoding.Raw,serialization.PublicFormat.Raw)
+    b64=lambda value:base64.urlsafe_b64encode(value).decode().rstrip('=')
+    seedfile=private(directory/'signing.seed',b64(seed))
+    trust={name:provenance[name] for name in ['evaluator','mode','policy_package','policy_version','policy_package_digest']}
+    trust.update(schema='secrets-engine.pdp-trust.v1',keys={'synthetic-owner-key':b64(public)})
+    trustfile=directory/'pdp-trust.json';trustfile.write_text(json.dumps(trust))
+    port=free_port();origin='http://127.0.0.1:'+str(port)
+    args=[str(binary),'serve','--addr','127.0.0.1:'+str(port),'--registry',str(registry),'--policy',str(policy),
+          '--envelope-signing-key-file',str(seedfile),'--envelope-signing-kid','synthetic-owner-key',
+          '--caller-auth-mode','enforce','--caller-audience','flex-auth','--caller-kubernetes-url',reviewer.origin,
+          '--caller-reviewer-token-file',str(reviewer.reviewer_file),'--caller-ca-file',str(reviewer.ca),
+          '--caller-binding','secrets-engine='+PRINCIPAL]
+    process=subprocess.Popen(args,stdout=subprocess.DEVNULL,stderr=subprocess.DEVNULL)
+    for _ in range(100):
+        if process.poll() is not None:raise RuntimeError('source Access server exited')
+        try:
+            with socket.create_connection(('127.0.0.1',port),timeout=.1):break
+        except OSError:time.sleep(.05)
+    else:process.kill();process.wait();raise RuntimeError('source Access server failed to start')
+    try:yield SimpleNamespace(origin=origin,trustfile=trustfile,verifier=verifier,
+                              package=provenance['policy_package'],version=provenance['policy_version'],
+                              process=process,offline_envelope=produced)
+    finally:
+        process.terminate()
+        try:process.wait(timeout=5)
+        except subprocess.TimeoutExpired:process.kill();process.wait()

tests/test_task_execution_mediation_journal.py

@@ -0,0 +1,32 @@
+"""Durable CAS/confirmation call prefix cannot overlap native call ordinals."""
+import os
+from pathlib import Path
+import sys
+import uuid
+import pytest
+sys.path.insert(0,str(Path(__file__).resolve().parents[1]/'scripts'))
+from task_execution_journal import Binding,Cleanup,Journal,Refused
+
+
+def test_next_call_continues_auth_prefix_without_replay_or_resume(tmp_path):
+    os.chmod(tmp_path,0o700);journal=Journal(tmp_path)
+    binding=Binding('fixture-approval','sha256:'+'a'*64,str(uuid.uuid4()),'b'*64)
+    try:
+        with pytest.raises(Refused):journal.next_call_ordinal(binding)
+        with journal.locked():
+            journal.reserve(binding,(Cleanup('task-token-revoke','c'*64),))
+            assert journal.next_call_ordinal(binding)==0
+            for ordinal in [0,1]:
+                journal.before_call(binding,ordinal,'d'*64)
+                with pytest.raises(Refused):journal.next_call_ordinal(binding)
+                journal.after_call(binding,ordinal,'d'*64,'e'*64)
+            assert journal.next_call_ordinal(binding)==2
+            journal.transition(binding,'reserved','consume-confirmed')
+            journal.before_call(binding,2,'f'*64)
+            with pytest.raises(Refused):journal.before_call(binding,0,'f'*64)
+            journal.after_call(binding,2,'f'*64,'e'*64)
+            journal.before_cleanup(binding,0,Cleanup('task-token-revoke','c'*64))
+            with pytest.raises(Refused):journal.next_call_ordinal(binding)
+        with journal.locked():
+            with pytest.raises(Refused):journal.next_call_ordinal(binding)
+    finally:journal.close()

tests/test_task_execution_mediator.py

@@ -0,0 +1,160 @@
+"""Whole source-server/native fixture proof, never production owner acceptance."""
+from contextlib import contextmanager
+from dataclasses import replace
+import importlib
+import json
+from pathlib import Path
+import sys
+from types import SimpleNamespace
+import uuid
+
+import pytest
+sys.path.insert(0,str(Path(__file__).resolve().parents[1]/'scripts'))
+from test_task_execution_native_fixture import owner,bao_dev,compiler
+from task_execution_mediator import ConfirmationConfig,run_native_fixture_task
+from task_execution_journal import Refused
+from task_execution_source_servers import approval_server,reviewer_server,access_server,http,private
+from secrets_engine.approval_consume import _request_purpose
+from secrets_engine.approval_claim import binding_from_check_request
+from secrets_engine.authorization import build_action_request
+from secrets_engine.decision_check import check_decision
+from secrets_engine.execution_contract import ExecutionContract
+
+
+@contextmanager
+def whole_fixture(owner,tmp_path,*,actual_policy=False,confirmation_read=True):
+    native,journal,binding,call,compiler,descriptor=owner
+    snapshot={path:compiler.read(Path(path)) for path in compiler.pins}
+    entry=compiler._catalogue_entry(snapshot)
+    approval_id=str(uuid.uuid4())
+    entry=replace(entry,approval={**entry.approval,'authorization_id':approval_id})
+    request=build_action_request(entry,'execute-ccr',subject_id='secrets-engine',subject_type='service',
+                                 purpose=_request_purpose(entry),execution_contract=ExecutionContract.parse(descriptor))
+    with reviewer_server(tmp_path) as reviewer, access_server(tmp_path,descriptor,request,reviewer,actual_policy=actual_policy) as access, approval_server(tmp_path) as approval:
+        issue=access.offline_envelope
+        body={'id':approval_id,'binding':binding_from_check_request(issue['binding']),
+              'pdp_digest':issue['binding']['request_digest'],'pdp_path':True,'human_control':True,
+              'validity':{'not_before':'2026-01-01T00:00:00+00:00',
+                          'expires_at':reviewer_expiry()}}
+        status,obj=http(approval.origin,'POST','/v1/approvals',body,approval.service_token)
+        assert status==201
+        assert http(approval.origin,'POST','/v1/approvals/'+approval_id+'/entries',{},approval.human_token)[0]==200
+        approval_token=private(tmp_path/'approval.token',approval.service_token)
+        pdp_token=private(tmp_path/'pdp.token',reviewer.caller_token)
+        confirmation_token=private(tmp_path/'confirmation.token',approval.service_token if confirmation_read else approval.no_read_token)
+        cfg=SimpleNamespace(approval_url=approval.origin,approval_token_file=approval_token,
+                            approval_client_secret_file=None,pdp_url=access.origin,pdp_token_file=pdp_token,
+                            pdp_trust_file=access.trustfile,pdp_verifier=access.verifier,
+                            authorization_subject_id='secrets-engine',authorization_subject_type='service',
+                            authorization_policy_package=access.package,authorization_policy_version=access.version,
+                            clock_trust_file=None)
+        yield SimpleNamespace(native=native,journal=journal,call=call,compiler=compiler,
+                              descriptor=descriptor,entry=entry,cfg=cfg,approval=approval,
+                              confirmation=ConfirmationConfig(approval.origin,confirmation_token),
+                              approval_id=approval_id,request=request,reviewer=reviewer)
+
+
+def reviewer_expiry():
+    from datetime import datetime,timedelta,timezone
+    return (datetime.now(timezone.utc)+timedelta(minutes=10)).isoformat()
+
+
+def invoke(fixture):
+    return run_native_fixture_task(compiler=fixture.compiler,journal=fixture.journal,
+                plan_request=fixture.native.meta['request'],owner_descriptor=fixture.native.admission.descriptor,
+                secrets_cfg=fixture.cfg,entry=fixture.entry,confirmation=fixture.confirmation,
+                native_admission=fixture.native.admission)
+
+
+def test_real_source_servers_signed_fixture_allow_and_actual_task_policy_deny(owner,tmp_path):
+    # These two Checks test real source servers and independently installed keys;
+    # they are not a fabricated DecisionEnvelope or activation of a real policy.
+    for actual_policy in [False,True]:
+        directory=tmp_path/('actual-policy' if actual_policy else 'fixture-policy');directory.mkdir()
+        with whole_fixture(owner,directory,actual_policy=actual_policy) as fixture:
+            status,claim=http(fixture.approval.origin,'GET','/v1/approvals/'+fixture.approval_id+'/claim',token=fixture.approval.service_token)
+            assert status==200
+            request=json.loads(json.dumps(fixture.request));request['context']['approval']=claim
+            envelope=check_decision(base_url=fixture.cfg.pdp_url,token_file=fixture.cfg.pdp_token_file,request=request,
+                                    trust_file=fixture.cfg.pdp_trust_file,verifier=fixture.cfg.pdp_verifier)
+            assert envelope['effect']==('deny' if actual_policy else 'allow')
+            assert envelope['provenance']['caller']['mode']=='enforce'
+            assert all(fixture.reviewer.calls)
+            assert fixture.approval.engine.get(fixture.approval_id).status=='approved'
+            assert fixture.call('GET','sys/policies/acl/'+fixture.native.policy_name)[0]==404
+
+
+def test_missing_strict_dependency_never_reads_credentials_or_reserves(monkeypatch):
+    original=importlib.import_module
+    def missing(name,*args,**kwargs):
+        if name=='secrets_engine.task_execution_handoff':raise ImportError()
+        return original(name,*args,**kwargs)
+    monkeypatch.setattr(importlib,'import_module',missing)
+    with pytest.raises(Refused,match='dependency unavailable'):
+        run_native_fixture_task(compiler=None,journal=None,plan_request={},owner_descriptor=b'',
+                                secrets_cfg=None,entry=None,confirmation=None,native_admission=None)
+
+
+def test_whole_strict_consume_confirmation_precedes_real_native_effect(owner,tmp_path):
+    with whole_fixture(owner,tmp_path) as fixture:
+        result=invoke(fixture)
+        assert result['native_calls_completed']==6 and result['cleanup_confirmed']
+        assert result['native_apply']=='unverified' # total owner baseline/readback acceptance remains separate
+        obj=fixture.approval.engine.get(fixture.approval_id)
+        assert obj.status=='consumed'
+        assert obj.consumed_digest==result['request_digest']
+        calls=fixture.journal.db.execute('SELECT ordinal,outcome FROM calls ORDER BY ordinal').fetchall()
+        assert calls[:2]==[(0,'complete'),(1,'complete')]
+        assert len(calls)==13 # CAS + GET + five native grant calls + six metadata calls
+        assert fixture.approval.calls.count(('POST','/v1/approvals/'+fixture.approval_id+'/consume'))==1
+        assert fixture.approval.calls.count(('GET','/v1/approvals/'+fixture.approval_id))==1
+        assert fixture.call('GET','sys/policies/acl/'+fixture.native.policy_name)[0]==404
+        raw=(fixture.journal.directory/'journal.sqlite').read_bytes()
+        for token in [fixture.native.admission.owner_token,fixture.native.admission.identity_session.token,
+                      fixture.approval.service_token,fixture.reviewer.caller_token]:
+            assert token.encode() not in raw
+        with pytest.raises(Exception):invoke(fixture)
+        assert fixture.approval.calls.count(('POST','/v1/approvals/'+fixture.approval_id+'/consume'))==1
+
+
+def test_actual_confirmation_read_scope_denial_keeps_native_untouched(owner,tmp_path):
+    with whole_fixture(owner,tmp_path,confirmation_read=False) as fixture:
+        with pytest.raises(Refused,match='confirmation unconfirmed'):invoke(fixture)
+        assert fixture.approval.engine.get(fixture.approval_id).status=='consumed'
+        state=fixture.journal.db.execute('SELECT state FROM operations').fetchone()[0]
+        assert state=='unknown'
+        assert fixture.journal.db.execute("SELECT COUNT(*) FROM calls WHERE outcome='pending'").fetchone()[0]==1
+        assert fixture.call('GET','sys/policies/acl/'+fixture.native.policy_name)[0]==404
+        policy_path=fixture.compiler.before_call(fixture.native.plan,1).path
+        assert fixture.call('GET',policy_path)[0]==404
+        with pytest.raises(Exception):invoke(fixture)
+        assert fixture.approval.calls.count(('POST','/v1/approvals/'+fixture.approval_id+'/consume'))==1
+
+
+def test_actual_unadmitted_policy_refuses_before_cas_and_native(owner,tmp_path):
+    with whole_fixture(owner,tmp_path,actual_policy=True) as fixture:
+        with pytest.raises(Exception):invoke(fixture)
+        assert fixture.approval.engine.get(fixture.approval_id).status=='approved'
+        assert not any(method=='POST' and path.endswith('/consume') for method,path in fixture.approval.calls)
+        assert fixture.journal.db.execute('SELECT COUNT(*) FROM operations').fetchone()[0]==0
+        assert fixture.call('GET','sys/policies/acl/'+fixture.native.policy_name)[0]==404
+
+
+def test_entry_purpose_substitution_refuses_before_authorizer_or_cas(owner,tmp_path):
+    with whole_fixture(owner,tmp_path) as fixture:
+        fixture.entry=replace(fixture.entry,approval={**fixture.entry.approval,'purpose':'caller-substituted-purpose'})
+        before=list(fixture.approval.calls)
+        with pytest.raises(Refused,match='approval metadata differs'):invoke(fixture)
+        assert fixture.approval.calls==before
+        assert fixture.journal.db.execute('SELECT COUNT(*) FROM operations').fetchone()[0]==0
+        assert fixture.call('GET','sys/policies/acl/'+fixture.native.policy_name)[0]==404
+
+
+def test_catalogue_bool_integer_alias_refuses_before_authorizer(owner,tmp_path):
+    with whole_fixture(owner,tmp_path) as fixture:
+        assert fixture.entry.delivery_auth['secret_id_num_uses']==1
+        fixture.entry=replace(fixture.entry,delivery_auth={**fixture.entry.delivery_auth,'secret_id_num_uses':True})
+        before=list(fixture.approval.calls)
+        with pytest.raises(Refused,match='catalogue does not match'):invoke(fixture)
+        assert fixture.approval.calls==before
+        assert fixture.journal.db.execute('SELECT COUNT(*) FROM operations').fetchone()[0]==0