feat(integrations): Splunk data source + anomaly forwarding (#99)

core/api_routers/models.py

@@ -26,6 +26,8 @@ class TrainRequest(BaseModel):
     file_path: Optional[str] = None
     file_name: Optional[str] = None
     source_group_slug: Optional[str] = None
+    splunk_search: Optional[str] = None
+    splunk_index: Optional[str] = None
     query: Optional[Dict[str, Any]] = None
     version_id: Optional[str] = None
 
@@ -37,6 +39,8 @@ class ExecuteRequest(BaseModel):
     file_path: Optional[str] = None
     file_name: Optional[str] = None
     source_group_slug: Optional[str] = None
+    splunk_search: Optional[str] = None
+    splunk_index: Optional[str] = None
     query: Optional[Dict[str, Any]] = None
     version_id: Optional[str] = None
     artifact_id: Optional[str] = None
@@ -311,6 +315,8 @@ async def train_model(
     file_path: Optional[str] = Query(None, description="local file path"),
     file_name: Optional[str] = Query(None, description="local file name"),
     source_group_slug: Optional[str] = Query(None, description="source group slug"),
+    splunk_search: Optional[str] = Query(None, description="splunk search query (SPL)"),
+    splunk_index: Optional[str] = Query(None, description="splunk index to search"),
     db: Session = Depends(get_db),
     current_user: dict = Depends(require_permission("models", "write"))
 ):
@@ -328,6 +334,8 @@ async def train_model(
     _file_path = (body.file_path if body else None) or file_path
     _file_name = (body.file_name if body else None) or file_name
     _source_group_slug = (body.source_group_slug if body else None) or source_group_slug
+    _splunk_search = (body.splunk_search if body else None) or splunk_search
+    _splunk_index = (body.splunk_index if body else None) or splunk_index
     _query = (body.query if body else None) or {"match_all": {}}
     _version_id = None
     if body and body.version_id:
@@ -362,6 +370,10 @@ async def train_model(
             input_data["file_name"] = _file_name
         elif _data_source == "source_group" and _source_group_slug:
             input_data["source_group_slug"] = _source_group_slug
+        elif _data_source == "splunk" and _splunk_search:
+            input_data["splunk_search"] = _splunk_search
+            if _splunk_index:
+                input_data["splunk_index"] = _splunk_index
 
     try:
         orchestrator = ModelOrchestrator()
@@ -410,6 +422,8 @@ async def execute_model(
     file_path: Optional[str] = Query(None, description="local file path"),
     file_name: Optional[str] = Query(None, description="local file name"),
     source_group_slug: Optional[str] = Query(None, description="source group slug"),
+    splunk_search: Optional[str] = Query(None, description="splunk search query (SPL)"),
+    splunk_index: Optional[str] = Query(None, description="splunk index to search"),
     db: Session = Depends(get_db),
     current_user: dict = Depends(require_permission("models", "write"))
 ):
@@ -427,6 +441,8 @@ async def execute_model(
     _file_path = (body.file_path if body else None) or file_path
     _file_name = (body.file_name if body else None) or file_name
     _source_group_slug = (body.source_group_slug if body else None) or source_group_slug
+    _splunk_search = (body.splunk_search if body else None) or splunk_search
+    _splunk_index = (body.splunk_index if body else None) or splunk_index
     _query = (body.query if body else None) or {"match_all": {}}
     _version_id = None
     if body and body.version_id:
@@ -466,6 +482,10 @@ async def execute_model(
             input_data["file_name"] = _file_name
         elif _data_source == "source_group" and _source_group_slug:
             input_data["source_group_slug"] = _source_group_slug
+        elif _data_source == "splunk" and _splunk_search:
+            input_data["splunk_search"] = _splunk_search
+            if _splunk_index:
+                input_data["splunk_index"] = _splunk_index
 
     try:
         orchestrator = ModelOrchestrator()

core/api_routers/settings.py

@@ -17,7 +17,7 @@
 router = APIRouter()
 logger = logging.getLogger(__name__)
 
-VALID_INTEGRATION_TYPES = {"ollama", "openai", "claude", "gemini", "elasticsearch", "spark"}
+VALID_INTEGRATION_TYPES = {"ollama", "openai", "claude", "gemini", "elasticsearch", "spark", "splunk"}
 
 
 class IntegrationConfigUpdate(BaseModel):
@@ -183,6 +183,8 @@ async def test_integration(
             return await _test_elasticsearch(config)
         elif integration_type == "spark":
             return await _test_spark(config)
+        elif integration_type == "splunk":
+            return await _test_splunk(config)
     except Exception as e:
         logger.error(f"test {integration_type} failed: {e}")
         return {"status": "error", "message": str(e)}
@@ -283,10 +285,16 @@ async def _test_spark(config: dict) -> dict:
         return {"status": "error", "message": f"HTTP {resp.status_code}"}
 
 
+async def _test_splunk(config: dict) -> dict:
+    '''verify Splunk connectivity via the shared SplunkConnector'''
+    from core.integrations.splunk import SplunkConnector
+    return SplunkConnector.from_config(config).test_connection()
+
+
 def _mask_sensitive_fields(config: dict) -> dict:
-    '''mask api keys in config for safe display'''
+    '''mask api keys / tokens / passwords in config for safe display'''
     masked = dict(config)
-    for key in ("api_key",):
+    for key in ("api_key", "token", "hec_token", "password"):
         if key in masked and masked[key]:
             val = str(masked[key])
             if len(val) > 8:

core/integrations/__init__.py

@@ -5,9 +5,11 @@
 
 from core.integrations.spark import SparkConnector
 from core.integrations.elasticsearch import ElasticsearchConnector
+from core.integrations.splunk import SplunkConnector
 
 __all__ = [
     "SparkConnector",
     "ElasticsearchConnector",
+    "SplunkConnector",
 ]
 

core/integrations/splunk.py

@@ -0,0 +1,244 @@
+'''
+Copyright 2019-Present The OpenUBA Platform Authors
+splunk integration
+
+Two directions:
+  * INPUT  — run a Splunk search and return events (via the REST export API)
+  * OUTPUT — forward anomalies/events to Splunk (via the HTTP Event Collector)
+
+Config comes from constructor args with SPLUNK_* env fallbacks, matching the
+ElasticsearchConnector pattern. Nothing here is Splunk-SDK-dependent — it uses
+plain `requests`, so it also works unchanged inside the self-contained model
+runner container.
+'''
+
+import json
+import logging
+import os
+from typing import Any, Dict, List, Optional
+
+import requests
+
+logger = logging.getLogger(__name__)
+
+
+def _env_flag(name: str, default: bool) -> bool:
+    val = os.getenv(name)
+    if val is None:
+        return default
+    return val.strip().lower() in ("1", "true", "yes", "on")
+
+
+class SplunkConnector:
+    '''
+    connector for Splunk Enterprise / Cloud.
+
+    host      — management REST endpoint, e.g. https://splunk:8089 (for search)
+    token     — bearer token for the REST API (preferred), OR
+    username/password — basic auth for the REST API
+    hec_url   — HTTP Event Collector base, e.g. https://splunk:8088 (for output)
+    hec_token — HEC token
+    '''
+
+    def __init__(
+        self,
+        host: Optional[str] = None,
+        token: Optional[str] = None,
+        username: Optional[str] = None,
+        password: Optional[str] = None,
+        hec_url: Optional[str] = None,
+        hec_token: Optional[str] = None,
+        verify_ssl: Optional[bool] = None,
+        timeout: int = 60,
+    ):
+        self.host = (host or os.getenv("SPLUNK_HOST", "")).rstrip("/")
+        self.token = token or os.getenv("SPLUNK_TOKEN", "")
+        self.username = username or os.getenv("SPLUNK_USERNAME", "")
+        self.password = password or os.getenv("SPLUNK_PASSWORD", "")
+        self.hec_url = (hec_url or os.getenv("SPLUNK_HEC_URL", "")).rstrip("/")
+        self.hec_token = hec_token or os.getenv("SPLUNK_HEC_TOKEN", "")
+        if verify_ssl is None:
+            verify_ssl = _env_flag("SPLUNK_VERIFY_SSL", True)
+        self.verify_ssl = verify_ssl
+        self.timeout = timeout
+
+    @classmethod
+    def from_config(cls, config: Dict[str, Any]) -> "SplunkConnector":
+        '''build a connector from an integration_settings config dict'''
+        return cls(
+            host=config.get("host"),
+            token=config.get("token"),
+            username=config.get("username"),
+            password=config.get("password"),
+            hec_url=config.get("hec_url"),
+            hec_token=config.get("hec_token"),
+            verify_ssl=config.get("verify_ssl", True),
+        )
+
+    # ── auth ─────────────────────────────────────────────────────────
+
+    def _rest_headers(self) -> Dict[str, str]:
+        headers = {"Content-Type": "application/x-www-form-urlencoded"}
+        if self.token:
+            headers["Authorization"] = f"Bearer {self.token}"
+        return headers
+
+    def _rest_auth(self):
+        '''basic auth tuple when no bearer token is configured'''
+        if not self.token and self.username:
+            return (self.username, self.password)
+        return None
+
+    # ── INPUT: search ────────────────────────────────────────────────
+
+    @staticmethod
+    def _normalize_query(query: str) -> str:
+        '''Splunk searches must start with `search` or a generating command (`|`)'''
+        q = (query or "").strip()
+        if not q:
+            raise ValueError("splunk search query is empty")
+        if q.startswith("|") or q.lower().startswith("search "):
+            return q
+        return f"search {q}"
+
+    def search(
+        self,
+        query: str,
+        earliest_time: str = "-24h",
+        latest_time: str = "now",
+        max_count: int = 1000,
+    ) -> List[Dict[str, Any]]:
+        '''
+        run a blocking Splunk search via the export endpoint and return the
+        list of result rows (each a flat dict of field → value).
+        '''
+        if not self.host:
+            raise ValueError("splunk host not configured")
+
+        url = f"{self.host}/services/search/jobs/export"
+        payload = {
+            "search": self._normalize_query(query),
+            "output_mode": "json",
+            "earliest_time": earliest_time,
+            "latest_time": latest_time,
+            "count": max_count,
+        }
+        logger.info(f"running splunk search on {self.host} (count<={max_count})")
+        resp = requests.post(
+            url,
+            data=payload,
+            headers=self._rest_headers(),
+            auth=self._rest_auth(),
+            verify=self.verify_ssl,
+            timeout=self.timeout,
+        )
+        if resp.status_code != 200:
+            raise ValueError(
+                f"splunk search failed ({resp.status_code}): {resp.text[:500]}"
+            )
+
+        results: List[Dict[str, Any]] = []
+        # the export endpoint streams newline-delimited JSON objects
+        for line in resp.text.splitlines():
+            line = line.strip()
+            if not line:
+                continue
+            try:
+                obj = json.loads(line)
+            except json.JSONDecodeError:
+                continue
+            row = obj.get("result")
+            if isinstance(row, dict):
+                results.append(row)
+        logger.info(f"splunk search returned {len(results)} rows")
+        return results
+
+    # ── OUTPUT: HTTP Event Collector ─────────────────────────────────
+
+    def send_event(
+        self,
+        event: Dict[str, Any],
+        sourcetype: str = "openuba",
+        source: str = "openuba",
+        index: Optional[str] = None,
+    ) -> bool:
+        '''forward a single event to Splunk via HEC. returns True on success.'''
+        if not self.hec_url or not self.hec_token:
+            logger.debug("splunk HEC not configured — skipping event forward")
+            return False
+
+        url = f"{self.hec_url}/services/collector/event"
+        body: Dict[str, Any] = {
+            "event": event,
+            "sourcetype": sourcetype,
+            "source": source,
+        }
+        if index:
+            body["index"] = index
+        try:
+            resp = requests.post(
+                url,
+                data=json.dumps(body),
+                headers={"Authorization": f"Splunk {self.hec_token}"},
+                verify=self.verify_ssl,
+                timeout=self.timeout,
+            )
+            if resp.status_code == 200:
+                return True
+            logger.error(f"splunk HEC send failed ({resp.status_code}): {resp.text[:300]}")
+            return False
+        except Exception as e:
+            logger.error(f"splunk HEC send error: {e}")
+            return False
+
+    def send_anomaly(self, anomaly: Dict[str, Any], index: Optional[str] = None) -> bool:
+        '''forward an OpenUBA anomaly to Splunk as an openuba:anomaly event'''
+        return self.send_event(anomaly, sourcetype="openuba:anomaly", index=index)
+
+    def send_anomalies(self, anomalies: List[Dict[str, Any]], index: Optional[str] = None) -> int:
+        '''forward a batch of anomalies; returns the count successfully sent'''
+        sent = 0
+        for a in anomalies:
+            if self.send_anomaly(a, index=index):
+                sent += 1
+        return sent
+
+    # ── connectivity ─────────────────────────────────────────────────
+
+    def test_connection(self) -> Dict[str, Any]:
+        '''probe the REST management endpoint (or HEC if only that is set)'''
+        try:
+            if self.host:
+                resp = requests.get(
+                    f"{self.host}/services/server/info",
+                    params={"output_mode": "json"},
+                    headers=self._rest_headers(),
+                    auth=self._rest_auth(),
+                    verify=self.verify_ssl,
+                    timeout=10,
+                )
+                if resp.status_code == 200:
+                    version = ""
+                    try:
+                        entries = resp.json().get("entry", [])
+                        if entries:
+                            version = entries[0].get("content", {}).get("version", "")
+                    except Exception:
+                        pass
+                    return {"status": "connected", "version": version}
+                return {"status": "error", "message": f"HTTP {resp.status_code}"}
+
+            if self.hec_url and self.hec_token:
+                resp = requests.get(
+                    f"{self.hec_url}/services/collector/health",
+                    headers={"Authorization": f"Splunk {self.hec_token}"},
+                    verify=self.verify_ssl,
+                    timeout=10,
+                )
+                if resp.status_code == 200:
+                    return {"status": "connected", "channel": "hec"}
+                return {"status": "error", "message": f"HEC HTTP {resp.status_code}"}
+
+            return {"status": "error", "message": "host or hec_url not configured"}
+        except Exception as e:
+            return {"status": "error", "message": str(e)}