# -*- coding: utf-8 -*- """utils.callback_authority_validator — envelope vs actual owner authority gate. task-2646 CALLBACK_REGISTRATION_AUTHORITY_GATE. Validates whether the actual schedule owner key matches the envelope collector_key and is an independent ANU key. Layer A / NO-CRON / NO-WRITE / NO-SUBPROCESS: pure function, zero side-effects. """ from __future__ import annotations from dataclasses import dataclass, field from typing import List, Optional, Sequence, Tuple # Inline constants from dispatch.callback_owner_enforcer to avoid dispatch # package resolution issues (dispatch/ dir vs dispatch.py shim, task-2646). # These values are pinned by 회장 §10 and must match exactly. _ANU_KEY_2553 = "c119085addb0f8b7" DEFAULT_ANU_KEYS: frozenset = frozenset({_ANU_KEY_2553}) def is_anu_key(key, anu_keys) -> bool: """True iff key is a non-empty configured independent ANU key.""" return bool(key) and key in set(anu_keys) # ── module constants ────────────────────────────────────────────────────────── SCHEMA = "utils.callback_authority_validator.v1" PASS = "PASS" FAIL = "FAIL" # ── authority markers ───────────────────────────────────────────────────────── MARKER_AUTHORITATIVE = "AUTHORITATIVE_CALLBACK_COLLECTOR_PROCESSED" MARKER_NON_AUTHORITATIVE = "NON_AUTHORITATIVE_SELF_COLLECTOR" # ── state enum subset used by this module ───────────────────────────────────── STATE_NON_AUTHORITATIVE_SELF_COLLECTOR = "NON_AUTHORITATIVE_SELF_COLLECTOR" STATE_OWNER_KEY_MISMATCH = "OWNER_KEY_MISMATCH" STATE_OWNER_KEY_VERIFIED = "OWNER_KEY_VERIFIED" @dataclass class AuthorityVerdict: """Result of an authority validation check.""" schema: str verdict: str # PASS | FAIL state: str # one of state enum envelope_collector_key: str actual_owner_key: str executor_key: str authority_marker: Optional[str] reasons: List[str] = field(default_factory=list) @property def ok(self) -> bool: return self.verdict == PASS def to_json(self) -> dict: return { "schema": self.schema, "verdict": self.verdict, "state": self.state, "envelope_collector_key": self.envelope_collector_key, "actual_owner_key": self.actual_owner_key, "executor_key": self.executor_key, "authority_marker": self.authority_marker, "reasons": list(self.reasons), } def validate_authority( *, envelope_collector_key: str, actual_owner_key: str, executor_key: str, anu_keys: Sequence[str] = tuple(DEFAULT_ANU_KEYS), ) -> AuthorityVerdict: """Gate: validate callback registration authority. Decision logic (in priority order): 1. actual == executor_key → state=NON_AUTHORITATIVE_SELF_COLLECTOR, verdict=FAIL, marker=NON_AUTHORITATIVE_SELF_COLLECTOR (self-key callback is always forbidden, regardless of envelope) If envelope=ANU, reason includes "ENVELOPE_ACTUAL_MISMATCH" (ANCHOR-1) 2. envelope_collector_key ∈ anu_keys BUT actual_owner_key ∉ anu_keys (actual != executor — handled above) → state=OWNER_KEY_MISMATCH, verdict=FAIL, marker=NON_AUTHORITATIVE_SELF_COLLECTOR reason includes "ENVELOPE_ACTUAL_MISMATCH" (ANCHOR-1) 3. envelope_collector_key ∉ anu_keys AND actual_owner_key ∉ anu_keys → state=OWNER_KEY_MISMATCH, verdict=FAIL 4. actual_owner_key ∈ anu_keys AND envelope_collector_key == actual_owner_key → state=OWNER_KEY_VERIFIED, verdict=PASS, marker=AUTHORITATIVE_CALLBACK_COLLECTOR_PROCESSED """ reasons: List[str] = [] envelope_is_anu = is_anu_key(envelope_collector_key, anu_keys) actual_is_anu = is_anu_key(actual_owner_key, anu_keys) # Rule 1: actual == executor → self-key forbidden (ANCHOR-5) # Self-key is always FAIL regardless of envelope. # If envelope=ANU, also include ENVELOPE_ACTUAL_MISMATCH reason (ANCHOR-1). if actual_owner_key == executor_key: if envelope_is_anu: reasons.append( "ENVELOPE_ACTUAL_MISMATCH: envelope.collector_key is ANU key " f"({envelope_collector_key!r}) but actual_owner_key " f"{actual_owner_key!r} == executor_key → " "NON_AUTHORITATIVE_SELF_COLLECTOR (self-key + ANU envelope " "mismatch; ANCHOR-1 + ANCHOR-5 fail-closed)." ) else: reasons.append( f"actual_owner_key {actual_owner_key!r} == executor_key → " "NON_AUTHORITATIVE_SELF_COLLECTOR (self-key callback " "forbidden, fail-closed)." ) return AuthorityVerdict( schema=SCHEMA, verdict=FAIL, state=STATE_NON_AUTHORITATIVE_SELF_COLLECTOR, envelope_collector_key=envelope_collector_key, actual_owner_key=actual_owner_key, executor_key=executor_key, authority_marker=MARKER_NON_AUTHORITATIVE, reasons=reasons, ) # Rule 2: envelope=ANU but actual is non-ANU non-self (ANCHOR-1 mismatch) if envelope_is_anu and not actual_is_anu: reasons.append( "ENVELOPE_ACTUAL_MISMATCH: envelope.collector_key is ANU key " f"({envelope_collector_key!r}) but actual schedule owner " f"{actual_owner_key!r} is not an ANU key (ANCHOR-1 fail-closed)." ) return AuthorityVerdict( schema=SCHEMA, verdict=FAIL, state=STATE_OWNER_KEY_MISMATCH, envelope_collector_key=envelope_collector_key, actual_owner_key=actual_owner_key, executor_key=executor_key, authority_marker=MARKER_NON_AUTHORITATIVE, reasons=reasons, ) # Rule 3: neither key is ANU (actual != executor — handled above) if not actual_is_anu: reasons.append( f"actual_owner_key {actual_owner_key!r} is not an ANU key " f"(anu_keys={list(anu_keys)}) → OWNER_KEY_MISMATCH." ) return AuthorityVerdict( schema=SCHEMA, verdict=FAIL, state=STATE_OWNER_KEY_MISMATCH, envelope_collector_key=envelope_collector_key, actual_owner_key=actual_owner_key, executor_key=executor_key, authority_marker=None, reasons=reasons, ) # Rule 4: actual is ANU + envelope matches actual if actual_is_anu and envelope_collector_key == actual_owner_key: reasons.append( f"actual_owner_key {actual_owner_key!r} ∈ anu_keys AND " f"envelope_collector_key {envelope_collector_key!r} == actual → " "OWNER_KEY_VERIFIED / AUTHORITATIVE_CALLBACK_COLLECTOR_PROCESSED." ) return AuthorityVerdict( schema=SCHEMA, verdict=PASS, state=STATE_OWNER_KEY_VERIFIED, envelope_collector_key=envelope_collector_key, actual_owner_key=actual_owner_key, executor_key=executor_key, authority_marker=MARKER_AUTHORITATIVE, reasons=reasons, ) # Fallback: actual is ANU but envelope doesn't match if actual_is_anu and not envelope_is_anu: reasons.append( f"actual_owner_key {actual_owner_key!r} ∈ anu_keys but " f"envelope_collector_key {envelope_collector_key!r} is not ANU → " "OWNER_KEY_MISMATCH (envelope not authoritative)." ) return AuthorityVerdict( schema=SCHEMA, verdict=FAIL, state=STATE_OWNER_KEY_MISMATCH, envelope_collector_key=envelope_collector_key, actual_owner_key=actual_owner_key, executor_key=executor_key, authority_marker=None, reasons=reasons, ) # Both are ANU but differ (key rotation / multiple ANU keys) reasons.append( f"actual_owner_key {actual_owner_key!r} ∈ anu_keys (PASS — " "envelope/actual key difference within ANU key set is acceptable)." ) return AuthorityVerdict( schema=SCHEMA, verdict=PASS, state=STATE_OWNER_KEY_VERIFIED, envelope_collector_key=envelope_collector_key, actual_owner_key=actual_owner_key, executor_key=executor_key, authority_marker=MARKER_AUTHORITATIVE, reasons=reasons, ) __all__ = [ "SCHEMA", "PASS", "FAIL", "MARKER_AUTHORITATIVE", "MARKER_NON_AUTHORITATIVE", "AuthorityVerdict", "validate_authority", ]