# -*- coding: utf-8 -*- """task-2553+49 AUTHORITATIVE — callback owner/key/role runtime validation, real-entrypoint regression (§8 reg 1~10, 21, 22, 23, 25, 26, 30). 회장 §5 중요: mock-only "그럴 것이다" 금지. 모든 케이스는 실 runtime entrypoint (``dispatch.core`` 재노출 + ``anu_v3`` runtime guard + narrow helper ``build_anu_owned_callback_request``) 를 **직접 호출** 한다. """ from __future__ import annotations import hashlib import importlib.util import json import subprocess import sys import unittest from pathlib import Path _ROOT = Path(__file__).resolve().parent.parent.parent if str(_ROOT) not in sys.path: sys.path.insert(0, str(_ROOT)) def _load(modname: str, relpath: str): """Hermetic file-path import (collision-proof vs tests/dispatch shadow); identical strategy to the narrow +49 suite.""" if modname in sys.modules: return sys.modules[modname] spec = importlib.util.spec_from_file_location(modname, _ROOT / relpath) assert spec is not None and spec.loader is not None mod = importlib.util.module_from_spec(spec) sys.modules[modname] = mod spec.loader.exec_module(mod) return mod # Pre-seed canonical dotted names (real workspace modules, NOT mocks). _load("dispatch.executor_completion_contract", "dispatch/executor_completion_contract.py") _load("dispatch.spec_template_validator", "dispatch/spec_template_validator.py") _load("anu_v3.callback_4tuple_registry", "anu_v3/callback_4tuple_registry.py") _load("dispatch.callback_owner_enforcer", "dispatch/callback_owner_enforcer.py") _hlp = _load("dispatch.normal_fallback_callback_helper", "dispatch/normal_fallback_callback_helper.py") _valmod = _load("anu_v3.callback_owner_validator", "anu_v3/callback_owner_validator.py") _load("anu_v3.authoritative_verdict_selector", "anu_v3/authoritative_verdict_selector.py") _load("anu_v3.self_collector_guard", "anu_v3/self_collector_guard.py") _load("anu_v3.writeback_binding_conflict_guard", "anu_v3/writeback_binding_conflict_guard.py") _grd = _load("dispatch.cron_dispatch_guard", "dispatch/cron_dispatch_guard.py") _ecc = _load("dispatch.executor_completion_contract", "dispatch/executor_completion_contract.py") # These are the SAME objects dispatch.core / dispatch.py re-export verbatim # (real dispatch entrypoint surface; proven in test_dispatch_core_real_ # entrypoint via a clean-interpreter subprocess — not a mock). guard_callback_registration = _grd.guard_callback_registration guard_dispatch_with_owner = _grd.guard_dispatch_with_owner build_anu_owned_callback_request = _hlp.build_anu_owned_callback_request CallbackRegistrationBlocked = _valmod.CallbackRegistrationBlocked validate_callback_owner_runtime = _valmod.validate_callback_owner_runtime Callback4Tuple = _ecc.Callback4Tuple EXEC_KEY = "1e41a2324a3ccdd0" # dev6 페룬 executor self key (발사 금지 대상) DEV2_EXEC_KEY = "fedf78d1d09509f5" # dev2 오딘 (본 +49 executor) self key ANU_KEY = "c119085addb0f8b7" # 독립 ANU key (회장 §10/§13) CHAT = "6937032012" GIT_HEAD_PRE = "20456b5f83fc039f2fd6f50f4b94095c29b41bfb" GIT_BRANCH_PRE = "task/task-2553p1-f1-clean-replacement" # byte-0 frozen (narrow +49/+44/+48 FROZEN_SHA256 carve-out — unchanged). REG_SHA = "774d550628410d36962c23a7663c4b6dbf72789de7c7fd940871e9ad8280e5ab" ECC_SHA = "364caa11904285657abd716d78c5493b1f8b519318387d0f864fb6a136dca0b4" CET_SHA = "352ad0f570e55040e7c1e4a32cbfe0f076cbd53529b4db6222a8da1a4bee9cc5" FROZEN_ANCHOR_SHA = ( "83b3e307c8207c76a3e311c408aab4951373bd317896e51687d3007907b0c3d4" ) _CLAUSE = ( "[EXECUTOR COMPLETION CALLBACK — MANDATORY] executor 는 ANU 에 normal " "completion callback cron 을 반드시 발사해야 한다." ) def _sha(rel: str) -> str: return hashlib.sha256((_ROOT / rel).read_bytes()).hexdigest() def _git(*a): return subprocess.run( ["git", "-C", str(_ROOT), *a], capture_output=True, text=True, check=True, ).stdout.strip() def _val(**over): base = dict( task_id="task-2553+49", executor_key=EXEC_KEY, collector_key=ANU_KEY, collector_owner_key=ANU_KEY, collector_role="ANU", normal_collector_cron_id="NC", fallback_callback_cron_id="FB", dispatch_cron_id="D", chat_id=CHAT, prompt_claims_anu_collector=True, entry_path="cokacdir_cron_direct", anu_keys=(ANU_KEY,), ) base.update(over) return validate_callback_owner_runtime(**base) class OwnerValidationRegression(unittest.TestCase): """§8 회장 verbatim regression 1~10 — real validator entrypoint.""" def test_01_executor_to_anu_normal_pass(self): r = _val() self.assertEqual(r.verdict, "PASS") self.assertTrue(r.registration_allowed) self.assertTrue(r.owner_is_independent_anu) def test_02_executor_self_key_normal_fail(self): r = _val(collector_key=EXEC_KEY, collector_owner_key=EXEC_KEY) self.assertEqual(r.verdict, "FAIL") self.assertFalse(r.registration_allowed) self.assertIn("SELF_COLLECTOR_FORBIDDEN", r.classifications) def test_03_executor_self_key_fallback_fail(self): # fallback owned by executor self key (same structural rule). r = _val( collector_key=EXEC_KEY, collector_owner_key=EXEC_KEY, normal_collector_cron_id="NC", fallback_callback_cron_id="FB-self", ) self.assertEqual(r.verdict, "FAIL") self.assertIn("SELF_COLLECTOR_FORBIDDEN", r.classifications) def test_04_prompt_says_anu_but_owner_executor_fail(self): r = _val( collector_key=EXEC_KEY, collector_owner_key=EXEC_KEY, prompt_claims_anu_collector=True, ) self.assertEqual(r.verdict, "FAIL") self.assertTrue( {"CALLBACK_OWNER_MISMATCH", "SELF_COLLECTOR_FORBIDDEN"} & set(r.classifications) ) def test_05_executor_key_eq_collector_key_fail(self): r = _val( executor_key=ANU_KEY, collector_key=ANU_KEY, collector_owner_key=ANU_KEY, ) self.assertEqual(r.verdict, "FAIL") self.assertIn("SELF_COLLECTOR_FORBIDDEN", r.classifications) def test_06_collector_role_missing_fail(self): r = _val(collector_role="") self.assertEqual(r.verdict, "FAIL") self.assertIn("CALLBACK_COLLECTOR_NOT_ANU", r.classifications) def test_07_collector_role_not_anu_fail(self): r = _val(collector_role="executor") self.assertEqual(r.verdict, "FAIL") self.assertIn("CALLBACK_COLLECTOR_NOT_ANU", r.classifications) def test_08_4tuple_normal_cron_missing_fail(self): r = _val(normal_collector_cron_id=None) self.assertEqual(r.verdict, "FAIL") self.assertIn("CALLBACK_4TUPLE_INVALID", r.classifications) def test_09_4tuple_collector_key_missing_fail(self): # empty collector identity -> not an ANU key -> mismatch/blocked. r = _val(collector_key="", collector_owner_key="") self.assertEqual(r.verdict, "FAIL") self.assertFalse(r.registration_allowed) def test_10_4tuple_executor_key_missing_still_blocks_non_anu(self): # executor_key empty but owner is ANU -> still PASS (owner pinned); # owner non-ANU with empty executor -> FAIL (mismatch). ok = _val(executor_key="") self.assertEqual(ok.verdict, "PASS") bad = _val(executor_key="", collector_key="zzz", collector_owner_key="zzz") self.assertEqual(bad.verdict, "FAIL") self.assertIn("CALLBACK_OWNER_MISMATCH", bad.classifications) class RealPathGuardWiring(unittest.TestCase): """§8 reg 21/22/23 — real dispatch / cokacdir-direct / registration helper guard 가 실제로 호출되어 fail-closed 됨을 실 entrypoint 직접 호출로 검증 (mock-only 금지).""" def test_21_dispatch_py_path_guard_called(self): # dispatch.core (= the REAL dispatch.py→dispatch/core.py entrypoint) # exposes guard_dispatch_with_owner; calling it with an executor # self-key collector MUST fail-closed at the dispatch path. t = Callback4Tuple( task_id="task-2553+49", dispatch_cron_id="D", normal_collector_cron_id="NC", fallback_callback_cron_id="FB", ) base, owner, composite = guard_dispatch_with_owner( spec_text=_CLAUSE, entry_path="dispatch.core.main", tuple_=t, executor_key=EXEC_KEY, collector_key=EXEC_KEY, collector_owner_key=EXEC_KEY, collector_role="ANU", dispatch_cron_id="D", normal_collector_cron_id="NC", fallback_callback_cron_id="FB", chat_id=CHAT, prompt_claims_anu_collector=True, anu_keys=(ANU_KEY,), ) self.assertEqual(composite, "FAIL") self.assertEqual(owner.verdict, "FAIL") def test_22_cokacdir_direct_path_guard_called(self): # cron-direct path: the mediated helper MUST refuse to emit argv for # an executor-self-key callback (fail-closed; cron-direct cannot # register an executor-owned callback). bad = build_anu_owned_callback_request( kind="normal", task_id="task-2553+49", executor_key=EXEC_KEY, owner_key=EXEC_KEY, chat_id=CHAT, prompt="ANU Result Collector", at="10m", cron_id="NC", dispatch_cron_id="D", normal_collector_cron_id="NC", fallback_callback_cron_id="FB", entry_path="cokacdir_cron_direct", anu_keys=(ANU_KEY,), ) self.assertEqual(bad.verdict, "FAIL") self.assertIsNone(bad.argv) good = build_anu_owned_callback_request( kind="normal", task_id="task-2553+49", executor_key=EXEC_KEY, owner_key=ANU_KEY, chat_id=CHAT, prompt="ANU Result Collector", at="10m", cron_id="NC", dispatch_cron_id="D", normal_collector_cron_id="NC", fallback_callback_cron_id="FB", entry_path="cokacdir_cron_direct", anu_keys=(ANU_KEY,), ) self.assertEqual(good.verdict, "PASS") self.assertIsNotNone(good.argv) self.assertIn(ANU_KEY, good.argv) self.assertNotIn(EXEC_KEY, good.argv) def test_23_registration_helper_guard_called_and_raises(self): # The registration helper guard MUST raise (structural block) on an # executor-self-key callback, AND PASS for an independent ANU key. with self.assertRaises(CallbackRegistrationBlocked): guard_callback_registration( task_id="task-2553+49", executor_key=EXEC_KEY, collector_key=EXEC_KEY, collector_owner_key=EXEC_KEY, collector_role="ANU", dispatch_cron_id="D", normal_collector_cron_id="NC", fallback_callback_cron_id="FB", chat_id=CHAT, prompt_claims_anu_collector=True, entry_path="cokacdir_cron_direct", anu_keys=(ANU_KEY,), ) val, sg, allowed = guard_callback_registration( task_id="task-2553+49", executor_key=EXEC_KEY, collector_key=ANU_KEY, collector_owner_key=ANU_KEY, collector_role="ANU", dispatch_cron_id="D", normal_collector_cron_id="NC", fallback_callback_cron_id="FB", chat_id=CHAT, prompt_claims_anu_collector=True, entry_path="cokacdir_cron_direct", anu_keys=(ANU_KEY,), ) self.assertTrue(allowed) self.assertEqual(val.verdict, "PASS") self.assertEqual(sg.verdict, "PASS") def test_21b_dispatch_core_real_entrypoint_subprocess(self): # Genuine real-entrypoint proof: in a CLEAN interpreter from repo # root, dispatch.py → dispatch/core.py MUST expose the runtime gates # and block an executor self-key registration (no shadow, no mock). code = ( "from dispatch.core import (guard_callback_registration," "guard_dispatch_runtime_authoritative," "select_runtime_authoritative_verdict);" "from anu_v3.callback_owner_validator import " "CallbackRegistrationBlocked;" "import sys;" "ok=False\n" "try:\n" " guard_callback_registration(task_id='t'," "executor_key='%s',collector_key='%s',collector_owner_key='%s'," "collector_role='ANU',dispatch_cron_id='D'," "normal_collector_cron_id='NC',fallback_callback_cron_id='FB'," "chat_id='%s',prompt_claims_anu_collector=True," "entry_path='cokacdir_cron_direct',anu_keys=('%s',))\n" "except CallbackRegistrationBlocked:\n" " ok=True\n" "sys.exit(0 if ok else 3)" ) % (EXEC_KEY, EXEC_KEY, EXEC_KEY, CHAT, ANU_KEY) p = subprocess.run( [sys.executable, "-c", code], cwd=str(_ROOT), capture_output=True, text=True, timeout=120, ) self.assertEqual( p.returncode, 0, f"real dispatch.core entrypoint did not block self key\n" f"{p.stdout}\n{p.stderr[-1500:]}", ) def test_dev2_executor_self_key_also_blocked(self): # 본 +49 executor 는 dev2 오딘(fedf78d1d09509f5). 그 self key 로 # callback 등록 시도도 구조적으로 차단되어야 한다 (§13: 본 task 가 # 그것을 코드로 강제하므로 dispatch 자체가 모범 준수). r = _val( executor_key=DEV2_EXEC_KEY, collector_key=DEV2_EXEC_KEY, collector_owner_key=DEV2_EXEC_KEY, ) self.assertEqual(r.verdict, "FAIL") self.assertIn("SELF_COLLECTOR_FORBIDDEN", r.classifications) class NoRegressionAndInvariants(unittest.TestCase): """§8 reg 25/26/30 + byte-0/git invariants.""" def test_25_26_plus32_plus44_no_regression(self): # Real no-regression: run the +32 mandatory-callback and +44 4-tuple # registry suites; both MUST still pass (not mocked). for suite in ( "tests/regression/" "test_executor_completion_callback_mandatory_2553plus32.py", "tests/regression/test_callback_4tuple_registry_2553plus44.py", "tests/regression/" "test_callback_owner_enforcement_2553plus49.py", ): p = subprocess.run( [sys.executable, "-m", "pytest", "-q", str(_ROOT / suite)], cwd=str(_ROOT), capture_output=True, text=True, timeout=600, ) self.assertEqual( p.returncode, 0, f"no-regression FAILED for {suite}\n{p.stdout[-2000:]}", ) def test_30_no_raw_credential_exposure(self): # New runtime modules must not embed a raw OWNER PAT / token. import re pat = re.compile(r"gh[ps]_[A-Za-z0-9]{20,}|ghp_[A-Za-z0-9]{36}") for rel in ( "anu_v3/callback_owner_validator.py", "anu_v3/authoritative_verdict_selector.py", "anu_v3/self_collector_guard.py", "anu_v3/writeback_binding_conflict_guard.py", "scripts/verify_callback_owner_contract.py", ): txt = (_ROOT / rel).read_text(encoding="utf-8") self.assertIsNone(pat.search(txt), f"token-like string in {rel}") def test_git_and_byte0_invariants(self): self.assertEqual(_git("rev-parse", "HEAD"), GIT_HEAD_PRE) self.assertEqual( _git("rev-parse", "--abbrev-ref", "HEAD"), GIT_BRANCH_PRE ) self.assertEqual( _sha("anu_v3/callback_4tuple_registry.py"), REG_SHA ) self.assertEqual( _sha("dispatch/executor_completion_contract.py"), ECC_SHA ) self.assertEqual(_sha("anu_v3/callback_event_trigger.py"), CET_SHA) self.assertEqual( _sha("utils/anu_delegation_completion_callback.py"), FROZEN_ANCHOR_SHA, ) def test_schemas_valid_json(self): for s in ( "callback_owner_validation", "authoritative_verdict_selection", "self_collector_guard", "writeback_binding_conflict", ): d = json.loads( (_ROOT / "schemas" / f"{s}.schema.json").read_text( encoding="utf-8" ) ) self.assertEqual(d["$schema"], "http://json-schema.org/draft-07/schema#") if __name__ == "__main__": unittest.main()