""" Optimus Research Center - Mission 22D Portable Package Hygiene Test Suite Tests package security, forbidden file exclusions, path traversal defenses, and independent installation integrity in temporary environments. """ import os import sys import zipfile import tempfile import shutil import hashlib from pathlib import Path import pytest BASE_DIR = Path(__file__).resolve().parent.parent sys.path.insert(0, str(BASE_DIR)) from scripts.package_laboratory import ( validate_file_hygiene, select_package_files, audit_zip_archive, create_portable_package, FORBIDDEN_DIR_NAMES, FORBIDDEN_EXTENSIONS, FORBIDDEN_EXACT_NAMES ) from scripts.test_portable_installation import ( audit_archive_for_extraction, safe_extract_all, run_smoke_test_in_temp_dir ) def compute_sha256(file_path: Path) -> str: h = hashlib.sha256() with open(file_path, "rb") as f: while chunk := f.read(65536): h.update(chunk) return h.hexdigest() # -------------------------------------------------------------------------- # PRUEBA A: Inclusiones Prohibidas en Paquete Existente / Generado # -------------------------------------------------------------------------- def test_package_hygiene_no_forbidden_files_in_distribution(): """Prueba A: Verifica que el paquete generado en dist no contenga ningún archivo prohibido.""" dist_dir = BASE_DIR / "dist" zip_path = dist_dir / "optimus_research_center_portable_v1.0.zip" # Generate package if not present if not zip_path.exists(): create_portable_package(output_dir=dist_dir) assert zip_path.exists(), "Portable package zip does not exist in dist/" with zipfile.ZipFile(zip_path, "r") as zf: entries = zf.namelist() for entry in entries: norm = entry.replace("\\", "/") leaf = Path(norm).name.lower() # 1. No .env files of any kind assert not leaf.startswith(".env"), f"Forbidden .env file packaged: {entry}" assert ".env" not in [p.lower() for p in Path(norm).parts], f".env found in path: {entry}" # 2. No SQLite / database files assert not leaf.endswith((".db", ".sqlite", ".sqlite3", ".db-wal", ".db-shm")), f"Database file packaged: {entry}" # 3. No secrets or credentials assert not leaf.endswith((".key", ".pem", ".crt", ".pfx", ".p12", ".id_rsa")), f"Secret/key packaged: {entry}" # 4. No cache or bytecode assert "__pycache__" not in norm, f"__pycache__ found in package: {entry}" assert not leaf.endswith((".pyc", ".pyo", ".pyd")), f"Bytecode found in package: {entry}" # 5. No logs or PID assert not leaf.endswith((".log", ".pid")), f"Log/PID found in package: {entry}" # -------------------------------------------------------------------------- # PRUEBA B: Exclusión Rigurosa de Casos Sintéticos Nuevos (en Temp Dir) # -------------------------------------------------------------------------- def test_hygiene_filter_rejects_synthetic_secrets_and_databases(tmp_path): """ Prueba B: Crea una copia temporal sintética con múltiples archivos prohibidos y verifica que el empaquetador los excluya o rechace categóricamente. NUNCA crea estos archivos en el proyecto operativo. """ # Create mock project structure (tmp_path / "app").mkdir() (tmp_path / "app" / "main.py").write_text("# main", encoding="utf-8") (tmp_path / "app" / "config.py").write_text("# config", encoding="utf-8") (tmp_path / "storage" / "db").mkdir(parents=True) (tmp_path / "config").mkdir() # Create synthetic forbidden files forbidden_files = [ tmp_path / ".env", tmp_path / ".env.example", tmp_path / ".env.production", tmp_path / "config" / ".env.local", tmp_path / "storage" / "research_center.db", tmp_path / "storage" / "db" / "research_center.db", tmp_path / "storage" / "db" / "research_center.db-wal", tmp_path / "storage" / "db" / "research_center.db-shm", tmp_path / "app" / "server.key", tmp_path / "app" / "cert.pem", tmp_path / "app" / "secret_token.txt", tmp_path / "app" / "debug.log", tmp_path / "app" / "server.pid", tmp_path / "app" / "__pycache__" / "cached.pyc" ] (tmp_path / "app" / "__pycache__").mkdir(exist_ok=True) for f in forbidden_files: f.write_text("DUMMY_SENSITIVE_CONTENT", encoding="utf-8") # Scrutinize each forbidden file through validate_file_hygiene for f in forbidden_files: rel_p = f.relative_to(tmp_path) allowed, reason = validate_file_hygiene(rel_p, tmp_path) assert not allowed, f"Hygiene validator improperly ALLOWED forbidden file: {rel_p} (Reason: {reason})" # Verify select_package_files excludes all of them selected = select_package_files(tmp_path) selected_rel_paths = {str(item[1]).replace("\\", "/") for item in selected} for f in forbidden_files: rel_str = str(f.relative_to(tmp_path)).replace("\\", "/") assert rel_str not in selected_rel_paths, f"select_package_files included forbidden file: {rel_str}" # -------------------------------------------------------------------------- # PRUEBA C: Rutas Maliciosas (Traversal, Absolutas, Symlinks) # -------------------------------------------------------------------------- def test_audit_archive_rejects_malicious_paths(tmp_path): """ Prueba C: Genera ZIPs sintéticos con traversal, rutas absolutas y nombres duplicados y verifica que tanto audit_zip_archive como audit_archive_for_extraction los rechacen. """ # 1. ZIP with path traversal (..) bad_zip_traversal = tmp_path / "malicious_traversal.zip" with zipfile.ZipFile(bad_zip_traversal, "w") as zf: zf.writestr("app/../../escaped.txt", "MALICIOUS") with pytest.raises(ValueError, match="Insecure path detected|Directory traversal"): audit_archive_for_extraction(bad_zip_traversal) valid, errors = audit_zip_archive(bad_zip_traversal) assert not valid assert any("traversal" in err.lower() for err in errors) # 2. ZIP with absolute path bad_zip_abs = tmp_path / "malicious_absolute.zip" with zipfile.ZipFile(bad_zip_abs, "w") as zf: zf.writestr("/etc/passwd", "MALICIOUS") with pytest.raises(ValueError, match="Insecure path detected|Absolute path"): audit_archive_for_extraction(bad_zip_abs) valid, errors = audit_zip_archive(bad_zip_abs) assert not valid assert any("absolute" in err.lower() for err in errors) # 3. Safe extraction blocks traversal extract_target = tmp_path / "safe_extract_target" extract_target.mkdir() with pytest.raises(ValueError, match="Attempted path traversal extraction|Insecure path"): safe_extract_all(bad_zip_traversal, extract_target) # -------------------------------------------------------------------------- # PRUEBA D: Componentes Necesarios Presentes # -------------------------------------------------------------------------- def test_package_contains_all_necessary_scientific_components(): """ Prueba D: Comprobar que se incluyen el coordinador, los laboratorios, el registro de capacidades y los adaptadores desconectados. """ dist_dir = BASE_DIR / "dist" zip_path = dist_dir / "optimus_research_center_portable_v1.0.zip" if not zip_path.exists(): create_portable_package(output_dir=dist_dir) required_components = [ "app/main.py", "app/config.py", "app/core/coordinator.py", "app/core/database.py", "app/core/models.py", "app/core/lab_contract.py", "app/core/lab_adapters.py", "app/core/lab_registry.py", "app/core/adapters/titan_adapter.py", "app/core/adapters/neurolab_adapter.py", "app/lab_algorithms/router.py", "app/lab_algorithms/service.py", "app/lab_ai/router.py", "app/lab_ai/service.py", "run.py", "run_scenarios.py", "requirements.txt", "ORC22D_ENVIRONMENT_CONFIGURATION.md" ] with zipfile.ZipFile(zip_path, "r") as zf: entry_set = set(zf.namelist()) for req in required_components: assert req in entry_set, f"Required component missing from portable package: {req}" # -------------------------------------------------------------------------- # PRUEBA E: Portabilidad Real en Carpeta con Espacios # -------------------------------------------------------------------------- def test_portable_installation_end_to_end_in_temp_dir(): """ Prueba E: Extrae el paquete en una carpeta temporal con espacios en el nombre, inicializa datos limpios y ejecuta el circuito científico completo. """ dist_dir = BASE_DIR / "dist" zip_path = dist_dir / "optimus_research_center_portable_v1.0.zip" if not zip_path.exists(): create_portable_package(output_dir=dist_dir) success = run_smoke_test_in_temp_dir(zip_path) assert success is True, "Portable installation smoke test failed!" # -------------------------------------------------------------------------- # PRUEBA G: Adaptadores Desconectados # -------------------------------------------------------------------------- def test_adapters_strictly_disconnected_in_package(): """ Prueba G: Confirma que Titan y NeuroLab permanecen en DISCONNECTED sin conexiones reales en la configuración por defecto. """ from app.core.database import SessionLocal from app.core.adapters.titan_adapter import TitanAdapter from app.core.adapters.neurolab_adapter import NeuroLabAdapter db = SessionLocal() try: titan = TitanAdapter(db=db) assert titan.connection_status == "DISCONNECTED" hs = titan.get_handshake_contract() assert hs["connection_status"] == "DISCONNECTED" neurolab = NeuroLabAdapter(db=db) assert neurolab.connection_status == "DISCONNECTED" meta = neurolab.get_adapter_metadata() assert meta["connection_status"] == "DISCONNECTED" finally: db.close() # -------------------------------------------------------------------------- # PRUEBA H: Integridad Operativa Inmutable # -------------------------------------------------------------------------- def test_canonical_database_and_protected_files_remain_unmodified(): """ Prueba H: Comprobar que los archivos científicos protegidos y la SQLite operativa conservan exactamente su hash SHA-256 inicial. Separa formalmente: 1. Base canónica protegida (C:\\xampp\\htdocs\\centro de investigacion\\storage\\db\\research_center.db). 2. Base inicial del laboratorio candidato (BASE_DIR / storage/db/research_center.db). 3. Base experimental temporal (os.environ['ORC_TEST_DB_PATH']) que cambia legítimamente en tests. 4. Línea base histórica congelada ORC-22D vs Manifiesto versionado ORC-29 del candidato. """ import json # Expected baseline hash computed at start of Mission 22D (preserved verbatim) expected_db_hash = "E4A5A8DAE3B9BF9C24DF50ABC243CB4969F4CF4B29CC98F1013091CF01F9CDE9".lower() # Historical frozen hashes from Mission 22D (preserved verbatim) protected_files = { "app/core/coordinator.py": "B3D070FB5E707673115A1A07AE548DC9982962173377BE85921D170112625EC6".lower(), "app/core/ai_provider.py": "726EECE5293B5B60B80835333EE5604699A509B009DCDDC19B80FF34B0366543".lower(), "app/core/database.py": "CB8E5CA254FCA88394F3D7390D80CB36E9B5FCF499C583749E2E051ACB9C2B54".lower(), "app/core/models.py": "467D9FFB6D628B6A08C293B988A104B008D6FB0E9472DFC39EF265674FC48BE2".lower(), "app/core/lab_contract.py": "52E741DD27659E749B321A4EBDB3A057E06F110F096B4FE01CA7BC6C524AF86D".lower(), "app/core/lab_adapters.py": "E44B116DC62D3FFD569CF0C62BFD09C6C2FB4A4CA5E35C16FEC00A906B0BCF01".lower(), "app/core/lab_registry.py": "7144FD3DAA4CCFFF56B5B2604CF5D2733C64105B1750F648128404B2C5A06B79".lower(), "app/config.py": "95B05DEC6043D1CF1F16DFC4A17DD1218A7AA58F73B3CCFDB3CD825BEDC3EEF9".lower() } # 1. Verify Protected Production Installation (if present on host) against historical ORC-22D frozen hashes protected_prod_root = Path(r"C:\xampp\htdocs\centro de investigacion") if protected_prod_root.is_dir() and (protected_prod_root / "app").is_dir(): prod_db = protected_prod_root / "storage" / "db" / "research_center.db" assert prod_db.is_file(), "Protected production database file not found" prod_db_hash = compute_sha256(prod_db).lower() assert prod_db_hash == expected_db_hash, ( f"CRITICAL: Protected production database was altered! Original: {expected_db_hash}, Current: {prod_db_hash}" ) for rel_str, exp_hash in protected_files.items(): fp = protected_prod_root / rel_str assert fp.is_file(), f"Protected production file missing: {rel_str}" actual_hash = compute_sha256(fp).lower() assert actual_hash == exp_hash, ( f"Protected production file modified! {rel_str}: expected {exp_hash}, got {actual_hash}" ) # 2. Verify Candidate Initial Baseline Database (Role 2) remains unmodified by experimental tests (Role 3) canonical_db = BASE_DIR / "storage" / "db" / "research_center.db" assert canonical_db.is_file(), "Candidate initial baseline database file not found" current_db_hash = compute_sha256(canonical_db).lower() assert current_db_hash == expected_db_hash, ( f"CRITICAL: Candidate initial baseline database was altered! Original: {expected_db_hash}, Current: {current_db_hash}" ) # 3. Verify Candidate Core Files against ORC-29 Versioned Manifest (or ORC-22D if unversioned) orc29_manifest_path = BASE_DIR / "ORC29_CANDIDATE_MANIFEST.json" if orc29_manifest_path.is_file(): manifest_data = json.loads(orc29_manifest_path.read_text(encoding="utf-8")) candidate_hashes = { k: v.lower() for k, v in manifest_data.get("core_and_modified_files_sha256", {}).items() } assert len(candidate_hashes) >= 8, "ORC29_CANDIDATE_MANIFEST.json missing core_and_modified_files_sha256 entries" for rel_str, exp_orc22d_hash in protected_files.items(): fp = BASE_DIR / rel_str assert fp.is_file(), f"Candidate core file missing: {rel_str}" actual_hash = compute_sha256(fp).lower() expected_candidate_hash = candidate_hashes.get(rel_str, exp_orc22d_hash) assert actual_hash == expected_candidate_hash, ( f"Candidate file hash mismatch against ORC-29 manifest! {rel_str}: expected {expected_candidate_hash}, got {actual_hash}" ) else: for rel_str, exp_hash in protected_files.items(): fp = BASE_DIR / rel_str assert fp.is_file(), f"Protected file missing: {rel_str}" actual_hash = compute_sha256(fp).lower() assert actual_hash == exp_hash, ( f"Protected file modified! {rel_str}: expected {exp_hash}, got {actual_hash}" )