From a4a224b284cf2ea96bb32c02301b5692df1eaab5 Mon Sep 17 00:00:00 2001 From: PTah Date: Thu, 28 May 2026 09:46:12 +1000 Subject: [PATCH] feat: Problems correlation fingerprint, count, last_seen (d1-2) - Migration 003; windowed host+type+rule correlation on ingest - GET /problems filters; GET /problems/{id} with event timeline; ack/resolve 409 guard - Tests and SAC_PROBLEM_CORRELATION_WINDOW_MINUTES config Co-authored-by: Cursor --- .../versions/003_problem_correlation.py | 49 ++++ backend/app/api/v1/problems.py | 273 +++++++++++------- backend/app/config.py | 3 + backend/app/models/problem.py | 5 +- backend/app/services/problems.py | 71 +++-- backend/tests/conftest.py | 7 + backend/tests/test_problems.py | 69 +++++ deploy/env.native.example | 3 + docs/work-plan.md | 8 +- frontend/src/api.ts | 3 + 10 files changed, 365 insertions(+), 126 deletions(-) create mode 100644 backend/alembic/versions/003_problem_correlation.py create mode 100644 backend/tests/test_problems.py diff --git a/backend/alembic/versions/003_problem_correlation.py b/backend/alembic/versions/003_problem_correlation.py new file mode 100644 index 0000000..8c9c02f --- /dev/null +++ b/backend/alembic/versions/003_problem_correlation.py @@ -0,0 +1,49 @@ +"""problem correlation fields + +Revision ID: 003 +Revises: 002 +""" + +from typing import Sequence, Union + +import sqlalchemy as sa +from alembic import op + +revision: str = "003" +down_revision: Union[str, None] = "002" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + op.add_column("problems", sa.Column("fingerprint", sa.String(length=128), nullable=True)) + op.add_column( + "problems", + sa.Column("event_count", sa.Integer(), nullable=False, server_default="1"), + ) + op.add_column("problems", sa.Column("last_seen_at", sa.DateTime(timezone=True), nullable=True)) + op.create_index("ix_problems_fingerprint", "problems", ["fingerprint"]) + op.create_index("ix_problems_last_seen_at", "problems", ["last_seen_at"]) + + op.execute( + """ + UPDATE problems p SET + fingerprint = 'h' || COALESCE(p.host_id::text, '0') || ':r' || COALESCE(p.rule_id, 'unknown'), + event_count = COALESCE( + (SELECT COUNT(*)::int FROM problem_events pe WHERE pe.problem_id = p.id), + 1 + ), + last_seen_at = COALESCE(p.updated_at, p.created_at) + """ + ) + + op.alter_column("problems", "fingerprint", nullable=False) + op.alter_column("problems", "last_seen_at", nullable=False) + + +def downgrade() -> None: + op.drop_index("ix_problems_last_seen_at", table_name="problems") + op.drop_index("ix_problems_fingerprint", table_name="problems") + op.drop_column("problems", "last_seen_at") + op.drop_column("problems", "event_count") + op.drop_column("problems", "fingerprint") diff --git a/backend/app/api/v1/problems.py b/backend/app/api/v1/problems.py index b85aa06..f4b56a9 100644 --- a/backend/app/api/v1/problems.py +++ b/backend/app/api/v1/problems.py @@ -1,99 +1,174 @@ -from datetime import datetime - -from fastapi import APIRouter, Depends, HTTPException, Query -from pydantic import BaseModel -from sqlalchemy import func, select -from sqlalchemy.orm import Session, joinedload - -from app.auth.jwt_auth import get_current_user -from app.database import get_db -from app.models import Problem - -router = APIRouter(prefix="/problems", tags=["problems"]) - - -class ProblemSummary(BaseModel): - id: int - host_id: int | None - hostname: str | None - title: str - summary: str - severity: str - status: str - rule_id: str | None - created_at: datetime - updated_at: datetime - - model_config = {"from_attributes": True} - - -class ProblemListResponse(BaseModel): - items: list[ProblemSummary] - total: int - page: int - page_size: int - - -@router.get("", response_model=ProblemListResponse) -def list_problems( - page: int = Query(1, ge=1), - page_size: int = Query(50, ge=1, le=200), - status: str | None = Query(None), - db: Session = Depends(get_db), - _user: str = Depends(get_current_user), -) -> ProblemListResponse: - stmt = select(Problem).options(joinedload(Problem.host)) - count_stmt = select(func.count()).select_from(Problem) - if status: - stmt = stmt.where(Problem.status == status) - count_stmt = count_stmt.where(Problem.status == status) - - total = db.scalar(count_stmt) or 0 - rows = db.scalars( - stmt.order_by(Problem.updated_at.desc()).offset((page - 1) * page_size).limit(page_size) - ).all() - - items = [ - ProblemSummary( - id=p.id, - host_id=p.host_id, - hostname=p.host.hostname if p.host else None, - title=p.title, - summary=p.summary, - severity=p.severity, - status=p.status, - rule_id=p.rule_id, - created_at=p.created_at, - updated_at=p.updated_at, - ) - for p in rows - ] - return ProblemListResponse(items=items, total=total, page=page, page_size=page_size) - - -@router.post("/{problem_id}/ack") -def ack_problem( - problem_id: int, - db: Session = Depends(get_db), - _user: str = Depends(get_current_user), -) -> dict: - problem = db.get(Problem, problem_id) - if problem is None: - raise HTTPException(status_code=404, detail="Problem not found") - problem.status = "acknowledged" - db.commit() - return {"id": problem.id, "status": problem.status} - - -@router.post("/{problem_id}/resolve") -def resolve_problem( - problem_id: int, - db: Session = Depends(get_db), - _user: str = Depends(get_current_user), -) -> dict: - problem = db.get(Problem, problem_id) - if problem is None: - raise HTTPException(status_code=404, detail="Problem not found") - problem.status = "resolved" - db.commit() - return {"id": problem.id, "status": problem.status} +from datetime import datetime + +from fastapi import APIRouter, Depends, HTTPException, Query +from pydantic import BaseModel +from sqlalchemy import func, select +from sqlalchemy.orm import Session, joinedload + +from app.auth.jwt_auth import get_current_user +from app.database import get_db +from app.models import Event, Host, Problem, ProblemEvent + +router = APIRouter(prefix="/problems", tags=["problems"]) + + +class ProblemSummary(BaseModel): + id: int + host_id: int | None + hostname: str | None + title: str + summary: str + severity: str + status: str + rule_id: str | None + fingerprint: str + event_count: int + last_seen_at: datetime + created_at: datetime + updated_at: datetime + + +class ProblemListResponse(BaseModel): + items: list[ProblemSummary] + total: int + page: int + page_size: int + + +class ProblemEventItem(BaseModel): + id: int + event_id: str + occurred_at: datetime + type: str + severity: str + title: str + summary: str + + +class ProblemDetail(ProblemSummary): + events: list[ProblemEventItem] + + +def _problem_summary(p: Problem) -> ProblemSummary: + return ProblemSummary( + id=p.id, + host_id=p.host_id, + hostname=p.host.hostname if p.host else None, + title=p.title, + summary=p.summary, + severity=p.severity, + status=p.status, + rule_id=p.rule_id, + fingerprint=p.fingerprint, + event_count=p.event_count, + last_seen_at=p.last_seen_at, + created_at=p.created_at, + updated_at=p.updated_at, + ) + + +@router.get("", response_model=ProblemListResponse) +def list_problems( + page: int = Query(1, ge=1), + page_size: int = Query(50, ge=1, le=200), + status: str | None = Query(None), + severity: str | None = Query(None), + host_id: int | None = Query(None), + hostname: str | None = Query(None), + db: Session = Depends(get_db), + _user: str = Depends(get_current_user), +) -> ProblemListResponse: + stmt = select(Problem).join(Host, isouter=True).options(joinedload(Problem.host)) + count_stmt = select(func.count()).select_from(Problem).join(Host, isouter=True) + + if status: + stmt = stmt.where(Problem.status == status) + count_stmt = count_stmt.where(Problem.status == status) + if severity: + stmt = stmt.where(Problem.severity == severity) + count_stmt = count_stmt.where(Problem.severity == severity) + if host_id is not None: + stmt = stmt.where(Problem.host_id == host_id) + count_stmt = count_stmt.where(Problem.host_id == host_id) + if hostname: + like = f"%{hostname}%" + stmt = stmt.where(Host.hostname.ilike(like) | Host.display_name.ilike(like)) + count_stmt = count_stmt.where(Host.hostname.ilike(like) | Host.display_name.ilike(like)) + + total = db.scalar(count_stmt) or 0 + rows = db.scalars( + stmt.order_by(Problem.last_seen_at.desc()).offset((page - 1) * page_size).limit(page_size) + ).all() + + items = [_problem_summary(p) for p in rows] + return ProblemListResponse(items=items, total=total, page=page, page_size=page_size) + + +@router.get("/{problem_id}", response_model=ProblemDetail) +def get_problem( + problem_id: int, + db: Session = Depends(get_db), + _user: str = Depends(get_current_user), +) -> ProblemDetail: + problem = db.scalar( + select(Problem).where(Problem.id == problem_id).options(joinedload(Problem.host)) + ) + if problem is None: + raise HTTPException(status_code=404, detail="Problem not found") + + event_rows = db.scalars( + select(Event) + .join(ProblemEvent, ProblemEvent.event_id == Event.id) + .where(ProblemEvent.problem_id == problem_id) + .order_by(Event.occurred_at.desc()) + ).all() + + base = _problem_summary(problem) + return ProblemDetail( + **base.model_dump(), + events=[ + ProblemEventItem( + id=e.id, + event_id=e.event_id, + occurred_at=e.occurred_at, + type=e.type, + severity=e.severity, + title=e.title, + summary=e.summary, + ) + for e in event_rows + ], + ) + + +@router.post("/{problem_id}/ack") +def ack_problem( + problem_id: int, + db: Session = Depends(get_db), + _user: str = Depends(get_current_user), +) -> dict: + problem = db.get(Problem, problem_id) + if problem is None: + raise HTTPException(status_code=404, detail="Problem not found") + if problem.status == "resolved": + raise HTTPException(status_code=409, detail="Problem already resolved") + problem.status = "acknowledged" + db.commit() + return {"id": problem.id, "status": problem.status} + + +@router.post("/{problem_id}/resolve") +def resolve_problem( + problem_id: int, + db: Session = Depends(get_db), + _user: str = Depends(get_current_user), +) -> dict: + problem = db.get(Problem, problem_id) + if problem is None: + raise HTTPException(status_code=404, detail="Problem not found") + if problem.status == "resolved": + raise HTTPException(status_code=409, detail="Problem already resolved") + problem.status = "resolved" + db.commit() + return {"id": problem.id, "status": problem.status} + \ No newline at end of file diff --git a/backend/app/config.py b/backend/app/config.py index e679591..6ce67ea 100644 --- a/backend/app/config.py +++ b/backend/app/config.py @@ -53,6 +53,9 @@ class Settings(BaseSettings): # Порог «живости» агента: agent.heartbeat раз в ~12 ч (ssh-monitor) sac_heartbeat_stale_minutes: int = 780 + # Окно корреляции Problems: host + type + rule в одном open Problem + sac_problem_correlation_window_minutes: int = 60 + @lru_cache def get_settings() -> Settings: diff --git a/backend/app/models/problem.py b/backend/app/models/problem.py index 3c09a29..5a63d11 100644 --- a/backend/app/models/problem.py +++ b/backend/app/models/problem.py @@ -1,6 +1,6 @@ from datetime import datetime -from sqlalchemy import DateTime, ForeignKey, String, Text, func +from sqlalchemy import DateTime, ForeignKey, Integer, String, Text, func from sqlalchemy.orm import Mapped, mapped_column, relationship from app.database import Base @@ -16,6 +16,9 @@ class Problem(Base): severity: Mapped[str] = mapped_column(String(16), index=True) status: Mapped[str] = mapped_column(String(32), index=True, default="open") rule_id: Mapped[str | None] = mapped_column(String(64)) + fingerprint: Mapped[str] = mapped_column(String(128), index=True) + event_count: Mapped[int] = mapped_column(Integer, default=1) + last_seen_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), index=True) created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now()) updated_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), server_default=func.now(), onupdate=func.now() diff --git a/backend/app/services/problems.py b/backend/app/services/problems.py index aed1914..fe29885 100644 --- a/backend/app/services/problems.py +++ b/backend/app/services/problems.py @@ -1,11 +1,13 @@ -"""Auto-create Problems from ingested events (MVP rules).""" +"""Auto-create Problems from ingested events (MVP rules + correlation).""" + +from datetime import datetime, timedelta, timezone from sqlalchemy import select from sqlalchemy.orm import Session +from app.config import get_settings from app.models import Event, Problem, ProblemEvent -# event types that always open a problem PROBLEM_TYPES = frozenset( { "ssh.ip.banned", @@ -16,41 +18,63 @@ PROBLEM_TYPES = frozenset( HIGH_SEVERITIES = frozenset({"high", "critical"}) -def maybe_create_problem(db: Session, event: Event) -> tuple[Problem | None, bool]: - if event.severity not in HIGH_SEVERITIES and event.type not in PROBLEM_TYPES: - return None, False +def problem_rule_id(event: Event) -> str | None: + if event.severity in HIGH_SEVERITIES: + return "high_severity" + if event.type in PROBLEM_TYPES: + return f"type:{event.type}" + return None - rule_id = "high_severity" if event.severity in HIGH_SEVERITIES else f"type:{event.type}" - existing = db.scalar( - select(Problem) - .join(ProblemEvent) - .where( - Problem.status == "open", - Problem.rule_id == rule_id, - Problem.host_id == event.host_id, +def problem_fingerprint(host_id: int, event_type: str, rule_id: str) -> str: + return f"h{host_id}:t{event_type}:r{rule_id}" + + +def _correlation_cutoff(now: datetime) -> datetime: + minutes = get_settings().sac_problem_correlation_window_minutes + return now - timedelta(minutes=minutes) + + +def _append_event(db: Session, problem: Problem, event: Event) -> None: + linked = db.scalar( + select(ProblemEvent).where( + ProblemEvent.problem_id == problem.id, ProblemEvent.event_id == event.id, ) - .limit(1) ) - if existing: - return existing, False + if linked is not None: + return + db.add(ProblemEvent(problem_id=problem.id, event_id=event.id)) + problem.event_count = (problem.event_count or 0) + 1 + problem.last_seen_at = event.occurred_at + problem.updated_at = datetime.now(timezone.utc) + if event.severity in HIGH_SEVERITIES and problem.severity not in HIGH_SEVERITIES: + problem.severity = event.severity + db.flush() + + +def maybe_create_problem(db: Session, event: Event) -> tuple[Problem | None, bool]: + rule_id = problem_rule_id(event) + if rule_id is None: + return None, False + + fingerprint = problem_fingerprint(event.host_id, event.type, rule_id) + now = datetime.now(timezone.utc) + cutoff = _correlation_cutoff(now) - # dedupe: one open problem per host+rule (append event link) open_problem = db.scalar( select(Problem) .where( Problem.status == "open", - Problem.rule_id == rule_id, + Problem.fingerprint == fingerprint, Problem.host_id == event.host_id, + Problem.last_seen_at >= cutoff, ) - .order_by(Problem.created_at.desc()) + .order_by(Problem.last_seen_at.desc()) .limit(1) ) if open_problem: - db.add(ProblemEvent(problem_id=open_problem.id, event_id=event.id)) - open_problem.updated_at = event.received_at - db.flush() + _append_event(db, open_problem, event) return open_problem, False problem = Problem( @@ -60,6 +84,9 @@ def maybe_create_problem(db: Session, event: Event) -> tuple[Problem | None, boo severity=event.severity, status="open", rule_id=rule_id, + fingerprint=fingerprint, + event_count=1, + last_seen_at=event.occurred_at, ) db.add(problem) db.flush() diff --git a/backend/tests/conftest.py b/backend/tests/conftest.py index b5bc5ed..ee4ec14 100644 --- a/backend/tests/conftest.py +++ b/backend/tests/conftest.py @@ -14,6 +14,7 @@ from sqlalchemy.orm import sessionmaker from sqlalchemy.pool import StaticPool from app.auth.api_key import hash_api_key +from app.auth.jwt_auth import create_access_token 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 @@ -80,3 +81,9 @@ def client(db_session, monkeypatch): @pytest.fixture def auth_headers(): return {"Authorization": f"Bearer {TEST_API_KEY}"} + + +@pytest.fixture +def jwt_headers(): + token = create_access_token("test-admin") + return {"Authorization": f"Bearer {token}"} diff --git a/backend/tests/test_problems.py b/backend/tests/test_problems.py new file mode 100644 index 0000000..73d34d6 --- /dev/null +++ b/backend/tests/test_problems.py @@ -0,0 +1,69 @@ +"""Problems correlation and API smoke tests.""" + +import uuid +from datetime import datetime, timezone + +from tests.test_ingest import VALID_EVENT + + +def _event_payload(**overrides): + now = datetime.now(timezone.utc).astimezone().isoformat(timespec="seconds") + base = { + **VALID_EVENT, + "event_id": str(uuid.uuid4()), + "occurred_at": now, + "severity": "high", + "type": "ssh.ip.banned", + "title": "Ban test", + "summary": "test problem correlation", + } + base.update(overrides) + return base + + +def test_problem_correlates_two_events(client, auth_headers, jwt_headers): + e1 = _event_payload() + e2 = _event_payload(title="Ban test 2") + assert client.post("/api/v1/events", json=e1, headers=auth_headers).status_code == 201 + assert client.post("/api/v1/events", json=e2, headers=auth_headers).status_code == 201 + + r = client.get("/api/v1/problems?status=open", headers=jwt_headers) + assert r.status_code == 200 + data = r.json() + assert data["total"] == 1 + problem = data["items"][0] + assert problem["event_count"] == 2 + assert problem["fingerprint"].startswith("h") + assert ":tssh.ip.banned:" in problem["fingerprint"] + + detail = client.get(f"/api/v1/problems/{problem['id']}", headers=jwt_headers) + assert detail.status_code == 200 + assert len(detail.json()["events"]) == 2 + + +def test_problem_ack_and_resolve(client, auth_headers, jwt_headers): + payload = _event_payload() + client.post("/api/v1/events", json=payload, headers=auth_headers) + pid = client.get("/api/v1/problems?status=open", headers=jwt_headers).json()["items"][0]["id"] + + ack = client.post(f"/api/v1/problems/{pid}/ack", headers=jwt_headers) + assert ack.status_code == 200 + assert ack.json()["status"] == "acknowledged" + + res = client.post(f"/api/v1/problems/{pid}/resolve", headers=jwt_headers) + assert res.status_code == 200 + assert res.json()["status"] == "resolved" + + +def test_new_problem_after_resolve(client, auth_headers, jwt_headers): + p1 = _event_payload() + client.post("/api/v1/events", json=p1, headers=auth_headers) + pid = client.get("/api/v1/problems?status=open", headers=jwt_headers).json()["items"][0]["id"] + client.post(f"/api/v1/problems/{pid}/resolve", headers=jwt_headers) + + p2 = _event_payload(summary="after resolve") + client.post("/api/v1/events", json=p2, headers=auth_headers) + open_items = client.get("/api/v1/problems?status=open", headers=jwt_headers).json()["items"] + assert len(open_items) == 1 + assert open_items[0]["event_count"] == 1 + assert open_items[0]["id"] != pid diff --git a/deploy/env.native.example b/deploy/env.native.example index ded4f16..1af07e0 100644 --- a/deploy/env.native.example +++ b/deploy/env.native.example @@ -26,4 +26,7 @@ TELEGRAM_MIN_SEVERITY=high # Статус хоста по agent.heartbeat (минуты; ssh-monitor шлёт ~раз в 12 ч) SAC_HEARTBEAT_STALE_MINUTES=780 +# Корреляция Problems: host + type + rule в одном open Problem (минуты) +SAC_PROBLEM_CORRELATION_WINDOW_MINUTES=60 + CORS_ORIGINS=* diff --git a/docs/work-plan.md b/docs/work-plan.md index ab2693e..221acb5 100644 --- a/docs/work-plan.md +++ b/docs/work-plan.md @@ -82,7 +82,7 @@ ### День 1 - [x] `d1-1` Ingest: `event_id` UUID обязателен, UNIQUE индекс, коды duplicate, логи, docs/schema -- [ ] `d1-2` MVP Problems: модель, корреляция `host+type+окно`, API `GET/ack/resolve` +- [x] `d1-2` MVP Problems: модель, корреляция `host+type+окно`, API `GET/ack/resolve` - [ ] `d1-3` Правила v1: `brute-force burst`, `privilege spike`, `host silence` ### День 2 @@ -109,9 +109,9 @@ - [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 +- [x] `13:00–14:30` модель Problems + миграция +- [x] `14:30–16:00` корреляция при ingest (`fingerprint`, `count`, `last_seen`) +- [x] `16:00–17:30` API `GET /problems`, `ack`, `resolve` + smoke - [ ] `18:00–19:00` правила `brute/privilege/silence` + unit tests #### День 2 diff --git a/frontend/src/api.ts b/frontend/src/api.ts index 5a2d6f4..eb60443 100644 --- a/frontend/src/api.ts +++ b/frontend/src/api.ts @@ -97,6 +97,9 @@ export interface ProblemSummary { severity: string; status: string; rule_id: string | null; + fingerprint?: string; + event_count?: number; + last_seen_at?: string; created_at: string; updated_at: string; }