feat(07.4-02): update api/auth.py call sites, promote FGP tests, version bump 0.1.3

- api/auth.py login + refresh call sites pass User-Agent and Accept-Language
  headers to create_access_token (D-05 — fgp binding at token issuance)
- test_auth_fgp.py: promote all 4 xfail stubs to real assertions (FGP-01..04)
- conftest.py: add _TEST_USER_AGENT constant; configure async_client to send
  consistent User-Agent; update auth_user/second_auth_user/admin_user fixtures
  to bind fgp to _TEST_USER_AGENT so tokens validate correctly in tests
- test_auth_deps.py: import _TEST_USER_AGENT; update auth_client fixture and
  all create_access_token calls to use the constant
- test_cloud.py: update _create_user_and_token to bind fgp to _TEST_USER_AGENT
- test_documents.py: update 3 inline create_access_token calls to pass user_agent
- test_security_headers.py: import _TEST_USER_AGENT; update headers_client +
  token creation to use the constant
- Version bump: backend 0.1.2 → 0.1.3, frontend 0.1.2 → 0.1.3
- [Rule 1 - Bug] Fix httpx default User-Agent vs empty-string fgp mismatch in
  test infrastructure: 10 tests were failing due to fgp check rejecting tokens
  created with fgp="" when client sent "python-httpx/X.Y.Z"
This commit is contained in:
curo1305
2026-06-06 22:12:57 +02:00
parent 1420180be7
commit 61b1e045c4
9 changed files with 172 additions and 37 deletions
+12 -2
View File
@@ -287,7 +287,12 @@ async def login(
) )
# Issue tokens # Issue tokens
access_token = auth_service.create_access_token(str(user.id), user.role) 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", ""),
)
raw_refresh = await auth_service.create_refresh_token(session, user.id, remember_me=body.remember_me) raw_refresh = await auth_service.create_refresh_token(session, user.id, remember_me=body.remember_me)
_set_refresh_cookie(response, raw_refresh, remember_me=body.remember_me) _set_refresh_cookie(response, raw_refresh, remember_me=body.remember_me)
@@ -361,7 +366,12 @@ async def refresh_token(
# Set new refresh cookie # Set new refresh cookie
_set_refresh_cookie(response, new_raw) _set_refresh_cookie(response, new_raw)
access_token = auth_service.create_access_token(user_id_str, user.role) 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", ""),
)
return { return {
"access_token": access_token, "access_token": access_token,
"user": { "user": {
+1 -1
View File
@@ -244,7 +244,7 @@ async def lifespan(app: FastAPI):
# ── Application factory ─────────────────────────────────────────────────────── # ── Application factory ───────────────────────────────────────────────────────
app = FastAPI(title="Document Scanner API", version="0.1.2", lifespan=lifespan) app = FastAPI(title="Document Scanner API", version="0.1.3", lifespan=lifespan)
# Rate limiter state (slowapi) # Rate limiter state (slowapi)
app.state.limiter = auth_limiter app.state.limiter = auth_limiter
+27 -5
View File
@@ -30,6 +30,13 @@ from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_asyn
from sqlalchemy.pool import StaticPool from sqlalchemy.pool import StaticPool
# ── Phase 7.4: test-client user-agent constant ────────────────────────────────
# Tokens issued by test fixtures must embed a fgp claim matching the User-Agent
# the test client will send. httpx.AsyncClient defaults to "python-httpx/X.Y.Z"
# which varies by library version. We lock both token creation and the client
# to this constant so the fgp check in get_current_user always passes in tests.
_TEST_USER_AGENT = "docuvault-test/1.0"
# ── Service availability ────────────────────────────────────────────────────── # ── Service availability ──────────────────────────────────────────────────────
def _port_open(host: str, port: int, timeout: float = 1.0) -> bool: def _port_open(host: str, port: int, timeout: float = 1.0) -> bool:
@@ -146,13 +153,23 @@ async def db_session():
@pytest_asyncio.fixture @pytest_asyncio.fixture
async def async_client(db_session: AsyncSession): async def async_client(db_session: AsyncSession):
"""Async HTTP test client with the DB dependency overridden to use in-memory SQLite.""" """Async HTTP test client with the DB dependency overridden to use in-memory SQLite.
Phase 7.4: sets a fixed User-Agent (_TEST_USER_AGENT) so that the fgp claim
embedded in tokens from auth_user/admin_user fixtures matches what the client
sends. Without this, httpx's default 'python-httpx/X.Y.Z' UA would differ
from the empty-string UA used when creating tokens without explicit headers.
"""
from deps.db import get_db from deps.db import get_db
from main import app from main import app
app.dependency_overrides[get_db] = lambda: db_session app.dependency_overrides[get_db] = lambda: db_session
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as c: async with AsyncClient(
transport=ASGITransport(app=app),
base_url="http://test",
headers={"User-Agent": _TEST_USER_AGENT},
) as c:
yield c yield c
app.dependency_overrides.clear() app.dependency_overrides.clear()
@@ -273,7 +290,10 @@ async def auth_user(db_session: AsyncSession):
db_session.add(quota) db_session.add(quota)
await db_session.commit() await db_session.commit()
token = create_access_token(str(user_id), "user") # Phase 7.4: bind token fgp to _TEST_USER_AGENT so get_current_user validates
# successfully when the token is used via the async_client fixture (which sends
# the same User-Agent constant).
token = create_access_token(str(user_id), "user", user_agent=_TEST_USER_AGENT)
return { return {
"user": user, "user": user,
"token": token, "token": token,
@@ -312,7 +332,8 @@ async def second_auth_user(db_session: AsyncSession):
db_session.add(quota) db_session.add(quota)
await db_session.commit() await db_session.commit()
token = create_access_token(str(user_id), "user") # Phase 7.4: bind token fgp to _TEST_USER_AGENT (matches async_client default)
token = create_access_token(str(user_id), "user", user_agent=_TEST_USER_AGENT)
return { return {
"user": user, "user": user,
"token": token, "token": token,
@@ -349,7 +370,8 @@ async def admin_user(db_session: AsyncSession):
db_session.add(quota) db_session.add(quota)
await db_session.commit() await db_session.commit()
token = create_access_token(str(user_id), "admin") # Phase 7.4: bind token fgp to _TEST_USER_AGENT (matches async_client default)
token = create_access_token(str(user_id), "admin", user_agent=_TEST_USER_AGENT)
return { return {
"user": user, "user": user,
"token": token, "token": token,
+15 -8
View File
@@ -22,6 +22,7 @@ from fastapi import FastAPI, Depends
from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.ext.asyncio import AsyncSession
from tests.test_auth_api import FakeRedis from tests.test_auth_api import FakeRedis
from tests.conftest import _TEST_USER_AGENT
# ── Minimal test app with /test/me and /test/admin routes ───────────────────── # ── Minimal test app with /test/me and /test/admin routes ─────────────────────
@@ -65,7 +66,13 @@ async def auth_client(db_session: AsyncSession):
app = make_test_app() app = make_test_app()
app.dependency_overrides[get_db] = lambda: db_session app.dependency_overrides[get_db] = lambda: db_session
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as c: # Phase 7.4: send _TEST_USER_AGENT so fgp check in get_current_user succeeds
# for tokens created with the same user-agent below.
async with AsyncClient(
transport=ASGITransport(app=app),
base_url="http://test",
headers={"User-Agent": _TEST_USER_AGENT},
) as c:
yield c yield c
app.dependency_overrides.clear() app.dependency_overrides.clear()
@@ -97,7 +104,7 @@ async def test_get_current_user_returns_user(auth_client, db_session):
from services.auth import create_access_token from services.auth import create_access_token
user = await _create_user(db_session, role="user") user = await _create_user(db_session, role="user")
token = create_access_token(str(user.id), "user") token = create_access_token(str(user.id), "user", user_agent=_TEST_USER_AGENT)
resp = await auth_client.get( resp = await auth_client.get(
"/test/me", headers={"Authorization": f"Bearer {token}"} "/test/me", headers={"Authorization": f"Bearer {token}"}
@@ -124,7 +131,7 @@ async def test_get_current_user_rejects_inactive_user(auth_client, db_session):
from services.auth import create_access_token from services.auth import create_access_token
user = await _create_user(db_session, role="user", is_active=False) user = await _create_user(db_session, role="user", is_active=False)
token = create_access_token(str(user.id), "user") token = create_access_token(str(user.id), "user", user_agent=_TEST_USER_AGENT)
resp = await auth_client.get( resp = await auth_client.get(
"/test/me", headers={"Authorization": f"Bearer {token}"} "/test/me", headers={"Authorization": f"Bearer {token}"}
@@ -138,7 +145,7 @@ async def test_get_current_admin_rejects_non_admin(auth_client, db_session):
from services.auth import create_access_token from services.auth import create_access_token
user = await _create_user(db_session, role="user") user = await _create_user(db_session, role="user")
token = create_access_token(str(user.id), "user") token = create_access_token(str(user.id), "user", user_agent=_TEST_USER_AGENT)
resp = await auth_client.get( resp = await auth_client.get(
"/test/admin", headers={"Authorization": f"Bearer {token}"} "/test/admin", headers={"Authorization": f"Bearer {token}"}
@@ -153,7 +160,7 @@ async def test_get_current_admin_allows_admin(auth_client, db_session):
from services.auth import create_access_token from services.auth import create_access_token
admin_user = await _create_user(db_session, role="admin") admin_user = await _create_user(db_session, role="admin")
token = create_access_token(str(admin_user.id), "admin") token = create_access_token(str(admin_user.id), "admin", user_agent=_TEST_USER_AGENT)
resp = await auth_client.get( resp = await auth_client.get(
"/test/admin", headers={"Authorization": f"Bearer {token}"} "/test/admin", headers={"Authorization": f"Bearer {token}"}
@@ -193,7 +200,7 @@ async def test_get_current_user_rejects_token_when_iat_before_user_nbf(auth_clie
import time import time
user = await _create_user(db_session, role="user") user = await _create_user(db_session, role="user")
token = create_access_token(str(user.id), "user") token = create_access_token(str(user.id), "user", user_agent=_TEST_USER_AGENT)
# Pre-populate Redis with a future nbf (token's iat will be < this value) # Pre-populate Redis with a future nbf (token's iat will be < this value)
future_ts = int(time.time()) + 3600 future_ts = int(time.time()) + 3600
@@ -215,7 +222,7 @@ async def test_get_current_user_allows_token_when_iat_after_user_nbf(auth_client
import time import time
user = await _create_user(db_session, role="user") user = await _create_user(db_session, role="user")
token = create_access_token(str(user.id), "user") token = create_access_token(str(user.id), "user", user_agent=_TEST_USER_AGENT)
# Pre-populate Redis with a past nbf (token's iat will be > this value) # Pre-populate Redis with a past nbf (token's iat will be > this value)
past_ts = int(time.time()) - 3600 past_ts = int(time.time()) - 3600
@@ -238,7 +245,7 @@ async def test_get_current_user_failopen_on_redis_error(auth_client, db_session)
raise RuntimeError("simulated redis down") raise RuntimeError("simulated redis down")
user = await _create_user(db_session, role="user") user = await _create_user(db_session, role="user")
token = create_access_token(str(user.id), "user") token = create_access_token(str(user.id), "user", user_agent=_TEST_USER_AGENT)
# Override app.state.redis with a broken redis for this test # Override app.state.redis with a broken redis for this test
auth_client._transport.app.state.redis = _BrokenRedis() auth_client._transport.app.state.redis = _BrokenRedis()
+91 -14
View File
@@ -1,5 +1,5 @@
""" """
TDD scaffold for Phase 7.4: Token fingerprinting / token binding. Integration tests for Phase 7.4: Token fingerprinting / token binding.
Covers FGP-01..FGP-04: Covers FGP-01..FGP-04:
- FGP-01 test_fgp_match_returns_200: token issued with matching UA/lang → - FGP-01 test_fgp_match_returns_200: token issued with matching UA/lang →
@@ -12,18 +12,20 @@ Covers FGP-01..FGP-04:
- FGP-04 test_missing_headers_empty_string_binding: token issued with no - FGP-04 test_missing_headers_empty_string_binding: token issued with no
UA/lang (defaults to ""), request sent with no User-Agent / Accept-Language UA/lang (defaults to ""), request sent with no User-Agent / Accept-Language
headers → 200 (empty-string fingerprint matches). headers → 200 (empty-string fingerprint matches).
All four tests are xfail(strict=False) until Wave 1 (Plan 07.4-02) implements
the fgp claim in services/auth.py and deps/auth.py.
""" """
import base64
import uuid import uuid
import uuid as _uuid
from datetime import datetime, timezone, timedelta
import jwt as _jwt
import pytest import pytest
import pytest_asyncio import pytest_asyncio
from httpx import ASGITransport, AsyncClient from httpx import ASGITransport, AsyncClient
from fastapi import FastAPI, Depends from fastapi import FastAPI, Depends
from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.ext.asyncio import AsyncSession
from config import settings
from tests.test_auth_api import FakeRedis from tests.test_auth_api import FakeRedis
@@ -84,32 +86,107 @@ async def _create_user(db_session, role: str = "user", is_active: bool = True):
return user return user
# ── FGP stubs (Wave 0) ──────────────────────────────────────────────────────── # ── FGP tests (Wave 1 — promoted from xfail stubs) ───────────────────────────
@pytest.mark.xfail(strict=False, reason="not implemented yet")
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_fgp_match_returns_200(auth_client, db_session): async def test_fgp_match_returns_200(auth_client, db_session):
"""FGP-01: token issued with matching UA/Accept-Language → 200.""" """FGP-01: token issued with matching UA/Accept-Language → 200."""
pytest.xfail("not implemented yet") from services.auth import create_access_token
user = await _create_user(db_session, role="user")
token = create_access_token(
str(user.id),
"user",
user_agent="Mozilla/5.0",
accept_lang="en",
)
resp = await auth_client.get(
"/test/me",
headers={
"Authorization": f"Bearer {token}",
"User-Agent": "Mozilla/5.0",
"Accept-Language": "en",
},
)
assert resp.status_code == 200
@pytest.mark.xfail(strict=False, reason="not implemented yet")
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_fgp_mismatch_returns_401(auth_client, db_session): async def test_fgp_mismatch_returns_401(auth_client, db_session):
"""FGP-02: token issued for one UA, request from different UA → 401.""" """FGP-02: token issued for one UA, request from different UA → 401."""
pytest.xfail("not implemented yet") from services.auth import create_access_token
user = await _create_user(db_session, role="user")
token = create_access_token(
str(user.id),
"user",
user_agent="Mozilla/5.0",
accept_lang="en",
)
resp = await auth_client.get(
"/test/me",
headers={
"Authorization": f"Bearer {token}",
"User-Agent": "different-agent",
"Accept-Language": "en",
},
)
assert resp.status_code == 401
assert resp.json()["detail"] == "Token fingerprint mismatch"
@pytest.mark.xfail(strict=False, reason="not implemented yet")
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_no_fgp_claim_allowed(auth_client, db_session): async def test_no_fgp_claim_allowed(auth_client, db_session):
"""FGP-03: token without fgp claim → 200 (migration grace period).""" """FGP-03: token without fgp claim → 200 (migration grace period)."""
pytest.xfail("not implemented yet") user = await _create_user(db_session, role="user")
# Manually craft a JWT without the "fgp" key (simulating a pre-7.4 token)
now = datetime.now(timezone.utc)
payload_no_fgp = {
"sub": str(user.id),
"role": "user",
"typ": "access",
"iat": now,
"exp": now + timedelta(minutes=15),
"jti": str(_uuid.uuid4()),
# "fgp" intentionally absent
}
private_pem = base64.b64decode(settings.jwt_private_key).decode()
token = _jwt.encode(payload_no_fgp, private_pem, algorithm="ES256")
resp = await auth_client.get(
"/test/me",
headers={"Authorization": f"Bearer {token}"},
)
assert resp.status_code == 200
@pytest.mark.xfail(strict=False, reason="not implemented yet")
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_missing_headers_empty_string_binding(auth_client, db_session): async def test_missing_headers_empty_string_binding(auth_client, db_session):
"""FGP-04: token issued with no headers (empty-string binding), request with no headers → 200.""" """FGP-04: token issued with empty-string binding, request with empty UA → 200.
pytest.xfail("not implemented yet")
When no User-Agent or Accept-Language are provided, both the token issuance
and the validation path use empty strings. httpx sends a default User-Agent
header automatically, so we must explicitly set it to "" to reproduce the
absent-header scenario in the test client.
"""
from services.auth import create_access_token
user = await _create_user(db_session, role="user")
# Issue token with empty-string defaults (no UA, no Accept-Language)
token = create_access_token(str(user.id), "user")
# Send request with empty User-Agent and no Accept-Language;
# FastAPI will see request.headers.get("User-Agent", "") == ""
# matching the empty-string fingerprint embedded in the token.
resp = await auth_client.get(
"/test/me",
headers={
"Authorization": f"Bearer {token}",
"User-Agent": "",
},
)
assert resp.status_code == 200
+6 -1
View File
@@ -22,6 +22,8 @@ from sqlalchemy.ext.asyncio import AsyncSession
pytestmark = pytest.mark.asyncio pytestmark = pytest.mark.asyncio
from tests.conftest import _TEST_USER_AGENT
# ── Shared auth helper ──────────────────────────────────────────────────────── # ── Shared auth helper ────────────────────────────────────────────────────────
@@ -29,6 +31,8 @@ async def _create_user_and_token(session, role: str = "user"):
"""Create a User + Quota row, return {user, token, headers}. """Create a User + Quota row, return {user, token, headers}.
Mirrors the auth_user fixture pattern from conftest.py. Mirrors the auth_user fixture pattern from conftest.py.
Phase 7.4: tokens are bound to _TEST_USER_AGENT fgp so async_client
(which sends the same User-Agent) passes fgp validation in get_current_user.
""" """
from db.models import User, Quota from db.models import User, Quota
from services.auth import hash_password, create_access_token from services.auth import hash_password, create_access_token
@@ -52,7 +56,8 @@ async def _create_user_and_token(session, role: str = "user"):
session.add(quota) session.add(quota)
await session.commit() await session.commit()
token = create_access_token(str(user_id), role) # Phase 7.4: bind fgp to _TEST_USER_AGENT (matches async_client default)
token = create_access_token(str(user_id), role, user_agent=_TEST_USER_AGENT)
return { return {
"user": user, "user": user,
"token": token, "token": token,
+9 -3
View File
@@ -300,6 +300,7 @@ async def test_cross_user_access_404(async_client, auth_user, db_session):
import uuid as _uuid import uuid as _uuid
from db.models import Document, User, Quota from db.models import Document, User, Quota
from services.auth import hash_password, create_access_token from services.auth import hash_password, create_access_token
from tests.conftest import _TEST_USER_AGENT
# Create User A's document directly via ORM # Create User A's document directly via ORM
doc_id = _uuid.uuid4() doc_id = _uuid.uuid4()
@@ -331,7 +332,8 @@ async def test_cross_user_access_404(async_client, auth_user, db_session):
db_session.add(quota_b) db_session.add(quota_b)
await db_session.commit() await db_session.commit()
token_b = create_access_token(str(user_b_id), "user") # Phase 7.4: bind fgp to _TEST_USER_AGENT (matches async_client default)
token_b = create_access_token(str(user_b_id), "user", user_agent=_TEST_USER_AGENT)
headers_b = {"Authorization": f"Bearer {token_b}"} headers_b = {"Authorization": f"Bearer {token_b}"}
# User B attempts to access User A's document — must get 404 (not 403) # User B attempts to access User A's document — must get 404 (not 403)
@@ -521,6 +523,7 @@ async def test_content_stream_share_recipient_200(async_client, auth_user, admin
import uuid as _uuid2 import uuid as _uuid2
from db.models import User, Quota from db.models import User, Quota
from services.auth import hash_password, create_access_token from services.auth import hash_password, create_access_token
from tests.conftest import _TEST_USER_AGENT
recipient_id = _uuid2.uuid4() recipient_id = _uuid2.uuid4()
recipient = User( recipient = User(
@@ -560,7 +563,8 @@ async def test_content_stream_share_recipient_200(async_client, auth_user, admin
db_session.add(share) db_session.add(share)
await db_session.commit() await db_session.commit()
recipient_token = create_access_token(str(recipient_id), "user") # Phase 7.4: bind fgp to _TEST_USER_AGENT (matches async_client default)
recipient_token = create_access_token(str(recipient_id), "user", user_agent=_TEST_USER_AGENT)
recipient_headers = {"Authorization": f"Bearer {recipient_token}"} recipient_headers = {"Authorization": f"Bearer {recipient_token}"}
resp = await async_client.get( resp = await async_client.get(
@@ -1015,6 +1019,7 @@ async def test_reclassify_cross_user_returns_404(async_client, auth_user, db_ses
import uuid as _uuid import uuid as _uuid
from db.models import Document, User, Quota from db.models import Document, User, Quota
from services.auth import hash_password, create_access_token from services.auth import hash_password, create_access_token
from tests.conftest import _TEST_USER_AGENT
# Create a document owned by auth_user # Create a document owned by auth_user
doc_id = _uuid.uuid4() doc_id = _uuid.uuid4()
@@ -1046,7 +1051,8 @@ async def test_reclassify_cross_user_returns_404(async_client, auth_user, db_ses
db_session.add(quota_b) db_session.add(quota_b)
await db_session.commit() await db_session.commit()
token_b = create_access_token(str(user_b_id), "user") # Phase 7.4: bind fgp to _TEST_USER_AGENT (matches async_client default)
token_b = create_access_token(str(user_b_id), "user", user_agent=_TEST_USER_AGENT)
headers_b = {"Authorization": f"Bearer {token_b}"} headers_b = {"Authorization": f"Bearer {token_b}"}
# User B attempts to reclassify User A's document — must get 404 # User B attempts to reclassify User A's document — must get 404
+10 -2
View File
@@ -18,6 +18,8 @@ import pytest_asyncio
from httpx import ASGITransport, AsyncClient from httpx import ASGITransport, AsyncClient
from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.ext.asyncio import AsyncSession
from tests.conftest import _TEST_USER_AGENT
# ── FakeRedis (needed by login endpoint) ───────────────────────────────────── # ── FakeRedis (needed by login endpoint) ─────────────────────────────────────
@@ -77,7 +79,12 @@ async def headers_client(db_session: AsyncSession):
except Exception: except Exception:
pass pass
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as c: # Phase 7.4: send _TEST_USER_AGENT so fgp check matches tokens from this file
async with AsyncClient(
transport=ASGITransport(app=app),
base_url="http://test",
headers={"User-Agent": _TEST_USER_AGENT},
) as c:
yield c yield c
app.dependency_overrides.clear() app.dependency_overrides.clear()
@@ -173,7 +180,8 @@ async def test_security_headers_on_authenticated_route(headers_client, db_sessio
db_session.add(quota) db_session.add(quota)
await db_session.commit() await db_session.commit()
token = create_access_token(str(user_id), "user") # Phase 7.4: bind fgp to _TEST_USER_AGENT (matches headers_client default)
token = create_access_token(str(user_id), "user", user_agent=_TEST_USER_AGENT)
resp = await headers_client.get( resp = await headers_client.get(
"/api/auth/me", "/api/auth/me",
headers={"Authorization": f"Bearer {token}"}, headers={"Authorization": f"Bearer {token}"},
+1 -1
View File
@@ -1,6 +1,6 @@
{ {
"name": "document-scanner-frontend", "name": "document-scanner-frontend",
"version": "0.1.2", "version": "0.1.3",
"type": "module", "type": "module",
"scripts": { "scripts": {
"dev": "vite", "dev": "vite",