test: add pytest suite covering auth, invites, admin and OIDC flows (59 tests)

2026-06-03 22:10:40 +02:00
parent b61aa39b3b
commit 1d3c25f855
6 changed files with 934 additions and 2 deletions
@@ -0,0 +1,64 @@
+"""Shared fixtures for bincio-auth tests."""
+
+from __future__ import annotations
+
+import pytest
+from fastapi.testclient import TestClient
+
+from bincio.auth import deps
+from bincio.auth.db import create_invite, create_user, open_db
+from bincio.auth.server import app
+
+
+@pytest.fixture(autouse=True)
+def _reset_deps(tmp_path):
+    """Isolate each test: fresh DB, clean module state."""
+    deps.data_dir = tmp_path
+    deps.jwt_secret = "test-secret-32-bytes-long-enough!"
+    deps._db = None
+    deps.oidc_private_key_pem = ""
+    deps.oidc_issuer = ""
+    deps._login_attempts.clear()
+    deps._register_attempts.clear()
+    yield
+    if deps._db is not None:
+        deps._db.close()
+    deps._db = None
+
+
+@pytest.fixture()
+def db():
+    return deps._get_db()
+
+
+@pytest.fixture()
+def client():
+    return TestClient(app, raise_server_exceptions=True)
+
+
+@pytest.fixture()
+def admin(db):
+    return create_user(db, "admin", "Admin", "adminpass1", is_admin=True)
+
+
+@pytest.fixture()
+def user(db):
+    return create_user(db, "alice", "Alice", "alicepass1")
+
+
+@pytest.fixture()
+def invite(db, admin):
+    return create_invite(db, admin.handle)
+
+
+def login(client: TestClient, handle: str, password: str) -> str:
+    """Helper: POST /api/auth/login and return the session cookie value."""
+    r = client.post("/api/auth/login", json={"handle": handle, "password": password})
+    assert r.status_code == 200
+    return r.cookies["bincio_session"]
+
+
+def auth_cookies(handle: str, password: str, client: TestClient) -> dict:
+    """Return a cookies dict suitable for authenticated requests."""
+    token = login(client, handle, password)
+    return {"bincio_session": token}
@@ -0,0 +1,96 @@
+"""Tests for admin user-management endpoints."""
+
+from __future__ import annotations
+
+from .conftest import auth_cookies
+
+
+def test_list_users_admin(client, admin, user):
+    cookies = auth_cookies("admin", "adminpass1", client)
+    r = client.get("/api/admin/users", cookies=cookies)
+    assert r.status_code == 200
+    handles = [u["handle"] for u in r.json()]
+    assert "admin" in handles
+    assert "alice" in handles
+
+
+def test_list_users_non_admin(client, user):
+    cookies = auth_cookies("alice", "alicepass1", client)
+    r = client.get("/api/admin/users", cookies=cookies)
+    assert r.status_code == 403
+
+
+def test_list_users_unauthenticated(client):
+    r = client.get("/api/admin/users")
+    assert r.status_code == 401
+
+
+def test_suspend_and_unsuspend(client, admin, user):
+    cookies = auth_cookies("admin", "adminpass1", client)
+
+    r = client.post("/api/admin/users/alice/suspend", cookies=cookies)
+    assert r.status_code == 200
+    assert r.json()["status"] == "suspended"
+
+    # Suspended user can't log in
+    r2 = client.post("/api/auth/login", json={"handle": "alice", "password": "alicepass1"})
+    assert r2.status_code == 401
+
+    r3 = client.post("/api/admin/users/alice/unsuspend", cookies=cookies)
+    assert r3.status_code == 200
+
+    r4 = client.post("/api/auth/login", json={"handle": "alice", "password": "alicepass1"})
+    assert r4.status_code == 200
+
+
+def test_suspend_self(client, admin):
+    cookies = auth_cookies("admin", "adminpass1", client)
+    r = client.post("/api/admin/users/admin/suspend", cookies=cookies)
+    assert r.status_code == 400
+
+
+def test_delete_user(client, admin, user):
+    cookies = auth_cookies("admin", "adminpass1", client)
+    r = client.delete("/api/admin/users/alice", cookies=cookies)
+    assert r.status_code == 200
+
+    users = client.get("/api/admin/users", cookies=cookies).json()
+    assert not any(u["handle"] == "alice" for u in users)
+
+
+def test_delete_self(client, admin):
+    cookies = auth_cookies("admin", "adminpass1", client)
+    r = client.delete("/api/admin/users/admin", cookies=cookies)
+    assert r.status_code == 400
+
+
+def test_delete_nonexistent_user(client, admin):
+    cookies = auth_cookies("admin", "adminpass1", client)
+    r = client.delete("/api/admin/users/ghost", cookies=cookies)
+    assert r.status_code == 404
+
+
+def test_set_access_flags(client, admin, user):
+    cookies = auth_cookies("admin", "adminpass1", client)
+
+    r = client.patch("/api/admin/users/alice/access",
+                     json={"activity_access": True, "wiki_access": False},
+                     cookies=cookies)
+    assert r.status_code == 200
+
+    users = client.get("/api/admin/users", cookies=cookies).json()
+    alice = next(u for u in users if u["handle"] == "alice")
+    assert alice["activity_access"] is True
+    assert alice["wiki_access"] is False
+
+
+def test_set_access_non_admin(client, user):
+    cookies = auth_cookies("alice", "alicepass1", client)
+    r = client.patch("/api/admin/users/alice/access", json={"wiki_access": False}, cookies=cookies)
+    assert r.status_code == 403
+
+
+def test_reset_password_code_for_unknown_user(client, admin):
+    cookies = auth_cookies("admin", "adminpass1", client)
+    r = client.post("/api/admin/users/ghost/reset-password-code", cookies=cookies)
+    assert r.status_code == 404
@@ -0,0 +1,235 @@
+"""Tests for authentication, registration and password reset endpoints."""
+
+from __future__ import annotations
+
+import time
+
+from bincio.auth import deps
+from bincio.auth.db import (
+    change_password,
+    create_email_reset_token,
+    create_invite,
+    create_reset_code,
+    set_suspended,
+    set_user_email,
+)
+
+from .conftest import auth_cookies, login
+
+
+# ── Login ─────────────────────────────────────────────────────────────────────
+
+def test_login_success(client, admin):
+    r = client.post("/api/auth/login", json={"handle": "admin", "password": "adminpass1"})
+    assert r.status_code == 200
+    data = r.json()
+    assert data["ok"] is True
+    assert data["handle"] == "admin"
+    assert data["is_admin"] is True
+    assert "bincio_session" in r.cookies
+
+
+def test_login_wrong_password(client, admin):
+    r = client.post("/api/auth/login", json={"handle": "admin", "password": "wrong"})
+    assert r.status_code == 401
+
+
+def test_login_unknown_handle(client):
+    r = client.post("/api/auth/login", json={"handle": "nobody", "password": "x"})
+    assert r.status_code == 401
+
+
+def test_login_suspended(client, db, admin):
+    set_suspended(db, "admin", True)
+    r = client.post("/api/auth/login", json={"handle": "admin", "password": "adminpass1"})
+    assert r.status_code == 401
+
+
+def test_login_handle_case_insensitive(client, admin):
+    r = client.post("/api/auth/login", json={"handle": "ADMIN", "password": "adminpass1"})
+    assert r.status_code == 200
+
+
+# ── Logout ────────────────────────────────────────────────────────────────────
+
+def test_logout_clears_cookie(client, admin):
+    cookies = auth_cookies("admin", "adminpass1", client)
+    r = client.post("/api/auth/logout", cookies=cookies)
+    assert r.status_code == 200
+    assert r.cookies.get("bincio_session") == "" or "bincio_session" not in r.cookies
+
+
+# ── /me ───────────────────────────────────────────────────────────────────────
+
+def test_me_authenticated(client, admin):
+    cookies = auth_cookies("admin", "adminpass1", client)
+    r = client.get("/api/me", cookies=cookies)
+    assert r.status_code == 200
+    assert r.json()["handle"] == "admin"
+
+
+def test_me_unauthenticated(client):
+    r = client.get("/api/me")
+    assert r.status_code == 401
+
+
+def test_me_suspended_after_login(client, db, admin):
+    cookies = auth_cookies("admin", "adminpass1", client)
+    set_suspended(db, "admin", True)
+    r = client.get("/api/me", cookies=cookies)
+    assert r.status_code == 401
+
+
+# ── Registration ──────────────────────────────────────────────────────────────
+
+def test_register_success(client, invite):
+    r = client.post("/api/register", json={
+        "code": invite,
+        "handle": "newuser",
+        "password": "newpass99",
+        "display_name": "New User",
+    })
+    assert r.status_code == 200
+    assert r.json()["handle"] == "newuser"
+    assert "bincio_session" in r.cookies
+
+
+def test_register_invalid_invite(client):
+    r = client.post("/api/register", json={
+        "code": "BADCODE",
+        "handle": "newuser",
+        "password": "newpass99",
+    })
+    assert r.status_code == 400
+
+
+def test_register_used_invite(client, db, admin, invite):
+    # Use the invite first
+    client.post("/api/register", json={"code": invite, "handle": "first", "password": "firstpass1"})
+    r = client.post("/api/register", json={"code": invite, "handle": "second", "password": "secondpass1"})
+    assert r.status_code == 400
+
+
+def test_register_duplicate_handle(client, db, admin, invite):
+    client.post("/api/register", json={"code": invite, "handle": "alice", "password": "alicepass1"})
+    invite2 = create_invite(db, admin.handle)
+    r = client.post("/api/register", json={"code": invite2, "handle": "alice", "password": "alicepass1"})
+    assert r.status_code == 409
+
+
+def test_register_invalid_handle(client, invite):
+    # Handles must start with a letter/digit; underscore prefix is invalid
+    r = client.post("/api/register", json={
+        "code": invite,
+        "handle": "_invalid",
+        "password": "password99",
+    })
+    assert r.status_code == 400
+
+
+def test_register_short_password(client, invite):
+    r = client.post("/api/register", json={
+        "code": invite,
+        "handle": "valid",
+        "password": "short",
+    })
+    assert r.status_code == 400
+
+
+def test_register_grants_activity(client, db, admin):
+    code = create_invite(db, admin.handle, grants_activity=True)
+    r = client.post("/api/register", json={"code": code, "handle": "actuser", "password": "actpass99"})
+    assert r.status_code == 200
+    cookies = {"bincio_session": r.cookies["bincio_session"]}
+    me = client.get("/api/me", cookies=cookies).json()
+    assert me["activity_access"] is True
+
+
+# ── Admin-issued password reset ───────────────────────────────────────────────
+
+def test_reset_password_admin_code(client, db, admin, user):
+    cookies = auth_cookies("admin", "adminpass1", client)
+    r = client.post(f"/api/admin/users/alice/reset-password-code", cookies=cookies)
+    assert r.status_code == 200
+    code = r.json()["code"]
+
+    r2 = client.post("/api/auth/reset-password", json={
+        "handle": "alice",
+        "code": code,
+        "password": "newpassword1",
+    })
+    assert r2.status_code == 200
+
+    # Old password should no longer work
+    r3 = client.post("/api/auth/login", json={"handle": "alice", "password": "alicepass1"})
+    assert r3.status_code == 401
+
+    # New password works
+    r4 = client.post("/api/auth/login", json={"handle": "alice", "password": "newpassword1"})
+    assert r4.status_code == 200
+
+
+def test_reset_password_wrong_handle(client, db, admin, user):
+    cookies = auth_cookies("admin", "adminpass1", client)
+    r = client.post("/api/admin/users/alice/reset-password-code", cookies=cookies)
+    code = r.json()["code"]
+    r2 = client.post("/api/auth/reset-password", json={
+        "handle": "admin",  # wrong handle for this code
+        "code": code,
+        "password": "newpassword1",
+    })
+    assert r2.status_code == 400
+
+
+def test_reset_password_reuse_code(client, db, admin, user):
+    cookies = auth_cookies("admin", "adminpass1", client)
+    code = client.post("/api/admin/users/alice/reset-password-code", cookies=cookies).json()["code"]
+    client.post("/api/auth/reset-password", json={"handle": "alice", "code": code, "password": "newpass111"})
+    r = client.post("/api/auth/reset-password", json={"handle": "alice", "code": code, "password": "anotherpass1"})
+    assert r.status_code == 400
+
+
+# ── Self-service email password reset ────────────────────────────────────────
+
+def test_request_reset_always_200(client, db, user):
+    # Should never leak whether email is registered
+    r = client.post("/api/auth/request-reset", json={"email": "notregistered@example.com"})
+    assert r.status_code == 200
+
+    set_user_email(db, "alice", "alice@example.com")
+    r2 = client.post("/api/auth/request-reset", json={"email": "alice@example.com"})
+    assert r2.status_code == 200
+
+
+def test_reset_via_email_token(client, db, user):
+    set_user_email(db, "alice", "alice@example.com")
+    token = create_email_reset_token(db, "alice")
+
+    r = client.post("/api/auth/reset-password-token", json={"token": token, "password": "brandnewpass1"})
+    assert r.status_code == 200
+
+    r2 = client.post("/api/auth/login", json={"handle": "alice", "password": "brandnewpass1"})
+    assert r2.status_code == 200
+
+
+def test_reset_via_email_token_reuse(client, db, user):
+    set_user_email(db, "alice", "alice@example.com")
+    token = create_email_reset_token(db, "alice")
+    client.post("/api/auth/reset-password-token", json={"token": token, "password": "brandnewpass1"})
+    r = client.post("/api/auth/reset-password-token", json={"token": token, "password": "anotherpass1"})
+    assert r.status_code == 400
+
+
+def test_reset_via_email_token_invalid(client, db, user):
+    r = client.post("/api/auth/reset-password-token", json={"token": "bogus", "password": "somepass123"})
+    assert r.status_code == 400
+
+
+# ── Email preference ──────────────────────────────────────────────────────────
+
+def test_set_and_get_email(client, user):
+    cookies = auth_cookies("alice", "alicepass1", client)
+    client.post("/api/me/email", json={"email": "alice@example.com"}, cookies=cookies)
+    r = client.get("/api/me/email", cookies=cookies)
+    assert r.status_code == 200
+    assert r.json()["email"] == "alice@example.com"
@@ -0,0 +1,67 @@
+"""Tests for invite management endpoints."""
+
+from __future__ import annotations
+
+from bincio.auth.db import create_invite
+
+from .conftest import auth_cookies
+
+
+def test_list_invites_empty(client, user):
+    cookies = auth_cookies("alice", "alicepass1", client)
+    r = client.get("/api/invites", cookies=cookies)
+    assert r.status_code == 200
+    assert r.json() == []
+
+
+def test_create_and_list_invite(client, user):
+    cookies = auth_cookies("alice", "alicepass1", client)
+    r = client.post("/api/invites", json={}, cookies=cookies)
+    assert r.status_code == 200
+    code = r.json()["code"]
+    assert len(code) > 0
+
+    r2 = client.get("/api/invites", cookies=cookies)
+    codes = [i["code"] for i in r2.json()]
+    assert code in codes
+
+
+def test_invite_limit_regular_user(client, user):
+    cookies = auth_cookies("alice", "alicepass1", client)
+    # Regular users capped at 3
+    for _ in range(3):
+        r = client.post("/api/invites", json={}, cookies=cookies)
+        assert r.status_code == 200
+    r = client.post("/api/invites", json={}, cookies=cookies)
+    assert r.status_code == 400
+
+
+def test_admin_has_no_invite_limit(client, admin):
+    cookies = auth_cookies("admin", "adminpass1", client)
+    for _ in range(5):
+        r = client.post("/api/invites", json={}, cookies=cookies)
+        assert r.status_code == 200
+
+
+def test_create_invite_unauthenticated(client):
+    r = client.post("/api/invites", json={})
+    assert r.status_code == 401
+
+
+def test_list_invites_unauthenticated(client):
+    r = client.get("/api/invites")
+    assert r.status_code == 401
+
+
+def test_invite_grants_activity_flag(client, admin):
+    cookies = auth_cookies("admin", "adminpass1", client)
+    r = client.post("/api/invites", json={"grants_activity": True}, cookies=cookies)
+    assert r.status_code == 200
+    assert r.json()["grants_activity"] is True
+
+
+def test_regular_user_cannot_grant_activity_they_dont_have(client, user):
+    # alice has no activity_access
+    cookies = auth_cookies("alice", "alicepass1", client)
+    r = client.post("/api/invites", json={"grants_activity": True}, cookies=cookies)
+    assert r.status_code == 400
@@ -0,0 +1,344 @@
+"""Tests for the OIDC / OAuth2 authorization code flow."""
+
+from __future__ import annotations
+
+import base64
+import hashlib
+import secrets
+from urllib.parse import parse_qs, urlparse
+
+import jwt
+import pytest
+from cryptography.hazmat.primitives import serialization
+from cryptography.hazmat.primitives.asymmetric import rsa
+
+from bincio.auth import deps
+from bincio.auth.db import upsert_oauth2_client
+
+from .conftest import auth_cookies, login
+
+
+# ── Fixtures ──────────────────────────────────────────────────────────────────
+
+@pytest.fixture(autouse=True)
+def oidc_config(tmp_path):
+    """Enable OIDC mode for all tests in this module."""
+    private_key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
+    pem = private_key.private_bytes(
+        encoding=serialization.Encoding.PEM,
+        format=serialization.PrivateFormat.TraditionalOpenSSL,
+        encryption_algorithm=serialization.NoEncryption(),
+    ).decode()
+    deps.oidc_private_key_pem = pem
+    deps.oidc_issuer = "https://auth.example.test"
+
+
+@pytest.fixture()
+def confidential_client(db):
+    upsert_oauth2_client(
+        db,
+        client_id="test-app",
+        client_secret="test-secret",
+        name="Test App",
+        redirect_uris=["https://app.example.test/callback"],
+    )
+    return {"client_id": "test-app", "client_secret": "test-secret",
+            "redirect_uri": "https://app.example.test/callback"}
+
+
+@pytest.fixture()
+def public_client(db):
+    upsert_oauth2_client(
+        db,
+        client_id="mobile-app",
+        client_secret=None,
+        name="Mobile App",
+        redirect_uris=["myapp://callback"],
+    )
+    return {"client_id": "mobile-app", "redirect_uri": "myapp://callback"}
+
+
+# ── Discovery ─────────────────────────────────────────────────────────────────
+
+def test_openid_configuration(client):
+    r = client.get("/.well-known/openid-configuration")
+    assert r.status_code == 200
+    data = r.json()
+    assert data["issuer"] == "https://auth.example.test"
+    assert "authorization_endpoint" in data
+    assert "token_endpoint" in data
+    assert "jwks_uri" in data
+
+
+def test_jwks(client):
+    r = client.get("/.well-known/jwks.json")
+    assert r.status_code == 200
+    keys = r.json()["keys"]
+    assert len(keys) == 1
+    assert keys[0]["kty"] == "RSA"
+    assert keys[0]["alg"] == "RS256"
+
+
+def test_discovery_disabled_without_oidc(client):
+    deps.oidc_issuer = ""
+    r = client.get("/.well-known/openid-configuration")
+    assert r.status_code == 503
+
+
+# ── Authorization endpoint ────────────────────────────────────────────────────
+
+def test_authorize_redirects_to_login_when_unauthenticated(client, confidential_client):
+    r = client.get(
+        "/oauth2/authorize",
+        params={
+            "client_id": confidential_client["client_id"],
+            "redirect_uri": confidential_client["redirect_uri"],
+            "response_type": "code",
+            "scope": "openid profile",
+        },
+        follow_redirects=False,
+    )
+    assert r.status_code == 302
+    assert "/login/" in r.headers["location"]
+
+
+def test_authorize_issues_code_when_authenticated(client, admin, confidential_client):
+    cookies = auth_cookies("admin", "adminpass1", client)
+    r = client.get(
+        "/oauth2/authorize",
+        params={
+            "client_id": confidential_client["client_id"],
+            "redirect_uri": confidential_client["redirect_uri"],
+            "response_type": "code",
+            "scope": "openid profile",
+            "state": "xyz",
+        },
+        cookies=cookies,
+        follow_redirects=False,
+    )
+    assert r.status_code == 302
+    location = r.headers["location"]
+    qs = parse_qs(urlparse(location).query)
+    assert "code" in qs
+    assert qs["state"][0] == "xyz"
+
+
+def test_authorize_unknown_client(client, admin):
+    cookies = auth_cookies("admin", "adminpass1", client)
+    r = client.get(
+        "/oauth2/authorize",
+        params={
+            "client_id": "unknown",
+            "redirect_uri": "https://somewhere.test/cb",
+            "response_type": "code",
+        },
+        cookies=cookies,
+        follow_redirects=False,
+    )
+    assert r.status_code == 400
+
+
+def test_authorize_unregistered_redirect_uri(client, admin, confidential_client):
+    cookies = auth_cookies("admin", "adminpass1", client)
+    r = client.get(
+        "/oauth2/authorize",
+        params={
+            "client_id": confidential_client["client_id"],
+            "redirect_uri": "https://evil.example.test/steal",
+            "response_type": "code",
+        },
+        cookies=cookies,
+        follow_redirects=False,
+    )
+    assert r.status_code == 400
+
+
+# ── Full authorization code flow (confidential client) ────────────────────────
+
+def _get_auth_code(client, cookies, confidential_client, **extra_params) -> str:
+    r = client.get(
+        "/oauth2/authorize",
+        params={
+            "client_id": confidential_client["client_id"],
+            "redirect_uri": confidential_client["redirect_uri"],
+            "response_type": "code",
+            "scope": "openid profile",
+            **extra_params,
+        },
+        cookies=cookies,
+        follow_redirects=False,
+    )
+    assert r.status_code == 302
+    qs = parse_qs(urlparse(r.headers["location"]).query)
+    return qs["code"][0]
+
+
+def test_token_exchange_confidential_client(client, admin, confidential_client):
+    cookies = auth_cookies("admin", "adminpass1", client)
+    code = _get_auth_code(client, cookies, confidential_client)
+
+    r = client.post("/oauth2/token", data={
+        "grant_type": "authorization_code",
+        "code": code,
+        "redirect_uri": confidential_client["redirect_uri"],
+        "client_id": confidential_client["client_id"],
+        "client_secret": confidential_client["client_secret"],
+    })
+    assert r.status_code == 200
+    data = r.json()
+    assert "id_token" in data
+    assert data["token_type"] == "Bearer"
+
+
+def test_token_exchange_wrong_secret(client, admin, confidential_client):
+    cookies = auth_cookies("admin", "adminpass1", client)
+    code = _get_auth_code(client, cookies, confidential_client)
+
+    r = client.post("/oauth2/token", data={
+        "grant_type": "authorization_code",
+        "code": code,
+        "redirect_uri": confidential_client["redirect_uri"],
+        "client_id": confidential_client["client_id"],
+        "client_secret": "wrong-secret",
+    })
+    assert r.status_code == 401
+
+
+def test_token_exchange_code_reuse(client, admin, confidential_client):
+    cookies = auth_cookies("admin", "adminpass1", client)
+    code = _get_auth_code(client, cookies, confidential_client)
+
+    payload = {
+        "grant_type": "authorization_code",
+        "code": code,
+        "redirect_uri": confidential_client["redirect_uri"],
+        "client_id": confidential_client["client_id"],
+        "client_secret": confidential_client["client_secret"],
+    }
+    client.post("/oauth2/token", data=payload)
+    r = client.post("/oauth2/token", data=payload)
+    assert r.status_code == 400
+
+
+def test_id_token_claims(client, admin, confidential_client):
+    cookies = auth_cookies("admin", "adminpass1", client)
+    code = _get_auth_code(client, cookies, confidential_client)
+
+    r = client.post("/oauth2/token", data={
+        "grant_type": "authorization_code",
+        "code": code,
+        "redirect_uri": confidential_client["redirect_uri"],
+        "client_id": confidential_client["client_id"],
+        "client_secret": confidential_client["client_secret"],
+    })
+    id_token = r.json()["id_token"]
+    # Decode without verifying signature to inspect claims
+    payload = jwt.decode(id_token, options={"verify_signature": False})
+    assert payload["sub"] == "admin"
+    assert payload["iss"] == "https://auth.example.test"
+    assert payload["aud"] == "test-app"
+    assert payload["is_admin"] is True
+
+
+# ── PKCE flow (public client) ─────────────────────────────────────────────────
+
+def _pkce_pair() -> tuple[str, str]:
+    verifier = secrets.token_urlsafe(48)
+    challenge = base64.urlsafe_b64encode(
+        hashlib.sha256(verifier.encode()).digest()
+    ).rstrip(b"=").decode()
+    return verifier, challenge
+
+
+def test_pkce_flow(client, admin, public_client):
+    cookies = auth_cookies("admin", "adminpass1", client)
+    verifier, challenge = _pkce_pair()
+
+    # Authorize with PKCE challenge
+    r = client.get(
+        "/oauth2/authorize",
+        params={
+            "client_id": public_client["client_id"],
+            "redirect_uri": public_client["redirect_uri"],
+            "response_type": "code",
+            "scope": "openid profile",
+            "code_challenge": challenge,
+            "code_challenge_method": "S256",
+        },
+        cookies=cookies,
+        follow_redirects=False,
+    )
+    assert r.status_code == 302
+    code = parse_qs(urlparse(r.headers["location"]).query)["code"][0]
+
+    # Exchange with verifier
+    r2 = client.post("/oauth2/token", data={
+        "grant_type": "authorization_code",
+        "code": code,
+        "redirect_uri": public_client["redirect_uri"],
+        "client_id": public_client["client_id"],
+        "code_verifier": verifier,
+    })
+    assert r2.status_code == 200
+    assert "id_token" in r2.json()
+
+
+def test_pkce_wrong_verifier(client, admin, public_client):
+    cookies = auth_cookies("admin", "adminpass1", client)
+    verifier, challenge = _pkce_pair()
+
+    r = client.get(
+        "/oauth2/authorize",
+        params={
+            "client_id": public_client["client_id"],
+            "redirect_uri": public_client["redirect_uri"],
+            "response_type": "code",
+            "scope": "openid profile",
+            "code_challenge": challenge,
+            "code_challenge_method": "S256",
+        },
+        cookies=cookies,
+        follow_redirects=False,
+    )
+    code = parse_qs(urlparse(r.headers["location"]).query)["code"][0]
+
+    r2 = client.post("/oauth2/token", data={
+        "grant_type": "authorization_code",
+        "code": code,
+        "redirect_uri": public_client["redirect_uri"],
+        "client_id": public_client["client_id"],
+        "code_verifier": "wrong-verifier",
+    })
+    assert r2.status_code == 400
+
+
+# ── Userinfo endpoint ─────────────────────────────────────────────────────────
+
+def test_userinfo(client, admin, confidential_client):
+    cookies = auth_cookies("admin", "adminpass1", client)
+    code = _get_auth_code(client, cookies, confidential_client)
+
+    token_resp = client.post("/oauth2/token", data={
+        "grant_type": "authorization_code",
+        "code": code,
+        "redirect_uri": confidential_client["redirect_uri"],
+        "client_id": confidential_client["client_id"],
+        "client_secret": confidential_client["client_secret"],
+    })
+    access_token = token_resp.json()["access_token"]
+
+    r = client.get("/oauth2/userinfo", headers={"Authorization": f"Bearer {access_token}"})
+    assert r.status_code == 200
+    data = r.json()
+    assert data["sub"] == "admin"
+    assert data["preferred_username"] == "admin"
+
+
+def test_userinfo_invalid_token(client):
+    r = client.get("/oauth2/userinfo", headers={"Authorization": "Bearer bogus"})
+    assert r.status_code == 401
+
+
+def test_userinfo_no_token(client):
+    r = client.get("/oauth2/userinfo")
+    assert r.status_code == 401