# -*- coding: utf-8 -*- """ AR-88 Negative Test Suite: Strict Physical Verification of Closures 1 to 5. Directly exercises: A. Telemetry supervisor HMAC key inaccessible to candidate code. B. Candidate process cannot sign telemetry without supervisor key. C. Missing or unverified telemetry blocks auto-promotion. D. All canonical subfolders of C:/xampp/htdocs/digital optimus are strictly protected. E. Tampered rollback snapshot or mismatched SHA256 bytes is rejected. F. Valid authorized rollback succeeds and restores exact baseline bytes. G. Non-existent hook or invalid task does not produce ACCEPTED. H. Isolated environment whitelist lacks production secrets / keys. """ import copy import hashlib import os import shutil import sys import tempfile from pathlib import Path import pytest _LAB_DIR = Path(__file__).resolve().parents[1] _CANDIDATE_WS = str(_LAB_DIR / "candidate_workspace") if _CANDIDATE_WS not in sys.path: sys.path.insert(0, _CANDIDATE_WS) from app.remediation.self_repair.telemetry_contract import ( TelemetrySupervisorAuthority, AuthenticTelemetryEvaluator, TelemetryAuditRecord, ) from app.remediation.self_repair.promotion_guard import PromotionGuard from app.remediation.self_repair.promotion_executor import PromotionExecutor from app.remediation.self_repair.schemas import ( CandidateVerdict, PatchCandidate, PatchOperationType, PromotionExecutionResult, RollbackArtifact, ) from app.remediation.self_repair.live_shadow_hook import ( dispatch_live_shadow_hook, default_canonical_task_state_updater, ) @pytest.fixture def clean_env(): old_role = os.environ.get("OPTIMUS_AUTHORITY_ROLE") yield if old_role is not None: os.environ["OPTIMUS_AUTHORITY_ROLE"] = old_role else: os.environ.pop("OPTIMUS_AUTHORITY_ROLE", None) def test_scenario_a_candidate_cannot_obtain_supervisor_key(clean_env): """Scenario A: Candidate process cannot obtain supervisor HMAC key.""" os.environ.pop("OPTIMUS_AUTHORITY_ROLE", None) os.environ.pop("TEITAN_SUPERVISOR_KEY", None) # Internal secret key retrieval must return None or block assert not hasattr(TelemetrySupervisorAuthority, "get_internal_verification_key") assert not hasattr(TelemetrySupervisorAuthority, "get_supervisor_telemetry_key") # Attempting to sign without supervisor role raises PermissionError with pytest.raises(PermissionError) as exc_info: TelemetrySupervisorAuthority.create_and_sign_audit_record( mission_id="m_candidate_unauth", instance_id="inst_1", workspace_root="C:\\temp\\ws", observer_active=True, monitored_scopes=["URLLIB_HTTP", "RAW_SOCKET"], detected_attempts=[], ) assert "AUTHORITY_ISOLATION_BLOCKED" in str(exc_info.value) def test_scenario_b_missing_telemetry_blocks_promotion(clean_env): """Scenario B: Missing or invalid telemetry blocks auto-promotion.""" temp_ws = tempfile.mkdtemp(prefix="ar88_test_ws_") try: guard = PromotionGuard(temp_ws) # validate_provider_safety strictly evaluates require_authenticated_provenance is_safe = guard.validate_provider_safety( provider_called="FALSE_VERIFIED", network_used="FALSE_VERIFIED", telemetry_audit=None, # Missing telemetry audit record require_authenticated_provenance=True, ) assert is_safe is False finally: shutil.rmtree(temp_ws, ignore_errors=True) def test_scenario_c_canonical_subfolders_protected(): """Scenario C: All canonical production subfolders strictly protected against unauthorized promotion.""" guard = PromotionGuard() prod_root = r"C:\xampp\htdocs\Digital Optimus" protected_targets = [ r"app\remediation\self_repair\telemetry_contract.py", r"artifacts\brain_host\brain_host_tls.py", r"continuity.db", r"teitan_runtime\core.py", ] for pt in protected_targets: is_crit = guard.is_target_critical(pt, prod_root) is_allowed = guard.is_target_allowed(pt, prod_root) assert is_crit is True or is_allowed is False, f"Path {pt} in {prod_root} should be blocked!" def test_scenario_d_tampered_rollback_artifact_rejected(): """Scenario D: Tampered rollback artifact or mismatched SHA256 rejected.""" executor = PromotionExecutor() temp_dir = tempfile.mkdtemp(prefix="ar88_rollback_ws_") try: target_f = os.path.join(temp_dir, "test_file.py") orig_content = b"ORIGINAL_BASELINE_CONTENT" tampered_content = b"TAMPERED_BASELINE_CONTENT" with open(target_f, "wb") as f: f.write(tampered_content) orig_sha = hashlib.sha256(orig_content).hexdigest().upper() cand_sha = hashlib.sha256(tampered_content).hexdigest().upper() art = RollbackArtifact( promotion_id="promo_fake", target_path=target_f, original_hash=orig_sha, candidate_hash=cand_sha, original_bytes_snapshot=orig_content, # snapshot has orig_content ) # Tamper snapshot bytes in artifact art.original_bytes_snapshot = tampered_content # Rollback attempt with tampered snapshot bytes res = executor.rollback(art) assert res is False finally: shutil.rmtree(temp_dir, ignore_errors=True) def test_scenario_e_authorized_rollback_verifies_sha256(): """Scenario E: Authorized rollback verifies exact SHA256 bytes and restores file.""" executor = PromotionExecutor() temp_dir = tempfile.mkdtemp(prefix="ar88_rollback_ws2_") try: target_f = os.path.join(temp_dir, "test_file2.py") orig_content = b"EXACT_BASELINE_AUTHORIZED_CONTENT" with open(target_f, "wb") as f: f.write(b"MODIFIED_CONTENT") orig_sha = hashlib.sha256(orig_content).hexdigest().upper() cand_sha = hashlib.sha256(b"MODIFIED_CONTENT").hexdigest().upper() art = RollbackArtifact( promotion_id="promo_auth", target_path=target_f, original_hash=orig_sha, candidate_hash=cand_sha, original_bytes_snapshot=orig_content, ) res = executor.rollback(art) assert res is True # Verify target restored exactly with open(target_f, "rb") as f: assert f.read() == orig_content finally: shutil.rmtree(temp_dir, ignore_errors=True) def test_scenario_f_non_existent_hook_fails_safely(): """Scenario F: Non-dict or empty payload does not crash and returns False/None.""" res = default_canonical_task_state_updater({}) assert res is False res2 = default_canonical_task_state_updater({"task_id": "non_existent_task_99999"}) assert res2 is False res3 = dispatch_live_shadow_hook(None) assert res3 is None def test_scenario_g_whitelisted_env_lacks_prod_secrets(): """Scenario G: Environment whitelisting filters out sensitive keys.""" sys.path.insert(0, str(_LAB_DIR)) from test_isolated_brain_route import build_whitelisted_test_env temp_d = tempfile.mkdtemp(prefix="ar88_env_") try: os.environ["SECRET_PROD_KEY"] = "TOP_SECRET_123" os.environ["OPTIMUS_PROD_PASSWORD"] = "CONFIDENTIAL" w_env = build_whitelisted_test_env(temp_d, 5555, "devices.json") assert "SECRET_PROD_KEY" not in w_env assert "OPTIMUS_PROD_PASSWORD" not in w_env assert w_env["BRAIN_HOST_PORT"] == "5555" assert w_env["OPTIMUS_RUN_MODE"] == "LAB_TEST" finally: os.environ.pop("SECRET_PROD_KEY", None) os.environ.pop("OPTIMUS_PROD_PASSWORD", None) shutil.rmtree(temp_d, ignore_errors=True)