from __future__ import annotations from sqlmodel import Session, SQLModel, create_engine from backend.app.db.models import DeploymentJob from backend.app.schemas.enrollment import EnrollmentTrustConfigV1 from backend.app.schemas.forge_bootstrap import ForgeBootstrapPreflightV1 from backend.app.services.enrollment_plane import EnrollmentPlane from backend.app.services.forge_bootstrap_preflight import ( ForgeBootstrapPreflightService, ) def _profile(**overrides): value = { "api_version": "itworx.forge-bootstrap-preflight/v1", "revision": "a" * 40, "workspace_image_digest": "sha256:" + "b" * 64, "frontdoor_image_digest": "sha256:" + "c" * 64, } value.update(overrides) return ForgeBootstrapPreflightV1.model_validate(value) def _session(tmp_path): engine = create_engine(f"sqlite:///{tmp_path / 'forge-preflight.db'}") SQLModel.metadata.create_all(engine) return Session(engine) def test_forge_profile_is_strict_and_cannot_select_runtime_inputs(): _profile() for key, value in ( ("host_path", "/etc"), ("endpoint", "https://attacker.invalid"), ("secret_ref", "OTHER_SECRET"), ("command", "docker compose up"), ("repository", "other/repo"), ("target", "other-host"), ): try: _profile(**{key: value}) except ValueError: pass else: # pragma: no cover - assertion clarity raise AssertionError(f"caller-controlled {key} must be rejected") for field, value in ( ("revision", "a" * 39), ("workspace_image_digest", "latest"), ("frontdoor_image_digest", "sha256:" + "D" * 64), ): try: _profile(**{field: value}) except ValueError: pass else: # pragma: no cover - assertion clarity raise AssertionError(f"invalid {field} must be rejected") class _Broker: def __init__(self, result): self.result = result self.calls = [] def forge_bootstrap_preflight(self, profile, *, secret): self.calls.append((profile.model_dump(mode="json"), secret)) return self.result def _trust(session, *, root="/var/lib/docker/volumes/itworx-managed-appdata/_data"): EnrollmentPlane.set_trust_config( session, EnrollmentTrustConfigV1( controller_identity="tower-deployment-controller", bootstrap_secret_ref="CONTROLLER_BOOTSTRAP", trusted_appdata_root=root, managed_appdata_backup_acknowledged=( root == "/var/lib/docker/volumes/itworx-managed-appdata/_data" ), approved_networks=("itworx-mcp-hub-npm",), ), ) def test_forge_preflight_uses_stored_trust_and_preserves_unknowns(tmp_path, monkeypatch): from backend.app.services import forge_bootstrap_preflight as service_module broker = _Broker( { "apiVersion": "itworx.forge-bootstrap-preflight/v1", "status": "preflight_blocked", "revision": "a" * 40, "workspaceImageDigest": "sha256:" + "b" * 64, "frontdoorImageDigest": "sha256:" + "c" * 64, "checks": { "exact_main": {"status": "passed"}, "source_manifest": {"status": "passed"}, "managed_appdata": {"status": "blocked"}, "proxy_network": {"status": "passed"}, "resource_collisions": {"status": "passed"}, "push_validation": {"status": "unknown"}, "image_registry": {"status": "unknown"}, "oidc": {"status": "unknown"}, "secret_delivery": {"status": "unknown"}, }, } ) monkeypatch.setattr(service_module, "get_secret", lambda *_args, **_kwargs: "s" * 48) monkeypatch.setattr(service_module, "has_secret", lambda *_args, **_kwargs: True) with _session(tmp_path) as session: _trust(session) result = ForgeBootstrapPreflightService(broker=broker).preflight( session, _profile(), idempotency_key="forge-preflight-test-0001" ) assert result["status"] == "preflight_failed" assert result["preflight"]["status"] == "blocked" assert "push_validation" in { item["name"] for item in result["preflight"]["checks"] if item["status"] == "unknown" } assert broker.calls == [(_profile().model_dump(mode="json"), "s" * 48)] job = session.get(DeploymentJob, result["id"]) assert job is not None assert job.status == "preflight_failed" assert ForgeBootstrapPreflightService().refresh(session, job.id)["id"] == job.id assert any(item["id"] == job.id for item in ForgeBootstrapPreflightService.list_jobs(session)) def test_forge_preflight_idempotency_rejects_profile_change(tmp_path, monkeypatch): from backend.app.services import forge_bootstrap_preflight as service_module broker = _Broker( { "apiVersion": "itworx.forge-bootstrap-preflight/v1", "status": "preflight_unknown", "revision": "a" * 40, "workspaceImageDigest": "sha256:" + "b" * 64, "frontdoorImageDigest": "sha256:" + "c" * 64, "checks": { "exact_main": {"status": "unknown"}, "source_manifest": {"status": "unknown"}, "managed_appdata": {"status": "unknown"}, "proxy_network": {"status": "unknown"}, "resource_collisions": {"status": "unknown"}, "push_validation": {"status": "unknown"}, "image_registry": {"status": "unknown"}, "oidc": {"status": "unknown"}, "secret_delivery": {"status": "unknown"}, }, } ) monkeypatch.setattr(service_module, "get_secret", lambda *_args, **_kwargs: "s" * 48) monkeypatch.setattr(service_module, "has_secret", lambda *_args, **_kwargs: True) with _session(tmp_path) as session: _trust(session) service = ForgeBootstrapPreflightService(broker=broker) service.preflight(session, _profile(), idempotency_key="forge-preflight-test-0002") replay = service.preflight(session, _profile(), idempotency_key="forge-preflight-test-0002") assert replay["idempotentReplay"] is True assert len(broker.calls) == 1 try: service.preflight( session, _profile(revision="d" * 40), idempotency_key="forge-preflight-test-0002", ) except ValueError as exc: assert "different Forge profile" in str(exc) else: # pragma: no cover raise AssertionError("idempotency key reuse with changed Forge profile must fail") def test_legacy_storage_trust_still_collects_read_only_controller_blockers(tmp_path, monkeypatch): from backend.app.services import forge_bootstrap_preflight as service_module checks = { name: {"status": "unknown"} for name in ( "exact_main", "source_manifest", "managed_appdata", "proxy_network", "resource_collisions", "push_validation", "image_registry", "oidc", "secret_delivery", ) } checks["managed_appdata"] = {"status": "blocked"} broker = _Broker( { "apiVersion": "itworx.forge-bootstrap-preflight/v1", "status": "preflight_blocked", "revision": "a" * 40, "workspaceImageDigest": "sha256:" + "b" * 64, "frontdoorImageDigest": "sha256:" + "c" * 64, "checks": checks, } ) monkeypatch.setattr(service_module, "get_secret", lambda *_args, **_kwargs: "s" * 48) monkeypatch.setattr(service_module, "has_secret", lambda *_args, **_kwargs: True) with _session(tmp_path) as session: _trust(session, root="/srv/example/appdata") result = ForgeBootstrapPreflightService(broker=broker).preflight( session, _profile(), idempotency_key="forge-legacy-trust-0001" ) assert broker.calls == [(_profile().model_dump(mode="json"), "s" * 48)] assert "managed_appdata_root_selection" in result["preflight"]["blockers"] controller_checks = { item["name"]: item["status"] for item in result["preflight"]["checks"] } assert controller_checks["managed_appdata"] == "blocked" assert controller_checks["proxy_network"] == "unknown"