""" 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 → request with same headers returns 200. - FGP-02 test_fgp_mismatch_returns_401: token issued for one UA, request arrives with different UA → 401 "Token fingerprint mismatch". - FGP-03 test_no_fgp_claim_allowed: token manually crafted without "fgp" key → 200 (graceful migration; tokens issued before Phase 7.4 keep working). - 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). """ 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 # ── Minimal test app with /test/me route ────────────────────────────────────── def make_test_app(): """Create a minimal FastAPI app that exercises the auth deps. Phase 7.4: app.state.redis is set to a FakeRedis() instance so the NBF check and fgp check in get_current_user can call request.app.state.redis without raising AttributeError. """ 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} # Mount FakeRedis so NBF / fgp checks in get_current_user work test_app.state.redis = FakeRedis() return test_app @pytest_asyncio.fixture async def auth_client(db_session: AsyncSession): """Async HTTP test client for the auth-dep test app.""" 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() async def _create_user(db_session, role: str = "user", is_active: bool = True): """Helper: insert a minimal User row into the test DB.""" 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 # ── FGP tests (Wave 1 — promoted from xfail stubs) ─────────────────────────── @pytest.mark.asyncio async def test_fgp_match_returns_200(auth_client, db_session): """FGP-01: token issued with matching UA/Accept-Language → 200.""" 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.asyncio async def test_fgp_mismatch_returns_401(auth_client, db_session): """FGP-02: token issued for one UA, request from different UA → 401.""" 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.asyncio async def test_no_fgp_claim_allowed(auth_client, db_session): """FGP-03: token without fgp claim → 200 (migration grace period).""" 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.asyncio async def test_fgp_no_concatenation_collision(auth_client, db_session): """FGP-05 regression: token bound to UA='foobar'+AL='' must not match UA='foo'+AL='bar'. Without a separator in the HMAC input, both produce the same bytes ('foobar'), allowing a different header split to pass fingerprint validation. """ from services.auth import create_access_token user = await _create_user(db_session, role="user") # Token issued for UA='foobar', no Accept-Language token = create_access_token(str(user.id), "user", user_agent="foobar", accept_lang="") # Request arrives from a split: UA='foo', AL='bar' — same concatenation without sep resp = await auth_client.get( "/test/me", headers={ "Authorization": f"Bearer {token}", "User-Agent": "foo", "Accept-Language": "bar", }, ) assert resp.status_code == 401 assert resp.json()["detail"] == "Token fingerprint mismatch" @pytest.mark.asyncio async def test_missing_headers_empty_string_binding(auth_client, db_session): """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