kite/.planning/phases/07.2-security-jti-claim-redis-access-token-revocation-inserted/07.2-02-PLAN.md

phase, plan, type, wave, depends_on, files_modified, autonomous, requirements, tags, must_haves

phase	plan	type	wave	depends_on	files_modified	autonomous	requirements	tags	must_haves
07.2-security-jti-claim-redis-access-token-revocation-inserted	02	execute	1	07.2-01	backend/services/auth.py backend/deps/auth.py	true	CONCERNS:JTI-CLAIM CONCERNS:JTI-REVOKE-REDIS	security jwt redis access-control	truths artifacts key_links 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) path provides contains backend/services/auth.py create_access_token with jti=str(uuid.uuid4()) in payload jti path provides contains backend/deps/auth.py user_nbf Redis check in get_current_user with fail-open + HTTPException re-raise guard user_nbf from to via pattern backend/services/auth.py create_access_token PyJWT encode payload payload dict with new 'jti' key "jti":\s*str(uuid.uuid4 from to via pattern backend/deps/auth.py get_current_user request.app.state.redis await redis.get(f"user_nbf:{payload['sub']}") + int comparison vs payload['iat'] user_nbf:|app.state.redis from to via pattern Wave 1 implementation Wave 0 xfail stubs (test_auth_deps NBF check, test_task2_auth_service JTI) stubs flip to XPASS or are promoted by Wave 2/end-of-phase cleanup 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.

<execution_context> @$HOME/.claude/get-shit-done/workflows/execute-plan.md @$HOME/.claude/get-shit-done/templates/summary.md </execution_context>

@.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.

<threat_model>

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)
</threat_model>

- `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

<success_criteria>

• 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 </success_criteria>

Create `.planning/phases/07.2-security-jti-claim-redis-access-token-revocation-inserted/07.2-02-SUMMARY.md` when done.

Required fields in SUMMARY: artifacts (services/auth.py + deps/auth.py), patterns_established (NBF check pattern in get_current_user; HTTPException re-raise guard before broad catch; bytes/str tolerance in Redis return values), patterns_to_avoid (do NOT use <= in NBF comparison; do NOT add a get_redis dependency; do NOT touch get_current_admin/get_regular_user), provides (jti claim issuance + NBF rejection enforcement — ready for Wave 2 write sites).