# -*- coding: utf-8 -*- """utils.callback_registration — single registration helper for callback registration authority gate (task-2646 CALLBACK_REGISTRATION_AUTHORITY_GATE). Wraps dispatch.normal_fallback_callback_helper.launch_callback and adds: - XOR path enforcement (dispatch_path XOR direct_cron_path) - prompt byte classification - post-registration owner verification - state enum mapping (12 states) Layer A / NO-CRON / NO-WRITE / NO-SUBPROCESS: pure function wrapper. """ from __future__ import annotations import sys from dataclasses import dataclass, field from datetime import datetime, timezone 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). # 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) # dispatch.normal_fallback_callback_helper is imported lazily inside # register_callback() to avoid dispatch package resolution issues at module # load time (dispatch/ package vs dispatch.py shim, task-2646). # ── module constants ────────────────────────────────────────────────────────── HELPER_SCHEMA = "utils.callback_registration.v1" ANU_KEY = "c119085addb0f8b7" # ── 12 state enum ───────────────────────────────────────────────────────────── STATE_AUTHORITATIVE_CALLBACK_COLLECTOR_PROCESSED = ( "AUTHORITATIVE_CALLBACK_COLLECTOR_PROCESSED" ) STATE_NON_AUTHORITATIVE_SELF_COLLECTOR = "NON_AUTHORITATIVE_SELF_COLLECTOR" STATE_CALLBACK_MISSING = "CALLBACK_MISSING" STATE_CALLBACK_PROMPT_TOO_LARGE = "CALLBACK_PROMPT_TOO_LARGE" STATE_DISPATCH_SUBMITTED_UNVERIFIED = "DISPATCH_SUBMITTED_UNVERIFIED" STATE_OWNER_KEY_MISMATCH = "OWNER_KEY_MISMATCH" STATE_OWNER_KEY_VERIFIED = "OWNER_KEY_VERIFIED" STATE_REGISTRATION_HELPER_BYPASSED = "REGISTRATION_HELPER_BYPASSED" STATE_SCHEDULE_HISTORY_PENDING = "SCHEDULE_HISTORY_PENDING" STATE_CRON_LIST_AUTODELETED_FIRED = "CRON_LIST_AUTODELETED_FIRED" STATE_RESULT_ARTIFACT_SELF_ATTESTED = "RESULT_ARTIFACT_SELF_ATTESTED" STATE_SOURCE_CROSS_CHECK_PARTIAL = "SOURCE_CROSS_CHECK_PARTIAL" ALL_STATES: Tuple[str, ...] = ( STATE_AUTHORITATIVE_CALLBACK_COLLECTOR_PROCESSED, STATE_NON_AUTHORITATIVE_SELF_COLLECTOR, STATE_CALLBACK_MISSING, STATE_CALLBACK_PROMPT_TOO_LARGE, STATE_DISPATCH_SUBMITTED_UNVERIFIED, STATE_OWNER_KEY_MISMATCH, STATE_OWNER_KEY_VERIFIED, STATE_REGISTRATION_HELPER_BYPASSED, STATE_SCHEDULE_HISTORY_PENDING, STATE_CRON_LIST_AUTODELETED_FIRED, STATE_RESULT_ARTIFACT_SELF_ATTESTED, STATE_SOURCE_CROSS_CHECK_PARTIAL, ) # ── authority markers ───────────────────────────────────────────────────────── MARKER_AUTHORITATIVE = "AUTHORITATIVE_CALLBACK_COLLECTOR_PROCESSED" MARKER_NON_AUTHORITATIVE = "NON_AUTHORITATIVE_SELF_COLLECTOR" # ── verdict ─────────────────────────────────────────────────────────────────── VERDICT_PASS = "PASS" VERDICT_FAIL = "FAIL" # ── prompt byte classification boundaries ───────────────────────────────────── _BYTE_OK_TARGET_MAX = 3200 _BYTE_OK_ABOVE_TARGET_MAX = 3499 _BYTE_WARNING_BUT_ALLOWED_MAX = 3900 def classify_prompt_bytes(prompt: str) -> str: """Classify a prompt's UTF-8 byte length into one of 4 ranges. ≤3200 → OK_TARGET 3201-3499 → OK_ABOVE_TARGET 3500-3900 → WARNING_BUT_ALLOWED >3900 → HARD_BLOCK """ n = len((prompt or "").encode("utf-8")) if n <= _BYTE_OK_TARGET_MAX: return "OK_TARGET" if n <= _BYTE_OK_ABOVE_TARGET_MAX: return "OK_ABOVE_TARGET" if n <= _BYTE_WARNING_BUT_ALLOWED_MAX: return "WARNING_BUT_ALLOWED" return "HARD_BLOCK" @dataclass class RegistrationResult: """Result of a single callback registration attempt.""" schema: str verdict: str # PASS | FAIL state: str # one of 12 state enum kind: str task_id: str owner_key: str chat_id: str prompt_utf8_bytes: int prompt_byte_classification: str argv: Optional[List[str]] launch_decision: Optional[dict] authority_marker: Optional[str] reasons: List[str] = field(default_factory=list) registered_at_iso: Optional[str] = None @property def ok(self) -> bool: return self.verdict == VERDICT_PASS def to_json(self) -> dict: return { "schema": self.schema, "verdict": self.verdict, "state": self.state, "kind": self.kind, "task_id": self.task_id, "owner_key": self.owner_key, "chat_id": self.chat_id, "prompt_utf8_bytes": self.prompt_utf8_bytes, "prompt_byte_classification": self.prompt_byte_classification, "argv": list(self.argv) if self.argv is not None else None, "launch_decision": self.launch_decision, "authority_marker": self.authority_marker, "reasons": list(self.reasons), "registered_at_iso": self.registered_at_iso, } def replace(self, **kwargs) -> "RegistrationResult": """Return a new RegistrationResult with updated fields.""" import dataclasses return dataclasses.replace(self, **kwargs) def register_callback( *, kind: str, task_id: str, executor_key: str, owner_key: str, chat_id: str, prompt: str, at: str, canonical_root: Optional[str] = None, cron_id: Optional[str] = None, anu_keys: Sequence[str] = tuple(DEFAULT_ANU_KEYS), require_envelope: bool = True, dispatch_path: bool = False, direct_cron_path: bool = False, ) -> RegistrationResult: """Single entry point for callback registration (task-2646 ANCHOR-2). Exactly one of dispatch_path / direct_cron_path must be True. Both False → REGISTRATION_HELPER_BYPASSED (FAIL). Both True → also REGISTRATION_HELPER_BYPASSED (FAIL). Internally calls launch_callback from dispatch.normal_fallback_callback_helper and maps the result to one of the 12 state enum values. """ utf8_bytes = len((prompt or "").encode("utf-8")) byte_class = classify_prompt_bytes(prompt) registered_at = datetime.now(timezone.utc).isoformat() # XOR gate: exactly one path must be True if not (dispatch_path ^ direct_cron_path): return RegistrationResult( schema=HELPER_SCHEMA, verdict=VERDICT_FAIL, state=STATE_REGISTRATION_HELPER_BYPASSED, kind=kind, task_id=task_id, owner_key=owner_key, chat_id=str(chat_id), prompt_utf8_bytes=utf8_bytes, prompt_byte_classification=byte_class, argv=None, launch_decision=None, authority_marker=MARKER_NON_AUTHORITATIVE, reasons=[ "XOR path gate failed: exactly one of dispatch_path / " "direct_cron_path must be True — helper bypassed " "(REGISTRATION_HELPER_BYPASSED fail-closed)." ], registered_at_iso=registered_at, ) # Lazy import to avoid dispatch package resolution at module load time. from dispatch.normal_fallback_callback_helper import ( # noqa: PLC0415 LAUNCH_FAIL_CLOSED, LAUNCH_PASS, STATUS_CALLBACK_PROMPT_TOO_LARGE, launch_callback, ) # Delegate to the canonical launch_callback dec = launch_callback( kind=kind, task_id=task_id, executor_key=executor_key, owner_key=owner_key, chat_id=str(chat_id), prompt=prompt, at=at, canonical_root=canonical_root, cron_id=cron_id, anu_keys=anu_keys, require_envelope=require_envelope, ) dec_json = dec.to_json() # Map launch result → state enum if dec.verdict == LAUNCH_PASS and is_anu_key(owner_key, anu_keys): # Registered successfully; owner is ANU but not yet verified via # actual schedule history — DISPATCH_SUBMITTED_UNVERIFIED state = STATE_DISPATCH_SUBMITTED_UNVERIFIED verdict = VERDICT_PASS marker = MARKER_AUTHORITATIVE # provisional — to be confirmed by verify_actual_owner reasons = list(dec.reasons) + [ "launch_callback PASS + owner_key is ANU key → " "DISPATCH_SUBMITTED_UNVERIFIED (owner not yet confirmed via " "schedule_history; call verify_actual_owner to confirm)." ] elif dec.verdict == LAUNCH_FAIL_CLOSED and dec.status == STATUS_CALLBACK_PROMPT_TOO_LARGE: state = STATE_CALLBACK_PROMPT_TOO_LARGE verdict = VERDICT_FAIL marker = MARKER_NON_AUTHORITATIVE reasons = list(dec.reasons) elif dec.verdict == LAUNCH_FAIL_CLOSED and owner_key == executor_key: # Self-key attempted state = STATE_NON_AUTHORITATIVE_SELF_COLLECTOR verdict = VERDICT_FAIL marker = MARKER_NON_AUTHORITATIVE reasons = list(dec.reasons) + [ "owner_key == executor_key → NON_AUTHORITATIVE_SELF_COLLECTOR " "(self-key callback forbidden, fail-closed)." ] else: # General launch fail state = STATE_NON_AUTHORITATIVE_SELF_COLLECTOR verdict = VERDICT_FAIL marker = MARKER_NON_AUTHORITATIVE reasons = list(dec.reasons) return RegistrationResult( schema=HELPER_SCHEMA, verdict=verdict, state=state, kind=kind, task_id=task_id, owner_key=owner_key, chat_id=str(chat_id), prompt_utf8_bytes=utf8_bytes, prompt_byte_classification=byte_class, argv=dec.argv, launch_decision=dec_json, authority_marker=marker, reasons=reasons, registered_at_iso=registered_at, ) def verify_actual_owner( *, registration_result: RegistrationResult, observed_owner_key: str, observed_chat_id: str, observed_role: str, anu_keys: Sequence[str] = tuple(DEFAULT_ANU_KEYS), expected_chat_id: Optional[str] = None, ) -> RegistrationResult: """Verify the actual schedule owner key post-registration. Updates the state of a RegistrationResult that is DISPATCH_SUBMITTED_UNVERIFIED: - envelope.owner_key is ANU AND observed_owner_key ∈ anu_keys → state = OWNER_KEY_VERIFIED, marker = AUTHORITATIVE_CALLBACK_COLLECTOR_PROCESSED - envelope.owner_key is ANU but observed is self-key → state = OWNER_KEY_MISMATCH, marker = NON_AUTHORITATIVE_SELF_COLLECTOR (with "ENVELOPE_ACTUAL_MISMATCH" reason — ANCHOR-1) - observed_owner_key not in anu_keys → state = OWNER_KEY_MISMATCH """ rr = registration_result reasons: List[str] = list(rr.reasons) envelope_owner_is_anu = is_anu_key(rr.owner_key, anu_keys) observed_is_anu = is_anu_key(observed_owner_key, anu_keys) # Chat id mismatch check if expected_chat_id is not None and str(observed_chat_id) != str(expected_chat_id): reasons.append( f"chat_id mismatch: observed={observed_chat_id!r} " f"expected={expected_chat_id!r} (post-registration binding check)." ) if envelope_owner_is_anu and observed_is_anu and rr.owner_key == observed_owner_key: # Full match: envelope ANU == actual ANU new_state = STATE_OWNER_KEY_VERIFIED new_verdict = VERDICT_PASS new_marker = MARKER_AUTHORITATIVE reasons.append( f"verify_actual_owner PASS: envelope.owner_key={rr.owner_key!r} " f"== observed_owner_key={observed_owner_key!r} (both ANU) → " "OWNER_KEY_VERIFIED / AUTHORITATIVE_CALLBACK_COLLECTOR_PROCESSED." ) elif envelope_owner_is_anu and not observed_is_anu: # ANCHOR-1: envelope says ANU but actual is non-ANU (self-key) new_state = STATE_OWNER_KEY_MISMATCH new_verdict = VERDICT_FAIL new_marker = MARKER_NON_AUTHORITATIVE reasons.append( "ENVELOPE_ACTUAL_MISMATCH: envelope.collector_key is ANU key " f"({rr.owner_key!r}) but actual schedule owner " f"{observed_owner_key!r} is not ANU (ANCHOR-1 fail-closed)." ) elif not observed_is_anu: new_state = STATE_OWNER_KEY_MISMATCH new_verdict = VERDICT_FAIL new_marker = MARKER_NON_AUTHORITATIVE reasons.append( f"observed_owner_key {observed_owner_key!r} is not in anu_keys → " "OWNER_KEY_MISMATCH." ) else: # observed is ANU but envelope owner != observed (key rotation case) new_state = STATE_OWNER_KEY_VERIFIED new_verdict = VERDICT_PASS new_marker = MARKER_AUTHORITATIVE reasons.append( f"verify_actual_owner PASS: observed_owner_key={observed_owner_key!r} " "∈ anu_keys → OWNER_KEY_VERIFIED." ) verified_at = datetime.now(timezone.utc).isoformat() return rr.replace( state=new_state, verdict=new_verdict, authority_marker=new_marker, reasons=reasons, registered_at_iso=rr.registered_at_iso or verified_at, ) def selftest() -> dict: """Self-test: verify core registration + classification logic. Callable from CLI: python3 utils/callback_registration.py selftest """ failures: List[str] = [] # Test 1: classify_prompt_bytes tests_byte = [ ("", "OK_TARGET"), ("x" * 3200, "OK_TARGET"), ("x" * 3201, "OK_ABOVE_TARGET"), ("x" * 3499, "OK_ABOVE_TARGET"), ("x" * 3500, "WARNING_BUT_ALLOWED"), ("x" * 3900, "WARNING_BUT_ALLOWED"), ("x" * 3901, "HARD_BLOCK"), ] for prompt_str, expected in tests_byte: got = classify_prompt_bytes(prompt_str) if got != expected: failures.append( f"classify_prompt_bytes({len(prompt_str)} chars): " f"expected {expected!r}, got {got!r}" ) # Test 2: XOR path gate — both False → REGISTRATION_HELPER_BYPASSED rr = register_callback( kind="normal", task_id="selftest-0", executor_key="c38fb9955616e24d", owner_key=ANU_KEY, chat_id="chat-0", prompt="task_id=selftest-0\nowner_key=c119085addb0f8b7\n", at="10m", dispatch_path=False, direct_cron_path=False, ) if rr.state != STATE_REGISTRATION_HELPER_BYPASSED: failures.append( f"XOR gate (both False): expected REGISTRATION_HELPER_BYPASSED, " f"got {rr.state!r}" ) # Test 3: dispatch_path=True + ANU owner → DISPATCH_SUBMITTED_UNVERIFIED rr2 = register_callback( kind="normal", task_id="selftest-1", executor_key="c38fb9955616e24d", owner_key=ANU_KEY, chat_id="chat-1", prompt="task_id=selftest-1\nowner_key=c119085addb0f8b7\nat=10m\n", at="10m", dispatch_path=True, direct_cron_path=False, require_envelope=False, ) if rr2.state != STATE_DISPATCH_SUBMITTED_UNVERIFIED: failures.append( f"dispatch_path=True + ANU owner: expected " f"DISPATCH_SUBMITTED_UNVERIFIED, got {rr2.state!r}" ) # Test 4: verify_actual_owner → OWNER_KEY_VERIFIED rr3 = verify_actual_owner( registration_result=rr2, observed_owner_key=ANU_KEY, observed_chat_id="chat-1", observed_role="ANU", ) if rr3.state != STATE_OWNER_KEY_VERIFIED: failures.append( f"verify_actual_owner (ANU→ANU): expected OWNER_KEY_VERIFIED, " f"got {rr3.state!r}" ) # Test 5: dispatch_path=True + self-key → FAIL (NON_AUTHORITATIVE) rr4 = register_callback( kind="normal", task_id="selftest-2", executor_key="c38fb9955616e24d", owner_key="c38fb9955616e24d", # self-key chat_id="chat-2", prompt="task_id=selftest-2\nowner_key=c38fb9955616e24d\n", at="10m", dispatch_path=True, direct_cron_path=False, require_envelope=False, ) if rr4.verdict != VERDICT_FAIL: failures.append( f"self-key registration: expected FAIL, got {rr4.verdict!r}" ) # Test 6: All 12 states are defined if len(ALL_STATES) != 12: failures.append(f"ALL_STATES should have 12 entries, got {len(ALL_STATES)}") return { "schema": HELPER_SCHEMA, "ok": len(failures) == 0, "failures": failures, "tests_run": 6, } if __name__ == "__main__": import json as _json if len(sys.argv) > 1 and sys.argv[1] == "selftest": result = selftest() print(_json.dumps(result, ensure_ascii=False, indent=2)) sys.exit(0 if result["ok"] else 1) else: print("Usage: python3 utils/callback_registration.py selftest") sys.exit(1)