ea119ffaa5
`clip` в `services/session_audit.py` применяется в `record_login` (общая точка LOGIN/LOGIN_FAILED): заголовок User-Agent и email из токена присылает клиент, без ограничения запись в журнал была бы произвольной длины. Тесты: помощник, запись `SessionAudit` без БД и длинный User-Agent при 401 через API. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_012vJkmBGKKqMoEMzvsXG3y2
295 lines
12 KiB
Python
295 lines
12 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 (
|
||
DESCRIPTION_MAX,
|
||
FIRST_REQUEST_DESCRIPTION,
|
||
USER_AGENT_MAX,
|
||
SessionAudit,
|
||
clip,
|
||
)
|
||
|
||
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}
|
||
|
||
|
||
# --- ограничение длины полей события входа (ревью R-1) ------------------------------------------------------------
|
||
|
||
|
||
def test_clip_limits_text_length() -> None:
|
||
assert clip(None, 5) is None
|
||
assert clip("abc", 5) == "abc"
|
||
assert clip("a" * 10, 5) == "aaaaa"
|
||
|
||
|
||
async def test_login_event_fields_are_clipped() -> None:
|
||
session = _RecordingSession(fail=False)
|
||
audit = SessionAudit(_fake_transaction(session)) # type: ignore[arg-type]
|
||
|
||
await audit.login_failed(None, "x" * 5000, ip_address="10.0.0.1", user_agent="U" * 5000)
|
||
|
||
(event,) = session.added
|
||
assert isinstance(event, AuditEvent)
|
||
assert event.user_agent == "U" * USER_AGENT_MAX
|
||
assert event.description == "x" * DESCRIPTION_MAX
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_long_user_agent_is_clipped_in_journal(app: FastAPI, seeded: AsyncSession) -> None:
|
||
async with _client(app) as client:
|
||
response = await client.get(
|
||
"/api/v1/me", headers={**_bearer(GARBAGE_TOKEN), "User-Agent": "A" * (USER_AGENT_MAX * 4)}
|
||
)
|
||
assert response.status_code == HTTP_UNAUTHORIZED
|
||
|
||
(event,) = await _events(seeded)
|
||
assert event.user_agent == "A" * USER_AGENT_MAX
|