ef7458eddb
Раздел SESSION, object_type app_user, IP и user-agent запроса (`services/session_audit.py`): - LOGIN — выдача dev-токена (в транзакции запроса) и первый успешный `current_user` пользователя после старта процесса (`SessionAudit.logged_in`, в памяти процесса; параллельные первые запросы — одно событие); - LOGIN_FAILED — 401 (токен не принят: user_id пуст, причина без токена) и 403 аутентификации (не зарегистрирован / отключён / неоднозначен: email в description, user_id — если найден). `AuthenticationFailedError` (подкласс HTTPException) обрабатывается после отката транзакции запроса, запись — в отдельной транзакции `session_transaction`, поэтому остаётся в журнале. Без заголовка Authorization и 403 «недостаточно прав» — не события входа; ошибка записи аудита входа ответ не меняет. Тесты: `test_login_audit.py` (в т.ч. фиксация при откате запроса на `committed_app`); счётчики следов импорта не учитывают события SESSION. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_012vJkmBGKKqMoEMzvsXG3y2
256 lines
11 KiB
Python
256 lines
11 KiB
Python
"""Аудит входа LOGIN / LOGIN_FAILED (OQ-64, issue #9 п.5): раздел SESSION, `object_type = "app_user"`.
|
||
|
||
LOGIN — успешная выдача dev-токена и первый успешный запрос пользователя с токеном после старта процесса;
|
||
LOGIN_FAILED — 401 (токен не принят) и 403 аутентификации (пользователь не найден, отключён, неоднозначен).
|
||
Запрос без заголовка `Authorization` и 403 «недостаточно прав» входом не считаются. LOGIN_FAILED пишется в отдельной
|
||
транзакции и переживает откат транзакции запроса — проверяется на `committed_app` (сессия запроса как в проде).
|
||
"""
|
||
|
||
import asyncio
|
||
import uuid
|
||
from collections.abc import AsyncIterator
|
||
from contextlib import asynccontextmanager
|
||
|
||
import httpx
|
||
import pytest
|
||
from fastapi import FastAPI
|
||
from sqlalchemy import select, update
|
||
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
|
||
|
||
from src.core.auth import DevHs256Verifier
|
||
from src.core.config import get_settings
|
||
from src.models.access import AppUser
|
||
from src.models.audit_ex import AuditEvent
|
||
from src.models.enums import AuditAction, AuditSection
|
||
from src.services.session_audit import FIRST_REQUEST_DESCRIPTION, SessionAudit
|
||
|
||
HTTP_OK = 200
|
||
HTTP_UNAUTHORIZED = 401
|
||
HTTP_FORBIDDEN = 403
|
||
KA_RF = "ka_rf@armka.local"
|
||
EXPERT = "expert@armka.local"
|
||
USER_AGENT = "login-audit-test/1.0"
|
||
GARBAGE_TOKEN = "garbage.token.value" # noqa: S105 - заведомо неверный токен, не секрет
|
||
|
||
|
||
def _client(app: FastAPI) -> httpx.AsyncClient:
|
||
return httpx.AsyncClient(
|
||
transport=httpx.ASGITransport(app=app), base_url="http://test", headers={"User-Agent": USER_AGENT}
|
||
)
|
||
|
||
|
||
def _token(email: str) -> str:
|
||
return DevHs256Verifier(get_settings().AUTH_DEV_SECRET).issue_token(email)
|
||
|
||
|
||
def _bearer(token: str) -> dict[str, str]:
|
||
return {"Authorization": f"Bearer {token}"}
|
||
|
||
|
||
async def _events(session: AsyncSession) -> list[AuditEvent]:
|
||
rows = await session.scalars(
|
||
select(AuditEvent)
|
||
.where(AuditEvent.action.in_([AuditAction.LOGIN, AuditAction.LOGIN_FAILED]))
|
||
.order_by(AuditEvent.occurred_at)
|
||
)
|
||
return list(rows.all())
|
||
|
||
|
||
async def _user_id(session: AsyncSession, email: str) -> uuid.UUID:
|
||
user_id = await session.scalar(select(AppUser.id).where(AppUser.email == email))
|
||
assert user_id is not None
|
||
return user_id
|
||
|
||
|
||
def _assert_session_event(event: AuditEvent, action: AuditAction, user_id: uuid.UUID | None) -> None:
|
||
assert event.action is action
|
||
assert event.section is AuditSection.SESSION
|
||
assert event.object_type == "app_user"
|
||
assert event.user_id == user_id
|
||
assert event.object_id == user_id
|
||
assert event.ip_address == "127.0.0.1"
|
||
assert event.user_agent == USER_AGENT
|
||
|
||
|
||
# --- LOGIN ------------------------------------------------------------------------------------------------------
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_dev_token_issue_records_login(app: FastAPI, seeded: AsyncSession) -> None:
|
||
async with _client(app) as client:
|
||
response = await client.post("/api/v1/dev/token", json={"email": KA_RF})
|
||
assert response.status_code == HTTP_OK
|
||
|
||
(event,) = await _events(seeded)
|
||
_assert_session_event(event, AuditAction.LOGIN, await _user_id(seeded, KA_RF))
|
||
assert event.description is not None
|
||
assert "dev-токен" in event.description
|
||
assert response.json()["access_token"] not in event.description
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_first_request_with_token_records_login_once_per_user(app: FastAPI, seeded: AsyncSession) -> None:
|
||
async with _client(app) as client:
|
||
for _ in range(3):
|
||
assert (await client.get("/api/v1/me", headers=_bearer(_token(KA_RF)))).status_code == HTTP_OK
|
||
assert (await client.get("/api/v1/me", headers=_bearer(_token(EXPERT)))).status_code == HTTP_OK
|
||
|
||
events = await _events(seeded)
|
||
assert [(e.action, e.user_id) for e in events] == [
|
||
(AuditAction.LOGIN, await _user_id(seeded, KA_RF)),
|
||
(AuditAction.LOGIN, await _user_id(seeded, EXPERT)),
|
||
]
|
||
for event in events:
|
||
_assert_session_event(event, AuditAction.LOGIN, event.user_id)
|
||
assert event.description == FIRST_REQUEST_DESCRIPTION
|
||
assert app.state.session_audit.logged_in == {e.user_id for e in events}
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_login_set_is_per_process(app: FastAPI, seeded: AsyncSession) -> None:
|
||
"""Рестарт процесса (новый `SessionAudit`) — первый запрос снова пишет LOGIN."""
|
||
async with _client(app) as client:
|
||
await client.get("/api/v1/me", headers=_bearer(_token(KA_RF)))
|
||
app.state.session_audit = SessionAudit(app.state.session_audit.transaction)
|
||
await client.get("/api/v1/me", headers=_bearer(_token(KA_RF)))
|
||
|
||
assert [e.action for e in await _events(seeded)] == [AuditAction.LOGIN, AuditAction.LOGIN]
|
||
|
||
|
||
# --- LOGIN_FAILED -----------------------------------------------------------------------------------------------
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_invalid_token_records_login_failed_without_token(app: FastAPI, seeded: AsyncSession) -> None:
|
||
async with _client(app) as client:
|
||
response = await client.get("/api/v1/me", headers=_bearer(GARBAGE_TOKEN))
|
||
assert response.status_code == HTTP_UNAUTHORIZED
|
||
|
||
(event,) = await _events(seeded)
|
||
_assert_session_event(event, AuditAction.LOGIN_FAILED, None)
|
||
assert event.description is not None
|
||
assert response.json()["detail"] in event.description # причина отказа
|
||
assert GARBAGE_TOKEN not in event.description
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_unknown_user_records_login_failed_with_email(app: FastAPI, seeded: AsyncSession) -> None:
|
||
async with _client(app) as client:
|
||
response = await client.get("/api/v1/me", headers=_bearer(_token("ghost@example.com")))
|
||
assert response.status_code == HTTP_FORBIDDEN
|
||
|
||
(event,) = await _events(seeded)
|
||
_assert_session_event(event, AuditAction.LOGIN_FAILED, None)
|
||
assert event.description is not None
|
||
assert "ghost@example.com" in event.description
|
||
assert response.json()["detail"] in event.description
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_disabled_user_records_login_failed_with_user_id(app: FastAPI, seeded: AsyncSession) -> None:
|
||
user_id = await _user_id(seeded, EXPERT)
|
||
await seeded.execute(update(AppUser).where(AppUser.id == user_id).values(is_active=False))
|
||
async with _client(app) as client:
|
||
response = await client.get("/api/v1/me", headers=_bearer(_token(EXPERT)))
|
||
assert response.status_code == HTTP_FORBIDDEN
|
||
|
||
(event,) = await _events(seeded)
|
||
_assert_session_event(event, AuditAction.LOGIN_FAILED, user_id)
|
||
assert event.description is not None
|
||
assert EXPERT in event.description
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_missing_header_and_missing_permission_are_not_login_events(
|
||
committed_app: FastAPI, committed_sessionmaker: async_sessionmaker[AsyncSession]
|
||
) -> None:
|
||
"""Без заголовка — не попытка входа; 403 «недостаточно прав» — авторизация, не аутентификация.
|
||
|
||
LOGIN первого запроса expert записан, хотя сам запрос (403 по праву) откатил свою транзакцию.
|
||
"""
|
||
async with _client(committed_app) as client:
|
||
assert (await client.get("/api/v1/me")).status_code == HTTP_UNAUTHORIZED
|
||
expert = _bearer(_token(EXPERT))
|
||
assert (await client.get("/api/v1/admin/users", headers=expert)).status_code == HTTP_FORBIDDEN
|
||
|
||
async with committed_sessionmaker() as session:
|
||
events = await _events(session)
|
||
expert_id = await _user_id(session, EXPERT)
|
||
assert [(e.action, e.user_id) for e in events] == [(AuditAction.LOGIN, expert_id)]
|
||
|
||
|
||
# --- отдельная транзакция: LOGIN_FAILED переживает откат запроса -------------------------------------------------
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_login_failed_is_committed_although_request_rolls_back(
|
||
committed_app: FastAPI, committed_sessionmaker: async_sessionmaker[AsyncSession]
|
||
) -> None:
|
||
async with _client(committed_app) as client:
|
||
unauthorized = await client.get("/api/v1/me", headers=_bearer(GARBAGE_TOKEN))
|
||
forbidden = await client.get("/api/v1/me", headers=_bearer(_token("ghost@example.com")))
|
||
ok = await client.get("/api/v1/me", headers=_bearer(_token(KA_RF)))
|
||
assert (unauthorized.status_code, forbidden.status_code, ok.status_code) == (
|
||
HTTP_UNAUTHORIZED,
|
||
HTTP_FORBIDDEN,
|
||
HTTP_OK,
|
||
)
|
||
|
||
async with committed_sessionmaker() as session:
|
||
events = await _events(session)
|
||
ka_rf = await _user_id(session, KA_RF)
|
||
assert [(e.action, e.user_id) for e in events] == [
|
||
(AuditAction.LOGIN_FAILED, None),
|
||
(AuditAction.LOGIN_FAILED, None),
|
||
(AuditAction.LOGIN, ka_rf),
|
||
]
|
||
|
||
|
||
# --- SessionAudit без БД ----------------------------------------------------------------------------------------
|
||
|
||
|
||
class _RecordingSession:
|
||
def __init__(self, fail: bool) -> None: # noqa: FBT001
|
||
self.fail = fail
|
||
self.added: list[object] = []
|
||
|
||
def add(self, obj: object) -> None:
|
||
self.added.append(obj)
|
||
|
||
async def flush(self) -> None:
|
||
if self.fail:
|
||
raise RuntimeError("подставная ошибка БД")
|
||
|
||
|
||
def _fake_transaction(session: _RecordingSession): # noqa: ANN202
|
||
@asynccontextmanager
|
||
async def transaction() -> AsyncIterator[_RecordingSession]:
|
||
await asyncio.sleep(0) # точка переключения: параллельные первые запросы
|
||
yield session
|
||
|
||
return transaction
|
||
|
||
|
||
async def test_first_request_is_recorded_once_even_for_concurrent_requests() -> None:
|
||
session = _RecordingSession(fail=False)
|
||
audit = SessionAudit(_fake_transaction(session)) # type: ignore[arg-type]
|
||
user_id = uuid.uuid4()
|
||
|
||
await asyncio.gather(*(audit.first_request(user_id, ip_address="10.0.0.1", user_agent=None) for _ in range(5)))
|
||
|
||
assert len(session.added) == 1
|
||
assert audit.logged_in == {user_id}
|
||
|
||
|
||
async def test_failed_login_write_does_not_raise_and_is_retried() -> None:
|
||
session = _RecordingSession(fail=True)
|
||
audit = SessionAudit(_fake_transaction(session)) # type: ignore[arg-type]
|
||
user_id = uuid.uuid4()
|
||
|
||
await audit.first_request(user_id, ip_address=None, user_agent=None) # ошибка записи — в лог, не в ответ
|
||
|
||
assert audit.logged_in == set() # следующий запрос попробует записать LOGIN снова
|
||
session.fail = False
|
||
await audit.first_request(user_id, ip_address=None, user_agent=None)
|
||
assert audit.logged_in == {user_id}
|