165 lines
9.2 KiB
Python
165 lines
9.2 KiB
Python
"""Synthetic owner admissions/effect; actual issuer client and Warden kernel.
|
|
|
|
Runs only inside the private cross-owner fixture. No production case, catalog,
|
|
owner assent, application mutation, grant, credential or execution admission.
|
|
"""
|
|
from dataclasses import asdict
|
|
import hashlib
|
|
import json
|
|
from pathlib import Path
|
|
import subprocess
|
|
import sys
|
|
import uuid
|
|
|
|
import yaml
|
|
from warden.case_receipt_state import Outcome, Refused
|
|
from warden.case_run import Adapter, Descriptor, Executor, Journal, OwnerResult, digest
|
|
|
|
cfg = json.loads(Path(sys.argv[1]).read_text())
|
|
phase = sys.argv[2]
|
|
base = Path(cfg['directory'])
|
|
assert base.is_relative_to('/tmp')
|
|
base.mkdir(exist_ok=True, mode=0o700)
|
|
target = base / 'fixture-target.json'
|
|
records_file = base / 'fixture-owner-records.json'
|
|
records = json.loads(records_file.read_text()) if records_file.exists() else {'checked': {}, 'closed': {}, 'calls': []}
|
|
checks = []
|
|
|
|
def check(name, condition):
|
|
checks.append({'check': name, 'passed': bool(condition)})
|
|
if not condition:
|
|
raise AssertionError(name)
|
|
|
|
def save():
|
|
records_file.write_text(json.dumps(records))
|
|
records_file.chmod(0o600)
|
|
|
|
def cli(mode, reference):
|
|
entry = cfg['sessions'][reference]
|
|
output = base / ('observation-' + uuid.uuid4().hex + '.json')
|
|
command = [cfg['binary'], mode, '--admission-file', cfg['admission'],
|
|
'--token-file', entry['file'], '--token-sha256', entry['digest'], '--out', str(output)]
|
|
if mode == 'check-current-token':
|
|
command += ['--subject', cfg['subject'], '--tenant', cfg['tenant'], '--principal-type', 'service']
|
|
result = subprocess.run(command, capture_output=True, timeout=20)
|
|
if result.returncode:
|
|
check('failed_client_removes_observation', not output.exists())
|
|
return None
|
|
value = json.loads(output.read_text())
|
|
check('observation_private', output.stat().st_mode & 0o777 == 0o600)
|
|
if mode == 'check-current-token':
|
|
current = value['current']
|
|
check('current_expected_identity', current['subject'] == cfg['subject'] and current['tenant'] == cfg['tenant'])
|
|
check('current_exact_token', current['token_sha256'] == entry['digest'])
|
|
# Checked timestamps change. Retain immutable identity/token facts only.
|
|
immutable = {k:v for k,v in current.items() if k != 'checked_at'}
|
|
previous = records['checked'].get(reference)
|
|
if previous is not None:
|
|
check('original_token_facts_stable', {k:v for k,v in previous.items() if k != 'checked_at'} == immutable)
|
|
records['checked'][reference] = current
|
|
if reference == 'inverse':
|
|
check('inverse_is_distinct_grant', current['jti'] != records['checked']['initial']['jti'])
|
|
save()
|
|
else:
|
|
inactive = value['inactive_for_client']
|
|
check('negative_exact_client_and_token', inactive['client_id'] == cfg['client_id'] and inactive['token_sha256'] == entry['digest'])
|
|
# Negative status cannot establish original ownership on its own.
|
|
check('prior_positive_ownership_required', reference in records['checked'])
|
|
return value
|
|
|
|
descriptor = Descriptor('warden.case.execution.v1', 'fixture-change', 'fixture-owner',
|
|
'fixture-owner/change@v1', digest(Path(__file__).read_bytes()),
|
|
'agent', 'fixture-lane', 'target_id', 'change', ('fixture-state',),
|
|
False, ('parent-session',))
|
|
descriptor_bytes = json.dumps(asdict(descriptor)).encode()
|
|
type_bytes = yaml.safe_dump({'spec': {'id': 'current-session-fixture', 'version': '1',
|
|
'status': 'acknowledged', 'parameters': [{'name':'target_id','sensitivity':'operational','case_settable':True}],
|
|
'execution_descriptors': {'fixture-change':digest(descriptor_bytes)},
|
|
'follow_through': [{'id':'fixture-change','owner':'fixture-owner','lane':'fixture-lane','warden_executes':False,'attended':False}]}}).encode()
|
|
case_bytes = yaml.safe_dump({'spec': {'id':'fixture-case','type':'current-session-fixture@1',
|
|
'status':'materialized','acceptance':'fixture-acceptance','parameters':{'target_id':'fixture-target'}}}).encode()
|
|
catalog_bytes = yaml.safe_dump({'entries':[{'id':'fixture-lane','title':'fixture only','need_keywords':['fixture'],
|
|
'owner_repo':'fixture-owner','subsystem':'fixture','warden_executes':False,
|
|
'wiki_ref':'wiki/fixture.md','canon_ref':'fixture/canon.md','reviewed':'2026-10-18',
|
|
'status':'active','risk':'high','workload_ref':{'applicability':'not-applicable','reason':'synthetic fixture'},
|
|
'delegation':{'mode':'native','intended_owner':'fixture-owner','reviewed':'2026-10-18'}}]}).encode()
|
|
accepted = {('case-acceptance',digest(case_bytes),'fixture-acceptance'),
|
|
('ops-mason-materialization',digest(case_bytes),'fixture-materialization'),
|
|
('owner-descriptor-acceptance:fixture-owner',digest(descriptor_bytes),'fixture-descriptor'),
|
|
('type-machine-seam-acceptance',digest(type_bytes),'fixture-descriptor'),
|
|
('catalog-routing-acceptance',digest(catalog_bytes),'fixture-descriptor')}
|
|
active_reference = 'initial' if phase == 'run' else 'inverse'
|
|
|
|
def verify(purpose, source, evidence):
|
|
if purpose == 'caller-session:agent:fixture-operator:fixture-lane':
|
|
return source == digest(case_bytes) and evidence in cfg['sessions'] and cli('check-current-token', evidence) is not None
|
|
if purpose.startswith('session-revocation:'):
|
|
parts = purpose.split(':')
|
|
if len(parts) != 4 or source != digest(case_bytes):
|
|
return False
|
|
_, lane, operation, attempt = parts
|
|
record = records['closed'].get(attempt)
|
|
return (lane == 'fixture-lane' and record is not None and record['operation'] == operation
|
|
and evidence == 'closed-' + attempt
|
|
and cli('check-inactive-token', record['reference']) is not None)
|
|
return (purpose,source,evidence) in accepted
|
|
|
|
def effect(invocation, *, inverse=False):
|
|
records['calls'].append(invocation.operation)
|
|
expected = b'{"fixture":"after"}' if inverse else b'{"fixture":"before"}'
|
|
check('exact_fixture_before', target.read_bytes() == expected)
|
|
target.write_bytes(b'{"fixture":"before"}' if inverse else b'{"fixture":"after"}')
|
|
check('actual_parent_cleanup', cli('revoke-token', active_reference) is not None)
|
|
records['closed'][invocation.attempt_id] = {'reference':active_reference,'operation':invocation.operation}
|
|
save()
|
|
return OwnerResult(Outcome.ROLLED_BACK if inverse else Outcome.SUCCESS,
|
|
'fixture-result-' + invocation.attempt_id, () if inverse else ('fixture-state',), True,
|
|
session_revocation_evidence='closed-' + invocation.attempt_id,
|
|
attempt_id=invocation.attempt_id, operation=invocation.operation)
|
|
|
|
adapter = Adapter(descriptor.adapter, descriptor.adapter_source, effect,
|
|
lambda inv,result: result.evidence == 'fixture-result-' + inv.attempt_id,
|
|
lambda inv,objects: effect(inv,inverse=True))
|
|
executor = Executor(Journal(base/'journal'), {adapter.reference:adapter}, verify)
|
|
kwargs = dict(principal='fixture-operator',actor='agent',lane='fixture-lane',
|
|
acceptance_evidence='fixture-acceptance',materialization_evidence='fixture-materialization',
|
|
descriptor_evidence='fixture-descriptor',session_evidence=active_reference)
|
|
sources = (case_bytes,type_bytes,descriptor_bytes,catalog_bytes)
|
|
|
|
def row_counts():
|
|
return tuple(executor.journal.db.execute(f'SELECT COUNT(*) FROM {table}').fetchone()[0] for table in ('receipts','intents'))
|
|
|
|
if phase == 'run':
|
|
target.write_bytes(b'{"fixture":"before"}')
|
|
admitted = executor.admit(*sources,**kwargs)
|
|
check('actual_guarded_run', executor.run(*admitted).last.outcome is Outcome.SUCCESS)
|
|
before = row_counts()
|
|
try:
|
|
executor.rollback(*admitted)
|
|
raise AssertionError('revoked original session reused')
|
|
except Refused:
|
|
pass
|
|
check('revoked_parent_no_journal_or_callback', row_counts() == before and records['calls'] == ['run'])
|
|
records['pins'] = asdict(admitted[1].pins)
|
|
save()
|
|
elif phase == 'inverse':
|
|
admitted = executor.admit(*sources,**kwargs,original_session_reference='initial')
|
|
check('fresh_process_original_pins_preserved', asdict(admitted[1].pins) == records['pins'])
|
|
before = row_counts()
|
|
try:
|
|
executor.run(*admitted)
|
|
raise AssertionError('consumed run replayed')
|
|
except Refused:
|
|
pass
|
|
check('readmission_no_business_replay', records['calls'] == ['run'] and row_counts() == before)
|
|
check('actual_guarded_inverse', executor.rollback(*admitted).last.outcome is Outcome.ROLLED_BACK)
|
|
check('exact_fixture_restore', target.read_bytes() == b'{"fixture":"before"}')
|
|
check('exact_callback_denominator', records['calls'] == ['run','rollback'])
|
|
else:
|
|
raise AssertionError('unknown fixture phase')
|
|
|
|
result = {'phase':phase,'checks':checks,'caller_status_and_cleanup':'actual producer CLI/HTTPS issuer',
|
|
'case_catalog_owner_acceptance_and_target':'synthetic fixture only','native_acceptance':False,
|
|
'source_kernel':'actual Warden Executor/Journal','original_journal_pins':records.get('pins')}
|
|
(base/(phase+'-result.json')).write_text(json.dumps(result))
|
|
print(json.dumps({'phase':phase,'checks':len(checks),'passed':True,'native_acceptance':False}))
|