Publish curated Unraid-AppOps-Gateway source from ad0a390e431166c3198a2ac3bec010c2b44f2dea

backend/tests/test_managed_storage_action.py

@@ -0,0 +1,273 @@
+from __future__ import annotations
+
+import pytest
+from sqlmodel import Session, SQLModel, create_engine, select
+
+from backend.app.db.models import AppSetting, AuditEvent, JobRun, Policy
+from backend.app.services.action_execution import (
+    approval_execution_supported,
+    execute_governed_action,
+    mcp_execution_supported,
+)
+from backend.app.services.action_registry import (
+    ACTION_RISK,
+    ACTIONS,
+    action_default_target,
+)
+from backend.app.services.policy_engine import evaluate_policy
+
+ACTION = "controller.managed-storage.initialize"
+TARGET = "tower/deployment-controller"
+
+
+def test_managed_storage_initialization_is_typed_high_risk_and_fixed_target():
+    assert ACTION in ACTIONS
+    assert ACTION_RISK[ACTION] == "high"
+    assert action_default_target(ACTION) == TARGET
+    assert not approval_execution_supported(ACTION)
+    assert mcp_execution_supported(ACTION)
+
+
+def test_managed_storage_policy_defaults_to_blocked_and_requires_exact_manual_allow(tmp_path):
+    engine = create_engine(f"sqlite:///{tmp_path / 'managed-storage-policy.db'}")
+    SQLModel.metadata.create_all(engine)
+    with Session(engine) as session:
+        assert evaluate_policy(session, ACTION, TARGET, "manual")["decision"] == "blocked"
+        session.add(Policy(
+            id="managed-storage-wildcard",
+            action=ACTION,
+            target="*",
+            voice="blocked",
+            manual="allowed",
+            risk="high",
+        ))
+        session.commit()
+        assert evaluate_policy(session, ACTION, TARGET, "manual")["decision"] == "blocked"
+
+        session.add(Policy(
+            id="managed-storage-exact",
+            action=ACTION,
+            target=TARGET,
+            voice="allowed",
+            manual="allowed",
+            risk="high",
+        ))
+        session.commit()
+        assert evaluate_policy(session, ACTION, TARGET, "manual")["decision"] == "allowed"
+        assert evaluate_policy(session, ACTION, "tower/other-controller", "manual")["decision"] == "blocked"
+        assert evaluate_policy(session, ACTION, TARGET, "voice")["decision"] == "blocked"
+
+    engine.dispose()
+
+
+def _executor_engine(tmp_path):
+    engine = create_engine(f"sqlite:///{tmp_path / 'managed-storage-executor.db'}")
+    SQLModel.metadata.create_all(engine)
+    return engine
+
+
+def _managed_trust():
+    from backend.app.schemas.enrollment import EnrollmentTrustConfigV1
+
+    return EnrollmentTrustConfigV1.model_validate({
+        "controller_identity": "unraid-controller",
+        "bootstrap_secret_ref": "ENROLLMENT_BOOTSTRAP",
+        "trusted_appdata_root": "/var/lib/docker/volumes/itworx-managed-appdata/_data",
+        "managed_appdata_backup_acknowledged": True,
+        "approved_networks": ["itworx-mcp-hub-npm"],
+        "allowed_domain_suffixes": ["example.com"],
+        "autonomous_profiles": ["repopulse-authenticated"],
+        "repopulse_proxy_access_list_id": 42,
+        "oidc_issuer": None,
+        "oidc_role_claim": "roles",
+        "administrator_group": None,
+        "support_agent_group": None,
+        "read_only_group": None,
+    })
+
+
+def _execute_initializer(session):
+    return execute_governed_action(
+        session,
+        action=ACTION,
+        target=TARGET,
+        source="chatgpt-mcp",
+        risk="high",
+        reason="Operator explicitly selected managed storage and acknowledged its backup scope.",
+        original_policy_decision="allowed",
+    )
+
+
+@pytest.mark.parametrize("trust_state", ["missing", "legacy"])
+def test_managed_storage_executor_refuses_missing_selection_or_ack_without_broker_call(
+    tmp_path, monkeypatch, trust_state
+):
+    from backend.app.services.deployment_broker import DeploymentBrokerClient
+
+    engine = _executor_engine(tmp_path)
+    if trust_state == "legacy":
+        legacy = _managed_trust().model_copy(update={
+            "trusted_appdata_root": "/srv/example/appdata",
+            "managed_appdata_backup_acknowledged": False,
+        })
+        with Session(engine) as setup:
+            setup.add(AppSetting(
+                key="deployment_enrollment:trust",
+                value_json=legacy.model_dump(mode="json"),
+                is_secret=False,
+            ))
+            setup.commit()
+
+    monkeypatch.setattr(
+        DeploymentBrokerClient,
+        "initialize_managed_appdata",
+        lambda *_args, **_kwargs: pytest.fail("broker must not be called without managed trust opt-in"),
+    )
+    with Session(engine) as session:
+        result = _execute_initializer(session)
+        assert result["success"] is False
+        assert result["result"] == "blocked"
+        assert result["auditId"]
+        assert session.exec(select(JobRun)).all() == []
+        event = session.exec(select(AuditEvent)).one()
+        assert event.result == "blocked"
+        assert event.redacted_metadata == {"executor": ACTION, "trustOptIn": False}
+    engine.dispose()
+
+
+def test_managed_storage_executor_requires_bootstrap_secret_without_broker_call(tmp_path, monkeypatch):
+    from backend.app.services.deployment_broker import DeploymentBrokerClient
+    from backend.app.services.enrollment_plane import EnrollmentPlane
+
+    engine = _executor_engine(tmp_path)
+    monkeypatch.setattr(EnrollmentPlane, "get_trust_config", staticmethod(lambda _session: _managed_trust()))
+    monkeypatch.setattr("backend.app.services.secret_store.get_secret", lambda *_args, **_kwargs: None)
+    monkeypatch.setattr(
+        DeploymentBrokerClient,
+        "initialize_managed_appdata",
+        lambda *_args, **_kwargs: pytest.fail("broker must not be called without bootstrap secret"),
+    )
+    with Session(engine) as session:
+        result = _execute_initializer(session)
+        assert result["success"] is False
+        assert result["result"] == "failed"
+        assert result["auditId"]
+        assert session.exec(select(JobRun)).all() == []
+        event = session.exec(select(AuditEvent)).one()
+        assert event.result == "failed"
+        assert event.redacted_metadata == {"executor": ACTION, "secretMaterialReturned": "[REDACTED]"}
+    engine.dispose()
+
+
+@pytest.mark.parametrize("status", ["initialized", "already_initialized"])
+def test_managed_storage_executor_records_durable_success_without_secret_material(
+    tmp_path, monkeypatch, status
+):
+    from backend.app.services.deployment_broker import DeploymentBrokerClient
+    from backend.app.services.enrollment_plane import EnrollmentPlane
+
+    engine = _executor_engine(tmp_path)
+    monkeypatch.setattr(EnrollmentPlane, "get_trust_config", staticmethod(lambda _session: _managed_trust()))
+    monkeypatch.setattr("backend.app.services.secret_store.get_secret", lambda *_args, **_kwargs: "temporary-bootstrap-secret")
+    broker_calls = []
+
+    def initialize(_client, *, secret):
+        broker_calls.append(secret)
+        return {
+            "status": status,
+            "trustedAppdataRoot": "/var/lib/docker/volumes/itworx-managed-appdata/_data",
+            "backupLocation": "dockerRootDir",
+        }
+
+    monkeypatch.setattr(DeploymentBrokerClient, "initialize_managed_appdata", initialize)
+    with Session(engine) as session:
+        result = _execute_initializer(session)
+        assert result["success"] is True
+        assert result["status"] == "success"
+        assert result["jobId"] and result["auditId"]
+        assert broker_calls == ["temporary-bootstrap-secret"]
+
+    with Session(engine) as session:
+        job = session.get(JobRun, result["jobId"])
+        event = session.get(AuditEvent, result["auditId"])
+        assert job is not None and job.status == "succeeded"
+        assert event is not None and event.result == "success"
+        assert event.after_state == {
+            "status": status,
+            "trustedAppdataRoot": "/var/lib/docker/volumes/itworx-managed-appdata/_data",
+            "backupLocation": "dockerRootDir",
+        }
+        assert event.redacted_metadata["secretMaterialReturned"] == "[REDACTED]"
+        assert "temporary-bootstrap-secret" not in repr(event.model_dump())
+    assert "temporary-bootstrap-secret" not in repr(result)
+    engine.dispose()
+
+
+@pytest.mark.parametrize(
+    "broker_result",
+    [
+        {"status": "initialized", "trustedAppdataRoot": "/tmp/wrong-root", "backupLocation": "dockerRootDir"},
+        {"status": "initialized", "trustedAppdataRoot": "/var/lib/docker/volumes/itworx-managed-appdata/_data", "backupLocation": "external-share"},
+        {
+            "status": "secret-sentinel-/var/lib/private",
+            "trustedAppdataRoot": "/var/lib/docker/volumes/itworx-managed-appdata/_data",
+            "backupLocation": "dockerRootDir",
+        },
+    ],
+    ids=["wrong-root", "wrong-backup-location", "untrusted-status"],
+)
+def test_managed_storage_executor_fails_closed_on_mismatched_broker_attestation(
+    tmp_path, monkeypatch, broker_result
+):
+    from backend.app.services.deployment_broker import DeploymentBrokerClient
+    from backend.app.services.enrollment_plane import EnrollmentPlane
+
+    engine = _executor_engine(tmp_path)
+    monkeypatch.setattr(EnrollmentPlane, "get_trust_config", staticmethod(lambda _session: _managed_trust()))
+    monkeypatch.setattr("backend.app.services.secret_store.get_secret", lambda *_args, **_kwargs: "temporary-bootstrap-secret")
+    monkeypatch.setattr(DeploymentBrokerClient, "initialize_managed_appdata", lambda *_args, **_kwargs: broker_result)
+    with Session(engine) as session:
+        result = _execute_initializer(session)
+        assert result["success"] is False
+        assert result["result"] == "failed"
+        assert "temporary-bootstrap-secret" not in repr(result)
+        assert "/tmp/wrong-root" not in repr(result)
+        assert "secret-sentinel-/var/lib/private" not in repr(result)
+        event = session.get(AuditEvent, result["auditId"])
+        assert event is not None and event.result == "failed"
+        assert event.after_state["trustedAppdataRoot"] is None
+        assert event.after_state["backupLocation"] is None
+        if broker_result["status"].startswith("secret-sentinel"):
+            assert event.after_state["status"] == "refused"
+        assert "/tmp/wrong-root" not in repr(event.model_dump())
+        assert "temporary-bootstrap-secret" not in repr(event.model_dump())
+        assert "secret-sentinel-/var/lib/private" not in repr(event.model_dump())
+        job = session.get(JobRun, result["jobId"])
+        assert job is not None and job.status == "failed"
+        assert "secret-sentinel-/var/lib/private" not in repr(job.model_dump())
+    engine.dispose()
+
+
+def test_managed_storage_executor_sanitizes_broker_exception(tmp_path, monkeypatch):
+    from backend.app.services.deployment_broker import DeploymentBrokerClient
+    from backend.app.services.enrollment_plane import EnrollmentPlane
+
+    engine = _executor_engine(tmp_path)
+    monkeypatch.setattr(EnrollmentPlane, "get_trust_config", staticmethod(lambda _session: _managed_trust()))
+    monkeypatch.setattr("backend.app.services.secret_store.get_secret", lambda *_args, **_kwargs: "temporary-bootstrap-secret")
+
+    def raise_broker_error(*_args, **_kwargs):
+        raise RuntimeError("temporary-bootstrap-secret failed under /var/lib/private")
+
+    monkeypatch.setattr(DeploymentBrokerClient, "initialize_managed_appdata", raise_broker_error)
+    with Session(engine) as session:
+        result = _execute_initializer(session)
+        assert result["success"] is False
+        assert result["message"] == "Managed storage initialization was refused or unavailable."
+        assert "temporary-bootstrap-secret" not in repr(result)
+        assert "/var/lib/private" not in repr(result)
+        event = session.get(AuditEvent, result["auditId"])
+        assert event is not None and event.result == "failed"
+        assert "temporary-bootstrap-secret" not in repr(event.model_dump())
+        assert "/var/lib/private" not in repr(event.model_dump())
+    engine.dispose()