From c24873f195b69883d7689dcc232f17c65ac527cb Mon Sep 17 00:00:00 2001
From: Thor Whalen <1906276+thorwhalen@users.noreply.github.com>
Date: Tue, 22 Sep 2026 16:14:33 +0000
Subject: [PATCH 1/2] Revoke connector refresh families and shared-password
 cookies on credential change

One on_credentials_changed(email, *, keep=None) hook, built by the plugin and
injected into the auth and admin routers, now runs on every path that changes an
account's credentials (admin delete, admin password set, self-service change,
reset-link redemption; the set-password CLI does the same). It revokes the
account's browser sessions as before and, when the OAuth server keeps refresh
tokens, tombstones and deletes every refresh family issued to that account and
drops its unredeemed authorization codes.

shared_auth_<app> cookies now sign a keyed fingerprint of the app's current
shared-password hash instead of a constant; the middleware and the shared-login
page compare it, so rotating a shared password ends older cookies, and an app
with no configured password admits no cookie.

SessionStore takes an optional max_age and sweeps a bounded batch of expired
records whenever a session is created.

Closes #26

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
---
 README.md                           |  12 ++
 enlace_auth/__main__.py             |  33 +++-
 enlace_auth/admin/routes.py         |  15 +-
 enlace_auth/auth/middleware.py      |  16 +-
 enlace_auth/auth/oauth_server.py    |  46 ++---
 enlace_auth/auth/revocation.py      | 219 ++++++++++++++++++++++
 enlace_auth/auth/routes.py          |  35 +++-
 enlace_auth/auth/sessions.py        |  46 ++++-
 enlace_auth/plugin.py               |  31 ++-
 tests/test_auth_middleware.py       |  35 +++-
 tests/test_credential_revocation.py | 280 ++++++++++++++++++++++++++++
 11 files changed, 719 insertions(+), 49 deletions(-)
 create mode 100644 enlace_auth/auth/revocation.py
 create mode 100644 tests/test_credential_revocation.py

diff --git a/README.md b/README.md
index c0f7fcb..1b3ed33 100644
--- a/README.md
+++ b/README.md
@@ -129,6 +129,18 @@ within one access-token lifetime rather than one refresh-token lifetime. Keep
 `access_token_ttl_seconds` short for that reason: with refresh in place, a short
 access token costs nothing and is what bounds revocation lag.
 
+**A credential change ends connector sessions too.** Deleting a user, an admin
+password set, a self-service password change, a reset-link redemption and the
+`set-password` CLI all revoke the account's browser sessions *and* its refresh
+families (plus any unredeemed authorization codes); the connector must be
+re-authorized. Already-issued access tokens live out their TTL.
+
+**Rotating a shared password ends its cookies.** A `shared_auth_<app>` cookie
+carries a keyed fingerprint of the app's shared-password hash at the time it was
+minted, so after you change the hash (and restart), every cookie from the old
+password is refused. Upgrading to this version signs everyone out of
+shared-password apps once.
+
 Plus environment variables:
 
 - `ENLACE_SIGNING_KEY` — signing key (32+ chars). Generate with `python -c
diff --git a/enlace_auth/__main__.py b/enlace_auth/__main__.py
index e5d8a1e..5482183 100644
--- a/enlace_auth/__main__.py
+++ b/enlace_auth/__main__.py
@@ -95,6 +95,33 @@ def _load_session_store(toml_path: Path = Path("platform.toml")):
     return SessionStore(factory("sessions"))
 
 
+def _connector_tombstone_ttl(toml_path: Path = Path("platform.toml")) -> int:
+    """The refresh-family tombstone lifetime the configured OAuth server uses."""
+    from enlace_auth.auth.revocation import refresh_tombstone_ttl
+
+    osc = coerce_auth_config(PlatformConfig.from_toml(toml_path).auth).oauth_server
+    return refresh_tombstone_ttl(
+        refresh_token_ttl=osc.refresh_token_ttl_seconds,
+        refresh_reuse_detection=osc.refresh_reuse_detection_seconds,
+    )
+
+
+def _revoke_connector_subject(email: str, toml_path: Path) -> int:
+    """Revoke *email*'s connector refresh families (tombstoned); return count."""
+    from enlace_auth.auth.revocation import revoke_refresh_subject
+    from enlace_auth.stores import make_file_store_factory
+
+    auth = coerce_auth_config(PlatformConfig.from_toml(toml_path).auth)
+    factory = make_file_store_factory(auth.stores.path)
+    return revoke_refresh_subject(
+        factory("oauth_refresh_tokens"),
+        email,
+        reason="revoked by the enlace-auth CLI",
+        tombstone_ttl=_connector_tombstone_ttl(toml_path),
+        code_store=factory("oauth_codes"),
+    )
+
+
 def _load_user_store(toml_path: Path = Path("platform.toml")):
     """Open the platform's user store (email -> {password_hash, ...})."""
     from enlace_auth.stores import make_file_store_factory
@@ -228,7 +255,11 @@ def set_password(email: str, *, toml: str = "platform.toml"):
     store[key] = updated
     # Same rule as the HTTP reset paths: a new password ends the old sessions.
     revoked = _load_session_store(Path(toml)).revoke_user(key)
-    print(f"Password updated for {key}; {revoked} existing session(s) revoked.")
+    families = _revoke_connector_subject(key, Path(toml))
+    print(
+        f"Password updated for {key}; {revoked} existing session(s) and "
+        f"{families} connector session(s) revoked."
+    )
 
 
 def reset_link(
diff --git a/enlace_auth/admin/routes.py b/enlace_auth/admin/routes.py
index fb0b09d..5479a58 100644
--- a/enlace_auth/admin/routes.py
+++ b/enlace_auth/admin/routes.py
@@ -86,6 +86,7 @@ def make_admin_router(
     reset_link_ttl: int = DEFAULT_HANDOFF_TTL,
     resource_allowlist: Optional[Mapping[str, list[str]]] = None,
     public_base_url: Optional[str] = None,
+    on_credentials_changed=None,  # CredentialsChanged; default: sessions only
 ) -> APIRouter:
     """Build a FastAPI router exposing ``/_admin/api/*`` endpoints.
 
@@ -115,7 +116,17 @@ def make_admin_router(
     (it isn't) or, worse, to make some *other* app public by analogy (which
     would be). Passing the allow-list lets the dashboard show who can actually
     reach each one.
+
+    ``on_credentials_changed`` (``hook(email, *, keep=None)``) runs after a
+    user is deleted or has their password set. The default revokes the
+    account's browser sessions; the plugin injects one that also revokes its
+    OAuth connector refresh families (``enlace_auth.auth.revocation``).
     """
+    from enlace_auth.auth.revocation import make_on_credentials_changed
+
+    credentials_changed = on_credentials_changed or make_on_credentials_changed(
+        session_store
+    )
     admin_set = frozenset(e.lower() for e in admin_emails)
     apps_snapshot = list(apps)
     app_by_name = {a.name: a for a in apps_snapshot}
@@ -214,7 +225,7 @@ async def delete_user(email: str, request: Request) -> dict[str, Any]:
         # a deleted account keeps working until its cookie expires unless its
         # sessions go too. (An actor deleting themselves is logged out.)
         _ = actor
-        session_store.revoke_user(target)
+        credentials_changed(target)
         return {"ok": True, "email": target}
 
     @router.post("/users/{email}/password")
@@ -234,7 +245,7 @@ async def admin_reset_password(
         user_store[target] = record
         # An admin reset is how a compromised account is recovered: whoever
         # holds a session opened with the old password must lose it.
-        session_store.revoke_user(target)
+        credentials_changed(target)
         return {"ok": True, "email": target}
 
     @router.post("/users/{email}/reset-link")
diff --git a/enlace_auth/auth/middleware.py b/enlace_auth/auth/middleware.py
index 221ec5a..97aa2cd 100644
--- a/enlace_auth/auth/middleware.py
+++ b/enlace_auth/auth/middleware.py
@@ -36,6 +36,7 @@
 from urllib.parse import unquote
 
 from enlace_auth.auth.cookies import verify_cookie
+from enlace_auth.auth.revocation import shared_cookie_valid
 from enlace_auth.auth.sessions import SessionStore
 
 _logger = logging.getLogger("enlace_auth.middleware")
@@ -291,16 +292,21 @@ async def __call__(self, scope, receive, send):
             app_id = rule.app_id if rule is not None else ""
             name = f"shared_auth_{app_id}"
             token = cookies.get(name)
-            if (
-                not token
-                or verify_cookie(
+            # The cookie must carry the fingerprint of the app's CURRENT shared
+            # password hash: rotating the password ends every older cookie, and
+            # an app with no configured password admits no cookie at all.
+            value = (
+                verify_cookie(
                     token,
                     self._signing_key,
                     max_age=self._max_age,
                     salt=f"shared:{app_id}",
                 )
-                is None
-            ):
+                if token
+                else None
+            )
+            expected_hash = rule.shared_password_hash if rule is not None else None
+            if not shared_cookie_valid(value, expected_hash, self._signing_key):
                 return await self._deny(scope, send, "shared")
             state["user_id"] = "shared"
             await self.app(scope, receive, send)
diff --git a/enlace_auth/auth/oauth_server.py b/enlace_auth/auth/oauth_server.py
index 631bbd2..64665a6 100644
--- a/enlace_auth/auth/oauth_server.py
+++ b/enlace_auth/auth/oauth_server.py
@@ -57,6 +57,11 @@
 
 from enlace_auth.auth import pages
 from enlace_auth.auth.cookies import sign_cookie, verify_cookie
+from enlace_auth.auth.revocation import (
+    refresh_tombstone_ttl,
+    revoke_refresh_family,
+    revoked_family_key,
+)
 from enlace_auth.auth.sessions import SessionStore
 from enlace_auth.stores.validation import sanitize_key
 
@@ -301,6 +306,10 @@ def _norm_resource(resource: str) -> str:
     # One flag, consulted everywhere: a store with a zero TTL is NOT refresh
     # support, and metadata that says otherwise mints tokens dead on arrival.
     _refresh_enabled = refresh_store is not None and refresh_token_ttl > 0
+    _tombstone_ttl = refresh_tombstone_ttl(
+        refresh_token_ttl=refresh_token_ttl,
+        refresh_reuse_detection=refresh_reuse_detection,
+    )
     _local_claims: set[str] = set()
     _local_claim_lock = threading.Lock()
     _client_sweep_cursor: dict = {"pos": 0}
@@ -364,40 +373,9 @@ def _revoke_family(family: str, *, reason: str) -> int:
         """
         if refresh_store is None:
             return 0
-        # Write the marker FIRST. Revocation used to be expressed as the ABSENCE
-        # of records, which cannot work while another worker is concurrently
-        # creating them: a successor written after the scan snapshot survived,
-        # and a tombstone write resurrected a parent the scan had just deleted,
-        # so a chain that happened to be rotating defeated detection outright
-        # (measured at ~22% of attempts). A single positive fact cannot be raced.
-        now = _now()
-        refresh_store[_revoked_key(family)] = {
-            "revoked_at": now,
-            "reason": reason,
-            # Must outlive every token that could still belong to this family.
-            "exp": now + max(refresh_token_ttl, refresh_reuse_detection),
-        }
-        revoked = 0
-        for key in list(refresh_store):
-            try:
-                record = refresh_store[key]
-            except KeyError:
-                continue
-            if (record or {}).get("family") != family:
-                continue  # (the marker itself carries no "family" key)
-            try:
-                del refresh_store[key]
-                revoked += 1
-            except KeyError:
-                pass
-        _logger.warning(
-            "oauth: revoked refresh family %s (%d token(s)) — %s. The connector "
-            "using it is now dead until a human re-authorizes it.",
-            family,
-            revoked,
-            reason,
+        return revoke_refresh_family(
+            refresh_store, family, reason=reason, tombstone_ttl=_tombstone_ttl
         )
-        return revoked
 
     def _issue_refresh(
         *,
@@ -439,7 +417,7 @@ def _touch_client(client_id: str, now: int) -> None:
 
     def _revoked_key(family: str) -> str:
         """Store key for the tombstone that marks a whole family revoked."""
-        return f"family:{family}"
+        return revoked_family_key(family)
 
     def _family_revoked(family: Optional[str]) -> bool:
         return bool(family) and refresh_store.get(_revoked_key(family)) is not None
diff --git a/enlace_auth/auth/revocation.py b/enlace_auth/auth/revocation.py
new file mode 100644
index 0000000..a6171b1
--- /dev/null
+++ b/enlace_auth/auth/revocation.py
@@ -0,0 +1,219 @@
+"""Credential revocation: end everything an account holds when its credentials change.
+
+An account's access outlives a password change in three places, each with its own
+lifetime: browser **sessions** (``session_max_age``), OAuth connector
+**refresh-token families** (``refresh_family_max_lifetime``), and the per-app
+**shared-password cookies** (which are not per account at all -- see
+:func:`shared_password_fingerprint`). This module is the one place that knows how
+to end the first two, so every path that changes an account's credentials
+(admin delete, admin password set, self-service change, reset-link redemption,
+the ``set-password`` CLI) calls a single :func:`make_on_credentials_changed` hook
+instead of each remembering its own list.
+
+Kept free of FastAPI/Authlib imports so the CLI can use it without the
+``[oauth]`` extra.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import hmac
+import logging
+import time
+from collections.abc import MutableMapping
+from typing import Any, Optional, Protocol
+
+_logger = logging.getLogger("enlace_auth")
+
+__all__ = [
+    "CredentialsChanged",
+    "make_on_credentials_changed",
+    "refresh_tombstone_ttl",
+    "revoke_refresh_family",
+    "revoke_refresh_subject",
+    "revoked_family_key",
+    "shared_cookie_valid",
+    "shared_password_fingerprint",
+]
+
+
+class CredentialsChanged(Protocol):
+    """``on_credentials_changed(email, *, keep=None) -> None``.
+
+    *keep* names one browser session id to spare (the browser that just changed
+    its own password stays signed in).
+    """
+
+    def __call__(self, email: str, *, keep: Optional[str] = None) -> None: ...
+
+
+def revoked_family_key(family: str) -> str:
+    """Store key of the tombstone that marks a whole refresh family revoked."""
+    return f"family:{family}"
+
+
+def refresh_tombstone_ttl(
+    *, refresh_token_ttl: int, refresh_reuse_detection: int
+) -> int:
+    """How long a family tombstone must live: past every token of the family."""
+    return max(refresh_token_ttl, refresh_reuse_detection)
+
+
+def revoke_refresh_family(
+    refresh_store: MutableMapping[str, Any],
+    family: str,
+    *,
+    reason: str,
+    tombstone_ttl: int,
+    now: Optional[int] = None,
+) -> int:
+    """Revoke one refresh-token family; return how many token records went.
+
+    The tombstone is written FIRST: revocation expressed only as the absence of
+    records loses to a worker concurrently rotating the family (its successor is
+    written after our scan). A positive marker cannot be raced -- the refresh
+    grant refuses any record whose family carries one.
+    """
+    now = int(time.time()) if now is None else now
+    refresh_store[revoked_family_key(family)] = {
+        "revoked_at": now,
+        "reason": reason,
+        "exp": now + tombstone_ttl,
+    }
+    revoked = 0
+    for key in list(refresh_store):
+        try:
+            record = refresh_store[key]
+        except KeyError:
+            continue
+        if (record or {}).get("family") != family:
+            continue  # (the marker itself carries no "family" key)
+        try:
+            del refresh_store[key]
+            revoked += 1
+        except KeyError:
+            pass
+    _logger.warning(
+        "oauth: revoked refresh family %s (%d token(s)) — %s. The connector "
+        "using it is now dead until a human re-authorizes it.",
+        family,
+        revoked,
+        reason,
+    )
+    return revoked
+
+
+def revoke_refresh_subject(
+    refresh_store: MutableMapping[str, Any],
+    email: str,
+    *,
+    reason: str,
+    tombstone_ttl: int,
+    code_store: Optional[MutableMapping[str, Any]] = None,
+    now: Optional[int] = None,
+) -> int:
+    """Revoke every refresh family issued to *email*; return how many families.
+
+    Matches the subject case-insensitively. Also drops the subject's unredeemed
+    authorization codes from *code_store* when given, so a code minted just
+    before the change cannot start a fresh family after it. Access JWTs already
+    issued are self-contained and live out their (short) TTL.
+    """
+    if not email:
+        raise ValueError("revoke_refresh_subject needs a non-empty email")
+    target = email.lower()
+
+    def _is_subjects(record: Any) -> bool:
+        subject = (record or {}).get("email") if isinstance(record, dict) else None
+        return isinstance(subject, str) and subject.lower() == target
+
+    families: set[str] = set()
+    for key in list(refresh_store):
+        try:
+            record = refresh_store[key]
+        except KeyError:
+            continue
+        if _is_subjects(record) and record.get("family"):
+            families.add(record["family"])
+    for family in sorted(families):
+        revoke_refresh_family(
+            refresh_store, family, reason=reason, tombstone_ttl=tombstone_ttl, now=now
+        )
+    if code_store is not None:
+        for key in list(code_store):
+            try:
+                record = code_store[key]
+            except KeyError:
+                continue
+            if _is_subjects(record):
+                try:
+                    del code_store[key]
+                except KeyError:
+                    pass
+    return len(families)
+
+
+def make_on_credentials_changed(
+    session_store,
+    *,
+    refresh_store: Optional[MutableMapping[str, Any]] = None,
+    code_store: Optional[MutableMapping[str, Any]] = None,
+    tombstone_ttl: int = 0,
+    reason: str = "the account's credentials changed",
+) -> CredentialsChanged:
+    """Return the hook every credential-changing path calls.
+
+    Always revokes the account's browser sessions (sparing *keep*); when a
+    *refresh_store* is given, also revokes the account's connector refresh
+    families (and unredeemed codes in *code_store*). A failure to revoke
+    connector families is logged loudly but does not undo the password change,
+    which has already been written by the caller.
+    """
+    if refresh_store is not None and tombstone_ttl <= 0:
+        raise ValueError("tombstone_ttl must be positive when refresh_store is given")
+
+    def on_credentials_changed(email: str, *, keep: Optional[str] = None) -> None:
+        session_store.revoke_user(email, keep=keep)
+        if refresh_store is None:
+            return
+        try:
+            revoke_refresh_subject(
+                refresh_store,
+                email,
+                reason=reason,
+                tombstone_ttl=tombstone_ttl,
+                code_store=code_store,
+            )
+        except Exception:  # noqa: BLE001 - sessions are already gone; say so loudly
+            _logger.exception(
+                "enlace_auth: could not revoke connector sessions for %r after a "
+                "credential change; revoke them with "
+                "`enlace-auth revoke-connector-session --email`",
+                email,
+            )
+
+    return on_credentials_changed
+
+
+def shared_password_fingerprint(password_hash: str, signing_key: str) -> str:
+    """A short keyed fingerprint of an app's CURRENT shared-password hash.
+
+    Signed into the ``shared_auth_<app>`` cookie and compared by the middleware,
+    so rotating the shared password invalidates every cookie minted under the
+    old one. Keyed with *signing_key* so the cookie (whose payload is readable,
+    only signed) reveals nothing about the hash.
+    """
+    digest = hmac.new(
+        signing_key.encode(), f"shared-pw:{password_hash}".encode(), hashlib.sha256
+    )
+    return digest.hexdigest()[:32]
+
+
+def shared_cookie_valid(
+    value: Optional[str], password_hash: Optional[str], signing_key: str
+) -> bool:
+    """True iff a verified shared-cookie *value* was minted under *password_hash*."""
+    if not value or not password_hash:
+        return False
+    expected = shared_password_fingerprint(password_hash, signing_key)
+    return hmac.compare_digest(str(value), expected)
diff --git a/enlace_auth/auth/routes.py b/enlace_auth/auth/routes.py
index 9a3e065..e4009d4 100644
--- a/enlace_auth/auth/routes.py
+++ b/enlace_auth/auth/routes.py
@@ -38,6 +38,12 @@
     reset_url,
     verify_reset_token,
 )
+from enlace_auth.auth.revocation import (
+    CredentialsChanged,
+    make_on_credentials_changed,
+    shared_cookie_valid,
+    shared_password_fingerprint,
+)
 from enlace_auth.auth.sessions import SessionStore
 
 
@@ -87,6 +93,7 @@ def make_auth_router(
     send_email: Optional[EmailSender] = None,
     reset_token_max_age: int = DEFAULT_EMAIL_TTL,
     public_base_url: Optional[str] = None,
+    on_credentials_changed: Optional[CredentialsChanged] = None,
 ) -> APIRouter:
     """Build a FastAPI router exposing ``/auth/*`` endpoints.
 
@@ -109,8 +116,16 @@ def make_auth_router(
             ``Host`` header, which the requester controls -- a forged ``Host``
             would mail the victim a link that hands their reset token to
             another site, unless a proxy in front only forwards known hosts.
+        on_credentials_changed: ``hook(email, *, keep=None)`` called after a
+            password change or reset. Defaults to revoking the account's
+            browser sessions only; the plugin injects one that also revokes the
+            account's OAuth connector refresh families (see
+            ``enlace_auth.auth.revocation``).
     """
     router = APIRouter(prefix="/auth")
+    credentials_changed: CredentialsChanged = (
+        on_credentials_changed or make_on_credentials_changed(session_store)
+    )
     # Distinguish "no delivery channel configured" from "a sender was wired":
     # the page copy must not promise an email the platform cannot send.
     email_delivery_configured = send_email is not None
@@ -230,7 +245,13 @@ async def shared_login_page(request: Request):
                 status_code=404,
             )
         existing = request.cookies.get(f"shared_auth_{app}")
-        if existing and verify_cookie(existing, signing_key, salt=f"shared:{app}"):
+        if existing and shared_cookie_valid(
+            verify_cookie(
+                existing, signing_key, max_age=session_max_age, salt=f"shared:{app}"
+            ),
+            shared_password_for(app),
+            signing_key,
+        ):
             return RedirectResponse(next_url, status_code=303)
         return HTMLResponse(pages.render_shared_login_page(app=app, next_url=next_url))
 
@@ -243,7 +264,13 @@ async def shared_login(
             raise HTTPException(status_code=404, detail=f"Unknown app '{body.app}'")
         if not verify_password(stored_hash, body.password):
             raise HTTPException(status_code=401, detail="Invalid password")
-        token = sign_cookie("1", signing_key, salt=f"shared:{body.app}")
+        # The cookie carries a fingerprint of the CURRENT password hash, so
+        # rotating the shared password ends every cookie minted under the old.
+        token = sign_cookie(
+            shared_password_fingerprint(stored_hash, signing_key),
+            signing_key,
+            salt=f"shared:{body.app}",
+        )
         cookie_name_shared = f"shared_auth_{body.app}"
         attrs = [
             f"{cookie_name_shared}={token}",
@@ -291,7 +318,7 @@ async def change_own_password(
         user_store[email] = record
         # Log out every OTHER browser holding this account: a password change
         # is what a user does when they suspect someone else is signed in.
-        session_store.revoke_user(email, keep=_current_session_id(request))
+        credentials_changed(email, keep=_current_session_id(request))
         return {"ok": True, "email": email}
 
     # ----- Password recovery ---------------------------------------------
@@ -425,7 +452,7 @@ async def password_reset_confirm(
         user_store[email] = record
         # A reset is the recovery path for a compromised account, so every
         # session opened with the old password ends here.
-        session_store.revoke_user(email)
+        credentials_changed(email)
         session_id = session_store.create(user_id=email, email=email)
         _set_session_cookie(
             response,
diff --git a/enlace_auth/auth/sessions.py b/enlace_auth/auth/sessions.py
index 0c0b6f6..1ce49fd 100644
--- a/enlace_auth/auth/sessions.py
+++ b/enlace_auth/auth/sessions.py
@@ -2,6 +2,11 @@
 
 A session is ``{"user_id": str, "email": str | None, "created_at": float}``.
 Session IDs are 32-byte urlsafe tokens. Revocation is a simple delete.
+
+Records are otherwise only deleted by logout, so given a *max_age* the store
+sweeps records older than it, a bounded batch at a time, whenever a session is
+created -- keeping the store (and :meth:`SessionStore.revoke_user`'s scan) from
+growing without bound.
 """
 
 from __future__ import annotations
@@ -9,16 +14,55 @@
 import secrets
 import time
 from collections.abc import MutableMapping
+from itertools import islice
 from typing import Any, Optional
 
 
 class SessionStore:
     """Thin adapter around a MutableMapping that speaks session semantics."""
 
-    def __init__(self, store: MutableMapping):
+    def __init__(
+        self,
+        store: MutableMapping,
+        *,
+        max_age: Optional[float] = None,
+        sweep_batch: int = 100,
+    ):
         self._store = store
+        self._max_age = max_age
+        self._sweep_batch = sweep_batch
+        self._sweep_pos = 0
+
+    def sweep_expired(self, *, now: Optional[float] = None) -> int:
+        """Delete up to *sweep_batch* records older than *max_age*; return count.
+
+        No-op without a *max_age*. A cursor carries the position across calls
+        (wrapping at the end) so successive sweeps walk the whole store. A
+        record without a numeric ``created_at`` is left alone.
+        """
+        if not self._max_age:
+            return 0
+        now = time.time() if now is None else now
+        start = self._sweep_pos
+        keys = list(islice(iter(self._store), start, start + self._sweep_batch))
+        self._sweep_pos = 0 if len(keys) < self._sweep_batch else start + len(keys)
+        removed = 0
+        for sid in keys:
+            try:
+                record = self._store[sid]
+            except KeyError:
+                continue
+            created = record.get("created_at") if isinstance(record, dict) else None
+            if isinstance(created, (int, float)) and now - created > self._max_age:
+                if self.delete(sid):
+                    removed += 1
+        return removed
 
     def create(self, user_id: str, email: Optional[str] = None) -> str:
+        try:
+            self.sweep_expired()
+        except Exception:  # noqa: BLE001 - housekeeping must never block a login
+            pass
         session_id = secrets.token_urlsafe(32)
         self._store[session_id] = {
             "user_id": user_id,
diff --git a/enlace_auth/plugin.py b/enlace_auth/plugin.py
index 360905f..f0995ff 100644
--- a/enlace_auth/plugin.py
+++ b/enlace_auth/plugin.py
@@ -334,7 +334,9 @@ def wire(parent: "FastAPI", config) -> None:
     platform_factory = make_file_store_factory(auth_cfg.stores.path)
     session_backend = platform_factory("sessions")
     user_backend = platform_factory("users")
-    session_store = SessionStore(session_backend)
+    session_store = SessionStore(
+        session_backend, max_age=auth_cfg.session_max_age_seconds
+    )
 
     # Runtime per-app access grants (additive, optional UTC expiry). Lives beside
     # sessions/ and users/ under the same persistent store root — outside the
@@ -417,6 +419,31 @@ def wire(parent: "FastAPI", config) -> None:
     # beneath a more specific prefix via longest-prefix match.
     access_rules.append(AccessRule(prefix="/", level="public", app_id="_root"))
 
+    # One hook for every path that changes an account's credentials: browser
+    # sessions always go; the account's OAuth connector refresh families go too
+    # when the OAuth server keeps them (i2mint/enlace_auth#26).
+    from enlace_auth.auth.revocation import (
+        make_on_credentials_changed,
+        refresh_tombstone_ttl,
+    )
+
+    osc = auth_cfg.oauth_server
+    _connector_revocation = (
+        dict(
+            refresh_store=platform_factory("oauth_refresh_tokens"),
+            code_store=platform_factory("oauth_codes"),
+            tombstone_ttl=refresh_tombstone_ttl(
+                refresh_token_ttl=osc.refresh_token_ttl_seconds,
+                refresh_reuse_detection=osc.refresh_reuse_detection_seconds,
+            ),
+        )
+        if osc.enabled and osc.refresh_token_ttl_seconds > 0
+        else {}
+    )
+    on_credentials_changed = make_on_credentials_changed(
+        session_store, **_connector_revocation
+    )
+
     auth_router = make_auth_router(
         session_store=session_store,
         user_store=user_backend,
@@ -428,6 +455,7 @@ def wire(parent: "FastAPI", config) -> None:
         can_register=can_register,
         send_email=email_sender,
         public_base_url=_public_base_url(config, auth_cfg),
+        on_credentials_changed=on_credentials_changed,
     )
     parent.include_router(auth_router)
 
@@ -559,6 +587,7 @@ def wire(parent: "FastAPI", config) -> None:
         signing_key=signing_key,
         resource_allowlist=auth_cfg.oauth_server.resource_allowlist,
         public_base_url=_public_base_url(config, auth_cfg),
+        on_credentials_changed=on_credentials_changed,
     )
     parent.include_router(admin_router)
     if admin_emails:
diff --git a/tests/test_auth_middleware.py b/tests/test_auth_middleware.py
index 054808f..5f731bb 100644
--- a/tests/test_auth_middleware.py
+++ b/tests/test_auth_middleware.py
@@ -203,12 +203,45 @@ def test_longest_prefix_wins():
 
 
 def test_protected_shared_with_valid_cookie():
+    # A valid shared cookie carries the fingerprint of the app's CURRENT
+    # shared-password hash (i2mint/enlace_auth#26), not a constant.
+    from enlace_auth.auth.revocation import shared_password_fingerprint
+
+    rule = AccessRule(
+        prefix="/s", level="protected:shared", app_id="s", shared_password_hash="h1"
+    )
+    mw = _make_mw([rule])
+    token = sign_cookie(
+        shared_password_fingerprint("h1", SIGNING_KEY), SIGNING_KEY, salt="shared:s"
+    )
+    cap = _Capture()
+    _run(mw(_http_scope("/s/page", {"shared_auth_s": token}), cap.receive, cap.send))
+    assert cap.status() == 200
+
+
+def test_protected_shared_cookie_minted_under_another_password_denied():
+    from enlace_auth.auth.revocation import shared_password_fingerprint
+
+    rule = AccessRule(
+        prefix="/s", level="protected:shared", app_id="s", shared_password_hash="h2"
+    )
+    mw = _make_mw([rule])
+    for value in ("1", shared_password_fingerprint("h1", SIGNING_KEY)):
+        token = sign_cookie(value, SIGNING_KEY, salt="shared:s")
+        cap = _Capture()
+        _run(
+            mw(_http_scope("/s/page", {"shared_auth_s": token}), cap.receive, cap.send)
+        )
+        assert cap.status() == 401, value
+
+
+def test_protected_shared_without_configured_password_admits_no_cookie():
     rule = AccessRule(prefix="/s", level="protected:shared", app_id="s")
     mw = _make_mw([rule])
     token = sign_cookie("1", SIGNING_KEY, salt="shared:s")
     cap = _Capture()
     _run(mw(_http_scope("/s/page", {"shared_auth_s": token}), cap.receive, cap.send))
-    assert cap.status() == 200
+    assert cap.status() == 401
 
 
 def test_protected_shared_without_cookie_denied():
diff --git a/tests/test_credential_revocation.py b/tests/test_credential_revocation.py
new file mode 100644
index 0000000..7786972
--- /dev/null
+++ b/tests/test_credential_revocation.py
@@ -0,0 +1,280 @@
+"""Credential changes end connector sessions and shared-password cookies too.
+
+i2mint/enlace_auth#26: #25 made a password change revoke an account's browser
+sessions; these tests pin the two credential kinds it left alive -- the
+account's OAuth connector refresh-token families, and ``shared_auth_<app>``
+cookies minted under a shared password that has since been rotated -- plus the
+bounded session sweep.
+"""
+
+from __future__ import annotations
+
+import time
+
+import pytest
+from enlace.base import PlatformConfig
+from enlace.compose import build_backend
+from enlace.discover import discover_apps
+from starlette.testclient import TestClient
+
+from enlace_auth import plugin as auth_plugin
+from enlace_auth.auth import hash_password
+from enlace_auth.auth.revocation import (
+    make_on_credentials_changed,
+    revoke_refresh_subject,
+    revoked_family_key,
+    shared_cookie_valid,
+    shared_password_fingerprint,
+)
+from enlace_auth.auth.sessions import SessionStore
+from enlace_auth.stores import make_file_store_factory
+from tests.test_admin import (
+    _SIGNING_KEY,
+    _csrf,
+    _login,
+    _register,
+    _write_dummy_app,
+)
+from tests.test_auth_e2e import _make_apps
+
+
+def _refresh_record(family: str, email: str) -> dict:
+    now = int(time.time())
+    return {
+        "family": family,
+        "client_id": "c",
+        "resource": "https://x/api/mcp",
+        "scope": "mcp:read",
+        "email": email,
+        "iat": now,
+        "exp": now + 3600,
+        "family_exp": now + 7200,
+        "consumed_at": None,
+        "successor": None,
+    }
+
+
+# ---- unit: revocation helpers ------------------------------------------------
+
+
+def test_revoke_refresh_subject_tombstones_only_that_subjects_families():
+    store = {
+        "t1": _refresh_record("f1", "Vic@Example.com"),
+        "t2": {**_refresh_record("f1", "vic@example.com"), "consumed_at": 1},
+        "t3": _refresh_record("f2", "vic@example.com"),
+        "t4": _refresh_record("f3", "bob@example.com"),
+    }
+    codes = {
+        "c1": {"email": "VIC@example.com", "exp": 1},
+        "c2": {"email": "bob@example.com", "exp": 1},
+    }
+    n = revoke_refresh_subject(
+        store, "vic@EXAMPLE.com", reason="t", tombstone_ttl=100, code_store=codes
+    )
+    assert n == 2
+    assert revoked_family_key("f1") in store and revoked_family_key("f2") in store
+    live = {k for k, v in store.items() if "family" in v}
+    assert live == {"t4"}, "bob's family must survive"
+    assert set(codes) == {"c2"}, "vic's unredeemed codes must go"
+
+
+def test_revoke_refresh_subject_rejects_empty_email():
+    with pytest.raises(ValueError):
+        revoke_refresh_subject({}, "", reason="t", tombstone_ttl=1)
+
+
+def test_hook_revokes_sessions_and_families_and_spares_keep():
+    sessions = SessionStore({})
+    keep = sessions.create("vic@example.com", "vic@example.com")
+    other = sessions.create("vic@example.com", "vic@example.com")
+    refresh = {"t1": _refresh_record("f1", "vic@example.com")}
+    hook = make_on_credentials_changed(
+        sessions, refresh_store=refresh, tombstone_ttl=10
+    )
+    hook("vic@example.com", keep=keep)
+    assert sessions.get(keep) is not None
+    assert sessions.get(other) is None
+    assert "t1" not in refresh and revoked_family_key("f1") in refresh
+
+
+def test_hook_requires_a_tombstone_ttl_with_a_refresh_store():
+    with pytest.raises(ValueError):
+        make_on_credentials_changed(SessionStore({}), refresh_store={})
+
+
+def test_shared_cookie_fingerprint_binds_the_current_hash():
+    fp = shared_password_fingerprint("h1", "k")
+    assert shared_cookie_valid(fp, "h1", "k")
+    assert not shared_cookie_valid(fp, "h2", "k")
+    assert not shared_cookie_valid("1", "h1", "k"), "legacy constant cookie"
+    assert not shared_cookie_valid(fp, None, "k")
+    assert "h1" not in fp
+
+
+def test_session_sweep_drops_only_expired_records():
+    backend: dict = {}
+    sessions = SessionStore(backend, max_age=60, sweep_batch=10)
+    backend["old"] = {"user_id": "a", "email": "a", "created_at": time.time() - 120}
+    backend["legacy"] = {"user_id": "a", "email": "a"}  # no created_at: kept
+    fresh = sessions.create("b", "b")  # create() sweeps
+    assert "old" not in backend
+    assert "legacy" in backend and fresh in backend
+
+
+def test_session_sweep_is_bounded_and_walks_the_whole_store():
+    backend = {
+        f"s{i}": {"user_id": "a", "created_at": time.time() - 999} for i in range(25)
+    }
+    sessions = SessionStore(backend, max_age=60, sweep_batch=10)
+    assert sessions.sweep_expired() == 10
+    while backend:
+        assert sessions.sweep_expired() <= 10
+    assert not backend
+
+
+def test_session_sweep_is_a_noop_without_max_age():
+    backend = {"old": {"user_id": "a", "created_at": 0}}
+    SessionStore(backend).create("b")
+    assert "old" in backend
+
+
+# ---- end to end through the plugin -------------------------------------------
+
+
+@pytest.fixture
+def platform(tmp_path, monkeypatch):
+    """A gateway with the OAuth server (and so refresh families) enabled."""
+    apps_dir = tmp_path / "apps"
+    apps_dir.mkdir()
+    _write_dummy_app(apps_dir)
+    monkeypatch.setenv("ENLACE_SIGNING_KEY", _SIGNING_KEY)
+    monkeypatch.setenv("ENLACE_ADMIN_EMAILS", "boss@example.com")
+    store_root = tmp_path / "platform"
+    config = PlatformConfig(
+        apps_dir=apps_dir,
+        auth={
+            "enabled": True,
+            "secure_cookies": False,
+            "registration_open": True,
+            "stores": {"backend": "file", "path": str(store_root)},
+            "oauth_server": {"enabled": True, "key_dir": str(tmp_path / "keys")},
+        },
+    )
+    app = build_backend(discover_apps(config), plugins=[auth_plugin])
+    factory = make_file_store_factory(str(store_root))
+    return TestClient(app), factory("oauth_refresh_tokens"), factory("oauth_codes")
+
+
+def _seed(refresh, codes):
+    refresh["tok-vic"] = _refresh_record("fvic", "vic@example.com")
+    refresh["tok-bob"] = _refresh_record("fbob", "bob@example.com")
+    codes["code-vic"] = {"email": "vic@example.com", "exp": int(time.time()) + 60}
+
+
+def _families_left(refresh) -> set:
+    out = set()
+    for k in refresh:
+        rec = refresh[k]
+        if "family" in rec:
+            out.add(rec["family"])
+    return out
+
+
+def test_admin_password_set_revokes_connector_families(platform):
+    client, refresh, codes = platform
+    csrf = _csrf(client)
+    _register(client, "boss@example.com", "bosspw1!", csrf)
+    r = client.post(
+        "/_admin/api/users",
+        json={"email": "vic@example.com", "password": "victim-pw1"},
+        headers=csrf,
+    )
+    assert r.status_code == 200, r.text
+    _seed(refresh, codes)
+    r = client.post(
+        "/_admin/api/users/vic@example.com/password",
+        json={"password": "brand-new-pw1"},
+        headers=csrf,
+    )
+    assert r.status_code == 200, r.text
+    assert _families_left(refresh) == {"fbob"}
+    assert refresh.get(revoked_family_key("fvic")) is not None
+    assert "code-vic" not in codes
+
+
+def test_admin_delete_revokes_connector_families(platform):
+    client, refresh, codes = platform
+    csrf = _csrf(client)
+    _register(client, "boss@example.com", "bosspw1!", csrf)
+    client.post(
+        "/_admin/api/users",
+        json={"email": "vic@example.com", "password": "victim-pw1"},
+        headers=csrf,
+    )
+    _seed(refresh, codes)
+    r = client.delete("/_admin/api/users/vic@example.com", headers=csrf)
+    assert r.status_code == 200, r.text
+    assert _families_left(refresh) == {"fbob"}
+
+
+def test_self_service_password_change_revokes_connector_families(platform):
+    client, refresh, codes = platform
+    csrf = _csrf(client)
+    _register(client, "vic@example.com", "victim-pw1", csrf)
+    assert _login(client, "vic@example.com", "victim-pw1", csrf).status_code == 200
+    _seed(refresh, codes)
+    r = client.post(
+        "/auth/me/password",
+        json={"old_password": "victim-pw1", "new_password": "brand-new-pw1"},
+        headers=_csrf(client),
+    )
+    assert r.status_code == 200, r.text
+    assert _families_left(refresh) == {"fbob"}
+    assert client.get("/auth/whoami").json()["email"] == "vic@example.com", (
+        "the browser that changed its own password stays signed in"
+    )
+
+
+def _shared_platform(tmp_path, monkeypatch, password: str) -> TestClient:
+    apps_dir = tmp_path / "apps"
+    apps_dir.mkdir(exist_ok=True)
+    if not (apps_dir / "shared_app").exists():
+        _make_apps(apps_dir)
+    monkeypatch.setenv("ENLACE_SIGNING_KEY", _SIGNING_KEY)
+    monkeypatch.setenv("SHARED_APP_PW", hash_password(password))
+    config = PlatformConfig(
+        apps_dir=apps_dir,
+        auth={
+            "enabled": True,
+            "secure_cookies": False,
+            "stores": {"backend": "file", "path": str(tmp_path / "platform")},
+        },
+    )
+    return TestClient(build_backend(discover_apps(config), plugins=[auth_plugin]))
+
+
+def test_rotating_a_shared_password_ends_old_cookies(tmp_path, monkeypatch):
+    before = _shared_platform(tmp_path, monkeypatch, "open-sesame")
+    csrf = _csrf_shared(before)
+    r = before.post(
+        "/auth/shared-login",
+        json={"app": "shared_app", "password": "open-sesame"},
+        headers=csrf,
+    )
+    assert r.status_code == 200, r.text
+    cookie = before.cookies.get("shared_auth_shared_app")
+    assert cookie
+    assert before.get("/api/shared_app/peek").status_code == 200
+
+    # Restart with a rotated password: the old cookie must no longer work.
+    after = _shared_platform(tmp_path, monkeypatch, "new-sesame")
+    after.cookies.set("shared_auth_shared_app", cookie)
+    assert after.get("/api/shared_app/peek").status_code == 401
+
+
+def _csrf_shared(client: TestClient) -> dict:
+    from enlace_auth.auth.cookies import verify_cookie
+
+    client.get("/api/public_app/ping")
+    raw = verify_cookie(client.cookies.get("enlace_csrf"), _SIGNING_KEY, salt="csrf")
+    return {"X-CSRF-Token": raw}

From 64164e540bbf01fcb931e27b7d46da14a30c4c3a Mon Sep 17 00:00:00 2001
From: Thor Whalen <1906276+thorwhalen@users.noreply.github.com>
Date: Tue, 22 Sep 2026 16:27:08 +0000
Subject: [PATCH 2/2] Close the revocation races found in review; throttle the
 session sweep

- revoke_refresh_subject first writes a per-subject marker (revoked_before=T,
  lives for the family max lifetime). The code grant refuses codes issued at or
  before T (codes now record when the session was read), and the refresh grant
  and replay path refuse families authorized at or before T (records carry
  auth_at). This covers a family whose code was consumed before the scan but
  written after it, and a code written after the code-store scan.
- The plugin wires connector revocation even while the OAuth server is off.
- The hook runs connector revocation even if session revocation raises.
- set-password reports clearly if connector revocation fails.
- list-connector-sessions skips revocation markers.
- SessionStore sweeps at most once per sweep_interval (default 1h) per
  process, so a login never pays a directory walk.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
---
 enlace_auth/__main__.py             |  14 +++-
 enlace_auth/auth/oauth_server.py    |  44 ++++++++++-
 enlace_auth/auth/revocation.py      |  86 +++++++++++++++------
 enlace_auth/auth/sessions.py        |  22 ++++--
 enlace_auth/plugin.py               |   9 ++-
 tests/test_credential_revocation.py | 112 ++++++++++++++++++++++++++++
 6 files changed, 254 insertions(+), 33 deletions(-)

diff --git a/enlace_auth/__main__.py b/enlace_auth/__main__.py
index 5482183..e1e6cf6 100644
--- a/enlace_auth/__main__.py
+++ b/enlace_auth/__main__.py
@@ -118,6 +118,7 @@ def _revoke_connector_subject(email: str, toml_path: Path) -> int:
         email,
         reason="revoked by the enlace-auth CLI",
         tombstone_ttl=_connector_tombstone_ttl(toml_path),
+        marker_ttl=auth.oauth_server.refresh_family_max_lifetime_seconds,
         code_store=factory("oauth_codes"),
     )
 
@@ -255,7 +256,16 @@ def set_password(email: str, *, toml: str = "platform.toml"):
     store[key] = updated
     # Same rule as the HTTP reset paths: a new password ends the old sessions.
     revoked = _load_session_store(Path(toml)).revoke_user(key)
-    families = _revoke_connector_subject(key, Path(toml))
+    try:
+        families = _revoke_connector_subject(key, Path(toml))
+    except Exception as e:  # noqa: BLE001 - the password IS changed; say what isn't
+        print(
+            f"Password updated for {key}; {revoked} existing session(s) revoked, "
+            f"but connector sessions were NOT revoked ({e}). Run "
+            f"`enlace-auth revoke-connector-session --email {key}`.",
+            file=sys.stderr,
+        )
+        sys.exit(1)
     print(
         f"Password updated for {key}; {revoked} existing session(s) and "
         f"{families} connector session(s) revoked."
@@ -444,6 +454,8 @@ def list_connector_sessions(*, json: bool = False, toml: str = "platform.toml"):
             continue
         if not record or record.get("consumed_at") is not None:
             continue  # spent tokens are tombstones, not sessions
+        if not record.get("family") or not record.get("email"):
+            continue  # revocation markers, not sessions
         families[record.get("family", key)] = {
             "family": record.get("family"),
             "email": record.get("email"),
diff --git a/enlace_auth/auth/oauth_server.py b/enlace_auth/auth/oauth_server.py
index 64665a6..0c4cd59 100644
--- a/enlace_auth/auth/oauth_server.py
+++ b/enlace_auth/auth/oauth_server.py
@@ -61,6 +61,7 @@
     refresh_tombstone_ttl,
     revoke_refresh_family,
     revoked_family_key,
+    subject_revoked_before,
 )
 from enlace_auth.auth.sessions import SessionStore
 from enlace_auth.stores.validation import sanitize_key
@@ -402,6 +403,8 @@ def _issue_refresh(
             "scope": scope,
             "email": email,
             "iat": now,
+            # When the family was authorized; carried through every rotation.
+            "auth_at": now,
             "exp": now + refresh_token_ttl,
             "family_exp": family_exp,
             "consumed_at": None,
@@ -422,6 +425,18 @@ def _revoked_key(family: str) -> str:
     def _family_revoked(family: Optional[str]) -> bool:
         return bool(family) and refresh_store.get(_revoked_key(family)) is not None
 
+    def _authorized_before_revocation(record: dict) -> bool:
+        """True if the subject's credentials changed after this family began."""
+        revoked_before = subject_revoked_before(refresh_store, record.get("email"))
+        if revoked_before is None:
+            return False
+        auth_at = record.get("auth_at")
+        if not isinstance(auth_at, (int, float)):
+            # Records minted before ``auth_at`` existed: derive it from the
+            # absolute ceiling, which is fixed at authorization.
+            auth_at = (record.get("family_exp") or 0) - refresh_family_max_lifetime
+        return auth_at <= revoked_before
+
     def _grace_key(key: str) -> str:
         """Store key for the short-lived retry copy of a successor plaintext."""
         return f"grace:{key}"
@@ -640,9 +655,15 @@ def _validate_authorize(
         )
         return auth, None
 
-    def _issue_code(auth: _Authorized, email: str) -> str:
+    def _issue_code(auth: _Authorized, email: str, *, issued_at: int) -> str:
+        """Mint a code; *issued_at* is when the session was read, not written.
+
+        A credential change between reading the session and writing the code
+        would otherwise miss this code (see ``subject_revoked_before``).
+        """
         code = secrets.token_urlsafe(32)
         code_store[code] = {
+            "iat": issued_at,
             "client_id": auth.client_id,
             "email": email,
             "redirect_uri": auth.redirect_uri,
@@ -668,6 +689,7 @@ async def authorize(request: Request):
                 auth.redirect_uri, "invalid_request", auth.state, "PKCE S256 required"
             )
 
+        session_read_at = _now()
         email = _current_email(request)
         if not email:
             # Reuse the platform login, returning here once authenticated.
@@ -680,7 +702,7 @@ async def authorize(request: Request):
             return HTMLResponse(_denied_page(email), status_code=403)
 
         if not require_consent:
-            code = _issue_code(auth, email)
+            code = _issue_code(auth, email, issued_at=session_read_at)
             return RedirectResponse(
                 _with_query(auth.redirect_uri, {"code": code, "state": auth.state}),
                 status_code=302,
@@ -703,6 +725,7 @@ async def authorize_consent(
         csrf: str = Form(...),
         decision: str = Form(...),
     ):
+        session_read_at = _now()
         email = _current_email(request)
         if not email:
             return JSONResponse({"error": "login_required"}, status_code=401)
@@ -726,7 +749,7 @@ async def authorize_consent(
             return _redirect_error(redirect_uri, "access_denied", state)
         if decision != "approve":
             return _redirect_error(redirect_uri, "access_denied", state)
-        code = _issue_code(auth, email)
+        code = _issue_code(auth, email, issued_at=session_read_at)
         return RedirectResponse(
             _with_query(redirect_uri, {"code": code, "state": state}),
             status_code=302,
@@ -847,6 +870,12 @@ def _grant_authorization_code(
             or not _verify_pkce_s256(code_verifier, data["code_challenge"])
         ):
             return JSONResponse({"error": "invalid_grant"}, status_code=400)
+        # Issued before the account's credentials changed: dead, even if the
+        # revocation's scan could not see it (i2mint/enlace_auth#26).
+        revoked_before = subject_revoked_before(refresh_store, data["email"])
+        code_iat = data.get("iat", data["exp"] - code_ttl)
+        if revoked_before is not None and code_iat <= revoked_before:
+            return JSONResponse({"error": "invalid_grant"}, status_code=400)
 
         body, _ = _token_payload(
             iss=_issuer(request),
@@ -887,6 +916,11 @@ def _grant_refresh_token(
         # that was meant to delete it.
         if _family_revoked(record.get("family")):
             return JSONResponse({"error": "invalid_grant"}, status_code=400)
+        if _authorized_before_revocation(record):
+            _revoke_family(
+                record.get("family"), reason="the account's credentials changed"
+            )
+            return JSONResponse({"error": "invalid_grant"}, status_code=400)
 
         if record.get("consumed_at") is not None:
             return _replayed(
@@ -1015,6 +1049,7 @@ def _rotate(
             "scope": scope,
             "email": email,
             "iat": now,
+            "auth_at": record.get("auth_at"),
             "exp": now + refresh_token_ttl,
             "family_exp": family_exp,
             "consumed_at": None,
@@ -1084,6 +1119,9 @@ def _replayed(
         family = record.get("family")
         if _family_revoked(family):
             return JSONResponse({"error": "invalid_grant"}, status_code=400)
+        if _authorized_before_revocation(record):
+            _revoke_family(family, reason="the account's credentials changed")
+            return JSONResponse({"error": "invalid_grant"}, status_code=400)
         within_grace = now - (record.get("consumed_at") or 0) < refresh_reuse_grace
         same_client = record.get("client_id") == client_id
 
diff --git a/enlace_auth/auth/revocation.py b/enlace_auth/auth/revocation.py
index a6171b1..60a5f3d 100644
--- a/enlace_auth/auth/revocation.py
+++ b/enlace_auth/auth/revocation.py
@@ -32,6 +32,8 @@
     "revoke_refresh_family",
     "revoke_refresh_subject",
     "revoked_family_key",
+    "subject_marker_key",
+    "subject_revoked_before",
     "shared_cookie_valid",
     "shared_password_fingerprint",
 ]
@@ -52,6 +54,34 @@ def revoked_family_key(family: str) -> str:
     return f"family:{family}"
 
 
+def subject_marker_key(email: str) -> str:
+    """Store key of the marker that says "nothing *email* authorized before T"."""
+    digest = hashlib.sha256(email.lower().encode()).hexdigest()
+    return f"subject:{digest}"
+
+
+def subject_revoked_before(
+    refresh_store: Optional[MutableMapping[str, Any]], email: Optional[str]
+) -> Optional[int]:
+    """The time before which every authorization by *email* is revoked, if any.
+
+    A scan-and-delete revocation cannot see a family that another worker is
+    creating at that very moment (its code already consumed, its first refresh
+    record not yet written), nor a code issued from a session read just before
+    the change. The marker closes both: the code grant refuses codes issued at
+    or before it, and the refresh grant refuses families authorized at or
+    before it.
+    """
+    if refresh_store is None or not email:
+        return None
+    try:
+        record = refresh_store.get(subject_marker_key(email))
+    except Exception:  # noqa: BLE001 - an unreadable marker is no marker
+        return None
+    value = (record or {}).get("revoked_before") if isinstance(record, dict) else None
+    return value if isinstance(value, (int, float)) else None
+
+
 def refresh_tombstone_ttl(
     *, refresh_token_ttl: int, refresh_reuse_detection: int
 ) -> int:
@@ -110,18 +140,27 @@ def revoke_refresh_subject(
     reason: str,
     tombstone_ttl: int,
     code_store: Optional[MutableMapping[str, Any]] = None,
+    marker_ttl: int = 0,
     now: Optional[int] = None,
 ) -> int:
     """Revoke every refresh family issued to *email*; return how many families.
 
-    Matches the subject case-insensitively. Also drops the subject's unredeemed
-    authorization codes from *code_store* when given, so a code minted just
-    before the change cannot start a fresh family after it. Access JWTs already
-    issued are self-contained and live out their (short) TTL.
+    Matches the subject case-insensitively. First writes a subject marker (see
+    :func:`subject_revoked_before`) that lives *marker_ttl* seconds -- give it
+    the family max lifetime -- so authorizations racing this call are refused
+    too. Then tombstones each existing family and drops the subject's
+    unredeemed authorization codes from *code_store* when given. Access JWTs
+    already issued are self-contained and live out their (short) TTL.
     """
     if not email:
         raise ValueError("revoke_refresh_subject needs a non-empty email")
     target = email.lower()
+    now = int(time.time()) if now is None else now
+    refresh_store[subject_marker_key(target)] = {
+        "revoked_before": now,
+        "reason": reason,
+        "exp": now + max(marker_ttl, tombstone_ttl),
+    }
 
     def _is_subjects(record: Any) -> bool:
         subject = (record or {}).get("email") if isinstance(record, dict) else None
@@ -159,6 +198,7 @@ def make_on_credentials_changed(
     refresh_store: Optional[MutableMapping[str, Any]] = None,
     code_store: Optional[MutableMapping[str, Any]] = None,
     tombstone_ttl: int = 0,
+    marker_ttl: int = 0,
     reason: str = "the account's credentials changed",
 ) -> CredentialsChanged:
     """Return the hook every credential-changing path calls.
@@ -173,24 +213,28 @@ def make_on_credentials_changed(
         raise ValueError("tombstone_ttl must be positive when refresh_store is given")
 
     def on_credentials_changed(email: str, *, keep: Optional[str] = None) -> None:
-        session_store.revoke_user(email, keep=keep)
-        if refresh_store is None:
-            return
+        # Connector revocation runs even if the session revocation fails, and
+        # vice versa; a session-store failure still surfaces to the caller.
         try:
-            revoke_refresh_subject(
-                refresh_store,
-                email,
-                reason=reason,
-                tombstone_ttl=tombstone_ttl,
-                code_store=code_store,
-            )
-        except Exception:  # noqa: BLE001 - sessions are already gone; say so loudly
-            _logger.exception(
-                "enlace_auth: could not revoke connector sessions for %r after a "
-                "credential change; revoke them with "
-                "`enlace-auth revoke-connector-session --email`",
-                email,
-            )
+            session_store.revoke_user(email, keep=keep)
+        finally:
+            if refresh_store is not None:
+                try:
+                    revoke_refresh_subject(
+                        refresh_store,
+                        email,
+                        reason=reason,
+                        tombstone_ttl=tombstone_ttl,
+                        marker_ttl=marker_ttl,
+                        code_store=code_store,
+                    )
+                except Exception:  # noqa: BLE001 - say so loudly, don't mask
+                    _logger.exception(
+                        "enlace_auth: could not revoke connector sessions for %r "
+                        "after a credential change; revoke them with "
+                        "`enlace-auth revoke-connector-session --email`",
+                        email,
+                    )
 
     return on_credentials_changed
 
diff --git a/enlace_auth/auth/sessions.py b/enlace_auth/auth/sessions.py
index 1ce49fd..17ae540 100644
--- a/enlace_auth/auth/sessions.py
+++ b/enlace_auth/auth/sessions.py
@@ -4,9 +4,9 @@
 Session IDs are 32-byte urlsafe tokens. Revocation is a simple delete.
 
 Records are otherwise only deleted by logout, so given a *max_age* the store
-sweeps records older than it, a bounded batch at a time, whenever a session is
-created -- keeping the store (and :meth:`SessionStore.revoke_user`'s scan) from
-growing without bound.
+sweeps records older than it, a bounded batch at a time, when a session is
+created (at most once per *sweep_interval* per process) -- keeping the
+store (and :meth:`SessionStore.revoke_user`'s scan) from growing without bound.
 """
 
 from __future__ import annotations
@@ -27,11 +27,14 @@ def __init__(
         *,
         max_age: Optional[float] = None,
         sweep_batch: int = 100,
+        sweep_interval: float = 3600.0,
     ):
         self._store = store
         self._max_age = max_age
         self._sweep_batch = sweep_batch
+        self._sweep_interval = sweep_interval
         self._sweep_pos = 0
+        self._last_sweep = float("-inf")
 
     def sweep_expired(self, *, now: Optional[float] = None) -> int:
         """Delete up to *sweep_batch* records older than *max_age*; return count.
@@ -59,10 +62,15 @@ def sweep_expired(self, *, now: Optional[float] = None) -> int:
         return removed
 
     def create(self, user_id: str, email: Optional[str] = None) -> str:
-        try:
-            self.sweep_expired()
-        except Exception:  # noqa: BLE001 - housekeeping must never block a login
-            pass
+        # Throttled: on a file store, iterating to the cursor walks the
+        # directory, which must not be paid on every login.
+        now = time.time()
+        if now - self._last_sweep >= self._sweep_interval:
+            self._last_sweep = now
+            try:
+                self.sweep_expired(now=now)
+            except Exception:  # noqa: BLE001 - housekeeping must never block a login
+                pass
         session_id = secrets.token_urlsafe(32)
         self._store[session_id] = {
             "user_id": user_id,
diff --git a/enlace_auth/plugin.py b/enlace_auth/plugin.py
index f0995ff..f148c1a 100644
--- a/enlace_auth/plugin.py
+++ b/enlace_auth/plugin.py
@@ -436,8 +436,15 @@ def wire(parent: "FastAPI", config) -> None:
                 refresh_token_ttl=osc.refresh_token_ttl_seconds,
                 refresh_reuse_detection=osc.refresh_reuse_detection_seconds,
             ),
+            marker_ttl=osc.refresh_family_max_lifetime_seconds,
         )
-        if osc.enabled and osc.refresh_token_ttl_seconds > 0
+        # Wired even while the OAuth server is off: families minted before it
+        # was disabled come back to life if it is re-enabled.
+        if refresh_tombstone_ttl(
+            refresh_token_ttl=osc.refresh_token_ttl_seconds,
+            refresh_reuse_detection=osc.refresh_reuse_detection_seconds,
+        )
+        > 0
         else {}
     )
     on_credentials_changed = make_on_credentials_changed(
diff --git a/tests/test_credential_revocation.py b/tests/test_credential_revocation.py
index 7786972..9716faa 100644
--- a/tests/test_credential_revocation.py
+++ b/tests/test_credential_revocation.py
@@ -278,3 +278,115 @@ def _csrf_shared(client: TestClient) -> dict:
     client.get("/api/public_app/ping")
     raw = verify_cookie(client.cookies.get("enlace_csrf"), _SIGNING_KEY, salt="csrf")
     return {"X-CSRF-Token": raw}
+
+
+# ---- races the scan cannot see (review of #29) -------------------------------
+
+
+def _pkce_pair():
+    import base64
+    import hashlib
+
+    verifier = "v" * 64
+    challenge = (
+        base64.urlsafe_b64encode(hashlib.sha256(verifier.encode()).digest())
+        .rstrip(b"=")
+        .decode()
+    )
+    return verifier, challenge
+
+
+def _revoke_vic_via_admin(client):
+    csrf = _csrf(client)
+    _register(client, "boss@example.com", "bosspw1!", csrf)
+    client.post(
+        "/_admin/api/users",
+        json={"email": "vic@example.com", "password": "victim-pw1"},
+        headers=csrf,
+    )
+    r = client.post(
+        "/_admin/api/users/vic@example.com/password",
+        json={"password": "brand-new-pw1"},
+        headers=csrf,
+    )
+    assert r.status_code == 200, r.text
+
+
+def test_family_created_during_revocation_cannot_refresh(platform):
+    """A family written after the scan (code already consumed) is still dead."""
+    from enlace_auth.auth.oauth_server import _hash_refresh
+
+    client, refresh, _codes = platform
+    _revoke_vic_via_admin(client)
+    # The racing code grant read the clock before the revocation and wrote its
+    # family after the scan.
+    rec = _refresh_record("frace", "vic@example.com")
+    rec["auth_at"] = int(time.time()) - 5
+    refresh[_hash_refresh("racer")] = rec
+    r = client.post(
+        "/auth/oauth/token",
+        data={
+            "grant_type": "refresh_token",
+            "client_id": "c",
+            "refresh_token": "racer",
+        },
+    )
+    assert r.status_code == 400
+    assert refresh.get(revoked_family_key("frace")) is not None
+
+
+def test_family_authorized_after_revocation_still_refreshes(platform):
+    from enlace_auth.auth.oauth_server import _hash_refresh
+
+    client, refresh, _codes = platform
+    _revoke_vic_via_admin(client)
+    rec = _refresh_record("flater", "vic@example.com")
+    rec["auth_at"] = int(time.time()) + 5
+    refresh[_hash_refresh("later")] = rec
+    r = client.post(
+        "/auth/oauth/token",
+        data={
+            "grant_type": "refresh_token",
+            "client_id": "c",
+            "refresh_token": "later",
+        },
+    )
+    assert r.status_code == 200, r.text
+
+
+def test_code_issued_before_revocation_but_written_after_is_refused(platform):
+    client, _refresh, codes = platform
+    _revoke_vic_via_admin(client)
+    verifier, challenge = _pkce_pair()
+    now = int(time.time())
+    codes["late-code"] = {
+        "iat": now - 5,  # session read before the change, code written after
+        "client_id": "c",
+        "email": "vic@example.com",
+        "redirect_uri": "http://localhost/cb",
+        "code_challenge": challenge,
+        "scope": "mcp:read",
+        "resource": "https://x/api/mcp",
+        "exp": now + 60,
+    }
+    r = client.post(
+        "/auth/oauth/token",
+        data={
+            "grant_type": "authorization_code",
+            "client_id": "c",
+            "code": "late-code",
+            "redirect_uri": "http://localhost/cb",
+            "code_verifier": verifier,
+        },
+    )
+    assert r.status_code == 400
+    assert r.json()["error"] == "invalid_grant"
+
+
+def test_session_sweep_is_throttled():
+    backend: dict = {}
+    sessions = SessionStore(backend, max_age=60, sweep_interval=3600)
+    sessions.create("a")
+    backend["old"] = {"user_id": "a", "created_at": time.time() - 999}
+    sessions.create("b")
+    assert "old" in backend, "second create within the interval must not sweep"