flex-auth/tools/exercise_sitting_review_policy.py

import json, time, copy, subprocess, tempfile
from pathlib import Path
import argparse

p = argparse.ArgumentParser()
p.add_argument('--binary', required=True)
p.add_argument('--receipt', type=Path, required=True)
args = p.parse_args()
r = Path(__file__).resolve().parents[1] / 'examples/informed-decision-sitting'
records = json.loads((r / 'records.json').read_text())
version = next(l.split(':', 1)[1].strip() for l in (r / 'policy.md').read_text().splitlines() if l.startswith('version:'))
results = []
with tempfile.TemporaryDirectory() as temp:
    request_path = Path(temp) / 'request.json'

    def check(name, request, expected):
        request_path.write_text(json.dumps(request))
        result = subprocess.run(
            [args.binary, 'check', '--registry', str(r / 'registry.json'),
             '--policy', str(r / 'policy.md'), '--request', str(request_path)],
            capture_output=True, text=True, check=True)
        d = json.loads(result.stdout)
        assert d['effect'] == expected, (name, d)
        results.append({'check': name, 'effect': d['effect']})
        return d

    for memo, record in records.items():
        request = {
            'id': 'local-regression',
            'tenant': 'tenant:platform',
            'subject': {
                'id': 'synthetic-reviewer',
                'type': 'human',
                'tenant': 'tenant:platform',
                'attributes': {
                    'groups': ['net-kingdom-admins'],
                    'roles': [],
                    'tenant_source': 'registration-supplied',
                    'principal_type_source': 'authentication-derived',
                    'assurance': {
                        'level': 'aal2',
                        'mfa': True,
                        'methods': ['pwd', 'otp'],
                        'source': 'key-cape',
                        'at': int(time.time()),
                    },
                },
            },
            'resource': {
                'id': memo,
                'type': 'decision-memo',
                'system': 'informed-decision',
                'tenant': 'tenant:platform',
            },
            'action': 'accept',
            'context': {
                'memo_version': 1,
                'approval_id': record['approval_id'],
                'approval_binding_digest': record['binding_digest'],
            },
            'policy_version': version,
        }
        label = record['label']
        for action in ['read', 'acknowledge', 'accept', 'return', 'discuss', 'decline']:
            check(label + ':' + action, request | {'action': action}, 'allow')
        for name, path, value in [
            ('wrong-group', ['subject', 'attributes', 'groups'], ['net-kingdom-users']),
            ('no-group', ['subject', 'attributes', 'groups'], []),
            ('service', ['subject', 'type'], 'service'),
            ('stale-mfa', ['subject', 'attributes', 'assurance', 'at'], int(time.time()) - 901),
            ('future-mfa', ['subject', 'attributes', 'assurance', 'at'], int(time.time()) + 300),
            ('no-mfa', ['subject', 'attributes', 'assurance', 'mfa'], False),
            ('forged-human-route', ['subject', 'attributes', 'principal_type_source'], 'registration-supplied'),
            ('wrong-tenant', ['subject', 'tenant'], 'tenant:other'),
            ('other-memo', ['resource', 'id'], 'memo:other'),
            ('omitted-c01', ['resource', 'id'], 'memo:infd-20260914-c01'),
            ('t03-memo', ['resource', 'id'], 'memo:SECRETS-WP-0010-T03-apply'),
            ('changed-version', ['context', 'memo_version'], 2),
            ('changed-approval', ['context', 'approval_id'], 'other'),
            ('changed-digest', ['context', 'approval_binding_digest'], 'sha256:' + '0' * 64),
            ('consume', ['action'], 'consume'),
        ]:
            candidate = copy.deepcopy(request)
            target = candidate
            for key in path[:-1]:
                target = target[key]
            target[path[-1]] = value
            check(label + ':' + name, candidate, 'deny')

    # --- v3 list (FLEX-WP-0032, FLEX-DEC-2026-017) ---------------------------
    # list ranges over the union of the act records and the list-only records;
    # every act stays scoped to the act records, with the 900 s window.
    list_only = json.loads((r / 'list_only_records.json').read_text())
    listable = {k: dict(v, memo_version=1) for k, v in records.items()} | list_only
    now = int(time.time())
    for memo, record in listable.items():
        request = {
            'id': 'local-regression',
            'tenant': 'tenant:platform',
            'subject': {
                'id': 'synthetic-reviewer', 'type': 'human', 'tenant': 'tenant:platform',
                'attributes': {
                    'groups': ['net-kingdom-admins'], 'roles': [],
                    'tenant_source': 'registration-supplied',
                    'principal_type_source': 'authentication-derived',
                    'assurance': {'level': 'aal2', 'mfa': True, 'methods': ['pwd', 'otp'],
                                  'source': 'key-cape', 'at': now},
                },
            },
            'resource': {'id': memo, 'type': 'decision-memo',
                         'system': 'informed-decision', 'tenant': 'tenant:platform'},
            'action': 'list',
            'context': {'memo_version': record['memo_version'],
                        'approval_id': record['approval_id'],
                        'approval_binding_digest': record['binding_digest']},
        }
        label = 'list:' + record['label']

        def variant(path, value, req=request):
            c = copy.deepcopy(req)
            t = c
            for key in path[:-1]:
                t = t[key]
            t[path[-1]] = value
            return c

        at = ['subject', 'attributes', 'assurance', 'at']
        check(label + ':fresh', request, 'allow')
        # The observed production refusal: 1744 s, allowed for list...
        check(label + ':observed-1744s', variant(at, now - 1744), 'allow')
        check(label + ':inside-12h', variant(at, now - 43200 + 60), 'allow')
        # ...but the 12 h bound is real, and read does not follow list.
        check(label + ':past-12h', variant(at, now - 43200 - 60), 'deny')
        is_act_record = memo in records
        read_stale = variant(at, now - 1744) | {'action': 'read'}
        check(label + ':read-stale-refused', read_stale, 'deny')
        if not is_act_record:
            # T03 records are listable only; no act widens to them, even fresh.
            for action in ['read', 'acknowledge', 'accept', 'return', 'discuss', 'decline']:
                check(label + ':act-' + action + '-not-widened', request | {'action': action}, 'deny')
        for name, path, value in [
            ('wrong-group', ['subject', 'attributes', 'groups'], ['net-kingdom-users']),
            ('no-group', ['subject', 'attributes', 'groups'], []),
            ('service', ['subject', 'type'], 'service'),
            ('agent', ['subject', 'type'], 'agent'),
            ('future-mfa', at, now + 300),
            ('no-mfa', ['subject', 'attributes', 'assurance', 'mfa'], False),
            ('aal1', ['subject', 'attributes', 'assurance', 'level'], 'aal1'),
            ('forged-human-route', ['subject', 'attributes', 'principal_type_source'], 'registration-supplied'),
            ('indeterminate-route', ['subject', 'attributes', 'principal_type_source'], 'indeterminate'),
            ('wrong-tenant', ['subject', 'tenant'], 'tenant:other'),
            ('other-memo', ['resource', 'id'], 'memo:other'),
            ('omitted-c01', ['resource', 'id'], 'memo:infd-20260914-c01'),
            ('other-system', ['resource', 'system'], 'secrets-engine'),
            ('changed-version', ['context', 'memo_version'], record['memo_version'] + 1),
            ('changed-approval', ['context', 'approval_id'], 'other'),
            ('changed-digest', ['context', 'approval_binding_digest'], 'sha256:' + '0' * 64),
        ]:
            check(label + ':' + name, variant(path, value), 'deny')

args.receipt.write_text(json.dumps({
    'scope': 'local actual evaluator with synthetic identity; no live human approvals; T03 package untouched',
    'checks': results,
}, indent=2) + '\n')
print(len(results), 'policy checks passed')