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

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

<objective>
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.
</objective>

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

<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

<interfaces>
<!-- Current create_access_token signature (services/auth.py:86–99) -->

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)

<!-- Current get_current_user signature (deps/auth.py:38–80) -->

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

<!-- Existing fail-open template — services/auth.py check_hibp (~line 394) -->

try:
    # ... call ...
except Exception as exc:
    logger.warning("HIBP check failed (fail-open): %s", exc)
    return False

<!-- Existing TTL constant convention — config.py -->

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 access pattern — already used in api/auth.py:197 and :566 -->

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"})

<!-- PyJWT verified behavior (RESEARCH.md L165) -->

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
</interfaces>
</context>

<tasks>

<task type="auto" tdd="true">
  <name>Task 1: Add jti claim to create_access_token</name>
  <files>backend/services/auth.py</files>
  <read_first>
    - 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)
  </read_first>
  <behavior>
    - 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.
  </behavior>
  <action>
    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.
  </action>
  <verify>
    <automated>cd backend &amp;&amp; pytest tests/test_task2_auth_service.py -v</automated>
  </verify>
  <acceptance_criteria>
    - `cd backend &amp;&amp; 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
  </acceptance_criteria>
  <done>Every access token contains a unique jti UUID claim; Wave 0 stub promoted; zero existing test regressions.</done>
</task>

<task type="auto" tdd="true">
  <name>Task 2: Add user_nbf Redis check to get_current_user with fail-open guard</name>
  <files>backend/deps/auth.py</files>
  <read_first>
    - 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)
  </read_first>
  <behavior>
    - 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"<ts_future>"`, 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"<ts_past>"`, 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).
  </behavior>
  <action>
    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.
  </action>
  <verify>
    <automated>cd backend &amp;&amp; pytest tests/test_auth_deps.py tests/test_auth_api.py -v</automated>
  </verify>
  <acceptance_criteria>
    - `cd backend &amp;&amp; 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 &gt;= 1
    - `grep -E "except HTTPException:\\s*$" backend/deps/auth.py` returns &gt;= 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 &gt;= 1
    - `grep -E "_logger.warning|logger.warning" backend/deps/auth.py` returns &gt;= 1
    - `grep -c "Session invalidated" backend/deps/auth.py` returns &gt;= 1
    - `grep -c "request.app.state.redis" backend/deps/auth.py` returns &gt;= 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)
  </acceptance_criteria>
  <done>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.</done>
</task>

</tasks>

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

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

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

<output>
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).
</output>