From 3addd4a4a82996e9e4476576802843ab7da13042 Mon Sep 17 00:00:00 2001
From: MrChengLen <lennart.seidel@icloud.com>
Date: Wed, 30 Sep 2026 14:54:54 +0200
Subject: [PATCH] =?UTF-8?q?fix(auth):=20one=20token=20check=20for=20every?=
 =?UTF-8?q?=20route=20=E2=80=94=20sign-ins=20last=2030=20days=20from=20log?=
 =?UTF-8?q?in?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Account routes, upload routes and /auth/refresh now resolve access and
refresh tokens through one check against the account, and a refresh hands
back the refresh token it was sent, so a sign-in lasts 30 days from login.

- app/core/tokens.py: access and refresh tokens carry the password-hash
  fingerprint (phv) that reset links already use; decode_session_token
  replaces decode_token_full (only tests used it).
- app/api/routes/auth.py: _session_user resolves a token to the live
  account (UUID cast, active, not deleted, matching phv) for
  get_current_user and /auth/refresh; _token_pair mints the pair for
  register, login and refresh. /auth/refresh needs a database (503
  without one, like login).
- Tokens minted before this change have no phv and are refused, so
  everyone who is signed in signs in once more after the deploy.
- Client: app/static/js/auth.js stores the refresh token it gets back,
  as before; no client change needed.
- docs/api-reference.md (refresh and reset rows) and
  docs/api-usage-guide.md (refresh section, flow diagram) describe the
  behaviour; tests/test_auth_refresh.py is new.

Alternatives considered: a per-user token_version column (a migration,
and a bump in every path that should end sessions); server-side
refresh-token rotation (a token table plus client changes, since the web
UI refreshes from parallel requests with one token).

Full suite 1499 green (72 skipped locally); ruff + format + gitleaks clean.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 CHANGELOG.md                         |  10 ++
 app/api/routes/auth.py               |  58 +++++----
 app/core/tokens.py                   |  62 +++++----
 docs/api-reference.md                |   4 +-
 docs/api-usage-guide.md              |   8 +-
 tests/test_auth_refresh.py           | 181 +++++++++++++++++++++++++++
 tests/test_email_verification.py     |   4 +-
 tests/test_jwt_claims.py             |  13 +-
 tests/test_password_reset.py         |  63 ++++++++++
 tests/test_rate_limit_routes.py      |  24 ++--
 tests/test_token_types.py            |  10 +-
 tests/test_upload_auth_resolution.py |  63 +++++++++-
 12 files changed, 415 insertions(+), 85 deletions(-)
 create mode 100644 tests/test_auth_refresh.py

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 5346c0f..16668fa 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -9,6 +9,16 @@ Versions follow [Semantic Versioning](https://semver.org/).
 
 ## [Unreleased]
 
+### Changed — a sign-in lasts 30 days from login
+
+- `POST /auth/refresh` returns a new access token together with the refresh
+  token it was sent, so a sign-in ends 30 days after login; without a
+  database it answers `503`, like login.
+- Everyone who is signed in has to sign in once more after this update.
+- `docs/api-reference.md` and `docs/api-usage-guide.md` describe refresh and
+  password reset; tests in `tests/test_auth_refresh.py` (new),
+  `tests/test_password_reset.py` and `tests/test_upload_auth_resolution.py`.
+
 ### Fixed — patch-policy's `cosign verify` names an image tag that exists
 
 `docs/patch-policy.md` told readers to verify the release image
diff --git a/app/api/routes/auth.py b/app/api/routes/auth.py
index cf22f78..34f5991 100644
--- a/app/api/routes/auth.py
+++ b/app/api/routes/auth.py
@@ -22,7 +22,7 @@
     create_refresh_token,
     decode_email_verify_token,
     decode_password_reset_token,
-    decode_token,
+    decode_session_token,
     password_hash_version,
 )
 from app.core.config import settings
@@ -160,12 +160,19 @@ async def get_current_user(
         )
     if not authorization or not authorization.startswith("Bearer "):
         raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Not authenticated.")
-    token = authorization.removeprefix("Bearer ")
-    user_id = decode_token(token, expected_type="access")
+    return await _session_user(db, authorization.removeprefix("Bearer "), expected_type="access")
+
+
+async def _session_user(db: AsyncSession, token: str, expected_type: str) -> User:
+    """Resolve an access or refresh token to its live user, or raise 401.
+
+    The token's ``phv`` must match the user's current password hash, so a
+    password reset ends every session issued before it."""
+    user_id, token_phv = decode_session_token(token, expected_type=expected_type)
     # asyncpg happily binds a str to a UUID column, but SQLAlchemy's generic
     # UUID type (used by the SQLite test engine) calls ``.hex`` on the value
-    # and blows up on bare strings. Cast explicitly so the dependency works
-    # on any backend and rejects malformed subjects cleanly.
+    # and blows up on bare strings. Cast explicitly so this works on any
+    # backend and rejects malformed subjects cleanly.
     try:
         user_uuid = uuid.UUID(user_id)
     except (ValueError, TypeError):
@@ -181,6 +188,10 @@ async def get_current_user(
     user = result.scalar_one_or_none()
     if not user:
         raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="User not found.")
+    if password_hash_version(user.password_hash) != token_phv:
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid or expired token."
+        )
     return user
 
 
@@ -238,6 +249,19 @@ def _db_required(db: AsyncSession | None) -> AsyncSession:
     return db
 
 
+def _token_pair(user: User, refresh_token: str | None = None) -> TokenResponse:
+    """Access + refresh token for ``user``, both bound to the current password.
+
+    ``/auth/refresh`` passes the presented refresh token back instead of
+    minting a new one, so refreshing never extends a sign-in: it ends when
+    that token expires, 30 days after login."""
+    phv = password_hash_version(user.password_hash)
+    return TokenResponse(
+        access_token=create_access_token(str(user.id), phv=phv, role=user.role.value),
+        refresh_token=refresh_token or create_refresh_token(str(user.id), phv=phv),
+    )
+
+
 def _email_hash(email: str) -> str:
     """Lowercased SHA-256 of an email address.
 
@@ -345,10 +369,7 @@ async def register(
         actor_ip=_client_ip(request),
         payload={"trigger": "register"},
     )
-    return TokenResponse(
-        access_token=create_access_token(str(user.id), role=user.role.value),
-        refresh_token=create_refresh_token(str(user.id)),
-    )
+    return _token_pair(user)
 
 
 @router.post("/login", response_model=TokenResponse)
@@ -387,10 +408,7 @@ async def login(request: Request, body: LoginRequest, db: AsyncSession | None =
         actor_user_id=user.id,
         actor_ip=_client_ip(request),
     )
-    return TokenResponse(
-        access_token=create_access_token(str(user.id), role=user.role.value),
-        refresh_token=create_refresh_token(str(user.id)),
-    )
+    return _token_pair(user)
 
 
 # Deliberately unlimited. A request costs one signature check (plus one
@@ -401,17 +419,9 @@ async def login(request: Request, body: LoginRequest, db: AsyncSession | None =
 @router.post("/refresh", response_model=TokenResponse)
 @limiter.exempt
 async def refresh(body: RefreshRequest, db: AsyncSession | None = Depends(get_db)):
-    user_id = decode_token(body.refresh_token, expected_type="refresh")
-    role = RoleEnum.user.value
-    if db is not None:
-        result = await db.execute(select(User).where(User.id == user_id, User.is_active.is_(True)))
-        user = result.scalar_one_or_none()
-        if user:
-            role = user.role.value
-    return TokenResponse(
-        access_token=create_access_token(user_id, role=role),
-        refresh_token=create_refresh_token(user_id),
-    )
+    db = _db_required(db)
+    user = await _session_user(db, body.refresh_token, expected_type="refresh")
+    return _token_pair(user, refresh_token=body.refresh_token)
 
 
 def _user_response(user: User) -> UserResponse:
diff --git a/app/core/tokens.py b/app/core/tokens.py
index c7f00ea..5cb38db 100644
--- a/app/core/tokens.py
+++ b/app/core/tokens.py
@@ -1,12 +1,12 @@
 # SPDX-License-Identifier: AGPL-3.0-or-later
 """JWT token primitives — issuance, decoding, and the password-hash-version
-fingerprint that binds reset tokens to a specific stored password.
+fingerprint that binds session and reset tokens to a specific stored password.
 
 Four token types share the JWT secret and are discriminated only by the
 ``type`` claim:
 
-- ``access``  — short-lived bearer credential (15 min)
-- ``refresh`` — rotating session token (30 d)
+- ``access``  — short-lived bearer credential (15 min, bound to ``phv``)
+- ``refresh`` — session token (30 d from login, bound to ``phv``)
 - ``reset``   — single-use password-reset link (30 min, bound to ``phv``)
 - ``verify``  — email-verification link (7 d, bound to ``eat``)
 
@@ -75,34 +75,40 @@ def _decode(token: str) -> dict[str, Any]:
 
 
 # ── Access / refresh tokens ───────────────────────────────────────────────────
+#
+# Both carry the ``phv`` of the password hash they were issued under (see
+# ``password_hash_version`` below). ``_session_user`` in ``app/api/routes/auth.py``
+# compares it with the current hash, so a password change ends every session
+# issued before it. A token without ``phv`` is rejected outright.
 
 
-def create_access_token(subject: str, role: str = "user") -> str:
+def create_access_token(subject: str, *, phv: str, role: str = "user") -> str:
     expire = datetime.now(timezone.utc) + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
     return jwt.encode(
-        _encode_claims({"sub": subject, "exp": expire, "type": "access", "role": role}),
+        _encode_claims({"sub": subject, "exp": expire, "type": "access", "role": role, "phv": phv}),
         settings.jwt_secret,
         algorithm=ALGORITHM,
     )
 
 
-def create_refresh_token(subject: str) -> str:
+def create_refresh_token(subject: str, *, phv: str) -> str:
     expire = datetime.now(timezone.utc) + timedelta(days=REFRESH_TOKEN_EXPIRE_DAYS)
     return jwt.encode(
-        _encode_claims({"sub": subject, "exp": expire, "type": "refresh"}),
+        _encode_claims({"sub": subject, "exp": expire, "type": "refresh", "phv": phv}),
         settings.jwt_secret,
         algorithm=ALGORITHM,
     )
 
 
 def decode_token(token: str, expected_type: str = "access") -> str:
-    """Return the subject claim. Use `decode_token_full` if the role claim is needed."""
-    sub, _role = decode_token_full(token, expected_type=expected_type)
+    """Return the subject claim. Resolving a token to a user needs the ``phv``
+    check too — see ``_session_user`` in ``app/api/routes/auth.py``."""
+    sub, _phv = decode_session_token(token, expected_type=expected_type)
     return sub
 
 
-def decode_token_full(token: str, expected_type: str = "access") -> tuple[str, str]:
-    """Return ``(subject, role)``. The role defaults to ``"user"`` for legacy tokens."""
+def decode_session_token(token: str, expected_type: str = "access") -> tuple[str, str]:
+    """Return ``(subject, phv)`` of an access or refresh token."""
     try:
         payload = _decode(token)
         if payload.get("type") != expected_type:
@@ -110,10 +116,10 @@ def decode_token_full(token: str, expected_type: str = "access") -> tuple[str, s
                 status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid token type."
             )
         sub: str | None = payload.get("sub")
-        if not sub:
+        phv: str | None = payload.get("phv")
+        if not sub or not phv:
             raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid token.")
-        role: str = payload.get("role", "user")
-        return sub, role
+        return sub, phv
     except JWTError:
         raise HTTPException(
             status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid or expired token."
@@ -125,22 +131,28 @@ def decode_token_full(token: str, expected_type: str = "access") -> tuple[str, s
 # A reset token is a short-lived JWT bound to a *version* of the user's
 # current password hash. Changing the password — either via a successful
 # reset or an admin intervention — rotates the version and silently
-# invalidates every outstanding reset token. No DB table, no cleanup job,
-# and single-use is implicit.
+# invalidates every outstanding reset token (and every session, whose
+# tokens carry the same version). No DB table, no cleanup job, and
+# single-use is implicit.
 
 
 def password_hash_version(password_hash: str) -> str:
     """Return a short stable fingerprint of a password hash.
 
-    We take the SHA-256 of the first 16 characters of the hash so a single
-    reset token cannot be replayed after a successful password change. The
-    bcrypt string starts with ``$2b$12$`` plus a 22-char salt; these 16
-    chars are enough entropy to diverge on any new hash.
-
-    If we ever migrate to argon2 the prefix shape changes — bump the reset
-    JWT ``type`` claim (e.g. ``reset`` → ``reset_v2``) at the same time so
-    in-flight tokens from the old scheme are rejected, then update this
-    function.
+    We take the SHA-256 of the first 16 characters of the hash so neither a
+    reset token nor a session token outlives a password change. The bcrypt
+    string starts with ``$2b$12$`` plus a 22-char salt; these 16 chars are
+    enough entropy to diverge on any new hash.
+
+    If we ever migrate to argon2 the prefix shape changes — and every
+    argon2id hash starts with the same 16 characters, so this function must
+    change with it or the fingerprint never changes again
+    (``test_every_new_hash_changes_the_phv`` pins that two hashes of one
+    password get different fingerprints). Bump the reset JWT ``type`` claim
+    (e.g. ``reset`` → ``reset_v2``) at the same time so in-flight tokens
+    from the old scheme are rejected. If login ever rehashes a password
+    (e.g. a higher bcrypt cost), that changes the fingerprint too and signs
+    the user's other sessions out.
     """
     return hashlib.sha256(password_hash[:16].encode()).hexdigest()
 
diff --git a/docs/api-reference.md b/docs/api-reference.md
index 13c9c78..84c3b7c 100644
--- a/docs/api-reference.md
+++ b/docs/api-reference.md
@@ -74,11 +74,11 @@ The endpoints in this section only respond when the Cloud overlay is configured
 |---|---|---|
 | `POST /api/v1/auth/register` | none | Create account; returns access + refresh tokens. Sends a verification email (fire-and-forget) in the request locale, and stores that locale as `preferred_lang`. |
 | `POST /api/v1/auth/login` | none | Exchange email + password for access (15 min) + refresh (30 d) tokens. |
-| `POST /api/v1/auth/refresh` | none (refresh-token in body) | Issue a new access token. |
+| `POST /api/v1/auth/refresh` | none (refresh-token in body) | Issue a new access token. The refresh token in the response is the one you sent, so a sign-in ends 30 d after login. |
 | `GET /api/v1/auth/me` | Bearer | Return the currently authenticated user (`id`, `email`, `tier`, `role`, `created_at`, `subscription_status`, `preferred_lang`). |
 | `PUT /api/v1/auth/account/language` | Bearer | Set the language for this user's transactional email. Body: `{"preferred_lang": "de"\|"en"}` — an unsupported value is a `422`. Returns the updated user object. This is **email** locale only; the web-UI locale stays URL-prefix driven (no cookie). |
 | `POST /api/v1/auth/forgot-password` | none | Issue a single-use password-reset link via email (30 min TTL). |
-| `POST /api/v1/auth/reset-password` | reset-token in body | Set a new password and invalidate older sessions via password-hash rotation. |
+| `POST /api/v1/auth/reset-password` | reset-token in body | Set a new password. Every access and refresh token issued before the reset stops working, so all existing sign-ins end. |
 | `POST /api/v1/auth/verify-email` | verify-token | Mark the user's email as verified. |
 | `POST /api/v1/auth/resend-verification` | Bearer | Re-send the verification mail (auth-required to avoid spam). |
 | `DELETE /api/v1/auth/account` | Bearer | Self-service account deletion (GDPR Art. 17). Requires re-confirmation: current password, registered email, and the literal string `DELETE`. Success is `204`. Free / never-paid accounts are hard-deleted; an account linked to Stripe is retained in a restricted state — only `email`, the Stripe customer id, the last `tier`, and `created_at` are kept for the 10-year HGB §257 / AO §147 invoice record (permitted by GDPR Art. 17(3)(b)), every other personal field is erased, and the row is hard-deleted at the end of the retention period. Any active Stripe subscription is cancelled first; a Stripe API error returns `500` and leaves the account unchanged. See [`docs/gdpr-account-deletion-design.md`](./gdpr-account-deletion-design.md). |
diff --git a/docs/api-usage-guide.md b/docs/api-usage-guide.md
index ae7b177..f7e4cbe 100644
--- a/docs/api-usage-guide.md
+++ b/docs/api-usage-guide.md
@@ -76,7 +76,7 @@ callers are also accepted — you get the `anonymous` tier.
      │  ◄──── no 401: runs on the anonymous tier          │
      │                                                    │
      │  POST /auth/refresh  { refresh_token }             │
-     │  ◄──── { access_token, refresh_token }  (rotated)  │
+     │  ◄──── { access_token, the same refresh_token }    │
      │                                                    │
      │  POST /api/v1/convert  (new bearer)                │
      │  ◄──── 200 OK                                      │
@@ -128,8 +128,10 @@ async function login(email, password) {
 ### Refresh the access token
 
 Before the access token expires, exchange your refresh token for a fresh
-access token. Refresh tokens rotate — store the new one and discard
-the old.
+access token. The response hands back the refresh token you sent —
+refreshing never extends a sign-in. It ends 30 days after login, or
+earlier when the account's password is reset; after that,
+`/auth/refresh` answers `401` and you log in again.
 
 Don't wait for a `401`. Endpoints that need an account — for example
 `/auth/me`, `/keys` and `/billing/checkout` — answer an expired token with
diff --git a/tests/test_auth_refresh.py b/tests/test_auth_refresh.py
new file mode 100644
index 0000000..9ad38a4
--- /dev/null
+++ b/tests/test_auth_refresh.py
@@ -0,0 +1,181 @@
+# SPDX-License-Identifier: AGPL-3.0-or-later
+"""POST /auth/refresh — when a refresh token still stands for its user.
+
+A refresh token (and the access token next to it) works only while the
+account is live — active, not deleted — and the password is the one it was
+issued under (the ``phv`` claim). Refreshing never extends a sign-in: the
+answer carries the presented refresh token back.
+
+Same self-contained harness as :mod:`tests.test_password_reset`: a StaticPool
+SQLite engine, the ``get_db`` override for this module, a wipe per test.
+"""
+
+from __future__ import annotations
+
+import asyncio
+from datetime import datetime, timedelta, timezone
+
+import pytest
+from jose import jwt
+from sqlalchemy import delete, update
+from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
+from sqlalchemy.pool import StaticPool
+
+from app.core.auth import hash_password
+from app.core.config import settings
+from app.core.tokens import ALGORITHM, password_hash_version
+from app.db.base import Base, get_db
+from app.db.models import User
+from app.main import app
+
+_test_engine = create_async_engine(
+    "sqlite+aiosqlite:///:memory:",
+    connect_args={"check_same_thread": False},
+    poolclass=StaticPool,
+    echo=False,
+)
+_TestSession = async_sessionmaker(_test_engine, expire_on_commit=False, class_=AsyncSession)
+
+_PASSWORD = "refresh-test-password"
+
+
+async def _setup_schema() -> None:
+    async with _test_engine.begin() as conn:
+        await conn.run_sync(Base.metadata.create_all)
+
+
+async def _reset_tables() -> None:
+    async with _TestSession() as s:
+        await s.execute(delete(User))
+        await s.commit()
+
+
+async def _override_get_db():
+    async with _TestSession() as session:
+        yield session
+
+
+@pytest.fixture(scope="module", autouse=True)
+def _install_overrides():
+    asyncio.run(_setup_schema())
+    app.dependency_overrides[get_db] = _override_get_db
+    yield
+    app.dependency_overrides.pop(get_db, None)
+
+
+@pytest.fixture(autouse=True)
+def _wipe_between_tests():
+    asyncio.run(_reset_tables())
+    yield
+
+
+async def _insert_user(email: str) -> User:
+    async with _TestSession() as s:
+        user = User(email=email, password_hash=hash_password(_PASSWORD))
+        s.add(user)
+        await s.commit()
+        await s.refresh(user)
+        return user
+
+
+async def _deactivate(user: User) -> None:
+    async with _TestSession() as s:
+        await s.execute(update(User).where(User.id == user.id).values(is_active=False))
+        await s.commit()
+
+
+async def _remove(user: User) -> None:
+    async with _TestSession() as s:
+        await s.execute(delete(User).where(User.id == user.id))
+        await s.commit()
+
+
+def _login(client, email: str) -> dict:
+    res = client.post("/api/v1/auth/login", json={"email": email, "password": _PASSWORD})
+    assert res.status_code == 200, res.text
+    return res.json()
+
+
+def _hand_mint(user: User, token_type: str, **claims) -> str:
+    """Sign a token the way ``app/core/tokens.py`` does — real secret, issuer
+    and audience — so only the claims given here differ from a minted one."""
+    base = {
+        "sub": str(user.id),
+        "exp": datetime.now(timezone.utc) + timedelta(minutes=10),
+        "type": token_type,
+        "iss": settings.jwt_issuer,
+        "aud": settings.jwt_audience,
+    }
+    return jwt.encode({**base, **claims}, settings.jwt_secret, algorithm=ALGORITHM)
+
+
+def test_refresh_issues_a_working_access_token(client):
+    asyncio.run(_insert_user("live@example.com"))
+    tokens = _login(client, "live@example.com")
+
+    res = client.post("/api/v1/auth/refresh", json={"refresh_token": tokens["refresh_token"]})
+    assert res.status_code == 200, res.text
+    me = client.get(
+        "/api/v1/auth/me", headers={"Authorization": f"Bearer {res.json()['access_token']}"}
+    )
+    assert me.status_code == 200
+    assert me.json()["email"] == "live@example.com"
+
+
+def test_refresh_hands_back_the_presented_refresh_token(client):
+    """Refreshing never extends a sign-in: the answer carries the refresh
+    token that was sent, so the sign-in ends when that token expires — 30
+    days after login at the latest. The token expires in one day here: one
+    minted at login would be byte-identical to a freshly minted one issued
+    in the same second. It also works twice in a row, as it must — the web
+    UI can refresh from several requests at once."""
+    user = asyncio.run(_insert_user("lifetime@example.com"))
+    presented = _hand_mint(
+        user,
+        "refresh",
+        exp=datetime.now(timezone.utc) + timedelta(days=1),
+        phv=password_hash_version(user.password_hash),
+    )
+
+    for _ in range(2):
+        res = client.post("/api/v1/auth/refresh", json={"refresh_token": presented})
+        assert res.status_code == 200, res.text
+        assert res.json()["refresh_token"] == presented
+
+
+@pytest.mark.parametrize("end_account", [_deactivate, _remove], ids=["deactivated", "removed"])
+def test_refresh_rejects_an_account_that_is_gone(client, end_account):
+    user = asyncio.run(_insert_user("gone@example.com"))
+    tokens = _login(client, "gone@example.com")
+    asyncio.run(end_account(user))
+
+    res = client.post("/api/v1/auth/refresh", json={"refresh_token": tokens["refresh_token"]})
+    assert res.status_code == 401
+
+
+def test_tokens_without_a_password_binding_are_rejected(client):
+    """A token without a ``phv`` claim is refused outright: there is nothing
+    to compare with the current password hash."""
+    user = asyncio.run(_insert_user("legacy@example.com"))
+    access = _hand_mint(user, "access", role="user")
+    refresh = _hand_mint(user, "refresh")
+
+    me = client.get("/api/v1/auth/me", headers={"Authorization": f"Bearer {access}"})
+    assert me.status_code == 401
+    assert me.json()["detail"] == "Invalid token."
+    res = client.post("/api/v1/auth/refresh", json={"refresh_token": refresh})
+    assert res.status_code == 401
+    assert res.json()["detail"] == "Invalid token."
+
+
+def test_refresh_needs_a_database(client):
+    """Without a database there are no accounts, so there is nothing to
+    refresh — 503, like login."""
+    user = asyncio.run(_insert_user("nodb@example.com"))
+    token = _hand_mint(user, "refresh", phv=password_hash_version(user.password_hash))
+    saved_override = app.dependency_overrides.pop(get_db)
+    try:
+        res = client.post("/api/v1/auth/refresh", json={"refresh_token": token})
+    finally:
+        app.dependency_overrides[get_db] = saved_override
+    assert res.status_code == 503
diff --git a/tests/test_email_verification.py b/tests/test_email_verification.py
index 44ae3af..448a731 100644
--- a/tests/test_email_verification.py
+++ b/tests/test_email_verification.py
@@ -168,8 +168,8 @@ def test_token_rejects_wrong_type():
 
     for bogus in (
         create_password_reset_token("user-123", "phv-deadbeef"),
-        create_access_token("user-123"),
-        create_refresh_token("user-123"),
+        create_access_token("user-123", phv="phv-deadbeef"),
+        create_refresh_token("user-123", phv="phv-deadbeef"),
     ):
         with pytest.raises(HTTPException) as exc:
             decode_email_verify_token(bogus)
diff --git a/tests/test_jwt_claims.py b/tests/test_jwt_claims.py
index 286b673..646580e 100644
--- a/tests/test_jwt_claims.py
+++ b/tests/test_jwt_claims.py
@@ -60,6 +60,7 @@ def _hand_mint(claims: dict) -> str:
         "exp": datetime.now(timezone.utc) + timedelta(minutes=10),
         "type": "access",
         "role": "user",
+        "phv": "phv-1",
     }
     return jwt.encode({**base, **claims}, settings.jwt_secret, algorithm=ALGORITHM)
 
@@ -68,13 +69,13 @@ def _hand_mint(claims: dict) -> str:
 
 
 def test_access_token_carries_iss_and_aud():
-    claims = _decode_unverified(create_access_token("user-1"))
+    claims = _decode_unverified(create_access_token("user-1", phv="phv-1"))
     assert claims["iss"] == settings.jwt_issuer
     assert claims["aud"] == settings.jwt_audience
 
 
 def test_refresh_token_carries_iss_and_aud():
-    claims = _decode_unverified(create_refresh_token("user-1"))
+    claims = _decode_unverified(create_refresh_token("user-1", phv="phv-1"))
     assert claims["iss"] == settings.jwt_issuer
     assert claims["aud"] == settings.jwt_audience
 
@@ -95,11 +96,14 @@ def test_verify_token_carries_iss_and_aud():
 
 
 def test_access_token_round_trips():
-    assert decode_token(create_access_token("user-42")) == "user-42"
+    assert decode_token(create_access_token("user-42", phv="phv-1")) == "user-42"
 
 
 def test_refresh_token_round_trips():
-    assert decode_token(create_refresh_token("user-42"), expected_type="refresh") == "user-42"
+    assert (
+        decode_token(create_refresh_token("user-42", phv="phv-1"), expected_type="refresh")
+        == "user-42"
+    )
 
 
 def test_reset_token_round_trips():
@@ -172,6 +176,7 @@ def test_access_decoder_rejects_token_without_iss_or_aud():
             "exp": datetime.now(timezone.utc) + timedelta(minutes=10),
             "type": "access",
             "role": "user",
+            "phv": "phv-1",
         },
         settings.jwt_secret,
         algorithm=ALGORITHM,
diff --git a/tests/test_password_reset.py b/tests/test_password_reset.py
index 0733935..6d9f1f9 100644
--- a/tests/test_password_reset.py
+++ b/tests/test_password_reset.py
@@ -246,3 +246,66 @@ def test_reset_password_short_password_rejected(client, mock_send_email):
     )
     # Pydantic validator fails before the route body runs → 422.
     assert res.status_code == 422
+
+
+# ── /reset-password — sessions ─────────────────────────────────────────────────
+
+
+def _login(client, email: str, password: str) -> dict:
+    res = client.post("/api/v1/auth/login", json={"email": email, "password": password})
+    assert res.status_code == 200, res.text
+    return res.json()
+
+
+def _reset_via_email_link(client, mock_send_email, email: str, new_password: str) -> None:
+    client.post("/api/v1/auth/forgot-password", json={"email": email})
+    token = _extract_token_from_mock(mock_send_email)
+    res = client.post(
+        "/api/v1/auth/reset-password",
+        json={"token": token, "new_password": new_password},
+    )
+    assert res.status_code == 200, res.text
+
+
+def _bearer(access_token: str) -> dict[str, str]:
+    return {"Authorization": f"Bearer {access_token}"}
+
+
+def test_reset_password_ends_sessions_from_before_the_reset(client, mock_send_email):
+    """A reset signs out every session that existed before it — the access
+    token and the refresh token alike, so the refresh token can't mint a new
+    access token either."""
+    asyncio.run(_insert_user(email="sess@example.com", password="old-password-123"))
+    before = _login(client, "sess@example.com", "old-password-123")
+
+    _reset_via_email_link(client, mock_send_email, "sess@example.com", "new-password-456")
+
+    me = client.get("/api/v1/auth/me", headers=_bearer(before["access_token"]))
+    assert me.status_code == 401
+    refreshed = client.post("/api/v1/auth/refresh", json={"refresh_token": before["refresh_token"]})
+    assert refreshed.status_code == 401
+
+
+def test_sessions_started_after_the_reset_work(client, mock_send_email):
+    asyncio.run(_insert_user(email="after@example.com", password="old-password-123"))
+    _reset_via_email_link(client, mock_send_email, "after@example.com", "new-password-456")
+
+    after = _login(client, "after@example.com", "new-password-456")
+    assert client.get("/api/v1/auth/me", headers=_bearer(after["access_token"])).status_code == 200
+
+    refreshed = client.post("/api/v1/auth/refresh", json={"refresh_token": after["refresh_token"]})
+    assert refreshed.status_code == 200, refreshed.text
+    me = client.get("/api/v1/auth/me", headers=_bearer(refreshed.json()["access_token"]))
+    assert me.status_code == 200
+    assert me.json()["email"] == "after@example.com"
+
+
+def test_every_new_hash_changes_the_phv():
+    """Reset links and sessions both end because a new password hash has a
+    new ``phv``. Hashing the same password twice must therefore give two
+    fingerprints — a hash format whose first 16 characters never vary (every
+    argon2id hash starts ``$argon2id$v=19$m``) would silently keep both
+    alive."""
+    first = password_hash_version(hash_password("same-password-123"))
+    second = password_hash_version(hash_password("same-password-123"))
+    assert first != second
diff --git a/tests/test_rate_limit_routes.py b/tests/test_rate_limit_routes.py
index de9e3b3..264e423 100644
--- a/tests/test_rate_limit_routes.py
+++ b/tests/test_rate_limit_routes.py
@@ -54,7 +54,7 @@
 from app.core.auth import hash_password
 from app.core.config import settings
 from app.core.rate_limit import account_or_ip, limiter
-from app.core.tokens import create_access_token, create_refresh_token
+from app.core.tokens import create_access_token, create_refresh_token, password_hash_version
 from app.db.base import Base, get_db
 from app.db.models import ApiKey, AuditEvent, TierEnum, User
 from app.main import app
@@ -170,7 +170,9 @@ async def _insert_user(
 
 
 def _bearer_headers(user: User) -> dict[str, str]:
-    token = create_access_token(str(user.id), role=user.role.value)
+    token = create_access_token(
+        str(user.id), phv=password_hash_version(user.password_hash), role=user.role.value
+    )
     return {"Authorization": f"Bearer {token}"}
 
 
@@ -464,25 +466,15 @@ def test_auth_refresh_is_exempt_from_rate_limiting(client, rate_limiter_enabled)
     session. 350 garbage tokens all fail at JWT decode (401, before any DB
     access) and must never turn into 429; a VALID token right after must
     still succeed.
-
-    The valid-token call runs with the ``get_db`` override popped: ``refresh``
-    compares the JWT subject (a plain ``str``) against the UUID ``User.id``
-    column with no ``uuid.UUID(...)`` cast, unlike ``get_current_user`` (which
-    casts defensively). That still raises under the SQLite test engine
-    whenever a database is configured — a separate, pre-existing issue,
-    unrelated to rate limiting and not fixed here.
     """
     for i in range(350):
         res = client.post("/api/v1/auth/refresh", json={"refresh_token": "x"})
         assert res.status_code == 401, f"junk request {i + 1}/350: {res.text}"
 
-    token = create_refresh_token(str(uuid4()))
-    saved_override = app.dependency_overrides.pop(get_db)
-    try:
-        res = client.post("/api/v1/auth/refresh", json={"refresh_token": token})
-        assert res.status_code == 200, res.text
-    finally:
-        app.dependency_overrides[get_db] = saved_override
+    user = asyncio.run(_insert_user(email=f"rl-refresh-{uuid4().hex}@example.com"))
+    token = create_refresh_token(str(user.id), phv=password_hash_version(user.password_hash))
+    res = client.post("/api/v1/auth/refresh", json={"refresh_token": token})
+    assert res.status_code == 200, res.text
 
 
 # ══ Privacy: slowapi's own logger must never leak an IP or account ═══════════
diff --git a/tests/test_token_types.py b/tests/test_token_types.py
index 7d6c69d..2970870 100644
--- a/tests/test_token_types.py
+++ b/tests/test_token_types.py
@@ -35,7 +35,7 @@
     create_refresh_token,
     decode_email_verify_token,
     decode_password_reset_token,
-    decode_token_full,
+    decode_session_token,
     password_hash_version,
 )
 
@@ -45,8 +45,8 @@
 # every other decoder with the same token and assert it raises.
 
 _TYPES: list[tuple[str, callable]] = [
-    ("access", lambda: create_access_token("user-id-1", role="user")),
-    ("refresh", lambda: create_refresh_token("user-id-1")),
+    ("access", lambda: create_access_token("user-id-1", phv="phv-1", role="user")),
+    ("refresh", lambda: create_refresh_token("user-id-1", phv="phv-1")),
     (
         "reset",
         lambda: create_password_reset_token(
@@ -58,11 +58,11 @@
 
 
 def _decode_as_access(token: str) -> None:
-    decode_token_full(token, expected_type="access")
+    decode_session_token(token, expected_type="access")
 
 
 def _decode_as_refresh(token: str) -> None:
-    decode_token_full(token, expected_type="refresh")
+    decode_session_token(token, expected_type="refresh")
 
 
 def _decode_as_reset(token: str) -> None:
diff --git a/tests/test_upload_auth_resolution.py b/tests/test_upload_auth_resolution.py
index 4465eef..55fa375 100644
--- a/tests/test_upload_auth_resolution.py
+++ b/tests/test_upload_auth_resolution.py
@@ -26,16 +26,18 @@
 
 import asyncio
 import io
-from datetime import datetime, timezone
+from datetime import datetime, timedelta, timezone
 
 import pytest
+from jose import jwt
 from sqlalchemy import delete, select, update
 from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
 from sqlalchemy.pool import StaticPool
 
 from app.core.auth import hash_password
+from app.core.config import settings
 from app.core.security import validate_api_key
-from app.core.tokens import create_access_token
+from app.core.tokens import ALGORITHM, create_access_token, password_hash_version
 from app.db.base import Base, get_db
 from app.db.models import ApiKey, TierEnum, User
 from app.main import app
@@ -123,7 +125,9 @@ def _mint_dashboard_key(client, user: User) -> str:
     """Mint a key the way the dashboard does (``POST /api/v1/keys``). It
     lands only in the DB ``api_keys`` table — never in the file store that
     self-host/CLI keys live in."""
-    token = create_access_token(str(user.id), role=user.role.value)
+    token = create_access_token(
+        str(user.id), phv=password_hash_version(user.password_hash), role=user.role.value
+    )
     res = client.post(
         "/api/v1/keys", json={"label": "cli"}, headers={"Authorization": f"Bearer {token}"}
     )
@@ -162,7 +166,9 @@ def test_batch_resolves_bearer_jwt_to_user_tier(client, sample_jpg):
     ``tier_for(user)`` sees ``business`` (limit 100) and a 2-file batch
     is not rejected with ``tier limit of 1``."""
     user = asyncio.run(_insert_business_user())
-    token = create_access_token(str(user.id), role=user.role.value)
+    token = create_access_token(
+        str(user.id), phv=password_hash_version(user.password_hash), role=user.role.value
+    )
 
     res = client.post(
         "/api/v1/convert/batch",
@@ -217,6 +223,55 @@ def test_batch_anonymous_still_capped_at_one(client, sample_jpg):
     assert "tier limit of 1" in res.json()["detail"]
 
 
+def _token_from_before_a_password_change(client, user: User) -> str:
+    login = client.post(
+        "/api/v1/auth/login", json={"email": user.email, "password": "test-password"}
+    )
+    assert login.status_code == 200, login.text
+    asyncio.run(_update_owner(user, password_hash=hash_password("changed-password")))
+    return login.json()["access_token"]
+
+
+def _token_without_a_password_binding(client, user: User) -> str:
+    """An access token without the ``phv`` claim."""
+    claims = {
+        "sub": str(user.id),
+        "exp": datetime.now(timezone.utc) + timedelta(minutes=10),
+        "type": "access",
+        "role": "user",
+        "iss": settings.jwt_issuer,
+        "aud": settings.jwt_audience,
+    }
+    return jwt.encode(claims, settings.jwt_secret, algorithm=ALGORITHM)
+
+
+@pytest.mark.parametrize(
+    "stale_token",
+    [_token_from_before_a_password_change, _token_without_a_password_binding],
+    ids=["password-changed", "no-phv-claim"],
+)
+def test_batch_ignores_a_bearer_that_no_longer_stands_for_the_account(
+    client, sample_jpg, stale_token
+):
+    """``get_optional_user`` resolves a Bearer token through the same check as
+    the account routes. A token issued before the password changed — or one
+    without the ``phv`` claim at all — no longer stands for the account: the
+    request falls back to the anonymous tier (no 401, no 500), where a
+    two-file batch hits the limit of 1."""
+    # Login validates the address, which rejects the reserved ``.test`` TLD.
+    user = asyncio.run(_insert_business_user(email="biz@example.com"))
+    token = stale_token(client, user)
+
+    res = client.post(
+        "/api/v1/convert/batch",
+        files=_two_jpegs(sample_jpg),
+        data={"target_formats": ["png", "png"]},
+        headers={"Authorization": f"Bearer {token}"},
+    )
+    assert res.status_code == 400
+    assert "tier limit of 1" in res.json()["detail"]
+
+
 # ── Upload gate (require_api_key) ─────────────────────────────────────────────