Add company welcome and verified NetKingdom sign-in

Assistant: codex
Assistant-Model: gpt-6-astra
Assistant-Session: 01a092fe-13b1-7f12-ac74-7d258af4d79c

vergabe_teilnahme/apps/accounts/test_oidc.py

@@ -0,0 +1,265 @@
+"""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()