test soc

README.md

@@ -62,6 +62,7 @@ Main sections:
 
 - Legacy integration APIs (`/wazuh/*`, `/shuffle/*`, `/action/*`)
 - MVP orchestration APIs (`/mvp/*`)
+- Wazuh-to-MVP sync API (`/wazuh/sync-to-mvp`)
 
 ### MVP endpoints
 
@@ -118,6 +119,19 @@ Scenarios:
 
 See `scripts/README.md` for details.
 
+Sync Wazuh alerts from indexer into MVP pipeline:
+
+```bash
+curl -X POST "http://localhost:8088/wazuh/sync-to-mvp?limit=50&minutes=120&q=*" \
+  -H 'X-Internal-API-Key: dev-internal-key'
+```
+
+Notes:
+
+- This sync reads from `wazuh-alerts-*` in Wazuh indexer.
+- Re-running sync is safe; dedupe is applied by `source + event_id`.
+- Your `send-wazuh-test-events.sh` traffic appears only after Wazuh rules generate alerts.
+
 ## Logs
 
 All logs (non-follow):

soc-integrator/.env

@@ -11,6 +11,9 @@ SOC_INTEGRATOR_DB_PASSWORD=soc_integrator_password
 WAZUH_BASE_URL=https://wazuh.manager:55000
 WAZUH_USERNAME=wazuh-wui
 WAZUH_PASSWORD=MyS3cr37P450r.*-
+WAZUH_INDEXER_URL=https://wazuh.indexer:9200
+WAZUH_INDEXER_USERNAME=admin
+WAZUH_INDEXER_PASSWORD=SecretPassword
 
 SHUFFLE_BASE_URL=http://shuffle-backend:5001
 SHUFFLE_API_KEY=95286d1a-1d02-4dd6-8de1-9832e326871f

soc-integrator/.env.example

@@ -11,6 +11,9 @@ SOC_INTEGRATOR_DB_PASSWORD=soc_integrator_password
 WAZUH_BASE_URL=https://wazuh.manager:55000
 WAZUH_USERNAME=wazuh-wui
 WAZUH_PASSWORD=MyS3cr37P450r.*-
+WAZUH_INDEXER_URL=https://wazuh.indexer:9200
+WAZUH_INDEXER_USERNAME=admin
+WAZUH_INDEXER_PASSWORD=SecretPassword
 
 SHUFFLE_BASE_URL=http://shuffle-backend:5001
 SHUFFLE_API_KEY=

soc-integrator/app/adapters/wazuh.py

@@ -4,10 +4,21 @@ import httpx
 
 
 class WazuhAdapter:
-    def __init__(self, base_url: str, username: str, password: str) -> None:
+    def __init__(
+        self,
+        base_url: str,
+        username: str,
+        password: str,
+        indexer_url: str | None = None,
+        indexer_username: str | None = None,
+        indexer_password: str | None = None,
+    ) -> None:
         self.base_url = base_url.rstrip("/")
         self.username = username
         self.password = password
+        self.indexer_url = (indexer_url or "").rstrip("/")
+        self.indexer_username = indexer_username
+        self.indexer_password = indexer_password
 
     async def _authenticate(self, client: httpx.AsyncClient) -> str:
         auth_url = f"{self.base_url}/security/user/authenticate?raw=true"
@@ -80,3 +91,41 @@ class WazuhAdapter:
         async with httpx.AsyncClient(verify=False, timeout=20.0) as client:
             token = await self._authenticate(client)
             return await self._get_with_bearer(client, token, "/manager/logs", params=params)
+
+    async def search_alerts(
+        self,
+        query: str,
+        limit: int = 50,
+        minutes: int = 120,
+    ) -> dict[str, Any]:
+        if not self.indexer_url:
+            raise RuntimeError("Wazuh indexer URL is not configured.")
+
+        body: dict[str, Any] = {
+            "size": limit,
+            "sort": [{"@timestamp": {"order": "desc"}}],
+            "query": {
+                "bool": {
+                    "must": [{"query_string": {"query": query}}],
+                    "filter": [
+                        {
+                            "range": {
+                                "@timestamp": {
+                                    "gte": f"now-{minutes}m",
+                                    "lte": "now",
+                                }
+                            }
+                        }
+                    ],
+                }
+            },
+        }
+
+        async with httpx.AsyncClient(
+            verify=False,
+            timeout=20.0,
+            auth=(self.indexer_username, self.indexer_password),
+        ) as client:
+            response = await client.post(f"{self.indexer_url}/wazuh-alerts-*/_search", json=body)
+            response.raise_for_status()
+            return response.json()

soc-integrator/app/config.py

@@ -18,6 +18,9 @@ class Settings(BaseSettings):
     wazuh_base_url: str = "https://wazuh.manager:55000"
     wazuh_username: str = "wazuh-wui"
     wazuh_password: str = "MyS3cr37P450r.*-"
+    wazuh_indexer_url: str = "https://wazuh.indexer:9200"
+    wazuh_indexer_username: str = "admin"
+    wazuh_indexer_password: str = "SecretPassword"
 
     shuffle_base_url: str = "http://shuffle-backend:5001"
     shuffle_api_key: str = ""

soc-integrator/app/db.py

@@ -81,5 +81,8 @@ def init_schema() -> None:
             "CREATE INDEX IF NOT EXISTS idx_incident_events_incident_key_created_at ON incident_events(incident_key, created_at DESC);"
         )
         cur.execute(
+            "CREATE INDEX IF NOT EXISTS idx_incident_events_source_event_id ON incident_events(source, event_id);"
+        )
+        cur.execute(
             "CREATE INDEX IF NOT EXISTS idx_escalation_audit_incident_key_attempted_at ON escalation_audit(incident_key, attempted_at DESC);"
         )

soc-integrator/app/main.py

@@ -1,4 +1,4 @@
-from fastapi import FastAPI, HTTPException
+from fastapi import Depends, FastAPI, HTTPException
 
 from app.adapters.iris import IrisAdapter
 from app.adapters.pagerduty import PagerDutyAdapter
@@ -25,6 +25,9 @@ wazuh_adapter = WazuhAdapter(
     base_url=settings.wazuh_base_url,
     username=settings.wazuh_username,
     password=settings.wazuh_password,
+    indexer_url=settings.wazuh_indexer_url,
+    indexer_username=settings.wazuh_indexer_username,
+    indexer_password=settings.wazuh_indexer_password,
 )
 shuffle_adapter = ShuffleAdapter(
     base_url=settings.shuffle_base_url,
@@ -297,3 +300,16 @@ async def wazuh_manager_logs(
     except Exception as exc:
         raise HTTPException(status_code=502, detail=f"Wazuh call failed: {exc}") from exc
     return ApiResponse(data={"wazuh": result})
+
+
+@app.post("/wazuh/sync-to-mvp", response_model=ApiResponse, dependencies=[Depends(require_internal_api_key)])
+async def wazuh_sync_to_mvp(
+    limit: int = 50,
+    minutes: int = 120,
+    q: str = "soc_mvp_test=true OR event_type:*",
+) -> ApiResponse:
+    try:
+        result = await mvp_service.sync_wazuh_alerts(query=q, limit=limit, minutes=minutes)
+    except Exception as exc:
+        raise HTTPException(status_code=502, detail=f"Wazuh sync failed: {exc}") from exc
+    return ApiResponse(data={"sync": result})

soc-integrator/app/repositories/mvp_repo.py

@@ -37,6 +37,14 @@ DEFAULT_POLICY: dict[str, Any] = {
 
 
 class MvpRepository:
+    def has_event(self, source: str, event_id: str) -> bool:
+        with get_conn() as conn, conn.cursor() as cur:
+            cur.execute(
+                "SELECT 1 FROM incident_events WHERE source = %s AND event_id = %s LIMIT 1",
+                (source, event_id),
+            )
+            return cur.fetchone() is not None
+
     def ensure_policy(self) -> None:
         with get_conn() as conn, conn.cursor() as cur:
             cur.execute("SELECT id FROM policy_config WHERE id = 1")

soc-integrator/app/services/mvp_service.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import hashlib
+import re
 import time
 from datetime import datetime, timezone
 from typing import Any
@@ -112,6 +113,93 @@ class MvpService:
             return str(data.get("case_id"))
         return None
 
+    def _parse_kv_pairs(self, text: str) -> dict[str, str]:
+        pattern = r"([A-Za-z0-9_]+)=('(?:[^']*)'|\"(?:[^\"]*)\"|[^\\s]+)"
+        out: dict[str, str] = {}
+        for key, raw in re.findall(pattern, text):
+            value = raw.strip().strip("'").strip('"')
+            out[key] = value
+        return out
+
+    def _severity_from_rule_level(self, rule_level: Any) -> str:
+        try:
+            level = int(rule_level)
+        except (TypeError, ValueError):
+            return "medium"
+        if level >= 12:
+            return "critical"
+        if level >= 8:
+            return "high"
+        if level >= 4:
+            return "medium"
+        return "low"
+
+    def _event_type_from_text(self, text: str, parsed: dict[str, str]) -> str:
+        explicit = parsed.get("event_type")
+        if explicit:
+            return explicit
+        lowered = text.lower()
+        if "vpn" in lowered and ("geo" in lowered or "country" in lowered):
+            return "vpn_geo_anomaly"
+        if "domain" in lowered or "dns" in lowered:
+            return "ioc_dns"
+        if "c2" in lowered or "ips" in lowered or "ip " in lowered:
+            return "ioc_ips"
+        if "auth" in lowered and "fail" in lowered:
+            return "auth_anomaly"
+        return "generic"
+
+    def _normalize_wazuh_hit(self, hit: dict[str, Any]) -> dict[str, Any]:
+        src = hit.get("_source", {})
+        full_log = str(src.get("full_log", ""))
+        parsed = self._parse_kv_pairs(full_log)
+        event_id = str(parsed.get("event_id") or src.get("id") or hit.get("_id") or f"wazuh-{int(time.time())}")
+        timestamp = (
+            src.get("@timestamp")
+            or src.get("timestamp")
+            or datetime.now(timezone.utc).isoformat()
+        )
+        rule = src.get("rule", {}) if isinstance(src.get("rule"), dict) else {}
+        rule_desc = str(rule.get("description") or "")
+        event_type = self._event_type_from_text(full_log, parsed)
+        severity = str(parsed.get("severity", "")).lower() or self._severity_from_rule_level(rule.get("level"))
+
+        src_ip = parsed.get("src_ip")
+        dst_ip = parsed.get("dst_ip")
+        domain = parsed.get("query") or parsed.get("domain")
+        country = parsed.get("country")
+        user = parsed.get("user") or (src.get("agent", {}) or {}).get("name")
+
+        title = rule_desc or f"Wazuh alert {rule.get('id', '')}".strip()
+        description = full_log or rule_desc or "Wazuh alert"
+
+        return {
+            "source": "wazuh",
+            "event_type": event_type,
+            "event_id": event_id,
+            "timestamp": timestamp,
+            "severity": severity if severity in {"low", "medium", "high", "critical"} else "medium",
+            "title": title,
+            "description": description,
+            "asset": {
+                "user": user,
+                "hostname": (src.get("agent", {}) or {}).get("name"),
+                "agent_id": (src.get("agent", {}) or {}).get("id"),
+            },
+            "network": {
+                "src_ip": src_ip,
+                "dst_ip": dst_ip,
+                "domain": domain,
+                "country": country,
+            },
+            "tags": ["wazuh", event_type, f"rule_{rule.get('id', 'unknown')}"],
+            "risk_context": {
+                "outside_thailand": bool(country and str(country).upper() != "TH"),
+            },
+            "raw": src,
+            "payload": {},
+        }
+
     async def ingest_incident(self, event: dict[str, Any]) -> dict[str, Any]:
         policy = self.repo.get_policy()
         incident_key = self._incident_key(event)
@@ -310,6 +398,52 @@ class MvpService:
             "escalation_stub_sent": ingest_result.get("escalation_stub_sent", False),
         }
 
+    async def sync_wazuh_alerts(
+        self,
+        query: str = "soc_mvp_test=true OR event_type:*",
+        limit: int = 50,
+        minutes: int = 120,
+    ) -> dict[str, Any]:
+        raw = await self.wazuh_adapter.search_alerts(query=query, limit=limit, minutes=minutes)
+        hits = (raw.get("hits", {}) or {}).get("hits", []) if isinstance(raw, dict) else []
+
+        processed = 0
+        ingested = 0
+        skipped_existing = 0
+        failed = 0
+        errors: list[str] = []
+        created_incidents: list[str] = []
+
+        for hit in hits:
+            processed += 1
+            event = self._normalize_wazuh_hit(hit)
+            event_id = str(event.get("event_id", "")).strip()
+            if event_id and self.repo.has_event("wazuh", event_id):
+                skipped_existing += 1
+                continue
+            try:
+                result = await self.ingest_incident(event)
+                ingested += 1
+                incident_key = str(result.get("incident_key", ""))
+                if incident_key:
+                    created_incidents.append(incident_key)
+            except Exception as exc:
+                failed += 1
+                errors.append(f"{event_id or 'unknown_event'}: {exc}")
+
+        return {
+            "query": query,
+            "window_minutes": minutes,
+            "limit": limit,
+            "processed": processed,
+            "ingested": ingested,
+            "skipped_existing": skipped_existing,
+            "failed": failed,
+            "incident_keys": created_incidents,
+            "errors": errors[:10],
+            "total_hits": (raw.get("hits", {}).get("total", {}) if isinstance(raw, dict) else {}),
+        }
+
     async def dependency_health(self) -> dict[str, Any]:
         out: dict[str, Any] = {}