diff --git a/backend/api/auth.py b/backend/api/auth.py index 31bde07..976d030 100644 --- a/backend/api/auth.py +++ b/backend/api/auth.py @@ -287,7 +287,12 @@ async def login( ) # 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) _set_refresh_cookie(response, raw_refresh, remember_me=body.remember_me) @@ -361,7 +366,12 @@ async def refresh_token( # Set new refresh cookie _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 { "access_token": access_token, "user": { diff --git a/backend/main.py b/backend/main.py index 8f7e7c6..5fdee79 100644 --- a/backend/main.py +++ b/backend/main.py @@ -244,7 +244,7 @@ async def lifespan(app: FastAPI): # ── 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) app.state.limiter = auth_limiter diff --git a/backend/tests/conftest.py b/backend/tests/conftest.py index 3ae5ecd..bc7cc82 100644 --- a/backend/tests/conftest.py +++ b/backend/tests/conftest.py @@ -30,6 +30,13 @@ from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_asyn 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 ────────────────────────────────────────────────────── def _port_open(host: str, port: int, timeout: float = 1.0) -> bool: @@ -146,13 +153,23 @@ async def db_session(): @pytest_asyncio.fixture 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 main import app 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 app.dependency_overrides.clear() @@ -273,7 +290,10 @@ async def auth_user(db_session: AsyncSession): db_session.add(quota) 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 { "user": user, "token": token, @@ -312,7 +332,8 @@ async def second_auth_user(db_session: AsyncSession): db_session.add(quota) 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 { "user": user, "token": token, @@ -349,7 +370,8 @@ async def admin_user(db_session: AsyncSession): db_session.add(quota) 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 { "user": user, "token": token, diff --git a/backend/tests/test_auth_deps.py b/backend/tests/test_auth_deps.py index a8fc9e2..7740dd2 100644 --- a/backend/tests/test_auth_deps.py +++ b/backend/tests/test_auth_deps.py @@ -22,6 +22,7 @@ from fastapi import FastAPI, Depends from sqlalchemy.ext.asyncio import AsyncSession from tests.test_auth_api import FakeRedis +from tests.conftest import _TEST_USER_AGENT # ── 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.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 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 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( "/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 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( "/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 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( "/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 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( "/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 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) 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 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) 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") 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 auth_client._transport.app.state.redis = _BrokenRedis() diff --git a/backend/tests/test_auth_fgp.py b/backend/tests/test_auth_fgp.py index 24818a1..7768c8b 100644 --- a/backend/tests/test_auth_fgp.py +++ b/backend/tests/test_auth_fgp.py @@ -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: - 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 UA/lang (defaults to ""), request sent with no User-Agent / Accept-Language 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 as _uuid +from datetime import datetime, timezone, timedelta +import jwt as _jwt import pytest import pytest_asyncio from httpx import ASGITransport, AsyncClient from fastapi import FastAPI, Depends from sqlalchemy.ext.asyncio import AsyncSession +from config import settings 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 -# ── FGP stubs (Wave 0) ──────────────────────────────────────────────────────── +# ── FGP tests (Wave 1 — promoted from xfail stubs) ─────────────────────────── -@pytest.mark.xfail(strict=False, reason="not implemented yet") @pytest.mark.asyncio async def test_fgp_match_returns_200(auth_client, db_session): """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 async def test_fgp_mismatch_returns_401(auth_client, db_session): """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 async def test_no_fgp_claim_allowed(auth_client, db_session): """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 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.""" - pytest.xfail("not implemented yet") + """FGP-04: token issued with empty-string binding, request with empty UA → 200. + + 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 diff --git a/backend/tests/test_cloud.py b/backend/tests/test_cloud.py index 55ce2af..bee5036 100644 --- a/backend/tests/test_cloud.py +++ b/backend/tests/test_cloud.py @@ -22,6 +22,8 @@ from sqlalchemy.ext.asyncio import AsyncSession pytestmark = pytest.mark.asyncio +from tests.conftest import _TEST_USER_AGENT + # ── 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}. 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 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) 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 { "user": user, "token": token, diff --git a/backend/tests/test_documents.py b/backend/tests/test_documents.py index 667a158..991cc5a 100644 --- a/backend/tests/test_documents.py +++ b/backend/tests/test_documents.py @@ -300,6 +300,7 @@ async def test_cross_user_access_404(async_client, auth_user, db_session): import uuid as _uuid from db.models import Document, User, Quota from services.auth import hash_password, create_access_token + from tests.conftest import _TEST_USER_AGENT # Create User A's document directly via ORM 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) 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}"} # 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 from db.models import User, Quota from services.auth import hash_password, create_access_token + from tests.conftest import _TEST_USER_AGENT recipient_id = _uuid2.uuid4() recipient = User( @@ -560,7 +563,8 @@ async def test_content_stream_share_recipient_200(async_client, auth_user, admin db_session.add(share) 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}"} 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 from db.models import Document, User, Quota from services.auth import hash_password, create_access_token + from tests.conftest import _TEST_USER_AGENT # Create a document owned by auth_user 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) 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}"} # User B attempts to reclassify User A's document — must get 404 diff --git a/backend/tests/test_security_headers.py b/backend/tests/test_security_headers.py index 001ec9f..ea969e5 100644 --- a/backend/tests/test_security_headers.py +++ b/backend/tests/test_security_headers.py @@ -18,6 +18,8 @@ import pytest_asyncio from httpx import ASGITransport, AsyncClient from sqlalchemy.ext.asyncio import AsyncSession +from tests.conftest import _TEST_USER_AGENT + # ── FakeRedis (needed by login endpoint) ───────────────────────────────────── @@ -77,7 +79,12 @@ async def headers_client(db_session: AsyncSession): except Exception: 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 app.dependency_overrides.clear() @@ -173,7 +180,8 @@ async def test_security_headers_on_authenticated_route(headers_client, db_sessio db_session.add(quota) 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( "/api/auth/me", headers={"Authorization": f"Bearer {token}"}, diff --git a/frontend/package.json b/frontend/package.json index b18eb21..b0793e6 100644 --- a/frontend/package.json +++ b/frontend/package.json @@ -1,6 +1,6 @@ { "name": "document-scanner-frontend", - "version": "0.1.2", + "version": "0.1.3", "type": "module", "scripts": { "dev": "vite",