Security hardening (maziggy/bambuddy-security #14)

backend/app/api/routes/auth.py

@@ -683,7 +683,9 @@ async def mint_websocket_token(
     username = current_user.username if current_user is not None else None
     # The socket only carries the printers its minter may see (#1727)
     api_key_id = api_key.id if api_key is not None else None
-    return {"token": await create_websocket_token(username, api_key_id)}
+    # The key's prefix binds the token to this key (SQLite reuses a deleted key's id)
+    api_key_prefix = api_key.key_prefix if api_key is not None else None
+    return {"token": await create_websocket_token(username, api_key_id, api_key_prefix)}
 
 
 @router.post("/media-token")

backend/app/api/routes/camera.py

@@ -950,6 +950,8 @@ async def create_stream_token(
         "token": await create_camera_stream_token(
             username=user.username if user is not None else None,
             api_key_id=api_key.id if api_key is not None else None,
+            # Binds the token to this key (SQLite reuses a deleted key's id)
+            api_key_prefix=api_key.key_prefix if api_key is not None else None,
         )
     }
 

backend/app/api/routes/connected_apps.py

@@ -44,6 +44,7 @@ from backend.app.core.auth import (
     get_user_by_username,
     is_auth_enabled,
     security,
+    session_dated,
     token_outlived_its_session,
     verify_password,
 )
@@ -281,33 +282,32 @@ async def authorize(
         db.add(ConnectedAppGrant(app_id=app.id, user_id=user.id))
 
     code = secrets.token_urlsafe(32)
-    db.add(
-        AuthEphemeralToken.new_connect_code(
+    # Read before the commit: a rollback below expires what this session loaded,
+    # and reading it then raises in async SQLAlchemy.
+    username, user_id, app_id, redirect_uri = user.username, user.id, app.id, app.redirect_uri
+
+    def new_code() -> AuthEphemeralToken:
+        return AuthEphemeralToken.new_connect_code(
             code_hash=_hash_code(code),
-            username=user.username,
-            app_id=app.id,
+            username=username,
+            app_id=app_id,
             code_challenge=data.code_challenge,
             expires_at=now + CODE_TTL,
+            # When the session that consents began (core/auth.session_dated)
+            **session_dated(username),
         )
-    )
+
+    db.add(new_code())
     try:
         await db.commit()
     except IntegrityError:
         # Two tabs consenting at once both inserted the grant; the other one won.
         await db.rollback()
-        db.add(
-            AuthEphemeralToken.new_connect_code(
-                code_hash=_hash_code(code),
-                username=user.username,
-                app_id=app.id,
-                code_challenge=data.code_challenge,
-                expires_at=now + CODE_TTL,
-            )
-        )
+        db.add(new_code())
         await db.commit()
 
-    logger.info("Connected app %s: code issued for user %s", app.id, user.id)
-    return ConnectAuthorizeResponse(code=code, redirect_uri=app.redirect_uri)
+    logger.info("Connected app %s: code issued for user %s", app_id, user_id)
+    return ConnectAuthorizeResponse(code=code, redirect_uri=redirect_uri)
 
 
 # --------------------------------------------------------------------------

backend/app/api/routes/mfa.py

@@ -48,6 +48,7 @@ from backend.app.core.auth import (
     get_user_by_username,
     is_auth_enabled,
     resolve_session_max_minutes,
+    session_dated,
     token_outlived_its_session,
     verify_password,
 )
@@ -928,6 +929,7 @@ async def enable_email_otp(
             # Reuse the nonce field to store the code hash
             nonce=code_hash,
             expires_at=now + timedelta(minutes=10),
+            **session_dated(current_user.username),
         )
     )
     await db.commit()

backend/app/core/auth.py

@@ -848,17 +848,70 @@ async def token_outlived_its_session(db: AsyncSession, username: str | None, min
     password change or reset, a 2FA removal or a restore. A token from before
     belongs to an earlier holder of the name or an ended session; deletes and
     renames drop those, but one minted while they commit would survive.
-    Truncated to the second as for JWTs (_is_token_fresh): a new user and the
-    first token it is given can be written in the same flush.
+
+    A token naming a user that does not exist counts as outlived: its callers
+    look the user up again on their own, and a user created with the name in
+    between would be served. Compared to the microsecond (both are written by
+    Python): every user is committed before any token is minted for it, so
+    there is nothing to round for, unlike a JWT's whole-second iat.
     """
-    if not username or minted_at is None:
+    if not username:
         return False
+    if minted_at is None:
+        return True
     changed = (
         await db.execute(select(User.password_changed_at).where(func.lower(User.username) == func.lower(username)))
-    ).scalar_one_or_none()
+    ).first()
     if changed is None:
-        return False
-    return minted_at.replace(tzinfo=None) < changed.replace(tzinfo=None, microsecond=0)
+        return True  # no such user
+    session_start = changed[0]
+    if session_start is None:
+        return False  # no session start recorded (an old row the start-up backfill has not reached)
+    return minted_at.replace(tzinfo=None) < session_start.replace(tzinfo=None)
+
+
+def _minting_session_start() -> datetime | None:
+    """When the session of the request minting a token began, as a created_at.
+
+    A token minted by a signed-in user counts from their JWT's issue time, not
+    from when its row is written: a password change, reset or 2FA removal that
+    commits in between (a stolen JWT minting in a loop) would otherwise leave a
+    token dated after it. The end of the iat second, as _is_token_fresh treats
+    a JWT from the second the session began as fresh. None outside a request
+    signed in with a JWT (API keys, auth off, sign-in steps).
+    """
+    cached = _authenticated_user.get()
+    if cached is None:
+        return None
+    try:
+        iat = jwt.decode(cached[0], SECRET_KEY, algorithms=[ALGORITHM]).get("iat")
+    except JWTError:
+        return None
+    if iat is None:
+        return None
+    return datetime.fromtimestamp(int(iat), tz=timezone.utc).replace(tzinfo=None) + timedelta(microseconds=999999)
+
+
+def session_dated(username: str | None) -> dict:
+    """created_at for a token a signed-in user mints for themself (see
+    _minting_session_start); nothing when the request's session is someone
+    else's (an admin minting for a user) or there is none."""
+    cached = _authenticated_user.get()
+    if not username or cached is None or cached[1].username.lower() != username.lower():
+        return {}
+    start = _minting_session_start()
+    return {"created_at": start} if start is not None else {}
+
+
+def _api_key_still_the_minter(current_prefix: str | None, minted_prefix: str | None) -> bool:
+    """Whether a key-minted token's key is still the one that minted it.
+
+    The prefix is recorded at mint (in ``nonce``): SQLite gives a deleted key's
+    id to the next key created, and a token minted while the key was deleted
+    would otherwise work with the new key's printers. Tokens from before the
+    prefix was recorded have none and pass (they expire within the hour).
+    """
+    return minted_prefix is None or current_prefix == minted_prefix
 
 
 # --- Camera stream tokens ---
@@ -874,7 +927,9 @@ async def token_outlived_its_session(db: AsyncSession, username: str | None, min
 CAMERA_STREAM_TOKEN_EXPIRE_MINUTES = 60
 
 
-async def create_camera_stream_token(username: str | None = None, api_key_id: int | None = None) -> str:
+async def create_camera_stream_token(
+    username: str | None = None, api_key_id: int | None = None, api_key_prefix: str | None = None
+) -> str:
     """Create a reusable token for camera stream/snapshot access.
 
     Records who minted it -- ``username`` for a user, ``api_key_id`` for an
@@ -898,7 +953,9 @@ async def create_camera_stream_token(username: str | None = None, api_key_id: in
                 token_type="camera_stream",
                 username=username or "",
                 api_key_id=api_key_id,
+                nonce=api_key_prefix if api_key_id is not None else None,
                 expires_at=expires_at,
+                **session_dated(username),
             )
         )
         await db.commit()
@@ -908,7 +965,9 @@ async def create_camera_stream_token(username: str | None = None, api_key_id: in
 WEBSOCKET_TOKEN_EXPIRE_MINUTES = 60
 
 
-async def create_websocket_token(username: str | None, api_key_id: int | None = None) -> str:
+async def create_websocket_token(
+    username: str | None, api_key_id: int | None = None, api_key_prefix: str | None = None
+) -> str:
     """Create a short-lived token for ``/api/v1/ws`` connections.
 
     Mirrors the camera-stream-token pattern: opaque random string stored
@@ -941,7 +1000,9 @@ async def create_websocket_token(username: str | None, api_key_id: int | None =
                 token_type="websocket",
                 username=username or "",
                 api_key_id=api_key_id,
+                nonce=api_key_prefix if api_key_id is not None else None,
                 expires_at=expires_at,
+                **session_dated(username),
             )
         )
         await db.commit()
@@ -969,6 +1030,10 @@ async def verify_websocket_token_principal(token: str) -> tuple[str, int | None]
         row = result.scalar_one_or_none()
         if row is None or await token_outlived_its_session(db, row.username, row.created_at):
             return None
+        if row.api_key_id is not None and not _api_key_still_the_minter(
+            await usable_api_key_prefix(db, row.api_key_id), row.nonce
+        ):
+            return None
         return row.username or "", row.api_key_id
 
 
@@ -1030,11 +1095,22 @@ async def principal_holds(
     return False
 
 
-async def user_id_holds(db: AsyncSession, user_id: int, permission: Permission) -> bool:
-    """principal_holds for the owner of a long-lived token, named by id."""
+async def user_id_holds(
+    db: AsyncSession, user_id: int, permission: Permission, minted_at: datetime | None = None
+) -> bool:
+    """principal_holds for the owner of a long-lived token, named by id.
+
+    With ``minted_at`` (the token's created_at), a user created after the token
+    is not its owner: SQLite gives a deleted user's id to the next user, and a
+    token minted while the user was deleted would stream with the new user's
+    printers for up to a year. Both created_at are set by the database, so they
+    share one clock.
+    """
     result = await db.execute(select(User).where(User.id == user_id).options(selectinload(User.groups)))
     user = result.scalar_one_or_none()
-    return user is not None and user.is_active and user.has_permission(permission.value)
+    if user is None or not user.is_active or not user.has_permission(permission.value):
+        return False
+    return minted_at is None or user.created_at is None or minted_at >= user.created_at
 
 
 async def principal_may_connect(db: AsyncSession, username: str | None, api_key_id: int | None) -> bool:
@@ -1078,9 +1154,9 @@ class StreamOpener:
     A live view runs for hours on a token checked once. Its opener is a user
     (a 60-minute token minted with their session), an API key, a long-lived
     token (Cam Wall, overlay, Home Assistant), or nobody while auth was off.
-    ``started_at`` is when the user's token was minted, not when the stream
-    opened: a session that ended in between (its token checked just before)
-    must still end the stream.
+    ``started_at`` is the token's created_at -- when the session that minted it
+    began -- not when the stream opened: a session that ended in between must
+    still end the stream.
     """
 
     auth_on: bool
@@ -1113,13 +1189,21 @@ async def camera_stream_access(token: str) -> tuple[PrinterScope, StreamOpener]
         row = result.scalar_one_or_none()
         if row is not None and await token_outlived_its_session(db, row.username, row.created_at):
             return None
+        if (
+            row is not None
+            and row.api_key_id is not None
+            and not _api_key_still_the_minter(await usable_api_key_prefix(db, row.api_key_id), row.nonce)
+        ):
+            return None
         if row is not None:
             # Minting it needed camera:view; a group or key change can take it away.
             if not await principal_holds(db, row.username, row.api_key_id, Permission.CAMERA_VIEW):
                 return None
             scope = await principal_printer_scope(db, row.username, row.api_key_id)
             prefix = await usable_api_key_prefix(db, row.api_key_id) if row.api_key_id is not None else None
-            minted = row.expires_at.replace(tzinfo=None) - timedelta(minutes=CAMERA_STREAM_TOKEN_EXPIRE_MINUTES)
+            # When the minting session began (see _minting_session_start), or
+            # when the token was minted for one that is not a user's session.
+            minted = row.created_at.replace(tzinfo=None)
             return scope, StreamOpener(
                 auth_on=True,
                 username=row.username or None,
@@ -1133,7 +1217,7 @@ async def camera_stream_access(token: str) -> tuple[PrinterScope, StreamOpener]
         from backend.app.services.long_lived_tokens import STREAM_SCOPES, verify_token as verify_long_lived
 
         record = await verify_long_lived(db, token, scope=STREAM_SCOPES)
-        if record is None or not await user_id_holds(db, record.user_id, Permission.CAMERA_VIEW):
+        if record is None or not await user_id_holds(db, record.user_id, Permission.CAMERA_VIEW, record.created_at):
             return None
         scope = await resolve_user_id_printer_scope(db, record.user_id)
         return scope, StreamOpener(
@@ -1183,7 +1267,7 @@ async def stream_opener_still_allowed(opener: StreamOpener, printer_id: int) ->
                     or record.user_id != opener.long_lived_user_id
                     or record.lookup_prefix != opener.long_lived_prefix
                     or _is_expired(record, datetime.now(timezone.utc))
-                    or not await user_id_holds(db, record.user_id, Permission.CAMERA_VIEW)
+                    or not await user_id_holds(db, record.user_id, Permission.CAMERA_VIEW, record.created_at)
                 ):
                     return False
                 scope = await resolve_user_id_printer_scope(db, record.user_id)
@@ -1227,7 +1311,7 @@ async def verify_camwall_token(token: str) -> PrinterScope | None:
 
         record = await verify_long_lived(db, token, scope="camwall")
         # Minting it needed camera:view; a group change can take it away.
-        if record is None or not await user_id_holds(db, record.user_id, Permission.CAMERA_VIEW):
+        if record is None or not await user_id_holds(db, record.user_id, Permission.CAMERA_VIEW, record.created_at):
             return None
         return await resolve_user_id_printer_scope(db, record.user_id)
 
@@ -1249,7 +1333,7 @@ async def verify_overlay_token(token: str) -> PrinterScope | None:
 
         record = await verify_long_lived(db, token, scope="overlay")
         # Minting it needed camera:view; a group change can take it away.
-        if record is None or not await user_id_holds(db, record.user_id, Permission.CAMERA_VIEW):
+        if record is None or not await user_id_holds(db, record.user_id, Permission.CAMERA_VIEW, record.created_at):
             return None
         return await resolve_user_id_printer_scope(db, record.user_id)
 
@@ -1301,6 +1385,7 @@ async def create_media_token(username: str | None) -> str:
                 token_type="media",
                 username=username or "",
                 expires_at=expires_at,
+                **session_dated(username),
             )
         )
         await db.commit()
@@ -1742,6 +1827,8 @@ async def get_current_user(
             )
         if not _is_token_fresh(iat, user):
             raise credentials_exception
+        # What this request mints counts from this session (_minting_session_start)
+        _authenticated_user.set((token, user))
         return user
 
 
@@ -1833,6 +1920,8 @@ async def require_auth_if_enabled(
                     detail="Could not validate credentials",
                     headers={"WWW-Authenticate": "Bearer"},
                 )
+            # What this request mints counts from this session (_minting_session_start)
+            _authenticated_user.set((credentials.credentials, user))
             return user
 
         # No credentials provided

backend/app/models/auth_ephemeral.py

@@ -43,6 +43,12 @@ class TokenType(str, Enum):
     EMAIL_OTP_SETUP = "email_otp_setup"
     SLICER_DOWNLOAD = "slicer_download"
     CONNECT_CODE = "connect_code"
+    # Written as bare strings where they are used; listed here so every kind is
+    # known (see test_token_outlived_session).
+    WEBSOCKET = "websocket"
+    CAMERA_STREAM = "camera_stream"
+    MEDIA = "media"
+    REVOKED_JTI = "revoked_jti"  # the logout list, not a credential
 
 
 class EventType(str, Enum):
@@ -198,6 +204,7 @@ class AuthEphemeralToken(Base):
         app_id: int,
         code_challenge: str,
         expires_at: datetime,
+        created_at: datetime | None = None,
     ) -> AuthEphemeralToken:
         """Create a connected-app authorization code.
 
@@ -212,6 +219,8 @@ class AuthEphemeralToken(Base):
             provider_id=app_id,
             nonce=code_challenge,
             expires_at=expires_at,
+            # When the minting session began (core/auth.session_dated), if given
+            **({"created_at": created_at} if created_at is not None else {}),
         )
 
 

backend/tests/integration/test_camera_stream_recheck.py

@@ -157,18 +157,19 @@ async def test_a_stream_opened_with_auth_off_ends_when_it_is_turned_on(async_cli
 
 async def test_a_stream_opened_as_the_session_ends_does_not_outlive_it(admin, printer, db_session):
     """The password change commits between the token check and the stream's
-    start: its sessions ended after the token was minted, so the stream ends."""
+    start: its sessions ended after the token was minted, so the stream ends.
+    (A token minted before the session began is refused outright as well.)"""
     token = await auth.create_camera_stream_token(username="camadmin")
+    opener = await _opener(token)  # checked just before the change commits
     await asyncio.sleep(0.01)
     await db_session.execute(
         text("UPDATE users SET password_changed_at = :t WHERE id = :id"), {"t": utcnow_naive(), "id": admin}
     )
     await db_session.commit()
-    await asyncio.sleep(0.01)
-    opener = await _opener(token)  # the token row was still there when it was read
 
     response = await camera.camera_stream(printer, MagicMock(), fps=5, opener=opener)
     assert await _ended(response.body_iterator)
+    assert await auth.camera_stream_access(token) is None
 
 
 @pytest.fixture

backend/tests/integration/test_security.py

@@ -452,15 +452,6 @@ class TestOIDCExchangeReplay:
     async def test_exchange_token_is_single_use(self, async_client: AsyncClient, db_session: AsyncSession):
         """The second call to /oidc/exchange with the same token returns 401."""
         exchange_token = secrets.token_urlsafe(32)
-        db_session.add(
-            AuthEphemeralToken(
-                token=exchange_token,
-                token_type="oidc_exchange",
-                username="oidc_replay_user",
-                expires_at=datetime.now(timezone.utc) + timedelta(minutes=5),
-            )
-        )
-        await db_session.commit()
 
         # Seed the user so the exchange can resolve it
         from backend.app.core.auth import get_password_hash
@@ -478,6 +469,18 @@ class TestOIDCExchangeReplay:
                 )
                 await db.commit()
 
+        # Minted after the user exists, as the OIDC callback does (a token from
+        # before the user's session began is refused).
+        db_session.add(
+            AuthEphemeralToken(
+                token=exchange_token,
+                token_type="oidc_exchange",
+                username="oidc_replay_user",
+                expires_at=datetime.now(timezone.utc) + timedelta(minutes=5),
+            )
+        )
+        await db_session.commit()
+
         first = await async_client.post("/api/v1/auth/oidc/exchange", json={"oidc_token": exchange_token})
         assert first.status_code == 200
 

backend/tests/integration/test_token_outlived_session.py

@@ -160,9 +160,9 @@ async def test_a_connect_code_from_before_the_session_is_refused(async_client: A
 async def test_every_token_kind_named_after_a_user_is_covered():
     """A new kind must get the check (and a test above)."""
     covered = {
-        "websocket",
-        "camera_stream",
-        "media",
+        TokenType.WEBSOCKET.value,
+        TokenType.CAMERA_STREAM.value,
+        TokenType.MEDIA.value,
         TokenType.PRE_AUTH.value,
         TokenType.PASSWORD_RESET.value,
         TokenType.OIDC_EXCHANGE.value,
@@ -172,5 +172,232 @@ async def test_every_token_kind_named_after_a_user_is_covered():
     named_by_nobody = {
         TokenType.OIDC_STATE.value,  # before anyone is signed in
         TokenType.SLICER_DOWNLOAD.value,  # bound to a resource, not a user
+        TokenType.REVOKED_JTI.value,  # the logout list, not a credential
     }
     assert {kind.value for kind in TokenType} - named_by_nobody <= covered
+
+
+async def test_every_token_kind_in_the_code_is_in_the_enum():
+    """A kind written as a bare string would escape the check above."""
+    import re
+    from pathlib import Path
+
+    import backend.app as app_pkg
+
+    root = Path(app_pkg.__file__).parent
+    kinds = set()
+    for path in root.rglob("*.py"):
+        kinds.update(re.findall(r'token_type\s*(?:==|=)\s*"([a-z_]+)"', path.read_text()))
+    kinds.discard("bearer")  # the OAuth token_type of a login response, not a stored kind
+    assert kinds <= {kind.value for kind in TokenType}, kinds - {kind.value for kind in TokenType}
+
+
+async def test_a_token_of_a_user_that_does_not_exist_is_refused(carol, db_session):
+    """Callers look the user up again on their own: a recreate in between would
+    otherwise be served as the new user."""
+    stale = await _token(db_session, "media", minted_ago_minutes=10, username="ghost")
+
+    assert await auth.verify_media_token(stale) is None
+
+
+async def test_a_token_from_the_same_second_but_before_the_session_is_refused(carol, db_session):
+    """Both timestamps keep microseconds; nothing writes a user and its first
+    token in one flush, so there is no reason to round."""
+    session_start = datetime.now(timezone.utc).replace(tzinfo=None, microsecond=600000)
+    await db_session.execute(
+        text("UPDATE users SET password_changed_at = :t WHERE id = :id"), {"t": session_start, "id": carol}
+    )
+    db_session.add(
+        AuthEphemeralToken(
+            token="same-second",
+            token_type="media",
+            username="carol",
+            created_at=session_start.replace(microsecond=300000),
+            expires_at=datetime.now(timezone.utc) + timedelta(hours=1),
+        )
+    )
+    await db_session.commit()
+
+    assert await auth.verify_media_token("same-second") is None
+
+
+async def test_a_token_minted_as_the_session_ends_is_dated_by_its_session(carol, db_session):
+    """A stolen JWT keeps minting while the owner changes the password: a mint
+    checked before the change committed wrote its row after. The token counts
+    from when the session that minted it began, so it is refused."""
+    import jwt as pyjwt
+
+    from backend.app.models.user import User
+
+    issued = datetime.now(timezone.utc) - timedelta(minutes=20)  # the session began 20 min ago
+    session_jwt = pyjwt.encode(
+        {"sub": "carol", "iat": issued, "exp": issued + timedelta(days=1), "jti": "stolen"},
+        auth.SECRET_KEY,
+        algorithm=auth.ALGORITHM,
+    )
+    user = await db_session.get(User, carol)
+    marker = auth._authenticated_user.set((session_jwt, user))
+    try:
+        token = await auth.create_websocket_token("carol")
+    finally:
+        auth._authenticated_user.reset(marker)
+    # The password change, which started before the row was written, commits now.
+    await db_session.execute(
+        text("UPDATE users SET password_changed_at = :t WHERE id = :id"), {"t": _ago(minutes=5), "id": carol}
+    )
+    await db_session.commit()
+
+    assert await auth.verify_websocket_token_principal(token) is None
+
+
+async def test_a_keys_token_does_not_pass_to_the_next_key_given_its_id(carol, db_session):
+    """Minted while the key was deleted: SQLite gives the id to the next key."""
+    from backend.app.models.api_key import APIKey
+
+    key = APIKey(name="old", key_hash="h", key_prefix="bb_old1")
+    db_session.add(key)
+    await db_session.commit()
+    key_id = key.id
+    ws = await auth.create_websocket_token(None, key_id, api_key_prefix="bb_old1")
+    cam = await auth.create_camera_stream_token(api_key_id=key_id, api_key_prefix="bb_old1")
+    await db_session.execute(text("DELETE FROM api_keys WHERE id = :id"), {"id": key_id})
+    await db_session.commit()
+    db_session.add(APIKey(id=key_id, name="new", key_hash="h2", key_prefix="bb_new1"))
+    await db_session.commit()
+
+    assert await auth.verify_websocket_token_principal(ws) is None
+    assert await auth.camera_stream_access(cam) is None
+
+
+async def test_a_long_lived_token_does_not_pass_to_the_next_user_given_its_id(carol, db_session):
+    """Minted while its user was deleted: SQLite gives the id to the next user,
+    and the token would stream with their printers for up to a year."""
+    from backend.app.services.long_lived_tokens import create_token
+
+    created = await create_token(db_session, user_id=carol, name="wall", expires_in_days=30, scope="camwall")
+    await db_session.execute(
+        text("UPDATE long_lived_tokens SET created_at = :t WHERE id = :id"),
+        {"t": _ago(minutes=60), "id": created.record.id},
+    )
+    await db_session.execute(
+        text("UPDATE users SET created_at = :t WHERE id = :id"), {"t": _ago(minutes=1), "id": carol}
+    )
+    await db_session.commit()  # carol now stands for the user created after the token
+
+    assert await auth.verify_camwall_token(created.plaintext) is None
+    assert await auth.camera_stream_access(created.plaintext) is None
+
+
+async def _session_then_password_change(client: AsyncClient, db, carol_id: int, mint) -> str:
+    """Sign carol in, mint through the real route a second later, then commit a
+    password change dated between the session's start and the token's row.
+    Returns the token the route handed out."""
+    import asyncio
+
+    login = await client.post("/api/v1/auth/login", json={"username": "carol", "password": "Userpass1!"})
+    headers = {"Authorization": f"Bearer {login.json()['access_token']}"}
+    await asyncio.sleep(1.1)  # the row is written in a later second than the session began
+    token = await mint(headers)
+    # The change commits just before the request wrote its row.
+    changed = datetime.now(timezone.utc).replace(tzinfo=None) - timedelta(milliseconds=50)
+    await db.execute(text("UPDATE users SET password_changed_at = :t WHERE id = :id"), {"t": changed, "id": carol_id})
+    await db.commit()
+    return token
+
+
+@pytest.mark.parametrize(
+    ("route", "verify"),
+    [
+        ("/api/v1/auth/ws-token", lambda t: auth.verify_websocket_token_principal(t)),
+        ("/api/v1/auth/media-token", lambda t: auth.verify_media_token(t)),
+        ("/api/v1/printers/camera/stream-token", lambda t: auth.camera_stream_access(t)),
+    ],
+)
+async def test_a_token_minted_through_the_route_counts_from_its_session(
+    async_client: AsyncClient, carol, db_session, route, verify
+):
+    async def mint(headers):
+        response = await async_client.post(route, headers=headers)
+        assert response.status_code == 200, response.text
+        return response.json()["token"]
+
+    token = await _session_then_password_change(async_client, db_session, carol, mint)
+
+    assert await verify(token) is None
+
+
+async def test_a_connect_code_counts_from_its_session(async_client: AsyncClient, carol, db_session):
+    from backend.tests.integration.test_connected_apps import _authorize, _exchange, _pkce, _register
+
+    login = await async_client.post("/api/v1/auth/login", json={"username": "tokadmin", "password": "AdminPass1!"})
+    app = await _register(async_client, login.json()["access_token"])
+    verifier, challenge = _pkce()
+    codes = []
+
+    async def mint(headers):
+        response = await _authorize(async_client, headers["Authorization"].split()[1], app, challenge)
+        assert response.status_code == 200, response.text
+        codes.append(response.json()["code"])
+        return codes[-1]
+
+    code = await _session_then_password_change(async_client, db_session, carol, mint)
+
+    assert (await _exchange(async_client, app, code, verifier)).status_code == 400
+
+
+async def test_an_email_otp_setup_token_counts_from_its_session(async_client: AsyncClient, carol, db_session):
+    from unittest.mock import AsyncMock, MagicMock, patch
+
+    await db_session.execute(text("UPDATE users SET email = 'carol@example.com' WHERE id = :id"), {"id": carol})
+    await db_session.commit()
+
+    async def mint(headers):
+        with (
+            patch("backend.app.api.routes.mfa.get_smtp_settings", AsyncMock(return_value=MagicMock())),
+            patch("backend.app.api.routes.mfa.send_email", MagicMock()),
+        ):
+            response = await async_client.post("/api/v1/auth/2fa/email/enable", headers=headers)
+        assert response.status_code == 200, response.text
+        return response.json()["setup_token"]
+
+    setup = await _session_then_password_change(async_client, db_session, carol, mint)
+    written = await db_session.execute(
+        text("SELECT created_at FROM auth_ephemeral_tokens WHERE token = :t"), {"t": setup}
+    )
+    changed = await db_session.execute(text("SELECT password_changed_at FROM users WHERE id = :id"), {"id": carol})
+    assert written.scalar_one() < changed.scalar_one()  # dated before the change, so refused on use
+
+
+async def test_a_connect_code_from_the_grant_race_retry_counts_from_its_session(
+    async_client: AsyncClient, carol, db_session
+):
+    """Two consents at once: the loser rolls back and mints the code again."""
+    from unittest.mock import patch
+
+    from sqlalchemy.exc import IntegrityError
+    from sqlalchemy.ext.asyncio import AsyncSession
+
+    from backend.tests.integration.test_connected_apps import _authorize, _exchange, _pkce, _register
+
+    login = await async_client.post("/api/v1/auth/login", json={"username": "tokadmin", "password": "AdminPass1!"})
+    app = await _register(async_client, login.json()["access_token"])
+    verifier, challenge = _pkce()
+    real_commit = AsyncSession.commit
+    raised = []
+
+    async def commit_losing_the_grant_race(self):
+        if not raised and any(getattr(o, "token_type", None) == TokenType.CONNECT_CODE for o in self.new):
+            raised.append(1)
+            raise IntegrityError("INSERT", {}, Exception("grant exists"))
+        return await real_commit(self)
+
+    async def mint(headers):
+        with patch.object(AsyncSession, "commit", commit_losing_the_grant_race):
+            response = await _authorize(async_client, headers["Authorization"].split()[1], app, challenge)
+        assert response.status_code == 200, response.text
+        return response.json()["code"]
+
+    code = await _session_then_password_change(async_client, db_session, carol, mint)
+
+    assert raised, "the retry path was not taken"
+    assert (await _exchange(async_client, app, code, verifier)).status_code == 400

backend/tests/unit/test_invalidate_all_sessions.py

@@ -27,7 +27,7 @@ async def test_signs_every_user_out_and_drops_short_lived_auth_records(test_engi
         ]
     )
     expires = datetime.now(timezone.utc) + timedelta(hours=1)
-    for token_type in ("revoked_jti", "media", "camera_stream", "websocket", *[t.value for t in TokenType]):
+    for token_type in {"revoked_jti", "media", "camera_stream", "websocket", *[t.value for t in TokenType]}:
         db_session.add(AuthEphemeralToken(token=f"t-{token_type}", token_type=token_type, expires_at=expires))
     for event_type in EventType:
         db_session.add(AuthRateLimitEvent(username="changed", event_type=event_type.value))