369 lines
13 KiB
Python
369 lines
13 KiB
Python
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
|