v2x-server/tests/auth/conftest.py

"""Token-forging fixtures for exercising the validator against real crypto.

A locally generated RSA keypair stands in for Keycloak's realm keys, and the
provider's JWKS fetch is redirected at it. That makes it possible to mint
tokens that are genuinely signed -- and genuinely wrong in one specific way per
test -- without a running Keycloak.
"""

from __future__ import annotations

import datetime as dt
import json
from collections.abc import Callable, Iterator
from typing import Any

import jwt
import pytest
from cryptography.hazmat.primitives.asymmetric import rsa
from jwt.utils import base64url_encode

from v2x_server.core.config import Settings

TEST_KID = "test-key-1"


@pytest.fixture(scope="session")
def rsa_key() -> rsa.RSAPrivateKey:
    return rsa.generate_private_key(public_exponent=65537, key_size=2048)


@pytest.fixture(scope="session")
def jwks(rsa_key: rsa.RSAPrivateKey) -> dict[str, Any]:
    """The public half, in the JWKS shape Keycloak would serve."""
    public_numbers = rsa_key.public_key().public_numbers()

    def _b64(value: int) -> str:
        length = (value.bit_length() + 7) // 8
        return base64url_encode(value.to_bytes(length, "big")).decode()

    return {
        "keys": [
            {
                "kty": "RSA",
                "kid": TEST_KID,
                "use": "sig",
                "alg": "RS256",
                "n": _b64(public_numbers.n),
                "e": _b64(public_numbers.e),
            }
        ]
    }


@pytest.fixture
def auth_settings() -> Settings:
    return Settings(
        app_env="test",
        keycloak_issuer="https://idp.test/realms/v2x",
        keycloak_internal_url="https://idp.test",
        keycloak_realm="v2x",
        keycloak_audience="v2x-api",
    )


@pytest.fixture
def make_token(rsa_key: rsa.RSAPrivateKey, auth_settings: Settings) -> Callable[..., str]:
    """Mint a signed token, overriding any claim or header field per test."""

    def _make(
        *,
        kid: str | None = TEST_KID,
        algorithm: str = "RS256",
        key: Any = None,
        expires_in: int = 300,
        drop: tuple[str, ...] = (),
        **claim_overrides: Any,
    ) -> str:
        now = dt.datetime.now(tz=dt.UTC)
        claims: dict[str, Any] = {
            "sub": "11111111-1111-1111-1111-111111111111",
            "iss": auth_settings.keycloak_issuer,
            "aud": auth_settings.keycloak_audience,
            "iat": now,
            "exp": now + dt.timedelta(seconds=expires_in),
            "preferred_username": "vera",
            "email": "[email protected]",
            "scope": "openid profile email",
            "realm_access": {"roles": ["viewer"]},
            "resource_access": {"v2x-api": {"roles": ["device.read"]}},
        }
        claims.update(claim_overrides)
        for claim in drop:
            claims.pop(claim, None)
        headers = {"kid": kid} if kid else {}
        return jwt.encode(
            claims,
            key if key is not None else rsa_key,
            algorithm=algorithm,
            headers=headers,
        )

    return _make


@pytest.fixture
def unsigned_token(auth_settings: Settings) -> str:
    """An `alg: none` token -- the classic downgrade attempt."""
    now = dt.datetime.now(tz=dt.UTC)
    header = base64url_encode(
        json.dumps({"alg": "none", "typ": "JWT", "kid": TEST_KID}).encode()
    ).decode()
    payload = base64url_encode(
        json.dumps(
            {
                "sub": "attacker",
                "iss": auth_settings.keycloak_issuer,
                "aud": auth_settings.keycloak_audience,
                "iat": int(now.timestamp()),
                "exp": int((now + dt.timedelta(minutes=5)).timestamp()),
                "realm_access": {"roles": ["admin"]},
            }
        ).encode()
    ).decode()
    return f"{header}.{payload}."


@pytest.fixture
def provider(auth_settings: Settings, jwks: dict[str, Any]) -> Iterator[Any]:
    """An OIDCProvider whose network calls are served from the fixtures above."""
    import httpx

    from v2x_server.auth.keycloak import OIDCProvider

    certs_url = (
        f"{auth_settings.keycloak_internal_url}"
        f"/realms/{auth_settings.keycloak_realm}/protocol/openid-connect/certs"
    )

    def handler(request: httpx.Request) -> httpx.Response:
        url = str(request.url)
        if url.endswith("/.well-known/openid-configuration"):
            return httpx.Response(
                200,
                json={
                    "issuer": auth_settings.keycloak_issuer,
                    "jwks_uri": certs_url,
                },
            )
        if url.endswith("/protocol/openid-connect/certs"):
            return httpx.Response(200, json=jwks)
        return httpx.Response(404)

    client = httpx.AsyncClient(transport=httpx.MockTransport(handler))
    yield OIDCProvider(auth_settings, client)