from __future__ import annotations import hashlib import http.client import importlib.util import json import sys import threading import types from datetime import datetime, timezone from pathlib import Path import httpx import pytest from fastapi.testclient import TestClient from sqlmodel import Session, SQLModel, create_engine from backend.app.core.time import utc_now from backend.app.db.models import DeploymentJob from backend.app.schemas.forge_runtime_readiness import ( ForgeRuntimeReadinessResultV1, ForgeRuntimeReadinessV1, ) from backend.app.services import user_access _REAL_CURRENT_USER = user_access.current_user _JOB_ID = "forge-install-" + "a" * 64 _REQUEST = "b" * 64 _REVISION = "c" * 40 _WORKSPACE_DIGEST = "sha256:" + "d" * 64 _FRONTDOOR_DIGEST = "sha256:" + "e" * 64 _SERVICES = ("api", "dashboard", "itworx-forge", "postgres", "worker") _ARTIFACT = { "controllerVersion": "1.2.3", "sourceRepository": "jens/unraid-appops-gateway", "sourceRevision": "f" * 40, "sourceTree": "1" * 40, "artifactSha256": "2" * 64, } _ROOT = Path(__file__).resolve().parents[2] def _profile() -> dict[str, str]: return { "api_version": "itworx.forge-first-install/v1", "revision": _REVISION, "workspace_image_digest": _WORKSPACE_DIGEST, "frontdoor_image_digest": _FRONTDOOR_DIGEST, "preflight_job_id": "forge-preflight-" + "3" * 16, "preflight_sha256": "4" * 64, } def _runtime_payload(*, nonce: str = "5" * 32, observed_at: int | None = None) -> dict: return { "apiVersion": "itworx.forge-runtime-readiness/v1", "status": "ready", "repository": "jens/itworx-forge", "operation": "forge_first_install", "request": _REQUEST, "profile": { "revision": _REVISION, "workspaceImageDigest": _WORKSPACE_DIGEST, "frontdoorImageDigest": _FRONTDOOR_DIGEST, }, "observedAt": int(datetime.now(timezone.utc).timestamp()) if observed_at is None else observed_at, "nonce": nonce, "health": { "status": "healthy", "services": [{"name": name, "status": "healthy"} for name in _SERVICES], }, "migrationCompleted": True, "controllerArtifact": dict(_ARTIFACT), } def test_readiness_schema_requires_exact_service_set_and_aggregate_consistency(): parsed = ForgeRuntimeReadinessV1.model_validate(_runtime_payload()) assert parsed.status == "ready" unavailable_child = _runtime_payload() unavailable_child["health"]["services"][0]["status"] = "unavailable" with pytest.raises(ValueError): ForgeRuntimeReadinessV1.model_validate(unavailable_child) duplicate_service = _runtime_payload() duplicate_service["health"]["services"][0]["name"] = "worker" with pytest.raises(ValueError): ForgeRuntimeReadinessV1.model_validate(duplicate_service) boolean_timestamp = _runtime_payload() boolean_timestamp["observedAt"] = True with pytest.raises(ValueError): ForgeRuntimeReadinessV1.model_validate(boolean_timestamp) unexpected_field = _runtime_payload() unexpected_field["diagnostic"] = "must never be projected" with pytest.raises(ValueError): ForgeRuntimeReadinessV1.model_validate(unexpected_field) @pytest.mark.parametrize("status", ["ready", "not_ready"]) def test_public_readiness_requires_bound_identity_for_observed_results(status): result = { "status": status, "message": f"forge_runtime_{status}", "health": { "status": "healthy", "services": [{"name": name, "status": "healthy"} for name in _SERVICES], }, "migrationCompleted": status == "ready", } with pytest.raises(ValueError, match="identity is incomplete"): ForgeRuntimeReadinessResultV1.model_validate(result) def test_public_readiness_allows_anonymous_fixed_unavailable_projection(): result = ForgeRuntimeReadinessResultV1.model_validate( { "status": "unavailable", "message": "forge_runtime_readiness_unavailable", "health": { "status": "unavailable", "services": [ {"name": name, "status": "unavailable"} for name in _SERVICES ], }, "migrationCompleted": False, } ) assert result.status == "unavailable" assert result.job_id is None def test_readiness_broker_sends_one_fixed_fresh_hmac_request(monkeypatch): from backend.app.services import deployment_broker from backend.app.services.deployment_broker import DeploymentBrokerClient observed = {} secret = "s" * 48 nonce = "6" * 32 payload = _runtime_payload(nonce=nonce, observed_at=2000) body = json.dumps(payload, separators=(",", ":")).encode() monotonic_now = [0.0] monkeypatch.setattr(deployment_broker.time, "time", lambda: 2000) monkeypatch.setattr(deployment_broker.time, "monotonic", lambda: monotonic_now[0]) monkeypatch.setattr( deployment_broker.uuid, "uuid4", lambda: types.SimpleNamespace(hex=nonce), ) class Response: status_code = 200 headers = httpx.Headers( {"content-length": str(len(body)), "content-type": "application/json"} ) def __enter__(self): return self def __exit__(self, *_args): return False def iter_raw(self): yield body class Client: def __init__(self, *, timeout, verify, follow_redirects): observed["client"] = (timeout, verify, follow_redirects) def __enter__(self): return self def __exit__(self, *_args): return False def stream(self, method, url, *, content, headers): observed.update(method=method, url=url, content=content, headers=headers) # The fixed controller may spend fifteen seconds completing its # six-container snapshot before it can send response headers. monotonic_now[0] = 16.0 return Response() monkeypatch.setattr(deployment_broker.httpx, "Client", Client) broker = DeploymentBrokerClient(base_url="https://controller.example", verify_tls=True) result = broker.forge_bootstrap_readiness(_REQUEST, secret=secret) path = f"/v1/appops/forge/bootstrap/readiness/{_REQUEST}" assert result == payload assert isinstance(observed["client"][0], httpx.Timeout) assert observed["client"][0].read == 20.0 assert monotonic_now[0] == 16.0 assert observed["client"][1:] == (True, False) assert observed["method"] == "GET" assert observed["url"] == f"https://controller.example{path}" assert observed["content"] == b"" assert observed["headers"]["Accept-Encoding"] == "identity" assert observed["headers"]["Cache-Control"] == "no-store" assert observed["headers"]["Pragma"] == "no-cache" assert observed["headers"]["X-ITWorx-Deployment-Signature"] == broker._signature( secret, "GET", path, "2000", nonce, b"", ) @pytest.mark.parametrize( "failure", [ "redirect", "stale", "wrong_nonce", "duplicate", "boolean_timestamp", "future", "oversized", "long_length", "stream_oversized", "slow_drip", "invalid_utf8", "deeply_nested", ], ) def test_readiness_broker_fails_closed_on_unfresh_or_ambiguous_response( monkeypatch, failure ): from backend.app.services import deployment_broker from backend.app.services.deployment_broker import DeploymentBrokerClient nonce = "7" * 32 current_time = 5000 payload = _runtime_payload(nonce=nonce, observed_at=current_time) body = json.dumps(payload, separators=(",", ":")).encode() status_code = 302 if failure == "redirect" else 200 if failure == "stale": payload["observedAt"] = current_time - 31 body = json.dumps(payload, separators=(",", ":")).encode() elif failure == "wrong_nonce": payload["nonce"] = "8" * 32 body = json.dumps(payload, separators=(",", ":")).encode() elif failure == "duplicate": body = body[:-1] + b',"status":"ready"}' elif failure == "boolean_timestamp": payload["observedAt"] = True body = json.dumps(payload, separators=(",", ":")).encode() elif failure == "future": payload["observedAt"] = current_time + 6 body = json.dumps(payload, separators=(",", ":")).encode() content_length = str(len(body)) if failure == "oversized": content_length = str(32 * 1024 + 1) elif failure == "long_length": content_length = "9" * 11 elif failure == "stream_oversized": content_length = None body = b"x" * (32 * 1024 + 1) elif failure == "invalid_utf8": body = b"\xff" elif failure == "deeply_nested": body = b"[" * 1200 + b"]" * 1200 monkeypatch.setattr(deployment_broker.time, "time", lambda: current_time) monkeypatch.setattr( deployment_broker.uuid, "uuid4", lambda: types.SimpleNamespace(hex=nonce), ) if failure == "slow_drip": clock_values = iter((0.0, 0.0, 36.0)) monkeypatch.setattr(deployment_broker.time, "monotonic", lambda: next(clock_values)) response_headers = {"content-type": "application/json"} if content_length is not None: response_headers["content-length"] = content_length class Response: headers = httpx.Headers(response_headers) def __init__(self): self.status_code = status_code def __enter__(self): return self def __exit__(self, *_args): return False def iter_raw(self): if failure == "slow_drip": yield body[:8] yield body[8:] else: yield body class Client: def __init__(self, **_kwargs): pass def __enter__(self): return self def __exit__(self, *_args): return False def stream(self, *_args, **_kwargs): return Response() monkeypatch.setattr(deployment_broker.httpx, "Client", Client) result = DeploymentBrokerClient( base_url="https://controller.example", verify_tls=True ).forge_bootstrap_readiness(_REQUEST, secret="s" * 48) assert result == {"status": "unavailable"} class _ReadinessBroker: def __init__(self, response): self.response = response self.calls = [] def forge_bootstrap_readiness(self, request_id, *, secret): self.calls.append((request_id, secret)) return self.response def _deployed_job(session, *, overrides=None): plan = _profile() if overrides: plan.update(overrides) canonical_plan = json.dumps(plan, sort_keys=True, separators=(",", ":")) historical = { "apiVersion": "itworx.forge-first-install/v1", "status": "deployed", "request": _REQUEST, "repository": "jens/itworx-forge", "revision": _REVISION, "operation": "forge_first_install", "workspaceImageDigest": _WORKSPACE_DIGEST, "frontdoorImageDigest": _FRONTDOOR_DIGEST, "receipt": { "revision": _REVISION, "workspaceImageDigest": _WORKSPACE_DIGEST, "frontdoorImageDigest": _FRONTDOOR_DIGEST, "migration": {"service": "migrate", "status": "completed", "result": "succeeded"}, "health": {"status": "healthy", "services": list(_SERVICES)}, }, } row = DeploymentJob( id=_JOB_ID, application_id="itworx-forge", repository="jens/itworx-forge", revision=_REVISION, plan_sha256=hashlib.sha256(canonical_plan.encode()).hexdigest(), idempotency_sha256="9" * 64, status="deployed", stage="controller_deployed", plan_json=plan, preflight_json={}, evidence_json={"profile": "forge-first-install", "controller": historical}, broker_request_id=_REQUEST, result_summary="completed", completed_at=utc_now(), ) session.add(row) session.commit() session.refresh(row) return row def test_historical_deployed_receipt_alone_never_means_current_runtime_ready( tmp_path, monkeypatch ): from backend.app.services import forge_bootstrap_apply from backend.app.services.forge_bootstrap_apply import ForgeBootstrapApplyService engine = create_engine(f"sqlite:///{tmp_path / 'readiness-historical.db'}") SQLModel.metadata.create_all(engine) with Session(engine) as session: _deployed_job(session) secret_reads = [] monkeypatch.setattr( forge_bootstrap_apply, "_controller_secret", lambda _session: secret_reads.append(True) or "s" * 48, ) broker = _ReadinessBroker( { "apiVersion": "itworx.forge-first-install/v1", "status": "deployed", "receipt": {"health": {"status": "healthy"}}, } ) result = ForgeBootstrapApplyService(broker=broker).current_readiness( session, _JOB_ID ) assert result["status"] == "unavailable" assert broker.calls == [(_REQUEST, "s" * 48)] assert secret_reads == [True] engine.dispose() def test_readiness_service_validates_current_identity_and_projects_no_nonce(tmp_path, monkeypatch): from backend.app.services import forge_bootstrap_apply from backend.app.services.forge_bootstrap_apply import ForgeBootstrapApplyService engine = create_engine(f"sqlite:///{tmp_path / 'readiness-current.db'}") SQLModel.metadata.create_all(engine) with Session(engine) as session: _deployed_job(session) current_job = session.get(DeploymentJob, _JOB_ID) from backend.app.schemas.forge_bootstrap_apply import ForgeBootstrapApplyV1 from backend.app.services.forge_bootstrap_apply import _classify_controller_result parsed_profile = ForgeBootstrapApplyV1.model_validate(current_job.plan_json) assert _classify_controller_result( current_job.evidence_json["controller"], request_id=_REQUEST, profile=parsed_profile, )[0] == "deployed" assert ForgeRuntimeReadinessV1.model_validate(_runtime_payload()).status == "ready" secret_reads = [] monkeypatch.setattr( forge_bootstrap_apply, "_controller_secret", lambda _session: secret_reads.append(True) or "s" * 48, ) broker = _ReadinessBroker(_runtime_payload()) result = ForgeBootstrapApplyService(broker=broker).current_readiness( session, _JOB_ID ) assert result["status"] == "ready" assert result["jobId"] == _JOB_ID assert result["controllerArtifact"] == _ARTIFACT assert "nonce" not in result assert broker.calls == [(_REQUEST, "s" * 48)] assert secret_reads == [True] engine.dispose() @pytest.mark.parametrize( "mutation", ["request", "revision", "workspace_digest", "artifact", "aggregate", "extra"], ) def test_readiness_service_rejects_unbound_or_malformed_current_response( tmp_path, monkeypatch, mutation ): from backend.app.services import forge_bootstrap_apply from backend.app.services.forge_bootstrap_apply import ForgeBootstrapApplyService engine = create_engine(f"sqlite:///{tmp_path / f'readiness-{mutation}.db'}") SQLModel.metadata.create_all(engine) response = _runtime_payload() if mutation == "request": response["request"] = "0" * 64 elif mutation == "revision": response["profile"]["revision"] = "0" * 40 elif mutation == "workspace_digest": response["profile"]["workspaceImageDigest"] = "sha256:" + "0" * 64 elif mutation == "artifact": response["controllerArtifact"]["sourceRepository"] = "another/repository" elif mutation == "aggregate": response["health"]["services"][0]["status"] = "unavailable" elif mutation == "extra": response["upstreamReason"] = "must not be projected" with Session(engine) as session: _deployed_job(session) monkeypatch.setattr(forge_bootstrap_apply, "_controller_secret", lambda _session: "s" * 48) result = ForgeBootstrapApplyService( broker=_ReadinessBroker(response) ).current_readiness(session, _JOB_ID) assert result["status"] == "unavailable" assert "upstreamReason" not in result engine.dispose() def test_invalid_or_not_deployed_job_never_reads_controller_secret_or_calls_broker( tmp_path, monkeypatch ): from backend.app.services import forge_bootstrap_apply from backend.app.services.forge_bootstrap_apply import ForgeBootstrapApplyService engine = create_engine(f"sqlite:///{tmp_path / 'readiness-invalid.db'}") SQLModel.metadata.create_all(engine) secret_reads = [] monkeypatch.setattr( forge_bootstrap_apply, "_controller_secret", lambda _session: secret_reads.append(True) or "s" * 48, ) broker = _ReadinessBroker(_runtime_payload()) with Session(engine) as session: _deployed_job(session) row = session.get(DeploymentJob, _JOB_ID) row.status = "applying" session.add(row) session.commit() service = ForgeBootstrapApplyService(broker=broker) invalid_format = service.current_readiness(session, "bad-job-id") not_deployed = service.current_readiness(session, _JOB_ID) absent = service.current_readiness(session, "forge-install-" + "0" * 64) assert invalid_format["status"] == not_deployed["status"] == absent["status"] == "unavailable" assert secret_reads == [] assert broker.calls == [] engine.dispose() @pytest.fixture def readiness_operator_client(tmp_path, monkeypatch): from backend.app.core import config from backend.app.db import session as db_session from backend.app.db.init_db import init_db from backend.app.main import app engine = create_engine( f"sqlite:///{tmp_path / 'readiness-api.db'}", connect_args={"check_same_thread": False}, ) monkeypatch.setattr(db_session, "engine", engine) monkeypatch.setenv("APPOPS_REQUIRE_OPERATOR_AUTH", "false") monkeypatch.setattr(config.settings, "operator_proxy_auth_enabled", False) monkeypatch.setattr(config.settings, "auth_admin_subjects", ["authentik:owner"]) monkeypatch.setattr(config.settings, "auth_admin_emails", []) monkeypatch.setattr(config.settings, "data_dir", str(tmp_path)) monkeypatch.setattr(user_access, "current_user", _REAL_CURRENT_USER) SQLModel.metadata.create_all(engine) init_db() client = TestClient(app, client=("198.18.10.20", 42000)) yield client engine.dispose() def _sign_in_operator(client): from backend.app.db import session as db_session from backend.app.services.user_access import issue_session, provision_user with Session(db_session.engine) as session: user = provision_user( session, "authentik", "https://auth.example/application/o/appops/", {"sub": "owner", "email": "owner@example.invalid"}, ) cookie = issue_session(session, user) client.cookies.set("appops_user_session", cookie) def test_readiness_route_requires_operator_and_rejects_query_without_runtime_call( readiness_operator_client, monkeypatch ): from backend.app.services.forge_bootstrap_apply import ForgeBootstrapApplyService client = readiness_operator_client path = f"/api/deployments/forge-bootstrap/first-install/{_JOB_ID}/readiness" assert client.get(path).status_code == 401 _sign_in_operator(client) calls = [] monkeypatch.setattr( ForgeBootstrapApplyService, "current_readiness", lambda self, _session, _job_id: calls.append(True) or {"status": "unavailable"}, ) query = client.get(path + "?profile=caller-selected") body = client.request( "GET", path, content=b"{}", headers={"Content-Type": "application/json"} ) assert query.status_code == 400 assert body.status_code == 400 assert calls == [] def test_readiness_route_returns_fixed_unavailable_for_unknown_job_without_broker_call( readiness_operator_client, monkeypatch ): from backend.app.services import forge_bootstrap_apply from backend.app.services.deployment_broker import DeploymentBrokerClient client = readiness_operator_client _sign_in_operator(client) secret_reads = [] monkeypatch.setattr( forge_bootstrap_apply, "_controller_secret", lambda _session: secret_reads.append(True) or "s" * 48, ) monkeypatch.setattr( DeploymentBrokerClient, "forge_bootstrap_readiness", lambda *_args, **_kwargs: pytest.fail("unknown jobs must not contact the controller"), ) response = client.get( "/api/deployments/forge-bootstrap/first-install/forge-install-" + "0" * 64 + "/readiness" ) invalid_path = client.get( "/api/deployments/forge-bootstrap/first-install/not-a-job/readiness" ) assert response.status_code == 200 assert response.json()["status"] == "unavailable" assert response.json()["message"] == "forge_runtime_readiness_unavailable" assert invalid_path.status_code == 200 assert invalid_path.json()["status"] == "unavailable" assert secret_reads == [] @pytest.fixture def forge_status_module(): path = _ROOT / "ops" / "gitea-deploy-broker" / "status_server.py" spec = importlib.util.spec_from_file_location("forge_runtime_readiness_status_test", path) assert spec and spec.loader module = importlib.util.module_from_spec(spec) sys.modules[spec.name] = module spec.loader.exec_module(module) return module def test_controller_readiness_route_is_signed_replay_safe_and_uses_live_probe_only( forge_status_module, tmp_path, monkeypatch ): module = forge_status_module secret = b"controller-readiness-secret-" + b"x" * 32 secret_root = tmp_path / "deployment-secrets" secret_root.mkdir(mode=0o700) secret_path = secret_root / module.FORGE_BOOTSTRAP_DEPLOYMENT_SECRET_NAME secret_path.write_bytes(secret) secret_path.chmod(0o600) nonce_root = tmp_path / "nonces" monkeypatch.setattr(module, "APPOPS_DEPLOYMENT_SECRET_ROOT", secret_root) monkeypatch.setattr(module, "APPOPS_NONCE_ROOT", nonce_root) request_id = "9" * 64 runtime_result = { "status": "ready", "request": request_id, "repository": "jens/itworx-forge", "operation": "forge_first_install", "profile": { "revision": _REVISION, "workspaceImageDigest": _WORKSPACE_DIGEST, "frontdoorImageDigest": _FRONTDOOR_DIGEST, }, "health": { "status": "healthy", "services": [{"name": name, "status": "healthy"} for name in _SERVICES], }, "migrationCompleted": True, } calls = [] class Controller: def runtime_readiness(self, supplied_request): calls.append(supplied_request) return runtime_result def status(self, _supplied_request): pytest.fail("readiness must not invoke the mutating status accessor") release = { **_ARTIFACT, "apiVersion": "itworx.controller-release/v1", "builtAt": "2026-10-04T00:00:00Z", } monkeypatch.setattr(module, "_forge_first_install_controller", Controller()) monkeypatch.setattr(module, "_forge_controller_artifact_identity", lambda: release) server = module.BoundedThreadingHTTPServer(("127.0.0.1", 0), module.Handler) serving = threading.Thread(target=server.serve_forever, daemon=True) serving.start() def get_readiness(nonce: str): path = f"{module.FORGE_BOOTSTRAP_READINESS_PREFIX}{request_id}" timestamp = str(int(module.time.time())) signature = module._request_signature( secret, method="GET", path=path, timestamp=timestamp, nonce=nonce, raw_body=b"", ) connection = http.client.HTTPConnection(*server.server_address, timeout=3) connection.request( "GET", path, headers={ "X-ITWorx-Deployment-Timestamp": timestamp, "X-ITWorx-Deployment-Nonce": nonce, "X-ITWorx-Deployment-Signature": signature, }, ) response = connection.getresponse() status = response.status payload = json.loads(response.read()) connection.close() return status, payload try: status, payload = get_readiness("a" * 32) assert status == 200 assert payload["apiVersion"] == "itworx.forge-runtime-readiness/v1" assert payload["nonce"] == "a" * 32 assert payload["controllerArtifact"] == _ARTIFACT assert set(payload) == { "apiVersion", "status", "request", "repository", "operation", "profile", "health", "migrationCompleted", "observedAt", "nonce", "controllerArtifact", } assert secret.decode() not in json.dumps(payload) assert calls == [request_id] replay_status, replay_payload = get_readiness("a" * 32) assert replay_status == 403 assert replay_payload == {"status": "authentication_rejected"} assert calls == [request_id] finally: server.shutdown() server.server_close() serving.join(timeout=2)