feat(auth): OpenID-Connect-SSO gegen Authentik (Backend + Web)

Zusaetzlicher Anmeldeweg per OIDC neben dem bestehenden Passwort-Login. Ohne gesetztes OIDC_PROVIDERS aendert sich nichts: /api/auth/oidc/providers meldet enabled:false und der Login-Screen bleibt unveraendert. Backend - oidc_config.py: Provider-Registry rein aus ENV. Ein zweiter Provider ist reine Konfiguration (OIDC_PROVIDERS + OIDC_<KEY>_* Block), keine Codeaenderung. - oidc_client.py: Discovery, JWKS, Token-Tausch, ID-Token-Pruefung ueber authlib.jose mit fest gepinnter Signatur-Allowlist (nur asymmetrisch, damit alg:none und Algorithmus-Confusion ausgeschlossen sind). - oidc_identity.py: Zuordnung ueber (provider_key, sub). Ein E-Mail-Treffer auf ein bestehendes Konto wird standardmaessig abgelehnt statt verknuepft — sonst erbt jeder, der im IdP eine beliebige E-Mail setzen kann, das passende Konto. Verknuepfen ist eine explizite, authentifizierte Aktion. Auto-Provisioning nur per ENV-Flag, standardmaessig aus, optional mit Domain-Allowlist. - routers/oidc_router.py: /providers, /start, /callback, /complete, /exchange, /link, /identities. PKCE S256; Flow-State als kurzlebiges signiertes JWT in einem HttpOnly-Cookie mit jti==state statt eines prozesslokalen Dicts, damit der Flow Neustarts und mehrere Worker uebersteht. Das Calendarr-JWT wird ueber ein Einmal-Cookie uebergeben, nie ueber URL oder Fragment. - /exchange als Kontrakt fuer die AppAuth-Clients: die App fuehrt den Flow als Public Client selbst aus und tauscht das ID-Token; kein Secret im Request. - models.py: neue Tabelle oidc_identities, users.auth_source; Migration in main.py im dortigen try/except-Stil. Nebenbefund, mitbehoben: auth.verify_password liess ValueError aus bcrypt.checkpw durch. Ein nicht wohlgeformter password_hash wurde damit zu HTTP 500 auf /api/auth/login und bei jedem CalDAV-Basic-Auth. SSO-Nutzer bekommen einen echten bcrypt-Hash ueber ein zufaelliges, nirgends gespeichertes Geheimnis; fuer CalDAV nutzen sie App-Passwoerter wie 2FA-Nutzer. Web - SSO-Block auf dem Login-Screen, aus /providers befuellt; Buttons sind Links, weil /start eine Top-Level-Navigation braucht. - consumeSsoHandoff() laeuft vor den Setup-/Token-Pruefungen in boot() und raeumt die Query per replaceState ab, bevor es awaited. - 401-Auto-Logout um die SSO-Pfade erweitert, sonst dreht ein fehlgeschlagener Handoff eine Reload-Schleife. - i18n-Keys in de + en (Fallback-Kette deckt die uebrigen acht Sprachen). - Service Worker laesst /api/auth/oidc/ unangetastet durch. 37 neue Tests (Gesamtsuite 60 gruen): Signaturpruefung mit echten RSA-Tokens, alg:none, falscher iss/aud/nonce, unbekannte client_id, Provisioning-Policies, Username-Kollision, Einmaligkeit des Handoff-Cookies, unerreichbarer Provider. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
2026-09-09 18:00:46 +02:00
parent a30375aa63
commit 31baf4b18c
17 changed files with 2008 additions and 5 deletions
@@ -0,0 +1,584 @@
+"""OpenID Connect SSO — protocol validation, account mapping and provisioning.
+
+ID tokens here are signed with a real RSA key and go through the real
+validation path; only the network (discovery, JWKS, token endpoint) is stubbed,
+at the single seam ``oidc_client._http_get``.
+"""
+
+import time
+
+import pytest
+from authlib.jose import JsonWebKey, jwt as authlib_jwt
+
+import models
+import oidc_client
+import oidc_config
+from conftest import auth, create_user, register_admin
+from database import SessionLocal
+
+ISSUER = "https://authentik.example.com/application/o/calendarr/"
+WEB_CLIENT = "calendarr-web"
+MOBILE_CLIENT = "calendarr-mobile"
+
+# One key pair for the whole module — generating RSA keys is slow.
+_KEY = JsonWebKey.generate_key("RSA", 2048, is_private=True)
+_KID = "test-key-1"
+
+
+def _jwks() -> dict:
+    pub = _KEY.as_dict(is_private=False)
+    pub["kid"] = _KID
+    return {"keys": [pub]}
+
+
+def _discovery() -> dict:
+    return {
+        "issuer": ISSUER,
+        "authorization_endpoint": ISSUER + "authorize/",
+        "token_endpoint": ISSUER + "token/",
+        "userinfo_endpoint": ISSUER + "userinfo/",
+        "jwks_uri": ISSUER + "jwks/",
+        "end_session_endpoint": ISSUER + "end-session/",
+    }
+
+
+def make_id_token(*, sub="ak-sub-1", aud=WEB_CLIENT, iss=ISSUER, nonce=None,
+                  email="user@example.com", email_verified=True,
+                  preferred_username="user", name="Test User",
+                  exp_delta=300, alg="RS256", key=None, extra=None):
+    now = int(time.time())
+    payload = {
+        "iss": iss, "sub": sub, "aud": aud,
+        "iat": now, "exp": now + exp_delta,
+    }
+    if nonce is not None:
+        payload["nonce"] = nonce
+    if email is not None:
+        payload["email"] = email
+        payload["email_verified"] = email_verified
+    if preferred_username is not None:
+        payload["preferred_username"] = preferred_username
+    if name is not None:
+        payload["name"] = name
+    if extra:
+        payload.update(extra)
+    header = {"alg": alg, "kid": _KID}
+    return authlib_jwt.encode(header, payload, key or _KEY).decode("ascii")
+
+
+@pytest.fixture
+def stub_network(monkeypatch):
+    """Serve discovery and JWKS locally; fail loudly on any other URL."""
+    def fake_get(url, timeout=10):
+        if url.endswith("/.well-known/openid-configuration"):
+            return _discovery()
+        if url == _discovery()["jwks_uri"]:
+            return _jwks()
+        raise AssertionError(f"unexpected network call: {url}")
+    monkeypatch.setattr(oidc_client, "_http_get", fake_get)
+
+
+@pytest.fixture
+def authentik_env(monkeypatch):
+    """Provider configuration only — the network is left alone."""
+    monkeypatch.setenv("OIDC_PROVIDERS", "authentik")
+    monkeypatch.setenv("OIDC_AUTHENTIK_NAME", "Authentik")
+    monkeypatch.setenv("OIDC_AUTHENTIK_ISSUER", ISSUER)
+    monkeypatch.setenv("OIDC_AUTHENTIK_CLIENT_ID", WEB_CLIENT)
+    monkeypatch.setenv("OIDC_AUTHENTIK_MOBILE_CLIENT_ID", MOBILE_CLIENT)
+    oidc_config.reset_cache()
+    oidc_client.reset_cache()
+    yield
+    oidc_config.reset_cache()
+    oidc_client.reset_cache()
+
+
+@pytest.fixture
+def authentik(authentik_env, stub_network):
+    """Configured provider with a working, stubbed IdP."""
+    yield
+
+
+def exchange(client, **over):
+    body = {"provider": "authentik", "client_id": MOBILE_CLIENT,
+            "id_token": make_id_token(aud=MOBILE_CLIENT)}
+    body.update(over)
+    return client.post("/api/auth/oidc/exchange", json=body)
+
+
+def db_session():
+    return SessionLocal()
+
+
+# ── unconfigured: SSO must be completely invisible ───────────────────────────
+
+def test_disabled_when_unconfigured(client):
+    oidc_config.reset_cache()
+    r = client.get("/api/auth/oidc/providers")
+    assert r.status_code == 200
+    assert r.json() == {"enabled": False, "providers": []}
+    assert client.get("/api/auth/oidc/authentik/start",
+                      follow_redirects=False).status_code == 404
+
+
+def test_password_login_still_works_with_sso_enabled(client, authentik):
+    register_admin(client, "admin", "pw")
+    r = client.post("/api/auth/login", json={"username": "admin", "password": "pw"})
+    assert r.status_code == 200
+    assert r.json()["user"]["username"] == "admin"
+
+
+# ── /providers ───────────────────────────────────────────────────────────────
+
+def test_providers_lists_endpoints(client, authentik):
+    body = client.get("/api/auth/oidc/providers").json()
+    assert body["enabled"] is True
+    p = body["providers"][0]
+    assert p["key"] == "authentik"
+    assert p["name"] == "Authentik"
+    assert p["mobile_client_id"] == MOBILE_CLIENT
+    assert p["authorization_endpoint"] == _discovery()["authorization_endpoint"]
+
+
+def test_providers_never_leaks_client_secret(client, authentik, monkeypatch):
+    secret = "super-secret-value-42"
+    monkeypatch.setenv("OIDC_AUTHENTIK_CLIENT_SECRET", secret)
+    oidc_config.reset_cache()
+    r = client.get("/api/auth/oidc/providers")
+    assert secret not in r.text
+
+
+def test_provider_without_client_id_is_dropped(client, monkeypatch):
+    monkeypatch.setenv("OIDC_PROVIDERS", "broken")
+    monkeypatch.setenv("OIDC_BROKEN_ISSUER", ISSUER)
+    oidc_config.reset_cache()
+    assert client.get("/api/auth/oidc/providers").json()["enabled"] is False
+    oidc_config.reset_cache()
+
+
+# ── browser flow: /start and /callback ───────────────────────────────────────
+
+def test_start_redirects_with_pkce_and_sets_flow_cookie(client, authentik):
+    register_admin(client, "admin", "pw")
+    r = client.get("/api/auth/oidc/authentik/start", follow_redirects=False)
+    assert r.status_code == 302
+    loc = r.headers["location"]
+    assert loc.startswith(_discovery()["authorization_endpoint"])
+    for expected in ("code_challenge=", "code_challenge_method=S256",
+                     "state=", "nonce=", "response_type=code"):
+        assert expected in loc
+    # The PKCE verifier must never travel to the provider.
+    assert "code_verifier" not in loc
+    cookie_header = r.headers["set-cookie"]
+    assert "clr_oidc_flow=" in cookie_header
+    assert "HttpOnly" in cookie_header
+    # Lax, not Strict: the cookie has to survive the IdP's cross-site redirect.
+    assert "SameSite=lax" in cookie_header.replace("samesite", "SameSite")
+
+
+def test_start_refused_before_initial_setup(client, authentik):
+    r = client.get("/api/auth/oidc/authentik/start", follow_redirects=False)
+    assert r.status_code == 302
+    assert "sso_error=setup_required" in r.headers["location"]
+
+
+def _break_network(monkeypatch):
+    """Fail at the `requests` layer, below _http_get, so the real error
+    translation in _http_get is what gets exercised."""
+    import requests as real_requests
+
+    def boom(*a, **k):
+        raise real_requests.ConnectionError("dns go boom")
+    monkeypatch.setattr(oidc_client.http_requests, "get", boom)
+    monkeypatch.setattr(oidc_client.http_requests, "post", boom)
+    oidc_client.reset_cache()
+
+
+def test_start_survives_unreachable_provider(client, authentik_env, monkeypatch):
+    """An unreachable IdP must be a clean error, not an HTTP 500."""
+    register_admin(client, "admin", "pw")
+    _break_network(monkeypatch)
+
+    r = client.get("/api/auth/oidc/authentik/start", follow_redirects=False)
+    assert r.status_code == 302
+    assert "sso_error=oidc_provider_unreachable" in r.headers["location"]
+
+
+def test_providers_stays_up_when_discovery_is_down(client, authentik_env, monkeypatch):
+    """The login screen must render even if the IdP is unreachable."""
+    _break_network(monkeypatch)
+    body = client.get("/api/auth/oidc/providers").json()
+    assert body["enabled"] is True
+    assert body["providers"][0]["degraded"] is True
+
+
+def test_exchange_survives_unreachable_jwks(client, authentik_env, monkeypatch):
+    """A JWKS fetch failure must be a 401, not an HTTP 500."""
+    _seed_identity(client)
+    _break_network(monkeypatch)
+    assert exchange(client).status_code == 401
+
+
+def test_callback_rejects_missing_cookie(client, authentik):
+    register_admin(client, "admin", "pw")
+    r = client.get("/api/auth/oidc/authentik/callback?code=x&state=y",
+                   follow_redirects=False)
+    assert "sso_error=state_mismatch" in r.headers["location"]
+
+
+def test_callback_rejects_state_mismatch(client, authentik, monkeypatch):
+    register_admin(client, "admin", "pw")
+    client.get("/api/auth/oidc/authentik/start", follow_redirects=False)
+    r = client.get("/api/auth/oidc/authentik/callback?code=x&state=tampered",
+                   follow_redirects=False)
+    assert "sso_error=state_mismatch" in r.headers["location"]
+    with db_session() as db:
+        assert db.query(models.User).count() == 1  # only the admin
+
+
+def test_callback_happy_path_logs_in_linked_user(client, authentik, monkeypatch):
+    admin_token = register_admin(client, "admin", "pw")
+    uid, _ = create_user(client, admin_token, "alice")
+
+    start = client.get("/api/auth/oidc/authentik/start", follow_redirects=False)
+    state = _state_from(start.headers["location"])
+
+    with db_session() as db:
+        db.add(models.OIDCIdentity(user_id=uid, provider_key="authentik",
+                                   issuer=ISSUER, subject="ak-alice"))
+        db.commit()
+
+    nonce = _nonce_from(start.headers["location"])
+    monkeypatch.setattr(oidc_client, "exchange_code", lambda *a, **k: {
+        "id_token": make_id_token(sub="ak-alice", nonce=nonce, email="alice@example.com"),
+    })
+
+    cb = client.get(f"/api/auth/oidc/authentik/callback?code=good&state={state}",
+                    follow_redirects=False)
+    assert cb.status_code == 302
+    assert cb.headers["location"] == "/?sso=1"
+    assert "clr_oidc_token=" in cb.headers["set-cookie"]
+
+    done = client.post("/api/auth/oidc/complete")
+    assert done.status_code == 200
+    assert done.json()["user"]["username"] == "alice"
+
+    # The handoff cookie is single-use.
+    assert client.post("/api/auth/oidc/complete").status_code == 401
+
+
+def _state_from(location: str) -> str:
+    from urllib.parse import parse_qs, urlparse
+    return parse_qs(urlparse(location).query)["state"][0]
+
+
+def _nonce_from(location: str) -> str:
+    from urllib.parse import parse_qs, urlparse
+    return parse_qs(urlparse(location).query)["nonce"][0]
+
+
+# ── /exchange: token validation (the mobile entry point) ─────────────────────
+
+def _seed_identity(client, sub="ak-sub-1", username="alice"):
+    admin_token = register_admin(client, "admin", "pw")
+    uid, _ = create_user(client, admin_token, username)
+    with db_session() as db:
+        db.add(models.OIDCIdentity(user_id=uid, provider_key="authentik",
+                                   issuer=ISSUER, subject=sub))
+        db.commit()
+    return uid
+
+
+def test_exchange_happy_path(client, authentik):
+    _seed_identity(client)
+    r = exchange(client)
+    assert r.status_code == 200, r.text
+    token = r.json()["access_token"]
+    me = client.get("/api/auth/me", headers=auth(token))
+    assert me.status_code == 200
+    assert me.json()["username"] == "alice"
+
+
+def test_exchange_rejects_alg_none(client, authentik):
+    """Regression guard for the pinned signature-algorithm allow-list."""
+    import base64
+    import json
+
+    def b64(raw):
+        return base64.urlsafe_b64encode(raw).rstrip(b"=").decode()
+
+    header = b64(json.dumps({"alg": "none", "kid": _KID}).encode())
+    payload = b64(json.dumps({
+        "iss": ISSUER, "sub": "ak-sub-1", "aud": MOBILE_CLIENT,
+        "iat": int(time.time()), "exp": int(time.time()) + 300,
+    }).encode())
+    _seed_identity(client)
+    r = exchange(client, id_token=f"{header}.{payload}.")
+    assert r.status_code == 401
+
+
+def test_exchange_rejects_bad_signature(client, authentik):
+    other_key = JsonWebKey.generate_key("RSA", 2048, is_private=True)
+    _seed_identity(client)
+    r = exchange(client, id_token=make_id_token(aud=MOBILE_CLIENT, key=other_key))
+    assert r.status_code == 401
+
+
+def test_exchange_rejects_wrong_issuer(client, authentik):
+    _seed_identity(client)
+    r = exchange(client, id_token=make_id_token(aud=MOBILE_CLIENT,
+                                                iss="https://evil.example.com/"))
+    assert r.status_code == 401
+
+
+def test_exchange_rejects_wrong_audience(client, authentik):
+    _seed_identity(client)
+    # A token minted for the web client replayed against the mobile client id.
+    r = exchange(client, id_token=make_id_token(aud=WEB_CLIENT))
+    assert r.status_code == 401
+
+
+def test_exchange_rejects_expired_token(client, authentik):
+    _seed_identity(client)
+    r = exchange(client, id_token=make_id_token(aud=MOBILE_CLIENT, exp_delta=-3600))
+    assert r.status_code == 401
+
+
+def test_exchange_rejects_unknown_client_id(client, authentik):
+    _seed_identity(client)
+    r = exchange(client, client_id="some-other-app",
+                 id_token=make_id_token(aud="some-other-app"))
+    assert r.status_code == 401
+
+
+def test_exchange_rejects_nonce_mismatch(client, authentik):
+    _seed_identity(client)
+    r = exchange(client, nonce="expected-nonce",
+                 id_token=make_id_token(aud=MOBILE_CLIENT, nonce="different"))
+    assert r.status_code == 401
+
+
+def test_exchange_requires_nonce_when_token_carries_one(client, authentik):
+    _seed_identity(client)
+    r = exchange(client, id_token=make_id_token(aud=MOBILE_CLIENT, nonce="from-appauth"))
+    assert r.status_code == 400
+
+
+def test_exchange_accepts_no_secret_field(client, authentik):
+    """A public client must not be able to authenticate with a secret."""
+    _seed_identity(client)
+    r = exchange(client, client_secret="anything")   # extra field is ignored
+    assert r.status_code == 200
+
+
+# ── account mapping and provisioning ─────────────────────────────────────────
+
+def test_autoprovision_disabled_by_default(client, authentik):
+    register_admin(client, "admin", "pw")
+    r = exchange(client, id_token=make_id_token(aud=MOBILE_CLIENT, sub="new-sub",
+                                                email="new@example.com"))
+    assert r.status_code == 403
+    assert r.json()["detail"] == "oidc_signup_disabled"
+    with db_session() as db:
+        assert db.query(models.User).count() == 1
+
+
+def test_autoprovision_creates_user_with_settings(client, authentik, monkeypatch):
+    monkeypatch.setenv("OIDC_AUTHENTIK_ALLOW_SIGNUP", "true")
+    oidc_config.reset_cache()
+    register_admin(client, "admin", "pw")
+
+    r = exchange(client, id_token=make_id_token(
+        aud=MOBILE_CLIENT, sub="new-sub", email="new@example.com",
+        preferred_username="Neuer Nutzer"))
+    assert r.status_code == 200, r.text
+
+    with db_session() as db:
+        user = db.query(models.User).filter(models.User.email == "new@example.com").one()
+        assert user.auth_source == "sso"
+        assert user.is_admin is False
+        assert user.username == "neuernutzer"     # normalised
+        # Without this row /api/settings would 500 for the new user.
+        assert db.query(models.UserSettings).filter(
+            models.UserSettings.user_id == user.id).count() == 1
+
+    settings = client.get("/api/settings/", headers=auth(r.json()["access_token"]))
+    assert settings.status_code == 200
+
+
+def test_domain_allowlist_blocks_foreign_domain(client, authentik, monkeypatch):
+    monkeypatch.setenv("OIDC_AUTHENTIK_ALLOW_SIGNUP", "true")
+    monkeypatch.setenv("OIDC_AUTHENTIK_ALLOWED_DOMAINS", "allowed.example")
+    oidc_config.reset_cache()
+    register_admin(client, "admin", "pw")
+
+    r = exchange(client, id_token=make_id_token(aud=MOBILE_CLIENT, sub="s2",
+                                                email="x@forbidden.example"))
+    assert r.status_code == 403
+    assert r.json()["detail"] == "oidc_domain_not_allowed"
+
+    ok = exchange(client, id_token=make_id_token(aud=MOBILE_CLIENT, sub="s3",
+                                                 email="y@allowed.example"))
+    assert ok.status_code == 200
+
+
+def test_email_match_on_existing_account_is_refused(client, authentik):
+    """The account-takeover guard: no silent linking by email address."""
+    admin_token = register_admin(client, "admin", "pw")
+    client.post("/api/users/", headers=auth(admin_token),
+                json={"username": "bob", "password": "pw", "email": "bob@example.com"})
+
+    r = exchange(client, id_token=make_id_token(aud=MOBILE_CLIENT, sub="ak-bob",
+                                                email="bob@example.com"))
+    assert r.status_code == 403
+    assert r.json()["detail"] == "oidc_account_not_linked"
+    with db_session() as db:
+        assert db.query(models.OIDCIdentity).count() == 0
+
+
+def test_link_by_email_requires_verified_email(client, authentik, monkeypatch):
+    monkeypatch.setenv("OIDC_AUTHENTIK_LINK_BY_EMAIL", "true")
+    oidc_config.reset_cache()
+    admin_token = register_admin(client, "admin", "pw")
+    client.post("/api/users/", headers=auth(admin_token),
+                json={"username": "bob", "password": "pw", "email": "bob@example.com"})
+
+    bad = exchange(client, id_token=make_id_token(
+        aud=MOBILE_CLIENT, sub="ak-bob", email="bob@example.com", email_verified=False))
+    assert bad.status_code == 403
+    assert bad.json()["detail"] == "oidc_email_not_verified"
+
+    good = exchange(client, id_token=make_id_token(
+        aud=MOBILE_CLIENT, sub="ak-bob", email="bob@example.com", email_verified=True))
+    assert good.status_code == 200
+    assert good.json()["user"]["username"] == "bob"
+
+
+def test_username_collision_gets_suffix(client, authentik, monkeypatch):
+    monkeypatch.setenv("OIDC_AUTHENTIK_ALLOW_SIGNUP", "true")
+    oidc_config.reset_cache()
+    register_admin(client, "admin", "pw")
+
+    # A different human whose preferred_username happens to be "admin".
+    r = exchange(client, id_token=make_id_token(
+        aud=MOBILE_CLIENT, sub="other-human", email="other@example.com",
+        preferred_username="admin"))
+    assert r.status_code == 200
+    assert r.json()["user"]["username"] == "admin-2"
+    with db_session() as db:
+        assert db.query(models.User).filter(models.User.username == "admin").one().is_admin
+
+
+def test_email_collision_during_signup_is_refused(client, authentik, monkeypatch):
+    monkeypatch.setenv("OIDC_AUTHENTIK_ALLOW_SIGNUP", "true")
+    oidc_config.reset_cache()
+    admin_token = register_admin(client, "admin", "pw")
+    client.post("/api/users/", headers=auth(admin_token),
+                json={"username": "carol", "password": "pw", "email": "carol@example.com"})
+
+    r = exchange(client, id_token=make_id_token(aud=MOBILE_CLIENT, sub="ak-carol",
+                                                email="carol@example.com"))
+    # Refused rather than forking one human into a second account.
+    assert r.status_code == 403
+    with db_session() as db:
+        assert db.query(models.User).filter(
+            models.User.email == "carol@example.com").count() == 1
+
+
+def test_second_login_reuses_the_same_account(client, authentik, monkeypatch):
+    monkeypatch.setenv("OIDC_AUTHENTIK_ALLOW_SIGNUP", "true")
+    oidc_config.reset_cache()
+    register_admin(client, "admin", "pw")
+
+    first = exchange(client, id_token=make_id_token(aud=MOBILE_CLIENT, sub="stable",
+                                                    email="a@example.com"))
+    # Same sub, different email and username at the provider.
+    second = exchange(client, id_token=make_id_token(
+        aud=MOBILE_CLIENT, sub="stable", email="renamed@example.com",
+        preferred_username="renamed"))
+    assert first.json()["user"]["id"] == second.json()["user"]["id"]
+    with db_session() as db:
+        assert db.query(models.OIDCIdentity).count() == 1
+
+
+# ── the password_hash trap ───────────────────────────────────────────────────
+
+def test_sso_user_password_login_fails_without_500(client, authentik, monkeypatch):
+    monkeypatch.setenv("OIDC_AUTHENTIK_ALLOW_SIGNUP", "true")
+    oidc_config.reset_cache()
+    register_admin(client, "admin", "pw")
+    exchange(client, id_token=make_id_token(aud=MOBILE_CLIENT, sub="s", email="s@example.com",
+                                            preferred_username="ssouser"))
+
+    r = client.post("/api/auth/login", json={"username": "ssouser", "password": ""})
+    assert r.status_code == 401, f"expected 401, got {r.status_code}"
+
+
+def test_verify_password_survives_malformed_hash():
+    from auth import verify_password
+    for broken in ("", "!", "not-a-bcrypt-hash"):
+        assert verify_password("anything", broken) is False
+
+
+# ── linking and unlinking from a session ─────────────────────────────────────
+
+def test_link_and_unlink_identity(client, authentik):
+    admin_token = register_admin(client, "admin", "pw")
+    _, token = create_user(client, admin_token, "dave")
+
+    r = client.post("/api/auth/oidc/authentik/link", headers=auth(token),
+                    json={"id_token": make_id_token(sub="ak-dave"),
+                          "client_id": WEB_CLIENT})
+    assert r.status_code == 200, r.text
+    identity_id = r.json()["id"]
+
+    listed = client.get("/api/auth/oidc/identities", headers=auth(token)).json()
+    assert len(listed["identities"]) == 1
+    assert listed["auth_source"] == "local"
+
+    # Now SSO login for that identity works.
+    ex = exchange(client, client_id=MOBILE_CLIENT,
+                  id_token=make_id_token(sub="ak-dave", aud=MOBILE_CLIENT))
+    assert ex.status_code == 200
+    assert ex.json()["user"]["username"] == "dave"
+
+    assert client.delete(f"/api/auth/oidc/identities/{identity_id}",
+                         headers=auth(token)).status_code == 200
+
+
+def test_cannot_link_identity_owned_by_someone_else(client, authentik):
+    admin_token = register_admin(client, "admin", "pw")
+    _, dave = create_user(client, admin_token, "dave")
+    _, eve = create_user(client, admin_token, "eve")
+
+    client.post("/api/auth/oidc/authentik/link", headers=auth(dave),
+                json={"id_token": make_id_token(sub="ak-dave"), "client_id": WEB_CLIENT})
+    r = client.post("/api/auth/oidc/authentik/link", headers=auth(eve),
+                    json={"id_token": make_id_token(sub="ak-dave"), "client_id": WEB_CLIENT})
+    assert r.status_code == 409
+
+
+def test_sso_only_user_cannot_unlink_last_identity(client, authentik, monkeypatch):
+    monkeypatch.setenv("OIDC_AUTHENTIK_ALLOW_SIGNUP", "true")
+    oidc_config.reset_cache()
+    register_admin(client, "admin", "pw")
+    r = exchange(client, id_token=make_id_token(aud=MOBILE_CLIENT, sub="lonely",
+                                                email="lonely@example.com"))
+    token = r.json()["access_token"]
+
+    identities = client.get("/api/auth/oidc/identities", headers=auth(token)).json()
+    only = identities["identities"][0]["id"]
+    assert identities["auth_source"] == "sso"
+
+    resp = client.delete(f"/api/auth/oidc/identities/{only}", headers=auth(token))
+    assert resp.status_code == 400
+    assert resp.json()["detail"] == "oidc_last_identity"
+
+
+# ── token type confusion ─────────────────────────────────────────────────────
+
+def test_flow_token_cannot_be_used_as_access_token(client, authentik):
+    from routers.oidc_router import _encode_flow
+    register_admin(client, "admin", "pw")
+    flow = _encode_flow({"p": "authentik", "n": "n", "cv": "v", "jti": "s"})
+    assert client.get("/api/auth/me", headers=auth(flow)).status_code == 401