# -*- coding: utf-8 -*- """ TEITAN SELF-REPAIR V1 — PATCH EXECUTOR (FASE 2 — AR-74 HARDENED) Applies structured atomic patches exclusively to sandbox candidate files. Enforces canonical directory containment, symlink/junction blocking, manifest membership verification on every read, write, and rollback operation, exact matching, and all-or-nothing rollback on any partial mismatch or invalid instruction. """ from __future__ import annotations import difflib import json import os import shutil from typing import Any, Dict, List, Optional, Tuple from app.remediation.self_repair.sandbox_builder import ( _assert_no_reparse_chain, _is_canonically_within, _is_reparse_point_or_symlink, ) from app.remediation.self_repair.schemas import ( PatchInstruction, PatchOperationType, PatchPlan, PatchResult, SandboxManifest, ) class PatchExecutor: """ Applies structured patches to candidate files within an isolated sandbox. Guarantees all-or-nothing atomicity and strict canonical containment. """ ALLOWED_OPERATIONS = { PatchOperationType.REPLACE_EXACT, PatchOperationType.INSERT_AFTER, PatchOperationType.INSERT_BEFORE, } FORBIDDEN_OPERATIONS = {"DELETE_FILE", "MOVE_FILE", "EXECUTE_ARBITRARY_SHELL", "PRODUCTION_WRITE"} @staticmethod def _resolve_and_verify_sandbox_path(base_dir: str, rel_path: str, role_label: str = "candidate") -> Tuple[str, str]: """ Canonically resolves rel_path inside base_dir. Blocks empty paths, absolute paths, drive letters, '..' traversal, prefix-collision paths, symlinks, and Windows junctions/reparse points. """ raw_str = str(rel_path or "").strip() if not raw_str: raise ValueError(f"Empty {role_label} file path in patch instruction") raw_posix = raw_str.replace("\\", "/") if ( os.path.isabs(raw_str) or (len(raw_str) >= 2 and raw_str[1] == ":") or raw_posix.startswith("/") or ".." in raw_posix.split("/") ): raise ValueError( f"SANDBOX_ESCAPE_BLOCKED: Path '{rel_path}' contains traversal or external absolute reference" ) clean_rel = os.path.normpath(raw_str).replace("\\", "/") if ( not clean_rel or clean_rel in (".", "..") or clean_rel.startswith("../") or ".." in clean_rel.split("/") or os.path.isabs(clean_rel) or ":" in clean_rel ): raise ValueError( f"SANDBOX_ESCAPE_BLOCKED: Normalized path '{clean_rel}' escapes {role_label} directory" ) base_real = os.path.realpath(os.path.abspath(base_dir)) _assert_no_reparse_chain(base_real, clean_rel) full_real = os.path.realpath(os.path.join(base_real, clean_rel)) if not _is_canonically_within(full_real, base_real): raise ValueError( f"SANDBOX_ESCAPE_BLOCKED: Target file '{rel_path}' resolves to '{full_real}' outside {role_label} directory '{base_real}'" ) if os.path.normcase(full_real) == os.path.normcase(base_real): raise ValueError( f"SANDBOX_ESCAPE_BLOCKED: Target file '{rel_path}' resolves to {role_label} directory itself" ) if os.path.lexists(full_real) and _is_reparse_point_or_symlink(full_real): raise ValueError( f"SANDBOX_SYMLINK_OR_JUNCTION_BLOCKED: Target '{full_real}' is a symlink or reparse point" ) return clean_rel, full_real @staticmethod def _verify_modifiable_membership(clean_rel: str, manifest: SandboxManifest) -> Tuple[bool, Optional[str]]: """ Verifies that clean_rel belongs to the candidate's registered modifiable files in manifest and is NOT a read-only verification dependency. """ v_deps = getattr(manifest, "verification_dependencies", []) or [] for vd in v_deps: vd_norm = os.path.normpath(str(vd)).replace("\\", "/") if clean_rel == vd_norm: return ( False, f"VERIFICATION_DEPENDENCY_WRITE_BLOCKED: '{clean_rel}' is a read-only verification dependency", ) modifiable_set: set[str] = set() for fc in getattr(manifest, "files_copied", []) or []: fc_rel = os.path.normpath(str(fc.get("relative_path") or "")).replace("\\", "/") fc_role = str(fc.get("role") or "") if fc_role == "READ_ONLY_VERIFICATION_DEPENDENCY" and fc_rel == clean_rel: return ( False, f"VERIFICATION_DEPENDENCY_WRITE_BLOCKED: '{clean_rel}' is registered as READ_ONLY_VERIFICATION_DEPENDENCY", ) if fc_role in ("MODIFIABLE_FILE", ""): if fc_rel: modifiable_set.add(fc_rel) for sh in getattr(manifest, "source_hashes", []) or []: sh_rel = os.path.normpath(str(getattr(sh, "relative_path", "") or "")).replace("\\", "/") if sh_rel: modifiable_set.add(sh_rel) if not modifiable_set or clean_rel not in modifiable_set: return ( False, f"UNAUTHORIZED_CANDIDATE_FILE_BLOCKED: '{clean_rel}' is not a registered modifiable file in SandboxManifest", ) return True, None def apply_patch_plan( self, plan: PatchPlan, manifest: SandboxManifest, incident_dir: Optional[str] = None, ) -> PatchResult: """ Executes instructions against manifest.candidate_dir. If any instruction fails, rolls back modified candidate files from source_dir. """ applied_instructions: List[Dict[str, Any]] = [] file_diffs: Dict[str, str] = {} files_modified_in_candidate: List[str] = [] for idx, inst in enumerate(plan.instructions): op = inst.operation if isinstance(op, str): try: op = PatchOperationType(op) except ValueError: op = None if op not in self.ALLOWED_OPERATIONS: err_msg = f"Disallowed or unsupported patch operation '{inst.operation}' in instruction {idx}" self._rollback_candidate(files_modified_in_candidate, manifest) res = PatchResult( patch_plan_id=plan.patch_plan_id, incident_id=plan.incident_id, repair_id=plan.repair_id, candidate_id=plan.candidate_id, status="FAIL", applied_instructions=[], failed_instruction=inst.to_dict(), error_message=err_msg, file_diffs={}, ) if incident_dir: self.persist_result(plan, res, incident_dir) return res # 1. Canonical path & symlink/junction security validation before reading try: clean_rel, cand_file_path = self._resolve_and_verify_sandbox_path( manifest.candidate_dir, inst.file_rel_path, role_label="candidate", ) except Exception as path_err: err_msg = f"Target file '{inst.file_rel_path}' resolves outside candidate directory or violates sandbox boundary: {path_err}" self._rollback_candidate(files_modified_in_candidate, manifest) res = PatchResult( patch_plan_id=plan.patch_plan_id, incident_id=plan.incident_id, repair_id=plan.repair_id, candidate_id=plan.candidate_id, status="FAIL", applied_instructions=[], failed_instruction=inst.to_dict(), error_message=err_msg, file_diffs={}, ) if incident_dir: self.persist_result(plan, res, incident_dir) return res # 2. Verify manifest membership (must be a registered MODIFIABLE_FILE) is_member, member_err = self._verify_modifiable_membership(clean_rel, manifest) if not is_member: err_msg = member_err or f"Target file '{clean_rel}' is not authorized in manifest" self._rollback_candidate(files_modified_in_candidate, manifest) res = PatchResult( patch_plan_id=plan.patch_plan_id, incident_id=plan.incident_id, repair_id=plan.repair_id, candidate_id=plan.candidate_id, status="FAIL", applied_instructions=[], failed_instruction=inst.to_dict(), error_message=err_msg, file_diffs={}, ) if incident_dir: self.persist_result(plan, res, incident_dir) return res if not os.path.isfile(cand_file_path) or _is_reparse_point_or_symlink(cand_file_path): err_msg = f"Candidate file not found or invalid file type: '{inst.file_rel_path}'" self._rollback_candidate(files_modified_in_candidate, manifest) res = PatchResult( patch_plan_id=plan.patch_plan_id, incident_id=plan.incident_id, repair_id=plan.repair_id, candidate_id=plan.candidate_id, status="FAIL", applied_instructions=[], failed_instruction=inst.to_dict(), error_message=err_msg, file_diffs={}, ) if incident_dir: self.persist_result(plan, res, incident_dir) return res # 3. Read content safely preserving original line endings (CRLF/LF) try: with open(cand_file_path, "r", encoding="utf-8", newline="") as f: content_before = f.read() except Exception as read_err: err_msg = f"Error reading candidate file: {read_err}" self._rollback_candidate(files_modified_in_candidate, manifest) res = PatchResult( patch_plan_id=plan.patch_plan_id, incident_id=plan.incident_id, repair_id=plan.repair_id, candidate_id=plan.candidate_id, status="FAIL", applied_instructions=[], failed_instruction=inst.to_dict(), error_message=err_msg, file_diffs={}, ) if incident_dir: self.persist_result(plan, res, incident_dir) return res # 4. Apply operation in memory success, new_content, op_err = self._apply_single_op(content_before, op, inst.before, inst.after) if not success: self._rollback_candidate(files_modified_in_candidate, manifest) res = PatchResult( patch_plan_id=plan.patch_plan_id, incident_id=plan.incident_id, repair_id=plan.repair_id, candidate_id=plan.candidate_id, status="FAIL", applied_instructions=[], failed_instruction=inst.to_dict(), error_message=op_err, file_diffs={}, ) if incident_dir: self.persist_result(plan, res, incident_dir) return res # 5. Re-verify canonical boundary and membership immediately before writing try: _, write_target_real = self._resolve_and_verify_sandbox_path( manifest.candidate_dir, clean_rel, role_label="candidate_write", ) is_mem_w, mem_err_w = self._verify_modifiable_membership(clean_rel, manifest) if not is_mem_w: raise RuntimeError(mem_err_w or "Manifest membership failed before write") tmp_path = write_target_real + ".ar74_patch.tmp" if _is_reparse_point_or_symlink(tmp_path): raise RuntimeError("Temporary patch path is a symlink or reparse point") with open(tmp_path, "w", encoding="utf-8", newline="") as f: f.write(new_content) os.replace(tmp_path, write_target_real) if clean_rel not in files_modified_in_candidate: files_modified_in_candidate.append(clean_rel) diff_lines = list(difflib.unified_diff( content_before.splitlines(keepends=True), new_content.splitlines(keepends=True), fromfile=f"a/{clean_rel}", tofile=f"b/{clean_rel}", )) file_diffs[clean_rel] = "".join(diff_lines) applied_instructions.append(inst.to_dict()) except Exception as write_err: err_msg = f"Error writing updated content to candidate: {write_err}" self._rollback_candidate(files_modified_in_candidate, manifest) res = PatchResult( patch_plan_id=plan.patch_plan_id, incident_id=plan.incident_id, repair_id=plan.repair_id, candidate_id=plan.candidate_id, status="FAIL", applied_instructions=[], failed_instruction=inst.to_dict(), error_message=err_msg, file_diffs={}, ) if incident_dir: self.persist_result(plan, res, incident_dir) return res res = PatchResult( patch_plan_id=plan.patch_plan_id, incident_id=plan.incident_id, repair_id=plan.repair_id, candidate_id=plan.candidate_id, status="SUCCESS", applied_instructions=applied_instructions, failed_instruction=None, error_message=None, file_diffs=file_diffs, ) if incident_dir: self.persist_result(plan, res, incident_dir) return res def _apply_single_op( self, content: str, op: PatchOperationType, before: str, after: str, ) -> Tuple[bool, str, Optional[str]]: if not before: return False, content, "Patch 'before' anchor cannot be empty." eff_before = before eff_after = after occurrences = content.count(eff_before) if occurrences == 0 and "\r\n" in content and "\r\n" not in before and "\n" in before: eff_before = before.replace("\n", "\r\n") eff_after = after.replace("\r\n", "\n").replace("\n", "\r\n") occurrences = content.count(eff_before) elif occurrences == 0 and "\r\n" not in content and "\r\n" in before: eff_before = before.replace("\r\n", "\n") eff_after = after.replace("\r\n", "\n") occurrences = content.count(eff_before) if occurrences == 0: return False, content, "Exact anchor 'before' not found in target file." if occurrences > 1: return False, content, f"Ambiguous anchor 'before': found {occurrences} occurrences in target file." if op == PatchOperationType.REPLACE_EXACT: new_content = content.replace(eff_before, eff_after, 1) return True, new_content, None elif op == PatchOperationType.INSERT_AFTER: new_content = content.replace(eff_before, eff_before + eff_after, 1) return True, new_content, None elif op == PatchOperationType.INSERT_BEFORE: new_content = content.replace(eff_before, eff_after + eff_before, 1) return True, new_content, None return False, content, f"Unsupported operation {op}" def _rollback_candidate(self, modified_rel_paths: List[str], manifest: SandboxManifest) -> None: """ Restores modified candidate files from source_dir after verifying canonical containment and modifiable membership for both source and candidate. Never touches any external file. """ for rel in modified_rel_paths: try: clean_rel, src_real = self._resolve_and_verify_sandbox_path( manifest.source_dir, rel, role_label="rollback_source", ) _, cand_real = self._resolve_and_verify_sandbox_path( manifest.candidate_dir, clean_rel, role_label="rollback_candidate", ) is_mem, _ = self._verify_modifiable_membership(clean_rel, manifest) if not is_mem: continue if os.path.isfile(src_real) and not _is_reparse_point_or_symlink(src_real) and not _is_reparse_point_or_symlink(cand_real): shutil.copy2(src_real, cand_real, follow_symlinks=False) except Exception: # Fail closed: never copy to/from an unverified or escaping path continue def persist_result(self, plan: PatchPlan, result: PatchResult, incident_dir: str) -> Tuple[str, str]: """Persists patch_plan.json and patch_result.json.""" os.makedirs(incident_dir, exist_ok=True) plan_p = os.path.join(incident_dir, "patch_plan.json") with open(plan_p, "w", encoding="utf-8") as f: json.dump(plan.to_dict(), f, indent=2, ensure_ascii=False) res_p = os.path.join(incident_dir, "patch_result.json") with open(res_p, "w", encoding="utf-8") as f: json.dump(result.to_dict(), f, indent=2, ensure_ascii=False) return plan_p, res_p