"""Minimal isolated HOLD boundary before any Aureon package or target import. Run this file directly with ``python -I -S -B``. It imports only the Python standard library, derives the repository root from its own fixed location, and never imports Aureon, invokes Git, starts a child, writes a file, or accepts an arbitrary entrypoint. Until a production guard and durable HNC sink exist, its only valid decision is HOLD. """ from __future__ import annotations import argparse import hashlib import json import re import time from pathlib import Path from types import MappingProxyType SCHEMA = "aureon.plumber.isolated-protected-bootstrap.v05" _TARGET_RE = re.compile(r"^[a-z][a-z0-9-]{0,63}$") _MAX_ARGUMENTS = 64 _MAX_ARGUMENT_BYTES = 4096 _MAX_ARGUMENT_AGGREGATE_BYTES = 64 * 1024 _ZERO_SHA256 = "0" * 64 _SOURCE_SCOPE_SCHEMA = "aureon.plumber.isolated-perimeter-source-scope.v05" _MAX_SOURCE_FILES = 256 _MAX_SOURCE_BYTES = 512 * 1024 * 1024 _MAX_SINGLE_SOURCE_BYTES = 128 * 1024 * 1024 _PROTECTION_SOURCE_PATHS = frozenset( { "aureon/autonomous/aureon_intrusion_protection_bridge.py", "aureon/autonomous/aureon_runtime_protection_proposal_vault_v05.py", "aureon/plumber/authorization_chain_v02.py", "aureon/plumber/magic_star_v02.py", "aureon/plumber/os_protection.py", "aureon/plumber/production_release_broker_v03.py", "aureon/plumber/protected_bootstrap_v05.py", "aureon/plumber/release_boundary_v02.py", "aureon/plumber/runtime_guard_v04.py", "aureon/plumber/runtime_intrusion_ledger_v04.py", "aureon/plumber/star_custody_v02.py", "protected_console_bootstrap_v05.py", "pyproject.toml", "scripts/bootstrap/protected_bootstrap_v05.py", "scripts/validation/audit_economic_mutation_boundaries.py", "scripts/validation/audit_os_protection_boundaries.py", } ) _NODE_TARGET_SOURCES = MappingProxyType( { "flameborn:runtime": "flameborn/runtime/server.mjs", "frontend:vite": "frontend/package.json", } ) # Only perimeter IDs are needed here. Inner fixed Python targets remain in the # package registry, but cannot be reached until this outer boundary is ready. _TARGETS = MappingProxyType( { "cloud-autonomous-worker": ( "python", "aureon.autonomous.aureon_autonomous_worker", ), "cloud-command-center": ( "python", "aureon.command_centers.aureon_command_center_ui", ), "cloud-kraken-cache": ( "python", "aureon.exchanges.kraken_cache_feeder", ), "cloud-market-cache": ( "python", "aureon.data_feeds.unified_market_cache", ), "cloud-orca": ("python", "aureon.bots.orca_complete_kill_cycle"), "cloud-pro-dashboard": ( "python", "aureon.monitors.aureon_pro_dashboard", ), "cloud-queen-redistribution": ( "python", "aureon.queen.queen_power_redistribution", ), "cloud-supervisor": ("native", "deploy/supervisord.conf"), "capability-demo": ("python", "aureon.saas.capability_demo"), "canonical-cloud-organism": ( "python", "scripts.operations.run_canonical_cloud_organism", ), "data-ocean": ("python", "aureon.autonomous.aureon_data_ocean"), "druidic-live-calibration": ( "python", "scripts.operations.run_live_druidic_calibration", ), "docker-runtime": ("native", "Dockerfile"), "ephemeris-pipeline": ("native", "Dockerfile.ephemeris"), "flameborn-runtime": ("node", "flameborn:runtime"), "frontend-vite": ("node", "frontend:vite"), "hnc": ("python", "aureon.core.hnc_live_daemon"), "ignition": ("python", "scripts/aureon_ignition.py"), "linux-supervisor": ("native", "deploy/supervisord.linux.conf"), "local-gui": ("python", "aureon.operator.local_gui_organism"), "master-launcher": ( "python", "aureon.autonomous.aureon_master_launcher", ), "mind-hub": ( "python", "aureon.autonomous.aureon_mind_thought_action_hub", ), "operator": ("python", "aureon.operator.operator_server"), "operator-wsgi": ("python", "aureon.operator.wsgi"), "organism": ("python", "aureon.core.organism_daemon"), "organism-observer": ( "python", "aureon.autonomous.aureon_organism_runtime_observer", ), "parallel-strategy-unity": ( "python", "aureon.trading.parallel_strategy_unity", ), "production-supervisor": ("native", "production/supervisord.conf"), "queen-eternal-machine": ( "python", "aureon.queen.queen_eternal_machine", ), "queen-web-dashboard": ( "python", "aureon.queen.queen_web_dashboard", ), "power-redistribution-engine": ( "python", "aureon.utils.aureon_power_redistribution_engine", ), "self-questioning": ( "python", "aureon.autonomous.aureon_self_questioning_ai", ), "scorm-benchmark": ( "python", "aureon.operator.scorm_cloud_runner", ), "autonomous-self-run": ( "python", "aureon.autonomous.aureon_autonomous_self_run_loop", ), "unified-market-status": ( "python", "aureon.exchanges.unified_market_status_server", ), "unified-market-trader": ( "python", "aureon.exchanges.unified_market_trader", ), "website": ("python", "aureon.operator.website_operator"), "windows-wake": ( "powershell", "scripts/launchers/AUREON_WAKE_UP_FULL_AUTONOMOUS.ps1", ), "ws-market-data-feeder": ( "python", "aureon.data_feeds.ws_market_data_feeder", ), } ) def _canonical_bytes(value: dict[str, object]) -> bytes: return json.dumps( value, sort_keys=True, separators=(",", ":"), ensure_ascii=True, allow_nan=False, ).encode("utf-8") def _sha256_bytes(value: bytes) -> str: return hashlib.sha256(value).hexdigest() def _target_commitment(target_id: str, runtime_kind: str, entrypoint: str) -> str: return _sha256_bytes( _canonical_bytes( { "entrypoint": entrypoint, "runtime_kind": runtime_kind, "schema": "aureon.plumber.isolated-bootstrap-target.v05", "target_id": target_id, } ) ) def _bounded_arguments(values: list[str]) -> tuple[str, ...]: if values[:1] == ["--"]: values = values[1:] if len(values) > _MAX_ARGUMENTS: raise ValueError("bounded_bootstrap_target_arguments_required") aggregate = 0 for value in values: if type(value) is not str or "\x00" in value: raise ValueError("exact_bootstrap_target_argument_required") if value == "--root" or value.startswith("--root="): raise ValueError("caller_controlled_bootstrap_root_forbidden") size = len(value.encode("utf-8")) if size > _MAX_ARGUMENT_BYTES: raise ValueError("bounded_bootstrap_target_argument_required") aggregate += size if aggregate > _MAX_ARGUMENT_AGGREGATE_BYTES: raise ValueError("bounded_bootstrap_target_argument_aggregate_required") return tuple(values) def _repo_root() -> Path: script = Path(__file__).resolve() root = script.parents[2] expected = (root / "scripts" / "bootstrap" / "protected_bootstrap_v05.py").resolve() if script != expected or not expected.is_file(): raise ValueError("fixed_bootstrap_path_invalid") return root def _bootstrap_source_sha256() -> str: try: return _sha256_bytes(Path(__file__).resolve().read_bytes()) except OSError: return _ZERO_SHA256 def _runtime_guard_source_sha256(root: Path) -> str: path = root / "aureon" / "plumber" / "runtime_guard_v04.py" try: return _sha256_bytes(path.read_bytes()) except OSError: return _ZERO_SHA256 def _safe_source_path(root: Path, relative: str) -> Path: candidate_relative = Path(relative.replace("\\", "/")) if ( not relative or candidate_relative.is_absolute() or any(part in {"", ".", ".."} for part in candidate_relative.parts) ): raise ValueError("bounded_bootstrap_source_path_required") root_resolved = root.resolve(strict=True) candidate = root_resolved.joinpath(candidate_relative) current = root_resolved for part in candidate_relative.parts: current = current / part if current.is_symlink(): raise ValueError("bootstrap_source_symlink_forbidden") resolved = candidate.resolve(strict=True) try: resolved.relative_to(root_resolved) except ValueError as exc: raise ValueError("bootstrap_source_path_escape") from exc if not resolved.is_file(): raise ValueError("bootstrap_source_file_required") return resolved def _stable_file_measurement(path: Path) -> tuple[str, int]: before = path.stat() if before.st_size < 0 or before.st_size > _MAX_SINGLE_SOURCE_BYTES: raise ValueError("bounded_bootstrap_source_file_required") digest = hashlib.sha256() size = 0 with path.open("rb") as handle: while True: chunk = handle.read(1024 * 1024) if not chunk: break size += len(chunk) if size > _MAX_SINGLE_SOURCE_BYTES: raise ValueError("bounded_bootstrap_source_file_required") digest.update(chunk) after = path.stat() identity_before = ( before.st_dev, before.st_ino, before.st_size, before.st_mtime_ns, ) identity_after = ( after.st_dev, after.st_ino, after.st_size, after.st_mtime_ns, ) if identity_before != identity_after or size != after.st_size: raise ValueError("bootstrap_source_changed_during_measurement") return digest.hexdigest(), size def _source_files(root: Path) -> tuple[tuple[str, Path], ...]: root_resolved = root.resolve(strict=True) relative_paths = set(_PROTECTION_SOURCE_PATHS) for runtime_kind, entrypoint in _TARGETS.values(): relative_paths.add( _target_source_relative(runtime_kind, entrypoint, root_resolved) ) if len(relative_paths) > _MAX_SOURCE_FILES: raise ValueError("bootstrap_source_file_capacity_exceeded") return tuple( (relative, _safe_source_path(root_resolved, relative)) for relative in sorted(relative_paths, key=lambda item: item.encode("utf-8")) ) def _measure_source_scope(root: Path) -> tuple[dict[str, object], dict[str, tuple[str, int]]]: files = _source_files(root) aggregate = hashlib.sha256() aggregate.update((_SOURCE_SCOPE_SCHEMA + "\n").encode("ascii")) measurements: dict[str, tuple[str, int]] = {} total_bytes = 0 for relative, path in files: digest, size = _stable_file_measurement(path) total_bytes += size if total_bytes > _MAX_SOURCE_BYTES: raise ValueError("bootstrap_source_byte_capacity_exceeded") measurements[relative] = (digest, size) aggregate.update( _canonical_bytes( { "path": relative, "sha256": digest, "size": size, } ) ) aggregate.update(b"\n") if not measurements: raise ValueError("bootstrap_source_inventory_required") return ( { "schema": _SOURCE_SCOPE_SCHEMA, "sha256": aggregate.hexdigest(), "file_count": len(measurements), "total_bytes": total_bytes, }, measurements, ) def _target_source_relative(runtime_kind: str, entrypoint: str, root: Path) -> str: if runtime_kind == "node": relative = _NODE_TARGET_SOURCES.get(entrypoint) if relative is None: raise ValueError("fixed_node_target_source_required") return relative if runtime_kind in {"native", "powershell", "shell"}: return entrypoint.replace("\\", "/") if runtime_kind != "python": raise ValueError("recognized_bootstrap_runtime_kind_required") if entrypoint.casefold().endswith(".py") or "/" in entrypoint or "\\" in entrypoint: return entrypoint.replace("\\", "/") base = entrypoint.replace(".", "/") candidates = (f"{base}.py", f"{base}/__init__.py") present = [item for item in candidates if (root / item).is_file()] if len(present) != 1: raise ValueError("exact_python_target_source_required") return present[0] def _receipt( *, target_id: str, arguments: tuple[str, ...], checked_at: float, ) -> dict[str, object]: root = _repo_root() target = _TARGETS.get(target_id) registered = target is not None runtime_kind = "" if target is None else target[0] entrypoint = "" if target is None else target[1] failures = { "durable_hnc_evidence", "full_os_protection", "native_outer_boundary", "runtime_guard_production_readiness", } if not registered: failures.add("fixed_bootstrap_target") if runtime_kind and runtime_kind != "python": failures.add(f"{runtime_kind}_outer_boundary") source_scope: dict[str, object] = { "schema": _SOURCE_SCOPE_SCHEMA, "sha256": _ZERO_SHA256, "file_count": 0, "total_bytes": 0, } source_scope_stable = False source_measurements: dict[str, tuple[str, int]] = {} try: first_scope, first_measurements = _measure_source_scope(root) second_scope, second_measurements = _measure_source_scope(root) source_scope_stable = ( first_scope == second_scope and first_measurements == second_measurements ) if not source_scope_stable: raise ValueError("bootstrap_source_scope_changed_during_measurement") source_scope = second_scope source_measurements = second_measurements except (OSError, ValueError): failures.add("source_scope_measurement") target_source_sha256 = _ZERO_SHA256 target_source_size = 0 target_source_measured = False target_source_in_scope = False if target is not None: try: target_relative = _target_source_relative(runtime_kind, entrypoint, root) target_path = _safe_source_path(root, target_relative) target_source_sha256, target_source_size = _stable_file_measurement( target_path ) recorded = source_measurements.get(target_relative) target_source_in_scope = recorded == ( target_source_sha256, target_source_size, ) target_source_measured = target_source_in_scope except (OSError, ValueError): target_source_measured = False if not target_source_measured: failures.add("target_source_measurement") argument_policy_attested = len(arguments) == 0 if not argument_policy_attested: failures.add("target_argument_policy") causal: dict[str, object] = { "schema": SCHEMA, "decision": "HOLD", "reason": "complete_isolated_protected_bootstrap_required", "failed_checks": sorted(failures), "target_id": target_id, "target_registered": registered, "target_runtime_kind": runtime_kind, "target_entrypoint_commitment": ( _ZERO_SHA256 if target is None else _target_commitment(target_id, runtime_kind, entrypoint) ), "target_argument_count": len(arguments), "target_arguments_sha256": _sha256_bytes( json.dumps(list(arguments), separators=(",", ":"), ensure_ascii=True).encode( "utf-8" ) ), "repo_root_sha256": _sha256_bytes(str(root).casefold().encode("utf-8")), "repo_root_derived_from_bootstrap_path": True, "bootstrap_path_bound": True, "bootstrap_source_sha256": _bootstrap_source_sha256(), "caller_controlled_root_accepted": False, "source_scope_schema": source_scope["schema"], "source_scope_sha256": source_scope["sha256"], "source_scope_file_count": source_scope["file_count"], "source_scope_total_bytes": source_scope["total_bytes"], "source_scope_measured": source_scope_stable, "source_scope_stable": source_scope_stable, "target_source_sha256": target_source_sha256, "target_source_size": target_source_size, "target_source_measured": target_source_measured, "target_source_in_source_scope": target_source_in_scope, "target_argument_policy": "empty-arguments-v1", "target_argument_policy_attested": argument_policy_attested, "runtime_guard_source_sha256": _runtime_guard_source_sha256(root), "runtime_guard_production_ready": False, "full_os_protection_evaluated": False, "durable_hnc_evidence_attested": False, "magic_star_durable_custody_attested": False, "external_head_anchor_attested": False, "target_imported": False, "target_called": False, "child_process_started": False, "bootstrap_subprocess_started": False, "git_invoked": False, "network_accessed": False, "file_written": False, "hnc_denial_recorded": False, "durable_hnc_denial_recorded": False, "process_start_authorized": False, "action_eligible": False, "economic_eligible": False, "operational_eligible": False, "production_ready": False, "checked_at": checked_at, } return { **causal, "receipt_id": f"bootstrap:isolated-v05:{_sha256_bytes(_canonical_bytes(causal))}", } def _invalid_receipt(*, checked_at: float) -> dict[str, object]: causal: dict[str, object] = { "schema": SCHEMA, "decision": "HOLD", "reason": "exact_isolated_bootstrap_request_required", "failed_checks": ["bootstrap_request"], "target_id": "", "target_imported": False, "target_called": False, "child_process_started": False, "bootstrap_subprocess_started": False, "git_invoked": False, "network_accessed": False, "file_written": False, "process_start_authorized": False, "production_ready": False, "checked_at": checked_at, } return { **causal, "receipt_id": f"bootstrap:isolated-v05:{_sha256_bytes(_canonical_bytes(causal))}", } def _parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--target-id", required=True) parser.add_argument("--pretty", action="store_true") parser.add_argument("target_arguments", nargs=argparse.REMAINDER) return parser def main(argv: list[str] | None = None) -> int: args = _parser().parse_args(argv) checked_at = time.time() try: if type(args.target_id) is not str or _TARGET_RE.fullmatch(args.target_id) is None: raise ValueError("exact_bootstrap_target_id_required") arguments = _bounded_arguments(args.target_arguments) receipt = _receipt( target_id=args.target_id, arguments=arguments, checked_at=checked_at, ) exit_code = 1 except (OSError, TypeError, ValueError): receipt = _invalid_receipt(checked_at=checked_at) exit_code = 2 print( json.dumps( receipt, sort_keys=True, indent=2 if args.pretty else None, separators=None if args.pretty else (",", ":"), ) ) return exit_code if __name__ == "__main__": raise SystemExit(main())