access-engine/tools/task_execution_interop_fixture.py
tegwick bc92f95e5f
All checks were successful
CI Smoke / host-smoke (push) Successful in 0s
CI Smoke / container-smoke (push) Successful in 1s
Build and Publish Container Image / build-and-push (push) Successful in 59s
Prepare closed service task execution input and policy contract
Assistant: codex
Assistant-Model: gpt-6.1-sol
Assistant-Session: 01a126e1-76f8-7643-8cf3-e93336539ade
2026-10-11 19:48:53 +02:00

29 lines
3.2 KiB
Python

"""Synthetic interoperability generator; never obtains credentials or native state."""
import hashlib, json, os, sys
from datetime import datetime, timedelta, timezone
sys.path.insert(0, os.environ.get('SECRETS_ENGINE_SOURCE', '/home/worsch/secrets-engine') + '/src')
from secrets_engine.execution_contract import ExecutionContract, catalog_target_digest
from secrets_engine.authorization import request_digest, approval_binding_digest
from secrets_engine.errors import DecisionError
h='sha256:'+'a'*64
catalog={'lane':'synthetic','stage':'prod'}
c={'contract_version':'SECRETS-EXEC-20261011-v1','operation_id':'cbfbced0-09d5-48c8-a1e7-a2ea4dc69153','operation':'metadata-apply-ccr','execution_actor':'service:secrets-engine','ccr_id':'CCR-2099-0001','execution_grant_ccr_id':'CCR-2099-0002','ccr_immutable_sha256':h,'grant_immutable_sha256':h,'catalog_sha256':catalog_target_digest(catalog),'expected_before':h,'expected_after':h,'source_bundle':{r:{'path':'/synthetic/'+r+'.py','sha256':h} for r in ['engine_dispatcher','platform_mediator','typed_ccr','policy_auth']},'backend_origin':'http://127.0.0.1:18200','cluster_uid':'synthetic-not-native','backend_budget':{'ttl_seconds':120,'uses':12},'operation_deadline':(datetime.now(timezone.utc)+timedelta(seconds=100)).isoformat(),'operations':[{'method':method,'path':path,'payload_sha256':h} for path in ['sys/policies/acl/synthetic','auth/synthetic/role/synthetic'] for method in ['GET','PUT','GET']]}
c=ExecutionContract.parse(c).as_dict()
p={'version':'proposed-task-execution-projection/v1','subject':'synthetic-human','principal_type':'human','tenant':'tenant:platform','tenant_source':'directory','mfa':True,'transport_principal':'system:serviceaccount:secrets-engine:secrets-engine','execution_contract_digest':catalog_target_digest(c),'resource_id':'catalog:synthetic','membership_source':'user-engine','tenant_state_source':'tenant-engine','member_active':True,'tenant_active':True,'not_after_ns':int((datetime.now(timezone.utc)+timedelta(seconds=60)).timestamp())*1000000000}
r={'tenant':'tenant:platform','subject':{'id':'secrets-engine','type':'service'},'action':'execute-ccr','resource':{'id':'catalog:synthetic','type':'secret-catalog-lane','system':'secrets-engine'},'context':{'execution_contract':c,'catalog_target':catalog,'human_control':True,'task_execution_projection':p,'approval':{'kind':'approval-claim','binding':{'human_control':True}}}}
# Preserve the exact earlier shape as an explicit unsupported counterexample.
# It is NOT rounded, relabeled as RFC3339 or admitted into the old human policy.
original_projection = p
try:
request_digest(r)
original_projection_refused = False
except DecisionError:
original_projection_refused = True
# Descriptor-only interoperability is independent of a missing identity contract.
del r['context']['task_execution_projection']
print(json.dumps({'request':r,'request_digest':request_digest(r),
'approval_binding_digest':approval_binding_digest(r),
'descriptor_digest':catalog_target_digest(c),
'original_nanosecond_projection':original_projection,
'original_projection_refused':original_projection_refused,
'identity_contract_changed':False}))