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)
2026-05-28 09:13:12 +10:00
parent 5cf76f492c
commit 2b76ae8497
11 changed files with 210 additions and 26 deletions
@@ -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:
@@ -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)
+    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
@@ -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]:
@@ -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}"}
@@ -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"]
@@ -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 (три репо) |

 ## Порядок чтения

@@ -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).

 ---

@@ -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": "created",
-  "event_id": "550e8400-e29b-41d4-a716-446655440000",
+  "event_id": "550e8400-e29b-41d4-a716-446655440000",
  "created": true,
-  "problem_id": null
+  "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 201** или **409**) — удалить файл из spool.
-4. Лимит размера spool (например 500 MB) — логировать WARN, не удалять без алерта.
+4. Лимит размера spool (например 500 MB) — логировать WARN, не удалять без алерта.

 ---

@@ -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
@@ -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
@@ -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 или совместимый редактор).

 ---