kite/backend/tests/test_auth_fgp.py

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