""" Optimus Research Center - Portable Packaging Script Version 1.0.0 (Mission 22D Corrected Architecture) Generates a lightweight, self-contained, clean and secure portable distribution zip package for Optimus Research Center. Key Architecture: 1. Component Allowlist: Only explicitly permitted functional components are packaged. 2. Multi-layer Hygiene Filtering: All paths, filenames and extensions are scrutinized case-insensitively against forbidden patterns (.env*, databases, credentials, logs, caches). 3. Pre-flight & Post-flight Validation: Validates files before packaging, creates archive in a temporary staging location, and runs comprehensive zip security tests (CRC, path traversal, duplicate collisions, required files, size limits) before publishing to dist/. 4. Verifiable Manifest: Computes exact SHA-256 hashes, file sizes, and environment specification. """ import os import sys import zipfile import hashlib import json import re import tempfile import shutil from pathlib import Path from datetime import datetime, timezone from typing import List, Dict, Tuple, Set, Optional # Resolve project root relative to script location BASE_DIR = Path(__file__).resolve().parent.parent # ---------------------------------------------------------------------- # 1. ALLOWLIST SPECIFICATION # ---------------------------------------------------------------------- # Directories permitted for recursive inclusion (subject to hygiene filters) ALLOWED_DIRECTORIES = { "app", "tests" } # Specific utility scripts permitted in portable package ALLOWED_SCRIPTS = { "package_laboratory.py", "test_portable_installation.py", "safe_stop.py", "verify_laboratory_portability.ps1" } # Explicit root-level files permitted in portable package ALLOWED_ROOT_FILES = { "run.py", "run_scenarios.py", "requirements.txt", "README.md", "start.bat", "stop.bat", "run_tests.bat", ".gitignore", "ORC22C_LABORATORY_ARCHITECTURE.md", "ORC22C_TITAN_INTEGRATION_CONTRACT.md", "ORC22C_NEUROLAB_INTEGRATION_CONTRACT.md", "ORC22C_PC_VPS_PREPARATION.md", "ORC22D_ENVIRONMENT_CONFIGURATION.md", "ORC29_CANDIDATE_MANIFEST.json" } # Directories that must exist in target installation (created as empty directory markers) REQUIRED_EMPTY_STORAGE_DIRS = [ "storage/artifacts", "storage/datasets", "storage/db", "storage/evidence", "storage/reports" ] # Mandatory core files that MUST be present in final package MANDATORY_ARCHIVE_FILES = { "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_ai/router.py", "run.py", "requirements.txt", "start.bat", "stop.bat", "run_tests.bat", "README.md", "ORC22D_ENVIRONMENT_CONFIGURATION.md" } # ---------------------------------------------------------------------- # 2. STRICT HYGIENE EXCLUSION SPECIFICATION (DEFENSE-IN-DEPTH) # ---------------------------------------------------------------------- # Forbidden directory names anywhere in path (case-insensitive) FORBIDDEN_DIR_NAMES = { "__pycache__", ".pytest_cache", ".git", ".svn", ".hg", "venv", ".venv", "env", "dist", "backups", "sandbox_tmp", "quarantine", "quarantine_audit23", ".cache", "htmlcov", ".idea", ".vscode" } # Forbidden file extensions (case-insensitive) FORBIDDEN_EXTENSIONS = { # Databases & WAL ".db", ".sqlite", ".sqlite3", ".db-wal", ".db-shm", ".db-journal", ".bak", ".backup", # Python bytecode & coverage ".pyc", ".pyo", ".pyd", ".coverage", # Logs & temporary files ".log", ".pid", ".tmp", ".swp", # Secrets & Certificates ".key", ".pem", ".crt", ".p12", ".pfx", ".id_rsa", ".pub" } # Exact forbidden filenames (case-insensitive) FORBIDDEN_EXACT_NAMES = { ".env", ".env.example", "server.pid", ".coverage" } # Forbidden regex patterns for sensitive or residual files FORBIDDEN_NAME_PATTERNS = [ re.compile(r"^\.env(\..+)?$", re.IGNORECASE), # .env, .env.example, .env.production, etc. re.compile(r"^.*secret.*$", re.IGNORECASE), # any file with 'secret' in name re.compile(r"^.*credential.*$", re.IGNORECASE), # any file with 'credential' in name re.compile(r"^.*password.*$", re.IGNORECASE), # any file with 'password' in name re.compile(r"^.*token.*$", re.IGNORECASE), # any file with 'token' in name re.compile(r"^research_center.*\.db.*$", re.IGNORECASE) # any research_center database file ] # Package size limits MAX_PACKAGE_SIZE_BYTES = 25 * 1024 * 1024 # 25 MB max compressed MAX_UNCOMPRESSED_BYTES = 60 * 1024 * 1024 # 60 MB max uncompressed def compute_sha256(file_path: Path) -> str: """Computes SHA-256 hash of a file efficiently.""" h = hashlib.sha256() with open(file_path, "rb") as f: while chunk := f.read(65536): h.update(chunk) return h.hexdigest() def validate_file_hygiene(rel_path: Path, base_dir: Path) -> Tuple[bool, Optional[str]]: """ Scrutinizes a candidate file against all security, privacy, and hygiene rules. Returns (is_allowed, rejection_reason). """ # 1. Symlink check: block symlinks to prevent directory escaping abs_path = base_dir / rel_path if abs_path.is_symlink(): return False, f"Symlink rejected: {rel_path}" norm_rel = str(rel_path).replace("\\", "/") parts = [p.lower() for p in rel_path.parts] filename_lower = rel_path.name.lower() # 2. Check forbidden directory components for part in parts[:-1]: if part in FORBIDDEN_DIR_NAMES: return False, f"Path contains forbidden directory: '{part}'" if part.startswith(".env"): return False, f"Path contains .env-like directory: '{part}'" # 3. Check exact forbidden names if filename_lower in FORBIDDEN_EXACT_NAMES: return False, f"Filename is explicitly forbidden: '{rel_path.name}'" # 4. Check forbidden extensions suffix_lower = rel_path.suffix.lower() if suffix_lower in FORBIDDEN_EXTENSIONS: return False, f"Forbidden file extension '{suffix_lower}'" # Check multi-part extensions like .db-wal or .db-shm if any(filename_lower.endswith(ext) for ext in FORBIDDEN_EXTENSIONS): return False, f"Forbidden compound extension detected: '{filename_lower}'" # 5. Check regex patterns (covers .env.*, secrets, tokens, db files) for pattern in FORBIDDEN_NAME_PATTERNS: if pattern.match(filename_lower): return False, f"Filename matches forbidden sensitive pattern: '{rel_path.name}'" # 6. Check database files in storage if parts[0] == "storage": if len(parts) >= 2 and parts[1] == "backups": return False, f"Backups are strictly excluded: '{rel_path}'" if filename_lower.endswith((".db", ".sqlite", ".sqlite3")): return False, f"Database file in storage excluded: '{rel_path}'" return True, None def select_package_files(base_dir: Path) -> List[Tuple[Path, Path]]: """ Selects files based strictly on the Allowlist, then scrutinizes each file through the hygiene validator. Returns list of (abs_path, rel_path). """ selected: List[Tuple[Path, Path]] = [] rejected: List[Tuple[str, str]] = [] # 1. Allowed root files for root_file in ALLOWED_ROOT_FILES: abs_p = base_dir / root_file if abs_p.is_file(): rel_p = Path(root_file) allowed, reason = validate_file_hygiene(rel_p, base_dir) if allowed: selected.append((abs_p, rel_p)) else: rejected.append((str(rel_p), reason)) # 2. Allowed scripts scripts_dir = base_dir / "scripts" if scripts_dir.is_dir(): for script_name in ALLOWED_SCRIPTS: abs_p = scripts_dir / script_name if abs_p.is_file(): rel_p = Path("scripts") / script_name allowed, reason = validate_file_hygiene(rel_p, base_dir) if allowed: selected.append((abs_p, rel_p)) else: rejected.append((str(rel_p), reason)) # 3. Allowed directories (app, tests) for dir_name in ALLOWED_DIRECTORIES: root_dir = base_dir / dir_name if not root_dir.is_dir(): continue for root, dirs, files in os.walk(root_dir): # Prune forbidden directories in-place dirs[:] = [d for d in dirs if d.lower() not in FORBIDDEN_DIR_NAMES and not d.lower().startswith(".env")] for file in files: abs_p = Path(root) / file rel_p = abs_p.relative_to(base_dir) allowed, reason = validate_file_hygiene(rel_p, base_dir) if allowed: selected.append((abs_p, rel_p)) else: rejected.append((str(rel_p), reason)) # Sort selected files for deterministic builds selected.sort(key=lambda item: str(item[1]).replace("\\", "/").lower()) return selected def audit_zip_archive(zip_path: Path) -> Tuple[bool, List[str]]: """ Performs comprehensive post-creation security and integrity audit on the ZIP file: 1. Tests CRC-32 integrity of all entries. 2. Rejects path traversal (..), absolute paths, and backslashes. 3. Rejects duplicate entry names, including case-insensitive collisions on Windows. 4. Rejects any forbidden extensions, sensitive files (.env*), databases, or secrets. 5. Verifies all mandatory core files are present. 6. Verifies size limits. """ errors: List[str] = [] seen_names_lower: Dict[str, str] = {} entry_paths: Set[str] = set() try: with zipfile.ZipFile(zip_path, "r") as zf: # 1. CRC test bad_crc = zf.testzip() if bad_crc is not None: errors.append(f"CRC check failed on archive member: {bad_crc}") total_uncompressed = 0 infolist = zf.infolist() for info in infolist: entry_name = info.filename entry_path = Path(entry_name) norm_name = entry_name.replace("\\", "/") total_uncompressed += info.file_size # 2. Check path safety if norm_name.startswith("/") or ":" in norm_name: errors.append(f"Absolute path in archive rejected: '{entry_name}'") if ".." in norm_name.split("/"): errors.append(f"Directory traversal ('..') in archive rejected: '{entry_name}'") if "\\" in entry_name: errors.append(f"Windows backslash in ZIP entry name rejected: '{entry_name}'") # 3. Check case collisions lower_name = norm_name.lower() if lower_name in seen_names_lower: errors.append( f"Case-insensitive collision detected: '{norm_name}' conflicts with '{seen_names_lower[lower_name]}'" ) else: seen_names_lower[lower_name] = norm_name # Skip directory entries for file checks if norm_name.endswith("/"): continue entry_paths.add(norm_name) # 4. Check forbidden names & extensions in archive basename_lower = entry_path.name.lower() if basename_lower in FORBIDDEN_EXACT_NAMES: errors.append(f"Forbidden exact file packaged in archive: '{entry_name}'") for pattern in FORBIDDEN_NAME_PATTERNS: if pattern.match(basename_lower): errors.append(f"Sensitive or forbidden file pattern in archive: '{entry_name}'") for ext in FORBIDDEN_EXTENSIONS: if basename_lower.endswith(ext): errors.append(f"Forbidden file extension in archive: '{entry_name}' (extension {ext})") # Check path components for forbidden directories parts = [p.lower() for p in entry_path.parts] for p in parts[:-1]: if p in FORBIDDEN_DIR_NAMES or p.startswith(".env"): errors.append(f"Forbidden directory component in archive: '{entry_name}'") # 5. Check mandatory files missing_mandatory = MANDATORY_ARCHIVE_FILES - entry_paths if missing_mandatory: errors.append(f"Mandatory files missing from archive: {sorted(missing_mandatory)}") # 6. Check size limits compressed_size = zip_path.stat().st_size if compressed_size > MAX_PACKAGE_SIZE_BYTES: errors.append( f"Package size ({compressed_size:,} bytes) exceeds limit ({MAX_PACKAGE_SIZE_BYTES:,} bytes)" ) if total_uncompressed > MAX_UNCOMPRESSED_BYTES: errors.append( f"Uncompressed size ({total_uncompressed:,} bytes) exceeds limit ({MAX_UNCOMPRESSED_BYTES:,} bytes)" ) except Exception as e: errors.append(f"Failed to read/audit zip archive: {e}") return (len(errors) == 0), errors def create_portable_package( output_dir: Optional[Path] = None, base_dir: Optional[Path] = None, package_filename: str = "optimus_research_center_portable_v1.0.zip" ) -> Path: """ Generates the portable distribution archive in a temporary staging location, performs independent validation, and publishes to the output directory only upon 100% success. """ project_root = (base_dir or BASE_DIR).resolve() out_dir = (output_dir or (project_root / "dist")).resolve() out_dir.mkdir(parents=True, exist_ok=True) target_zip = out_dir / package_filename manifest_path = out_dir / "manifest.json" print("=" * 75) print(" OPTIMUS RESEARCH CENTER - PORTABLE PACKAGING UTILITY (MISSION 22D)") print(f" Source Root: {project_root}") print(f" Target Archive: {target_zip}") print("=" * 75) # Step 1: Select files using Allowlist & Hygiene Validator print("\n[1/5] Selecting allowed components and running hygiene filters...") selected_files = select_package_files(project_root) print(f" -> Total files selected: {len(selected_files)}") # Step 2: Build archive in temporary staging location print("\n[2/5] Creating archive in isolated temporary staging file...") temp_fd, temp_zip_path_str = tempfile.mkstemp(prefix="orc_pkg_", suffix=".zip") os.close(temp_fd) staging_zip = Path(temp_zip_path_str) included_metadata: List[Dict] = [] total_uncompressed_bytes = 0 try: with zipfile.ZipFile(staging_zip, "w", zipfile.ZIP_DEFLATED) as zipf: # Write required empty storage directory entries for dir_marker in REQUIRED_EMPTY_STORAGE_DIRS: dir_entry = dir_marker.rstrip("/") + "/" zipinfo = zipfile.ZipInfo(dir_entry) zipinfo.external_attr = 0o755 << 16 | 0o040000 # Directory mode zipf.writestr(zipinfo, "") # Write selected files with forward-slash POSIX paths for abs_path, rel_path in selected_files: arcname = str(rel_path).replace("\\", "/") zipf.write(abs_path, arcname=arcname) sz = abs_path.stat().st_size sha = compute_sha256(abs_path) total_uncompressed_bytes += sz included_metadata.append({ "path": arcname, "size_bytes": sz, "sha256": sha }) # Step 3: Run comprehensive post-creation security audit print("\n[3/5] Performing independent pre-release security and integrity audit on archive...") audit_passed, audit_errors = audit_zip_archive(staging_zip) if not audit_passed: print("[CRITICAL ERROR] Archive failed pre-release security audit:") for err in audit_errors: print(f" - {err}") raise ValueError(f"Package security validation failed with {len(audit_errors)} error(s)!") print(" -> Security Audit: PASSED (Zero sensitive files, zero traversal, valid CRC, all mandatory components present)") # Step 4: Publish archive atomically to dist/ print(f"\n[4/5] Publishing validated archive to: {target_zip}") shutil.copyfile(staging_zip, target_zip) package_sha256 = compute_sha256(target_zip) package_size_bytes = target_zip.stat().st_size # Parse dependencies from requirements.txt if present req_file = project_root / "requirements.txt" dependencies = [] if req_file.is_file(): for line in req_file.read_text(encoding="utf-8").splitlines(): line = line.strip() if line and not line.startswith("#"): dependencies.append(line) # Step 5: Generate cryptographic manifest print("\n[5/5] Generating cryptographic manifest...") manifest = { "package_name": package_filename, "version": "1.0.0-m22d", "mission": "MISSION_22D_PORTABLE_HYGIENE_CORRECTION", "generated_at_utc": datetime.now(timezone.utc).isoformat(), "sha256": package_sha256, "size_bytes": package_size_bytes, "total_uncompressed_bytes": total_uncompressed_bytes, "file_count": len(included_metadata), "target_profile": "PC_INTEGRATION_READY", "profiles_supported": ["LAPTOP_STANDALONE", "PC_INTEGRATION_READY", "VPS_PREPARED"], "security_certification": { "secrets_included": False, "dot_env_files_included": False, "sqlite_databases_included": False, "path_traversal_safe": True, "zero_duplicate_names": True, "all_mandatory_files_present": True }, "excluded_categories": [ "All .env* files (.env, .env.example, .env.production, etc.)", "All SQLite database files (*.db, *.sqlite, *.sqlite3, -wal, -shm)", "All historical database backups (storage/backups/*)", "All credentials, certificates, tokens and private keys (*.key, *.pem, *.crt, etc.)", "All Python bytecode, caches and virtual environments (__pycache__, .pytest_cache, venv)", "All log files, server.pid and runtime temp files (*.log, *.pid, *.tmp)", "All unneeded historical audit reports from previous missions" ], "dependencies": dependencies, "installation_notes": ( "Extract archive into clean target folder. No .env file is required to start " "in LAPTOP_STANDALONE mode. Execute 'python -c \"from app.core.database import init_db; init_db()\"' " "to generate fresh local SQLite schema, then run 'python run.py' or 'start.bat'." ), "files": included_metadata } with open(manifest_path, "w", encoding="utf-8") as f: json.dump(manifest, f, indent=2) print("\n" + "=" * 75) print(" [SUCCESS] OPTIMUS RESEARCH CENTER PORTABLE PACKAGE CREATED & CERTIFIED") print(f" Archive: {target_zip}") print(f" Size: {package_size_bytes:,} bytes ({package_size_bytes / (1024*1024):.2f} MB)") print(f" File Count: {len(included_metadata)} files") print(f" SHA-256: {package_sha256}") print(f" Manifest: {manifest_path}") print("=" * 75) return target_zip finally: # Clean up staging zip if staging_zip.exists(): try: staging_zip.unlink() except Exception: pass if __name__ == "__main__": create_portable_package()