Unraid-AppOps-Gateway-Public/backend/tests/test_user_auth.py

import pytest
from fastapi.testclient import TestClient
from sqlmodel import Session, SQLModel, create_engine


@pytest.fixture
def identity_client(tmp_path, monkeypatch):
    from backend.app.core.config import settings
    from backend.app.db import session as database
    from backend.app.main import app

    engine = create_engine(f"sqlite:///{tmp_path / 'identities.db'}", connect_args={"check_same_thread": False})
    monkeypatch.setattr(database, "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", ["jens@example.com"], raising=False)
    SQLModel.metadata.create_all(engine)
    # Avoid unrelated schedulers and external connectors in boundary tests.
    from backend.app.db.init_db import init_db
    init_db()
    yield TestClient(app, client=("198.18.10.20", 42000))
    engine.dispose()


def sign_in(client, subject, **claims):
    from backend.app.db import session as database
    from backend.app.services.user_access import issue_session, provision_user

    with Session(database.engine) as db:
        user = provision_user(db, "authentik", "https://auth.example/application/o/appops/", {
            "sub": subject, "email": f"{subject}@example.com", **claims,
        })
        cookie = issue_session(db, user)
        user_id = user.id
    client.cookies.set("appops_user_session", cookie)
    return user_id


def test_shared_token_and_disabled_operator_switch_do_not_grant_access(identity_client):
    client = identity_client
    assert client.get("/api/apps").status_code == 401
    assert client.post("/api/actions/plan", json={}, headers={"X-AppOps-Operator-Token": "old-shared-token"}).status_code == 401
    assert client.get("/api/auth/session").json()["authenticated"] is False
    assert client.get("/api/health").status_code == 200


def test_new_user_is_read_only_and_cannot_change_permissions(identity_client):
    user_id = sign_in(identity_client, "reader")
    assert identity_client.get("/api/apps").status_code == 200
    status = identity_client.get("/api/auth/session").json()
    assert status["user"]["role"] == "read_only"
    assert status["permissions"] == ["read"]
    for path in ["/api/actions/execute", "/api/voice/execute", "/api/settings/connectors/portainer", f"/api/auth/users/{user_id}"]:
        assert identity_client.post(path, json={}, headers={"X-AppOps-Session-CSRF": "1"}).status_code == 403
    assert identity_client.get("/api/auth/users").status_code == 403


def test_admin_can_manage_users_without_unlock_and_changes_apply_immediately(identity_client):
    reader_id = sign_in(identity_client, "reader")
    sign_in(identity_client, "owner")
    status = identity_client.get("/api/auth/session").json()
    assert status["user"]["role"] == "admin"
    assert "users.manage" in status["permissions"]
    assert identity_client.get("/api/auth/users").status_code == 200
    update = identity_client.patch(f"/api/auth/users/{reader_id}", json={"role": "custom", "permissions": ["read", "inventory.manage"], "enabled": True}, headers={"X-AppOps-Session-CSRF": "1"})
    assert update.status_code == 200
    sign_in(identity_client, "reader")
    assert identity_client.get("/api/auth/session").json()["permissions"] == ["inventory.manage", "read"]
    assert identity_client.post("/api/actions/execute", json={}, headers={"X-AppOps-Session-CSRF": "1"}).status_code == 403


def test_csrf_logout_revocation_and_last_admin_protection(identity_client):
    admin_id = sign_in(identity_client, "owner")
    assert identity_client.patch(f"/api/auth/users/{admin_id}", json={"role": "read_only", "permissions": [], "enabled": True}, headers={"X-AppOps-Session-CSRF": "1"}).status_code == 409
    assert identity_client.post("/api/auth/logout").status_code == 403
    assert identity_client.post("/api/auth/logout", headers={"X-AppOps-Session-CSRF": "1"}).status_code == 200
    assert identity_client.get("/api/apps").status_code == 401


def test_last_enabled_admin_guard_counts_another_enabled_admin(identity_client):
    admin_id = sign_in(identity_client, "owner")
    second_id = sign_in(identity_client, "second-admin")
    sign_in(identity_client, "owner")
    headers = {"X-AppOps-Session-CSRF": "1"}

    promoted = identity_client.patch(
        f"/api/auth/users/{second_id}",
        json={"role": "admin", "enabled": True, "permissions": []},
        headers=headers,
    )
    assert promoted.status_code == 200

    disabled = identity_client.patch(
        f"/api/auth/users/{second_id}",
        json={"role": "admin", "enabled": False, "permissions": []},
        headers=headers,
    )
    assert disabled.status_code == 200

    last_admin = identity_client.patch(
        f"/api/auth/users/{admin_id}",
        json={"role": "admin", "enabled": False, "permissions": []},
        headers=headers,
    )
    assert last_admin.status_code == 409


def test_session_expiration_and_forgery_fail_closed(identity_client):
    from backend.app.db import session as database
    from backend.app.db.models import UserSession
    from backend.app.services.user_access import digest
    sign_in(identity_client, "reader")
    with Session(database.engine) as db:
        record = db.get(UserSession, digest(identity_client.cookies.get("appops_user_session")))
        record.expires_at = 1
        db.add(record)
        db.commit()
    assert identity_client.get("/api/apps").status_code == 401
    identity_client.cookies.set("appops_user_session", "forged")
    assert identity_client.get("/api/apps").status_code == 401


def test_email_bootstrap_requires_verified_claim_and_is_pinned(identity_client):
    sign_in(identity_client, "username-only", email=None, preferred_username="jens@example.com", email_verified=True)
    assert identity_client.get("/api/auth/session").json()["user"]["role"] == "read_only"
    sign_in(identity_client, "unverified", email="jens@example.com")
    assert identity_client.get("/api/auth/session").json()["user"]["role"] == "read_only"

    sign_in(identity_client, "verified", email="jens@example.com", email_verified=True)
    assert identity_client.get("/api/auth/session").json()["user"]["role"] == "admin"
    sign_in(identity_client, "replacement", email="jens@example.com", email_verified=True)
    assert identity_client.get("/api/auth/session").json()["user"]["role"] == "read_only"


def test_bootstrap_recovers_early_login_without_restoring_demoted_admin(identity_client, monkeypatch):
    from backend.app.core.config import settings
    user_id = sign_in(identity_client, "early")
    monkeypatch.setattr(settings, "auth_admin_subjects", ["authentik:early", "authentik:owner"])
    sign_in(identity_client, "early")
    assert identity_client.get("/api/auth/session").json()["user"]["role"] == "admin"
    sign_in(identity_client, "owner")
    response = identity_client.patch(f"/api/auth/users/{user_id}", json={"role": "read_only", "enabled": True, "permissions": []}, headers={"X-AppOps-Session-CSRF": "1"})
    assert response.status_code == 200
    sign_in(identity_client, "early")
    assert identity_client.get("/api/auth/session").json()["user"]["role"] == "read_only"

    monkeypatch.setattr(settings, "auth_admin_subjects", ["authentik:owner"])
    sign_in(identity_client, "early", email="jens@example.com", email_verified=True)
    assert identity_client.get("/api/auth/session").json()["user"]["role"] == "read_only"


def test_expired_disabled_and_forged_sessions_fail_closed(identity_client):
    from backend.app.db import session as database
    from backend.app.db.models import GatewayUser
    user_id = sign_in(identity_client, "reader")
    with Session(database.engine) as db:
        user = db.get(GatewayUser, user_id)
        user.enabled = False
        db.add(user)
        db.commit()
    assert identity_client.get("/api/apps").status_code == 401


def test_proxy_headers_require_trusted_peer_and_fixed_uid(identity_client, monkeypatch):
    from fastapi import HTTPException
    from starlette.requests import Request

    from backend.app.core.config import settings
    from backend.app.services.user_access import current_user
    monkeypatch.setattr(settings, "operator_proxy_auth_enabled", True)
    monkeypatch.setattr(settings, "operator_proxy_trusted_networks", ["203.0.113.8/32"])
    headers = [(b"x-authentik-uid", b"owner"), (b"x-authentik-username", b"jens"), (b"x-authentik-email", b"jens@example.com")]
    def request(peer, supplied_headers):
        return Request({"type": "http", "method": "GET", "path": "/api/apps", "headers": supplied_headers, "client": (peer, 1234)})
    with pytest.raises(HTTPException) as rejected:
        current_user(request("198.18.10.20", headers))
    assert rejected.value.status_code == 401
    assert current_user(request("203.0.113.8", headers)).role == "admin"
    with pytest.raises(HTTPException) as rejected:
        current_user(request("203.0.113.8", headers[1:]))
    assert rejected.value.status_code == 401


def test_custom_voice_and_approval_permissions_do_not_grant_execution(identity_client):
    reader_id = sign_in(identity_client, "reader")
    sign_in(identity_client, "owner")
    response = identity_client.patch(f"/api/auth/users/{reader_id}", json={"role": "custom", "enabled": True, "permissions": ["voice.use", "approvals.manage"]}, headers={"X-AppOps-Session-CSRF": "1"})
    assert response.status_code == 200
    sign_in(identity_client, "reader")
    for path in ["/api/voice/actions/container/plex/restart", "/api/approvals/approval-1/execute", "/api/connectors/portainer/bind", "/api/ai/rag/reindex"]:
        assert identity_client.post(path, json={}, headers={"X-AppOps-Session-CSRF": "1"}).status_code == 403


def test_operational_users_cannot_change_enrollment_trust(identity_client):
    reader_id = sign_in(identity_client, "reader")
    sign_in(identity_client, "owner")
    headers = {"X-AppOps-Session-CSRF": "1"}
    assert identity_client.patch(f"/api/auth/users/{reader_id}", json={"role": "operator", "enabled": True}, headers=headers).status_code == 200
    sign_in(identity_client, "reader")
    assert identity_client.put("/api/enrollments/trust", json={}, headers=headers).status_code == 403
    sign_in(identity_client, "owner")
    # Administrator reaches schema validation; an operator is rejected first.
    assert identity_client.put("/api/enrollments/trust", json={}, headers=headers).status_code == 422


def _enrollment_trust_payload(*, appdata_root: str, backup_acknowledged: bool) -> dict:
    return {
        "controller_identity": "unraid-controller",
        "bootstrap_secret_ref": "ENROLLMENT_BOOTSTRAP",
        "trusted_appdata_root": appdata_root,
        "managed_appdata_backup_acknowledged": backup_acknowledged,
        "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,
    }


@pytest.mark.parametrize(
    ("initial_root", "initial_ack", "requested_root", "requested_ack"),
    [
        (
            "/srv/example/appdata",
            False,
            "/var/lib/docker/volumes/itworx-managed-appdata/_data",
            True,
        ),
        (
            "/var/lib/docker/volumes/itworx-managed-appdata/_data",
            True,
            "/srv/example/appdata",
            False,
        ),
    ],
    ids=["initial-managed-selection", "managed-root-reset"],
)
def test_settings_manage_put_cannot_change_storage_root(
    identity_client, initial_root, initial_ack, requested_root, requested_ack
):
    from sqlmodel import Session, select

    from backend.app.db import session as database
    from backend.app.db.models import AppSetting, AuditEvent

    reader_id = sign_in(identity_client, "reader")
    sign_in(identity_client, "owner")
    headers = {"X-AppOps-Session-CSRF": "1"}
    assert identity_client.patch(
        f"/api/auth/users/{reader_id}",
        json={"role": "custom", "enabled": True, "permissions": ["settings.manage"]},
        headers=headers,
    ).status_code == 200

    before = _enrollment_trust_payload(
        appdata_root=initial_root, backup_acknowledged=initial_ack
    )
    with Session(database.engine) as db:
        db.add(AppSetting(key="deployment_enrollment:trust", value_json=before, is_secret=False))
        db.commit()
        audit_ids_before = set(db.exec(select(AuditEvent.id)).all())

    sign_in(identity_client, "reader")
    response = identity_client.put(
        "/api/enrollments/trust",
        json=_enrollment_trust_payload(
            appdata_root=requested_root, backup_acknowledged=requested_ack
        ),
        headers=headers,
    )
    assert response.status_code == 403

    with Session(database.engine) as db:
        saved = db.get(AppSetting, "deployment_enrollment:trust")
        assert saved is not None and saved.value_json == before
        assert set(db.exec(select(AuditEvent.id)).all()) == audit_ids_before


def test_settings_manage_put_cannot_create_initial_managed_trust(identity_client):
    from sqlmodel import Session, select

    from backend.app.db import session as database
    from backend.app.db.models import AppSetting, AuditEvent

    reader_id = sign_in(identity_client, "reader")
    sign_in(identity_client, "owner")
    headers = {"X-AppOps-Session-CSRF": "1"}
    assert identity_client.patch(
        f"/api/auth/users/{reader_id}",
        json={"role": "custom", "enabled": True, "permissions": ["settings.manage"]},
        headers=headers,
    ).status_code == 200
    with Session(database.engine) as db:
        audit_ids_before = set(db.exec(select(AuditEvent.id)).all())

    sign_in(identity_client, "reader")
    response = identity_client.put(
        "/api/enrollments/trust",
        json=_enrollment_trust_payload(
            appdata_root="/var/lib/docker/volumes/itworx-managed-appdata/_data",
            backup_acknowledged=True,
        ),
        headers=headers,
    )
    assert response.status_code == 403
    with Session(database.engine) as db:
        assert db.get(AppSetting, "deployment_enrollment:trust") is None
        assert set(db.exec(select(AuditEvent.id)).all()) == audit_ids_before


def test_settings_manage_put_can_create_initial_legacy_trust(identity_client):
    from sqlmodel import Session

    from backend.app.db import session as database
    from backend.app.db.models import AppSetting

    reader_id = sign_in(identity_client, "reader")
    sign_in(identity_client, "owner")
    headers = {"X-AppOps-Session-CSRF": "1"}
    assert identity_client.patch(
        f"/api/auth/users/{reader_id}",
        json={"role": "custom", "enabled": True, "permissions": ["settings.manage"]},
        headers=headers,
    ).status_code == 200
    sign_in(identity_client, "reader")

    legacy = _enrollment_trust_payload(
        appdata_root="/srv/example/appdata", backup_acknowledged=False
    )
    response = identity_client.put("/api/enrollments/trust", json=legacy, headers=headers)
    assert response.status_code == 200
    with Session(database.engine) as db:
        saved = db.get(AppSetting, "deployment_enrollment:trust")
        assert saved is not None
        assert saved.value_json["trusted_appdata_root"] == "/srv/example/appdata"
        assert saved.value_json["managed_appdata_backup_acknowledged"] is False


def test_settings_manage_put_keeps_other_trust_edits_and_admin_can_select_managed_root(identity_client):
    from sqlmodel import Session

    from backend.app.db import session as database
    from backend.app.db.models import AppSetting

    reader_id = sign_in(identity_client, "reader")
    sign_in(identity_client, "owner")
    headers = {"X-AppOps-Session-CSRF": "1"}
    assert identity_client.patch(
        f"/api/auth/users/{reader_id}",
        json={"role": "custom", "enabled": True, "permissions": ["settings.manage"]},
        headers=headers,
    ).status_code == 200
    legacy = _enrollment_trust_payload(
        appdata_root="/srv/example/appdata", backup_acknowledged=False
    )
    with Session(database.engine) as db:
        db.add(AppSetting(key="deployment_enrollment:trust", value_json=legacy, is_secret=False))
        db.commit()

    sign_in(identity_client, "reader")
    unrelated_update = {
        **legacy,
        "approved_networks": ["itworx-mcp-hub-npm", "itworx-npm"],
    }
    response = identity_client.put(
        "/api/enrollments/trust", json=unrelated_update, headers=headers
    )
    assert response.status_code == 200
    with Session(database.engine) as db:
        saved = db.get(AppSetting, "deployment_enrollment:trust")
        assert saved is not None
        assert saved.value_json["approved_networks"] == ["itworx-mcp-hub-npm", "itworx-npm"]
        assert saved.value_json["trusted_appdata_root"] == "/srv/example/appdata"
        assert saved.value_json["managed_appdata_backup_acknowledged"] is False

    sign_in(identity_client, "owner")
    managed = _enrollment_trust_payload(
        appdata_root="/var/lib/docker/volumes/itworx-managed-appdata/_data",
        backup_acknowledged=True,
    )
    response = identity_client.put("/api/enrollments/trust", json=managed, headers=headers)
    assert response.status_code == 200
    with Session(database.engine) as db:
        saved = db.get(AppSetting, "deployment_enrollment:trust")
        assert saved is not None
        assert saved.value_json["trusted_appdata_root"] == managed["trusted_appdata_root"]
        assert saved.value_json["managed_appdata_backup_acknowledged"] is True


def test_settings_manage_put_cas_preserves_concurrent_admin_storage_selection(identity_client):
    from sqlalchemy import event
    from sqlmodel import Session, select

    from backend.app.core.time import utc_now
    from backend.app.db import session as database
    from backend.app.db.models import AppSetting, AuditEvent
    from backend.app.services.enrollment_plane import TRUST_SETTING_KEY

    reader_id = sign_in(identity_client, "reader")
    sign_in(identity_client, "owner")
    headers = {"X-AppOps-Session-CSRF": "1"}
    assert identity_client.patch(
        f"/api/auth/users/{reader_id}",
        json={"role": "custom", "enabled": True, "permissions": ["settings.manage"]},
        headers=headers,
    ).status_code == 200

    original = _enrollment_trust_payload(
        appdata_root="/srv/example/appdata", backup_acknowledged=False
    )
    with database.engine.connect() as connection:
        connection.exec_driver_sql("PRAGMA journal_mode=WAL")
    with Session(database.engine) as db:
        db.add(AppSetting(key=TRUST_SETTING_KEY, value_json=original, is_secret=False))
        db.commit()
        audit_ids_before = set(db.exec(select(AuditEvent.id)).all())

    sign_in(identity_client, "reader")
    injected = False

    def select_managed_root_before_put_write(
        _connection, _cursor, statement, parameters, _context, _executemany
    ):
        nonlocal injected
        normalized = statement.lstrip().upper()
        if (
            not injected
            and normalized.startswith("UPDATE APP_SETTINGS")
            and TRUST_SETTING_KEY in str(parameters)
        ):
            injected = True
            with Session(database.engine) as concurrent_session:
                row = concurrent_session.get(AppSetting, TRUST_SETTING_KEY)
                assert row is not None
                row.value_json = {
                    **row.value_json,
                    "trusted_appdata_root": "/var/lib/docker/volumes/itworx-managed-appdata/_data",
                    "managed_appdata_backup_acknowledged": True,
                }
                row.updated_at = utc_now()
                concurrent_session.add(row)
                concurrent_session.commit()

    event.listen(database.engine, "before_cursor_execute", select_managed_root_before_put_write)
    try:
        response = identity_client.put(
            "/api/enrollments/trust",
            json={
                **original,
                "approved_networks": ["itworx-mcp-hub-npm", "itworx-npm"],
            },
            headers=headers,
        )
    finally:
        event.remove(database.engine, "before_cursor_execute", select_managed_root_before_put_write)

    assert injected is True
    assert response.status_code == 409
    with Session(database.engine) as db:
        saved = db.get(AppSetting, TRUST_SETTING_KEY)
        assert saved is not None
        assert saved.value_json["trusted_appdata_root"] == "/var/lib/docker/volumes/itworx-managed-appdata/_data"
        assert saved.value_json["managed_appdata_backup_acknowledged"] is True
        assert saved.value_json["approved_networks"] == original["approved_networks"]
        assert set(db.exec(select(AuditEvent.id)).all()) == audit_ids_before


def test_settings_manage_initial_legacy_put_loses_if_admin_creates_trust_concurrently(identity_client):
    from sqlalchemy import event
    from sqlmodel import Session, select

    from backend.app.db import session as database
    from backend.app.db.models import AppSetting, AuditEvent
    from backend.app.services.enrollment_plane import TRUST_SETTING_KEY

    reader_id = sign_in(identity_client, "reader")
    sign_in(identity_client, "owner")
    headers = {"X-AppOps-Session-CSRF": "1"}
    assert identity_client.patch(
        f"/api/auth/users/{reader_id}",
        json={"role": "custom", "enabled": True, "permissions": ["settings.manage"]},
        headers=headers,
    ).status_code == 200
    with database.engine.connect() as connection:
        connection.exec_driver_sql("PRAGMA journal_mode=WAL")
    with Session(database.engine) as db:
        audit_ids_before = set(db.exec(select(AuditEvent.id)).all())

    sign_in(identity_client, "reader")
    admin_value = _enrollment_trust_payload(
        appdata_root="/var/lib/docker/volumes/itworx-managed-appdata/_data",
        backup_acknowledged=True,
    )
    injected = False

    def create_admin_trust_before_put_insert(
        _connection, _cursor, statement, parameters, _context, _executemany
    ):
        nonlocal injected
        normalized = statement.lstrip().upper()
        if (
            not injected
            and normalized.startswith("INSERT INTO APP_SETTINGS")
            and TRUST_SETTING_KEY in str(parameters)
        ):
            injected = True
            with Session(database.engine) as concurrent_session:
                concurrent_session.add(
                    AppSetting(key=TRUST_SETTING_KEY, value_json=admin_value, is_secret=False)
                )
                concurrent_session.commit()

    event.listen(database.engine, "before_cursor_execute", create_admin_trust_before_put_insert)
    try:
        response = identity_client.put(
            "/api/enrollments/trust",
            json=_enrollment_trust_payload(
                appdata_root="/srv/example/appdata", backup_acknowledged=False
            ),
            headers=headers,
        )
    finally:
        event.remove(database.engine, "before_cursor_execute", create_admin_trust_before_put_insert)

    assert injected is True
    assert response.status_code == 409
    with Session(database.engine) as db:
        saved = db.get(AppSetting, TRUST_SETTING_KEY)
        assert saved is not None and saved.value_json == admin_value
        assert set(db.exec(select(AuditEvent.id)).all()) == audit_ids_before


def test_storage_root_patch_is_admin_only_and_preserves_other_trust_fields(identity_client):
    from sqlmodel import Session, select

    from backend.app.db import session as database
    from backend.app.db.models import AppSetting, AuditEvent

    sign_in(identity_client, "owner")
    original = {
        "controller_identity": "unraid-controller",
        "bootstrap_secret_ref": "ENROLLMENT_BOOTSTRAP",
        "trusted_appdata_root": "/srv/example/appdata",
        "managed_appdata_backup_acknowledged": False,
        "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,
    }
    with Session(database.engine) as db:
        db.add(AppSetting(key="deployment_enrollment:trust", value_json=original, is_secret=False))
        db.commit()

    headers = {"X-AppOps-Session-CSRF": "1"}
    managed_root = "/var/lib/docker/volumes/itworx-managed-appdata/_data"
    updated = identity_client.patch(
        "/api/enrollments/trust/storage-root",
        json={"trusted_appdata_root": managed_root, "managed_appdata_backup_acknowledged": True},
        headers=headers,
    )
    assert updated.status_code == 200
    assert updated.json()["trustedAppdataRoot"] == managed_root
    assert updated.json()["managedAppdataBackupAcknowledged"] is True

    with Session(database.engine) as db:
        saved = db.get(AppSetting, "deployment_enrollment:trust")
        assert saved is not None
        assert saved.value_json == {
            **original,
            "trusted_appdata_root": managed_root,
            "managed_appdata_backup_acknowledged": True,
        }
        event = db.exec(
            select(AuditEvent).where(AuditEvent.action_id == "enrollment.trust.storage_root.update")
        ).one()
        assert event.before_state == {
            "trustedAppdataRoot": "/srv/example/appdata",
            "managedAppdataBackupAcknowledged": False,
        }
        assert event.after_state == {
            "trustedAppdataRoot": managed_root,
            "managedAppdataBackupAcknowledged": True,
        }
        assert event.redacted_metadata == {
            "backupLocation": "dockerRootDir",
            "requestSource": "manual",
        }
        assert "bootstrap_secret_ref" not in str(event.before_state)

    assert identity_client.patch(
        "/api/enrollments/trust/storage-root",
        json={"trusted_appdata_root": managed_root, "managed_appdata_backup_acknowledged": False},
        headers=headers,
    ).status_code == 422
    assert identity_client.patch(
        "/api/enrollments/trust/storage-root",
        json={
            "trusted_appdata_root": "/srv/example/other",
            "managed_appdata_backup_acknowledged": False,
        },
        headers=headers,
    ).status_code == 422
    assert identity_client.patch(
        "/api/enrollments/trust/storage-root",
        json={
            "trusted_appdata_root": managed_root,
            "managed_appdata_backup_acknowledged": True,
            "bootstrap_secret_ref": "LEAKED_SECRET_REFERENCE",
        },
        headers=headers,
    ).status_code == 422

    legacy = identity_client.patch(
        "/api/enrollments/trust/storage-root",
        json={"trusted_appdata_root": "/srv/example/appdata", "managed_appdata_backup_acknowledged": False},
        headers=headers,
    )
    assert legacy.status_code == 200
    assert legacy.json()["trustedAppdataRoot"] == "/srv/example/appdata"
    assert legacy.json()["managedAppdataBackupAcknowledged"] is False


def test_storage_root_patch_returns_conflict_after_concurrent_trust_change(identity_client, monkeypatch):
    from sqlmodel import Session, select

    from backend.app.core.time import utc_now
    from backend.app.db import session as database
    from backend.app.db.models import AppSetting, AuditEvent
    from backend.app.schemas.enrollment import EnrollmentTrustConfigV1
    from backend.app.services.enrollment_plane import EnrollmentPlane

    sign_in(identity_client, "owner")
    original = {
        "controller_identity": "unraid-controller",
        "bootstrap_secret_ref": "ENROLLMENT_BOOTSTRAP",
        "trusted_appdata_root": "/srv/example/appdata",
        "managed_appdata_backup_acknowledged": False,
        "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,
    }
    with Session(database.engine) as db:
        db.add(AppSetting(key="deployment_enrollment:trust", value_json=original, is_secret=False))
        db.commit()

    stale_trust = EnrollmentTrustConfigV1.model_validate(original)

    def read_stale_after_concurrent_update(_session):
        with Session(database.engine) as concurrent_session:
            row = concurrent_session.get(AppSetting, "deployment_enrollment:trust")
            assert row is not None
            row.value_json = {**row.value_json, "approved_networks": ["itworx-mcp-hub-npm", "itworx-npm"]}
            row.updated_at = utc_now()
            concurrent_session.add(row)
            concurrent_session.commit()
        return stale_trust

    monkeypatch.setattr(
        EnrollmentPlane,
        "get_trust_config",
        staticmethod(read_stale_after_concurrent_update),
    )
    response = identity_client.patch(
        "/api/enrollments/trust/storage-root",
        json={
            "trusted_appdata_root": "/var/lib/docker/volumes/itworx-managed-appdata/_data",
            "managed_appdata_backup_acknowledged": True,
        },
        headers={"X-AppOps-Session-CSRF": "1"},
    )

    assert response.status_code == 409
    with Session(database.engine) as db:
        saved = db.get(AppSetting, "deployment_enrollment:trust")
        assert saved is not None
        assert saved.value_json["approved_networks"] == ["itworx-mcp-hub-npm", "itworx-npm"]
        assert saved.value_json["trusted_appdata_root"] == "/srv/example/appdata"
        assert db.exec(
            select(AuditEvent).where(AuditEvent.action_id == "enrollment.trust.storage_root.update")
        ).all() == []


def test_storage_root_patch_cas_rejects_change_after_snapshot(identity_client, monkeypatch):
    from sqlmodel import Session, select

    from backend.app.api import routes_enrollments
    from backend.app.core.time import utc_now
    from backend.app.db import session as database
    from backend.app.db.models import AppSetting, AuditEvent

    sign_in(identity_client, "owner")
    original = {
        "controller_identity": "unraid-controller",
        "bootstrap_secret_ref": "ENROLLMENT_BOOTSTRAP",
        "trusted_appdata_root": "/srv/example/appdata",
        "managed_appdata_backup_acknowledged": False,
        "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,
    }
    with database.engine.connect() as connection:
        connection.exec_driver_sql("PRAGMA journal_mode=WAL")
    with Session(database.engine) as db:
        db.add(AppSetting(key="deployment_enrollment:trust", value_json=original, is_secret=False))
        db.commit()

    actual_sql_update = routes_enrollments.sql_update
    concurrent_write_done = False

    def write_concurrently_after_snapshot(model):
        nonlocal concurrent_write_done
        if model is AppSetting and not concurrent_write_done:
            concurrent_write_done = True
            with Session(database.engine) as concurrent_session:
                row = concurrent_session.get(AppSetting, "deployment_enrollment:trust")
                assert row is not None
                row.value_json = {
                    **row.value_json,
                    "approved_networks": ["itworx-mcp-hub-npm", "itworx-npm"],
                }
                row.updated_at = utc_now()
                concurrent_session.add(row)
                concurrent_session.commit()
        return actual_sql_update(model)

    # The endpoint has already read and validated its selector snapshot when it
    # calls sql_update. Commit an independent trust edit at that boundary so the
    # route's compare-and-swap must reject the now-stale write.
    monkeypatch.setattr(routes_enrollments, "sql_update", write_concurrently_after_snapshot)
    response = identity_client.patch(
        "/api/enrollments/trust/storage-root",
        json={
            "trusted_appdata_root": "/var/lib/docker/volumes/itworx-managed-appdata/_data",
            "managed_appdata_backup_acknowledged": True,
        },
        headers={"X-AppOps-Session-CSRF": "1"},
    )

    assert concurrent_write_done is True
    assert response.status_code == 409
    with Session(database.engine) as db:
        saved = db.get(AppSetting, "deployment_enrollment:trust")
        assert saved is not None
        assert saved.value_json["approved_networks"] == ["itworx-mcp-hub-npm", "itworx-npm"]
        assert saved.value_json["trusted_appdata_root"] == "/srv/example/appdata"
        assert db.exec(
            select(AuditEvent).where(AuditEvent.action_id == "enrollment.trust.storage_root.update")
        ).all() == []


def test_read_only_and_operator_cannot_change_storage_root(identity_client):
    reader_id = sign_in(identity_client, "reader")
    sign_in(identity_client, "owner")
    headers = {"X-AppOps-Session-CSRF": "1"}
    assert identity_client.patch(
        f"/api/auth/users/{reader_id}",
        json={"role": "operator", "enabled": True, "permissions": []},
        headers=headers,
    ).status_code == 200

    selection = {
        "trusted_appdata_root": "/var/lib/docker/volumes/itworx-managed-appdata/_data",
        "managed_appdata_backup_acknowledged": True,
    }
    sign_in(identity_client, "reader")
    assert identity_client.patch(
        "/api/enrollments/trust/storage-root", json=selection, headers=headers
    ).status_code == 403
    sign_in(identity_client, "owner")
    assert identity_client.patch(
        f"/api/auth/users/{reader_id}",
        json={"role": "custom", "enabled": True, "permissions": ["settings.manage"]},
        headers=headers,
    ).status_code == 200
    sign_in(identity_client, "reader")
    assert identity_client.patch(
        "/api/enrollments/trust/storage-root", json=selection, headers=headers
    ).status_code == 403
    sign_in(identity_client, "read-only")
    assert identity_client.patch(
        "/api/enrollments/trust/storage-root", json=selection, headers=headers
    ).status_code == 403