From 11941caa202815cfa83b277aa76a6165b9dfe4bb Mon Sep 17 00:00:00 2001 From: Andrey Epifancev Date: Sat, 3 Oct 2026 22:14:10 +0400 Subject: [PATCH] =?UTF-8?q?feat:=20=D0=B7=D0=B0=D0=B2=D0=B8=D1=81=D0=B8?= =?UTF-8?q?=D0=BC=D0=BE=D1=81=D1=82=D0=B8=20current=5Fuser=20=D0=B8=20requ?= =?UTF-8?q?ire(code),=20=D1=80=D0=B5=D0=BF=D0=BE=D0=B7=D0=B8=D1=82=D0=BE?= =?UTF-8?q?=D1=80=D0=B8=D0=B9=20=D0=BF=D0=BE=D0=BB=D1=8C=D0=B7=D0=BE=D0=B2?= =?UTF-8?q?=D0=B0=D1=82=D0=B5=D0=BB=D0=B5=D0=B9,=20=D1=81=D1=85=D0=B5?= =?UTF-8?q?=D0=BC=D0=B0=20me?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Пользователь, филиал, роли и права читаются одним запросом; роль is_active=false прав не даёт; email без учёта регистра, неоднозначность — 403 (R-17). Co-Authored-By: Claude Fable 5.1 --- api/src/api/dependencies/auth.py | 109 +++++++ api/src/repositories/users.py | 73 +++++ api/src/schemas/me.py | 29 ++ api/tests/test_auth_deps.py | 479 +++++++++++++++++++++++++++++++ 4 files changed, 690 insertions(+) create mode 100644 api/src/api/dependencies/auth.py create mode 100644 api/src/repositories/users.py create mode 100644 api/src/schemas/me.py create mode 100644 api/tests/test_auth_deps.py diff --git a/api/src/api/dependencies/auth.py b/api/src/api/dependencies/auth.py new file mode 100644 index 0000000..69ad7f1 --- /dev/null +++ b/api/src/api/dependencies/auth.py @@ -0,0 +1,109 @@ +"""Зависимости FastAPI: личность из JWT, пользователь из БД, право `require(code)` (spec authn-authz, D-5, D-10). + +Цепочка: заголовок `Authorization: Bearer` → проверяющий токены (`TokenVerifier`) → ОДИН запрос в БД +(`load_principal_data`) → :class:`Principal`. Зависимости кэшируются FastAPI в пределах запроса, поэтому любое число +`require(...)` и `current_user` на одном маршруте даёт один запрос к БД. + +Ответы: 401 — нет/невалидный токен (с `WWW-Authenticate: Bearer`); 403 — пользователь не найден, неоднозначен, +отключён либо нет права. Тело — `{"detail": "<причина по-русски>"}` (форматирует обработчик `HTTPException`). +Права проверяются только по кодам прав: код роли в логике не используется. +""" + +from collections.abc import Awaitable, Callable +from typing import Annotated + +from fastapi import Depends, Header, HTTPException, Request, status +from sqlalchemy.ext.asyncio import AsyncSession + +from src.core.auth import AuthError, Claims, TokenVerifier, build_verifier +from src.core.config import get_settings +from src.db.db import get_async_session +from src.repositories.users import AmbiguousUserError, UserWithAccess, load_principal_data + +# Текущий пользователь вместе с филиалом, ролями и эффективными правами (то, что нужно `GET /me` и `require`) +Principal = UserWithAccess + +_BEARER = "Bearer" +_UNAUTHORIZED_HEADERS = {"WWW-Authenticate": _BEARER} +_MISSING_HEADER = "требуется заголовок Authorization: Bearer <токен>" +_BEARER_PARTS = 2 + + +def _unauthorized(detail: str) -> HTTPException: + return HTTPException(status.HTTP_401_UNAUTHORIZED, detail=detail, headers=dict(_UNAUTHORIZED_HEADERS)) + + +def _forbidden(detail: str) -> HTTPException: + return HTTPException(status.HTTP_403_FORBIDDEN, detail=detail) + + +def parse_bearer(authorization: str | None) -> str: + """Токен из значения заголовка `Authorization: Bearer `; иначе 401 (схема без учёта регистра).""" + if authorization is not None: + parts = authorization.split() + if len(parts) == _BEARER_PARTS and parts[0].lower() == _BEARER.lower(): + return parts[1] + raise _unauthorized(_MISSING_HEADER) + + +def get_token_verifier(request: Request) -> TokenVerifier | None: + """Проверяющий токены приложения: строится лениво по настройкам и кэшируется в `app.state.token_verifier`. + + `None` — проверять нечем (ничего не настроено). В тестах переопределяется через `app.dependency_overrides`. + """ + state = request.app.state + if not hasattr(state, "token_verifier"): + state.token_verifier = build_verifier(get_settings()) + verifier: TokenVerifier | None = state.token_verifier + return verifier + + +async def verified_claims( + verifier: Annotated[TokenVerifier | None, Depends(get_token_verifier)], + authorization: Annotated[str | None, Header()] = None, +) -> Claims: + """Claims проверенного токена. Не обращается к БД: все 401 решаются до неё.""" + token = parse_bearer(authorization) + if verifier is None: + raise _unauthorized("аутентификация не настроена") + try: + return await verifier.verify(token) + except AuthError as error: + raise _unauthorized(str(error)) from error + + +async def current_user( + claims: Annotated[Claims, Depends(verified_claims)], + session: Annotated[AsyncSession, Depends(get_async_session)], +) -> Principal: + """Пользователь по email из токена с филиалом, ролями и эффективными правами (один запрос к БД). + + `Claims.email` уже нормализован (`strip().lower()`), сравнение в БД — по `lower(app_user.email)` (R-17). + 403: пользователя нет в АРМ; учётная запись неоднозначна (несколько email, различающихся регистром) или отключена. + """ + try: + principal = await load_principal_data(session, claims.email) + except AmbiguousUserError as error: + raise _forbidden("неоднозначная учётная запись") from error + if principal is None: + raise _forbidden("пользователь не зарегистрирован в АРМ") + if not principal.user.is_active: + raise _forbidden("учётная запись отключена") + return principal + + +def require(code: str) -> Callable[..., Awaitable[Principal]]: + """Фабрика зависимости: пропускает, только если среди эффективных прав есть `code`; иначе 403 с названием права. + + Код права не сверяется с enum на этапе создания (проверяется лишь непустота): права — данные. Роли не + используются. Возвращает :class:`Principal`, так что зависимость можно использовать и как `current_user`. + """ + if not code or not code.strip(): + raise ValueError("код права для require() не задан") + + async def check(principal: Annotated[Principal, Depends(current_user)]) -> Principal: + if code not in principal.permissions: + raise _forbidden(f"недостаточно прав: требуется {code}") + return principal + + return check diff --git a/api/src/repositories/users.py b/api/src/repositories/users.py new file mode 100644 index 0000000..5c66039 --- /dev/null +++ b/api/src/repositories/users.py @@ -0,0 +1,73 @@ +"""Репозиторий пользователей: личность и эффективные права одним запросом (spec authn-authz, D-5, R-17). + +`effective_permissions(user) = ∪ permission.code по всем активным ролям пользователя` (CLAUDE.md: права — данные). +Код роли в проверках не участвует: роли возвращаются только для показа (`GET /me`). +""" + +from dataclasses import dataclass + +from sqlalchemy import and_, func, select +from sqlalchemy.ext.asyncio import AsyncSession +from sqlalchemy.orm import aliased + +from src.models.access import AppUser, Permission, Role, RolePermission, UserRole +from src.models.directories import RegionalOffice + +# Достаточно двух строк, чтобы заметить неоднозначность; больше пользователей читать незачем +_AMBIGUITY_PROBE_LIMIT = 2 + + +class AmbiguousUserError(Exception): + """По email найдено больше одного пользователя (email в БД уникален с учётом регистра, сравнение — без него).""" + + +@dataclass(frozen=True) +class UserWithAccess: + """Пользователь, его филиал, коды активных ролей и эффективные права (коды). + + `roles` — отсортированный кортеж кодов активных ролей; `permissions` — объединение прав этих ролей без дубликатов. + """ + + user: AppUser + regional_office: RegionalOffice + roles: tuple[str, ...] + permissions: frozenset[str] + + +async def load_principal_data(session: AsyncSession, email: str) -> UserWithAccess | None: + """Пользователь по email (без учёта регистра) вместе с филиалом, ролями и правами — ОДИН SQL-запрос. + + Возвращает `None`, если пользователя нет; бросает :class:`AmbiguousUserError`, если подходит больше одного + (fail-closed, R-17). Неактивный пользователь возвращается как есть (решение — за вызывающим). Роль с + `is_active=false` не даёт прав и в `roles` не попадает. + """ + matched = ( + select(AppUser) + .where(func.lower(AppUser.email) == email.lower()) + .order_by(AppUser.id) + .limit(_AMBIGUITY_PROBE_LIMIT) + .subquery() + ) + user = aliased(AppUser, matched) + statement = ( + select(user, RegionalOffice, Role.code, Permission.code) + .join(RegionalOffice, RegionalOffice.id == user.regional_office_id) + .outerjoin(UserRole, UserRole.user_id == user.id) + .outerjoin(Role, and_(Role.id == UserRole.role_id, Role.is_active.is_(True))) + .outerjoin(RolePermission, RolePermission.role_id == Role.id) + .outerjoin(Permission, Permission.id == RolePermission.permission_id) + ) + rows = (await session.execute(statement)).all() + if not rows: + return None + users = {row[0].id for row in rows} + if len(users) > 1: + raise AmbiguousUserError(email) + roles = {row[2] for row in rows if row[2] is not None} + permissions = frozenset(row[3] for row in rows if row[3] is not None) + return UserWithAccess( + user=rows[0][0], + regional_office=rows[0][1], + roles=tuple(sorted(roles)), + permissions=permissions, + ) diff --git a/api/src/schemas/me.py b/api/src/schemas/me.py new file mode 100644 index 0000000..ea11a2a --- /dev/null +++ b/api/src/schemas/me.py @@ -0,0 +1,29 @@ +"""Схема ответа `GET /api/v1/me` (spec authn-authz): профиль, роли и эффективные права текущего пользователя.""" + +import uuid + +from pydantic import BaseModel, ConfigDict + +from src.models.enums import BusinessSegment + + +class RegionalOfficeOut(BaseModel): + """Филиал пользователя.""" + + id: uuid.UUID + name: str + is_head_office: bool + + +class MeResponse(BaseModel): + """Текущий пользователь. `roles` и `permissions` — коды, отсортированные по алфавиту.""" + + model_config = ConfigDict(use_enum_values=True) + + id: uuid.UUID + email: str + full_name: str + regional_office: RegionalOfficeOut + business_segment: BusinessSegment | None + roles: list[str] + permissions: list[str] diff --git a/api/tests/test_auth_deps.py b/api/tests/test_auth_deps.py new file mode 100644 index 0000000..4123d94 --- /dev/null +++ b/api/tests/test_auth_deps.py @@ -0,0 +1,479 @@ +"""Зависимости аутентификации и прав (spec authn-authz, задача 17): `current_user`, `require(code)`, 401/403. + +БД-тесты (`integration`) работают на `db_session` с seed dev-данных; мини-приложение собрано в тесте без lifespan, +верификатор токенов подменён на `DevHs256Verifier`. Быстрые тесты (без БД) проверяют разбор заголовка и то, что +401-ветки не обращаются к сессии. +""" + +import ast +import uuid +from collections.abc import AsyncIterator, Callable, Iterator +from datetime import timedelta +from pathlib import Path +from typing import Annotated, Any + +import httpx +import pytest +import yaml +from fastapi import Depends, FastAPI, HTTPException +from sqlalchemy import event, select, update +from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession + +from scripts.seed import run_seed +from src.api.dependencies.auth import ( + Principal, + current_user, + get_token_verifier, + parse_bearer, + require, +) +from src.core.auth import DevHs256Verifier +from src.core.exception_handlers import register_exception_handlers +from src.db.db import get_async_session +from src.models.access import AppUser, Role, UserRole +from src.models.directories import RegionalOffice +from src.models.enums import BusinessSegment +from src.repositories.users import AmbiguousUserError, load_principal_data +from src.schemas.me import MeResponse, RegionalOfficeOut + +SECRET = "test-secret-for-auth-deps-0123456789abcdef" +OTHER_SECRET = "foreign-secret-for-auth-deps-0123456789ab" +SRC = Path(__file__).resolve().parent.parent / "src" +SEED_CORE = Path(__file__).resolve().parent.parent / "seed" / "core" + +NOT_REGISTERED = "пользователь не зарегистрирован в АРМ" +DISABLED = "учётная запись отключена" +AMBIGUOUS = "неоднозначная учётная запись" + + +class _ForbiddenSession: + """Заглушка сессии для 401-веток: любое обращение — ошибка теста.""" + + def __getattr__(self, name: str) -> Any: + raise AssertionError(f"обращение к БД при 401: session.{name}") + + +def _build_app(session_factory: Callable[[], AsyncIterator[Any]], verifier: DevHs256Verifier | None) -> FastAPI: + app = FastAPI() + register_exception_handlers(app) + + @app.get("/whoami") + async def whoami(principal: Annotated[Principal, Depends(current_user)]) -> dict[str, Any]: + return { + "email": principal.user.email, + "roles": list(principal.roles), + "permissions": sorted(principal.permissions), + "office": principal.regional_office.name, + } + + @app.get("/kfo-verify", dependencies=[Depends(require("KFO_VERIFY"))]) + async def kfo_verify() -> dict[str, bool]: + return {"ok": True} + + @app.get("/doc-edit", dependencies=[Depends(require("DOCUMENT_EDIT"))]) + async def doc_edit() -> dict[str, bool]: + return {"ok": True} + + @app.get("/both", dependencies=[Depends(require("KFO_VERIFY")), Depends(require("DOCUMENT_EDIT"))]) + async def both(principal: Annotated[Principal, Depends(current_user)]) -> dict[str, bool]: + return {"ok": True} + + app.dependency_overrides[get_async_session] = session_factory + app.dependency_overrides[get_token_verifier] = lambda: verifier + return app + + +def _client(app: FastAPI) -> httpx.AsyncClient: + return httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://test") + + +def _bearer(token: str) -> dict[str, str]: + return {"Authorization": f"Bearer {token}"} + + +# --- быстрые тесты без БД --------------------------------------------------------------------------------------- + + +def test_parse_bearer_ok() -> None: + assert parse_bearer("Bearer abc.def.ghi") == "abc.def.ghi" + assert parse_bearer("bearer abc") == "abc" + + +@pytest.mark.parametrize("header", [None, "", "Bearer", "Bearer ", "Bearer ", "Basic abc", "abc.def", "Bearer a b"]) +def test_parse_bearer_rejects(header: str | None) -> None: + with pytest.raises(HTTPException) as info: + parse_bearer(header) + assert info.value.status_code == 401 # noqa: PLR2004 + assert info.value.headers == {"WWW-Authenticate": "Bearer"} + assert "Authorization: Bearer" in info.value.detail + + +def test_require_rejects_empty_code() -> None: + with pytest.raises(ValueError, match="код права"): + require("") + with pytest.raises(ValueError, match="код права"): + require(" ") + + +@pytest.fixture +def untouchable_app() -> tuple[FastAPI, list[str]]: + """Приложение с заглушкой сессии; список `touched` пополняется, если зависимость сессии вообще вызвана.""" + touched: list[str] = [] + + async def session() -> AsyncIterator[_ForbiddenSession]: + touched.append("session") + yield _ForbiddenSession() + + verifier = DevHs256Verifier(SECRET) + return _build_app(session, verifier), touched + + +async def _expect_401(app: FastAPI, headers: dict[str, str]) -> httpx.Response: + async with _client(app) as client: + response = await client.get("/whoami", headers=headers) + assert response.status_code == 401 # noqa: PLR2004 + assert response.headers["www-authenticate"] == "Bearer" + assert isinstance(response.json()["detail"], str) + return response + + +async def test_401_without_header_does_not_touch_db(untouchable_app: tuple[FastAPI, list[str]]) -> None: + app, touched = untouchable_app + response = await _expect_401(app, {}) + assert "Authorization: Bearer" in response.json()["detail"] + assert touched == [] + + +async def test_401_not_bearer_does_not_touch_db(untouchable_app: tuple[FastAPI, list[str]]) -> None: + app, touched = untouchable_app + await _expect_401(app, {"Authorization": "Basic dXNlcjpwYXNz"}) + assert touched == [] + + +async def test_401_garbage_token_does_not_touch_db(untouchable_app: tuple[FastAPI, list[str]]) -> None: + app, touched = untouchable_app + response = await _expect_401(app, _bearer("not-a-jwt")) + assert response.json() == {"detail": "некорректный токен"} + assert touched == [] + + +async def test_401_expired_token_does_not_touch_db(untouchable_app: tuple[FastAPI, list[str]]) -> None: + app, touched = untouchable_app + token = DevHs256Verifier(SECRET).issue_token("ka_go@armka.local", ttl=timedelta(minutes=-5)) + response = await _expect_401(app, _bearer(token)) + assert response.json() == {"detail": "срок действия токена истёк"} + assert touched == [] + + +async def test_401_foreign_signature_does_not_touch_db(untouchable_app: tuple[FastAPI, list[str]]) -> None: + app, touched = untouchable_app + token = DevHs256Verifier(OTHER_SECRET).issue_token("ka_go@armka.local") + response = await _expect_401(app, _bearer(token)) + assert response.json() == {"detail": "недействительная подпись токена"} + assert touched == [] + + +async def test_401_when_verifier_not_configured() -> None: + touched: list[str] = [] + + async def session() -> AsyncIterator[_ForbiddenSession]: + touched.append("session") + yield _ForbiddenSession() + + app = _build_app(session, None) + token = DevHs256Verifier(SECRET).issue_token("ka_go@armka.local") + response = await _expect_401(app, _bearer(token)) + assert response.json() == {"detail": "аутентификация не настроена"} + assert touched == [] + + +def test_get_token_verifier_builds_lazily_and_caches(monkeypatch: pytest.MonkeyPatch) -> None: + """Проверяющий строится один раз на приложение (в том числе `None`) и кладётся в `app.state.token_verifier`.""" + from types import SimpleNamespace # noqa: PLC0415 + + calls: list[object] = [] + sentinel = DevHs256Verifier(SECRET) + + def fake_build(settings: object) -> DevHs256Verifier: + calls.append(settings) + return sentinel + + monkeypatch.setattr("src.api.dependencies.auth.build_verifier", fake_build) + request = SimpleNamespace(app=SimpleNamespace(state=SimpleNamespace())) + assert get_token_verifier(request) is sentinel # type: ignore[arg-type] + assert get_token_verifier(request) is sentinel # type: ignore[arg-type] + assert len(calls) == 1 + assert request.app.state.token_verifier is sentinel + + monkeypatch.setattr("src.api.dependencies.auth.build_verifier", lambda settings: None) + request_none = SimpleNamespace(app=SimpleNamespace(state=SimpleNamespace())) + assert get_token_verifier(request_none) is None # type: ignore[arg-type] + assert request_none.app.state.token_verifier is None + + +# --- статическая проверка: код роли в логике не используется ------------------------------------------------------ + + +def test_role_codes_are_not_used_in_auth_logic() -> None: + """В `dependencies/auth.py` и `repositories/users.py` нет строковых констант — кодов ролей (права — данные).""" + roles = yaml.safe_load((SEED_CORE / "roles.yaml").read_text(encoding="utf-8")) + role_codes = {item["code"] for item in roles} + assert {"KA_RF", "KA_GO", "EXPERT"} <= role_codes + for relative in ("api/dependencies/auth.py", "repositories/users.py"): + tree = ast.parse((SRC / relative).read_text(encoding="utf-8")) + literals = { + node.value for node in ast.walk(tree) if isinstance(node, ast.Constant) and isinstance(node.value, str) + } + assert not literals & role_codes, f"{relative}: код роли в логике: {sorted(literals & role_codes)}" + + +# --- интеграционные тесты на seed ------------------------------------------------------------------------------- + + +@pytest.fixture +async def seeded(db_session: AsyncSession) -> AsyncSession: + await run_seed(db_session, dev=True) + await db_session.flush() + return db_session + + +@pytest.fixture +async def app(seeded: AsyncSession) -> FastAPI: + async def session() -> AsyncIterator[AsyncSession]: + yield seeded + + return _build_app(session, DevHs256Verifier(SECRET)) + + +@pytest.fixture +def token_for() -> Callable[[str], dict[str, str]]: + verifier = DevHs256Verifier(SECRET) + + def make(email: str) -> dict[str, str]: + return _bearer(verifier.issue_token(email)) + + return make + + +async def _add_user(session: AsyncSession, email: str, *, active: bool = True, roles: tuple[str, ...] = ()) -> AppUser: + office_id = (await session.execute(select(RegionalOffice.id).limit(1))).scalar_one() + user = AppUser(email=email, full_name="Тест Тестов", regional_office_id=office_id, is_active=active) + session.add(user) + await session.flush() + for code in roles: + role_id = (await session.execute(select(Role.id).where(Role.code == code))).scalar_one() + session.add(UserRole(user_id=user.id, role_id=role_id)) + await session.flush() + return user + + +def _seed_permissions(*role_codes: str) -> set[str]: + matrix = yaml.safe_load((SEED_CORE / "role_permissions.yaml").read_text(encoding="utf-8")) + return {code for role in role_codes for code in matrix[role]} + + +@pytest.mark.integration +async def test_permission_present(app: FastAPI, token_for: Callable[[str], dict[str, str]]) -> None: + """Сценарий «Право есть»: KA_GO, KFO_VERIFY.""" + async with _client(app) as client: + response = await client.get("/kfo-verify", headers=token_for("ka_go@armka.local")) + assert response.status_code == 200 # noqa: PLR2004 + assert response.json() == {"ok": True} + + +@pytest.mark.integration +async def test_permission_absent(app: FastAPI, token_for: Callable[[str], dict[str, str]]) -> None: + """Сценарий «Права нет»: EXPERT, DOCUMENT_EDIT → 403 с названием права.""" + async with _client(app) as client: + response = await client.get("/doc-edit", headers=token_for("expert@armka.local")) + assert response.status_code == 403 # noqa: PLR2004 + assert response.json() == {"detail": "недостаточно прав: требуется DOCUMENT_EDIT"} + + +@pytest.mark.integration +async def test_whoami_returns_user_office_roles_and_permissions( + app: FastAPI, token_for: Callable[[str], dict[str, str]] +) -> None: + async with _client(app) as client: + response = await client.get("/whoami", headers=token_for("ka_go@armka.local")) + assert response.status_code == 200 # noqa: PLR2004 + body = response.json() + assert body["email"] == "ka_go@armka.local" + assert body["roles"] == ["KA_GO"] + assert body["permissions"] == sorted(_seed_permissions("KA_GO")) + assert body["office"] + + +@pytest.mark.integration +async def test_two_roles_union_without_duplicates( + seeded: AsyncSession, app: FastAPI, token_for: Callable[[str], dict[str, str]] +) -> None: + """Сценарий «Две роли»: KA_RF + CURATOR_GO → объединение прав без дубликатов.""" + await _add_user(seeded, "two.roles@armka.local", roles=("KA_RF", "CURATOR_GO")) + ka_rf, curator = _seed_permissions("KA_RF"), _seed_permissions("CURATOR_GO") + assert ka_rf & curator, "тест осмыслен, только если у ролей есть общие права" + async with _client(app) as client: + response = await client.get("/whoami", headers=token_for("two.roles@armka.local")) + assert response.status_code == 200 # noqa: PLR2004 + body = response.json() + assert body["roles"] == ["CURATOR_GO", "KA_RF"] + assert body["permissions"] == sorted(ka_rf | curator) + assert len(body["permissions"]) == len(set(body["permissions"])) + + +@pytest.mark.integration +async def test_unknown_user_is_403(app: FastAPI, token_for: Callable[[str], dict[str, str]]) -> None: + async with _client(app) as client: + response = await client.get("/whoami", headers=token_for("nobody@armka.local")) + assert response.status_code == 403 # noqa: PLR2004 + assert response.json() == {"detail": NOT_REGISTERED} + + +@pytest.mark.integration +async def test_inactive_user_is_403( + seeded: AsyncSession, app: FastAPI, token_for: Callable[[str], dict[str, str]] +) -> None: + await _add_user(seeded, "off@armka.local", active=False, roles=("KA_GO",)) + async with _client(app) as client: + response = await client.get("/kfo-verify", headers=token_for("off@armka.local")) + assert response.status_code == 403 # noqa: PLR2004 + assert response.json() == {"detail": DISABLED} + + +@pytest.mark.integration +async def test_email_case_variants_are_ambiguous_403( + seeded: AsyncSession, app: FastAPI, token_for: Callable[[str], dict[str, str]] +) -> None: + """R-17: два пользователя с email, различающимися регистром → 403, а не «первый попавшийся».""" + await _add_user(seeded, "Ivan@x.local", roles=("KA_GO",)) + await _add_user(seeded, "ivan@x.local", roles=("EXPERT",)) + async with _client(app) as client: + response = await client.get("/whoami", headers=token_for("IVAN@x.local")) + assert response.status_code == 403 # noqa: PLR2004 + assert response.json() == {"detail": AMBIGUOUS} + + +@pytest.mark.integration +async def test_token_email_case_insensitive(app: FastAPI, token_for: Callable[[str], dict[str, str]]) -> None: + async with _client(app) as client: + response = await client.get("/kfo-verify", headers=token_for("KA_GO@Armka.Local")) + assert response.status_code == 200 # noqa: PLR2004 + + +@pytest.mark.integration +async def test_user_email_stored_in_mixed_case_is_found( + seeded: AsyncSession, app: FastAPI, token_for: Callable[[str], dict[str, str]] +) -> None: + await _add_user(seeded, "Mixed.Case@Armka.Local", roles=("KA_GO",)) + async with _client(app) as client: + response = await client.get("/kfo-verify", headers=token_for("mixed.case@armka.local")) + assert response.status_code == 200 # noqa: PLR2004 + + +@pytest.mark.integration +async def test_inactive_role_grants_nothing( + seeded: AsyncSession, app: FastAPI, token_for: Callable[[str], dict[str, str]] +) -> None: + await _add_user(seeded, "mix@armka.local", roles=("KA_GO", "EXPERT")) + await seeded.execute(update(Role).where(Role.code == "KA_GO").values(is_active=False)) + async with _client(app) as client: + verify = await client.get("/kfo-verify", headers=token_for("mix@armka.local")) + who = await client.get("/whoami", headers=token_for("mix@armka.local")) + assert verify.status_code == 403 # noqa: PLR2004 + assert who.json()["roles"] == ["EXPERT"] + assert who.json()["permissions"] == sorted(_seed_permissions("EXPERT")) + + +@pytest.mark.integration +async def test_user_without_roles_has_no_permissions( + seeded: AsyncSession, app: FastAPI, token_for: Callable[[str], dict[str, str]] +) -> None: + await _add_user(seeded, "norole@armka.local") + async with _client(app) as client: + who = await client.get("/whoami", headers=token_for("norole@armka.local")) + verify = await client.get("/kfo-verify", headers=token_for("norole@armka.local")) + assert who.status_code == 200 # noqa: PLR2004 + assert who.json()["roles"] == [] + assert who.json()["permissions"] == [] + assert verify.status_code == 403 # noqa: PLR2004 + + +@pytest.fixture +def sql_log(db_engine: AsyncEngine) -> Iterator[list[str]]: + """Все SQL-выражения, выполненные через движок тестовой БД с момента запроса фикстуры.""" + statements: list[str] = [] + + def record(_conn: Any, _cursor: Any, statement: str, *_rest: Any) -> None: + statements.append(statement) + + event.listen(db_engine.sync_engine, "before_cursor_execute", record) + try: + yield statements + finally: + event.remove(db_engine.sync_engine, "before_cursor_execute", record) + + +def _principal_queries(statements: list[str]) -> list[str]: + return [s for s in statements if "app_user" in s] + + +@pytest.mark.integration +async def test_db_queried_once_per_request( + app: FastAPI, token_for: Callable[[str], dict[str, str]], sql_log: list[str] +) -> None: + """Два `require` и `current_user` на одном маршруте — один запрос к пользователю и правам.""" + async with _client(app) as client: + single = await client.get("/whoami", headers=token_for("ka_go@armka.local")) + assert single.status_code == 200 # noqa: PLR2004 + queries_single = len(_principal_queries(sql_log)) + sql_log.clear() + combined = await client.get("/both", headers=token_for("ka_go@armka.local")) + assert combined.status_code == 200 # noqa: PLR2004 + assert queries_single == 1 + assert len(_principal_queries(sql_log)) == 1 + assert len(sql_log) == 1, sql_log + + +@pytest.mark.integration +async def test_both_requires_checked_on_one_route(app: FastAPI, token_for: Callable[[str], dict[str, str]]) -> None: + async with _client(app) as client: + response = await client.get("/both", headers=token_for("expert@armka.local")) + assert response.status_code == 403 # noqa: PLR2004 + assert response.json() == {"detail": "недостаточно прав: требуется KFO_VERIFY"} + + +@pytest.mark.integration +async def test_repository_load_principal_data(seeded: AsyncSession) -> None: + assert await load_principal_data(seeded, "nobody@armka.local") is None + data = await load_principal_data(seeded, "ka_go@armka.local") + assert data is not None + assert data.user.email == "ka_go@armka.local" + assert data.roles == ("KA_GO",) + assert data.permissions == frozenset(_seed_permissions("KA_GO")) + assert data.regional_office.id == data.user.regional_office_id + + +@pytest.mark.integration +async def test_repository_raises_on_ambiguity(seeded: AsyncSession) -> None: + await _add_user(seeded, "Dup@armka.local", roles=("EXPERT",)) + await _add_user(seeded, "dup@armka.local", roles=("KA_GO",)) + with pytest.raises(AmbiguousUserError): + await load_principal_data(seeded, "dup@armka.local") + + +def test_me_response_schema_serializes_enum_and_null_segment() -> None: + office = RegionalOfficeOut(id=uuid.uuid4(), name="ГО", is_head_office=True) + fields: dict[str, Any] = { + "id": uuid.uuid4(), + "email": "a@b.c", + "full_name": "А", + "regional_office": office, + "roles": ["KA_GO"], + "permissions": ["KFO_VIEW"], + } + empty = MeResponse(business_segment=None, **fields) + assert empty.model_dump(mode="json")["business_segment"] is None + segment = next(iter(BusinessSegment)) + filled = MeResponse(business_segment=segment, **fields) + assert filled.model_dump()["business_segment"] == str(segment) + assert type(filled.model_dump()["business_segment"]) is str