vergabe-teilnahme/vergabe_teilnahme/apps/accounts/test_oidc.py

"""Exercise the actual signed-token and browser-session admission boundaries."""

import time
from types import SimpleNamespace
from urllib.parse import parse_qs, urlsplit

import jwt
import pytest
from cryptography.hazmat.primitives.asymmetric import rsa
from django.test import Client
from django.urls import get_script_prefix, set_script_prefix

from . import oidc
from .models import Mitarbeiter, OIDCIdentity

pytestmark = pytest.mark.django_db
ISSUER = "https://kc.example.test"
TENANT = "tenant:trial:demo-company"
CLIENT = "vergabe-demo-company"


@pytest.fixture
def configured(settings, monkeypatch):
    settings.NETKINGDOM_ENABLED = True
    settings.NETKINGDOM_ISSUER = ISSUER
    settings.NETKINGDOM_CLIENT_ID = CLIENT
    settings.NETKINGDOM_TENANT = TENANT
    settings.NETKINGDOM_CALLBACK = "https://vergabe.example.test/accounts/oidc/callback/"
    settings.COMPANY_DISPLAY_NAME = "Demo Company"
    settings.APP_BASE_PATH = ""
    metadata = {
        "issuer": ISSUER,
        "authorization_endpoint": ISSUER + "/authorize",
        "token_endpoint": ISSUER + "/token",
        "jwks_uri": ISSUER + "/jwks",
        "code_challenge_methods_supported": ["S256"],
        "id_token_signing_alg_values_supported": ["RS256"],
    }
    monkeypatch.setattr(oidc, "discovery", lambda *args: metadata)
    return settings


@pytest.fixture
def signed_flow(configured, monkeypatch):
    key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
    monkeypatch.setattr(
        oidc,
        "key_client",
        lambda uri: SimpleNamespace(
            get_signing_key_from_jwt=lambda token: SimpleNamespace(key=key.public_key()),
        ),
    )

    def exchange(client, changes=None, token_override=None):
        pending = client.session["oidc_pending"]
        claims = {
            "iss": ISSUER,
            "sub": "subject-1",
            "aud": CLIENT,
            "iat": int(time.time()),
            "exp": int(time.time()) + 600,
            "nonce": pending["nonce"],
            "tenant": TENANT,
            "tenant_source": "directory",
            "principal_type": "human",
            "groups": [TENANT + ":users"],
            "roles": ["user"],
            "assurance": {"level": "aal1"},
            "preferred_username": "Demo member",
        }
        claims.update(changes or {})
        token = token_override or jwt.encode(
            claims, key, algorithm="RS256", headers={"kid": "test"}
        )
        monkeypatch.setattr(oidc, "read_json", lambda *args: {"id_token": token})
        return client.get(
            "/accounts/oidc/callback/", {"code": "single-code", "state": pending["state"]}
        )

    return exchange


def test_welcome_does_not_log_in_and_start_requires_csrf(configured):
    client = Client(enforce_csrf_checks=True)
    response = client.get("/accounts/login/?next=https://evil.test&token=untrusted")
    assert b"Demo Company" in response.content
    assert b"Mit NetKingdom anmelden" in response.content
    assert "_auth_user_id" not in client.session
    assert client.get("/accounts/oidc/start/").status_code == 405
    assert client.post("/accounts/oidc/start/").status_code == 403
    response = client.post(
        "/accounts/oidc/start/",
        {
            "csrfmiddlewaretoken": client.cookies["csrftoken"].value,
        },
    )
    query = parse_qs(urlsplit(response.url).query)
    assert query["prompt"] == ["login"]
    assert query["code_challenge_method"] == ["S256"]
    assert query["tenant_hint"] == [TENANT]
    assert len(query["nonce"][0]) >= 32
    assert "evil.test" not in response.url


def test_verified_login_needs_explicit_confirmation_and_stable_mapping(client, signed_flow):
    assert client.post("/accounts/oidc/start/").status_code == 302
    assert signed_flow(client).url == "/accounts/oidc/confirm/"
    assert not Mitarbeiter.objects.exists()
    assert client.get("/ausschreibungen/").status_code == 302
    assert b"Demo member" in client.get("/accounts/oidc/confirm/").content
    assert client.post("/accounts/oidc/confirm/").url == "/"
    user = Mitarbeiter.objects.get()
    assert not user.is_staff and not user.is_superuser and not user.has_usable_password()
    assert OIDCIdentity.objects.get().subject == "subject-1"
    dashboard = client.get("/ausschreibungen/")
    assert dashboard.status_code == 200
    assert b"Demo Company" in dashboard.content
    assert "access_token" not in repr(dict(client.session))
    assert client.post("/accounts/oidc/start/").status_code == 302
    assert "_auth_user_id" not in client.session
    assert signed_flow(client, {"preferred_username": "New display name"}).status_code == 302
    assert client.post("/accounts/oidc/confirm/").status_code == 302
    assert Mitarbeiter.objects.count() == 1


@pytest.mark.parametrize(
    "changes",
    [
        {"iss": "https://evil.test"},
        {"aud": "other-client"},
        {"nonce": "wrong"},
        {"exp": 1},
        {"iat": int(time.time()) + 3600},
        {"nbf": int(time.time()) + 3600},
        {"tenant": "tenant:platform"},
        {"tenant_source": "registration"},
        {"principal_type": "service"},
        {"groups": []},
        {"groups": [TENANT + ":users", "netkingdom-suspended"]},
        {"roles": ["platform-operator"]},
        {"groups": [TENANT + ":users", "net-kingdom-admins"]},
        {"assurance": {"level": "aal0"}},
        {"sub": ""},
        {"roles": "user"},
        {"aud": [CLIENT, "other"], "azp": "other"},
    ],
)
def test_invalid_tokens_never_create_users(client, signed_flow, changes):
    client.post("/accounts/oidc/start/")
    response = signed_flow(client, changes)
    assert response.status_code == 403
    assert not Mitarbeiter.objects.exists()
    assert "_auth_user_id" not in client.session


def test_wrong_algorithm_denied(client, signed_flow):
    client.post("/accounts/oidc/start/")
    assert (
        signed_flow(
            client, token_override=jwt.encode({"sub": "forged"}, "x" * 32, algorithm="HS256")
        ).status_code
        == 403
    )


def test_state_is_browser_bound_expired_and_one_use(client, signed_flow):
    client.post("/accounts/oidc/start/")
    original = client.session["oidc_pending"]
    other = Client()
    response = other.get("/accounts/oidc/callback/", {"state": original["state"], "code": "x"})
    assert response.status_code == 403
    assert signed_flow(client).status_code == 302
    assert (
        client.get(
            "/accounts/oidc/callback/", {"state": original["state"], "code": "x"}
        ).status_code
        == 403
    )
    assert client.post("/accounts/oidc/confirm/").status_code == 403
    client.post("/accounts/oidc/start/")
    session = client.session
    session["oidc_pending"]["created"] -= 601
    session.save()
    assert signed_flow(client).status_code == 403


def test_confirmation_requires_csrf(client, signed_flow):
    client.post("/accounts/oidc/start/")
    signed_flow(client)
    guarded = Client(enforce_csrf_checks=True)
    guarded.cookies = client.cookies
    assert guarded.post("/accounts/oidc/confirm/").status_code == 403
    assert not Mitarbeiter.objects.exists()


def test_no_email_merge_and_inactive_or_staff_accounts_stay_denied(client, signed_flow):
    existing = Mitarbeiter.objects.create_user(username="Demo member", email="same@example.test")
    client.post("/accounts/oidc/start/")
    signed_flow(client, {"email": existing.email})
    client.post("/accounts/oidc/confirm/")
    linked = OIDCIdentity.objects.get().user
    assert linked.pk != existing.pk
    linked.is_active = False
    linked.save()
    assert client.get("/ausschreibungen/").status_code == 302
    client.post("/accounts/oidc/start/")
    signed_flow(client)
    assert client.post("/accounts/oidc/confirm/").status_code == 403
    linked.is_active = True
    linked.is_staff = True
    linked.save()
    client.post("/accounts/oidc/start/")
    signed_flow(client)
    assert client.post("/accounts/oidc/confirm/").status_code == 403


def test_absolute_session_expiry_and_wrong_identity_do_not_reuse_operator(client, signed_flow):
    operator = Mitarbeiter.objects.create_superuser(username="operator", password="test-only")
    client.force_login(operator)
    client.post("/accounts/oidc/start/")
    assert "_auth_user_id" not in client.session
    assert signed_flow(client, {"tenant": "tenant:platform"}).status_code == 403
    client.post("/accounts/oidc/start/")
    signed_flow(client)
    client.post("/accounts/oidc/confirm/")
    session = client.session
    session["oidc_expires"] = time.time() - 1
    session.save()
    assert client.get("/ausschreibungen/").status_code == 302


def test_exact_company_callback_and_prefixed_links(configured, client):
    old = get_script_prefix()
    configured.APP_BASE_PATH = "/demo-company"
    configured.FORCE_SCRIPT_NAME = "/demo-company"
    configured.NETKINGDOM_CALLBACK = (
        "https://vergabe.example.test/demo-company/accounts/oidc/callback/"
    )
    set_script_prefix("/demo-company")
    try:
        page = client.get("/accounts/login/")
        assert b"/demo-company/accounts/oidc/start/" in page.content
        target = client.post("/accounts/oidc/start/").url
        assert parse_qs(urlsplit(target).query)["redirect_uri"] == [configured.NETKINGDOM_CALLBACK]
        configured.NETKINGDOM_CALLBACK = (
            "https://vergabe.example.test/other/accounts/oidc/callback/"
        )
        assert client.post("/accounts/oidc/start/").status_code == 503
    finally:
        set_script_prefix(old)


def test_disabled_oidc_preserves_existing_pilot_login(settings, client):
    settings.NETKINGDOM_ENABLED = False
    assert b"Benutzername" in client.get("/accounts/login/").content
    assert client.post("/accounts/oidc/start/").status_code == 404
    assert client.get("/accounts/oidc/callback/").status_code == 404


def test_wrong_signature_denied(client, signed_flow):
    client.post("/accounts/oidc/start/")
    another = rsa.generate_private_key(public_exponent=65537, key_size=2048)
    token = jwt.encode({"sub": "forged"}, another, algorithm="RS256")
    assert signed_flow(client, token_override=token).status_code == 403
    assert not Mitarbeiter.objects.exists()