"""present_existing intake profile (INFD-WP-0010-T01). A registered submitter presents an approval it created itself in approval-engine. Intake verifies the object read-only and never creates, enters or consumes. Synthetic fixtures; the opt-in test at the end uses approval-engine's real store (INFD_APPROVAL_ENGINE_SOURCE) to produce the object shape and binding digest. """ from datetime import datetime, timedelta, timezone import json import os from pathlib import Path import sys import pytest from informed_decision.intake import IntakeRefused, PresentExisting, SubmitterGrant, validate_submission from informed_decision.intake_service import IntakeService, TokenReview, configuration, IntakeError from informed_decision.store import Store from test_intake import envelope as base_envelope from test_intake_service import request from informed_decision.intake_service import AUDIENCE EXECUTOR = 'system:serviceaccount:sso:nk-bind-executor' STATE_HUB = 'system:serviceaccount:state-hub:state-hub' ACTOR = 'service:nk-bind-executor' APPROVAL = '0f1e2d3c-4b5a-4968-8776-655443322110' DIGEST = 'sha256:' + 'b' * 64 TARGET = {'admission_id': 'BEA-2026-0001', 'admission_digest': 'sha256:' + 'c' * 64} PROFILE = PresentExisting(ACTOR, frozenset({'bounded-execution'})) GRANT = SubmitterGrant('nk-bind-executor', frozenset({'bounded-execution'}), frozenset({'net-kingdom-admins'})) def envelope(approval_id=APPROVAL, digest=DIGEST): env = base_envelope() env['idempotency_key'] = 'bea-2026-0001-r1' env['memo_class'] = 'bounded-execution' env['approval_intent'] = {'action': 'bounded-execution', 'principal': 'admin', 'purpose': 'Rotate one consumer bind password', 'target': dict(TARGET)} env['memo']['id'] = 'bea-2026-0001-r1' env['memo']['approval_id'] = approval_id env['memo']['approval_binding_digest'] = digest del env['serves'] return env def approval(**changes): now = datetime.now(timezone.utc) body = {'id': APPROVAL, 'status': 'requested', 'entries': [], 'required_count': 1, 'binding': {'action': 'bounded-execution', 'actor': ACTOR, 'principal': 'admin', 'purpose': 'Rotate one consumer bind password', 'target': dict(TARGET), 'digest': DIGEST, 'human_control': True, 'pdp_path': False, 'pdp_digest': None}, 'validity': {'not_before': (now - timedelta(minutes=1)).isoformat(), 'expires_at': (now + timedelta(hours=1)).isoformat()}} for key, value in changes.items(): if key.startswith('binding.'): body['binding'][key[8:]] = value else: body[key] = value return body def review(body): username = {'executor-token': EXECUTOR, 'hub-token': STATE_HUB}[body['spec']['token']] return {'status': {'authenticated': True, 'audiences': [AUDIENCE], 'user': {'username': username, 'uid': 'workload-uid'}}} @pytest.fixture def setup(tmp_path): private = tmp_path / 'private' private.mkdir(mode=0o700) store = Store(private / 'review.sqlite') submitters, groups = tmp_path / 'submitters.json', tmp_path / 'groups.json' submitters.write_text(json.dumps({ EXECUTOR: {'submitter_id': 'nk-bind-executor', 'memo_classes': ['bounded-execution'], 'review_groups': ['net-kingdom-admins'], 'daily_limit': 5, 'present_existing': {'approval_actor': ACTOR, 'actions': ['bounded-execution']}}, STATE_HUB: {'submitter_id': 'state-hub', 'memo_classes': ['credential-lane'], 'review_groups': ['net-kingdom-admins'], 'daily_limit': 5}})) groups.write_text(json.dumps({'net-kingdom-admins': {'subject': 'keycape-human', 'display_name': 'Reviewer'}})) created, reads = [], [] current = {'body': approval()} def create(body): created.append(body) return {'id': '12345678-1234-1234-1234-123456789abc', 'status': 'requested', 'binding': { **body['binding'], 'human_control': True, 'digest': 'sha256:' + 'a' * 64}} def read(approval_id): reads.append(approval_id) return json.loads(json.dumps(current['body'])) service = IntakeService(store, TokenReview(review), create, submitters_path=submitters, groups_path=groups, reader=read) return service, store, created, reads, current def test_presents_existing_without_creating(setup): service, store, created, reads, _ = setup status, receipt = request(service, envelope(), token='executor-token') assert status == 201 assert receipt == {'memo_id': 'bea-2026-0001-r1', 'approval_id': APPROVAL, 'approval_binding_digest': DIGEST} assert created == [] and reads == [APPROVAL] memo = store.memo('bea-2026-0001-r1') assert memo.approval_id == APPROVAL and memo.binding.principal.id == 'keycape-human' context = store.intake_context(memo) assert context['approval_requester'] == ACTOR assert context['intake_submitter'] == 'nk-bind-executor' assert context['approval_action'] == 'bounded-execution' # compact-sitting v14 # Replay is served from the ledger without another read. assert request(service, envelope(), token='executor-token') == (200, receipt) assert reads == [APPROVAL] @pytest.mark.parametrize('changes, code', [ ({'status': 'approved'}, 'approval_not_requested'), ({'status': 'consumed'}, 'approval_not_requested'), ({'entries': [{'subject': 'someone'}]}, 'approval_has_entries'), ({'binding.human_control': False}, 'approval_not_human_controlled'), ({'binding.actor': 'informed-decision'}, 'approval_actor_mismatch'), ({'binding.actor': 'service:other'}, 'approval_actor_mismatch'), ({'binding.digest': 'sha256:' + 'd' * 64}, 'approval_digest_mismatch'), ({'binding.purpose': 'Something else'}, 'approval_binding_mismatch'), ({'binding.target': {'admission_id': 'BEA-2026-0002', 'admission_digest': TARGET['admission_digest']}}, 'approval_binding_mismatch'), ({'required_count': 2}, 'approval_required_count_unsupported'), ({'validity': {'not_before': '2026-01-01T00:00:00+00:00', 'expires_at': '2026-01-02T00:00:00+00:00'}}, 'approval_expired'), ({'id': '99999999-1234-1234-1234-123456789abc'}, 'approval_mismatch'), ]) def test_refuses_unpresentable_approval_and_releases_reservation(setup, changes, code): service, store, created, _, current = setup current['body'] = approval(**changes) assert request(service, envelope(), token='executor-token') == (409, {'error': code}) assert created == [] with store._connection() as db: assert db.execute('SELECT COUNT(*) FROM memos').fetchone()[0] == 0 assert db.execute('SELECT COUNT(*) FROM intake_ledger').fetchone()[0] == 0 # Once the object is presentable the same submission succeeds. current['body'] = approval() assert request(service, envelope(), token='executor-token')[0] == 201 def test_unreadable_approval_fails_closed(setup): service, store, created, _, _ = setup def broken(approval_id): raise OSError('upstream detail never echoed') service.reader = broken assert request(service, envelope(), token='executor-token') == (503, {'error': 'approval_unreadable'}) service.reader = None assert request(service, envelope(), token='executor-token') == (503, {'error': 'present_existing_unavailable'}) assert created == [] def test_one_approval_one_memo(setup): service, _, _, _, _ = setup assert request(service, envelope(), token='executor-token')[0] == 201 second = envelope() second['idempotency_key'] = 'bea-2026-0001-r2' second['memo']['id'] = 'bea-2026-0001-r2' assert request(service, second, token='executor-token') == (409, {'error': 'approval_already_presented'}) def test_profile_rejects_unpinned_and_unadmitted_actions(setup): service, _, created, reads, _ = setup unpinned = envelope(approval_id=None, digest=None) assert request(service, unpinned, token='executor-token') == (400, {'error': 'invalid_text'}) other = envelope() other['approval_intent']['action'] = 'deploy' assert request(service, other, token='executor-token') == (400, {'error': 'action_not_admitted'}) assert created == [] and reads == [] def test_existing_submitters_keep_create_path(setup): service, store, created, reads, _ = setup status, receipt = request(service, base_envelope(), token='hub-token') assert status == 201 and len(created) == 1 and reads == [] assert created[0]['binding']['actor'] == 'informed-decision' context = store.intake_context(store.memo(receipt['memo_id'])) assert context['approval_requester'] == 'informed-decision' assert 'approval_action' not in context # v13 shape unchanged for the create path # A create-path submitter still cannot pin an approval. assert request(service, envelope(), token='hub-token')[0] == 400 def test_validation_actor_comes_from_profile_not_body(): out = validate_submission(envelope(), submitter_id='nk-bind-executor', grant=GRANT, present_existing=PROFILE) assert out.profile == 'present_existing' assert out.approval_binding == {'action': 'bounded-execution', 'actor': ACTOR, 'principal': 'admin', 'purpose': 'Rotate one consumer bind password', 'target': TARGET} with pytest.raises(IntakeRefused) as error: validate_submission(envelope(), submitter_id='nk-bind-executor', grant=GRANT) assert error.value.reason in ('unknown_field', 'approval_pinned_by_submitter') # BEA target or pin @pytest.mark.parametrize('profile', [ {'approval_actor': 'informed-decision', 'actions': ['bounded-execution']}, {'approval_actor': 'service:informed-decision', 'actions': ['bounded-execution']}, {'approval_actor': ACTOR, 'actions': []}, {'approval_actor': ACTOR, 'actions': ['Bad Action']}, {'approval_actor': ACTOR, 'actions': ['bounded-execution'], 'extra': True}, ]) def test_registry_profile_is_closed(tmp_path, profile): submitters, groups = tmp_path / 's.json', tmp_path / 'g.json' submitters.write_text(json.dumps({EXECUTOR: {'submitter_id': 'x', 'memo_classes': [], 'review_groups': [], 'daily_limit': 1, 'present_existing': profile}})) groups.write_text('{}') with pytest.raises(IntakeError): configuration(submitters, groups) def test_reader_adapter_is_get_only(): from informed_decision.intake_approval import IntakeApprovalReader calls = [] class Transport: def request(self, method, url, *, headers, body=None): calls.append((method, url, body)) return 200, approval() reader = IntakeApprovalReader('https://engine.example', lambda: 'read-token', transport=Transport()) assert reader(APPROVAL)['id'] == APPROVAL assert calls == [('GET', 'https://engine.example/v1/approvals/' + APPROVAL, None)] with pytest.raises(IntakeError): reader('../consume') def test_real_engine_object_is_presentable(setup, tmp_path): """Opt-in: approval-engine's own store creates the object as the executor would.""" source = os.environ.get('INFD_APPROVAL_ENGINE_SOURCE') if not source: pytest.skip('set INFD_APPROVAL_ENGINE_SOURCE for the real-engine contract check') assert (Path(source) / 'approval_engine' / 'store.py').is_file() sys.path.insert(0, source) from approval_engine.store import Engine service, store, created, _, _ = setup engine = Engine(tmp_path / 'approval.db') now = datetime.now(timezone.utc) record = engine.create({'action': 'bounded-execution', 'actor': ACTOR, 'principal': 'admin', 'purpose': 'Rotate one consumer bind password', 'target': dict(TARGET)}, {'not_before': (now - timedelta(seconds=1)).isoformat(), 'expires_at': (now + timedelta(minutes=30)).isoformat()}, human_control=True, approval_id=APPROVAL) real = engine.get(APPROVAL).as_dict() service.reader = lambda approval_id: engine.get(approval_id).as_dict() env = envelope(digest=real['binding']['digest']) status, receipt = request(service, env, token='executor-token') assert status == 201 and created == [] assert receipt['approval_binding_digest'] == real['binding']['digest'] assert engine.get(APPROVAL).as_dict()['entries'] == [] and record is not None # A wrong digest from the submitter is refused against the real object. other = envelope(digest='sha256:' + 'e' * 64) other['idempotency_key'] = 'bea-2026-0001-r9' other['memo']['id'] = 'bea-2026-0001-r9' assert request(service, other, token='executor-token') == (409, {'error': 'approval_already_presented'}) class _StopAtPolicy(Exception): pass class _RecordingPolicy: version = 'v14' def __init__(self): self.requests = [] def check(self, body): self.requests.append(body) raise _StopAtPolicy() def _review(store, live): from dataclasses import replace from informed_decision.review import ReviewController from test_policy_client import human class Engine: def get_approval(self, approval_id): return json.loads(json.dumps(live)) policy = _RecordingPolicy() controller = ReviewController(store, policy, lambda session: Engine(), intake_attributes=True) return controller, policy, replace(human(), subject='keycape-human') def test_review_sends_approval_action_to_the_pdp(setup): service, store, _, _, _ = setup assert request(service, envelope(), token='executor-token')[0] == 201 controller, policy, session = _review(store, approval()) with pytest.raises(_StopAtPolicy): controller.open(session, 'bea-2026-0001-r1') context = policy.requests[0]['context'] assert context['approval_action'] == 'bounded-execution' assert context['approval_requester'] == ACTOR and context['intake_submitter'] == 'nk-bind-executor' def test_review_refuses_changed_live_action_before_policy(setup): from informed_decision.review import ReviewError service, store, _, _, _ = setup assert request(service, envelope(), token='executor-token')[0] == 201 controller, policy, session = _review(store, approval(**{'binding.action': 'deploy'})) with pytest.raises(ReviewError, match='binding_changed'): controller.open(session, 'bea-2026-0001-r1') assert policy.requests == [] def test_build_request_admits_approval_action_only_for_present_existing(setup): from dataclasses import replace from informed_decision.policy import build_request from test_policy_client import human service, store, _, _, _ = setup assert request(service, envelope(), token='executor-token')[0] == 201 memo = store.memo('bea-2026-0001-r1') session = replace(human(), subject='keycape-human') context = store.intake_context(memo) assert build_request(session, memo, 'read', 'v14', intake_context=context)['context']['approval_action'] \ == 'bounded-execution' with pytest.raises(ValueError): build_request(session, memo, 'read', 'v14', intake_context={**context, 'approval_requester': 'informed-decision'}) with pytest.raises(ValueError): build_request(session, memo, 'read', 'v14', intake_context={**context, 'approval_scope': 'x'}) without = {k: v for k, v in context.items() if k != 'approval_action'} assert 'approval_action' not in build_request(session, memo, 'read', 'v13', intake_context=without)['context']