user-engine/tools/catalogue_bootstrap.py

"""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())