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:
+12
-2
@@ -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
@@ -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
|
||||||
|
|||||||
@@ -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,
|
||||||
|
|||||||
@@ -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()
|
||||||
|
|||||||
@@ -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
|
||||||
|
|||||||
@@ -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,
|
||||||
|
|||||||
@@ -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
|
||||||
|
|||||||
@@ -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,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",
|
||||||
|
|||||||
Reference in New Issue
Block a user