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

from __future__ import annotations

import os
from contextlib import contextmanager

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
_PROVISION_PATH = "/api/apps/authentik/forge/oidc/provision"


@pytest.fixture
def forge_identity_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
    from backend.app.services import forge_oidc_provisioning

    if os.name == "nt":
        # The shared Windows runner temp ACL is deliberately not the production
        # data-directory ACL; isolate route tests from that host configuration.
        monkeypatch.setattr(
            forge_oidc_provisioning,
            "_windows_private_path",
            lambda *_args, **_kwargs: None,
        )
    else:
        monkeypatch.setattr(
            forge_oidc_provisioning,
            "_FORGE_OIDC_LOCK_ROOT",
            tmp_path / "forge-oidc-runtime-lock",
        )

    engine = create_engine(
        f"sqlite:///{tmp_path / 'forge-oidc-auth.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(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 test_forge_oidc_route_requires_authenticated_operator_and_csrf(
    forge_identity_client: TestClient,
    monkeypatch: pytest.MonkeyPatch,
):
    from backend.app.services import setup_service

    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)
    unauthenticated = forge_identity_client.post(_PROVISION_PATH, json={})
    assert unauthenticated.status_code == 401

    _sign_in(forge_identity_client, "owner")
    missing_csrf = forge_identity_client.post(_PROVISION_PATH, json={})
    assert missing_csrf.status_code == 403
    assert loader_calls == []


def test_forge_oidc_route_rejects_read_only_user_and_nonempty_body(
    forge_identity_client: TestClient,
    monkeypatch: pytest.MonkeyPatch,
):
    from backend.app.services import setup_service

    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)
    _sign_in(forge_identity_client, "reader")
    denied = forge_identity_client.post(
        _PROVISION_PATH,
        json={},
        headers={"X-AppOps-Session-CSRF": "1"},
    )
    assert denied.status_code == 403

    _sign_in(forge_identity_client, "owner")
    invalid = forge_identity_client.post(
        _PROVISION_PATH,
        json={"client_id": "caller-selected"},
        headers={"X-AppOps-Session-CSRF": "1"},
    )
    assert invalid.status_code == 422
    missing_body = forge_identity_client.post(
        _PROVISION_PATH,
        headers={"X-AppOps-Session-CSRF": "1"},
    )
    assert missing_body.status_code == 422
    assert loader_calls == []


def test_forge_oidc_route_loads_stored_connector_settings_and_passes_actor(
    forge_identity_client: TestClient,
    monkeypatch: pytest.MonkeyPatch,
):
    from backend.app.core.config import settings
    from backend.app.db import session as db_session
    from backend.app.db.models import AppSetting
    from backend.app.services import forge_oidc_provisioning, setup_service
    from backend.app.services.secret_store import set_secret

    _sign_in(forge_identity_client, "owner")
    loader_calls = []
    constructed = []
    outcome_identities = []
    events = []
    protected_admin_token = "protected-authentik-admin-token-fixture"
    monkeypatch.setattr(settings, "environment", "production")
    monkeypatch.setattr(settings, "secret_key", "test-key-" + "x" * 36)
    monkeypatch.setattr(settings, "secret_key_source", "env")
    with Session(db_session.engine) as session:
        session.add(
            AppSetting(
                key="connector:authentik:config",
                value_json={"url": "https://auth.example", "verifyTls": True},
                is_secret=False,
            )
        )
        set_secret(session, "AUTHENTIK_TOKEN", protected_admin_token)
        session.commit()

    original_loader = setup_service.apply_stored_connector_settings
    original_lock = forge_oidc_provisioning._forge_oidc_process_lock
    original_store_status = forge_oidc_provisioning.secret_storage_status

    @contextmanager
    def tracked_lock():
        with original_lock():
            events.append("lock-enter")
            yield
            events.append("lock-exit")

    def tracked_store_status():
        events.append("store-check")
        return original_store_status()

    def tracked_loader(session, connector_ids=None):
        events.append("connector-loader")
        loader_calls.append(connector_ids)
        assert connector_ids == {"authentik"}
        return original_loader(session, connector_ids)

    class FakeAuthentikConnector:
        def __init__(self):
            constructed.append(
                (
                    settings.apis.get("AUTHENTIK_URL"),
                    settings.apis.get("AUTHENTIK_VERIFY_TLS"),
                    settings.apis.get("AUTHENTIK_TOKEN"),
                )
            )

        def ensure_forge_oidc_prerequisites(self, *, expected_client_secret):
            assert expected_client_secret is None
            return {"success": False, "status": "failed"}

    original_outcome = forge_oidc_provisioning._connector_outcome

    def tracked_outcome(session, *, identity, outcome, secret_created):
        outcome_identities.append(identity)
        return original_outcome(
            session,
            identity=identity,
            outcome=outcome,
            secret_created=secret_created,
        )

    monkeypatch.setattr(setup_service, "apply_stored_connector_settings", tracked_loader)
    monkeypatch.setattr(forge_oidc_provisioning, "_forge_oidc_process_lock", tracked_lock)
    monkeypatch.setattr(forge_oidc_provisioning, "secret_storage_status", tracked_store_status)
    monkeypatch.setattr(forge_oidc_provisioning, "AuthentikConnector", FakeAuthentikConnector)
    monkeypatch.setattr(forge_oidc_provisioning, "_connector_outcome", tracked_outcome)
    response = forge_identity_client.post(
        _PROVISION_PATH,
        json={},
        headers={"X-AppOps-Session-CSRF": "1"},
    )

    assert response.status_code == 200
    assert response.json()["status"] == "failed"
    assert response.json()["message"] == "forge_oidc_provisioning_failed"
    assert constructed == [("https://auth.example", True, protected_admin_token)]
    assert loader_calls == [{"authentik"}]
    assert events == ["lock-enter", "store-check", "connector-loader", "lock-exit"]
    assert outcome_identities == ["authentik:owner"]
    assert protected_admin_token not in response.text
    assert "client_secret" not in response.json()


def test_forge_oidc_production_store_not_ready_does_not_load_connector(
    forge_identity_client: TestClient,
    monkeypatch: pytest.MonkeyPatch,
):
    from backend.app.core.config import settings
    from backend.app.services import setup_service

    _sign_in(forge_identity_client, "owner")
    monkeypatch.setattr(settings, "environment", "production")
    monkeypatch.setattr(settings, "secret_key", "")
    monkeypatch.setattr(settings, "secret_key_source", "missing")
    loader_calls = []
    original_loader = setup_service.apply_stored_connector_settings

    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)
    response = forge_identity_client.post(
        _PROVISION_PATH,
        json={},
        headers={"X-AppOps-Session-CSRF": "1"},
    )

    assert response.status_code == 200
    assert response.json()["status"] == "blocked"
    assert response.json()["message"] == "forge_oidc_secret_store_not_production_ready"
    assert loader_calls == []


def test_forge_oidc_route_returns_blocked_when_exact_policy_is_disabled(
    forge_identity_client: TestClient,
    monkeypatch: pytest.MonkeyPatch,
):
    from backend.app.db import session as db_session
    from backend.app.db.models import Policy
    from backend.app.services import setup_service

    loader_calls = []
    original_loader = setup_service.apply_stored_connector_settings

    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)
    _sign_in(forge_identity_client, "owner")
    with Session(db_session.engine) as session:
        policy = session.exec(
            select(Policy).where(
                (Policy.action == "forge.oidc.provision")
                & (Policy.target == "itworx-forge")
            )
        ).one()
        policy.manual = "blocked"
        session.add(policy)
        session.commit()

    response = forge_identity_client.post(
        _PROVISION_PATH,
        json={},
        headers={"X-AppOps-Session-CSRF": "1"},
    )
    assert response.status_code == 200
    assert response.json()["status"] == "blocked"
    assert response.json()["message"] == "forge_oidc_exact_target_policy_required"
    assert loader_calls == []