fix(oidc): harden session security from PR review (CSRF, revocation, refresh)

Address the high/medium correctness findings on the OIDC/SCIM PR:

- Login CSRF / session fixation: bind `state` to a Secure/HttpOnly/SameSite=Lax
  cookie at login and require the callback to echo it, so a code+state captured
  from another browser can't silently sign a victim into the attacker's account.
- Require `exp` on session JWTs under AUTH_TYPE=oidc (require_exp), so an
  exp-less HS256 token signed with JWT_SECRET_KEY can't authenticate forever or
  outlive the denylist.
- Denylist now keys revocation on an `iat` watermark instead of a deletable
  flag: a fresh login (newer iat) self-supersedes a revocation without clearing
  it, so sessions revoked on other devices stay revoked. Drops the
  login/SCIM-reactivation denylist-clearing paths (allow_user/allow_idp_sub).
- Refresh: gate the disabled-account check on the post-grant identity (not just
  the old sub); attempt the IdP grant before consuming the refresh token and
  return a retryable 503 (restoring the token) on transient IdP errors instead
  of force-logging-out a live session.
- Gate the oidc blueprint at request time on AUTH_TYPE=oidc, so non-oidc
  deployments cleanly 404 these routes instead of 500-ing on an unset
  OIDC_ISSUER (mirrors SCIM_ENABLED).
- Surface revocation write failures: back-channel logout returns 502, and SCIM
  deactivation rolls back and returns 503, when the denylist write fails — so
  the IdP retries instead of recording a logout/deprovision that didn't revoke.
- Back-channel logout: require `jti`, run the replay check unconditionally, and
  reject stale `iat` beyond the replay-cache window.
- Make migration 0017 idempotent (IF NOT EXISTS) so re-apply can't wedge startup.
- SCIM userName matching is case-insensitive (caseExact=false) for the list
  filter and create-dedup.

Tests added/updated across test_oidc.py, test_scim.py, test_auth.py,
test_app_routes.py and the SCIM integration test.

application/alembic/versions/0017_oidc_scim.py

@@ -21,10 +21,13 @@ depends_on: Union[str, Sequence[str], None] = None
 
 
 def upgrade() -> None:
-    op.execute("ALTER TABLE users ADD COLUMN active BOOLEAN NOT NULL DEFAULT TRUE;")
+    # IF NOT EXISTS keeps the migration idempotent: re-applying it over a
+    # partially-migrated or out-of-band-patched schema must not abort the whole
+    # upgrade (which would wedge startup with alembic_version stuck behind head).
+    op.execute("ALTER TABLE users ADD COLUMN IF NOT EXISTS active BOOLEAN NOT NULL DEFAULT TRUE;")
     op.execute(
         """
-        CREATE TABLE auth_events (
+        CREATE TABLE IF NOT EXISTS auth_events (
             id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
             user_id TEXT NOT NULL,
             event TEXT NOT NULL,
@@ -36,7 +39,7 @@ def upgrade() -> None:
         """
     )
     op.execute(
-        "CREATE INDEX auth_events_user_idx ON auth_events (user_id, created_at DESC);"
+        "CREATE INDEX IF NOT EXISTS auth_events_user_idx ON auth_events (user_id, created_at DESC);"
     )
 
 

application/api/oidc/denylist.py

@@ -1,9 +1,16 @@
 """Redis-backed session denylist for OIDC revocation.
 
 Back-channel logout and SCIM deactivation drop identifiers here; the
-request path refuses any session token whose identifiers match. Entries
-live slightly longer than ``OIDC_SESSION_LIFETIME_SECONDS`` — every
-session minted before the revocation expires before its denylist entry
+request path refuses any session token whose identifiers match. Each entry
+stores a revocation *watermark* (a Unix timestamp): a session is denied
+only when it was issued (``iat``) at or before the watermark. Storing a
+watermark instead of a boolean is what lets a fresh login self-supersede a
+prior revocation — its newer ``iat`` simply sits above the watermark —
+without deleting the entry and thereby resurrecting still-live sessions
+that were revoked on other devices.
+
+Entries live slightly longer than ``OIDC_SESSION_LIFETIME_SECONDS`` —
+every session issued at or before the watermark expires before the entry
 does, so nothing needs to be stored durably.
 
 Revocation is best-effort by design: if Redis is unreachable the check
@@ -13,6 +20,7 @@ fails open (sessions keep working) rather than taking the whole API down.
 from __future__ import annotations
 
 import logging
+import time
 
 from application.cache import get_redis_instance
 from application.core.settings import settings
@@ -34,7 +42,9 @@ def _set(key: str) -> bool:
         logger.error("Redis unavailable — could not denylist %s", key)
         return False
     try:
-        redis.set(key, "1", ex=_ttl_seconds())
+        # Store the revocation instant; existing entries are overwritten with a
+        # newer watermark (revoking again only ever moves it forward in time).
+        redis.set(key, str(int(time.time())), ex=_ttl_seconds())
         return True
     except Exception:
         logger.error("Failed to denylist %s", key, exc_info=True)
@@ -56,28 +66,19 @@ def deny_sid(sid: str) -> bool:
     return _set(_SID_PREFIX + sid)
 
 
-def allow_user(user_id: str) -> None:
-    """Clear a user-level denylist entry (SCIM reactivation)."""
-    _delete(_USER_PREFIX + user_id)
-
-
-def allow_idp_sub(sub: str) -> None:
-    """Clear an IdP-sub denylist entry (fresh login supersedes a back-channel logout)."""
-    _delete(_SUB_PREFIX + sub)
-
-
-def _delete(key: str) -> None:
-    redis = get_redis_instance()
-    if redis is None:
-        return
+def _watermark(value) -> float:
+    """Parse a stored watermark to a float; unparseable values deny everything."""
+    if isinstance(value, bytes):
+        value = value.decode("utf-8", "ignore")
     try:
-        redis.delete(key)
-    except Exception:
-        logger.warning("Failed to clear denylist key %s", key, exc_info=True)
+        return float(value)
+    except (TypeError, ValueError):
+        # Corrupt/legacy entry: treat as "deny" (a watermark far in the future).
+        return float("inf")
 
 
 def is_denied(decoded_token: dict) -> bool:
-    """True when any identifier in a decoded session token is denylisted."""
+    """True when the token was issued at/before a matching revocation watermark."""
     keys = []
     if decoded_token.get("sub"):
         keys.append(_USER_PREFIX + str(decoded_token["sub"]))
@@ -91,7 +92,16 @@ def is_denied(decoded_token: dict) -> bool:
     if redis is None:
         return False
     try:
-        return any(value is not None for value in redis.mget(keys))
+        values = redis.mget(keys)
     except Exception:
         logger.warning("Denylist check failed — allowing request", exc_info=True)
         return False
+    try:
+        iat = float(decoded_token.get("iat"))
+    except (TypeError, ValueError):
+        # No usable issue time — if any revocation exists for this identity we
+        # cannot prove the token post-dates it, so deny.
+        return any(value is not None for value in values)
+    # Strict ``<``: a session issued in the same second as (or after) the
+    # revocation — e.g. an immediate re-login — is allowed.
+    return any(value is not None and iat < _watermark(value) for value in values)

application/api/oidc/provider.py

@@ -41,6 +41,14 @@ class OIDCError(Exception):
     """Raised when an OIDC flow step fails."""
 
 
+class OIDCTransientError(OIDCError):
+    """Raised when the IdP was unreachable or returned 5xx — the call can be retried.
+
+    Subclasses OIDCError so existing ``except OIDCError`` handlers still catch
+    it; callers that can retry (e.g. session refresh) catch this first.
+    """
+
+
 def reset_cache() -> None:
     """Clear the cached discovery document and JWKS (used by tests)."""
     with _lock:
@@ -180,6 +188,11 @@ def validate_logout_token(logout_token: str) -> dict:
             "require_iss": True,
             "require_aud": True,
             "require_iat": True,
+            # jti is REQUIRED by OIDC Back-Channel Logout 1.0 and underpins
+            # replay protection — reject tokens that omit it. exp is not a
+            # required logout-token claim, so it stays optional (the caller
+            # bounds replay via iat freshness instead).
+            "require_jti": True,
             "require_exp": False,
         },
     )
@@ -208,11 +221,16 @@ def _token_request(data: dict) -> dict:
     try:
         response = requests.post(discovery["token_endpoint"], **post_kwargs)
     except requests.RequestException as exc:
-        raise OIDCError(f"Token request failed: {exc}") from exc
+        # Network failure — retryable.
+        raise OIDCTransientError(f"Token request failed: {exc}") from exc
     if response.status_code != 200:
         logger.error(
             "OIDC token request failed (%s): %s", response.status_code, response.text[:500]
         )
+        # 5xx is an IdP-side hiccup (retryable); 4xx (e.g. invalid_grant) is a
+        # definitive rejection the caller must not retry.
+        if response.status_code >= 500:
+            raise OIDCTransientError(f"Token endpoint returned {response.status_code}")
         raise OIDCError(f"Token endpoint returned {response.status_code}")
     return response.json()
 

application/api/oidc/routes.py

@@ -11,6 +11,7 @@ from __future__ import annotations
 
 import base64
 import hashlib
+import hmac
 import json
 import logging
 import secrets
@@ -35,6 +36,11 @@ STATE_TTL_SECONDS = 600
 HANDOFF_TTL_SECONDS = 60
 LOGOUT_JTI_TTL_SECONDS = 600
 MAX_PICTURE_CLAIM_CHARS = 2048
+# Browser-bound CSRF guard for the login flow: the callback requires this
+# cookie to echo the ``state`` it received. Scoped to the oidc paths so it is
+# only ever sent on the callback.
+STATE_COOKIE_NAME = "oidc_state"
+STATE_COOKIE_PATH = "/api/auth/oidc/"
 
 
 def _state_key(state: str) -> str:
@@ -66,9 +72,17 @@ def _frontend_url() -> str:
     return (settings.OIDC_FRONTEND_URL or "").rstrip("/") or "/"
 
 
+def _state_cookie_secure() -> bool:
+    """Mark the state cookie ``Secure`` when the app is served over HTTPS."""
+    return (settings.OIDC_FRONTEND_URL or "").lower().startswith("https")
+
+
 def _frontend_redirect(fragment: str):
     base = (settings.OIDC_FRONTEND_URL or "").rstrip("/")
-    return redirect(f"{base}/#{fragment}", code=302)
+    response = redirect(f"{base}/#{fragment}", code=302)
+    # Every callback exit consumes the one-shot state cookie.
+    response.delete_cookie(STATE_COOKIE_NAME, path=STATE_COOKIE_PATH)
+    return response
 
 
 def _no_store(payload, status: int = 200) -> Response:
@@ -223,7 +237,20 @@ def oidc_login():
         "code_challenge": _pkce_challenge(code_verifier),
         "code_challenge_method": "S256",
     }
-    return redirect(f"{authorization_endpoint}?{urlencode(params)}", code=302)
+    response = redirect(f"{authorization_endpoint}?{urlencode(params)}", code=302)
+    # Bind this login to the browser: the callback rejects any state that
+    # isn't echoed by this cookie, so a code+state captured from another
+    # browser (login CSRF / session fixation) can't complete the flow.
+    response.set_cookie(
+        STATE_COOKIE_NAME,
+        state,
+        max_age=STATE_TTL_SECONDS,
+        httponly=True,
+        secure=_state_cookie_secure(),
+        samesite="Lax",
+        path=STATE_COOKIE_PATH,
+    )
+    return response
 
 
 def oidc_callback():
@@ -235,6 +262,14 @@ def oidc_callback():
     if not state or not code:
         return _frontend_redirect("oidc_error=invalid_state")
 
+    # Browser binding: the state must match the cookie set at login. Without
+    # this, an attacker could feed a victim a code+state from the attacker's
+    # own login and silently sign the victim into the attacker's account.
+    cookie_state = request.cookies.get(STATE_COOKIE_NAME)
+    if not cookie_state or not hmac.compare_digest(cookie_state, state):
+        logger.warning("OIDC callback rejected: state cookie missing or mismatched")
+        return _frontend_redirect("oidc_error=invalid_state")
+
     redis = get_redis_instance()
     if redis is None:
         logger.error("Redis unavailable during OIDC callback")
@@ -268,12 +303,10 @@ def oidc_callback():
     if not _gate_and_audit_login(user_id, effective, groups):
         return _frontend_redirect("oidc_error=account_disabled")
 
-    # A fresh IdP-blessed authentication supersedes session-level revocations
-    # (back-channel logout denylists the IdP sub; without this, re-login would
-    # stay blocked until the denylist TTL ran out).
-    denylist.allow_user(str(user_id))
-    denylist.allow_idp_sub(str(claims["sub"]))
-
+    # No need to lift prior revocations here: the denylist keys on a revocation
+    # timestamp and the session minted below carries a newer ``iat``, so it is
+    # allowed automatically while still-live sessions revoked on other devices
+    # stay denied.
     session_token, jti = _mint_session_token(
         {
             "sub": user_id,
@@ -313,6 +346,17 @@ def oidc_token():
     return jsonify({"token": token})
 
 
+def _is_account_disabled(user_id: str) -> bool:
+    """True only when a readable user row is marked inactive (DB outage fails open)."""
+    try:
+        with db_readonly() as conn:
+            row = UsersRepository(conn).get(user_id)
+    except Exception:
+        logger.error("User lookup failed during OIDC refresh", exc_info=True)
+        return False
+    return bool(row is not None and row.get("active") is False)
+
+
 def oidc_refresh():
     """Rotate the stored IdP refresh token and mint a fresh session JWT."""
     decoded = handle_auth(request)
@@ -330,13 +374,8 @@ def oidc_refresh():
     if denylist.is_denied(decoded):
         return make_response(jsonify({"error": "token_revoked"}), 401)
 
-    try:
-        with db_readonly() as conn:
-            row = UsersRepository(conn).get(str(decoded["sub"]))
-    except Exception:
-        logger.error("User lookup failed during OIDC refresh", exc_info=True)
-        row = None
-    if row is not None and row.get("active") is False:
+    # Gate the current session identity before spending the refresh token.
+    if _is_account_disabled(str(decoded["sub"])):
         return make_response(jsonify({"error": "account_disabled"}), 401)
 
     redis = get_redis_instance()
@@ -349,8 +388,24 @@ def oidc_refresh():
 
     try:
         tokens = provider.refresh_grant(refresh_token)
+    except provider.OIDCTransientError:
+        # IdP unreachable / 5xx — the refresh token is still valid. Put it back
+        # and tell the client to retry instead of killing a live session over a
+        # transient blip (the frontend reschedules a renewal on 503).
+        try:
+            redis.set(
+                _refresh_key(str(decoded["jti"])),
+                refresh_token,
+                ex=settings.OIDC_SESSION_LIFETIME_SECONDS,
+            )
+        except Exception:
+            logger.warning("Failed to restore refresh token after transient error", exc_info=True)
+        logger.warning("OIDC refresh grant failed transiently", exc_info=True)
+        return make_response(jsonify({"error": "refresh_unavailable"}), 503)
     except provider.OIDCError:
-        logger.warning("OIDC refresh grant failed", exc_info=True)
+        # invalid_grant / 4xx — the refresh token is spent or revoked; leave it
+        # consumed so the client falls back to a fresh login.
+        logger.warning("OIDC refresh grant rejected", exc_info=True)
         return make_response(jsonify({"error": "refresh_failed"}), 401)
 
     identity = {
@@ -384,7 +439,13 @@ def oidc_refresh():
             return make_response(jsonify({"error": "not_authorized"}), 401)
         user_id = effective.get(settings.OIDC_USER_ID_CLAIM)
         if user_id:
-            identity["sub"] = str(user_id)
+            user_id = str(user_id)
+            # The pre-grant gate only saw the old sub. If the refreshed identity
+            # maps to a different user id, re-check that account is enabled
+            # before minting a session for it.
+            if user_id != str(decoded["sub"]) and _is_account_disabled(user_id):
+                return make_response(jsonify({"error": "account_disabled"}), 401)
+            identity["sub"] = user_id
         for claim in ("email", "name", "picture"):
             if effective.get(claim):
                 identity[claim] = effective[claim]
@@ -429,24 +490,40 @@ def oidc_backchannel_logout():
         logger.warning("Rejected OIDC back-channel logout token", exc_info=True)
         return _no_store(jsonify({"error": "invalid_logout_token"}), 400)
 
+    # Reject stale tokens: past the jti replay-cache window we can no longer
+    # detect replays by jti, so bound acceptance to that window (logout tokens
+    # are short-lived). Combined with the always-on jti check below, a captured
+    # token can be replayed neither within the window (jti dedupe) nor after it
+    # (iat too old).
+    now = int(time.time())
+    iat = claims.get("iat")
+    if not isinstance(iat, (int, float)) or now - iat > LOGOUT_JTI_TTL_SECONDS:
+        logger.warning("Rejected stale OIDC back-channel logout token")
+        return _no_store(jsonify({"error": "invalid_logout_token"}), 400)
+
     jti = claims.get("jti")
-    if jti:
-        redis = get_redis_instance()
-        if redis is not None:
-            try:
-                fresh = redis.set(_logout_jti_key(str(jti)), "1", ex=LOGOUT_JTI_TTL_SECONDS, nx=True)
-            except Exception:
-                logger.warning("Logout-token jti replay check failed; accepting token", exc_info=True)
-                fresh = True
-            if not fresh:
-                return _no_store(jsonify({"error": "invalid_logout_token"}), 400)
+    redis = get_redis_instance()
+    if redis is not None and jti:
+        try:
+            fresh = redis.set(_logout_jti_key(str(jti)), "1", ex=LOGOUT_JTI_TTL_SECONDS, nx=True)
+        except Exception:
+            logger.warning("Logout-token jti replay check failed; accepting token", exc_info=True)
+            fresh = True
+        if not fresh:
+            return _no_store(jsonify({"error": "invalid_logout_token"}), 400)
 
     sub = claims.get("sub")
     sid = claims.get("sid")
+    revoked = True
     if sub:
-        denylist.deny_idp_sub(str(sub))
+        revoked = bool(denylist.deny_idp_sub(str(sub))) and revoked
     if sid:
-        denylist.deny_sid(str(sid))
+        revoked = bool(denylist.deny_sid(str(sid))) and revoked
+    if not revoked:
+        # The denylist write failed (Redis down). Report failure so the IdP
+        # retries rather than recording a logout that never took effect.
+        logger.error("Back-channel logout could not persist the revocation")
+        return _no_store(jsonify({"error": "revocation_unavailable"}), 502)
 
     user_id = str(sub) if sub else f"sid:{sid}"
     try:
@@ -480,8 +557,22 @@ def oidc_logout():
     return redirect(f"{end_session}?{urlencode(params)}", code=302)
 
 
+def _require_oidc_enabled() -> Response | None:
+    """404 every oidc route unless OIDC is the active auth mode.
+
+    Registration is unconditional (so the import-time auth mode doesn't matter),
+    but the endpoints only work under ``AUTH_TYPE=oidc`` — otherwise OIDC_ISSUER
+    is unset and ``get_discovery`` would dereference ``None`` and 500. Mirrors
+    SCIM's ``SCIM_ENABLED`` gate.
+    """
+    if settings.AUTH_TYPE != "oidc":
+        return make_response(jsonify({"error": "oidc_not_enabled"}), 404)
+    return None
+
+
 def register(bp: Blueprint) -> None:
     """Attach the oidc auth routes to ``bp``."""
+    bp.before_request(_require_oidc_enabled)
     bp.add_url_rule(
         "/api/auth/oidc/login", view_func=oidc_login, methods=["GET"], endpoint="login"
     )

application/api/scim/routes.py

@@ -18,7 +18,7 @@ from typing import Any, Optional
 from flask import Blueprint, Response, request
 from sqlalchemy import Connection
 
-from application.api.oidc.denylist import allow_user, deny_user
+from application.api.oidc.denylist import deny_user
 from application.core.settings import settings
 from application.storage.db.repositories.auth_events import AuthEventsRepository
 from application.storage.db.repositories.users import UsersRepository
@@ -74,7 +74,9 @@ _USER_SCHEMA = {
             "type": "string",
             "multiValued": False,
             "required": True,
-            "caseExact": True,
+            # RFC 7643 §8.7.1 defines userName as caseExact=false; matching is
+            # case-insensitive (see UsersRepository.list_paginated / create).
+            "caseExact": False,
             "mutability": "immutable",
             "returned": "default",
             "uniqueness": "server",
@@ -187,6 +189,14 @@ def _parse_filter(raw: Optional[str]) -> tuple[Optional[str], Optional[Response]
 # ----------------------------------------------------------------------
 # Side effects
 # ----------------------------------------------------------------------
+class _RevocationUnavailable(Exception):
+    """A SCIM deactivation could not persist its session revocation (Redis down)."""
+
+
+def _revocation_unavailable(_exc: _RevocationUnavailable) -> Response:
+    return _scim_error(503, "Session revocation backend unavailable; retry")
+
+
 def _audit(conn: Connection, user_id: str, event: str) -> None:
     """Best-effort audit insert in a savepoint; failure never fails the request."""
     try:
@@ -203,10 +213,16 @@ def _apply_active(conn: Connection, row: dict, desired: bool) -> dict:
     updated = UsersRepository(conn).set_active(str(row["id"]), desired) or row
     user_id = row["user_id"]
     if desired:
-        allow_user(user_id)
+        # Reactivation needs no denylist write: deactivation set a revocation
+        # watermark that already lets a fresh login through (newer iat) while
+        # keeping the pre-deactivation sessions revoked.
         _audit(conn, user_id, "scim_reactivated")
     else:
-        deny_user(user_id)
+        if not deny_user(user_id):
+            # Redis is down — we cannot revoke live sessions. Fail the request
+            # (the surrounding transaction rolls the deactivation back) so the
+            # IdP retries instead of recording a deprovision that didn't revoke.
+            raise _RevocationUnavailable(user_id)
         _audit(conn, user_id, "scim_deactivated")
     return updated
 
@@ -392,6 +408,7 @@ def group_detail(group_id: str):
 def register(bp: Blueprint) -> None:
     """Attach the SCIM routes and bearer-token gate to ``bp``."""
     bp.before_request(_enforce_scim_auth)
+    bp.register_error_handler(_RevocationUnavailable, _revocation_unavailable)
     bp.add_url_rule(
         "/scim/v2/ServiceProviderConfig", view_func=service_provider_config, methods=["GET"],
         endpoint="service_provider_config",

application/auth.py

@@ -12,14 +12,19 @@ def handle_auth(request, data={}):
 
         jwt_token = jwt_token.replace("Bearer ", "")
 
+        is_oidc = settings.AUTH_TYPE == "oidc"
         try:
             decoded_token = jwt.decode(
                 jwt_token,
                 settings.JWT_SECRET_KEY,
                 algorithms=["HS256"],
-                # oidc sessions are minted with an exp at the login callback;
-                # simple_jwt/session_jwt tokens never carried one.
-                options={"verify_exp": settings.AUTH_TYPE == "oidc"},
+                # oidc sessions are minted with an exp at the login callback and
+                # must carry one: require_exp rejects any exp-less HS256 token
+                # signed with JWT_SECRET_KEY (e.g. a legacy simple_jwt/session_jwt
+                # token), which would otherwise authenticate forever and be
+                # unrevocable. simple_jwt/session_jwt never carried an exp, so the
+                # requirement is scoped to oidc.
+                options={"verify_exp": is_oidc, "require_exp": is_oidc},
             )
             return decoded_token
         except ExpiredSignatureError:

application/storage/db/repositories/users.py

@@ -249,7 +249,20 @@ class UsersRepository:
     # SCIM provisioning
     # ------------------------------------------------------------------
     def create(self, user_id: str, active: bool = True) -> Optional[dict]:
-        """Insert a new user row; ``None`` means ``user_id`` already exists."""
+        """Insert a new user row; ``None`` means a user with that id already exists.
+
+        SCIM ``userName`` is case-insensitive (caseExact=false), so a row that
+        differs only by case counts as a duplicate. The pre-check keeps SCIM
+        from provisioning a second row for a user the OIDC login already
+        created under different casing; the ``ON CONFLICT`` clause remains an
+        exact-match backstop for the concurrent-insert race.
+        """
+        existing = self._conn.execute(
+            text("SELECT 1 FROM users WHERE lower(user_id) = lower(:user_id)"),
+            {"user_id": user_id},
+        ).first()
+        if existing is not None:
+            return None
         result = self._conn.execute(
             text(
                 """
@@ -302,7 +315,9 @@ class UsersRepository:
         where = ""
         filter_params: dict = {}
         if user_name is not None:
-            where = "WHERE user_id = :user_name"
+            # SCIM userName is case-insensitive (caseExact=false); match on
+            # lower(user_id) so the IdP finds the account regardless of casing.
+            where = "WHERE lower(user_id) = lower(:user_name)"
             filter_params = {"user_name": user_name}
         total = self._conn.execute(
             text(f"SELECT count(*) FROM users {where}"), filter_params

tests/integration/test_scim.py

@@ -58,10 +58,8 @@ def scim_env(monkeypatch):
     """Enable SCIM on the settings singleton and stub the Redis denylist."""
     monkeypatch.setattr(settings, "SCIM_ENABLED", True)
     monkeypatch.setattr(settings, "SCIM_TOKEN", SCIM_TOKEN)
-    with patch("application.api.scim.routes.deny_user") as deny_user_mock, patch(
-        "application.api.scim.routes.allow_user"
-    ) as allow_user_mock:
-        yield SimpleNamespace(deny_user=deny_user_mock, allow_user=allow_user_mock)
+    with patch("application.api.scim.routes.deny_user") as deny_user_mock:
+        yield SimpleNamespace(deny_user=deny_user_mock)
 
 
 @pytest.fixture
@@ -146,7 +144,8 @@ class TestScimLifecycle:
         assert response.status_code == 200
         assert response.get_json()["active"] is True
         assert _fetch_user(scim_user_name)["active"] is True
-        scim_env.allow_user.assert_called_once_with(scim_user_name)
+        # Reactivation no longer clears the denylist (watermark model).
+        scim_env.deny_user.assert_called_once_with(scim_user_name)
         assert any(e["event"] == "scim_reactivated" for e in _fetch_events(scim_user_name))
 
         # Soft delete deactivates again

tests/test_app_routes.py

@@ -127,12 +127,15 @@ class TestAuthenticateRequest:
     @pytest.mark.unit
     def test_oidc_auth_paths_exempt_from_jwt_check(self, client, app):
         # A stale/expired Bearer header must never 401 the oidc login
-        # endpoints — they are the only path back to a fresh session.
+        # endpoints — they are the only path back to a fresh session. The oidc
+        # routes are only live under AUTH_TYPE=oidc, so pin it here.
+        from application.core.settings import settings as _settings
+
         with patch(
             "application.app.handle_auth", return_value={"error": "invalid_token"}
         ), patch(
             "application.api.oidc.routes.get_redis_instance", return_value=None
-        ):
+        ), patch.object(_settings, "AUTH_TYPE", "oidc"):
             response = client.get(
                 "/api/auth/oidc/login", headers={"Authorization": "Bearer garbage"}
             )

tests/test_auth.py

@@ -46,7 +46,7 @@ class TestHandleAuth:
             "valid_token",
             "secret",
             algorithms=["HS256"],
-            options={"verify_exp": False},
+            options={"verify_exp": False, "require_exp": False},
         )
 
     def test_returns_error_on_invalid_jwt(self):
@@ -117,7 +117,7 @@ class TestHandleAuthOidc:
             "valid_token",
             "secret",
             algorithms=["HS256"],
-            options={"verify_exp": True},
+            options={"verify_exp": True, "require_exp": True},
         )
 
     def test_expired_token_returns_token_expired(self):
@@ -154,10 +154,10 @@ class TestHandleAuthOidc:
 
         assert result["error"] == "invalid_token"
 
-    def test_token_without_exp_passes_real_jose(self):
-        # Pins that e2e/integration helper tokens (signed without an exp
-        # claim) stay valid under oidc: python-jose only validates claims
-        # that are present.
+    def test_token_without_exp_rejected_under_oidc(self):
+        # Under oidc, exp is REQUIRED: an exp-less HS256 token signed with the
+        # shared secret (e.g. a legacy simple_jwt/session_jwt token) must not
+        # authenticate, or it would be valid forever and unrevocable.
         from jose import jwt as real_jwt
 
         from application.auth import handle_auth
@@ -171,7 +171,7 @@ class TestHandleAuthOidc:
             mock_settings.JWT_SECRET_KEY = "secret"
             result = handle_auth(mock_request)
 
-        assert result == {"sub": "helper_user"}
+        assert result["error"] == "invalid_token"
 
     def test_expired_token_real_jose(self):
         import time

tests/test_oidc.py

@@ -549,6 +549,56 @@ class TestValidateLogoutToken:
             self._validate(sign_id_token(logout_token_claims(aud="someone-else")))
 
 
+class _WatermarkRedis:
+    """Minimal redis stand-in for the denylist (set + mget)."""
+
+    def __init__(self):
+        self.store = {}
+
+    def set(self, key, value, ex=None):
+        self.store[key] = value.encode("utf-8") if isinstance(value, str) else value
+        return True
+
+    def mget(self, keys):
+        return [self.store.get(key) for key in keys]
+
+
+@pytest.mark.unit
+class TestDenylistWatermark:
+    """is_denied compares the token iat against the stored revocation timestamp."""
+
+    def _wire(self, monkeypatch):
+        from application.api.oidc import denylist
+
+        redis = _WatermarkRedis()
+        monkeypatch.setattr(denylist, "get_redis_instance", lambda: redis)
+        return denylist, redis
+
+    def test_session_before_revocation_denied_after_allowed(self, monkeypatch):
+        denylist, redis = self._wire(monkeypatch)
+        denylist.deny_user("u1")
+        watermark = int(redis.store[denylist._USER_PREFIX + "u1"])
+        # Issued before the revocation -> denied.
+        assert denylist.is_denied({"sub": "u1", "iat": watermark - 5}) is True
+        # Issued at/after the revocation (e.g. an immediate re-login) -> allowed.
+        assert denylist.is_denied({"sub": "u1", "iat": watermark + 5}) is False
+
+    def test_unrelated_identity_not_denied(self, monkeypatch):
+        denylist, redis = self._wire(monkeypatch)
+        denylist.deny_user("u1")
+        watermark = int(redis.store[denylist._USER_PREFIX + "u1"])
+        assert denylist.is_denied({"sub": "u2", "iat": watermark - 5}) is False
+
+    def test_missing_iat_is_conservatively_denied(self, monkeypatch):
+        denylist, _ = self._wire(monkeypatch)
+        denylist.deny_idp_sub("idp-1")
+        assert denylist.is_denied({"oidc_sub": "idp-1"}) is True
+
+    def test_no_entry_means_allowed(self, monkeypatch):
+        denylist, _ = self._wire(monkeypatch)
+        assert denylist.is_denied({"sub": "u1", "iat": 1}) is False
+
+
 @pytest.fixture(scope="module")
 def app():
     with patch("application.app.handle_auth", return_value={"sub": "test_user"}):
@@ -592,11 +642,14 @@ def db_mocks():
         yield SimpleNamespace(users=users_repo, events=events_repo)
 
 
-def _seed_state(fake_redis, state="state-1", nonce="nonce-1"):
+def _seed_state(client, fake_redis, state="state-1", nonce="nonce-1"):
     fake_redis.set(
         f"oidc:state:{state}",
         json.dumps({"code_verifier": "verifier-1", "nonce": nonce}),
     )
+    # Mirror the browser cookie the login route sets, so the callback's
+    # state-binding (CSRF) check passes.
+    client.set_cookie("oidc_state", state)
     return state, nonce
 
 
@@ -657,6 +710,21 @@ class TestLoginRoute:
         assert response.status_code == 503
 
 
+@pytest.mark.unit
+class TestOidcDisabled:
+    """Outside AUTH_TYPE=oidc the routes 404 cleanly instead of 500-ing."""
+
+    def test_routes_404_when_not_oidc(self, client, monkeypatch):
+        monkeypatch.setattr(settings, "AUTH_TYPE", "session_jwt")
+        assert client.get("/api/auth/oidc/login").status_code == 404
+        assert client.get("/api/auth/oidc/logout").status_code == 404
+        # Even a (would-be) RS256 logout token must not reach discovery.
+        post = client.post(
+            "/api/auth/oidc/backchannel-logout", data={"logout_token": "x"}
+        )
+        assert post.status_code == 404
+
+
 @pytest.mark.unit
 class TestCallbackRoute:
 
@@ -664,11 +732,11 @@ class TestCallbackRoute:
     def _db(self, db_mocks):
         self.db = db_mocks
 
-    def _seed_state(self, fake_redis, state="state-1", nonce="nonce-1"):
-        return _seed_state(fake_redis, state=state, nonce=nonce)
+    def _seed_state(self, client, fake_redis, state="state-1", nonce="nonce-1"):
+        return _seed_state(client, fake_redis, state=state, nonce=nonce)
 
     def test_happy_path_mints_session_and_redirects_with_handoff(self, client, fake_redis):
-        state, nonce = self._seed_state(fake_redis)
+        state, nonce = self._seed_state(client, fake_redis)
         with patch("application.api.oidc.provider.requests") as mock_requests:
             mock_requests.get.side_effect = make_fake_get([PUBLIC_JWK])
             mock_requests.post.return_value = _mint_id_token_response(nonce)
@@ -690,7 +758,7 @@ class TestCallbackRoute:
         assert f"oidc:state:{state}" not in fake_redis.store
 
     def test_replayed_state_rejected(self, client, fake_redis):
-        state, nonce = self._seed_state(fake_redis)
+        state, nonce = self._seed_state(client, fake_redis)
         with patch("application.api.oidc.provider.requests") as mock_requests:
             mock_requests.get.side_effect = make_fake_get([PUBLIC_JWK])
             mock_requests.post.return_value = _mint_id_token_response(nonce)
@@ -701,9 +769,28 @@ class TestCallbackRoute:
         assert replay.headers["Location"] == f"{FRONTEND_URL}/#oidc_error=invalid_state"
 
     def test_unknown_state_rejected(self, client, fake_redis):
+        # Cookie matches the state (passes the CSRF binding) but the state was
+        # never stored in Redis — the server-side lookup must still reject it.
+        client.set_cookie("oidc_state", "forged")
         response = client.get("/api/auth/oidc/callback?code=abc&state=forged")
         assert response.headers["Location"] == f"{FRONTEND_URL}/#oidc_error=invalid_state"
 
+    def test_missing_state_cookie_rejected(self, client, fake_redis):
+        # State is valid server-side, but no browser cookie binds it — a forged
+        # cross-browser callback (login CSRF) lands here.
+        self._seed_state(client, fake_redis)
+        client.delete_cookie("oidc_state")
+        response = client.get("/api/auth/oidc/callback?code=abc&state=state-1")
+        assert response.headers["Location"] == f"{FRONTEND_URL}/#oidc_error=invalid_state"
+
+    def test_mismatched_state_cookie_rejected(self, client, fake_redis):
+        # The cookie carries a different login attempt's state than the query
+        # param — reject rather than complete an attacker-seeded flow.
+        self._seed_state(client, fake_redis)
+        client.set_cookie("oidc_state", "attacker-state")
+        response = client.get("/api/auth/oidc/callback?code=abc&state=state-1")
+        assert response.headers["Location"] == f"{FRONTEND_URL}/#oidc_error=invalid_state"
+
     def test_idp_error_forwarded(self, client, fake_redis):
         response = client.get("/api/auth/oidc/callback?error=access_denied")
         assert (
@@ -712,7 +799,7 @@ class TestCallbackRoute:
         )
 
     def test_nonce_mismatch_fails_auth(self, client, fake_redis):
-        state, _ = self._seed_state(fake_redis, nonce="nonce-1")
+        state, _ = self._seed_state(client, fake_redis, nonce="nonce-1")
         with patch("application.api.oidc.provider.requests") as mock_requests:
             mock_requests.get.side_effect = make_fake_get([PUBLIC_JWK])
             mock_requests.post.return_value = _mint_id_token_response("evil-nonce")
@@ -722,7 +809,7 @@ class TestCallbackRoute:
 
     def test_missing_user_id_claim(self, client, fake_redis, monkeypatch):
         monkeypatch.setattr(settings, "OIDC_USER_ID_CLAIM", "preferred_username")
-        state, nonce = self._seed_state(fake_redis)
+        state, nonce = self._seed_state(client, fake_redis)
         with patch("application.api.oidc.provider.requests") as mock_requests:
             mock_requests.get.side_effect = make_fake_get([PUBLIC_JWK])
             mock_requests.post.return_value = _mint_id_token_response(nonce)
@@ -731,7 +818,7 @@ class TestCallbackRoute:
         assert response.headers["Location"] == f"{FRONTEND_URL}/#oidc_error=missing_claim"
 
     def test_optional_profile_claims_omitted_when_absent(self, client, fake_redis):
-        state, nonce = self._seed_state(fake_redis)
+        state, nonce = self._seed_state(client, fake_redis)
         claims = id_token_claims(nonce=nonce)
         del claims["email"]
         del claims["name"]
@@ -810,7 +897,7 @@ class TestCallbackGroups:
         self.db = db_mocks
 
     def _callback(self, client, fake_redis, claims, userinfo=None, userinfo_status=200):
-        state, nonce = _seed_state(fake_redis)
+        state, nonce = _seed_state(client, fake_redis)
         claims = {**claims, "nonce": nonce}
         with patch("application.api.oidc.provider.requests") as mock_requests:
             mock_requests.get.side_effect = make_fake_get(
@@ -909,7 +996,7 @@ class TestCallbackUserGate:
         self.db = db_mocks
 
     def _callback(self, client, fake_redis):
-        state, nonce = _seed_state(fake_redis)
+        state, nonce = _seed_state(client, fake_redis)
         with patch("application.api.oidc.provider.requests") as mock_requests:
             mock_requests.get.side_effect = make_fake_get([PUBLIC_JWK])
             mock_requests.post.return_value = _mint_id_token_response(nonce)
@@ -957,25 +1044,25 @@ class TestCallbackUserGate:
 
         assert "#oidc_code=" in response.headers["Location"]
 
-    def test_successful_login_clears_session_revocations(self, client, fake_redis):
-        # A back-channel logout denylists the IdP sub; a fresh IdP-blessed
-        # login must lift session-level revocations or re-login would stay
-        # blocked until the denylist TTL expires.
+    def test_successful_login_does_not_touch_denylist(self, client, fake_redis):
+        # The denylist keys on a revocation timestamp and the minted session
+        # carries a newer iat, so a fresh login is allowed without clearing any
+        # entry — clearing would resurrect sessions revoked on other devices.
         with patch("application.api.oidc.routes.denylist") as deny:
             response = self._callback(client, fake_redis)
 
         assert "#oidc_code=" in response.headers["Location"]
-        deny.allow_user.assert_called_once_with("oidc-user-1")
-        deny.allow_idp_sub.assert_called_once_with("oidc-user-1")
+        assert not deny.allow_user.called
+        assert not deny.allow_idp_sub.called
 
-    def test_denied_login_does_not_clear_revocations(self, client, fake_redis):
+    def test_denied_login_does_not_touch_denylist(self, client, fake_redis):
         self.db.users.get.return_value = {"user_id": "oidc-user-1", "active": False}
         with patch("application.api.oidc.routes.denylist") as deny:
             response = self._callback(client, fake_redis)
 
         assert "account_disabled" in response.headers["Location"]
-        deny.allow_user.assert_not_called()
-        deny.allow_idp_sub.assert_not_called()
+        assert not deny.allow_user.called
+        assert not deny.allow_idp_sub.called
 
 
 @pytest.mark.unit
@@ -986,7 +1073,7 @@ class TestSessionTokenMint:
         self.db = db_mocks
 
     def _login_decoded(self, client, fake_redis, claims=None, **token_extra):
-        state, nonce = _seed_state(fake_redis)
+        state, nonce = _seed_state(client, fake_redis)
         claims = {**(claims or id_token_claims()), "nonce": nonce}
         with patch("application.api.oidc.provider.requests") as mock_requests:
             mock_requests.get.side_effect = make_fake_get([PUBLIC_JWK])
@@ -1127,6 +1214,34 @@ class TestBackchannelLogoutRoute:
         assert replay.status_code == 400
         assert self.denylist.deny_idp_sub.call_count == 1
 
+    def test_missing_jti_rejected(self, client, fake_redis):
+        # jti is required (Back-Channel Logout 1.0) and underpins replay
+        # protection — a token without one must not be accepted.
+        claims = logout_token_claims()
+        del claims["jti"]
+        response = self._post(client, claims)
+
+        assert response.status_code == 400
+        self.denylist.deny_idp_sub.assert_not_called()
+
+    def test_stale_iat_rejected(self, client, fake_redis):
+        # Beyond the jti replay-cache window the token can no longer be deduped,
+        # so an old iat is rejected outright (bounds replay after eviction).
+        claims = logout_token_claims(jti="bcl-stale", iat=int(time.time()) - 10000)
+        response = self._post(client, claims)
+
+        assert response.status_code == 400
+        self.denylist.deny_idp_sub.assert_not_called()
+
+    def test_revocation_write_failure_returns_502(self, client, fake_redis):
+        # Redis down: report failure so the IdP retries instead of recording a
+        # logout that never revoked anything.
+        self.denylist.deny_idp_sub.return_value = False
+        response = self._post(client, logout_token_claims(jti="bcl-502"))
+
+        assert response.status_code == 502
+        assert response.get_json() == {"error": "revocation_unavailable"}
+
 
 @pytest.mark.unit
 class TestRefreshRoute:
@@ -1328,3 +1443,37 @@ class TestRefreshRoute:
 
         assert response.status_code == 503
         assert response.get_json() == {"error": "redis_unavailable"}
+
+    def test_transient_idp_failure_503_and_token_restored(self, client, fake_redis):
+        # 5xx from the IdP is transient: keep the (still-valid) refresh token and
+        # tell the client to retry rather than dropping a live session.
+        token = make_session_token()
+        fake_redis.store["oidc:refresh:jti-1"] = b"rt-old"
+        response, _ = self._refresh(client, token, idp_status=503)
+
+        assert response.status_code == 503
+        assert response.get_json() == {"error": "refresh_unavailable"}
+        assert fake_redis.store["oidc:refresh:jti-1"] == b"rt-old"
+
+    def test_remapped_disabled_identity_rejected(self, client, fake_redis):
+        # The refreshed id_token maps to a different (disabled) user id than the
+        # session sub — the post-grant gate must refuse it.
+        token = make_session_token()
+        fake_redis.store["oidc:refresh:jti-1"] = b"rt-old"
+        self.db.users.get.side_effect = (
+            lambda uid: {"user_id": uid, "active": False} if uid == "oidc-user-2" else None
+        )
+        id_claims = id_token_claims(sub="oidc-user-2")
+        del id_claims["nonce"]
+        response, _ = self._refresh(
+            client,
+            token,
+            idp_response={
+                "access_token": "at-2",
+                "refresh_token": "rt-new",
+                "id_token": sign_id_token(id_claims),
+            },
+        )
+
+        assert response.status_code == 401
+        assert response.get_json() == {"error": "account_disabled"}

tests/test_scim.py

@@ -63,17 +63,19 @@ def scim_mocks():
         "application.api.scim.routes.AuthEventsRepository"
     ) as audit_cls, patch(
         "application.api.scim.routes.deny_user"
-    ) as deny_user_mock, patch(
-        "application.api.scim.routes.allow_user"
-    ) as allow_user_mock:
+    ) as deny_user_mock:
         conn = MagicMock(name="conn")
         db_session_mock.return_value.__enter__.return_value = conn
         db_readonly_mock.return_value.__enter__.return_value = conn
+        # Don't let the mocked context manager swallow exceptions (a real
+        # ``engine.begin()`` __exit__ returns None) — _RevocationUnavailable
+        # must propagate to the blueprint error handler.
+        db_session_mock.return_value.__exit__.return_value = False
+        db_readonly_mock.return_value.__exit__.return_value = False
         yield SimpleNamespace(
             users=users_cls.return_value,
             audit=audit_cls.return_value,
             deny_user=deny_user_mock,
-            allow_user=allow_user_mock,
             conn=conn,
         )
 
@@ -355,7 +357,7 @@ class TestReplaceUser:
         assert response.get_json()["scimType"] == "mutability"
         scim_mocks.users.set_active.assert_not_called()
 
-    def test_put_active_true_triggers_allow(self, client, scim_settings, scim_mocks):
+    def test_put_active_true_reactivates(self, client, scim_settings, scim_mocks):
         row = _user_row(active=False)
         scim_mocks.users.get_by_pk.return_value = row
         scim_mocks.users.set_active.return_value = {**row, "active": True}
@@ -367,7 +369,8 @@ class TestReplaceUser:
         assert response.status_code == 200
         assert response.get_json()["active"] is True
         scim_mocks.users.set_active.assert_called_once_with(USER_PK, True)
-        scim_mocks.allow_user.assert_called_once_with("alice@example.com")
+        # Reactivation does not touch the denylist (the deactivation watermark
+        # already lets a fresh login through on a newer iat).
         scim_mocks.deny_user.assert_not_called()
         scim_mocks.audit.insert.assert_called_once_with(
             "alice@example.com", "scim_reactivated", metadata={"via": "scim"}
@@ -400,7 +403,6 @@ class TestReplaceUser:
         assert response.status_code == 200
         scim_mocks.users.set_active.assert_not_called()
         scim_mocks.deny_user.assert_not_called()
-        scim_mocks.allow_user.assert_not_called()
         scim_mocks.audit.insert.assert_not_called()
 
     def test_put_missing_user_returns_404(self, client, scim_settings, scim_mocks):
@@ -459,7 +461,7 @@ class TestPatchUser:
         assert response.status_code == 200
         assert response.get_json()["active"] is True
         scim_mocks.users.set_active.assert_called_once_with(USER_PK, True)
-        scim_mocks.allow_user.assert_called_once_with("alice@example.com")
+        scim_mocks.deny_user.assert_not_called()
         scim_mocks.audit.insert.assert_called_once_with(
             "alice@example.com", "scim_reactivated", metadata={"via": "scim"}
         )
@@ -537,6 +539,16 @@ class TestDeleteUser:
         response = client.delete(f"/scim/v2/Users/{USER_PK}", headers=AUTH)
         assert response.status_code == 404
 
+    def test_deactivation_revocation_failure_returns_503(self, client, scim_settings, scim_mocks):
+        # Redis down: deny_user returns False, so the deactivation rolls back and
+        # the IdP gets a 503 to retry instead of a false deprovision success.
+        row = _user_row(active=True)
+        scim_mocks.users.get_by_pk.return_value = row
+        scim_mocks.users.set_active.return_value = {**row, "active": False}
+        scim_mocks.deny_user.return_value = False
+        response = client.delete(f"/scim/v2/Users/{USER_PK}", headers=AUTH)
+        assert response.status_code == 503
+
 
 @pytest.mark.unit
 class TestGroups: