geointel/backend/tests/test_authentik_oidc_service.py

from __future__ import annotations

import time
from urllib.parse import parse_qs, urlsplit

import jwt
import pytest
from cryptography.hazmat.primitives.asymmetric import rsa
from pydantic import ValidationError

from app.core.config import Settings
from app.services.authentik_oidc_service import (
    MAX_OIDC_JSON_BYTES,
    AuthentikOidcService,
)


ISSUER = "https://auth.example.test/application/o/geointel"


def configured_settings(**overrides: object) -> Settings:
    values: dict[str, object] = {
        "auth_enabled": True,
        "auth_username": "ITWorx",
        "auth_password_hash": "pbkdf2_sha256$1$salt$digest",
        "auth_session_secret": "s" * 48,
        "authentik_issuer": ISSUER,
        "authentik_client_id": "geointel-client",
        "authentik_client_secret": "client-secret",
        "authentik_allowed_email": "operator@example.test",
        "public_base_url": "https://geointel.example.test",
    }
    values.update(overrides)
    return Settings(_env_file=None, **values)


def discovery_document() -> dict[str, str]:
    return {
        "issuer": ISSUER,
        "authorization_endpoint": f"{ISSUER}/authorize",
        "token_endpoint": f"{ISSUER}/token",
        "jwks_uri": f"{ISSUER}/jwks",
    }


def test_authentik_configuration_is_all_or_nothing_and_https_only() -> None:
    with pytest.raises(ValidationError, match="configured together"):
        configured_settings(authentik_client_secret=None)
    with pytest.raises(ValidationError, match="absolute HTTPS URL"):
        configured_settings(authentik_issuer="http://auth.example.test/issuer")
    with pytest.raises(ValidationError, match="must not contain a path"):
        configured_settings(public_base_url="https://geointel.example.test/app")


def test_start_uses_same_origin_discovery_and_pkce(monkeypatch: pytest.MonkeyPatch) -> None:
    service = AuthentikOidcService(configured_settings())
    monkeypatch.setattr(service, "_fetch_json", lambda *_args, **_kwargs: discovery_document())

    location, flow_cookie = service.start()

    parsed = urlsplit(location)
    query = parse_qs(parsed.query)
    flow = service.serializer.loads(flow_cookie, max_age=600)
    assert f"{parsed.scheme}://{parsed.netloc}{parsed.path}" == f"{ISSUER}/authorize"
    assert query["redirect_uri"] == [
        "https://geointel.example.test/api/v1/auth/authentik/callback"
    ]
    assert query["code_challenge_method"] == ["S256"]
    assert query["state"] == [flow["state"]]
    assert query["nonce"] == [flow["nonce"]]
    assert query["code_challenge"][0]


def test_discovery_rejects_cross_origin_endpoints(monkeypatch: pytest.MonkeyPatch) -> None:
    service = AuthentikOidcService(configured_settings())
    document = discovery_document()
    document["jwks_uri"] = "https://attacker.example.test/jwks"
    monkeypatch.setattr(service, "_fetch_json", lambda *_args, **_kwargs: document)

    with pytest.raises(ValueError, match="outside the configured issuer origin"):
        service._discovery()


def test_finish_verifies_signature_nonce_and_exact_allowed_email(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    service = AuthentikOidcService(configured_settings())
    state, nonce, verifier = "state-value", "nonce-value", "verifier-value"
    flow_cookie = service.serializer.dumps(
        {"state": state, "nonce": nonce, "verifier": verifier}
    )
    private_key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
    public_jwk = jwt.algorithms.RSAAlgorithm.to_jwk(
        private_key.public_key(), as_dict=True
    )
    public_jwk["kid"] = "operator-key"
    now = int(time.time())
    token = jwt.encode(
        {
            "iss": ISSUER,
            "aud": "geointel-client",
            "sub": "authentik-user-id",
            "iat": now,
            "exp": now + 300,
            "nonce": nonce,
            "email": "Operator@Example.Test",
            "email_verified": True,
        },
        private_key,
        algorithm="RS256",
        headers={"kid": "operator-key"},
    )
    token_holder = {"value": token}

    def fetch(url: str, data: dict[str, str] | None = None) -> dict:
        if url.endswith("openid-configuration"):
            return discovery_document()
        if url.endswith("/token"):
            assert data is not None
            assert data["code_verifier"] == verifier
            return {"id_token": token_holder["value"]}
        if url.endswith("/jwks"):
            return {"keys": [public_jwk]}
        raise AssertionError(url)

    monkeypatch.setattr(service, "_fetch_json", fetch)

    claims = service.finish(code="authorization-code", state=state, flow_cookie=flow_cookie)

    assert claims["sub"] == "authentik-user-id"
    token_holder["value"] = jwt.encode(
        {
            "iss": ISSUER,
            "aud": "geointel-client",
            "sub": "different-user",
            "iat": now,
            "exp": now + 300,
            "nonce": nonce,
            "email": "other@example.test",
            "email_verified": True,
        },
        private_key,
        algorithm="RS256",
        headers={"kid": "operator-key"},
    )
    with pytest.raises(ValueError, match="not authorized"):
        service.finish(code="authorization-code", state=state, flow_cookie=flow_cookie)
    with pytest.raises(ValueError, match="state mismatch"):
        service.finish(
            code="authorization-code",
            state="different-state",
            flow_cookie=flow_cookie,
        )


def test_fetch_json_rejects_declared_oversize_response(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    service = AuthentikOidcService(configured_settings())

    class OversizeResponse:
        headers = {"Content-Length": str(MAX_OIDC_JSON_BYTES + 1)}

        def __enter__(self):
            return self

        def __exit__(self, *_args: object) -> None:
            return None

        def read(self, _size: int) -> bytes:
            raise AssertionError("oversized responses must not be read")

    class Opener:
        def open(self, *_args: object, **_kwargs: object) -> OversizeResponse:
            return OversizeResponse()

    monkeypatch.setattr(
        "app.services.authentik_oidc_service.build_opener",
        lambda *_args: Opener(),
    )

    with pytest.raises(ValueError, match="size limit"):
        service._fetch_json(f"{ISSUER}/oversized")