4ecc3d689c
All checks were successful
CI / changes (pull_request) Successful in 10s
CI Trade-In / changes (pull_request) Successful in 10s
CI Trade-In / frontend-checks (pull_request) Has been skipped
CI / backend-tests (pull_request) Has been skipped
CI / frontend-tests (pull_request) Has been skipped
CI / openapi-codegen-check (pull_request) Has been skipped
CI Trade-In / backend-tests (pull_request) Successful in 1m25s
CRITICAL: _propagate_authenticated_user делала skip-if-present вместо
перезаписи — клиент-контролируемый X-Authenticated-User (Caddy шлёт его
на КАЖДЫЙ прод-запрос) выигрывал у резолвленной сессии для всего
downstream-трафика, читающего заголовок напрямую (_assert_estimate_access*,
account_quota, /trade-in/history, support.py) — в обоих auth_mode
(dual и db_only). Теперь заголовок безусловно перезаписывается сессионным
username (ASGI header-имена всегда lowercase bytes).
Medium: .encode("latin-1") без errors="replace" крашил бы 500-кой каждый
запрос кириллического username. Login timing-oracle — verify_password
короткозамыкалась на unknown-username/NULL-hash (~1мс vs ~100-300мс bcrypt)
→ теперь всегда сверяется против dummy-хеша при отсутствующем юзере/хеше.
Login rate-limit key length-prefixed — username с ':' (или IPv6 IP) больше
не может схлопнуть чужой бюджет.
Новые тесты подтверждают регрессию: прогнаны на старом коде (до фикса)
через временный откат rbac.py — все три (spoof dual-mode, spoof db_only,
кириллица) падали с 'victim' == 'alice' / UnicodeEncodeError; после
фикса — зелёные. test_rbac.py/test_internal_auth_secret.py без изменений.
536 lines
21 KiB
Python
536 lines
21 KiB
Python
"""Integration tests for #2552 auth-core: POST /login, /logout, dual-mode GET /me,
|
||
and rbac_guard session-cookie resolution.
|
||
|
||
Uses the REAL `rbac_guard` (app.core.rbac) + REAL `auth.router` / `me.router` wired
|
||
into an isolated FastAPI test app (same pattern as tests/test_rbac.py), with an
|
||
in-memory fake DB standing in for `tradein_users`/`tradein_sessions`:
|
||
- `app.core.rbac.SessionLocal` is monkeypatched (rbac_guard opens its own session,
|
||
it's middleware — no FastAPI DI available there).
|
||
- `app.core.db.get_db` is overridden via `app.dependency_overrides` (auth.py /
|
||
me.py use `Depends(get_db)`, the idiomatic FastAPI-testable path).
|
||
|
||
Both point at the SAME `_Store` instance per test, so a session created by POST
|
||
/login is immediately visible to rbac_guard's own DB round trip on the next request.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import os
|
||
from datetime import UTC, datetime, timedelta
|
||
from types import SimpleNamespace
|
||
from typing import Annotated, Any
|
||
|
||
os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
|
||
|
||
import pytest
|
||
from fastapi import FastAPI, Header
|
||
from fastapi.testclient import TestClient
|
||
|
||
from app.api.v1 import auth as auth_router
|
||
from app.api.v1 import me as me_router
|
||
from app.core import auth as auth_mod
|
||
from app.core import config
|
||
from app.core.db import get_db
|
||
from app.core.password import hash_password
|
||
from app.core.rbac import rbac_guard
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Fake DB backing tradein_users / tradein_sessions
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class _Store:
|
||
def __init__(self) -> None:
|
||
self.users: dict[str, dict[str, Any]] = {}
|
||
self.sessions: dict[str, dict[str, Any]] = {}
|
||
self._next_id = 1
|
||
|
||
def add_user(
|
||
self,
|
||
username: str,
|
||
password_hash: str | None,
|
||
*,
|
||
role: str = "employee",
|
||
is_active: bool = True,
|
||
display_name: str | None = "Alice A.",
|
||
org_name: str | None = "Org LLC",
|
||
email: str | None = "alice@example.com",
|
||
) -> int:
|
||
uid = self._next_id
|
||
self._next_id += 1
|
||
self.users[username] = {
|
||
"id": uid,
|
||
"username": username,
|
||
"password_hash": password_hash,
|
||
"role": role,
|
||
"is_active": is_active,
|
||
"display_name": display_name,
|
||
"org_name": org_name,
|
||
"email": email,
|
||
}
|
||
return uid
|
||
|
||
def user_by_id(self, uid: int) -> dict[str, Any] | None:
|
||
for u in self.users.values():
|
||
if u["id"] == uid:
|
||
return u
|
||
return None
|
||
|
||
def add_expired_session(self, token: str, user_id: int) -> None:
|
||
now = datetime.now(UTC)
|
||
self.sessions[token] = {
|
||
"user_id": user_id,
|
||
"expires_at": now - timedelta(minutes=1),
|
||
"last_seen_at": now - timedelta(minutes=1),
|
||
}
|
||
|
||
|
||
class _FakeDB:
|
||
"""Minimal Session stand-in dispatching on SQL text — see module docstring."""
|
||
|
||
def __init__(self, store: _Store) -> None:
|
||
self.store = store
|
||
|
||
def __enter__(self) -> _FakeDB:
|
||
return self
|
||
|
||
def __exit__(self, *exc: object) -> bool:
|
||
return False
|
||
|
||
def close(self) -> None:
|
||
pass
|
||
|
||
def commit(self) -> None:
|
||
pass
|
||
|
||
def rollback(self) -> None:
|
||
pass
|
||
|
||
def execute(self, stmt: object, params: dict[str, Any] | None = None) -> SimpleNamespace:
|
||
sql = str(stmt)
|
||
p = params or {}
|
||
|
||
if "INSERT INTO tradein_sessions" in sql:
|
||
now = datetime.now(UTC)
|
||
self.store.sessions[p["token"]] = {
|
||
"user_id": p["user_id"],
|
||
"expires_at": now + timedelta(hours=p["ttl_hours"]),
|
||
"last_seen_at": now,
|
||
}
|
||
return SimpleNamespace(fetchone=lambda: None)
|
||
|
||
if "UPDATE tradein_sessions" in sql and "SET last_seen_at" in sql:
|
||
sess = self.store.sessions.get(p["token"])
|
||
if sess is not None:
|
||
now = datetime.now(UTC)
|
||
sess["last_seen_at"] = now
|
||
sess["expires_at"] = now + timedelta(hours=p["ttl_hours"])
|
||
return SimpleNamespace(fetchone=lambda: None)
|
||
|
||
if "DELETE FROM tradein_sessions WHERE token" in sql:
|
||
self.store.sessions.pop(p["token"], None)
|
||
return SimpleNamespace(fetchone=lambda: None)
|
||
|
||
if "DELETE FROM tradein_sessions WHERE user_id" in sql:
|
||
uid = p["user_id"]
|
||
for tok in [t for t, s in self.store.sessions.items() if s["user_id"] == uid]:
|
||
del self.store.sessions[tok]
|
||
return SimpleNamespace(fetchone=lambda: None)
|
||
|
||
if "FROM tradein_sessions s" in sql and "JOIN tradein_users u" in sql:
|
||
sess = self.store.sessions.get(p["token"])
|
||
if sess is None:
|
||
return SimpleNamespace(fetchone=lambda: None)
|
||
user = self.store.user_by_id(sess["user_id"])
|
||
if user is None:
|
||
return SimpleNamespace(fetchone=lambda: None)
|
||
row = SimpleNamespace(
|
||
user_id=sess["user_id"],
|
||
expires_at=sess["expires_at"],
|
||
last_seen_at=sess["last_seen_at"],
|
||
username=user["username"],
|
||
role=user["role"],
|
||
display_name=user["display_name"],
|
||
org_name=user["org_name"],
|
||
email=user["email"],
|
||
is_active=user["is_active"],
|
||
)
|
||
return SimpleNamespace(fetchone=lambda: row)
|
||
|
||
if "FROM tradein_users" in sql:
|
||
user = self.store.users.get(p["username"])
|
||
if user is None:
|
||
return SimpleNamespace(fetchone=lambda: None)
|
||
row = SimpleNamespace(**user)
|
||
return SimpleNamespace(fetchone=lambda: row)
|
||
|
||
raise AssertionError(f"unhandled fake SQL in test_auth_api: {sql!r}")
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Test app
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def _build_test_app(store: _Store) -> FastAPI:
|
||
app = FastAPI()
|
||
app.middleware("http")(rbac_guard)
|
||
app.include_router(auth_router.router, prefix="/api/v1/auth", tags=["auth"])
|
||
app.include_router(me_router.router, prefix="/api/v1", tags=["me"])
|
||
|
||
@app.get("/api/v1/trade-in/dummy")
|
||
async def tradein_dummy() -> dict:
|
||
return {"ok": True}
|
||
|
||
@app.get("/api/v1/trade-in/whoami")
|
||
async def tradein_whoami(
|
||
x_authenticated_user: Annotated[str | None, Header(alias="X-Authenticated-User")] = None,
|
||
) -> dict:
|
||
"""Echoes the X-Authenticated-User header exactly as a downstream handler
|
||
(`_assert_estimate_access*`, `account_quota`, etc.) would see it — used to
|
||
assert session-identity wins over a client-forged header (#2552 spoof fix)."""
|
||
return {"user": x_authenticated_user}
|
||
|
||
def _override_get_db(): # generator dependency — matches app.core.db.get_db shape
|
||
yield _FakeDB(store)
|
||
|
||
app.dependency_overrides[get_db] = _override_get_db
|
||
return app
|
||
|
||
|
||
@pytest.fixture(autouse=True)
|
||
def _reset_state(monkeypatch: pytest.MonkeyPatch) -> None:
|
||
auth_mod.reset_cache_for_tests()
|
||
auth_router._LOGIN_LIMITER._hits.clear()
|
||
monkeypatch.setattr(config.settings, "auth_mode", "dual")
|
||
|
||
|
||
@pytest.fixture
|
||
def store() -> _Store:
|
||
return _Store()
|
||
|
||
|
||
@pytest.fixture
|
||
def client(store: _Store, monkeypatch: pytest.MonkeyPatch) -> TestClient:
|
||
monkeypatch.setattr("app.core.rbac.SessionLocal", lambda: _FakeDB(store))
|
||
# base_url=https:// — login sets the session cookie with Secure=True (real prod
|
||
# behaviour, not weakened for tests); httpx's cookie jar silently drops Secure
|
||
# cookies on a plain-http connection, so a plain http://testserver client would
|
||
# never resend the cookie on subsequent requests within the same test.
|
||
return TestClient(_build_test_app(store), base_url="https://testserver")
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# POST /login
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def test_login_happy_path_sets_cookie(client: TestClient, store: _Store) -> None:
|
||
store.add_user("alice", hash_password("Secret123!"), role="employee")
|
||
|
||
resp = client.post("/api/v1/auth/login", json={"username": "alice", "password": "Secret123!"})
|
||
|
||
assert resp.status_code == 200, resp.text
|
||
assert resp.json() == {"ok": True}
|
||
cookie_name = config.settings.session_cookie_name
|
||
assert cookie_name in resp.cookies
|
||
assert resp.cookies[cookie_name]
|
||
# Сессия реально создана в сторе под этим токеном.
|
||
assert resp.cookies[cookie_name] in store.sessions
|
||
|
||
|
||
def test_login_wrong_password_401_and_logs_failed_event(
|
||
client: TestClient, store: _Store, monkeypatch: pytest.MonkeyPatch
|
||
) -> None:
|
||
store.add_user("alice", hash_password("Secret123!"), role="employee")
|
||
events: list[dict[str, Any]] = []
|
||
monkeypatch.setattr(auth_router, "schedule_event", lambda **kw: events.append(kw))
|
||
|
||
resp = client.post("/api/v1/auth/login", json={"username": "alice", "password": "wrong"})
|
||
|
||
assert resp.status_code == 401
|
||
assert "detail" in resp.json()
|
||
assert config.settings.session_cookie_name not in resp.cookies
|
||
assert any(e["event_type"] == "login_failed" for e in events)
|
||
failed = next(e for e in events if e["event_type"] == "login_failed")
|
||
assert failed["username"] == "alice"
|
||
# Raw-пароль никогда не попадает в событие.
|
||
assert "wrong" not in str(failed)
|
||
|
||
|
||
def test_login_unknown_username_401_generic_message(client: TestClient) -> None:
|
||
resp = client.post("/api/v1/auth/login", json={"username": "ghost", "password": "whatever"})
|
||
assert resp.status_code == 401
|
||
# НЕ раскрываем, что юзера не существует — то же сообщение, что и wrong-password.
|
||
body_ghost = resp.json()["detail"]
|
||
|
||
resp2 = client.post("/api/v1/auth/login", json={"username": "ghost2", "password": "x"})
|
||
assert resp2.json()["detail"] == body_ghost
|
||
|
||
|
||
def test_login_inactive_user_401(client: TestClient, store: _Store) -> None:
|
||
store.add_user("bob", hash_password("Secret123!"), role="employee", is_active=False)
|
||
resp = client.post("/api/v1/auth/login", json={"username": "bob", "password": "Secret123!"})
|
||
assert resp.status_code == 401
|
||
|
||
|
||
def test_login_null_password_hash_401(client: TestClient, store: _Store) -> None:
|
||
store.add_user("carol", None, role="employee")
|
||
resp = client.post("/api/v1/auth/login", json={"username": "carol", "password": "anything"})
|
||
assert resp.status_code == 401
|
||
|
||
|
||
def test_login_always_calls_verify_password_timing_oracle_guard(
|
||
client: TestClient, store: _Store, monkeypatch: pytest.MonkeyPatch
|
||
) -> None:
|
||
"""#2552 post-review Medium 2: `verify_password` должен выполняться ровно
|
||
один раз на КАЖДУЮ попытку логина — включая неизвестный username и NULL
|
||
password_hash — иначе короткое замыкание даёт наблюдаемый timing-oracle
|
||
для user-enumeration. Тест не измеряет тайминг (флейки в CI), а проверяет
|
||
сам факт + аргумент вызова через monkeypatch-счётчик."""
|
||
store.add_user("alice", hash_password("Secret123!"), role="employee")
|
||
store.add_user("nullhash", None, role="employee")
|
||
|
||
calls: list[str] = []
|
||
real_verify = auth_router.verify_password
|
||
|
||
def _counting_verify(plain: str, hashed: str) -> bool:
|
||
calls.append(hashed)
|
||
return real_verify(plain, hashed)
|
||
|
||
monkeypatch.setattr(auth_router, "verify_password", _counting_verify)
|
||
|
||
resp_unknown = client.post("/api/v1/auth/login", json={"username": "ghost", "password": "x"})
|
||
assert resp_unknown.status_code == 401
|
||
|
||
resp_null_hash = client.post(
|
||
"/api/v1/auth/login", json={"username": "nullhash", "password": "x"}
|
||
)
|
||
assert resp_null_hash.status_code == 401
|
||
|
||
resp_wrong_pw = client.post(
|
||
"/api/v1/auth/login", json={"username": "alice", "password": "wrong"}
|
||
)
|
||
assert resp_wrong_pw.status_code == 401
|
||
|
||
assert len(calls) == 3
|
||
# Unknown user / NULL hash — сверяется против dummy-хеша, не против NULL.
|
||
assert calls[0] == auth_router._DUMMY_PASSWORD_HASH
|
||
assert calls[1] == auth_router._DUMMY_PASSWORD_HASH
|
||
# Реальный юзер с реальным hash — НЕ dummy.
|
||
assert calls[2] != auth_router._DUMMY_PASSWORD_HASH
|
||
|
||
|
||
def test_login_rate_limit_429(client: TestClient, store: _Store) -> None:
|
||
store.add_user("dave", hash_password("Secret123!"), role="employee")
|
||
limit = config.settings.login_rate_limit
|
||
|
||
for _ in range(limit):
|
||
resp = client.post("/api/v1/auth/login", json={"username": "dave", "password": "wrong"})
|
||
assert resp.status_code == 401
|
||
|
||
resp = client.post("/api/v1/auth/login", json={"username": "dave", "password": "wrong"})
|
||
assert resp.status_code == 429
|
||
assert "Retry-After" in resp.headers
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# POST /logout
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def test_logout_revokes_session_and_clears_cookie(client: TestClient, store: _Store) -> None:
|
||
store.add_user("alice", hash_password("Secret123!"), role="employee")
|
||
login_resp = client.post(
|
||
"/api/v1/auth/login", json={"username": "alice", "password": "Secret123!"}
|
||
)
|
||
token = login_resp.cookies[config.settings.session_cookie_name]
|
||
assert token in store.sessions
|
||
|
||
logout_resp = client.post("/api/v1/auth/logout")
|
||
assert logout_resp.status_code == 200
|
||
assert logout_resp.json() == {"ok": True}
|
||
assert token not in store.sessions
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# GET /me — session-first + dual-mode legacy fallback
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def test_me_with_session_cookie_returns_db_role(client: TestClient, store: _Store) -> None:
|
||
store.add_user(
|
||
"alice",
|
||
hash_password("Secret123!"),
|
||
role="employee",
|
||
display_name="Алиса",
|
||
org_name="ООО Ромашка",
|
||
email="alice@romashka.ru",
|
||
)
|
||
client.post("/api/v1/auth/login", json={"username": "alice", "password": "Secret123!"})
|
||
|
||
resp = client.get("/api/v1/me")
|
||
assert resp.status_code == 200, resp.text
|
||
body = resp.json()
|
||
assert body["username"] == "alice"
|
||
assert body["role"] == "employee"
|
||
assert "/trade-in/**" in body["allowed_paths"]
|
||
assert "/admin/**" in body["deny_paths"]
|
||
assert body["display_name"] == "Алиса"
|
||
assert body["org"] == "ООО Ромашка"
|
||
assert body["email"] == "alice@romashka.ru"
|
||
|
||
|
||
def test_me_manager_role_gets_team_path(client: TestClient, store: _Store) -> None:
|
||
store.add_user("mgr", hash_password("Secret123!"), role="manager")
|
||
client.post("/api/v1/auth/login", json={"username": "mgr", "password": "Secret123!"})
|
||
|
||
resp = client.get("/api/v1/me")
|
||
assert resp.status_code == 200
|
||
assert "/api/v1/team/**" in resp.json()["allowed_paths"]
|
||
|
||
|
||
def test_me_without_cookie_dual_mode_legacy_still_works(client: TestClient) -> None:
|
||
"""Regression guard: без сессии, auth_mode=dual — legacy X-Authenticated-User
|
||
путь через roles.yaml работает БЕЗ ИЗМЕНЕНИЙ."""
|
||
resp = client.get("/api/v1/me", headers={"X-Authenticated-User": "admin"})
|
||
assert resp.status_code == 200, resp.text
|
||
body = resp.json()
|
||
assert body["username"] == "admin"
|
||
assert body["role"] == "admin"
|
||
assert body["allowed_paths"] == ["/**"]
|
||
|
||
|
||
def test_me_no_cookie_no_header_401(client: TestClient) -> None:
|
||
resp = client.get("/api/v1/me")
|
||
assert resp.status_code == 401
|
||
|
||
|
||
def test_expired_session_falls_back_to_no_auth(client: TestClient, store: _Store) -> None:
|
||
"""Истёкшая сессия трактуется как отсутствие cookie: без legacy-заголовка — 401."""
|
||
uid = store.add_user("alice", hash_password("Secret123!"), role="employee")
|
||
store.add_expired_session("expired-tok", uid)
|
||
|
||
client.cookies.set(config.settings.session_cookie_name, "expired-tok")
|
||
resp = client.get("/api/v1/trade-in/dummy")
|
||
assert resp.status_code == 401
|
||
|
||
|
||
def test_expired_session_falls_back_to_legacy_header_in_dual_mode(
|
||
client: TestClient, store: _Store
|
||
) -> None:
|
||
"""Истёкшая сессия + валидный legacy header в dual-mode — header отрабатывает."""
|
||
uid = store.add_user("alice", hash_password("Secret123!"), role="employee")
|
||
store.add_expired_session("expired-tok", uid)
|
||
|
||
client.cookies.set(config.settings.session_cookie_name, "expired-tok")
|
||
resp = client.get(
|
||
"/api/v1/trade-in/dummy",
|
||
headers={"X-Authenticated-User": "kopylov"},
|
||
)
|
||
assert resp.status_code == 200
|
||
assert resp.json() == {"ok": True}
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# rbac_guard dual vs db_only
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def test_db_only_mode_rejects_legacy_header_without_session(
|
||
client: TestClient, monkeypatch: pytest.MonkeyPatch
|
||
) -> None:
|
||
monkeypatch.setattr(config.settings, "auth_mode", "db_only")
|
||
|
||
resp = client.get(
|
||
"/api/v1/trade-in/dummy",
|
||
headers={"X-Authenticated-User": "admin"},
|
||
)
|
||
assert resp.status_code == 401
|
||
assert "session" in resp.json()["detail"].lower()
|
||
|
||
|
||
def test_db_only_mode_accepts_valid_session(
|
||
client: TestClient, store: _Store, monkeypatch: pytest.MonkeyPatch
|
||
) -> None:
|
||
store.add_user("alice", hash_password("Secret123!"), role="employee")
|
||
client.post("/api/v1/auth/login", json={"username": "alice", "password": "Secret123!"})
|
||
|
||
monkeypatch.setattr(config.settings, "auth_mode", "db_only")
|
||
|
||
resp = client.get("/api/v1/trade-in/dummy")
|
||
assert resp.status_code == 200
|
||
assert resp.json() == {"ok": True}
|
||
|
||
|
||
def test_session_user_can_reach_tradein_but_not_admin(client: TestClient, store: _Store) -> None:
|
||
store.add_user("alice", hash_password("Secret123!"), role="employee")
|
||
client.post("/api/v1/auth/login", json={"username": "alice", "password": "Secret123!"})
|
||
|
||
ok = client.get("/api/v1/trade-in/dummy")
|
||
assert ok.status_code == 200
|
||
|
||
denied = client.get("/api/v1/admin/dummy")
|
||
# rbac_guard's admin-gate matches the path regex BEFORE routing even happens
|
||
# (route isn't registered on this test app) — role=employee != admin -> 403,
|
||
# never a 404 (a bare "any non-2xx" assertion would mask a rbac_guard typo).
|
||
assert denied.status_code == 403
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# #2552 post-review CRITICAL fix: session identity must win over a spoofed
|
||
# client-sent X-Authenticated-User header (was a skip-if-present bug — the
|
||
# forged header used to override the session for every downstream reader of
|
||
# the raw header: _assert_estimate_access*, account_quota, /trade-in/history,
|
||
# support.py — in BOTH auth_mode=dual and db_only).
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def test_session_identity_wins_over_spoofed_header_dual_mode(
|
||
client: TestClient, store: _Store
|
||
) -> None:
|
||
store.add_user("alice", hash_password("Secret123!"), role="employee")
|
||
store.add_user("victim", hash_password("Secret123!"), role="employee")
|
||
client.post("/api/v1/auth/login", json={"username": "alice", "password": "Secret123!"})
|
||
|
||
resp = client.get(
|
||
"/api/v1/trade-in/whoami",
|
||
headers={"X-Authenticated-User": "victim"},
|
||
)
|
||
assert resp.status_code == 200
|
||
assert resp.json()["user"] == "alice"
|
||
|
||
|
||
def test_session_identity_wins_over_spoofed_header_db_only_mode(
|
||
client: TestClient, store: _Store, monkeypatch: pytest.MonkeyPatch
|
||
) -> None:
|
||
store.add_user("alice", hash_password("Secret123!"), role="employee")
|
||
store.add_user("victim", hash_password("Secret123!"), role="employee")
|
||
client.post("/api/v1/auth/login", json={"username": "alice", "password": "Secret123!"})
|
||
|
||
monkeypatch.setattr(config.settings, "auth_mode", "db_only")
|
||
|
||
resp = client.get(
|
||
"/api/v1/trade-in/whoami",
|
||
headers={"X-Authenticated-User": "victim"},
|
||
)
|
||
assert resp.status_code == 200
|
||
assert resp.json()["user"] == "alice"
|
||
|
||
|
||
def test_cyrillic_username_session_propagation_does_not_500(
|
||
client: TestClient, store: _Store
|
||
) -> None:
|
||
"""#2552 post-review Medium 1: `.encode("latin-1")` без errors="replace" на
|
||
кириллическом username крашил бы КАЖДЫЙ запрос такого юзера с 500."""
|
||
store.add_user("алиса", hash_password("Secret123!"), role="employee")
|
||
login_resp = client.post(
|
||
"/api/v1/auth/login", json={"username": "алиса", "password": "Secret123!"}
|
||
)
|
||
assert login_resp.status_code == 200, login_resp.text
|
||
|
||
resp = client.get("/api/v1/trade-in/whoami")
|
||
assert resp.status_code == 200, resp.text
|
||
# latin-1 "replace" гарантированно не крашит — точное значение (что именно
|
||
# получится из non-latin1 байт) не является контрактом, важно отсутствие 500.
|
||
assert resp.json()["user"] is not None
|