featt: Add injectable Kibana and Elasticsearch client wrappers

src/detection_as_code/elastic_client.py

@@ -0,0 +1,57 @@
+from __future__ import annotations
+
+from typing import Any
+
+from .secrets import SecretProvider
+
+
+class ElasticQueryError(RuntimeError):
+    """Raised when Elasticsearch rejects a query or the request itself fails."""
+
+
+class ElasticClient:
+    """Wraps the Elasticsearch client for the two things the pipeline needs
+    before a rule ships: confirming the query is syntactically valid, and
+    backtesting it against recent data so a rule doesn't go live silent.
+    """
+
+    def __init__(self, url: str, api_key: str, client: Any = None):
+        if client is None:
+            from elasticsearch import Elasticsearch
+
+            client = Elasticsearch(url, api_key=api_key)
+        self._client = client
+
+    @classmethod
+    def from_secrets(cls, secrets: SecretProvider) -> ElasticClient:
+        return cls(url=secrets.get("elastic_url"), api_key=secrets.get("elastic_api_key"))
+
+    def validate_query(self, index_patterns: list[str], query: str) -> bool:
+        try:
+            response = self._client.indices.validate_query(
+                index=",".join(index_patterns), q=query, explain=True
+            )
+        except Exception as exc:  # noqa: BLE001 - surface as a domain error
+            raise ElasticQueryError(f"Query validation request failed: {exc}") from exc
+        return bool(response.get("valid", False))
+
+    def count_matches(
+        self, index_patterns: list[str], query: str, lookback: str = "now-7d/d"
+    ) -> int:
+        """Run the rule's query over the lookback window and return the hit
+        count, so a rule with zero backtest hits gets flagged before it ships
+        rather than discovered silent in production.
+        """
+        body = {
+            "query": {
+                "bool": {
+                    "must": [{"query_string": {"query": query}}],
+                    "filter": [{"range": {"@timestamp": {"gte": lookback, "lt": "now"}}}],
+                }
+            }
+        }
+        try:
+            response = self._client.search(index=",".join(index_patterns), body=body, size=0)
+        except Exception as exc:  # noqa: BLE001 - surface as a domain error
+            raise ElasticQueryError(f"Backtest search failed: {exc}") from exc
+        return int(response["hits"]["total"]["value"])

src/detection_as_code/kibana_client.py

@@ -0,0 +1,65 @@
+from __future__ import annotations
+
+from typing import Any
+
+import requests
+from requests.auth import HTTPBasicAuth
+
+from .secrets import SecretProvider
+
+
+class KibanaClientError(RuntimeError):
+    """Raised when the Kibana Detection Engine API rejects a request."""
+
+
+class KibanaClient:
+    """Thin, dependency-injected wrapper around the Kibana Detection Engine
+    API. The session is injectable so tests never touch the network.
+    """
+
+    def __init__(
+        self,
+        base_url: str,
+        username: str,
+        password: str,
+        session: requests.Session | None = None,
+        timeout: float = 15.0,
+    ):
+        self._base_url = base_url.rstrip("/")
+        self._auth = HTTPBasicAuth(username, password)
+        self._session = session or requests.Session()
+        self._timeout = timeout
+        self._headers = {"Content-Type": "application/json", "kbn-xsrf": "true"}
+
+    @classmethod
+    def from_secrets(cls, secrets: SecretProvider) -> KibanaClient:
+        return cls(
+            base_url=secrets.get("kibana_url"),
+            username=secrets.get("kibana_user"),
+            password=secrets.get("kibana_password"),
+        )
+
+    def list_connectors(self) -> list[dict[str, Any]]:
+        resp = self._session.get(
+            f"{self._base_url}/api/actions/connectors",
+            headers=self._headers,
+            auth=self._auth,
+            timeout=self._timeout,
+        )
+        if not resp.ok:
+            raise KibanaClientError(f"Failed to list connectors: {resp.status_code} {resp.text}")
+        return resp.json()
+
+    def create_rule(self, rule_payload: dict[str, Any]) -> dict[str, Any]:
+        resp = self._session.post(
+            f"{self._base_url}/api/detection_engine/rules",
+            headers=self._headers,
+            json=rule_payload,
+            auth=self._auth,
+            timeout=self._timeout,
+        )
+        if resp.status_code == 409:
+            raise KibanaClientError(f"Rule '{rule_payload.get('name')}' already exists")
+        if not resp.ok:
+            raise KibanaClientError(f"Failed to create rule: {resp.status_code} {resp.text}")
+        return resp.json()

tests/test_elastic_client.py

@@ -0,0 +1,38 @@
+from unittest.mock import MagicMock
+
+import pytest
+
+from detection_as_code.elastic_client import ElasticClient, ElasticQueryError
+
+
+def test_validate_query_true():
+    fake_es = MagicMock()
+    fake_es.indices.validate_query.return_value = {"valid": True}
+    client = ElasticClient(url="https://es.example.com", api_key="key", client=fake_es)
+
+    assert client.validate_query(["winlogbeat-*"], 'process.name:"a.exe"') is True
+
+
+def test_validate_query_false():
+    fake_es = MagicMock()
+    fake_es.indices.validate_query.return_value = {"valid": False}
+    client = ElasticClient(url="https://es.example.com", api_key="key", client=fake_es)
+
+    assert client.validate_query(["winlogbeat-*"], "bad::syntax") is False
+
+
+def test_validate_query_wraps_exceptions():
+    fake_es = MagicMock()
+    fake_es.indices.validate_query.side_effect = RuntimeError("boom")
+    client = ElasticClient(url="https://es.example.com", api_key="key", client=fake_es)
+
+    with pytest.raises(ElasticQueryError, match="boom"):
+        client.validate_query(["winlogbeat-*"], 'process.name:"a.exe"')
+
+
+def test_count_matches_returns_hit_total():
+    fake_es = MagicMock()
+    fake_es.search.return_value = {"hits": {"total": {"value": 42}}}
+    client = ElasticClient(url="https://es.example.com", api_key="key", client=fake_es)
+
+    assert client.count_matches(["winlogbeat-*"], 'process.name:"a.exe"') == 42

tests/test_kibana_client.py

@@ -0,0 +1,50 @@
+from unittest.mock import MagicMock
+
+import pytest
+
+from detection_as_code.kibana_client import KibanaClient, KibanaClientError
+
+
+def _client_with_session(session):
+    return KibanaClient(
+        base_url="https://kibana.example.com", username="user", password="pass", session=session
+    )
+
+
+def test_list_connectors_success():
+    session = MagicMock()
+    session.get.return_value = MagicMock(ok=True, json=lambda: [{"id": "1", "name": "torq"}])
+    client = _client_with_session(session)
+
+    connectors = client.list_connectors()
+
+    assert connectors == [{"id": "1", "name": "torq"}]
+    session.get.assert_called_once()
+
+
+def test_list_connectors_failure_raises():
+    session = MagicMock()
+    session.get.return_value = MagicMock(ok=False, status_code=403, text="forbidden")
+    client = _client_with_session(session)
+
+    with pytest.raises(KibanaClientError, match="403"):
+        client.list_connectors()
+
+
+def test_create_rule_success():
+    session = MagicMock()
+    session.post.return_value = MagicMock(ok=True, json=lambda: {"id": "abc123"})
+    client = _client_with_session(session)
+
+    result = client.create_rule({"name": "my_rule"})
+
+    assert result["id"] == "abc123"
+
+
+def test_create_rule_conflict_raises_clear_error():
+    session = MagicMock()
+    session.post.return_value = MagicMock(ok=False, status_code=409, text="conflict")
+    client = _client_with_session(session)
+
+    with pytest.raises(KibanaClientError, match="already exists"):
+        client.create_rule({"name": "duplicate_rule"})