---
phase: 07.2-security-jti-claim-redis-access-token-revocation-inserted
plan: 02
type: execute
wave: 1
depends_on:
- 07.2-01
files_modified:
- backend/services/auth.py
- backend/deps/auth.py
autonomous: true
requirements:
- CONCERNS:JTI-CLAIM
- CONCERNS:JTI-REVOKE-REDIS
tags:
- security
- jwt
- redis
- access-control
must_haves:
truths:
- "Every access token issued by create_access_token contains a unique jti UUID claim (D-01)"
- "get_current_user rejects any access token whose iat predates user_nbf:{user_id} in Redis with HTTP 401 (D-02)"
- "get_current_user accepts tokens whose iat post-dates user_nbf (intended recovery path after refresh)"
- "A Redis outage during the NBF check does not block requests — fail-open with warning log (D-04)"
- "The HTTPException raised by the NBF check is NOT swallowed by the fail-open broad-catch (Pitfall 1)"
artifacts:
- path: "backend/services/auth.py"
provides: "create_access_token with jti=str(uuid.uuid4()) in payload"
contains: "jti"
- path: "backend/deps/auth.py"
provides: "user_nbf Redis check in get_current_user with fail-open + HTTPException re-raise guard"
contains: "user_nbf"
key_links:
- from: "backend/services/auth.py create_access_token"
to: "PyJWT encode payload"
via: "payload dict with new 'jti' key"
pattern: "\"jti\":\\s*str\\(uuid\\.uuid4"
- from: "backend/deps/auth.py get_current_user"
to: "request.app.state.redis"
via: "await redis.get(f\"user_nbf:{payload['sub']}\") + int comparison vs payload['iat']"
pattern: "user_nbf:|app\\.state\\.redis"
- from: "Wave 1 implementation"
to: "Wave 0 xfail stubs (test_auth_deps NBF check, test_task2_auth_service JTI)"
via: "stubs flip to XPASS or are promoted by Wave 2/end-of-phase cleanup"
pattern: "XPASS|xfail.*strict=False"
---
Add the jti claim to access token issuance and the user_nbf revocation check
to the central auth dependency.
Purpose: Closes the 15-minute window between Phase 7.1's refresh-token
revocation and natural access-token expiry. Implements D-01 (jti claim) and
D-02/D-03/D-04 (user-level NBF check with fail-open) verbatim.
This plan is single-concern (token issuance + validation) and self-contained:
no API handler or admin code is touched. Wave 2 (Plan 03) wires the actual
`user_nbf` writes into the four security-event handlers.
Output: Two surgical edits — one to services/auth.py (add jti), one to
deps/auth.py (add NBF check block). Wave 0's XFAIL stubs for jti-claim and
NBF-check flip to XPASS. Existing 373-test baseline stays green.
@$HOME/.claude/get-shit-done/workflows/execute-plan.md
@$HOME/.claude/get-shit-done/templates/summary.md
@.planning/phases/07.2-security-jti-claim-redis-access-token-revocation-inserted/07.2-CONTEXT.md
@.planning/phases/07.2-security-jti-claim-redis-access-token-revocation-inserted/07.2-RESEARCH.md
@backend/services/auth.py
@backend/deps/auth.py
@.planning/phases/07.2-security-jti-claim-redis-access-token-revocation-inserted/07.2-01-PLAN.md
def create_access_token(user_id: str, role: str) -> str:
- Returns jwt.encode(payload, settings.secret_key, algorithm="HS256")
- Current payload keys: sub, role, typ, iat, exp
- Wave 1 adds: jti=str(uuid.uuid4())
- uuid is ALREADY imported at module top (services/auth.py:25)
- Settings used: settings.access_token_expire_minutes (already in scope)
async def get_current_user(
request: Request, # already present, no change
credentials: HTTPAuthorizationCredentials = Depends(security),
session: AsyncSession = Depends(get_db),
) -> User:
- Step 1: payload = auth_service.decode_access_token(credentials.credentials) — raises ValueError → 401
- Step 2 (NEW Wave 1): user_nbf check via request.app.state.redis
- Step 3 (existing): uuid.UUID(payload["sub"]) → 401 on KeyError/ValueError
- Step 4 (existing): session.get(User, user_uuid); reject if None or not is_active → 401
- Step 5 (existing): request.state.current_user = user; return user
try:
# ... call ...
except Exception as exc:
logger.warning("HIBP check failed (fail-open): %s", exc)
return False
settings.access_token_expire_minutes: int = 15 (confirmed by RESEARCH.md L182)
TTL for user_nbf key = settings.access_token_expire_minutes * 60 = 900
NOTE: this Wave does NOT write user_nbf — only reads. TTL constant is used by Wave 2 writers.
redis_client = request.app.state.redis
nbf_bytes = await redis_client.get(f"user_nbf:{payload['sub']}") # returns bytes | None
# Compare:
if nbf_bytes is not None and payload["iat"] < int(nbf_bytes.decode()):
raise HTTPException(401, "Session invalidated", headers={"WWW-Authenticate": "Bearer"})
After jwt.decode():
- payload["iat"] is type int (Unix timestamp)
- payload["jti"] is verbatim the string we passed in (PyJWT does not validate jti format)
- Direct integer comparison: payload["iat"] < int(nbf_bytes.decode()) works correctly
Task 1: Add jti claim to create_access_token
backend/services/auth.py
- backend/services/auth.py lines 1–30 (module docstring + imports — confirm `import uuid` already present at line 25)
- backend/services/auth.py lines 86–117 (create_access_token + decode_access_token — current implementations)
- backend/tests/test_task2_auth_service.py (Wave 0 stub test_create_access_token_includes_jti_claim — this is the test that must flip from XFAIL to XPASS or PASS)
- .planning/phases/07.2-security-jti-claim-redis-access-token-revocation-inserted/07.2-RESEARCH.md "Pattern 1" section (lines 143–166)
- Test 1 (positive): test_create_access_token_includes_jti_claim — decoded payload contains key "jti"; value is a string parseable as uuid.UUID; promoted from Wave 0 XFAIL to PASS (remove the xfail decorator in this task).
- Test 2 (uniqueness): two consecutive calls to create_access_token produce tokens whose decoded jti values differ — add a NEW small test alongside the promotion edit (one-line guard: `assert t1["jti"] != t2["jti"]`).
- Test 3 (regression guard): existing test_create_access_token_jwt_format and test_decode_access_token_valid still pass — the payload remains a valid HS256 JWT and existing keys (sub, role, typ, iat, exp) are unchanged.
Edit `create_access_token` (backend/services/auth.py line 86–99): add exactly one key to the `payload` dict, placed after the existing `exp` line for grep-ability:
`"jti": str(uuid.uuid4()),`
Do NOT change the function signature. Do NOT change algorithm. Do NOT add a TTL constant here (Wave 2 handles user_nbf TTL).
Promote the Wave 0 stub `test_create_access_token_includes_jti_claim` in `backend/tests/test_task2_auth_service.py`: remove the `@pytest.mark.xfail(...)` decorator and replace any placeholder body with the real assertions:
- decoded = decode_access_token(token)
- `assert "jti" in decoded`
- `assert isinstance(decoded["jti"], str)`
- `uuid.UUID(decoded["jti"]) # raises ValueError if not a valid UUID`
Append one new test alongside it: `def test_create_access_token_jti_is_unique_per_call():` — mint two tokens for the same user, decode both, assert `t1["jti"] != t2["jti"]`. Imports per existing in-function-scope convention in this test file.
Do NOT modify decode_access_token — PyJWT decodes the jti claim automatically as part of the payload dict.
cd backend && pytest tests/test_task2_auth_service.py -v
- `cd backend && pytest tests/test_task2_auth_service.py -v` exits 0 with zero failures
- `grep -E "\"jti\":\\s*str\\(uuid\\.uuid4" backend/services/auth.py` returns 1 match
- `grep -v '^\\s*#' backend/services/auth.py | grep -E "\"jti\"" | head -1` shows the new jti line in create_access_token
- test_create_access_token_includes_jti_claim reports PASSED (no longer XFAIL)
- test_create_access_token_jti_is_unique_per_call reports PASSED
- test_create_access_token_jwt_format and test_decode_access_token_valid still PASSED (no regressions)
- Python REPL check: `from services.auth import create_access_token, decode_access_token; import uuid; t = create_access_token('u1','user'); p = decode_access_token(t); uuid.UUID(p['jti'])` returns a UUID object without raising
Every access token contains a unique jti UUID claim; Wave 0 stub promoted; zero existing test regressions.
Task 2: Add user_nbf Redis check to get_current_user with fail-open guard
backend/deps/auth.py
- backend/deps/auth.py (entire file — 115 lines; current get_current_user at lines 38–80)
- backend/services/auth.py lines 286–297 (existing TOTP replay pattern — the reference template for `await redis.get / await redis.set` and `bytes | None` return type)
- backend/services/auth.py near line 394 (check_hibp fail-open template — confirm exact except-clause pattern with logger.warning)
- backend/api/auth.py lines 195–200 and 564–568 (existing `request.app.state.redis` access pattern — confirm direct attribute access without dependency injection)
- .planning/phases/07.2-security-jti-claim-redis-access-token-revocation-inserted/07.2-RESEARCH.md "Pattern 3" section (lines 186–235) AND "Pitfall 1" section (lines 279–284) AND "Pitfall 2" section (lines 286–290)
- backend/tests/test_auth_deps.py (three Wave 0 stubs created in Plan 01 — must flip from XFAIL to PASS in this task)
- Test 1 (reject path): `test_get_current_user_rejects_token_when_iat_before_user_nbf` — pre-populates `app.state.redis` key `user_nbf:{uid}` = `b""`, calls /test/me with a token whose iat (encoded via `create_access_token`, hence `int(now())`) is less than `ts_future`, expects HTTP 401 with response JSON `{"detail": "Session invalidated"}` and `WWW-Authenticate: Bearer` header. Promote from XFAIL.
- Test 2 (allow path): `test_get_current_user_allows_token_when_iat_after_user_nbf` — pre-populates `user_nbf:{uid}` = `b""`, calls /test/me with a token whose iat > ts_past, expects 200. Promote from XFAIL.
- Test 3 (fail-open): `test_get_current_user_failopen_on_redis_error` — mounts a broken Redis (raises RuntimeError on .get()), calls /test/me, expects 200 (no 5xx, no 401). Promote from XFAIL.
- Test 4 (regression guard): all 34+ existing tests in test_auth_deps.py still PASS (token decode, missing-user 401, deactivated-user 401, admin role check).
- Test 5 (no key path): if `user_nbf:{uid}` key is ABSENT from Redis (the common path), get_current_user must NOT raise — request passes through to the existing DB lookup. This is implicitly covered by the existing passing tests once FakeRedis is mounted (Plan 01 Task 1).
Add a new block to `get_current_user` in backend/deps/auth.py, inserted AFTER the decode_access_token try/except block (currently lines 50–57) and BEFORE the `try: user_uuid = uuid.UUID(payload["sub"])` block (currently line 59).
Implementation requirements (concrete; do NOT inline code in this plan — read RESEARCH.md Pattern 3 for the exact shape):
1. Add module-level imports at top of deps/auth.py:
- `import logging`
- `_logger = logging.getLogger(__name__)` (module-level, after the imports)
2. Inside get_current_user, after the existing decode try/except, add a new try/except block:
- Inner work: `redis_client = request.app.state.redis`; `nbf_bytes = await redis_client.get(f"user_nbf:{payload['sub']}")`; if `nbf_bytes is not None and payload["iat"] < int(nbf_bytes.decode())` → raise `HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Session invalidated", headers={"WWW-Authenticate": "Bearer"})`.
- First except: `except HTTPException: raise` — MUST be present and MUST come BEFORE the broad except. This is Pitfall 1 from RESEARCH.md and is also documented in the threat model below (T-7.2-02).
- Second except: `except Exception as exc: _logger.warning("Redis user_nbf check failed (fail-open): %s", exc)` — mirrors the HIBP fail-open template at services/auth.py:~394.
3. Comparison direction MUST be `payload["iat"] < int(nbf_bytes.decode())` → reject (Pitfall 2). Do NOT use `<=` (a token issued at the exact same second as the event is allowed — edge case, but matches the "issued strictly before" semantic).
4. The `nbf_bytes` value handling: be tolerant of both `bytes` and `str` returns from Redis (FakeRedis in tests may return either depending on what was stored). Use `nbf_bytes.decode() if isinstance(nbf_bytes, (bytes, bytearray)) else nbf_bytes` before `int(...)`. Comment this with a reference to FakeRedis in test fixtures.
5. Do NOT add a `get_redis` FastAPI dependency. Per D-03 the access is direct via `request.app.state.redis`.
6. Do NOT change the function signature (request: Request is already present).
7. Do NOT touch `get_current_admin` or `get_regular_user` — they wrap get_current_user via Depends and inherit the check transparently.
Promote the three Wave 0 stubs in backend/tests/test_auth_deps.py:
- Remove the `@pytest.mark.xfail(...)` decorators from all three NBF-check tests created in Plan 01.
- Fill in the placeholder bodies with concrete assertions:
* Reject test: set `auth_client._transport.app.state.redis._store[f"user_nbf:{user.id}"] = (b"9999999999", None)` (FakeRedis tuple format: (value, expiry)) or use `await app.state.redis.set(f"user_nbf:{user.id}", b"9999999999")`. Mint token with create_access_token. GET /test/me with Bearer. Assert resp.status_code == 401 and resp.json()["detail"] == "Session invalidated".
* Allow test: set the key to `b"1"` (Unix ts way in the past). Mint token. Assert resp.status_code == 200.
* Fail-open test: replace app.state.redis with a small `class _BrokenRedis` whose async get/set raise RuntimeError. Mint a token. Assert resp.status_code == 200 (NOT 5xx, NOT 401).
Verify by direct grep that the HTTPException re-raise guard is in place — Pitfall 1 is the single highest-impact correctness risk in this plan.
cd backend && pytest tests/test_auth_deps.py tests/test_auth_api.py -v
- `cd backend && pytest tests/test_auth_deps.py tests/test_auth_api.py -v` exits 0 with zero failures
- `grep -c "user_nbf:" backend/deps/auth.py` returns >= 1
- `grep -E "except HTTPException:\\s*$" backend/deps/auth.py` returns >= 1 line — the re-raise guard is present
- `grep -E "except HTTPException" backend/deps/auth.py` line number is LESS than the `grep -n "except Exception" backend/deps/auth.py` line number (re-raise guard appears BEFORE broad catch — confirm with two grep -n calls and inspect ordering)
- `grep -E "import logging" backend/deps/auth.py` returns >= 1
- `grep -E "_logger.warning|logger.warning" backend/deps/auth.py` returns >= 1
- `grep -c "Session invalidated" backend/deps/auth.py` returns >= 1
- `grep -c "request.app.state.redis" backend/deps/auth.py` returns >= 1
- test_get_current_user_rejects_token_when_iat_before_user_nbf reports PASSED (no longer XFAIL)
- test_get_current_user_allows_token_when_iat_after_user_nbf reports PASSED
- test_get_current_user_failopen_on_redis_error reports PASSED
- All existing test_auth_deps tests (get_current_user_returns_user, deactivated user 401, missing user 401, admin role check) still PASSED
- All existing test_auth_api tests still PASSED (the NBF-write stubs from Plan 01 remain XFAIL — they are Wave 2's job to flip)
get_current_user enforces user_nbf with fail-open + HTTPException re-raise guard; three NBF-check stubs promoted to PASSED; zero regressions in 34+ baseline auth-dep tests.
## Trust Boundaries
| Boundary | Description |
|----------|-------------|
| client → API (Authorization: Bearer) | Untrusted JWT crosses here; signature + typ + NBF all validated in get_current_user |
| API → Redis (app.state.redis) | Trusted local network within Docker Compose; failures must not deny service (D-04 fail-open) |
## STRIDE Threat Register
| Threat ID | Category | Component | Disposition | Mitigation Plan |
|-----------|----------|-----------|-------------|-----------------|
| T-7.2-01 | Elevation of Privilege | get_current_user (NBF check) | mitigate | After successful decode, read `user_nbf:{payload['sub']}` from Redis; if present and `payload["iat"] < int(nbf_bytes.decode())`, raise HTTP 401 "Session invalidated" with WWW-Authenticate header. Closes the 15-minute window where a revoked session's live access token remains valid. |
| T-7.2-02 | Elevation of Privilege | fail-open except block in get_current_user | mitigate | Explicit `except HTTPException: raise` MUST precede the broad `except Exception` — otherwise the intentional 401 from the NBF check is swallowed and a revoked token leaks through. Verified by source assertion (grep ordering) in Task 2 acceptance criteria. |
| T-7.2-03 | Elevation of Privilege | NBF comparison direction | mitigate | Comparison is `payload["iat"] < int(nbf_bytes.decode())` — token issued BEFORE the event is rejected. Pitfall 2 from RESEARCH.md. Verified by test_get_current_user_rejects_token_when_iat_before_user_nbf. |
| T-7.2-04 | Denial of Service / EoP | Redis outage during NBF check | accept | D-04 mandates fail-open: if Redis raises Exception, log a warning and allow the request to proceed. Mirrors the HIBP fail-open pattern (services/auth.py:~394). Availability is prioritized over blocking-during-outage; the surface area is the 15-minute access-token lifetime. |
| T-7.2-05 | Information Disclosure | jti claim in JWT body | accept | jti is a UUIDv4 (no PII, no user identifier). RFC 7519 standard. Visible in any base64-decoded JWT — acceptable. |
| T-7.2-W1-SC | Tampering | npm/pip/cargo installs | n/a | No packages installed in this plan — uses only stdlib (uuid, logging) already imported in scope (RESEARCH.md "Package Legitimacy Audit" intentionally blank) |
- `cd backend && pytest tests/test_auth_deps.py tests/test_task2_auth_service.py -v` — zero failures; 4 stubs from Plan 01 (jti + 3 NBF-check) now PASSED
- `cd backend && pytest -v` — full suite green; zero new failures vs Phase 7.1's 373-test baseline (the 3 NBF-write stubs from Plan 01 Task 3 + 2 admin stubs from Plan 01 Task 4 remain XFAIL — they are Wave 2's job)
- Source assertions (Task 2 acceptance criteria) prove the Pitfall 1 re-raise guard is in place — this is the single most consequential correctness check in the plan
- Every access token issued after this plan deploys contains a `jti` UUID claim (verifiable: decode any new token)
- get_current_user rejects tokens issued before any user_nbf timestamp written for that user (verifiable: pytest tests/test_auth_deps.py -k nbf)
- Redis outages produce log warnings but do not block requests (verifiable: test_get_current_user_failopen_on_redis_error)
- HTTPException raised by the NBF check is NEVER caught by the fail-open broad-except (verifiable: source ordering grep + test behavior)
- Wave 2 (Plan 03) can write `user_nbf:{user_id}` from the four security-event handlers and the check will fire on the next request from any pre-event token