"""One-time supervised catalogue maintenance; no installed runtime endpoint. A private, exact-hash supervisor artifact is an out-of-band maintenance grant, not authentication of its JSON author. The operator must independently confirm the founder's native disposition before supplying it. No JWT role is invented. """ from __future__ import annotations import argparse from dataclasses import replace from datetime import datetime, timedelta, timezone import hashlib import json import os from pathlib import Path import re import stat from user_engine.adapters.postgres import PostgresUserEngineStore from user_engine.domain import (Actor, Application, ApplicationBinding, AuditRecord, AuthorizationDecision, AuthorizationEffect, PrincipalType) from user_engine.service import UserEngineService, AuthorizationDenied from user_engine.store_records import store_record_for APP = Application('informed-decision', 'Informed Decision', 'informed-decision') BINDING = ApplicationBinding('informed-decision', oidc_client_id='informed-decision-approver', protected_system_id='informed-decision') PROPOSAL = 'USER-CATALOGUE-INFDEC-20261011-R1' SCHEMA = 'user-engine.supervised-catalogue-maintenance.v1' KEYS = {'schema','proposal_id','approval_id','memo_id','memo_version','approval_receipt_sha256','executor_sha256','records_sha256','expires_at','operator_id','authority'} def digest(value): return hashlib.sha256(json.dumps(value, sort_keys=True, separators=(',', ':')).encode()).hexdigest() def records(): return [store_record_for(x).__dict__ for x in (APP, BINDING)] class Refused(ValueError): pass def validate_grant(grant, *, expected_grant_sha256, actual_executor_sha256, now): if (type(grant) is not dict or set(grant) != KEYS or grant['schema'] != SCHEMA or grant['proposal_id'] != PROPOSAL or grant['authority'] != 'founder-approved-one-time-administrative-exception' or grant['memo_id'] != 'infd-20261011-user-catalogue-bootstrap-r1' or type(grant['memo_version']) is not int or grant['memo_version'] != 1 or not re.fullmatch(r'[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}', grant['approval_id']) or not re.fullmatch(r'[a-zA-Z0-9:._-]{3,80}', grant['operator_id'])): raise Refused('closed grant required') for key in ('approval_receipt_sha256', 'executor_sha256', 'records_sha256'): if not re.fullmatch(r'[0-9a-f]{64}', grant[key]): raise Refused('exact evidence digests required') expires = datetime.fromisoformat(grant['expires_at'].replace('Z', '+00:00')) if (expires.tzinfo is None or now.tzinfo is None or now >= expires or expires > datetime(2026, 10, 12, 23, 59, 59, tzinfo=timezone.utc)): raise Refused('expired grant') if (digest(grant) != expected_grant_sha256 or grant['executor_sha256'] != actual_executor_sha256 or grant['records_sha256'] != digest(records())): raise Refused('approved source differs') return expires def _summary(cursor, audit_id): cursor.execute('SELECT summary FROM user_engine_audit_records WHERE audit_id = %s', (audit_id,)) row = cursor.fetchone() return None if row is None else json.loads(row[0]) def _event_ids(cursor, correlation): cursor.execute('SELECT event_id FROM user_engine_outbox_events WHERE correlation_id = %s ORDER BY event_id', (correlation,)) return [r[0] for r in cursor.fetchall()] def _actor(grant): return Actor(issuer='urn:supervised-maintenance', subject=grant['operator_id'], tenant='tenant:platform', principal_type=PrincipalType.SERVICE, audience=(), claims={'maintenance_proposal': PROPOSAL}) class _Capture: def __init__(self, ordinary): self.ordinary = ordinary self.result = None def check(self, request): self.result = self.ordinary.check(request) return self.result class _ExceptionAuthorization: def __init__(self, actor, correlation, expires): self.actor, self.correlation, self.expires = actor, correlation, expires self.used = False def check(self, request): now = datetime.now(timezone.utc) if (self.used or now >= self.expires or request.actor != self.actor or request.action != 'application.register' or request.tenant != 'tenant:platform' or request.resource_type != 'user-engine:application' or request.resource_id != APP.application_id or request.application_id != APP.application_id or request.correlation_id != self.correlation): raise Refused('exception outside fixed registration') self.used = True return AuthorizationDecision(effect=AuthorizationEffect.ALLOW, reason='explicit supervised maintenance exception; not a PDP allow', decision_id=None, binding='supervised-maintenance', issued_at=now, lifetime=min(timedelta(seconds=30), self.expires-now)) def execute(connection, ordinary_authorization, grant, *, expected_grant_sha256, actual_executor_sha256, mode='dry-run', now=None): now = now or datetime.now(timezone.utc) expires = validate_grant(grant, expected_grant_sha256=expected_grant_sha256, actual_executor_sha256=actual_executor_sha256, now=now) if mode not in ('dry-run', 'apply', 'rollback'): raise Refused('unknown operation') store = PostgresUserEngineStore(connection) if not store.ready: raise Refused('existing schema required; no migration') capture = _Capture(ordinary_authorization) service = UserEngineService(store=store, identity_adapter=object(), authorization=capture) actor = _actor(grant) ledger = 'aud_catalogue_bootstrap_' + grant['approval_id'] correlation = 'maintenance:' + grant['approval_id'] # Construct both services before BEGIN: the owner readiness probe rolls back # its connection. Constructing a service in the transaction would discard # the genuine denial audit and release the native serialization lock. exception_service = UserEngineService(store=store, identity_adapter=object(), authorization=_ExceptionAuthorization(actor, correlation, expires)) outcome = None try: with store.transaction(): with connection.cursor() as cursor: cursor.execute('SET TRANSACTION ISOLATION LEVEL SERIALIZABLE') cursor.execute('SELECT pg_advisory_xact_lock(hashtextextended(%s, 0))', ('user-engine:one-app-catalogue:informed-decision',)) cursor.execute('SELECT clock_timestamp()') if cursor.fetchone()[0] >= expires: raise Refused('native transaction grant expired') previous = _summary(cursor, ledger) rollback_done = _summary(cursor, ledger + '_rollback') if mode == 'rollback': if previous is None or rollback_done is not None or not previous['created']: raise Refused('no unused created-only recovery authority') if previous['grant_sha256'] != expected_grant_sha256: raise Refused('recovery grant differs from consumed authority') if store.application(APP.application_id) != APP or store.binding(APP.application_id) != BINDING: raise Refused('created metadata changed') cursor.execute("SELECT COUNT(*) FROM user_engine_records WHERE application_id = %s AND record_type NOT IN ('applications', 'application_bindings')", (APP.application_id,)) if cursor.fetchone()[0]: raise Refused('later dependent records exist') # Match the existing delivery worker's session advisory lock namespace. for event_id in previous['event_ids']: cursor.execute('SELECT pg_advisory_xact_lock(hashtextextended(%s, 0))', ('user-engine:tenant-lifecycle:outbox-event:' + event_id,)) for event_id in previous['event_ids']: cursor.execute('SELECT event_id FROM user_engine_outbox_events WHERE event_id = %s FOR UPDATE', (event_id,)) if cursor.fetchone() is None: raise Refused('created event missing') event = store.outbox_event(event_id) if (event is None or event.correlation_id != correlation or event.claimed_by is not None or event.claimed_at is not None or event.delivered_at is not None): raise Refused('claimed/delivered event needs owner reconciliation') store.save_outbox(replace(event, dead_lettered_at=now, failure_reason='supervised bootstrap rolled back before delivery')) for row in records(): cursor.execute('DELETE FROM user_engine_records WHERE record_type = %s AND record_key = %s AND payload = %s::jsonb', (row['record_type'], row['record_key'], json.dumps(row['payload']))) if cursor.rowcount != 1: raise Refused('created-only compare-and-delete failed') store.append_audit(AuditRecord(ledger + '_rollback', actor, 'application.metadata.bootstrap.rollback', APP.application_id, actor.tenant, correlation, summary=json.dumps({'maintenance_authority': grant['approval_id'], 'grant_sha256': expected_grant_sha256, 'removed': ['applications', 'application_bindings'], 'events_retained_dead_lettered': previous['event_ids']}))) outcome = {'result': 'rolled-back', 'ordinary_pdp_allow_claimed': False, 'created_entities_removed': 2, 'audit_and_event_history_retained': True} else: if previous is not None: raise Refused('native single-use grant already consumed') a, b = store.application(APP.application_id), store.binding(APP.application_id) if (a is None) != (b is None) or (a is not None and (a != APP or b != BINDING)): raise Refused('partial or conflicting metadata') created = a is None if created: try: service.register_application(actor, APP, binding=BINDING, correlation_id=correlation) except AuthorizationDenied: refusal = capture.result if (refusal is None or refusal.allowed or refusal.decision_id is None or refusal.reason == 'authorization service unavailable'): raise Refused('real ordinary PDP refusal required') else: raise Refused('ordinary authorization unexpectedly allowed; use normal owner path') exception_service.register_application(actor, APP, binding=BINDING, correlation_id=correlation) if store.application(APP.application_id) != APP or store.binding(APP.application_id) != BINDING: raise Refused('post-write readback differs') event_ids = _event_ids(cursor, correlation) for event_id in event_ids: event = store.outbox_event(event_id) store.save_outbox(replace(event, payload={**event.payload, 'maintenance_exception': True, 'proposal_id': PROPOSAL, 'approval_id': grant['approval_id'], 'operator_id': actor.subject})) summary = {'created': created, 'grant_sha256': expected_grant_sha256, 'records_sha256': grant['records_sha256'], 'executor_sha256': actual_executor_sha256, 'approval_receipt_sha256': grant['approval_receipt_sha256'], 'ordinary_pdp_denial_id': capture.result.decision_id if created else None, 'event_ids': event_ids, 'authority': grant['authority'], 'not_human_authentication': True} store.append_audit(AuditRecord(ledger, actor, 'application.metadata.bootstrap', APP.application_id, actor.tenant, correlation, summary=json.dumps(summary))) outcome = {'result': 'created' if created else 'no-change', 'ordinary_pdp_refusal_retained': created, 'ordinary_pdp_allow_claimed': False, 'created_entities': 2 if created else 0, 'created_event_ids': event_ids, 'grant_native_single_use': True} cursor.execute('SELECT clock_timestamp()') if cursor.fetchone()[0] >= expires: raise Refused('grant expired before transaction commit') if mode == 'dry-run': raise _DryRun(outcome) except _DryRun as dry: return {**dry.outcome, 'transaction_committed': False, 'dry_run_rolled_back': True} return {**outcome, 'transaction_committed': True} class _DryRun(Exception): def __init__(self, outcome): self.outcome = outcome def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument('--grant-file', required=True, type=Path) parser.add_argument('--expected-grant-sha256', required=True) parser.add_argument('--mode', choices=['dry-run','apply','rollback'], default='dry-run') args = parser.parse_args() # Whole stdout/stderr, including dependency/driver errors, stays contained. saved = os.dup(1) null = os.open(os.devnull, os.O_WRONLY) os.dup2(null, 1); os.dup2(null, 2) try: info = args.grant_file.lstat() if (not stat.S_ISREG(info.st_mode) or info.st_uid != os.getuid() or stat.S_IMODE(info.st_mode) != 0o600 or info.st_size > 8192): raise Refused('private reviewed grant artifact required') grant = json.loads(args.grant_file.read_text()) source_hash = hashlib.sha256(Path(__file__).read_bytes()).hexdigest() import psycopg from user_engine.adapters.flex_auth import FlexAuthHTTPAdapter ordinary = FlexAuthHTTPAdapter(base_url=os.environ['USER_ENGINE_FLEX_AUTH_URL'], bearer_token_file=os.environ['USER_ENGINE_FLEX_AUTH_TOKEN_FILE'], timeout_seconds=3) with psycopg.connect(os.environ['USER_ENGINE_DATABASE_URL']) as connection: result = execute(connection, ordinary, grant, expected_grant_sha256=args.expected_grant_sha256, actual_executor_sha256=source_hash, mode=args.mode) status = 0 except Exception: result, status = {'result': 'refused-or-rolled-back', 'raw_output_suppressed': True}, 2 os.dup2(saved, 1); os.close(saved); os.close(null) print(json.dumps(result, sort_keys=True)) return status if __name__ == '__main__': raise SystemExit(main())