feat(web): Gitea OIDC auth, signed session, allowlist (W5)
This commit is contained in:
@@ -0,0 +1,465 @@
|
||||
"""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 = ["/", "/vms", "/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")
|
||||
Reference in New Issue
Block a user