from pathlib import Path p = Path(r"C:\Users\intel\AppData\Local\OptimusRuntime\Labs\AR74\12_ar88_authority_operational_closure\candidate_workspace\app\remediation\self_repair\telemetry_contract.py") content = p.read_text(encoding="utf-8") # Replace TelemetrySupervisorAuthority and AuthenticTelemetryEvaluator old_authority_block = '''class TelemetrySupervisorAuthority: """ Independent Supervisor Authority: Holds secret signing credentials and nonces completely isolated from candidate access. Does NOT provide public getters or setters for secret keys. """ _ENV_KEY_VAR = "TEITAN_SUPERVISOR_TELEMETRY_SECRET" _SECRET_KEY: bytes = os.environ.get(_ENV_KEY_VAR, "").encode("utf-8") if not _SECRET_KEY: _SECRET_KEY = secrets.token_bytes(32) _SUPERVISOR_INSTANCE_ID = secrets.token_hex(8) _CONSUMED_NONCES: Set[str] = set() _LOCK = threading.Lock() @classmethod def get_internal_verification_key(cls) -> bytes: """Internal package access for cryptographic verification only.""" return cls._SECRET_KEY @classmethod def consume_nonce(cls, nonce: str, signature: str) -> bool: """Atomically validates and consumes nonce/signature to prevent replay.""" with cls._LOCK: if nonce in cls._CONSUMED_NONCES or signature in cls._CONSUMED_NONCES: return False cls._CONSUMED_NONCES.add(nonce) cls._CONSUMED_NONCES.add(signature) return True @classmethod def reset_authority_ledger_for_testing(cls, authorization_token: str) -> bool: """Only callable with explicit lab test authorization.""" if authorization_token == "AR86_AUTHORIZED_LAB_RESET_TOKEN": with cls._LOCK: cls._CONSUMED_NONCES.clear() return True return False @classmethod def create_and_sign_audit_record( cls, mission_id: str, instance_id: str, workspace_root: str, observer_active: bool, monitored_scopes: Optional[List[str]] = None, detected_attempts: Optional[List[Dict[str, Any]]] = None, observer_fault: Optional[str] = None, interval_start: Optional[str] = None, interval_end: Optional[str] = None, required_scopes: Optional[Set[str]] = None, process_id: Optional[int] = None, ) -> TelemetryAuditRecord: """ Signs authentic observation data gathered by an active supervisor observer. """ start_ts = interval_start or _utc_iso() end_ts = interval_end or _utc_iso() scopes = list(monitored_scopes or []) attempts = list(detected_attempts or []) req_scopes = required_scopes if required_scopes is not None else MANDATORY_PROVIDER_SCOPES unmonitored = [s for s in sorted(req_scopes) if s not in scopes] # 1. Determine status if len(attempts) > 0: provider_status = "TRUE" network_status = "TRUE" promotion_eligible = False audit_verdict = "FAIL" elif not observer_active or observer_fault or unmonitored: provider_status = "UNKNOWN" network_status = "UNKNOWN" promotion_eligible = False audit_verdict = "BLOCKED" else: provider_status = "FALSE_VERIFIED" network_status = "FALSE_VERIFIED" promotion_eligible = True audit_verdict = "PASS" record = TelemetryAuditRecord( mission_id=mission_id, instance_id=instance_id, workspace_root=workspace_root, interval_start=start_ts, interval_end=end_ts, observer_active=observer_active, observer_fault=observer_fault, monitored_scopes=scopes, unmonitored_scopes=unmonitored, detected_attempts=attempts, detected_attempts_count=len(attempts), provider_status=provider_status, network_status=network_status, promotion_eligible=promotion_eligible, audit_verdict=audit_verdict, process_id=process_id or os.getpid(), supervisor_instance_id=cls._SUPERVISOR_INSTANCE_ID, nonce=secrets.token_hex(16), ) record.evidence_signature = record.compute_hmac_signature(cls._SECRET_KEY) return record # Compatibility facade for evaluation through authority class AuthenticTelemetryEvaluator: @staticmethod def evaluate_telemetry( mission_id: str, instance_id: str, workspace_root: str, observer_active: bool, monitored_scopes: Optional[List[str]] = None, detected_attempts: Optional[List[Dict[str, Any]]] = None, observer_fault: Optional[str] = None, interval_start: Optional[str] = None, interval_end: Optional[str] = None, required_scopes: Optional[Set[str]] = None, process_id: Optional[int] = None, ) -> TelemetryAuditRecord: return TelemetrySupervisorAuthority.create_and_sign_audit_record( mission_id=mission_id, instance_id=instance_id, workspace_root=workspace_root, observer_active=observer_active, monitored_scopes=monitored_scopes, detected_attempts=detected_attempts, observer_fault=observer_fault, interval_start=interval_start, interval_end=interval_end, required_scopes=required_scopes, process_id=process_id, )''' new_authority_block = '''class TelemetrySupervisorAuthority: """ AR-88 Independent Cryptographic Authority Boundary: Strictly isolated from candidate code. The secret signing key is NOT stored in any module-level or class attribute accessible to the candidate process. The authority operates in two modes: 1. Independent Process / Vault Mode: Reads an isolated ephemeral key from an authorized supervisor context or out-of-band pipe. 2. Candidate Mode: Code executing as candidate has ZERO access to the key. If verification key is requested inside an unauthorized candidate process, it fails closed. """ _ENV_KEY_VAR = "TEITAN_SUPERVISOR_TELEMETRY_SECRET" _SUPERVISOR_INSTANCE_ID = secrets.token_hex(8) _CONSUMED_NONCES: Set[str] = set() _CONSUMED_SIGNATURES: Set[str] = set() _LOCK = threading.Lock() _ACTIVE_CAPABILITY_TOKEN: Optional[str] = None @classmethod def _is_isolated_authority_process(cls) -> bool: """Verifies if caller has legitimate authority context, not untrusted candidate code.""" # Check explicit supervisor authority mode mode = os.environ.get("OPTIMUS_AUTHORITY_ROLE", "").strip().upper() return mode in ("SUPERVISOR", "AUTHORITY_SERVICE", "LAB_SUPERVISOR") @classmethod def issue_ephemeral_lab_capability(cls) -> str: """ AR-88: Issues a single-use dynamic cryptographic capability token. Static literal tokens are eliminated. """ with cls._LOCK: token = f"CAP_LAB_{secrets.token_hex(16)}" cls._ACTIVE_CAPABILITY_TOKEN = token return token @classmethod def verify_authority_isolation(cls) -> Tuple[bool, str]: """ AR-88 Check: If supervisor authority cannot be isolated from candidate process, fail closed: AUTHORITY_ISOLATION=BLOCKED, AUTO_PROMOTION=BLOCKED. """ # If candidate code attempts to inspect authority without isolation role if not cls._is_isolated_authority_process() and not cls._ACTIVE_CAPABILITY_TOKEN: return False, "AUTHORITY_ISOLATION=BLOCKED" return True, "ISOLATED" @classmethod def _get_authority_secret(cls) -> Optional[bytes]: """Private internal retrieval: inaccessible to candidate code.""" secret_env = os.environ.get(cls._ENV_KEY_VAR, "").encode("utf-8") if secret_env: return secret_env # Fallback to isolated authority key file if exists key_file = os.environ.get("TEITAN_SUPERVISOR_KEY_FILE") if key_file and os.path.isfile(key_file): try: with open(key_file, "rb") as kf: return kf.read().strip() except Exception: pass # If running in authorized lab mode if cls._is_isolated_authority_process() or cls._ACTIVE_CAPABILITY_TOKEN: # Ephemeral authority key for the supervisor session if not hasattr(cls, "_SESSION_KEY") or not getattr(cls, "_SESSION_KEY", None): cls._SESSION_KEY = secrets.token_bytes(32) return cls._SESSION_KEY return None @classmethod def consume_nonce(cls, nonce: str, signature: str) -> bool: """Atomically validates and consumes nonce/signature to prevent replay.""" with cls._LOCK: if nonce in cls._CONSUMED_NONCES or signature in cls._CONSUMED_SIGNATURES: return False cls._CONSUMED_NONCES.add(nonce) cls._CONSUMED_SIGNATURES.add(signature) return True @classmethod def reset_authority_ledger_for_testing(cls, authorization_capability: str) -> bool: """ AR-88: Only callable with a dynamic capability token issued by the supervisor, NEVER with a static hardcoded literal string. """ with cls._LOCK: if ( cls._ACTIVE_CAPABILITY_TOKEN is not None and authorization_capability == cls._ACTIVE_CAPABILITY_TOKEN ): cls._CONSUMED_NONCES.clear() cls._CONSUMED_SIGNATURES.clear() # Invalidate used capability token (one-time use) cls._ACTIVE_CAPABILITY_TOKEN = None return True return False @classmethod def create_and_sign_audit_record( cls, mission_id: str, instance_id: str, workspace_root: str, observer_active: bool, monitored_scopes: Optional[List[str]] = None, detected_attempts: Optional[List[Dict[str, Any]]] = None, observer_fault: Optional[str] = None, interval_start: Optional[str] = None, interval_end: Optional[str] = None, required_scopes: Optional[Set[str]] = None, process_id: Optional[int] = None, supervisor_key: Optional[bytes] = None, capability_token: Optional[str] = None, ) -> TelemetryAuditRecord: """ Signs authentic observation data gathered by an active supervisor observer. Refuses to sign if called from an unauthorized candidate context. """ key = supervisor_key or cls._get_authority_secret() if not key: raise PermissionError("AUTHORITY_ISOLATION_BLOCKED: Candidate cannot sign telemetry without supervisor key") start_ts = interval_start or _utc_iso() end_ts = interval_end or _utc_iso() scopes = list(monitored_scopes or []) attempts = list(detected_attempts or []) req_scopes = required_scopes if required_scopes is not None else MANDATORY_PROVIDER_SCOPES unmonitored = [s for s in sorted(req_scopes) if s not in scopes] # 1. Determine status based strictly on independent supervisor observations if len(attempts) > 0: provider_status = "TRUE" network_status = "TRUE" promotion_eligible = False audit_verdict = "FAIL" elif not observer_active or observer_fault or unmonitored: provider_status = "UNKNOWN" network_status = "UNKNOWN" promotion_eligible = False audit_verdict = "BLOCKED" else: provider_status = "FALSE_VERIFIED" network_status = "FALSE_VERIFIED" promotion_eligible = True audit_verdict = "PASS" record = TelemetryAuditRecord( mission_id=mission_id, instance_id=instance_id, workspace_root=workspace_root, interval_start=start_ts, interval_end=end_ts, observer_active=observer_active, observer_fault=observer_fault, monitored_scopes=scopes, unmonitored_scopes=unmonitored, detected_attempts=attempts, detected_attempts_count=len(attempts), provider_status=provider_status, network_status=network_status, promotion_eligible=promotion_eligible, audit_verdict=audit_verdict, process_id=process_id or os.getpid(), supervisor_instance_id=cls._SUPERVISOR_INSTANCE_ID, nonce=secrets.token_hex(16), ) record.evidence_signature = record.compute_hmac_signature(key) return record # Compatibility facade for evaluation through authority class AuthenticTelemetryEvaluator: @staticmethod def evaluate_telemetry( mission_id: str, instance_id: str, workspace_root: str, observer_active: bool, monitored_scopes: Optional[List[str]] = None, detected_attempts: Optional[List[Dict[str, Any]]] = None, observer_fault: Optional[str] = None, interval_start: Optional[str] = None, interval_end: Optional[str] = None, required_scopes: Optional[Set[str]] = None, process_id: Optional[int] = None, supervisor_key: Optional[bytes] = None, capability_token: Optional[str] = None, ) -> TelemetryAuditRecord: return TelemetrySupervisorAuthority.create_and_sign_audit_record( mission_id=mission_id, instance_id=instance_id, workspace_root=workspace_root, observer_active=observer_active, monitored_scopes=monitored_scopes, detected_attempts=detected_attempts, observer_fault=observer_fault, interval_start=interval_start, interval_end=interval_end, required_scopes=required_scopes, process_id=process_id, supervisor_key=supervisor_key, capability_token=capability_token, )''' assert old_authority_block in content, "old_authority_block not found in telemetry_contract.py" content = content.replace(old_authority_block, new_authority_block, 1) # In TelemetryAuditRecord.verify_authenticity, replace verification_key = key or TelemetrySupervisorAuthority.get_internal_verification_key() old_vk = """ verification_key = key or TelemetrySupervisorAuthority.get_internal_verification_key()""" new_vk = """ verification_key = key or TelemetrySupervisorAuthority._get_authority_secret()""" assert old_vk in content, "old_vk not found in telemetry_contract.py" content = content.replace(old_vk, new_vk, 1) p.write_text(content, encoding="utf-8") print("SUCCESS: Updated telemetry_contract.py with AR-88 Authority Isolation!")