From 2b76ae84974d50454f7a83909824060e959ebb6c Mon Sep 17 00:00:00 2001 From: PTah Date: Thu, 28 May 2026 09:13:12 +1000 Subject: [PATCH] feat: ingest HTTP 201/409 for event_id idempotency - Validate UUID format; log created/duplicate/rejected - Tests for 201, 409, 422; update agent-integration and work-plan - Docs: neutral IDE wording (no product-specific editor names) --- backend/app/api/v1/events.py | 42 +++++++++---- backend/app/services/ingest.py | 10 ++- backend/app/services/schema_validate.py | 2 +- backend/tests/conftest.py | 82 +++++++++++++++++++++++++ backend/tests/test_ingest.py | 63 +++++++++++++++++++ docs/INDEX.md | 2 +- docs/TZ.md | 2 +- docs/agent-integration.md | 19 ++++-- docs/diagrams/data-flow.md | 4 +- docs/work-plan.md | 8 +-- docs/workspace-three-repos.md | 2 +- 11 files changed, 210 insertions(+), 26 deletions(-) create mode 100644 backend/tests/conftest.py create mode 100644 backend/tests/test_ingest.py diff --git a/backend/app/api/v1/events.py b/backend/app/api/v1/events.py index 3aa2b8d..72b04ba 100644 --- a/backend/app/api/v1/events.py +++ b/backend/app/api/v1/events.py @@ -1,7 +1,9 @@ +import logging from datetime import datetime from typing import Any from fastapi import APIRouter, Body, Depends, HTTPException, Query +from fastapi.responses import JSONResponse from pydantic import BaseModel from sqlalchemy import func, select from sqlalchemy.orm import Session, joinedload @@ -18,6 +20,7 @@ from app.services.schema_validate import validate_event_payload from app.services.telegram_notify import notify_event, notify_problem router = APIRouter(prefix="/events", tags=["events"]) +logger = logging.getLogger("sac.ingest") class IngestResponse(BaseModel): @@ -28,14 +31,32 @@ class IngestResponse(BaseModel): problem_id: int | None = None -@router.post("", status_code=202, response_model=IngestResponse) +def _ingest_response(event: Event, *, created: bool) -> IngestResponse: + settings = get_settings() + base = settings.sac_public_url.rstrip("/") + return IngestResponse( + status="created" if created else "duplicate", + event_id=event.event_id, + created=created, + sac_event_url=f"{base}/api/v1/events/{event.id}", + problem_id=None, + ) + + +@router.post("", response_model=IngestResponse) def post_event( payload: dict[str, Any] = Body(...), db: Session = Depends(get_db), _api_key: str = Depends(get_api_key_auth), -) -> IngestResponse: +) -> JSONResponse: errors = validate_event_payload(payload) if errors: + event_id_hint = payload.get("event_id") if isinstance(payload.get("event_id"), str) else None + logger.warning( + "ingest rejected status=422 event_id=%s errors=%s", + event_id_hint, + errors[:5], + ) raise HTTPException(status_code=422, detail={"schema_errors": errors[:20]}) event, created = ingest_event(db, payload) @@ -46,17 +67,16 @@ def post_event( notify_event(event) if problem is not None and problem_created: notify_problem(problem, event) + logger.info("ingest created event_id=%s type=%s host_id=%s", event.event_id, event.type, event.host_id) + else: + logger.info("ingest duplicate event_id=%s", event.event_id) db.commit() - settings = get_settings() - base = settings.sac_public_url.rstrip("/") - return IngestResponse( - status="accepted", - event_id=event.event_id, - created=created, - sac_event_url=f"{base}/api/v1/events/{event.id}", - problem_id=problem.id if problem else None, - ) + body = _ingest_response(event, created=created) + if problem is not None: + body.problem_id = problem.id + status_code = 201 if created else 409 + return JSONResponse(status_code=status_code, content=body.model_dump(mode="json")) def _parse_optional_dt(value: str | None) -> datetime | None: diff --git a/backend/app/services/ingest.py b/backend/app/services/ingest.py index 2e67901..6c14bba 100644 --- a/backend/app/services/ingest.py +++ b/backend/app/services/ingest.py @@ -1,6 +1,7 @@ from datetime import datetime, timezone from sqlalchemy import select +from sqlalchemy.exc import IntegrityError from sqlalchemy.orm import Session from app.models import Event, Host @@ -85,5 +86,12 @@ def ingest_event(db: Session, payload: dict) -> tuple[Event, bool]: payload=payload, ) db.add(event) - db.flush() + try: + db.flush() + except IntegrityError: + db.rollback() + raced = db.scalar(select(Event).where(Event.event_id == event_id)) + if raced is not None: + return raced, False + raise return event, True diff --git a/backend/app/services/schema_validate.py b/backend/app/services/schema_validate.py index d2376c1..a9a99ef 100644 --- a/backend/app/services/schema_validate.py +++ b/backend/app/services/schema_validate.py @@ -15,7 +15,7 @@ def get_event_validator() -> Draft202012Validator: path = Path(__file__).resolve().parents[3] / "schemas" / "event-schema-v1.json" schema = json.loads(path.read_text(encoding="utf-8")) Draft202012Validator.check_schema(schema) - return Draft202012Validator(schema) + return Draft202012Validator(schema, format_checker=Draft202012Validator.FORMAT_CHECKER) def validate_event_payload(payload: dict) -> list[str]: diff --git a/backend/tests/conftest.py b/backend/tests/conftest.py new file mode 100644 index 0000000..b5bc5ed --- /dev/null +++ b/backend/tests/conftest.py @@ -0,0 +1,82 @@ +"""SQLite in-memory fixtures for API tests.""" + +import os + +# Must be set before app.database imports create_engine +os.environ.setdefault("DATABASE_URL", "sqlite:///:memory:") +os.environ.setdefault("SAC_BOOTSTRAP_API_KEY", "sac_test_key_for_pytest_only") + +import pytest +from fastapi.testclient import TestClient +from sqlalchemy import JSON, create_engine, event +from sqlalchemy.dialects.postgresql import JSONB +from sqlalchemy.orm import sessionmaker +from sqlalchemy.pool import StaticPool + +from app.auth.api_key import hash_api_key +from app.database import Base, get_db +from app.main import app as fastapi_app +import app.models # noqa: F401 — register all tables on Base.metadata +from app.models import ApiKey + +TEST_API_KEY = os.environ["SAC_BOOTSTRAP_API_KEY"] + + +@event.listens_for(Base.metadata, "before_create") +def _sqlite_jsonb_as_json(metadata, connection, **_kwargs) -> None: + if connection.dialect.name != "sqlite": + return + for table in metadata.tables.values(): + for column in table.columns: + if isinstance(column.type, JSONB): + column.type = JSON() + + +@pytest.fixture +def db_engine(): + engine = create_engine( + "sqlite:///:memory:", + connect_args={"check_same_thread": False}, + poolclass=StaticPool, + ) + Base.metadata.create_all(engine) + yield engine + engine.dispose() + + +@pytest.fixture +def db_session(db_engine): + Session = sessionmaker(bind=db_engine, autocommit=False, autoflush=False) + session = Session() + session.add( + ApiKey( + name="test", + key_prefix=TEST_API_KEY[:12], + key_hash=hash_api_key(TEST_API_KEY), + is_active=True, + ) + ) + session.commit() + yield session + session.close() + + +@pytest.fixture +def client(db_session, monkeypatch): + monkeypatch.setattr("app.main.bootstrap_api_key", lambda: None) + + def override_get_db(): + try: + yield db_session + finally: + pass + + fastapi_app.dependency_overrides[get_db] = override_get_db + with TestClient(fastapi_app) as c: + yield c + fastapi_app.dependency_overrides.clear() + + +@pytest.fixture +def auth_headers(): + return {"Authorization": f"Bearer {TEST_API_KEY}"} diff --git a/backend/tests/test_ingest.py b/backend/tests/test_ingest.py new file mode 100644 index 0000000..d3e8ba2 --- /dev/null +++ b/backend/tests/test_ingest.py @@ -0,0 +1,63 @@ +"""Ingest idempotency and HTTP status contract.""" + +import uuid + +from app.services.schema_validate import validate_event_payload + +VALID_EVENT = { + "schema_version": "1.0", + "event_id": "550e8400-e29b-41d4-a716-446655440000", + "occurred_at": "2026-05-27T10:00:00+03:00", + "source": {"product": "ssh-monitor", "product_version": "1.2.3-SAC"}, + "host": {"hostname": "test-host", "os_family": "linux"}, + "category": "agent", + "type": "agent.test", + "severity": "info", + "title": "Test", + "summary": "pytest ingest", +} + + +def test_schema_rejects_non_uuid_event_id(): + bad = {**VALID_EVENT, "event_id": "not-a-uuid"} + errors = validate_event_payload(bad) + assert errors + + +def test_schema_rejects_missing_event_id(): + payload = {k: v for k, v in VALID_EVENT.items() if k != "event_id"} + errors = validate_event_payload(payload) + assert errors + + +def test_ingest_created_201(client, auth_headers): + event_id = str(uuid.uuid4()) + payload = {**VALID_EVENT, "event_id": event_id} + r = client.post("/api/v1/events", json=payload, headers=auth_headers) + assert r.status_code == 201 + data = r.json() + assert data["created"] is True + assert data["status"] == "created" + assert data["event_id"] == event_id + + +def test_ingest_duplicate_409(client, auth_headers): + event_id = str(uuid.uuid4()) + payload = {**VALID_EVENT, "event_id": event_id} + assert client.post("/api/v1/events", json=payload, headers=auth_headers).status_code == 201 + r = client.post("/api/v1/events", json=payload, headers=auth_headers) + assert r.status_code == 409 + data = r.json() + assert data["created"] is False + assert data["status"] == "duplicate" + assert data["event_id"] == event_id + + +def test_ingest_invalid_payload_422(client, auth_headers): + r = client.post( + "/api/v1/events", + json={"event_id": "bad", "summary": "x"}, + headers=auth_headers, + ) + assert r.status_code == 422 + assert "schema_errors" in r.json()["detail"] diff --git a/docs/INDEX.md b/docs/INDEX.md index 028b289..ae7218e 100644 --- a/docs/INDEX.md +++ b/docs/INDEX.md @@ -15,7 +15,7 @@ | [ssl-certificate.md](ssl-certificate.md) | Wildcard TLS: формат, scp, пути `/etc/ssl/sac/` | | [deployment.md](deployment.md) | Эксплуатация, backup, TLS, `sac-deploy.sh` | | [operations-prod-status.md](operations-prod-status.md) | **Статус prod** и следующие шаги | -| [workspace-three-repos.md](workspace-three-repos.md) | Cursor multi-root | +| [workspace-three-repos.md](workspace-three-repos.md) | Multi-root workspace (три репо) | ## Порядок чтения diff --git a/docs/TZ.md b/docs/TZ.md index be3522d..2c0cf4c 100644 --- a/docs/TZ.md +++ b/docs/TZ.md @@ -44,7 +44,7 @@ | 2 | `RDP-login-monitor` | Агент на Windows Server | | 3 | `security-alert-center` | Центральный сервер (этот проект) | -Совместная разработка в Cursor: multi-root workspace — см. [workspace-three-repos.md](workspace-three-repos.md). +Совместная разработка в multi-root workspace — см. [workspace-three-repos.md](workspace-three-repos.md). --- diff --git a/docs/agent-integration.md b/docs/agent-integration.md index 1475ab9..b79d369 100644 --- a/docs/agent-integration.md +++ b/docs/agent-integration.md @@ -44,7 +44,7 @@ $SacSpoolDir = "D:\Soft\Logs\sac-spool" - `UseSAC=off` — без изменений: хотя бы один канал `NOTIFY_CHAIN` (как сейчас). - `UseSAC≠off` — обязательны `SAC_URL`, `SAC_API_KEY`; HTTP `GET {base}/health` OK. -- `--check-sac` / `Test-SacConnection` — отправка `agent.test`, ожидание `202`. +- `--check-sac` / `Test-SacConnection` — отправка `agent.test`, ожидание **HTTP 201** (повтор с тем же `event_id` — **409**, тоже успех для spool). --- @@ -64,20 +64,31 @@ Idempotency-Key: 550e8400-e29b-41d4-a716-446655440000 ### 2.2. Ответ +| HTTP | Значение | `status` в теле | `created` | +|------|----------|-----------------|-----------| +| **201** | Событие записано впервые | `created` | `true` | +| **409** | Тот же `event_id` уже есть (идемпотентность) | `duplicate` | `false` | +| **422** | Ошибка JSON Schema | — | — | + +Пример **201**: + ```json { - "status": "accepted", + "status": "created", "event_id": "550e8400-e29b-41d4-a716-446655440000", - "sac_event_url": "/events/12345", + "created": true, + "sac_event_url": "https://sac.kalinamall.ru/api/v1/events/12345", "problem_id": null } ``` +`event_id` — обязательный **UUID** (RFC 4122), уникален глобально; в БД индекс `UNIQUE (event_id)`. + ### 2.3. Spool при ошибке 1. Записать JSON в `SAC_SPOOL_DIR/{event_id}.json`. 2. **ssh-monitor:** каждую итерацию цикла вызывается `sac_flush_spool` (до 20 файлов). -3. После успеха (HTTP 202) — удалить файл. +3. После успеха (**HTTP 201** или **409**) — удалить файл из spool. 4. Лимит размера spool (например 500 MB) — логировать WARN, не удалять без алерта. --- diff --git a/docs/diagrams/data-flow.md b/docs/diagrams/data-flow.md index dea585d..473c246 100644 --- a/docs/diagrams/data-flow.md +++ b/docs/diagrams/data-flow.md @@ -1,11 +1,11 @@ # Диаграмма потоков данных (рис. 1 к ТЗ) Используется в [TZ.md](../TZ.md), раздел 3. -В GitLab/GitHub и в VS Code/Cursor с поддержкой Mermaid диаграмма рендерится автоматически. +В GitLab/GitHub и в VS Code с поддержкой Mermaid диаграмма рендерится автоматически. ## Экспорт в PNG/SVG -1. Открыть этот файл в Cursor preview или на https://mermaid.live +1. Открыть этот файл в preview редактора или на https://mermaid.live 2. Экспортировать как изображение для презентаций ```mermaid diff --git a/docs/work-plan.md b/docs/work-plan.md index ce3fb57..ab2693e 100644 --- a/docs/work-plan.md +++ b/docs/work-plan.md @@ -81,7 +81,7 @@ ### День 1 -- [ ] `d1-1` Ingest: `event_id` UUID обязателен, UNIQUE индекс, коды duplicate, логи, docs/schema +- [x] `d1-1` Ingest: `event_id` UUID обязателен, UNIQUE индекс, коды duplicate, логи, docs/schema - [ ] `d1-2` MVP Problems: модель, корреляция `host+type+окно`, API `GET/ack/resolve` - [ ] `d1-3` Правила v1: `brute-force burst`, `privilege spike`, `host silence` @@ -106,9 +106,9 @@ #### День 1 -- [ ] `09:00–10:00` миграция `event_id` + UNIQUE, валидация ingest -- [ ] `10:00–11:00` поведение duplicate (`201/409`) и логирование reject -- [ ] `11:00–12:00` обновить `event-schema-v1.json` и `agent-integration.md` (в т.ч. `host.display_name` для агентов) +- [x] `09:00–10:00` миграция `event_id` + UNIQUE, валидация ingest +- [x] `10:00–11:00` поведение duplicate (`201/409`) и логирование reject +- [x] `11:00–12:00` обновить `event-schema-v1.json` и `agent-integration.md` (в т.ч. `host.display_name` для агентов) - [ ] `13:00–14:30` модель Problems + миграция - [ ] `14:30–16:00` корреляция при ingest (`fingerprint`, `count`, `last_seen`) - [ ] `16:00–17:30` API `GET /problems`, `ack`, `resolve` + smoke diff --git a/docs/workspace-three-repos.md b/docs/workspace-three-repos.md index 47efce0..9534a26 100644 --- a/docs/workspace-three-repos.md +++ b/docs/workspace-three-repos.md @@ -1,6 +1,6 @@ # Multi-root workspace: три репозитория -Как работать в Cursor над **ssh-monitor**, **RDP-login-monitor** и **security-alert-center** одновременно. +Как работать в одном multi-root workspace над **ssh-monitor**, **RDP-login-monitor** и **security-alert-center** одновременно (VS Code или совместимый редактор). ---