From 6c43b8637e04dc2afa18cc82c7a10e56c2f1b4db Mon Sep 17 00:00:00 2001 From: PTah Date: Fri, 3 Jul 2026 09:05:28 +1000 Subject: [PATCH] fix: dedupe host_silence problems under concurrent scans (0.4.11) Treat acknowledged as active, dedupe by hostname, add PostgreSQL advisory lock and a unique index. Migration closes existing duplicate open problems. Co-authored-by: Cursor --- README.md | 2 +- README_en.md | 2 +- .../versions/026_host_silence_dedupe.py | 74 ++++++++ backend/app/services/host_silence_scan.py | 162 +++++++++++++++--- backend/app/services/problems.py | 2 +- backend/app/version.py | 2 +- backend/tests/test_host_silence_scan.py | 117 +++++++++++++ deploy/env.native.example | 2 + frontend/src/version.ts | 2 +- 9 files changed, 334 insertions(+), 31 deletions(-) create mode 100644 backend/alembic/versions/026_host_silence_dedupe.py diff --git a/README.md b/README.md index a5512ad..600f0a5 100644 --- a/README.md +++ b/README.md @@ -13,7 +13,7 @@ | **security-alert-center** | Сервер SAC (Ubuntu 24.04) | | [seaca](https://git.kalinamall.ru/PapaTramp/seaca) | Android-клиент | -**Версия:** `0.4.10` · **Деплой:** `sudo /opt/sac-deploy.sh` +**Версия:** `0.4.11` · **Деплой:** `sudo /opt/sac-deploy.sh` ## Возможности diff --git a/README_en.md b/README_en.md index 2136141..9c9b337 100644 --- a/README_en.md +++ b/README_en.md @@ -13,7 +13,7 @@ Self-hosted hub for security events from Linux and Windows agents: ingest, corre | **security-alert-center** | SAC server (Ubuntu 24.04) | | [seaca](https://git.kalinamall.ru/PapaTramp/seaca) | Android client | -**Version:** `0.4.10` · **Deploy:** `sudo /opt/sac-deploy.sh` +**Version:** `0.4.11` · **Deploy:** `sudo /opt/sac-deploy.sh` ## Features diff --git a/backend/alembic/versions/026_host_silence_dedupe.py b/backend/alembic/versions/026_host_silence_dedupe.py new file mode 100644 index 0000000..e965c74 --- /dev/null +++ b/backend/alembic/versions/026_host_silence_dedupe.py @@ -0,0 +1,74 @@ +"""dedupe host_silence problems + unique active index + +Revision ID: 026_host_silence_dedupe +Revises: 025_auto_rdp_flap_disconnect +Create Date: 2026-07-03 + +""" + +from alembic import op + +revision = "026_host_silence_dedupe" +down_revision = "025_auto_rdp_flap_disconnect" +branch_labels = None +depends_on = None + + +def _dedupe_host_silence_problems() -> None: + op.execute( + """ + WITH ranked AS ( + SELECT + p.id, + ROW_NUMBER() OVER ( + PARTITION BY p.host_id + ORDER BY p.last_seen_at DESC, p.id DESC + ) AS rn + FROM problems p + WHERE p.rule_id = 'rule:host_silence' + AND p.status IN ('open', 'acknowledged') + ) + UPDATE problems + SET status = 'resolved', + resolved_by = 'auto', + updated_at = NOW() + WHERE id IN (SELECT id FROM ranked WHERE rn > 1) + """ + ) + op.execute( + """ + WITH ranked AS ( + SELECT + p.id, + ROW_NUMBER() OVER ( + PARTITION BY lower(h.hostname) + ORDER BY p.last_seen_at DESC, p.id DESC + ) AS rn + FROM problems p + JOIN hosts h ON h.id = p.host_id + WHERE p.rule_id = 'rule:host_silence' + AND p.status IN ('open', 'acknowledged') + ) + UPDATE problems + SET status = 'resolved', + resolved_by = 'auto', + updated_at = NOW() + WHERE id IN (SELECT id FROM ranked WHERE rn > 1) + """ + ) + + +def upgrade() -> None: + _dedupe_host_silence_problems() + op.execute( + """ + CREATE UNIQUE INDEX IF NOT EXISTS uq_problems_host_silence_active + ON problems (host_id) + WHERE rule_id = 'rule:host_silence' + AND status IN ('open', 'acknowledged') + """ + ) + + +def downgrade() -> None: + op.execute("DROP INDEX IF EXISTS uq_problems_host_silence_active") diff --git a/backend/app/services/host_silence_scan.py b/backend/app/services/host_silence_scan.py index 2d9adb1..920e02b 100644 --- a/backend/app/services/host_silence_scan.py +++ b/backend/app/services/host_silence_scan.py @@ -6,15 +6,19 @@ import logging from dataclasses import dataclass from datetime import datetime, timedelta, timezone -from sqlalchemy import select +from sqlalchemy import func, select, text +from sqlalchemy.exc import IntegrityError from sqlalchemy.orm import Session from app.config import get_settings from app.models import Host, Problem -from app.services.problem_rules import RuleMatch, build_fingerprint, host_silence_match_for_host +from app.services.problem_rules import RULE_HOST_SILENCE, RuleMatch, build_fingerprint, host_silence_match_for_host logger = logging.getLogger(__name__) +ACTIVE_HOST_SILENCE_STATUSES = ("open", "acknowledged") +_HOST_SILENCE_SCAN_LOCK_KEY = 915_001_001 + @dataclass(frozen=True) class HostSilenceScanResult: @@ -24,19 +28,92 @@ class HostSilenceScanResult: created: bool +def _host_silence_scan_lock(db: Session) -> bool: + """Один scan за раз (uvicorn workers + systemd timer).""" + bind = db.get_bind() + if bind.dialect.name != "postgresql": + return True + acquired = db.execute( + text("SELECT pg_try_advisory_xact_lock(:key)"), + {"key": _HOST_SILENCE_SCAN_LOCK_KEY}, + ).scalar() + return bool(acquired) + + +def _find_active_host_silence_problem( + db: Session, + *, + host_id: int, + hostname: str | None, +) -> Problem | None: + active = db.scalar( + select(Problem) + .where( + Problem.host_id == host_id, + Problem.rule_id == RULE_HOST_SILENCE, + Problem.status.in_(ACTIVE_HOST_SILENCE_STATUSES), + ) + .order_by(Problem.last_seen_at.desc()) + .limit(1) + ) + if active is not None: + return active + + name = (hostname or "").strip() + if not name: + return None + + sibling = db.scalar( + select(Problem) + .join(Host, Problem.host_id == Host.id) + .where( + Problem.rule_id == RULE_HOST_SILENCE, + Problem.status.in_(ACTIVE_HOST_SILENCE_STATUSES), + func.lower(Host.hostname) == name.lower(), + ) + .order_by(Problem.last_seen_at.desc()) + .limit(1) + ) + if sibling is not None and sibling.host_id != host_id: + sibling.host_id = host_id + db.flush() + return sibling + + +def _apply_host_silence_refresh( + problem: Problem, + *, + host_id: int, + match: RuleMatch, + fingerprint: str, + ref: datetime, +) -> None: + problem.host_id = host_id + problem.summary = match.summary + problem.title = match.title + problem.fingerprint = fingerprint + problem.last_seen_at = ref + problem.updated_at = ref + + def open_or_refresh_host_silence_problem( db: Session, host_id: int, match: RuleMatch, *, now: datetime | None = None, + hostname: str | None = None, ) -> tuple[Problem | None, bool]: """ - Открыть или обновить open Problem host_silence без ingest-события. + Открыть или обновить open/ack Problem host_silence без ingest-события. Returns (problem, created). problem is None when suppressed by manual-resolve cooldown. """ ref = now or datetime.now(timezone.utc) + if hostname is None: + host = db.get(Host, host_id) + hostname = host.hostname if host is not None else None + fingerprint = build_fingerprint( host_id, match.correlation_type, @@ -44,24 +121,17 @@ def open_or_refresh_host_silence_problem( match.fingerprint_suffix, ) - open_problem = db.scalar( - select(Problem) - .where( - Problem.status == "open", - Problem.rule_id == match.rule_id, - Problem.host_id == host_id, + active_problem = _find_active_host_silence_problem(db, host_id=host_id, hostname=hostname) + if active_problem is not None: + _apply_host_silence_refresh( + active_problem, + host_id=host_id, + match=match, + fingerprint=fingerprint, + ref=ref, ) - .order_by(Problem.last_seen_at.desc()) - .limit(1) - ) - if open_problem: - open_problem.summary = match.summary - open_problem.title = match.title - open_problem.fingerprint = fingerprint - open_problem.last_seen_at = ref - open_problem.updated_at = ref db.flush() - return open_problem, False + return active_problem, False settings = get_settings() cooldown_hours = settings.sac_host_silence_manual_resolve_cooldown_hours @@ -103,12 +173,27 @@ def open_or_refresh_host_silence_problem( .limit(1) ) if manual_expired is not None: + active_problem = _find_active_host_silence_problem(db, host_id=host_id, hostname=hostname) + if active_problem is not None: + _apply_host_silence_refresh( + active_problem, + host_id=host_id, + match=match, + fingerprint=fingerprint, + ref=ref, + ) + db.flush() + return active_problem, False + manual_expired.status = "open" manual_expired.resolved_by = None - manual_expired.summary = match.summary - manual_expired.title = match.title - manual_expired.last_seen_at = ref - manual_expired.updated_at = ref + _apply_host_silence_refresh( + manual_expired, + host_id=host_id, + match=match, + fingerprint=fingerprint, + ref=ref, + ) db.flush() return manual_expired, True @@ -123,13 +208,32 @@ def open_or_refresh_host_silence_problem( event_count=0, last_seen_at=ref, ) - db.add(problem) - db.flush() + try: + with db.begin_nested(): + db.add(problem) + db.flush() + except IntegrityError: + active_problem = _find_active_host_silence_problem(db, host_id=host_id, hostname=hostname) + if active_problem is None: + raise + _apply_host_silence_refresh( + active_problem, + host_id=host_id, + match=match, + fingerprint=fingerprint, + ref=ref, + ) + db.flush() + return active_problem, False return problem, True def run_host_silence_scan(db: Session, *, now: datetime | None = None) -> list[HostSilenceScanResult]: """Найти stale-хосты и создать/обновить Problems rule:host_silence.""" + if not _host_silence_scan_lock(db): + logger.debug("host_silence scan skipped (another runner holds advisory lock)") + return [] + ref = now or datetime.now(timezone.utc) hosts = db.scalars(select(Host).order_by(Host.id)).all() results: list[HostSilenceScanResult] = [] @@ -138,7 +242,13 @@ def run_host_silence_scan(db: Session, *, now: datetime | None = None) -> list[H match = host_silence_match_for_host(db, host.id, hostname=host.hostname, now=ref) if match is None: continue - problem, created = open_or_refresh_host_silence_problem(db, host.id, match, now=ref) + problem, created = open_or_refresh_host_silence_problem( + db, + host.id, + match, + now=ref, + hostname=host.hostname, + ) if problem is None: continue results.append( diff --git a/backend/app/services/problems.py b/backend/app/services/problems.py index 838fc28..c4b2ccd 100644 --- a/backend/app/services/problems.py +++ b/backend/app/services/problems.py @@ -51,7 +51,7 @@ def open_or_append_problem( if ref.tzinfo is None: ref = ref.replace(tzinfo=timezone.utc) problem, created = open_or_refresh_host_silence_problem( - db, event.host_id, match, now=ref + db, event.host_id, match, now=ref, hostname=event.host.hostname if event.host else None ) if problem is None: return None, False diff --git a/backend/app/version.py b/backend/app/version.py index 99b3444..664d6fd 100644 --- a/backend/app/version.py +++ b/backend/app/version.py @@ -1,5 +1,5 @@ """Единый источник версии SAC (API, health, логи, OpenAPI).""" APP_NAME = "Security Alert Center" -APP_VERSION = "0.4.10" +APP_VERSION = "0.4.11" APP_VERSION_LABEL = f"{APP_NAME} v.{APP_VERSION}" diff --git a/backend/tests/test_host_silence_scan.py b/backend/tests/test_host_silence_scan.py index 1f12782..3c8a3be 100644 --- a/backend/tests/test_host_silence_scan.py +++ b/backend/tests/test_host_silence_scan.py @@ -211,6 +211,123 @@ def test_scan_reopens_after_manual_cooldown_expires(db_session, scan_settings, m assert later[0].problem.resolved_by is None +def test_scan_does_not_duplicate_acknowledged_problem(db_session, scan_settings): + hb = _ingest( + db_session, + type="agent.heartbeat", + category="agent", + severity="info", + title="hb", + summary="heartbeat", + ) + hb.received_at = datetime.now(timezone.utc) - timedelta(hours=2) + db_session.flush() + + now = datetime.now(timezone.utc) + first = run_host_silence_scan(db_session, now=now) + assert first[0].created is True + first[0].problem.status = "acknowledged" + db_session.flush() + + second = run_host_silence_scan(db_session, now=now + timedelta(minutes=5)) + assert len(second) == 1 + assert second[0].created is False + assert second[0].problem.id == first[0].problem.id + + problems = db_session.scalars( + select(Problem).where( + Problem.host_id == hb.host_id, + Problem.rule_id == RULE_HOST_SILENCE, + Problem.status.in_(("open", "acknowledged")), + ) + ).all() + assert len(problems) == 1 + + +def test_scan_dedupes_by_hostname_across_duplicate_hosts(db_session, scan_settings): + from app.models import Host + + now = datetime.now(timezone.utc) + host_a = Host( + hostname="COMM-PC", + os_family="windows", + product="rdp-login-monitor", + ipv4="192.168.162.33", + ) + host_b = Host( + hostname="COMM-PC", + os_family="windows", + product="rdp-login-monitor", + ipv4="192.168.162.33", + ) + db_session.add_all([host_a, host_b]) + db_session.flush() + + hb = _ingest( + db_session, + host={"hostname": host_a.hostname, "os_family": "windows", "ipv4": host_a.ipv4}, + type="agent.heartbeat", + category="agent", + severity="info", + title="hb", + summary="heartbeat", + ) + hb.host_id = host_a.id + hb.received_at = now - timedelta(hours=2) + db_session.flush() + + first = run_host_silence_scan(db_session, now=now) + assert len(first) == 1 + assert first[0].created is True + + hb_b = _ingest( + db_session, + event_id=str(uuid.uuid4()), + host={"hostname": host_b.hostname, "os_family": "windows", "ipv4": host_b.ipv4}, + source={"product": "rdp-login-monitor", "product_version": "2.0.0-SAC"}, + type="agent.heartbeat", + category="agent", + severity="info", + title="hb2", + summary="heartbeat", + ) + hb_b.host_id = host_b.id + hb_b.received_at = now - timedelta(hours=2) + db_session.flush() + + second = run_host_silence_scan(db_session, now=now + timedelta(minutes=5)) + assert len(second) == 2 + assert all(not item.created for item in second) + assert {item.problem.id for item in second} == {first[0].problem.id} + + active = db_session.scalars( + select(Problem).where( + Problem.rule_id == RULE_HOST_SILENCE, + Problem.status.in_(("open", "acknowledged")), + ) + ).all() + assert len(active) == 1 + + +def test_scan_skipped_when_advisory_lock_not_acquired(db_session, scan_settings, monkeypatch): + monkeypatch.setattr( + "app.services.host_silence_scan._host_silence_scan_lock", + lambda _db: False, + ) + hb = _ingest( + db_session, + type="agent.heartbeat", + category="agent", + severity="info", + title="hb", + summary="heartbeat", + ) + hb.received_at = datetime.now(timezone.utc) - timedelta(hours=2) + db_session.flush() + + assert run_host_silence_scan(db_session) == [] + + def test_scan_notifies_only_on_create(db_session, scan_settings): hb = _ingest( db_session, diff --git a/deploy/env.native.example b/deploy/env.native.example index f3d26e4..b066bca 100644 --- a/deploy/env.native.example +++ b/deploy/env.native.example @@ -65,6 +65,8 @@ SAC_DAILY_REPORT_REQUIRE_ACTIVITY=true SAC_HEARTBEAT_STALE_MINUTES=300 # Проактивный алерт host_silence (без ожидания нового события с хоста) +# Если включён systemd timer sac-host-silence-scan.timer — лучше SAC_HOST_SILENCE_SCAN_ENABLED=false +# (иначе scan идёт параллельно из каждого uvicorn worker + timer). SAC_HOST_SILENCE_SCAN_ENABLED=true SAC_HOST_SILENCE_SCAN_INTERVAL_MINUTES=5 diff --git a/frontend/src/version.ts b/frontend/src/version.ts index eaf5f65..a09c5d6 100644 --- a/frontend/src/version.ts +++ b/frontend/src/version.ts @@ -1,4 +1,4 @@ /** Fallback до загрузки /health; при релизе держите в sync с backend/app/version.py */ export const APP_NAME = "Security Alert Center"; -export const APP_VERSION = "0.4.10"; +export const APP_VERSION = "0.4.11"; export const APP_VERSION_LABEL = `${APP_NAME} v.${APP_VERSION}`;