from __future__ import annotations import ast import json from copy import deepcopy from types import SimpleNamespace from backend.app.services.repopulse_adoption_preflight import ( _EXPECTED_REGISTRY_KEYS, _EXPECTED_TARGET, _EXPECTED_VALIDATION, _PREFLIGHT_SCRIPT, RepoPulseAdoptionPreflight, evaluate_facts, ) def _healthy_facts() -> dict: return { "registry": { "present": True, "alias_matches": True, "entry_type": "object", "keys": sorted(_EXPECTED_REGISTRY_KEYS), "target": { key: value for key, value in _EXPECTED_TARGET.items() if key != "environment_keys" }, "environment_keys": [], "validation": deepcopy(_EXPECTED_VALIDATION), "repository_id": 64, "webhook_secret_file_matches": True, }, "state": { "schema_version": 3, "repository": "jens/repopulse", "revision": "12431667760d2bce8c7533b57f462aac8d2e289e", "live_revision": "12431667760d2bce8c7533b57f462aac8d2e289e", "live_status": "healthy", "live_legacy_provenance": False, "status": "success", }, "git": { "bare_repo_present": True, "main_revision": "12431667760d2bce8c7533b57f462aac8d2e289e", "live_revision": "12431667760d2bce8c7533b57f462aac8d2e289e", "live_revision_resolves": True, "live_is_ancestor": True, "compose_present_on_main": True, }, "runtime": { "name_matches": True, "compose_project_matches": True, "container_id_sha256": "a" * 64, "image_sha256": "b" * 64, "state": "running", "health": "healthy", "image": "repopulse-repopulse:latest", "network_mode": "repopulse_default", "mounts": [ { "destination": "/data", "type": "volume", "name": "repopulse_data", "read_write": True, }, { "destination": "/scan", "type": "bind", "source": "/mnt/cache/appdata/gitea-deploy-runners/sources", "read_write": False, }, ], "read_only_root": False, "cap_drop_all": False, "no_new_privileges": True, "host_ports": [{"ip": "0.0.0.0", "port": 1254}], }, "volume": {"name_matches": True, "mountpoint_matches": True}, "scan_access": { "runtime_uid": 10001, "runtime_gid": 10001, "root_readable": True, "repopulse_source_readable": True, "privatemuse_source_readable": True, }, "appdata": { "root_is_safe_directory": True, "deploy_env_is_regular_file": True, "managed_env_exists": False, "managed_env_is_regular_file": False, "managed_env_is_private": False, "target_data_exists": False, "target_data_is_symlink": False, "sources_exists": False, "sources_is_safe_directory": True, "unexpected_entry_count": 0, }, "health": {"http_ok": True, "status": "ok"}, "meta": { "http_ok": True, "product_matches": True, "repository_count": 71, "queue": {"queued": 0, "active": 0, "completed": 568, "failed": 0}, "storage": { "database_bytes": 44_408_832, "journal_bytes": 0, "scan_count": 567, "compacted_scan_count": 94, "mirror_count": 69, "mirror_cache_bytes": 3_000_000_000, "worktree_count": 0, }, "database_recovery_status": "healthy", "mutation_authentication": "perimeter_required", }, } def _blocked_codes(report: dict) -> set[str]: return {check["code"] for check in report["checks"] if check["status"] != "passed"} def test_healthy_legacy_facts_pass_compatibility_but_not_target_acceptance() -> None: report = evaluate_facts(_healthy_facts()) assert report["status"] == "preflight_passed" assert report["legacyCompatibility"]["status"] == "passed" assert report["targetAcceptance"]["status"] == "blocked" assert report["repository"] == "jens/repopulse" assert report["readOnly"] is True assert report["adoptionAuthorized"] is False assert report["evidence"]["host_ports"] == [{"ip": "0.0.0.0", "port": 1254}] assert report["evidence"]["operator_token_configured"] is False assert report["evidence"]["queue"] == { "queued": 0, "active": 0, "completed": 568, "failed": 0, } def test_registry_extra_keys_and_webhook_reference_mismatch_block() -> None: facts = _healthy_facts() facts["registry"]["keys"].append("unreviewed_target_option") facts["registry"]["webhook_secret_file_matches"] = False assert "legacy_registry" in _blocked_codes(evaluate_facts(facts)) def test_malformed_registry_key_list_fails_closed() -> None: facts = _healthy_facts() facts["registry"]["keys"] = [[], "branch"] assert "legacy_registry" in _blocked_codes(evaluate_facts(facts)) def test_registry_json_types_must_match_the_fixed_contract() -> None: facts = _healthy_facts() facts["registry"]["target"]["exact_source"] = 1 facts["registry"]["validation"]["validation_gate_version"] = True assert "legacy_registry" in _blocked_codes(evaluate_facts(facts)) def test_missing_registry_state_and_runtime_facts_fail_closed() -> None: facts = _healthy_facts() facts["registry"]["present"] = False facts["state"] = None facts["runtime"] = None blocked = _blocked_codes(evaluate_facts(facts)) assert {"legacy_registry", "controller_state", "runtime_identity_and_health"} <= blocked def test_wrong_live_sha_or_non_ancestor_blocks() -> None: facts = _healthy_facts() facts["git"]["live_is_ancestor"] = False assert "gitea_ancestry" in _blocked_codes(evaluate_facts(facts)) def test_changed_data_volume_or_writable_scan_mount_blocks() -> None: facts = _healthy_facts() facts["runtime"]["mounts"][0]["name"] = "other_volume" facts["runtime"]["mounts"][1]["read_write"] = True assert "legacy_storage_topology" in _blocked_codes(evaluate_facts(facts)) def test_unhealthy_or_exposed_container_blocks() -> None: facts = _healthy_facts() facts["runtime"]["health"] = "unhealthy" facts["runtime"]["host_ports"] = [{"ip": "0.0.0.0", "port": 1254}] report = evaluate_facts(facts) assert report["status"] == "blocked" assert "runtime_identity_and_health" in _blocked_codes(report) assert "host_exposure" in _blocked_codes(report) def test_container_identity_requires_redacted_id_and_image_fingerprints() -> None: facts = _healthy_facts() facts["runtime"]["container_id_sha256"] = None facts["runtime"]["image_sha256"] = "not-a-sha256" report = evaluate_facts(facts) assert report["legacyCompatibility"]["status"] == "blocked" assert "runtime_identity_and_health" in _blocked_codes(report) assert report["evidence"]["container_id_sha256"] is None assert report["evidence"]["container_image_sha256"] is None def test_host_network_container_blocks_even_without_port_bindings() -> None: facts = _healthy_facts() facts["runtime"]["network_mode"] = "host" report = evaluate_facts(facts) assert report["legacyCompatibility"]["status"] == "passed" assert "managed_network" in _blocked_codes(report) def test_data_target_collision_and_unsafe_legacy_runtime_block_target_acceptance_only() -> None: facts = _healthy_facts() facts["appdata"]["target_data_exists"] = True facts["runtime"]["read_only_root"] = False facts["runtime"]["cap_drop_all"] = False facts["runtime"]["no_new_privileges"] = False report = evaluate_facts(facts) assert report["status"] == "preflight_passed" assert report["legacyCompatibility"]["status"] == "passed" blocked = _blocked_codes(report) assert {"managed_data_target", "runtime_hardening"} <= blocked def test_optional_sources_staging_directory_must_be_safe() -> None: facts = _healthy_facts() facts["appdata"]["sources_exists"] = True facts["appdata"]["sources_is_safe_directory"] = True assert evaluate_facts(facts)["legacyCompatibility"]["status"] == "passed" facts["appdata"]["sources_is_safe_directory"] = False assert "legacy_appdata" in _blocked_codes(evaluate_facts(facts)) def test_managed_env_may_be_staged_only_as_private_regular_file() -> None: facts = _healthy_facts() facts["appdata"].update( managed_env_exists=True, managed_env_is_regular_file=True, managed_env_is_private=True, managed_env_contents="SECRET_VALUE", ) report = evaluate_facts(facts) assert report["legacyCompatibility"]["status"] == "passed" assert "managed_environment_file" not in _blocked_codes(report) assert "SECRET_VALUE" not in str(report) facts["appdata"]["managed_env_is_private"] = False report = evaluate_facts(facts) assert "legacy_appdata" in _blocked_codes(report) assert "managed_environment_file" in _blocked_codes(report) assert report["evidence"]["managed_environment_file_ready"] is False def test_scan_access_is_a_managed_acceptance_gate_for_runtime_uid() -> None: facts = _healthy_facts() facts["scan_access"]["repopulse_source_readable"] = False report = evaluate_facts(facts) assert report["legacyCompatibility"]["status"] == "passed" assert "scan_access" in _blocked_codes(report) assert report["evidence"]["scan_access_uid_10001"] is False def test_missing_or_busy_queue_and_bad_database_recovery_block() -> None: facts = _healthy_facts() facts["meta"]["queue"]["active"] = 1 facts["meta"]["database_recovery_status"] = "unknown" blocked = _blocked_codes(evaluate_facts(facts)) assert {"queue_quiescence", "database_health"} <= blocked facts["meta"]["queue"].pop("failed") assert "queue_quiescence" in _blocked_codes(evaluate_facts(facts)) def test_report_never_echoes_untrusted_or_secret_values() -> None: facts = _healthy_facts() facts["state"]["last_attempt"] = {"message": "TOP_SECRET_VALUE"} facts["meta"]["gitea"] = {"token": "TOP_SECRET_VALUE"} facts["runtime"]["operator_token"] = "TOP_SECRET_VALUE" report_text = json.dumps(evaluate_facts(facts)) assert "TOP_SECRET_VALUE" not in report_text class _FakeResponse: def __init__(self, *, payload=None, text=""): self.payload = payload self.text = text def json(self): return self.payload class _FakePortainer: def __init__(self, facts): self.facts = facts self.calls = [] def get_container_status(self, name): assert name == "gitea-deploy-control" return { "status": "ok", "container": {"id": "controller-id", "name": name, "state": "running"}, } def _config(self): return SimpleNamespace(endpoint_id="3") def _request(self, method, path, **kwargs): self.calls.append((method, path, kwargs)) if path.endswith("/containers/controller-id/exec"): return _FakeResponse(payload={"Id": "exec-1"}) if path.endswith("/exec/exec-1/start"): return _FakeResponse(text=json.dumps(self.facts)) raise AssertionError(path) def test_live_service_uses_only_fixed_read_only_collector() -> None: fake = _FakePortainer(_healthy_facts()) report = RepoPulseAdoptionPreflight(fake).run() assert report["status"] == "preflight_passed" assert len(fake.calls) == 2 create = fake.calls[0] assert create[0] == "POST" assert create[2]["json"]["Cmd"] == ["python3", "-c", _PREFLIGHT_SCRIPT] assert create[2]["json"]["AttachStdout"] is True assert create[2]["json"]["AttachStderr"] is False assert create[2]["json"]["Tty"] is True start = fake.calls[1] assert start[2]["json"] == {"Detach": False, "Tty": True} assert "docker rm" not in _PREFLIGHT_SCRIPT assert "docker start" not in _PREFLIGHT_SCRIPT assert "write_text(" not in _PREFLIGHT_SCRIPT def test_fixed_collector_source_is_valid_python() -> None: ast.parse(_PREFLIGHT_SCRIPT) def test_collector_failure_is_sanitized() -> None: class BrokenPortainer(_FakePortainer): def _request(self, method, path, **kwargs): if path.endswith("/containers/controller-id/exec"): return _FakeResponse(payload={"Id": "exec-1"}) return _FakeResponse(text="not json with secret=TOP_SECRET_VALUE") report = RepoPulseAdoptionPreflight(BrokenPortainer(_healthy_facts())).run() report_text = json.dumps(report) assert report["status"] == "unavailable" assert "TOP_SECRET_VALUE" not in report_text