from __future__ import annotations import base64 import json import os import re import pytest from fastapi.testclient import TestClient from sqlmodel import Session, SQLModel, create_engine, select from backend.app.services import user_access _REAL_CURRENT_USER = user_access.current_user _DELIVERY_PATH = "/api/deployments/forge-bootstrap/secrets" def test_forge_secret_preparation_has_a_default_denied_exact_operator_policy( tmp_path, monkeypatch ): from backend.app.db import session as db_session from backend.app.db.init_db import init_db from backend.app.db.models import Policy from backend.app.services.action_registry import ( ACTION_RISK, ACTIONS, action_default_target, ) from backend.app.services.policy_engine import evaluate_policy engine = create_engine(f"sqlite:///{tmp_path / 'forge-secret-policy.db'}") monkeypatch.setattr(db_session, "engine", engine) try: init_db() with Session(engine) as session: rows = session.exec( select(Policy).where(Policy.action == "forge.secrets.prepare") ).all() assert "forge.secrets.prepare" in ACTIONS assert ACTION_RISK["forge.secrets.prepare"] == "critical" assert action_default_target("forge.secrets.prepare") == "itworx-forge" assert len(rows) == 1 policy = rows[0] assert (policy.id, policy.target, policy.manual, policy.voice, policy.risk) == ( "policy-forge-secret-prepare", "itworx-forge", "blocked", "blocked", "critical", ) denied = evaluate_policy( session, action="forge.secrets.prepare", target="itworx-forge", source="manual", ) assert denied["decision"] == "blocked" assert denied["scope_decision"] == "blocked" policy.manual = "allowed" policy.voice = "blocked" session.add(policy) session.commit() init_db() with Session(engine) as session: policy = session.exec( select(Policy).where(Policy.action == "forge.secrets.prepare") ).one() assert policy.manual == "allowed" allowed = evaluate_policy( session, action="forge.secrets.prepare", target="itworx-forge", source="manual", ) assert allowed["decision"] == "allowed" assert allowed["scope_decision"] == "allowed" assert allowed["policy_id"] == "policy-forge-secret-prepare" finally: engine.dispose() def _profile(): from backend.app.schemas.forge_bootstrap import ForgeBootstrapPreflightV1 return ForgeBootstrapPreflightV1( api_version="itworx.forge-bootstrap-preflight/v1", revision="a" * 40, workspace_image_digest="sha256:" + "b" * 64, frontdoor_image_digest="sha256:" + "c" * 64, ) @pytest.fixture def forge_secret_operator_client(tmp_path, monkeypatch: pytest.MonkeyPatch): from backend.app.core.config import settings from backend.app.db import session as db_session from backend.app.db.init_db import init_db from backend.app.main import app if os.name == "nt": from backend.app.services import forge_oidc_provisioning monkeypatch.setattr( forge_oidc_provisioning, "_windows_private_path", lambda *_args, **_kwargs: None, ) else: from backend.app.services import forge_oidc_provisioning monkeypatch.setattr( forge_oidc_provisioning, "_FORGE_OIDC_LOCK_ROOT", tmp_path / "forge-secret-api-lock", ) engine = create_engine( f"sqlite:///{tmp_path / 'forge-secret-operator.db'}", connect_args={"check_same_thread": False}, ) monkeypatch.setattr(db_session, "engine", engine) monkeypatch.setenv("APPOPS_REQUIRE_OPERATOR_AUTH", "false") monkeypatch.setattr(settings, "operator_proxy_auth_enabled", False) monkeypatch.setattr(settings, "auth_admin_subjects", ["authentik:owner"], raising=False) monkeypatch.setattr(settings, "auth_admin_emails", [], raising=False) monkeypatch.setattr(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: TestClient, subject: str) -> None: 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": subject, "email": f"{subject}@example.com"}, ) cookie = issue_session(session, user) client.cookies.set("appops_user_session", cookie) def _spy_actual_setup_loader(monkeypatch: pytest.MonkeyPatch): from backend.app.services import setup_service original = setup_service.apply_stored_connector_settings calls = [] def tracked(session, connector_ids=None): calls.append(connector_ids) return original(session, connector_ids) monkeypatch.setattr(setup_service, "apply_stored_connector_settings", tracked) return calls def test_secret_delivery_broker_sends_one_bounded_hmac_signed_fixed_request( monkeypatch, ): from backend.app.services import deployment_broker from backend.app.services.deployment_broker import DeploymentBrokerClient config_bytes = b'{"schema":"itworx.forge-first-install-config/v4"}' oidc_secret = b"o" * 40 reader_token = b"r" * 48 broker_secret = "b" * 48 response_body = b'{"status":"bundle_ready"}' observed = {} class Response: status_code = 200 headers = {"content-length": str(len(response_body))} def __enter__(self): return self def __exit__(self, *_args): return False def iter_bytes(self, *, chunk_size): assert chunk_size == 16 * 1024 yield response_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) return Response() monkeypatch.setattr(deployment_broker.httpx, "Client", Client) client = DeploymentBrokerClient( base_url="https://controller.example", verify_tls=True, ) result = client.forge_secret_delivery( _profile(), config_bytes=config_bytes, oidc_client_secret=oidc_secret, authentik_read_token=reader_token, secret=broker_secret, ) path = "/v1/appops/forge/bootstrap/secrets" expected_payload = { "apiVersion": "itworx.forge-secret-delivery/v1", "target": "itworx-forge", "profile": _profile().model_dump(mode="json"), "configBase64": base64.b64encode(config_bytes).decode("ascii"), "oidcClientSecretBase64": base64.b64encode(oidc_secret).decode("ascii"), "authentikReadTokenBase64": base64.b64encode(reader_token).decode("ascii"), } expected_body = json.dumps(expected_payload, sort_keys=True, separators=(",", ":")).encode() headers = observed["headers"] assert observed["client"] == (client.timeout_seconds, True, False) assert observed["method"] == "POST" assert observed["url"] == f"https://controller.example{path}" assert observed["content"] == expected_body assert len(expected_body) <= 16_384 assert headers["Content-Type"] == "application/json" assert headers["Accept-Encoding"] == "identity" assert re.fullmatch(r"[0-9]{10}", headers["X-ITWorx-Deployment-Timestamp"]) assert re.fullmatch(r"[0-9a-f]{32}", headers["X-ITWorx-Deployment-Nonce"]) assert headers["X-ITWorx-Deployment-Signature"] == DeploymentBrokerClient._signature( broker_secret, "POST", path, headers["X-ITWorx-Deployment-Timestamp"], headers["X-ITWorx-Deployment-Nonce"], expected_body, ) assert result == {"status": "response", "response": {"status": "bundle_ready"}} for secret in (broker_secret.encode(), oidc_secret, reader_token): assert secret not in json.dumps(result).encode() @pytest.mark.parametrize( "failure", ["http_error_with_secret", "duplicate_json", "oversized", "pathological_content_length"], ) def test_secret_delivery_broker_never_echoes_untrusted_controller_responses( monkeypatch, failure ): from backend.app.services import deployment_broker from backend.app.services.deployment_broker import DeploymentBrokerClient secret = b"response-secret-must-never-escape" if failure == "http_error_with_secret": body = b'{"message":"response-secret-must-never-escape"}' status = 502 elif failure == "duplicate_json": body = b'{"status":"bundle_ready","status":"bad"}' status = 200 elif failure == "pathological_content_length": body = b'{"status":"bundle_ready"}' status = 200 else: body = b"x" * (64 * 1024 + 1) status = 200 class Response: status_code = status headers = ( {"content-length": "9" * 5000} if failure == "pathological_content_length" else {} ) def __enter__(self): return self def __exit__(self, *_args): return False def iter_bytes(self, *, chunk_size): for offset in range(0, len(body), chunk_size): yield body[offset : offset + chunk_size] 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_secret_delivery( _profile(), config_bytes=b"{}", oidc_client_secret=b"o" * 40, authentik_read_token=b"r" * 48, secret="b" * 48, ) assert result["status"] == "outcome_unknown" assert secret.decode() not in json.dumps(result) def _delivery_setup(tmp_path, monkeypatch, *, manual="allowed"): from sqlmodel import SQLModel from backend.app.core.config import settings from backend.app.db.models import AppSetting, Policy from backend.app.schemas.enrollment import ( MANAGED_DOCKER_APPDATA_ROOT, EnrollmentTrustConfigV1, ) from backend.app.services.enrollment_plane import TRUST_SETTING_KEY from backend.app.services.secret_store import set_secret engine = create_engine(f"sqlite:///{tmp_path / 'forge-secret-delivery.db'}") SQLModel.metadata.create_all(engine) session = Session(engine) monkeypatch.setattr(settings, "environment", "production") monkeypatch.setattr(settings, "secret_key", "test-key-" + "x" * 35) monkeypatch.setattr(settings, "secret_key_source", "env") for key, value in { "AUTHENTIK_OIDC_AUTHORIZATION_FLOW": "forge-authz-flow", "AUTHENTIK_OIDC_INVALIDATION_FLOW": "forge-logout-flow", "AUTHENTIK_FORGE_OIDC_SIGNING_KEY_ID": "11234567-89ab-4cde-8fab-0123456789ab", "FORGE_PRIVATE_PACKAGE_OWNER": "itworx-private", "AUTHENTIK_URL": "https://auth.example-host.com", "AUTHENTIK_TOKEN": "a" * 48, "AUTHENTIK_VERIFY_TLS": True, "AUTHENTIK_TIMEOUT_SECONDS": "10", }.items(): monkeypatch.setitem(settings.apis, key, value) trust = EnrollmentTrustConfigV1( controller_identity="unraid-controller", bootstrap_secret_ref="ENROLLMENT_BOOTSTRAP", trusted_appdata_root=MANAGED_DOCKER_APPDATA_ROOT, managed_appdata_backup_acknowledged=True, approved_networks=("itworx-mcp-hub-npm",), ) session.add( AppSetting( key=TRUST_SETTING_KEY, is_secret=False, value_json=trust.model_dump(mode="json"), ) ) session.add( Policy( id="policy-forge-secret-prepare", action="forge.secrets.prepare", target="itworx-forge", manual=manual, voice="blocked", risk="critical", reason="unit test policy", ) ) session.commit() for name, value in { "ENROLLMENT_BOOTSTRAP": "e" * 48, "FORGE_DEPLOYMENT_BROKER": "b" * 48, "AUTHENTIK_TOKEN": "a" * 48, "ITWORX_FORGE_OIDC_CLIENT_SECRET": "o" * 40, "AUTHENTIK_FORGE_OIDC_BINDING": '{"version":1}', "AUTHENTIK_FORGE_READER_BINDING": '{"version":1}', "ITWORX_FORGE_AUTHENTIK_READ_TOKEN": "r" * 48, }.items(): set_secret(session, name, value) return engine, session def _successful_controller_response(profile): return { "apiVersion": "itworx.forge-secret-delivery/v1", "target": "itworx-forge", "status": "bundle_ready", "profile": profile.model_dump(mode="json"), "controllerArtifact": { "controllerVersion": "2.4.1", "sourceRepository": "jens/unraid-appops-gateway", "sourceRevision": "d" * 40, "sourceTree": "e" * 40, "artifactSha256": "f" * 64, }, "secretFilesReady": 7, "masterKeyBackupVerified": True, "applyPreflight": "not_evaluated", } def _wire_delivery_for_unit_test(monkeypatch, *, broker_response=None): from contextlib import contextmanager from types import SimpleNamespace from backend.app.services import deployment_broker, forge_bootstrap_apply from backend.app.services import forge_controller_reader as reader from backend.app.services import forge_secret_delivery as delivery @contextmanager def unlocked(): yield verified = reader._VerifiedForgeDeliveryInputs( context={ "applicationId": "01234567-89ab-4cde-8fab-0123456789ab", "providerId": "7", "scopeMappingIds": { "openid": "01234567-89ab-4cde-8fab-0123456789ab", "email": "21234567-89ab-4cde-8fab-0123456789ab", "profile": "31234567-89ab-4cde-8fab-0123456789ab", }, "groupIds": { "forge-viewers": "41234567-89ab-4cde-8fab-0123456789ab", "forge-operators": "51234567-89ab-4cde-8fab-0123456789ab", "forge-owners": "61234567-89ab-4cde-8fab-0123456789ab", }, "signingKeyId": "11234567-89ab-4cde-8fab-0123456789ab", }, oidc_client_secret=b"o" * 40, authentik_read_token=b"r" * 48, ) monkeypatch.setattr(delivery, "_forge_oidc_process_lock", unlocked) monkeypatch.setattr(delivery, "secret_storage_status", lambda: {"productionReady": True}) monkeypatch.setattr(forge_bootstrap_apply, "_controller_secret", lambda _session: "b" * 48) monkeypatch.setattr(delivery, "_fixed_authentik_connector", lambda _session: SimpleNamespace()) monkeypatch.setattr(reader, "_verify_forge_delivery_inputs", lambda *_args: verified) monkeypatch.setattr(delivery, "AuthentikConnector", lambda: SimpleNamespace()) calls = [] def deliver(_self, profile, **kwargs): calls.append((profile, kwargs)) return {"status": "response", "response": broker_response or _successful_controller_response(profile)} monkeypatch.setattr(deployment_broker.DeploymentBrokerClient, "forge_secret_delivery", deliver) return delivery, verified, calls def test_delivery_policy_must_be_one_exact_matching_manual_policy(tmp_path, monkeypatch): from backend.app.db.models import Policy from backend.app.services import forge_secret_delivery as delivery engine, session = _delivery_setup(tmp_path, monkeypatch) try: assert delivery._exact_secret_prepare_policy_allowed(session) == "policy-forge-secret-prepare" session.add( Policy( id="policy-forge-secret-wildcard", action="forge.secrets.prepare", target="*", manual="allowed", voice="blocked", risk="critical", ) ) session.commit() assert delivery._exact_secret_prepare_policy_allowed(session) is None finally: session.close() engine.dispose() def test_delivery_policy_requires_allowed_scope_and_same_exact_policy_id( tmp_path, monkeypatch ): from backend.app.services import forge_secret_delivery as delivery engine, session = _delivery_setup(tmp_path, monkeypatch) try: monkeypatch.setattr( delivery, "evaluate_policy", lambda *_args, **_kwargs: { "decision": "allowed", "scope_decision": None, "policy_id": "policy-forge-secret-prepare", }, ) assert delivery._exact_secret_prepare_policy_allowed(session) is None monkeypatch.setattr( delivery, "evaluate_policy", lambda *_args, **_kwargs: { "decision": "allowed", "scope_decision": "allowed", "policy_id": "some-other-policy", }, ) assert delivery._exact_secret_prepare_policy_allowed(session) is None finally: session.close() engine.dispose() def test_delivery_policy_duplicate_exact_row_fails_closed(tmp_path, monkeypatch): from backend.app.db.models import Policy from backend.app.services import forge_secret_delivery as delivery engine, session = _delivery_setup(tmp_path, monkeypatch) try: session.add( Policy( id="policy-forge-secret-prepare-duplicate", action="forge.secrets.prepare", target="itworx-forge", manual="allowed", voice="blocked", risk="critical", ) ) session.commit() assert delivery._exact_secret_prepare_policy_allowed(session) is None finally: session.close() engine.dispose() def test_delivery_policy_rejection_precedes_secret_and_connector_access(tmp_path, monkeypatch): from backend.app.services import forge_secret_delivery as delivery engine, session = _delivery_setup(tmp_path, monkeypatch, manual="blocked") secret_reads = [] connector_calls = [] monkeypatch.setattr(delivery, "has_secret", lambda *_args: secret_reads.append("has") or True) monkeypatch.setattr(delivery, "AuthentikConnector", lambda: connector_calls.append("connector")) try: result = delivery.prepare_forge_secret_delivery( session, profile=_profile(), idempotency_key="prepare-0001", actor_identity="operator@example.test", ) assert result["status"] == "blocked" assert result["message"] == "forge_secret_exact_target_policy_required" assert secret_reads == [] assert connector_calls == [] finally: session.close() engine.dispose() def test_delivery_uses_only_protected_config_and_returns_safe_verified_projection( tmp_path, monkeypatch ): from backend.app.db.models import AuditEvent from backend.app.services.secret_store import get_secret engine, session = _delivery_setup(tmp_path, monkeypatch) delivery, verified, calls = _wire_delivery_for_unit_test(monkeypatch) try: result = delivery.prepare_forge_secret_delivery( session, profile=_profile(), idempotency_key="prepare-0001", actor_identity="operator@example.test", ) assert result["success"] is True assert result["status"] == "bundle_ready" assert result["applyPreflight"] == "not_evaluated" assert result["secretFilesReady"] == 7 assert result["masterKeyBackupVerified"] is True assert len(calls) == 1 profile, delivered = calls[0] assert profile.model_dump(mode="json") == _profile().model_dump(mode="json") assert delivered["oidc_client_secret"] == b"o" * 40 assert delivered["authentik_read_token"] == b"r" * 48 assert delivered["secret"] == "b" * 48 assert delivered["secret"] != "e" * 48 config_bytes = delivered["config_bytes"] config = json.loads(config_bytes) assert list(config) == sorted( [ "schema", "issuer", "audience", "client_id", "authentik_api_url", "launch_url", "authorization_flow", "invalidation_flow", "authentik_openid_mapping_id", "authentik_email_mapping_id", "authentik_profile_mapping_id", "authentik_provider_id", "authentik_signing_key_id", "private_package_owner", ] ) assert config == { "schema": "itworx.forge-first-install-config/v4", "issuer": "https://auth.example-host.com/application/o/itworx-forge/", "audience": "itworx-forge", "client_id": "itworx-forge", "authentik_api_url": "https://auth.example-host.com", "launch_url": "https://forge.example.com", "authorization_flow": "forge-authz-flow", "invalidation_flow": "forge-logout-flow", "authentik_openid_mapping_id": verified.context["scopeMappingIds"]["openid"], "authentik_email_mapping_id": verified.context["scopeMappingIds"]["email"], "authentik_profile_mapping_id": verified.context["scopeMappingIds"]["profile"], "authentik_provider_id": "7", "authentik_signing_key_id": verified.context["signingKeyId"], "private_package_owner": "itworx-private", } assert config_bytes == json.dumps(config, sort_keys=True, separators=(",", ":")).encode() assert b"o" * 40 not in config_bytes assert b"r" * 48 not in config_bytes response_text = json.dumps(result) for held in ("b" * 48, "o" * 40, "r" * 48): assert held not in response_text audit_rows = session.exec(select(AuditEvent)).all() assert len(audit_rows) == 1 assert audit_rows[0].action_id == "forge.secrets.prepare" audit_text = json.dumps(audit_rows[0].redacted_metadata or {}) for held in ("b" * 48, "o" * 40, "r" * 48): assert held not in audit_text assert get_secret(session, "ITWORX_FORGE_AUTHENTIK_READ_TOKEN", touch=False) == "r" * 48 finally: session.close() engine.dispose() def test_invalid_controller_response_and_post_delivery_audit_failure_are_unknown( tmp_path, monkeypatch ): from backend.app.services import deployment_broker engine, session = _delivery_setup(tmp_path, monkeypatch) raw_secret = "r" * 48 malformed = _successful_controller_response(_profile()) malformed["profile"]["revision"] = "c" * 40 delivery, _verified, calls = _wire_delivery_for_unit_test( monkeypatch, broker_response=malformed ) try: result = delivery.prepare_forge_secret_delivery( session, profile=_profile(), idempotency_key="prepare-0001", actor_identity="operator@example.test", ) assert result["status"] == "outcome_unknown" assert raw_secret not in json.dumps(result) assert len(calls) == 1 monkeypatch.setattr( delivery, "log_event", lambda *_args, **_kwargs: (_ for _ in ()).throw(RuntimeError(raw_secret)), ) monkeypatch.setattr( deployment_broker.DeploymentBrokerClient, "forge_secret_delivery", lambda *_args, **_kwargs: {"status": "response", "response": _successful_controller_response(_profile())}, ) audit_failure = delivery.prepare_forge_secret_delivery( session, profile=_profile(), idempotency_key="prepare-0002", actor_identity="operator@example.test", ) assert audit_failure["status"] == "outcome_unknown" assert raw_secret not in json.dumps(audit_failure) finally: session.close() engine.dispose() def test_missing_dedicated_broker_secret_blocks_before_connector_and_delivery( tmp_path, monkeypatch ): from backend.app.services import forge_bootstrap_apply engine, session = _delivery_setup(tmp_path, monkeypatch) delivery, _verified, calls = _wire_delivery_for_unit_test(monkeypatch) monkeypatch.setattr(forge_bootstrap_apply, "_controller_secret", lambda _session: None) monkeypatch.setattr(delivery, "AuthentikConnector", lambda: pytest.fail("connector must not load")) try: result = delivery.prepare_forge_secret_delivery( session, profile=_profile(), idempotency_key="prepare-0001", actor_identity="operator@example.test", ) assert result["status"] == "blocked" assert result["message"] == "forge_secret_controller_trust_required" assert calls == [] finally: session.close() engine.dispose() @pytest.mark.parametrize( "failure", ["production_flag_integer", "managed_root_drift", "backup_ack_integer", "missing_reader_token"], ) def test_delivery_prerequisite_drift_blocks_before_verification_or_broker( tmp_path, monkeypatch, failure ): from sqlalchemy import update from backend.app.db.models import AppSetting from backend.app.services.enrollment_plane import TRUST_SETTING_KEY from backend.app.services.secret_store import delete_secret engine, session = _delivery_setup(tmp_path, monkeypatch) delivery, _verified, calls = _wire_delivery_for_unit_test(monkeypatch) if failure == "production_flag_integer": monkeypatch.setattr(delivery, "secret_storage_status", lambda: {"productionReady": 1}) elif failure in {"managed_root_drift", "backup_ack_integer"}: trust = session.get(AppSetting, TRUST_SETTING_KEY) if failure == "managed_root_drift": trust.value_json = { **trust.value_json, "trusted_appdata_root": "/mnt/cache/appdata", } session.add(trust) session.commit() else: # AppSetting's typed assignment coerces nested integer 1 to bool. # Write malformed persisted JSON directly to exercise strict decode. session.execute( update(AppSetting) .where(AppSetting.key == TRUST_SETTING_KEY) .values( value_json={ **trust.value_json, "managed_appdata_backup_acknowledged": 1, } ) ) session.commit() session.expire_all() else: delete_secret(session, "ITWORX_FORGE_AUTHENTIK_READ_TOKEN") try: result = delivery.prepare_forge_secret_delivery( session, profile=_profile(), idempotency_key="prepare-prereq", actor_identity="operator@example.test", ) assert result["status"] == "blocked" assert calls == [] assert result["message"] in { "forge_secret_store_not_production_ready", "forge_secret_controller_trust_required", "forge_secret_verified_reader_required", } finally: session.close() engine.dispose() def test_fixed_authentik_runtime_requires_matching_encrypted_admin_secret( tmp_path, monkeypatch ): from backend.app.core.config import settings from backend.app.services import forge_secret_delivery as delivery from backend.app.services.secret_store import delete_secret, get_secret engine, session = _delivery_setup(tmp_path, monkeypatch) try: assert delivery._fixed_authentik_connector(session) is not None assert get_secret(session, "AUTHENTIK_TOKEN", touch=False) == settings.apis["AUTHENTIK_TOKEN"] delete_secret(session, "AUTHENTIK_TOKEN") assert settings.apis["AUTHENTIK_TOKEN"] == "a" * 48 assert delivery._fixed_authentik_connector(session) is None finally: session.close() engine.dispose() def test_delivery_refuses_legacy_plaintext_authentik_secret_without_migrating( tmp_path, monkeypatch ): from backend.app.db.models import AppSetting from backend.app.services import forge_secret_delivery as delivery from backend.app.services.secret_store import get_secret engine, session = _delivery_setup(tmp_path, monkeypatch) real_connector_factory = delivery._fixed_authentik_connector delivery, _verified, calls = _wire_delivery_for_unit_test(monkeypatch) monkeypatch.setattr(delivery, "_fixed_authentik_connector", real_connector_factory) legacy = AppSetting( key="connector:authentik:secret", is_secret=True, value_json={"value": "legacy-admin-token-must-not-be-migrated"}, ) session.add(legacy) session.commit() original = dict(legacy.value_json) try: result = delivery.prepare_forge_secret_delivery( session, profile=_profile(), idempotency_key="prepare-legacy", actor_identity="operator@example.test", ) assert result["status"] == "blocked" assert result["message"] == "forge_secret_authentik_configuration_required" assert not calls assert session.get(AppSetting, "connector:authentik:secret").value_json == original assert get_secret(session, "AUTHENTIK_TOKEN", touch=False) == "a" * 48 assert "legacy-admin-token-must-not-be-migrated" not in json.dumps(result) finally: session.close() engine.dispose() def test_lock_exit_failure_after_controller_delivery_is_outcome_unknown(tmp_path, monkeypatch): from contextlib import contextmanager engine, session = _delivery_setup(tmp_path, monkeypatch) delivery, _verified, calls = _wire_delivery_for_unit_test(monkeypatch) @contextmanager def lock_fails_on_exit(): yield raise RuntimeError("post-delivery cleanup failure") monkeypatch.setattr(delivery, "_forge_oidc_process_lock", lock_fails_on_exit) try: result = delivery.prepare_forge_secret_delivery( session, profile=_profile(), idempotency_key="prepare-0001", actor_identity="operator@example.test", ) assert len(calls) == 1 assert result["status"] == "outcome_unknown" assert result["message"] == "forge_secret_controller_outcome_unknown" assert "post-delivery cleanup failure" not in json.dumps(result) finally: session.close() engine.dispose() def test_unrecognized_reader_failure_code_never_reaches_result_or_audit(tmp_path, monkeypatch): from backend.app.services import forge_controller_reader as reader engine, session = _delivery_setup(tmp_path, monkeypatch) delivery, _verified, calls = _wire_delivery_for_unit_test(monkeypatch) marker = "reader-error-contains-held-secret" * 2 def invalid_verifier(*_args): raise reader._ReaderFailure("conflict", marker) monkeypatch.setattr(reader, "_verify_forge_delivery_inputs", invalid_verifier) try: result = delivery.prepare_forge_secret_delivery( session, profile=_profile(), idempotency_key="prepare-0001", actor_identity="operator@example.test", ) assert result["status"] == "conflict" assert result["message"] == "forge_secret_reader_proof_invalid" assert marker not in json.dumps(result) assert calls == [] finally: session.close() engine.dispose() def test_forge_secret_delivery_api_requires_authenticated_operator_and_csrf( forge_secret_operator_client: TestClient, monkeypatch: pytest.MonkeyPatch, ): loader_calls = _spy_actual_setup_loader(monkeypatch) unauthenticated = forge_secret_operator_client.post( _DELIVERY_PATH, json={"profile": _profile().model_dump(mode="json")}, ) assert unauthenticated.status_code == 401 _sign_in_operator(forge_secret_operator_client, "owner") missing_csrf = forge_secret_operator_client.post( _DELIVERY_PATH, json={"profile": _profile().model_dump(mode="json")}, headers={"Idempotency-Key": "prepare-api-01"}, ) assert missing_csrf.status_code == 403 assert loader_calls == [] def test_forge_secret_delivery_api_rejects_read_only_and_extra_profile_fields( forge_secret_operator_client: TestClient, monkeypatch: pytest.MonkeyPatch, ): loader_calls = _spy_actual_setup_loader(monkeypatch) profile = _profile().model_dump(mode="json") _sign_in_operator(forge_secret_operator_client, "reader") denied = forge_secret_operator_client.post( _DELIVERY_PATH, json={"profile": profile}, headers={ "X-AppOps-Session-CSRF": "1", "Idempotency-Key": "prepare-api-02", }, ) assert denied.status_code == 403 _sign_in_operator(forge_secret_operator_client, "owner") profile["caller_config"] = "must-not-be-forwarded" invalid = forge_secret_operator_client.post( _DELIVERY_PATH, json={"profile": profile}, headers={ "X-AppOps-Session-CSRF": "1", "Idempotency-Key": "prepare-api-03", }, ) assert invalid.status_code == 422 assert loader_calls == [] def test_forge_secret_delivery_api_delegates_only_profile_key_and_actor( forge_secret_operator_client: TestClient, monkeypatch: pytest.MonkeyPatch, ): from backend.app.services import forge_secret_delivery _sign_in_operator(forge_secret_operator_client, "owner") calls = [] safe_result = { "success": True, "status": "bundle_ready", "message": "forge_secret_bundle_ready", "target": "itworx-forge", "revision": "a" * 40, "controllerArtifact": { "controllerVersion": "2.4.1", "sourceRepository": "jens/unraid-appops-gateway", "sourceRevision": "d" * 40, "sourceTree": "e" * 40, "artifactSha256": "f" * 64, }, "secretFilesReady": 7, "masterKeyBackupVerified": True, "applyPreflight": "not_evaluated", } def prepare(session, *, profile, idempotency_key, actor_identity): calls.append((session, profile, idempotency_key, actor_identity)) return safe_result monkeypatch.setattr(forge_secret_delivery, "prepare_forge_secret_delivery", prepare) response = forge_secret_operator_client.post( _DELIVERY_PATH, json={"profile": _profile().model_dump(mode="json")}, headers={ "X-AppOps-Session-CSRF": "1", "Idempotency-Key": "prepare-api-04", }, ) assert response.status_code == 200 assert response.json() == safe_result assert len(calls) == 1 assert calls[0][1].model_dump(mode="json") == _profile().model_dump(mode="json") assert calls[0][2:] == ("prepare-api-04", "authentik:owner") def test_forge_secret_delivery_api_default_deny_precedes_loader_secret_and_broker( forge_secret_operator_client: TestClient, monkeypatch: pytest.MonkeyPatch, ): from backend.app.services import forge_secret_delivery as delivery from backend.app.services import setup_service _sign_in_operator(forge_secret_operator_client, "owner") original_loader = setup_service.apply_stored_connector_settings loader_calls = [] def tracked_loader(session, connector_ids=None): loader_calls.append(connector_ids) return original_loader(session, connector_ids) monkeypatch.setattr(setup_service, "apply_stored_connector_settings", tracked_loader) monkeypatch.setattr( delivery, "_fixed_authentik_connector", lambda *_args: pytest.fail("default-deny must precede connector loading"), ) monkeypatch.setattr( delivery, "has_secret", lambda *_args: pytest.fail("default-deny must precede protected secret lookups"), ) response = forge_secret_operator_client.post( _DELIVERY_PATH, json={"profile": _profile().model_dump(mode="json")}, headers={ "X-AppOps-Session-CSRF": "1", "Idempotency-Key": "prepare-default-deny", }, ) assert response.status_code == 200 assert response.json()["status"] == "blocked" assert response.json()["message"] == "forge_secret_exact_target_policy_required" assert loader_calls == []