from __future__ import annotations import hashlib import importlib.util import json import os import socket import subprocess import sys import threading import urllib.parse from contextlib import contextmanager from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path from types import SimpleNamespace from typing import Any, Iterator ROOT = Path(__file__).resolve().parents[3] EXPORTER = ROOT / "scripts" / "backup" / "signoz-metadata-export.py" VERIFIER = ROOT / "scripts" / "backup" / "verify-signoz-metadata-export.py" RESTORE = ROOT / "scripts" / "backup" / "signoz-metadata-restore-drill.py" ISOLATED_CONTROLLER = ( ROOT / "scripts" / "backup" / "signoz-metadata-isolated-restore.py" ) POLICY = ROOT / "config" / "signoz" / "metadata-export-policy.json" TRACE_ID = "trace-P0-OBS-002-metadata" RUN_ID = "run-P0-OBS-002-metadata" WORK_ITEM_ID = "P0-OBS-002" TOKEN = "test-only-signoz-token-reference" SOURCE_ASSETS: dict[str, Any] = { "/api/v1/dashboards": {"data": [{"id": "dash-1", "title": "Ops"}]}, "/api/v1/rules": { "status": "success", "data": {"rules": [{"id": "rule-1", "name": "Latency"}]}, }, "/api/v1/explorer/views?sourcePage=logs": { "data": [ { "id": "view-1", "name": "Errors", "sourcePage": "logs", } ] }, "/api/v1/explorer/views?sourcePage=meter": {"data": None}, "/api/v1/explorer/views?sourcePage=metrics": {"data": {}}, "/api/v1/explorer/views?sourcePage=traces": {"data": []}, "/api/v1/roles": {"data": [{"id": "role-1", "name": "Viewer"}]}, "/api/v1/org/preferences": {"data": [{"name": "theme", "value": "dark"}]}, "/api/v1/user/preferences": {"data": [{"name": "timezone", "value": "UTC"}]}, "/api/v1/global/config": {"data": {"setupCompleted": True}}, } class MetadataHandler(BaseHTTPRequestHandler): assets: dict[str, Any] = {} drift_path: str | None = None get_counts: dict[str, int] = {} post_count = 0 def log_message(self, format: str, *args: object) -> None: return def _authorized(self) -> bool: return ( self.headers.get("SIGNOZ-API-KEY") == TOKEN or self.headers.get("Authorization") == f"Bearer {TOKEN}" ) def _send_json(self, status: int, value: Any) -> None: payload = json.dumps(value, separators=(",", ":")).encode("utf-8") self.send_response(status) self.send_header("Content-Type", "application/json") self.send_header("Content-Length", str(len(payload))) self.end_headers() self.wfile.write(payload) def do_GET(self) -> None: if not self._authorized(): self._send_json(401, {"error": "unauthorized"}) return path = self.path.split("?", 1)[0] asset_key = self.path if self.path in self.assets else path if asset_key not in self.assets: self._send_json(404, {"error": "not found"}) return self.get_counts[asset_key] = self.get_counts.get(asset_key, 0) + 1 value = json.loads(json.dumps(self.assets[asset_key])) if self.drift_path in {path, asset_key} and self.get_counts[asset_key] >= 2: value["data"].append({"id": "drift", "title": "changed"}) self._send_json(200, value) def do_POST(self) -> None: if not self._authorized(): self._send_json(401, {"error": "unauthorized"}) return path = self.path.split("?", 1)[0] if path != "/api/v1/explorer/views" and path not in self.assets: self._send_json(404, {"error": "not found"}) return length = int(self.headers.get("Content-Length", "0")) value = json.loads(self.rfile.read(length)) if not isinstance(value, dict): self._send_json(400, {"error": "object required"}) return if path == "/api/v1/explorer/views": source_page = value.get("sourcePage") if source_page not in {"logs", "meter", "metrics", "traces"}: self._send_json(400, {"error": "sourcePage required"}) return asset_key = f"{path}?{urllib.parse.urlencode({'sourcePage': source_page})}" if asset_key not in self.assets: self._send_json(404, {"error": "not found"}) return data = self.assets[asset_key].get("data") if data is None or data == {}: self.assets[asset_key]["data"] = [] data = self.assets[asset_key]["data"] collection = data else: data = self.assets[path].get("data") collection = ( data.get("rules") if path == "/api/v1/rules" and isinstance(data, dict) else data ) if not isinstance(collection, list): self._send_json(404, {"error": "not found"}) return type(self).post_count += 1 restored = {"id": f"server-{type(self).post_count}", **value} collection.append(restored) self._send_json(201, {"data": restored}) @contextmanager def metadata_server( assets: dict[str, Any], *, drift_path: str | None = None, ) -> Iterator[tuple[str, type[MetadataHandler]]]: handler = type("BoundMetadataHandler", (MetadataHandler,), {}) handler.assets = json.loads(json.dumps(assets)) handler.drift_path = drift_path handler.get_counts = {} handler.post_count = 0 server = ThreadingHTTPServer(("127.0.0.1", 0), handler) thread = threading.Thread(target=server.serve_forever, daemon=True) thread.start() try: host, port = server.server_address yield f"http://{host}:{port}", handler finally: server.shutdown() server.server_close() thread.join(timeout=5) def credential_file(tmp_path: Path, *, mode: int = 0o600) -> Path: path = tmp_path / "signoz-api-key.ref" path.write_text(f"{TOKEN}\n", encoding="ascii") path.chmod(mode) return path def export_command( *, base_url: str, output_dir: Path, credential: Path | None, apply: bool, receipt_file: Path | None = None, policy: Path | None = None, ) -> list[str]: command = [ sys.executable, str(EXPORTER), "--apply" if apply else "--check", "--base-url", base_url, "--output-dir", str(output_dir), "--trace-id", TRACE_ID, "--run-id", RUN_ID, "--work-item-id", WORK_ITEM_ID, ] if credential is not None: command.extend(["--credential-file", str(credential)]) if policy is not None: command.extend(["--policy", str(policy)]) if apply: terminal_path = receipt_file or output_dir.with_name( f"{output_dir.name}.terminal.json" ) command.extend(["--receipt-file", str(terminal_path)]) return command def create_bundle(tmp_path: Path) -> tuple[Path, Path]: credential = credential_file(tmp_path) bundle = tmp_path / "bundle" with metadata_server(SOURCE_ASSETS) as (base_url, _): result = subprocess.run( export_command( base_url=base_url, output_dir=bundle, credential=credential, apply=True, ), text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=20, check=False, ) assert result.returncode == 0, result.stderr return bundle, credential def load_isolated_controller() -> Any: backup_dir = str(ROOT / "scripts" / "backup") if backup_dir not in sys.path: sys.path.insert(0, backup_dir) spec = importlib.util.spec_from_file_location( "signoz_isolated_restore_contract_integration", ISOLATED_CONTROLLER, ) assert spec is not None and spec.loader is not None module = importlib.util.module_from_spec(spec) sys.modules[spec.name] = module spec.loader.exec_module(module) return module def isolation_receipt(tmp_path: Path, target_base_url: str) -> Path: policy = json.loads(POLICY.read_text(encoding="utf-8")) identity = hashlib.sha256( TRACE_ID.encode("utf-8") + b"\0" + RUN_ID.encode("utf-8") + b"\0" + WORK_ITEM_ID.encode("utf-8") ).hexdigest() short_id = identity[:16] prefix = f"awoooi-signoz-md-restore-{short_id}" port_text = target_base_url.rsplit(":", 1)[-1] port = int(port_text) if port_text.isdigit() else 443 production_before = { entry["container"]: { "id": hashlib.sha256(entry["container"].encode()).hexdigest(), "image_id": entry["image_id"], "image_ref": entry["image_ref"], "running": True, "started_at": "2026-07-22T00:00:00Z", "restart_count": 0, } for entry in policy["restore_isolation"]["production_continuity"].values() } value = { "schema": "awoooi_signoz_metadata_restore_isolation_v1", "trace_id": TRACE_ID, "run_id": RUN_ID, "work_item_id": WORK_ITEM_ID, "target": { "identity": identity, "prefix": prefix, "canonical_id": f"ephemeral-signoz-metadata-restore-{short_id}", "base_url_sha256": hashlib.sha256( target_base_url.encode("utf-8") ).hexdigest(), "image_ref": "signoz/signoz:v0.113.0", "image_id": policy["restore_isolation"]["required_image_id"], "images": policy["restore_isolation"]["images"], "image_present_locally": True, "image_pull_performed": False, "image_build_performed": False, "production_network_attached": False, "production_volume_attached": False, "ephemeral_volumes_only": True, "cleanup_controller_armed": True, "zero_residue_verifier_armed": True, "network_scope": "loopback_and_internal_only", "resource_label_key": policy["restore_isolation"]["resource_label_key"], "resource_label_value": RUN_ID, "identity_label_key": "com.awoooi.restore.identity", "identity_label_value": identity, "resources": { "network": f"{prefix}-net", "containers": [ f"{prefix}-server", f"{prefix}-clickhouse", f"{prefix}-zookeeper", ], "volumes": [ f"{prefix}-zk-data", f"{prefix}-zk-log", f"{prefix}-ch-data", f"{prefix}-server-data", ], "workspace": f"/tmp/{prefix}-private", }, "listener": {"host": "127.0.0.1", "port": port}, "authentication_mode": "isolated_session_bearer", }, "production_before": production_before, "image_inventory_before": ["sha256:test|test:fixture"], } path = tmp_path / "isolation.json" path.write_text(json.dumps(value), encoding="utf-8") path.chmod(0o600) return path def restore_command( *, mode: str, bundle: Path, target_base_url: str, credential: Path | None, isolation: Path, receipt_file: Path | None = None, ) -> list[str]: command = [ sys.executable, str(RESTORE), mode, "--bundle-dir", str(bundle), "--target-base-url", target_base_url, "--auth-mode", "isolated_session_bearer", "--isolation-receipt", str(isolation), "--trace-id", TRACE_ID, "--run-id", RUN_ID, "--work-item-id", WORK_ITEM_ID, ] if credential is not None: command.extend(["--credential-file", str(credential)]) if mode == "--apply": terminal_path = receipt_file or bundle.parent / "restore-terminal.json" command.extend(["--receipt-file", str(terminal_path)]) return command def test_policy_is_explicitly_non_raw_and_non_completion() -> None: policy = json.loads(POLICY.read_text(encoding="utf-8")) assert policy["source_runtime"]["raw_database_access_allowed"] is False assert policy["source_runtime"]["raw_volume_access_allowed"] is False assert policy["source_runtime"]["github_supply_chain_allowed"] is False assert policy["completion_contract"]["portable_export_alone_is_completion"] is False assert ( policy["completion_contract"]["full_sqlite_completion_claim_allowed"] is False ) assert {item["path"] for item in policy["forbidden_endpoints"]} >= { "/api/v1/pats", "/api/v1/domains", "/api/v1/channels", } alert_rules = next(item for item in policy["assets"] if item["id"] == "alert_rules") assert alert_rules["endpoint"] == "/api/v1/rules" assert alert_rules["response_pointer"] == "/data/rules" explorer_views = [ item for item in policy["assets"] if item["id"].startswith("explorer_views_") ] assert {item["query_parameters"]["sourcePage"] for item in explorer_views} == { "logs", "meter", "metrics", "traces", } assert all(item["response_pointer"] == "/data" for item in explorer_views) assert all( item["empty_collection_sentinels"] == ["null", "empty_object"] for item in explorer_views ) def test_policy_rejects_unhashable_empty_collection_sentinel_as_contract_error( tmp_path: Path, ) -> None: policy_value = json.loads(POLICY.read_text(encoding="utf-8")) explorer = next( item for item in policy_value["assets"] if item["id"] == "explorer_views_logs" ) explorer["empty_collection_sentinels"] = [{}] malformed_policy = tmp_path / "malformed-policy.json" malformed_policy.write_text(json.dumps(policy_value), encoding="utf-8") result = subprocess.run( export_command( base_url="https://signoz.invalid", output_dir=tmp_path / "must-not-exist", credential=None, apply=False, policy=malformed_policy, ), text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=10, check=False, ) assert result.returncode == 1 assert "policy_empty_collection_sentinels_invalid_explorer_views_logs" in ( result.stderr ) assert "Traceback" not in result.stderr def test_policy_pins_reviewed_isolated_restore_runtime() -> None: policy = json.loads(POLICY.read_text(encoding="utf-8")) isolation = policy["restore_isolation"] assert isolation["required_image_id"] == ( "sha256:c17991d10966aedcb0d4d83805a0baf5310f1553d90611629036fcfaee8d969b" ) assert isolation["startup_contract"]["reviewed"] is True assert isolation["startup_contract"]["server_entrypoint"] == [ "./signoz", "server", ] assert isolation["startup_contract"]["server_environment"] == { "SIGNOZ_TELEMETRYSTORE_CLICKHOUSE_DSN": "tcp://clickhouse:9000", "SIGNOZ_SQLSTORE_SQLITE_PATH": "/var/lib/signoz/signoz.db", } def test_check_mode_writes_nothing(tmp_path: Path) -> None: output = tmp_path / "must-not-exist" result = subprocess.run( export_command( base_url="https://signoz.invalid", output_dir=output, credential=None, apply=False, ), text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=10, check=False, ) assert result.returncode == 0, result.stderr assert json.loads(result.stdout)["terminal"] == "check_pass_no_write" assert not output.exists() def test_apply_and_independent_verifier_pass(tmp_path: Path) -> None: bundle, _ = create_bundle(tmp_path) result = subprocess.run( [ sys.executable, str(VERIFIER), "--bundle-dir", str(bundle), "--expected-trace-id", TRACE_ID, "--expected-run-id", RUN_ID, "--expected-work-item-id", WORK_ITEM_ID, ], text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=10, check=False, ) assert result.returncode == 0, result.stderr assert json.loads(result.stdout)["terminal"] == ( "pass_export_verified_restore_pending" ) manifest = json.loads((bundle / "manifest.json").read_text(encoding="utf-8")) assert manifest["stable_read_count"] == 2 assert manifest["raw_sqlite_read"] is False assert manifest["sensitive_value_persisted"] is False assert manifest["completion_claim"] is False assert oct((bundle / "manifest.json").stat().st_mode & 0o777) == "0o600" def test_explorer_views_are_source_scoped_and_empty_sentinels_are_bounded( tmp_path: Path, ) -> None: credential = credential_file(tmp_path) bundle = tmp_path / "bundle" with metadata_server(SOURCE_ASSETS) as (base_url, handler): result = subprocess.run( export_command( base_url=base_url, output_dir=bundle, credential=credential, apply=True, ), text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=20, check=False, ) assert result.returncode == 0, result.stderr expected_keys = { f"/api/v1/explorer/views?sourcePage={source_page}" for source_page in ("logs", "meter", "metrics", "traces") } assert expected_keys <= set(handler.get_counts) assert all(handler.get_counts[key] == 2 for key in expected_keys) assert "/api/v1/explorer/views" not in handler.get_counts manifest = json.loads((bundle / "manifest.json").read_text(encoding="utf-8")) view_counts = { item["id"]: item["item_count"] for item in manifest["assets"] if item["id"].startswith("explorer_views_") } assert view_counts == { "explorer_views_logs": 1, "explorer_views_meter": 0, "explorer_views_metrics": 0, "explorer_views_traces": 0, } view_queries = { item["id"]: item["query_parameters"] for item in manifest["assets"] if item["id"].startswith("explorer_views_") } assert view_queries == { "explorer_views_logs": {"sourcePage": "logs"}, "explorer_views_meter": {"sourcePage": "meter"}, "explorer_views_metrics": {"sourcePage": "metrics"}, "explorer_views_traces": {"sourcePage": "traces"}, } def test_explorer_views_nonempty_scalar_fails_closed(tmp_path: Path) -> None: assets = json.loads(json.dumps(SOURCE_ASSETS)) assets["/api/v1/explorer/views?sourcePage=metrics"]["data"] = "unexpected" output = tmp_path / "bundle" credential = credential_file(tmp_path) with metadata_server(assets) as (base_url, _): result = subprocess.run( export_command( base_url=base_url, output_dir=output, credential=credential, apply=True, ), text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=20, check=False, ) assert result.returncode == 1 assert "asset_collection_shape_invalid_explorer_views_metrics" in result.stderr assert not output.exists() def test_explorer_view_item_must_match_scoped_source_page(tmp_path: Path) -> None: assets = json.loads(json.dumps(SOURCE_ASSETS)) assets["/api/v1/explorer/views?sourcePage=logs"]["data"][0]["sourcePage"] = "traces" output = tmp_path / "bundle" credential = credential_file(tmp_path) with metadata_server(assets) as (base_url, _): result = subprocess.run( export_command( base_url=base_url, output_dir=output, credential=credential, apply=True, ), text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=20, check=False, ) assert result.returncode == 1 assert "asset_collection_scope_invalid_explorer_views_logs" in result.stderr assert not output.exists() def test_secret_key_fails_closed_without_bundle(tmp_path: Path) -> None: assets = json.loads(json.dumps(SOURCE_ASSETS)) assets["/api/v1/global/config"]["data"]["clientSecret"] = "must-not-persist" output = tmp_path / "bundle" credential = credential_file(tmp_path) with metadata_server(assets) as (base_url, _): result = subprocess.run( export_command( base_url=base_url, output_dir=output, credential=credential, apply=True, ), text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=20, check=False, ) assert result.returncode == 1 assert "forbidden_key_detected" in result.stderr assert "must-not-persist" not in result.stderr assert not output.exists() def test_source_drift_fails_closed_without_bundle(tmp_path: Path) -> None: output = tmp_path / "bundle" credential = credential_file(tmp_path) with metadata_server(SOURCE_ASSETS, drift_path="/api/v1/dashboards") as ( base_url, _, ): result = subprocess.run( export_command( base_url=base_url, output_dir=output, credential=credential, apply=True, ), text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=20, check=False, ) assert result.returncode == 1 assert "source_metadata_drift_between_stable_reads" in result.stderr assert not output.exists() def test_tamper_breaks_independent_verifier(tmp_path: Path) -> None: bundle, _ = create_bundle(tmp_path) asset = bundle / "assets" / "dashboards.json" asset.write_text('{"data":[]}\n', encoding="utf-8") asset.chmod(0o600) result = subprocess.run( [ sys.executable, str(VERIFIER), "--bundle-dir", str(bundle), "--expected-trace-id", TRACE_ID, "--expected-run-id", RUN_ID, ], text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=10, check=False, ) assert result.returncode == 1 assert "manifest_asset_hash_mismatch_dashboards" in result.stderr def test_unexpected_bundle_file_breaks_independent_verifier(tmp_path: Path) -> None: bundle, _ = create_bundle(tmp_path) unexpected = bundle / "unlisted.json" unexpected.write_text('{"not":"manifested"}\n', encoding="utf-8") unexpected.chmod(0o600) result = subprocess.run( [ sys.executable, str(VERIFIER), "--bundle-dir", str(bundle), "--expected-trace-id", TRACE_ID, "--expected-run-id", RUN_ID, ], text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=10, check=False, ) assert result.returncode == 1 assert "bundle_tree_contains_unexpected_entries" in result.stderr def test_restore_check_and_semantic_apply_are_bounded(tmp_path: Path) -> None: bundle, credential = create_bundle(tmp_path) target_assets = { "/api/v1/dashboards": {"data": []}, "/api/v1/rules": {"status": "success", "data": {"rules": []}}, "/api/v1/explorer/views?sourcePage=logs": {"data": []}, "/api/v1/explorer/views?sourcePage=meter": {"data": None}, "/api/v1/explorer/views?sourcePage=metrics": {"data": {}}, "/api/v1/explorer/views?sourcePage=traces": {"data": []}, } with metadata_server(target_assets) as (target_base_url, handler): isolation = isolation_receipt(tmp_path, target_base_url) check_result = subprocess.run( restore_command( mode="--check", bundle=bundle, target_base_url=target_base_url, credential=credential, isolation=isolation, ), text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=20, check=False, ) assert check_result.returncode == 0, check_result.stderr assert json.loads(check_result.stdout)["terminal"] == "check_pass_no_write" assert handler.post_count == 0 apply_result = subprocess.run( restore_command( mode="--apply", bundle=bundle, target_base_url=target_base_url, credential=credential, isolation=isolation, ), text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=20, check=False, ) assert apply_result.returncode == 0, apply_result.stderr assert json.loads(apply_result.stdout)["terminal"] == ( "partial_degraded_cleanup_pending" ) apply_receipt = json.loads(apply_result.stdout) assert apply_receipt["restored_asset_counts"] == { "dashboards": 1, "alert_rules": 1, "explorer_views_logs": 1, "explorer_views_meter": 0, "explorer_views_metrics": 0, "explorer_views_traces": 0, } assert len(apply_receipt["portable_semantic_sha256"]) == 64 assert apply_receipt["completion_scope"] == "portable_assets_only" assert apply_receipt["full_sqlite_completion_claim"] is False phase_receipts = apply_receipt["phase_receipts"] assert [item["phase"] for item in phase_receipts] == [ "sensor_source", "normalized_asset_identity", "source_of_truth_diff", "ai_decision", "risk_policy_decision", "check_mode", "bounded_execution", "independent_post_verifier", "rollback", "learning_writeback", ] assert phase_receipts[-1]["terminal"] == "pending" assert handler.post_count == 3 def test_isolated_controller_invokes_real_restore_driver_contract( tmp_path: Path, ) -> None: controller = load_isolated_controller() bundle, credential = create_bundle(tmp_path) target_assets = { "/api/v1/dashboards": {"data": []}, "/api/v1/rules": {"status": "success", "data": {"rules": []}}, "/api/v1/explorer/views?sourcePage=logs": {"data": []}, "/api/v1/explorer/views?sourcePage=meter": {"data": None}, "/api/v1/explorer/views?sourcePage=metrics": {"data": {}}, "/api/v1/explorer/views?sourcePage=traces": {"data": []}, } args = SimpleNamespace( bundle_dir=bundle, policy=POLICY, trace_id=TRACE_ID, run_id=RUN_ID, work_item_id=WORK_ITEM_ID, ) def real_child_runner(argv: list[str], timeout: int) -> Any: assert argv[0] == controller.PYTHON completed = subprocess.run( [sys.executable, *argv[1:]], stdin=subprocess.DEVNULL, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, timeout=timeout, check=False, ) return controller.CommandResult(completed.returncode, completed.stdout) with metadata_server(target_assets) as (target_base_url, handler): isolation = isolation_receipt(tmp_path, target_base_url) check = controller.invoke_restore_driver( args, real_child_runner, target_base_url, isolation, credential, "--check", ) apply = controller.invoke_restore_driver( args, real_child_runner, target_base_url, isolation, credential, "--apply", ) assert check["terminal"] == "check_pass_no_write" assert apply["terminal"] == "partial_degraded_cleanup_pending" assert set(apply["restored_asset_counts"]) == set(controller.PORTABLE_ASSET_IDS) assert handler.post_count == 3 child_receipt = tmp_path / "restore-driver-apply-receipt.json" assert child_receipt.is_file() assert child_receipt.read_bytes() == controller.canonical_json_bytes(apply) def test_restore_rejects_non_loopback_target_before_writes(tmp_path: Path) -> None: bundle, credential = create_bundle(tmp_path) target = "https://signoz.example.invalid" isolation = isolation_receipt(tmp_path, target) result = subprocess.run( restore_command( mode="--apply", bundle=bundle, target_base_url=target, credential=credential, isolation=isolation, ), text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=20, check=False, ) assert result.returncode == 1 assert "restore_target_must_be_loopback" in result.stderr def test_restore_rejects_non_exact_isolation_image_before_writes( tmp_path: Path, ) -> None: bundle, credential = create_bundle(tmp_path) with socket.socket() as sock: sock.bind(("127.0.0.1", 0)) port = sock.getsockname()[1] target = f"http://127.0.0.1:{port}" isolation = isolation_receipt(tmp_path, target) value = json.loads(isolation.read_text(encoding="utf-8")) value["target"]["image_id"] = f"sha256:{'a' * 64}" isolation.write_text(json.dumps(value), encoding="utf-8") isolation.chmod(0o600) result = subprocess.run( restore_command( mode="--apply", bundle=bundle, target_base_url=target, credential=credential, isolation=isolation, ), text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=20, check=False, ) assert result.returncode == 1 assert "isolation_target_image_id_mismatch" in result.stderr def test_cleanup_verifier_requires_zero_labeled_residue(tmp_path: Path) -> None: bundle, _ = create_bundle(tmp_path) fake_bin = tmp_path / "bin" fake_bin.mkdir() docker = fake_bin / "docker" policy = json.loads(POLICY.read_text(encoding="utf-8")) production = { entry["container"]: { "Id": hashlib.sha256(entry["container"].encode()).hexdigest(), "Image": entry["image_id"], "Config": {"Image": entry["image_ref"]}, "State": { "Running": True, "StartedAt": "2026-07-22T00:00:00Z", }, "RestartCount": 0, } for entry in policy["restore_isolation"]["production_continuity"].values() } docker.write_text( "#!/usr/bin/env python3\n" "import json, sys\n" f"production = {production!r}\n" "args = sys.argv[1:]\n" "if args[:2] == ['container', 'inspect'] and args[2] in production:\n" " print(json.dumps([production[args[2]]]))\n" " raise SystemExit(0)\n" "if args[:3] == ['image', 'ls', '--no-trunc']:\n" " print('sha256:test|test:fixture')\n" " raise SystemExit(0)\n" "if len(args) >= 2 and args[1] == 'inspect':\n" " raise SystemExit(1)\n" "raise SystemExit(0)\n", encoding="utf-8", ) docker.chmod(0o755) with socket.socket() as sock: sock.bind(("127.0.0.1", 0)) port = sock.getsockname()[1] target = f"http://127.0.0.1:{port}" isolation = isolation_receipt(tmp_path, target) env = {**os.environ, "PATH": f"{fake_bin}:{os.environ['PATH']}"} result = subprocess.run( restore_command( mode="--verify-cleanup", bundle=bundle, target_base_url=target, credential=None, isolation=isolation, ), env=env, text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=20, check=False, ) assert result.returncode == 0, result.stderr assert json.loads(result.stdout)["terminal"] == "pass_zero_residue_verified" def test_credential_reference_mode_fails_before_network(tmp_path: Path) -> None: credential = credential_file(tmp_path, mode=0o644) output = tmp_path / "bundle" result = subprocess.run( export_command( base_url="https://signoz.invalid", output_dir=output, credential=credential, apply=True, ), text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=10, check=False, ) assert result.returncode == 1 assert "credential_reference_mode_must_be_0400_or_0600" in result.stderr assert not output.exists() def test_credential_symlink_fails_before_network(tmp_path: Path) -> None: credential = credential_file(tmp_path) link = tmp_path / "credential-link" link.symlink_to(credential) output = tmp_path / "bundle" result = subprocess.run( export_command( base_url="https://signoz.invalid", output_dir=output, credential=link, apply=True, ), text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=10, check=False, ) assert result.returncode == 1 assert "credential_reference_symlink_component_forbidden" in result.stderr assert not output.exists() def test_apply_failure_writes_terminal_receipt_without_secret(tmp_path: Path) -> None: assets = json.loads(json.dumps(SOURCE_ASSETS)) secret_value = "must-not-appear-in-terminal-receipt" assets["/api/v1/global/config"]["data"]["clientSecret"] = secret_value output = tmp_path / "bundle" receipt_file = tmp_path / "terminal.json" credential = credential_file(tmp_path) with metadata_server(assets) as (base_url, _): command = export_command( base_url=base_url, output_dir=output, credential=credential, apply=True, receipt_file=receipt_file, ) result = subprocess.run( command, text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=20, check=False, ) assert result.returncode == 1 terminal = json.loads(receipt_file.read_text(encoding="utf-8")) assert terminal["terminal"] == "failed_export_not_closable" assert secret_value not in receipt_file.read_text(encoding="utf-8") assert not output.exists()