From 8550fb066cd09780c4229e5705307f8b60d2dc66 Mon Sep 17 00:00:00 2001 From: Dan Cohen Vaxman Date: Wed, 15 Jul 2026 15:04:46 +0200 Subject: [PATCH] Add injectable Kibana and Elasticsearch client wrappers --- src/detection_as_code/elastic_client.py | 57 ++++++++++++++++++++++ src/detection_as_code/kibana_client.py | 65 +++++++++++++++++++++++++ tests/test_elastic_client.py | 38 +++++++++++++++ tests/test_kibana_client.py | 50 +++++++++++++++++++ 4 files changed, 210 insertions(+) create mode 100644 src/detection_as_code/elastic_client.py create mode 100644 src/detection_as_code/kibana_client.py create mode 100644 tests/test_elastic_client.py create mode 100644 tests/test_kibana_client.py diff --git a/src/detection_as_code/elastic_client.py b/src/detection_as_code/elastic_client.py new file mode 100644 index 0000000..f87f79e --- /dev/null +++ b/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"]) diff --git a/src/detection_as_code/kibana_client.py b/src/detection_as_code/kibana_client.py new file mode 100644 index 0000000..ac27bdf --- /dev/null +++ b/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() diff --git a/tests/test_elastic_client.py b/tests/test_elastic_client.py new file mode 100644 index 0000000..2c32030 --- /dev/null +++ b/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 diff --git a/tests/test_kibana_client.py b/tests/test_kibana_client.py new file mode 100644 index 0000000..355b86f --- /dev/null +++ b/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"})