fix(dlp): block self-test case creation

aw-server/aw-ru-patch.js

@@ -936,9 +936,10 @@
     const hideSuppressed = center.querySelector("[data-aw-ru-hide-suppressed]") && center.querySelector("[data-aw-ru-hide-suppressed]").checked;
     const rows = [];
     for (const event of state.events) {
+      const data = event.data || {};
+      if (String(data.signalType || "").toLowerCase() === "self_test") continue;
       const matchedRule = state.activeRules.find(function (rule) { return ruleMatchesEvent(rule, event); }) || null;
       if (hideSuppressed && matchedRule) continue;
-      const data = event.data || {};
       const eventKey = buildDlpKey(event);
       rows.push(
         '<tr class="aw-ru-dlp-row' + (matchedRule ? ' aw-ru-dlp-muted' : '') + '" data-aw-ru-dlp-key="' + escapeHtml(eventKey) + '">' +
@@ -1051,6 +1052,9 @@
 
   async function createCaseFromEvent(host, event, row) {
     const data = event.data || {};
+    if (String(data.signalType || "").toLowerCase() === "self_test") {
+      throw new Error("self_test не должен превращаться в кейс");
+    }
     const verdict = row.querySelector("[data-aw-ru-dlp-verdict]").value;
     const category = row.querySelector("[data-aw-ru-dlp-category]").value.trim();
     const comment = row.querySelector("[data-aw-ru-dlp-comment]").value.trim();

aw-server/dlp-case-management/case_rules.py

@@ -0,0 +1,8 @@
+#!/usr/bin/env python3
+from __future__ import annotations
+
+
+def is_self_test_case(incident_id: str | None, title: str | None) -> bool:
+    incident = str(incident_id or "").lower()
+    label = str(title or "").lower()
+    return "|self_test|" in incident or label.startswith("dlp self_test")

aw-server/dlp-case-management/case_service.py

@@ -8,6 +8,7 @@ from typing import Any
 from fastapi import FastAPI, HTTPException, Query
 from fastapi.middleware.cors import CORSMiddleware
 
+from case_rules import is_self_test_case
 from case_schema import CaseCommentCreate, CaseCreate, CaseHayabusaLink, CaseUpdate
 from case_storage import CaseStorage
 
@@ -30,6 +31,8 @@ def health() -> dict[str, Any]:
 
 @APP.post("/api/0/dlp/cases")
 def create_case(payload: CaseCreate) -> dict[str, Any]:
+    if is_self_test_case(payload.incident_id, payload.title):
+        raise HTTPException(status_code=422, detail="self_test cases are not allowed")
     return STORE.create_case(payload.model_dump(exclude_none=True), actor="api")
 
 

aw-server/dlp-case-management/case_storage.py

@@ -127,6 +127,17 @@ class CaseStorage:
             )
             evidence_digest = normalized_evidence.get("latest_sha256") or evidence_sha256(payload.get("evidence"))
         with self.conn() as c:
+            existing = c.execute(
+                """
+                SELECT * FROM cases
+                WHERE incident_id = ? AND COALESCE(host, '') = COALESCE(?, '')
+                ORDER BY id DESC
+                LIMIT 1
+                """,
+                (payload["incident_id"], payload.get("host")),
+            ).fetchone()
+            if existing:
+                return self._to_case_dict(existing)
             cur = c.execute(
                 """
                 INSERT INTO cases (

aw-server/dlp-case-management/test_case_rules.py

@@ -0,0 +1,21 @@
+#!/usr/bin/env python3
+from __future__ import annotations
+
+import unittest
+
+from case_rules import is_self_test_case
+
+
+class CaseRulesTest(unittest.TestCase):
+    def test_is_self_test_case_by_incident_id(self) -> None:
+        self.assertTrue(is_self_test_case("2026-05-14T17:00:52.186Z|self_test|Администратор|||", "Normal"))
+
+    def test_is_self_test_case_by_title(self) -> None:
+        self.assertTrue(is_self_test_case("inc-1", "DLP self_test · Администратор"))
+
+    def test_is_self_test_case_false_for_normal_case(self) -> None:
+        self.assertFalse(is_self_test_case("inc-1", "DLP print incident"))
+
+
+if __name__ == "__main__":
+    unittest.main()

aw-server/dlp-case-management/test_case_storage.py

@@ -46,6 +46,22 @@ class CaseStorageHayabusaLinkTest(unittest.TestCase):
             audit = storage.list_audit(int(created["id"]))
             self.assertTrue(any(row.get("action") == "link_hayabusa" for row in audit))
 
+    def test_create_case_deduplicates_by_incident_and_host(self) -> None:
+        with tempfile.TemporaryDirectory() as tmpdir:
+            db_path = Path(tmpdir) / "cases.db"
+            storage = CaseStorage(db_path)
+            payload = {
+                "incident_id": "2026-05-14T17:00:52.186Z|self_test|Администратор|||",
+                "host": "SHARKON2025",
+                "title": "DLP self_test · Администратор",
+                "severity": "medium",
+            }
+            created = storage.create_case(payload, actor="test")
+            duplicate = storage.create_case(payload, actor="test")
+            self.assertEqual(created["id"], duplicate["id"])
+            cases = storage.list_cases(host="SHARKON2025", limit=10)
+            self.assertEqual(len(cases), 1)
+
 
 if __name__ == "__main__":
     unittest.main()