"""Executable permission binding for the first-customer invite controller.""" from __future__ import annotations import sys from dataclasses import dataclass from pathlib import Path from gateway.platforms import telegram_customer_bootstrap as bootstrap from first_customer_invite_contract import ( CANDIDATE, HERMES_WHEEL, HERMES_WHEEL_SHA256, MODULE_SHA256, PROFILE_WHEEL, PROFILE_WHEEL_SHA256, TAGGED_SEAL, TAGGED_SEAL_SHA256, ControllerError, json_object, require_private_file, require_service_stopped, sha256_file, ) @dataclass(frozen=True, slots=True) class Permission: candidate_digest: str hermes_wheel_sha256: str profile_wheel_sha256: str tagged_seal_sha256: str profile: Path service: str expected_python: Path unit_path: Path unit_sha256: str draft_sha256: str module_sha256: str controller_sha256: str contract_sha256: str operations_sha256: str permission_module_sha256: str approval_event_path: Path approval_event_sha256: str launch_authorization_path: Path launch_authorization_sha256: str maximum_invites: int preflight_receipt_path: Path prepare_receipt_path: Path handoff_path: Path verify_receipt_path: Path install_receipt_path: Path provider_receipt_path: Path start_receipt_path: Path allowed_modes: frozenset[str] def load_permission(path: Path) -> Permission: require_private_file(path) value = json_object(path) modes = value.get("allowed_modes") maximum = value.get("maximum_invites") if ( not isinstance(modes, list) or any(not isinstance(item, str) for item in modes) or type(maximum) is not int ): raise ControllerError("permission seal is malformed") try: return Permission( candidate_digest=str(value["candidate_digest"]), hermes_wheel_sha256=str(value["hermes_wheel_sha256"]), profile_wheel_sha256=str(value["profile_wheel_sha256"]), tagged_seal_sha256=str(value["tagged_seal_sha256"]), profile=Path(str(value["profile"])).resolve(), service=str(value["service"]), expected_python=Path(str(value["expected_python"])), unit_path=Path(str(value["unit_path"])).resolve(), unit_sha256=str(value["unit_sha256"]), draft_sha256=str(value["draft_sha256"]), module_sha256=str(value["module_sha256"]), controller_sha256=str(value["controller_sha256"]), contract_sha256=str(value["contract_sha256"]), operations_sha256=str(value["operations_sha256"]), permission_module_sha256=str(value["permission_module_sha256"]), approval_event_path=Path(str(value["approval_event_path"])).resolve(), approval_event_sha256=str(value["approval_event_sha256"]), launch_authorization_path=Path( str(value["launch_authorization_path"]) ).resolve(), launch_authorization_sha256=str(value["launch_authorization_sha256"]), maximum_invites=maximum, preflight_receipt_path=Path( str(value["preflight_receipt_path"]) ).resolve(), prepare_receipt_path=Path(str(value["prepare_receipt_path"])).resolve(), handoff_path=Path(str(value["handoff_path"])).resolve(), verify_receipt_path=Path(str(value["verify_receipt_path"])).resolve(), install_receipt_path=Path(str(value["install_receipt_path"])).resolve(), provider_receipt_path=Path( str(value["provider_receipt_path"]) ).resolve(), start_receipt_path=Path(str(value["start_receipt_path"])).resolve(), allowed_modes=frozenset( item for item in modes if isinstance(item, str) ), ) except KeyError as exc: raise ControllerError("permission seal is malformed") from exc def verify_permission( mode: str, profile: Path, draft: Path, permission_path: Path, controller_path: Path, contract_path: Path, operations_path: Path, permission_module_path: Path, receipt_path: Path, handoff_path: Path | None, ) -> Permission: permission = load_permission(permission_path) expected = ( permission.candidate_digest == CANDIDATE and permission.hermes_wheel_sha256 == HERMES_WHEEL_SHA256 and permission.profile_wheel_sha256 == PROFILE_WHEEL_SHA256 and permission.tagged_seal_sha256 == TAGGED_SEAL_SHA256 and permission.profile == profile.resolve() and permission.expected_python == Path(sys.executable) and permission.unit_sha256 == sha256_file(permission.unit_path) and permission.draft_sha256 == sha256_file(draft) and permission.module_sha256 == MODULE_SHA256 and permission.controller_sha256 == sha256_file(controller_path) and permission.contract_sha256 == sha256_file(contract_path) and permission.operations_sha256 == sha256_file(operations_path) and permission.permission_module_sha256 == sha256_file(permission_module_path) and permission.approval_event_sha256 == sha256_file(permission.approval_event_path) and permission.launch_authorization_sha256 == sha256_file(permission.launch_authorization_path) and permission.maximum_invites == 1 and mode in permission.allowed_modes ) if not expected: raise ControllerError("permission seal binding mismatch") authorization = json_object(permission.launch_authorization_path) if ( authorization.get("status") != "AUTHORIZED_ONE_FIRST_CUSTOMER_LINK" or authorization.get("candidate_digest") != CANDIDATE or authorization.get("approval_event_sha256") != permission.approval_event_sha256 or authorization.get("maximum_invites") != 1 ): raise ControllerError("launch authorization binding mismatch") expected_receipt = { "preflight": permission.preflight_receipt_path, "prepare": permission.prepare_receipt_path, "verify": permission.verify_receipt_path, }[mode] if receipt_path.resolve() != expected_receipt: raise ControllerError("receipt output path is not authorized") if mode in {"prepare", "verify"} and ( handoff_path is None or handoff_path.resolve() != permission.handoff_path ): raise ControllerError("private handoff path is not authorized") if receipt_path.exists() or (mode == "prepare" and permission.handoff_path.exists()): raise ControllerError("authorized output path is already consumed") if ( sha256_file(HERMES_WHEEL) != HERMES_WHEEL_SHA256 or sha256_file(PROFILE_WHEEL) != PROFILE_WHEEL_SHA256 or sha256_file(TAGGED_SEAL) != TAGGED_SEAL_SHA256 or sha256_file(Path(bootstrap.__file__).resolve()) != MODULE_SHA256 ): raise ControllerError("loaded candidate bytes mismatch") installed = json_object(permission.install_receipt_path) provider = json_object(permission.provider_receipt_path) installed_count = installed.get("installed_regular_file_count") if ( installed.get("schema") != "dualcoach-first-customer-install-receipt-v1" or installed.get("status") != "PASS_INSTALLED" or installed.get("candidate_digest") != CANDIDATE or installed.get("bootstrap_module_sha256") != MODULE_SHA256 or installed.get("invite_ttl_seconds") != 86_400 or installed.get("venv_mode") != "0700" or installed.get("link_mode") != "copy" or installed.get("installer") != "pip" or type(installed_count) is not int or installed_count <= 0 or installed.get("unit_sha256") != sha256_file(permission.unit_path) or provider.get("schema") != "dualcoach-first-customer-provider-receipt-v1" or provider.get("status") != "PASS_PROVIDER_READY" or provider.get("candidate_digest") != CANDIDATE or provider.get("install_receipt_sha256") != sha256_file(permission.install_receipt_path) or provider.get("unit_sha256") != sha256_file(permission.unit_path) ): raise ControllerError("deployment readiness binding mismatch") if mode == "verify": started = json_object(permission.start_receipt_path) if ( started.get("schema") != "dualcoach-first-customer-start-receipt-v1" or started.get("status") != "PASS_STARTED" or started.get("candidate_digest") != CANDIDATE ): raise ControllerError("gateway start binding mismatch") unit = permission.unit_path.read_text(encoding="utf-8") if f"ExecStart={permission.expected_python} -m hermes_cli.main" not in unit: raise ControllerError("service unit runtime mismatch") if mode in {"preflight", "prepare"}: require_service_stopped(permission.service) return permission