from pathlib import Path # 1. Update promotion_guard.py guard_path = Path(r"C:\Users\intel\AppData\Local\OptimusRuntime\Labs\AR74\12_ar88_authority_operational_closure\candidate_workspace\app\remediation\self_repair\promotion_guard.py") guard_content = guard_path.read_text(encoding="utf-8") old_guard_check = """ # AR-86 Trust Closure: # In production environments (or when telemetry_audit is provided), authenticated cryptographic # provenance is strictly required. An explicit lab authority token can only be honored if the # target workspace is not production. explicit_lab_auth = details.get("explicit_lab_authority") is_authorized_lab_bypass = (explicit_lab_auth == "AR86_LAB_PROMOTION_AUTHORITY") req_auth = details.get("require_authenticated_provenance") if req_auth is None: # In Phase 5A unit tests where build_promotion_decision is called directly with synthetic flags # and no telemetry_audit or explicit token is set, require_authenticated_provenance is False # ONLY IF tele_rec is None AND caller did not specify require_authenticated_provenance. # But if require_authenticated_provenance is explicitly passed or in production, enforce True! req_auth = True if tele_rec is not None else False elif req_auth is False and not is_authorized_lab_bypass and tele_rec is not None: # Caller attempted to bypass provenance while passing a telemetry record: forbidden req_auth = True""" new_guard_check = """ # AR-88 Inconditional Protection: # 1. Canonical protection of all production subfolders. w_canon = os.path.realpath(os.path.abspath(str(w_root))).replace("\\\\", "/").lower() prod_canon = "c:/xampp/htdocs/digital optimus" is_in_prod_tree = (w_canon == prod_canon or w_canon.startswith(prod_canon + "/")) # 2. Dynamic capability token check (no static literal string bypass). lab_cap = details.get("lab_capability_token") from app.remediation.self_repair.telemetry_contract import TelemetrySupervisorAuthority is_valid_cap = False if lab_cap and not is_in_prod_tree: # Verify dynamically against the supervisor authority active token if getattr(TelemetrySupervisorAuthority, "_ACTIVE_CAPABILITY_TOKEN", None) == lab_cap: is_valid_cap = True # 3. Cryptographic provenance is UNCONDITIONALLY required for auto-promotion. # No default fallback to False. if is_in_prod_tree: req_auth = True else: req_auth = details.get("require_authenticated_provenance", True) if not is_valid_cap: # Without valid supervisor capability token, cannot bypass provenance check req_auth = True""" assert old_guard_check in guard_content, "old_guard_check not found in promotion_guard.py" guard_content = guard_content.replace(old_guard_check, new_guard_check, 1) # Also check target_p canonical resolution in is_target_allowed and is_target_critical old_guard_target_crit = """ # Check if absolute path resolves into project root's critical areas if workspace_root or self.workspace_root: w_root = Path(workspace_root or self.workspace_root).resolve() try: target_p = Path(target_file) if not target_p.is_absolute(): target_p = (w_root / target_p).resolve() else: target_p = target_p.resolve()""" new_guard_target_crit = """ # Check if absolute path resolves into project root's critical areas (Canonical Protection AR-88) prod_canon = "c:/xampp/htdocs/digital optimus" target_canon = os.path.realpath(os.path.abspath(str(target_file))).replace("\\\\", "/").lower() if target_canon == prod_canon or target_canon.startswith(prod_canon + "/"): # If target resolves anywhere into production tree, it is protected return True if workspace_root or self.workspace_root: w_root = Path(workspace_root or self.workspace_root).resolve() try: target_p = Path(target_file) if not target_p.is_absolute(): target_p = (w_root / target_p).resolve() else: target_p = target_p.resolve()""" assert old_guard_target_crit in guard_content, "old_guard_target_crit not found in promotion_guard.py" guard_content = guard_content.replace(old_guard_target_crit, new_guard_target_crit, 1) guard_path.write_text(guard_content, encoding="utf-8") print("SUCCESS: Updated promotion_guard.py with AR-88 canonical protection!") # 2. Update promotion_executor.py exec_path = Path(r"C:\Users\intel\AppData\Local\OptimusRuntime\Labs\AR74\12_ar88_authority_operational_closure\candidate_workspace\app\remediation\self_repair\promotion_executor.py") exec_content = exec_path.read_text(encoding="utf-8") old_exec_canon = """ ws_str = str(w_root).replace("\\\\", "/").lower() is_prod_workspace = (ws_str in ("c:/xampp/htdocs/digital optimus", "c:/xampp/htdocs/digital optimus/")) # AR-86: Eliminate path heuristics (temp/tmp/pytest). # In production, require_authenticated_provenance is unconditionally True. # In isolated test suites with no telemetry_audit, if caller does not require authenticated provenance, # fallback is permitted ONLY if not in production and not targeting production files. explicit_lab_auth = rollback_plan.get("explicit_lab_authority") if isinstance(rollback_plan, dict) else None is_authorized_lab = (explicit_lab_auth == "AR86_LAB_PROMOTION_AUTHORITY") eff_req_auth = require_authenticated_provenance if eff_req_auth is None: if is_prod_workspace: eff_req_auth = True else: eff_req_auth = True if telemetry_audit is not None else False elif eff_req_auth is False and is_prod_workspace: # Caller attempted to pass False inside production workspace: strictly prohibited eff_req_auth = True details_payload: Dict[str, Any] = { "promotion_id": promo_id, "incident_id": incident_id, "mission_id": incident_id, "telemetry_audit": telemetry_audit, "require_authenticated_provenance": eff_req_auth, "explicit_lab_authority": explicit_lab_auth if is_authorized_lab else None, }""" new_exec_canon = """ ws_canon = os.path.realpath(os.path.abspath(str(w_root))).replace("\\\\", "/").lower() prod_canon = "c:/xampp/htdocs/digital optimus" is_prod_workspace = (ws_canon == prod_canon or ws_canon.startswith(prod_canon + "/")) # AR-88: Canonical Protection & Dynamic Supervisor Capability # No static token strings ("AR86_LAB_PROMOTION_AUTHORITY"). lab_cap = rollback_plan.get("lab_capability_token") if isinstance(rollback_plan, dict) else None from app.remediation.self_repair.telemetry_contract import TelemetrySupervisorAuthority is_authorized_lab = False if lab_cap and not is_prod_workspace: if getattr(TelemetrySupervisorAuthority, "_ACTIVE_CAPABILITY_TOKEN", None) == lab_cap: is_authorized_lab = True # In AR-88, require_authenticated_provenance is strictly True by default eff_req_auth = require_authenticated_provenance if eff_req_auth is None: eff_req_auth = False if is_authorized_lab else True elif eff_req_auth is False and (is_prod_workspace or not is_authorized_lab): eff_req_auth = True details_payload: Dict[str, Any] = { "promotion_id": promo_id, "incident_id": incident_id, "mission_id": incident_id, "telemetry_audit": telemetry_audit, "require_authenticated_provenance": eff_req_auth, "lab_capability_token": lab_cap if is_authorized_lab else None, }""" assert old_exec_canon in exec_content, "old_exec_canon not found in promotion_executor.py" exec_content = exec_content.replace(old_exec_canon, new_exec_canon, 1) # In rollback(): Add authorized rollback validation (AR-88 Requirement 3) old_rollback_start = """ def rollback(self, artifact: RollbackArtifact) -> bool: \"\"\" Executes an immediate rollback restoring the target file to its exact original state. Verifies that restored hash strictly matches original_hash. \"\"\" artifact.rollback_triggered = True if artifact.original_bytes_snapshot is None: artifact.error = "ROLLBACK_BYTES_MISSING" artifact.rollback_completed = False artifact.production_restored = False artifact.original_hash_restored = False logger.error("[PromotionExecutor] Rollback failed: original_bytes_snapshot is None") return False""" new_rollback_start = """ def rollback(self, artifact: RollbackArtifact, target_override_path: Optional[str] = None) -> bool: \"\"\" AR-88 Authorized Rollback: 1. Verifies promotion identity and artifact integrity. 2. Validates canonical path authorization: rejects unauthorized paths or production targets. 3. Validates original bytes and hash match prior to write. 4. Atomic replace with verified post-rollback hash check. \"\"\" artifact.rollback_triggered = True if not artifact or not getattr(artifact, "promotion_id", None): logger.error("[PromotionExecutor] Rollback rejected: invalid or arbitrary artifact") if artifact: artifact.error = "ARBITRARY_ARTIFACT_REJECTED" return False # Verify artifact ownership if tracked if artifact.promotion_id in self.rollback_artifacts: known_art = self.rollback_artifacts[artifact.promotion_id] if known_art.original_hash != artifact.original_hash: artifact.error = "ARTIFACT_OWNERSHIP_MISMATCH" logger.error("[PromotionExecutor] Rollback rejected: original_hash mismatch with tracked artifact") return False effective_target = target_override_path or artifact.target_path target_canon = os.path.realpath(os.path.abspath(str(effective_target))).replace("\\\\", "/").lower() prod_canon = "c:/xampp/htdocs/digital optimus" # Unauthorized target protection: strictly block rollback targeting production unless authorized if target_canon == prod_canon or target_canon.startswith(prod_canon + "/"): artifact.error = "UNAUTHORIZED_PRODUCTION_ROLLBACK_BLOCKED" artifact.rollback_completed = False artifact.production_restored = False logger.error("[PromotionExecutor] Rollback blocked: target resolves to protected production path %s", target_canon) return False if artifact.original_bytes_snapshot is None: artifact.error = "ROLLBACK_BYTES_MISSING" artifact.rollback_completed = False artifact.production_restored = False artifact.original_hash_restored = False logger.error("[PromotionExecutor] Rollback failed: original_bytes_snapshot is None") return False # Verify physical integrity of snapshot bytes before restoring snap_hash = hashlib.sha256(artifact.original_bytes_snapshot).hexdigest().upper() if snap_hash != artifact.original_hash: artifact.error = "SNAPSHOT_BYTES_HASH_MISMATCH" artifact.rollback_completed = False logger.error("[PromotionExecutor] Rollback failed: snapshot bytes hash %s != original_hash %s", snap_hash, artifact.original_hash) return False""" assert old_rollback_start in exec_content, "old_rollback_start not found in promotion_executor.py" exec_content = exec_content.replace(old_rollback_start, new_rollback_start, 1) # In rollback(): update target_p to effective_target old_rb_target_p = """ target_p = Path(artifact.target_path)""" new_rb_target_p = """ target_p = Path(effective_target)""" assert old_rb_target_p in exec_content, "old_rb_target_p not found" exec_content = exec_content.replace(old_rb_target_p, new_rb_target_p, 1) exec_path.write_text(exec_content, encoding="utf-8") print("SUCCESS: Updated promotion_executor.py with AR-88 authorized rollback and canonical protection!")