docs(07.4): create phase plan — token fingerprinting / token binding
2 plans (2 waves): Wave 0 xfail stubs for test_auth_fgp.py (4 FGP tests), Wave 1 production implementation (_compute_fgp helper, create_access_token fgp claim, get_current_user fgp validation, login/refresh call-site updates, test promotion) + version bump to 0.1.3. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
co-authored by
Claude Sonnet 4.6
parent
de0f3fb7c3
commit
22fcb53d9d
+386
@@ -0,0 +1,386 @@
|
||||
# Phase 7.4: Security — Token Fingerprinting / Token Binding - Pattern Map
|
||||
|
||||
**Mapped:** 2026-06-06
|
||||
**Files analyzed:** 4
|
||||
**Analogs found:** 4 / 4
|
||||
|
||||
---
|
||||
|
||||
## File Classification
|
||||
|
||||
| New/Modified File | Role | Data Flow | Closest Analog | Match Quality |
|
||||
|-------------------|------|-----------|----------------|---------------|
|
||||
| `backend/services/auth.py` | service | request-response | `backend/services/auth.py` itself (extend existing) | exact — same file, additive change |
|
||||
| `backend/deps/auth.py` | middleware/dependency | request-response | `backend/deps/auth.py` itself (extend user_nbf block) | exact — same file, additive block |
|
||||
| `backend/api/auth.py` | controller | request-response | `backend/api/auth.py` itself (update 2 call sites) | exact — same file, call-site tweak |
|
||||
| `backend/tests/test_auth_fgp.py` | test | request-response | `backend/tests/test_auth_deps.py` (NBF test pattern) | exact — same app harness, same fixture style |
|
||||
|
||||
---
|
||||
|
||||
## Pattern Assignments
|
||||
|
||||
### `backend/services/auth.py` (service, request-response)
|
||||
|
||||
**Analog:** Same file — additive change. Extend `create_access_token` and add `_compute_fgp` helper before it.
|
||||
|
||||
**Existing imports block** (lines 18-39):
|
||||
```python
|
||||
from __future__ import annotations
|
||||
|
||||
import base64
|
||||
import hashlib
|
||||
import hmac
|
||||
import logging
|
||||
import re
|
||||
import secrets
|
||||
import uuid
|
||||
from datetime import datetime, timezone, timedelta
|
||||
from typing import Optional
|
||||
|
||||
import httpx
|
||||
import jwt
|
||||
import pyotp
|
||||
from pwdlib import PasswordHash
|
||||
from pwdlib.hashers.argon2 import Argon2Hasher
|
||||
from sqlalchemy import select, update
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from config import settings
|
||||
from db.models import BackupCode, Quota, RefreshToken, User
|
||||
```
|
||||
Both `import hashlib` (line 21) and `import hmac` (line 22) are already present. No new imports required.
|
||||
|
||||
**Existing `create_access_token` signature** (line 87):
|
||||
```python
|
||||
def create_access_token(user_id: str, role: str) -> str:
|
||||
```
|
||||
|
||||
**Existing JWT payload dict** (lines 93-100) — `"fgp"` claim inserts here alongside existing claims:
|
||||
```python
|
||||
payload = {
|
||||
"sub": str(user_id),
|
||||
"role": role,
|
||||
"typ": "access",
|
||||
"iat": now,
|
||||
"exp": now + timedelta(minutes=settings.access_token_expire_minutes),
|
||||
"jti": str(uuid.uuid4()),
|
||||
}
|
||||
```
|
||||
|
||||
**Existing `hmac.compare_digest` usage pattern** (lines 419-425) — copy constant-time comparison idiom for the fgp check in `deps/auth.py`:
|
||||
```python
|
||||
if hmac.compare_digest(candidate_suffix.upper(), suffix):
|
||||
```
|
||||
|
||||
**New `_compute_fgp` helper to add** (insert before `create_access_token` at line 87, per D-04):
|
||||
```python
|
||||
def _compute_fgp(user_agent: str, accept_lang: str) -> str:
|
||||
"""Return 16-char hex fingerprint binding a token to its client context (D-04)."""
|
||||
return hmac.new(
|
||||
settings.secret_key.encode(),
|
||||
(user_agent + accept_lang).encode(),
|
||||
hashlib.sha256,
|
||||
).hexdigest()[:16]
|
||||
```
|
||||
|
||||
**Updated `create_access_token` signature** (D-05 — backward-compatible, empty-string defaults):
|
||||
```python
|
||||
def create_access_token(
|
||||
user_id: str,
|
||||
role: str,
|
||||
user_agent: str = "",
|
||||
accept_lang: str = "",
|
||||
) -> str:
|
||||
```
|
||||
Add `"fgp": _compute_fgp(user_agent, accept_lang)` to the `payload` dict.
|
||||
|
||||
---
|
||||
|
||||
### `backend/deps/auth.py` (dependency/middleware, request-response)
|
||||
|
||||
**Analog:** Same file — additive block inserted after the existing `user_nbf` check.
|
||||
|
||||
**Existing imports block** (lines 23-33):
|
||||
```python
|
||||
import logging
|
||||
import uuid
|
||||
|
||||
from fastapi import Depends, HTTPException, Request, status
|
||||
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from db.models import User
|
||||
from deps.db import get_db
|
||||
from services import auth as auth_service
|
||||
```
|
||||
`hmac` is NOT imported in `deps/auth.py`. The fgp comparison uses `auth_service._compute_fgp` (called via the already-imported `auth_service`) plus `hmac.compare_digest`. Add `import hmac` at the top of this file.
|
||||
|
||||
**`get_current_user` signature** (lines 41-45) — `request: Request` already present:
|
||||
```python
|
||||
async def get_current_user(
|
||||
request: Request,
|
||||
credentials: HTTPAuthorizationCredentials = Depends(security),
|
||||
session: AsyncSession = Depends(get_db),
|
||||
) -> User:
|
||||
```
|
||||
|
||||
**Existing `user_nbf` block** (lines 62-85) — this is the exact structural model to follow for placement and exception guard order:
|
||||
```python
|
||||
# ── user_nbf check (D-02, D-03, D-04) ──────────────────────────────────────
|
||||
try:
|
||||
redis_client = request.app.state.redis
|
||||
nbf_bytes = await redis_client.get(f"user_nbf:{payload['sub']}")
|
||||
if nbf_bytes is not None:
|
||||
nbf_str = nbf_bytes.decode() if isinstance(nbf_bytes, (bytes, bytearray)) else nbf_bytes
|
||||
if payload["iat"] < int(nbf_str):
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_401_UNAUTHORIZED,
|
||||
detail="Session invalidated",
|
||||
headers={"WWW-Authenticate": "Bearer"},
|
||||
)
|
||||
except HTTPException:
|
||||
raise # re-raise the 401 we just constructed (T-7.2-02: Pitfall 1 guard)
|
||||
except Exception as exc:
|
||||
_logger.warning("Redis user_nbf check failed (fail-open): %s", exc)
|
||||
# ── end user_nbf check ──────────────────────────────────────────────────────
|
||||
```
|
||||
|
||||
**Critical ordering rule** (line 81-84): `except HTTPException: raise` MUST precede `except Exception` — this guard re-raises any intentional HTTPException without it being swallowed. The fgp block raises `HTTPException` directly inside a try block; this guard is what ensures it propagates correctly.
|
||||
|
||||
**fgp validation block to insert after line 85** (per D-06, RESEARCH.md Pattern 3):
|
||||
```python
|
||||
# ── fgp check (D-06, Phase 7.4) ────────────────────────────────────────────
|
||||
fgp_claim = payload.get("fgp", "")
|
||||
if fgp_claim:
|
||||
fgp_actual = auth_service._compute_fgp(
|
||||
request.headers.get("User-Agent", ""),
|
||||
request.headers.get("Accept-Language", ""),
|
||||
)
|
||||
if not hmac.compare_digest(fgp_claim, fgp_actual):
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_401_UNAUTHORIZED,
|
||||
detail="Token fingerprint mismatch",
|
||||
headers={"WWW-Authenticate": "Bearer"},
|
||||
)
|
||||
# ── end fgp check ───────────────────────────────────────────────────────────
|
||||
```
|
||||
|
||||
**Key difference from user_nbf block:** No try/except wrapper needed — `_compute_fgp` is pure computation with no I/O, so there is no infrastructure-failure path requiring fail-open. The fgp mismatch must always be enforced (D-03).
|
||||
|
||||
**Insertion point:** After line 85 (end of `user_nbf` block comment), before line 87 (`try: user_uuid = uuid.UUID(payload["sub"])`).
|
||||
|
||||
---
|
||||
|
||||
### `backend/api/auth.py` (controller, request-response)
|
||||
|
||||
**Analog:** Same file — two targeted call-site updates.
|
||||
|
||||
**Existing imports** (lines 21-42) — no new imports needed; `Request` already imported:
|
||||
```python
|
||||
from fastapi import APIRouter, Depends, HTTPException, Request, Response, status
|
||||
```
|
||||
|
||||
**Login handler signature** (lines 190-197) — `request: Request` already present:
|
||||
```python
|
||||
@router.post("/login")
|
||||
@limiter.limit("10/minute")
|
||||
async def login(
|
||||
request: Request,
|
||||
body: LoginRequest,
|
||||
response: Response,
|
||||
session: AsyncSession = Depends(get_db),
|
||||
):
|
||||
```
|
||||
|
||||
**Login call site** (line 290) — current:
|
||||
```python
|
||||
access_token = auth_service.create_access_token(str(user.id), user.role)
|
||||
```
|
||||
Updated form (pass headers through):
|
||||
```python
|
||||
access_token = auth_service.create_access_token(
|
||||
str(user.id),
|
||||
user.role,
|
||||
user_agent=request.headers.get("User-Agent", ""),
|
||||
accept_lang=request.headers.get("Accept-Language", ""),
|
||||
)
|
||||
```
|
||||
|
||||
**Refresh handler signature** (lines 320-326) — `request: Request` already present:
|
||||
```python
|
||||
@router.post("/refresh")
|
||||
@limiter.limit("10/minute")
|
||||
async def refresh_token(
|
||||
request: Request,
|
||||
response: Response,
|
||||
session: AsyncSession = Depends(get_db),
|
||||
):
|
||||
```
|
||||
|
||||
**Refresh call site** (line 364) — current:
|
||||
```python
|
||||
access_token = auth_service.create_access_token(user_id_str, user.role)
|
||||
```
|
||||
Updated form:
|
||||
```python
|
||||
access_token = auth_service.create_access_token(
|
||||
user_id_str,
|
||||
user.role,
|
||||
user_agent=request.headers.get("User-Agent", ""),
|
||||
accept_lang=request.headers.get("Accept-Language", ""),
|
||||
)
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
### `backend/tests/test_auth_fgp.py` (test, request-response)
|
||||
|
||||
**Analog:** `backend/tests/test_auth_deps.py` — copy the full harness pattern (FakeRedis, `make_test_app`, `auth_client` fixture, `_create_user` helper).
|
||||
|
||||
**Test file module docstring pattern** (lines 1-15 of `test_auth_deps.py`):
|
||||
```python
|
||||
"""
|
||||
Tests for backend/deps/auth.py — FastAPI authentication dependency chain.
|
||||
...
|
||||
"""
|
||||
```
|
||||
|
||||
**Standard imports block** (lines 17-24 of `test_auth_deps.py`):
|
||||
```python
|
||||
import uuid
|
||||
|
||||
import pytest
|
||||
import pytest_asyncio
|
||||
from httpx import ASGITransport, AsyncClient
|
||||
from fastapi import FastAPI, Depends
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from tests.test_auth_api import FakeRedis
|
||||
```
|
||||
New test file also needs: `import hmac`, `import hashlib`, `import base64`, `import time` for crafting tokens without `fgp` claim (FGP-03) and for testing hmac values.
|
||||
|
||||
**`make_test_app()` helper** (lines 29-52 of `test_auth_deps.py`) — copy verbatim; it creates a minimal `/test/me` route wired to `get_current_user`, with `app.state.redis = FakeRedis()`:
|
||||
```python
|
||||
def make_test_app():
|
||||
from deps.auth import get_current_user
|
||||
from db.models import User
|
||||
|
||||
test_app = FastAPI()
|
||||
|
||||
@test_app.get("/test/me")
|
||||
async def get_me(current_user: User = Depends(get_current_user)):
|
||||
return {"id": str(current_user.id), "role": current_user.role}
|
||||
|
||||
test_app.state.redis = FakeRedis()
|
||||
return test_app
|
||||
```
|
||||
|
||||
**`auth_client` fixture** (lines 55-71 of `test_auth_deps.py`) — copy verbatim; sets `get_db` override:
|
||||
```python
|
||||
@pytest_asyncio.fixture
|
||||
async def auth_client(db_session: AsyncSession):
|
||||
from deps.db import get_db
|
||||
|
||||
app = make_test_app()
|
||||
app.dependency_overrides[get_db] = lambda: db_session
|
||||
|
||||
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as c:
|
||||
yield c
|
||||
|
||||
app.dependency_overrides.clear()
|
||||
```
|
||||
|
||||
**`_create_user` helper** (lines 74-89 of `test_auth_deps.py`) — copy verbatim:
|
||||
```python
|
||||
async def _create_user(db_session, role: str = "user", is_active: bool = True):
|
||||
from db.models import User
|
||||
from services.auth import hash_password
|
||||
|
||||
user = User(
|
||||
id=uuid.uuid4(),
|
||||
handle=f"user_{uuid.uuid4().hex[:6]}",
|
||||
email=f"{uuid.uuid4().hex[:6]}@example.com",
|
||||
password_hash=hash_password("testpassword"),
|
||||
role=role,
|
||||
is_active=is_active,
|
||||
)
|
||||
db_session.add(user)
|
||||
await db_session.commit()
|
||||
return user
|
||||
```
|
||||
|
||||
**Standard test function pattern** (lines 94-108 of `test_auth_deps.py`) — `@pytest.mark.asyncio`, inline import of `create_access_token`, `auth_client.get` with `Authorization` header:
|
||||
```python
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_current_user_returns_user(auth_client, db_session):
|
||||
from services.auth import create_access_token
|
||||
|
||||
user = await _create_user(db_session, role="user")
|
||||
token = create_access_token(str(user.id), "user")
|
||||
|
||||
resp = await auth_client.get(
|
||||
"/test/me", headers={"Authorization": f"Bearer {token}"}
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
```
|
||||
|
||||
**FGP-03 — crafting a token without `fgp` claim** — pattern from `test_auth_es256.py` (lines 52-75): use PyJWT directly with the ES256 private key extracted from `settings`. In the test file import `import jwt as _jwt`, `import config`, decode `settings.jwt_private_key` via `base64.b64decode`, and call `_jwt.encode(payload_no_fgp, private_pem, algorithm="ES256")`. The `conftest._patch_es256_test_keys` autouse fixture (session-scoped) is already wired to all tests in `tests/` — no extra fixture needed in the new test file.
|
||||
|
||||
**Access to conftest `_patch_es256_test_keys` autouse fixture** — because it is `autouse=True` in `conftest.py`, every test in `backend/tests/` automatically gets it. The new `test_auth_fgp.py` benefits from it without any explicit reference.
|
||||
|
||||
---
|
||||
|
||||
## Shared Patterns
|
||||
|
||||
### `hmac.compare_digest` — constant-time comparison
|
||||
**Source:** `backend/services/auth.py` line 419
|
||||
**Apply to:** fgp validation block in `backend/deps/auth.py`
|
||||
```python
|
||||
if not hmac.compare_digest(fgp_claim, fgp_actual):
|
||||
```
|
||||
Never use `==` for token/fingerprint comparison. `hmac.compare_digest` is mandatory (CLAUDE.md SEC-06).
|
||||
|
||||
### `except HTTPException: raise` guard — before broad `except Exception`
|
||||
**Source:** `backend/deps/auth.py` lines 81-84
|
||||
**Apply to:** Any try block in `get_current_user` that raises an intentional `HTTPException`
|
||||
```python
|
||||
except HTTPException:
|
||||
raise # re-raise the 401 we just constructed (T-7.2-02: Pitfall 1 guard)
|
||||
except Exception as exc:
|
||||
_logger.warning("Redis user_nbf check failed (fail-open): %s", exc)
|
||||
```
|
||||
The fgp block does NOT need this try/except (pure computation, no I/O). If placed inside the `user_nbf` try block, the guard order already handles it. If placed as a standalone `if` block outside any try, no wrapper is needed at all.
|
||||
|
||||
### `request.headers.get("X", "")` — header extraction with empty-string fallback
|
||||
**Source:** `backend/api/auth.py` — idiomatic FastAPI; `request: Request` already injected in all callers
|
||||
**Apply to:** Both `api/auth.py` call sites and `deps/auth.py` fgp validation block
|
||||
```python
|
||||
request.headers.get("User-Agent", "")
|
||||
request.headers.get("Accept-Language", "")
|
||||
```
|
||||
|
||||
### Error response format for 401
|
||||
**Source:** `backend/deps/auth.py` lines 75-80
|
||||
**Apply to:** fgp mismatch raise
|
||||
```python
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_401_UNAUTHORIZED,
|
||||
detail="Token fingerprint mismatch",
|
||||
headers={"WWW-Authenticate": "Bearer"},
|
||||
)
|
||||
```
|
||||
All 401 raises in `get_current_user` include `headers={"WWW-Authenticate": "Bearer"}`.
|
||||
|
||||
---
|
||||
|
||||
## No Analog Found
|
||||
|
||||
All four files have close analogs. No gaps.
|
||||
|
||||
---
|
||||
|
||||
## Metadata
|
||||
|
||||
**Analog search scope:** `backend/services/`, `backend/deps/`, `backend/api/`, `backend/tests/`
|
||||
**Files scanned:** 5 (services/auth.py, deps/auth.py, api/auth.py, tests/test_auth_deps.py, tests/test_auth_es256.py, tests/conftest.py)
|
||||
**Pattern extraction date:** 2026-06-06
|
||||
Reference in New Issue
Block a user