local-ci-cd-system/tests/python/test_web_auth.py

"""Tests for Gitea OIDC auth, signed sessions, and the allowlist (W5).

SECURITY-CRITICAL: this suite is the contract for the whole auth layer.
No real network call, no real keyring, no real Gitea:

- The OS keyring is faked by monkeypatching
  ``KeyringCredentialStore.get`` (see :func:`_install_fake_keyring`) to
  return deterministic secrets for two known keyring key *names*.
- Authlib's OAuth client is faked by swapping ``app.state.oauth.gitea``
  (built by ``register_oauth`` from real ``OAuth()``, which never touches
  the network on its own -- OIDC discovery is lazy) with
  :class:`_StubOAuthClient`, whose ``authorize_redirect`` /
  ``authorize_access_token`` / ``userinfo`` are async and return a chosen
  claim set. This is the approach the W5 plan calls out explicitly.
"""

from __future__ import annotations

import logging
import shutil
from pathlib import Path
from typing import Any

import pytest
from fastapi.responses import RedirectResponse
from fastapi.testclient import TestClient

from ci_orchestrator.config import (
    BackendConfig,
    CargoCacheConfig,
    Config,
    Paths,
    WebConfig,
    WebOidcConfig,
)
from ci_orchestrator.credentials import Credential, KeyringCredentialStore
from ci_orchestrator.web.app import create_app, get_backend
from ci_orchestrator.web.auth import (
    _read_secret,
    configure_auth,
    register_oauth,
    require_user,
)

_SESSION_CRED_NAME = "TestWebSessionSecret"
_OIDC_CRED_NAME = "TestWebOidcClientSecret"
_FAKE_SESSION_SECRET = "sess-secret-do-not-leak-abc123"
_FAKE_OIDC_CLIENT_SECRET = "oidc-client-secret-do-not-leak-xyz987"

_PROTECTED_ROUTES = ["/", "/executions", "/cache", "/host"]


# ── shared fixtures / fakes ─────────────────────────────────────────────────


def _install_fake_keyring(monkeypatch: pytest.MonkeyPatch) -> None:
    """No real ``keyring`` package touched -- deterministic secrets only."""
    secrets = {
        _SESSION_CRED_NAME: _FAKE_SESSION_SECRET,
        _OIDC_CRED_NAME: _FAKE_OIDC_CLIENT_SECRET,
    }

    def fake_get(self: KeyringCredentialStore, target: str) -> Credential:
        if target not in secrets:
            raise KeyError(target)
        return Credential(username=target, password=secrets[target])

    monkeypatch.setattr(KeyringCredentialStore, "get", fake_get)


def _quiet_host(monkeypatch: pytest.MonkeyPatch) -> None:
    """No real disk/systemctl calls leak into route rendering."""
    from collections import namedtuple

    _DiskUsage = namedtuple("_DiskUsage", ["total", "used", "free"])
    gb = 1024**3
    monkeypatch.setattr(
        shutil, "disk_usage", lambda _p: _DiskUsage(total=100 * gb, used=40 * gb, free=60 * gb)
    )
    monkeypatch.setattr(shutil, "which", lambda _name: None)


def _make_config(
    root: Path, *, active: bool, allowed_users: list[str] | None = None
) -> Config:
    oidc = None
    if active:
        oidc = WebOidcConfig(
            issuer="https://gitea.example.lan",
            client_id="ci-web",
            redirect_url="http://127.0.0.1:8800/auth/callback",
            client_secret_cred=_OIDC_CRED_NAME,
            scopes=["openid", "profile"],
        )
    return Config(
        paths=Paths(
            root=root,
            templates=root / "templates",
            build_vms=root / "build-vms",
            artifacts=root / "artifacts",
            keys=root / "keys",
        ),
        backend=BackendConfig(),
        cargo_cache=CargoCacheConfig(
            enabled=True, root=root / "cache" / "cargo", retention_days=30
        ),
        web=WebConfig(
            enabled=active,
            allowed_users=allowed_users if allowed_users is not None else [],
            session_secret_cred=_SESSION_CRED_NAME,
            oidc=oidc,
        ),
    )


class _StubOAuthClient:
    """Fake Authlib ``gitea`` client -- no network, chosen claims."""

    def __init__(
        self,
        *,
        token: dict[str, Any] | None = None,
        userinfo_claims: dict[str, Any] | None = None,
    ) -> None:
        self.token: dict[str, Any] = token if token is not None else {}
        self.userinfo_claims = userinfo_claims
        self.userinfo_calls = 0
        self.authorize_redirect_calls: list[dict[str, Any]] = []

    async def authorize_redirect(
        self, request: Any, redirect_uri: str | None = None, **kwargs: Any
    ) -> RedirectResponse:
        self.authorize_redirect_calls.append({"redirect_uri": redirect_uri, **kwargs})
        return RedirectResponse(
            f"https://gitea.example.lan/login/oauth/authorize?redirect_uri={redirect_uri}",
            status_code=302,
        )

    async def authorize_access_token(self, request: Any, **kwargs: Any) -> dict[str, Any]:
        return self.token

    async def userinfo(self, **kwargs: Any) -> dict[str, Any]:
        self.userinfo_calls += 1
        return self.userinfo_claims or {}


def _install_stub_oauth(app: Any, stub: _StubOAuthClient) -> None:
    """Swap the real (lazy-discovery) ``gitea`` client for a fake one."""
    app.state.oauth.gitea = stub


# ── 1. auth inactive: open mode, W4 parity ─────────────────────────────────


def test_auth_inactive_root_is_200(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
    _install_fake_keyring(monkeypatch)
    _quiet_host(monkeypatch)
    app = create_app(_make_config(tmp_path, active=False))
    app.dependency_overrides[get_backend] = lambda: None
    client = TestClient(app)

    resp = client.get("/")

    assert resp.status_code == 200


def test_auth_inactive_login_routes_not_registered(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    """Open mode never wires /login etc. -- they simply don't exist."""
    _install_fake_keyring(monkeypatch)
    app = create_app(_make_config(tmp_path, active=False))
    client = TestClient(app, follow_redirects=False)

    resp = client.get("/login")

    assert resp.status_code == 404


# ── 2. auth active, no session: redirect to /login ──────────────────────────


def test_auth_active_no_session_redirects_to_login(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    _install_fake_keyring(monkeypatch)
    app = create_app(_make_config(tmp_path, active=True, allowed_users=["simone"]))
    client = TestClient(app, follow_redirects=False)

    resp = client.get("/")

    assert resp.status_code in (302, 303)
    assert "/login" in resp.headers["location"]


@pytest.mark.parametrize("route", _PROTECTED_ROUTES)
def test_auth_active_no_session_redirects_every_protected_route(
    route: str, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    _install_fake_keyring(monkeypatch)
    app = create_app(_make_config(tmp_path, active=True, allowed_users=["simone"]))
    client = TestClient(app, follow_redirects=False)

    resp = client.get(route)

    assert resp.status_code in (302, 303)
    assert "/login" in resp.headers["location"]


# ── 3. /healthz stays open even when auth is active ─────────────────────────


def test_healthz_is_200_without_session_when_auth_active(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    _install_fake_keyring(monkeypatch)
    app = create_app(_make_config(tmp_path, active=True, allowed_users=["simone"]))
    client = TestClient(app, follow_redirects=False)

    resp = client.get("/healthz")

    assert resp.status_code == 200
    assert resp.json() == {"status": "ok"}


def test_static_stays_open_when_auth_active(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    _install_fake_keyring(monkeypatch)
    app = create_app(_make_config(tmp_path, active=True, allowed_users=["simone"]))
    client = TestClient(app, follow_redirects=False)

    resp = client.get("/static/app.css")

    assert resp.status_code == 200


# ── 4. allowlisted callback -> session set -> subsequent GET / works ────────


def test_callback_allowlisted_user_sets_session_and_redirects_home(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    _install_fake_keyring(monkeypatch)
    _quiet_host(monkeypatch)
    app = create_app(_make_config(tmp_path, active=True, allowed_users=["simone"]))
    app.dependency_overrides[get_backend] = lambda: None
    stub = _StubOAuthClient(token={"userinfo": {"preferred_username": "simone"}})
    _install_stub_oauth(app, stub)
    client = TestClient(app, follow_redirects=False)

    callback_resp = client.get("/auth/callback")

    assert callback_resp.status_code == 303
    assert callback_resp.headers["location"] == "/"

    home_resp = client.get("/")
    assert home_resp.status_code == 200


def test_callback_falls_back_to_userinfo_endpoint_when_token_has_none(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    _install_fake_keyring(monkeypatch)
    app = create_app(_make_config(tmp_path, active=True, allowed_users=["simone"]))
    stub = _StubOAuthClient(
        token={"access_token": "at"}, userinfo_claims={"preferred_username": "simone"}
    )
    _install_stub_oauth(app, stub)
    client = TestClient(app, follow_redirects=False)

    resp = client.get("/auth/callback")

    assert resp.status_code == 303
    assert stub.userinfo_calls == 1


def test_callback_falls_back_to_login_claim_when_preferred_username_absent(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    _install_fake_keyring(monkeypatch)
    app = create_app(_make_config(tmp_path, active=True, allowed_users=["alice"]))
    stub = _StubOAuthClient(token={"userinfo": {"login": "alice"}})
    _install_stub_oauth(app, stub)
    client = TestClient(app, follow_redirects=False)

    resp = client.get("/auth/callback")

    assert resp.status_code == 303


# ── 5. non-allowlisted user -> 403, no session ───────────────────────────────


def test_callback_non_allowlisted_user_is_403_and_creates_no_session(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    _install_fake_keyring(monkeypatch)
    app = create_app(_make_config(tmp_path, active=True, allowed_users=["simone"]))
    stub = _StubOAuthClient(token={"userinfo": {"preferred_username": "mallory"}})
    _install_stub_oauth(app, stub)
    client = TestClient(app, follow_redirects=False)

    callback_resp = client.get("/auth/callback")
    assert callback_resp.status_code == 403

    home_resp = client.get("/")
    assert home_resp.status_code in (302, 303)
    assert "/login" in home_resp.headers["location"]


def test_callback_userinfo_with_no_recognized_username_claim_is_403(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    _install_fake_keyring(monkeypatch)
    app = create_app(_make_config(tmp_path, active=True, allowed_users=["simone"]))
    stub = _StubOAuthClient(token={"userinfo": {"email": "[email protected]"}})
    _install_stub_oauth(app, stub)
    client = TestClient(app, follow_redirects=False)

    resp = client.get("/auth/callback")

    assert resp.status_code == 403


# ── 6. empty allowlist -> fail closed, even a "valid" OIDC user is denied ───


def test_empty_allowlist_denies_even_a_valid_oidc_user(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    _install_fake_keyring(monkeypatch)
    app = create_app(_make_config(tmp_path, active=True, allowed_users=[]))
    stub = _StubOAuthClient(token={"userinfo": {"preferred_username": "simone"}})
    _install_stub_oauth(app, stub)
    client = TestClient(app, follow_redirects=False)

    resp = client.get("/auth/callback")

    assert resp.status_code == 403


# ── 7. logout clears the session ────────────────────────────────────────────


def test_logout_clears_session_and_subsequent_get_redirects_to_login(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    _install_fake_keyring(monkeypatch)
    _quiet_host(monkeypatch)
    app = create_app(_make_config(tmp_path, active=True, allowed_users=["simone"]))
    app.dependency_overrides[get_backend] = lambda: None
    stub = _StubOAuthClient(token={"userinfo": {"preferred_username": "simone"}})
    _install_stub_oauth(app, stub)
    client = TestClient(app, follow_redirects=False)

    client.get("/auth/callback")
    assert client.get("/").status_code == 200  # logged in

    logout_resp = client.get("/logout")
    assert logout_resp.status_code == 303

    after_resp = client.get("/")
    assert after_resp.status_code in (302, 303)
    assert "/login" in after_resp.headers["location"]


# ── 8. no secret value ever leaks into a response body or the logs ─────────


def test_no_secret_value_leaks_into_responses_or_logs(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
) -> None:
    _install_fake_keyring(monkeypatch)
    _quiet_host(monkeypatch)
    caplog.set_level(logging.DEBUG)

    app = create_app(_make_config(tmp_path, active=True, allowed_users=["simone"]))
    app.dependency_overrides[get_backend] = lambda: None
    allowlisted_stub = _StubOAuthClient(token={"userinfo": {"preferred_username": "simone"}})
    _install_stub_oauth(app, allowlisted_stub)
    client = TestClient(app, follow_redirects=False)

    bodies: list[str] = []

    bodies.append(client.get("/").text)  # redirect to /login
    bodies.append(client.get("/login").text)  # real authorize_redirect not stubbed here
    bodies.append(client.get("/auth/callback").text)  # session set
    bodies.append(client.get("/").text)  # 200, dashboard
    bodies.append(client.get("/logout").text)

    denied_stub = _StubOAuthClient(token={"userinfo": {"preferred_username": "mallory"}})
    _install_stub_oauth(app, denied_stub)
    bodies.append(client.get("/auth/callback").text)  # 403

    for body in bodies:
        assert _FAKE_SESSION_SECRET not in body
        assert _FAKE_OIDC_CLIENT_SECRET not in body

    log_text = caplog.text
    assert _FAKE_SESSION_SECRET not in log_text
    assert _FAKE_OIDC_CLIENT_SECRET not in log_text


# ── require_user / _read_secret / register_oauth: direct unit coverage ─────


def test_require_user_inactive_returns_anonymous_without_touching_session(
    tmp_path: Path,
) -> None:
    from types import SimpleNamespace

    config = _make_config(tmp_path, active=False)
    fake_request = SimpleNamespace(app=SimpleNamespace(state=SimpleNamespace(config=config)))

    assert require_user(fake_request) == "anonymous"  # type: ignore[arg-type]


def test_read_secret_returns_keyring_password(monkeypatch: pytest.MonkeyPatch) -> None:
    _install_fake_keyring(monkeypatch)

    assert _read_secret(_SESSION_CRED_NAME) == _FAKE_SESSION_SECRET


def test_read_secret_raises_runtime_error_naming_only_the_key(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    _install_fake_keyring(monkeypatch)

    with pytest.raises(RuntimeError) as exc_info:
        _read_secret("SomeMissingCred")

    message = str(exc_info.value)
    assert "SomeMissingCred" in message
    assert _FAKE_SESSION_SECRET not in message
    assert _FAKE_OIDC_CLIENT_SECRET not in message


def test_register_oauth_raises_without_oidc_config(tmp_path: Path) -> None:
    config = _make_config(tmp_path, active=False)

    with pytest.raises(RuntimeError):
        register_oauth(config)


def test_register_oauth_builds_gitea_client_without_network(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    _install_fake_keyring(monkeypatch)
    config = _make_config(tmp_path, active=True, allowed_users=["simone"])

    oauth = register_oauth(config)

    assert oauth.gitea is not None


def test_configure_auth_noop_when_oidc_none(tmp_path: Path) -> None:
    from fastapi import FastAPI

    app = FastAPI()
    config = _make_config(tmp_path, active=False)

    configure_auth(app, config)

    assert not hasattr(app.state, "oauth")