Fix Support username and pw for kibana auth analog to elastic which is already supported according to Docu but wasnt implemented

CLI.md

@@ -32,7 +32,7 @@ Currently supported arguments:
 #### Using environment variables
 
 Environment variables using the argument format: `DR_<UPPERCASED_ARG_NAME>` will be parsed in commands which expect it.
-EX: `DR_USER=joe`
+EX: `DR_ES_USER=joe`
 
 
 Using the environment variable `DR_BYPASS_NOTE_VALIDATION_AND_PARSE` will bypass the Detection Rules validation on the `note` field in toml files.
@@ -149,6 +149,8 @@ Options:
   --ignore-ssl-errors TEXT
   --space TEXT              Kibana space
   --api-key TEXT
+  --kibana-user TEXT
+  --kibana-password TEXT
   --cloud-id TEXT           ID of the cloud instance.
   --kibana-url TEXT
   -h, --help                Show this message and exit.
@@ -177,6 +179,8 @@ Options:
   --ignore-ssl-errors TEXT
   --space TEXT              Kibana space
   --api-key TEXT
+  --kibana-user TEXT
+  --kibana-password TEXT
   --cloud-id TEXT           ID of the cloud instance.
   --kibana-url TEXT
 
@@ -237,6 +241,8 @@ Options:
   --ignore-ssl-errors TEXT
   --space TEXT              Kibana space
   --api-key TEXT
+  --kibana-user TEXT
+  --kibana-password TEXT
   --cloud-id TEXT           ID of the cloud instance.
   --kibana-url TEXT
 
@@ -445,6 +451,8 @@ Options:
   --ignore-ssl-errors TEXT
   --space TEXT              Kibana space
   --api-key TEXT
+  --kibana-user TEXT
+  --kibana-password TEXT
   --cloud-id TEXT           ID of the cloud instance.
   --kibana-url TEXT
 
@@ -479,6 +487,8 @@ Options:
   --ignore-ssl-errors TEXT
   --space TEXT              Kibana space
   --api-key TEXT
+  --kibana-user TEXT
+  --kibana-password TEXT
   --cloud-id TEXT           ID of the cloud instance.
   --kibana-url TEXT
 

detection_rules/misc.py

@@ -387,7 +387,9 @@ def get_elasticsearch_client(  # noqa: PLR0913
 
 def get_kibana_client(
     *,
-    api_key: str,
+    api_key: str | None = None,
+    kibana_user: str | None = None,
+    kibana_password: str | None = None,
     cloud_id: str | None = None,
     kibana_url: str | None = None,
     space: str | None = None,
@@ -399,14 +401,25 @@ def get_kibana_client(
         raise_client_error("Missing required --cloud-id or --kibana-url")
 
     verify = not ignore_ssl_errors
-    return Kibana(cloud_id=cloud_id, kibana_url=kibana_url, space=space, verify=verify, api_key=api_key, **kwargs)
+    return Kibana(
+        cloud_id=cloud_id,
+        kibana_url=kibana_url,
+        api_key=api_key,
+        username=kibana_user,
+        password=kibana_password,
+        space=space,
+        verify=verify,
+        **kwargs,
+    )
 
 
 client_options = {
     "kibana": {
         "kibana_url": click.Option(["--kibana-url"], default=getdefault("kibana_url")),
         "cloud_id": click.Option(["--cloud-id"], default=getdefault("cloud_id"), help="ID of the cloud instance."),
         "api_key": click.Option(["--api-key"], default=getdefault("api_key")),
+        "kibana_user": click.Option(["--kibana-user", "-ku"], default=getdefault("kibana_user")),
+        "kibana_password": click.Option(["--kibana-password", "-kp"], default=getdefault("kibana_password")),
         "space": click.Option(["--space"], default=None, help="Kibana space"),
         "ignore_ssl_errors": click.Option(["--ignore-ssl-errors"], default=getdefault("ignore_ssl_errors")),
     },

detection_rules/remote_validation.py

@@ -42,7 +42,8 @@ class RemoteConnector:
 
     def __init__(self, parse_config: bool = False, **kwargs: Any) -> None:
         es_args = ["cloud_id", "ignore_ssl_errors", "elasticsearch_url", "es_user", "es_password", "timeout"]
-        kibana_args = ["cloud_id", "ignore_ssl_errors", "kibana_url", "api_key", "space"]
+        kibana_args = ["cloud_id", "ignore_ssl_errors", "kibana_url", "api_key",
+                       "kibana_user", "kibana_password", "space"]
 
         if parse_config:
             es_kwargs = {arg: getdefault(arg)() for arg in es_args}
@@ -88,7 +89,9 @@ def auth_es(  # noqa: PLR0913
     def auth_kibana(
         self,
         *,
-        api_key: str,
+        api_key: str | None = None,
+        kibana_user: str | None = None,
+        kibana_password: str | None = None,
         cloud_id: str | None = None,
         kibana_url: str | None = None,
         space: str | None = None,
@@ -101,6 +104,8 @@ def auth_kibana(
             ignore_ssl_errors=ignore_ssl_errors,
             kibana_url=kibana_url,
             api_key=api_key,
+            kibana_user=kibana_user,
+            kibana_password=kibana_password,
             space=space,
             **kwargs,
         )

lib/kibana/kibana/connector.py

@@ -21,7 +21,17 @@
 class Kibana:
     """Wrapper around the Kibana SIEM APIs."""
 
-    def __init__(self, cloud_id=None, kibana_url=None, api_key=None, verify=True, elasticsearch=None, space=None):
+    def __init__(
+        self,
+        cloud_id=None,
+        kibana_url=None,
+        api_key=None,
+        username=None,
+        password=None,
+        verify=True,
+        elasticsearch=None,
+        space=None,
+    ):
         """"Open a session to the platform."""
         self.authenticated = False
 
@@ -35,6 +45,9 @@ def __init__(self, cloud_id=None, kibana_url=None, api_key=None, verify=True, el
                     "Authorization": f"ApiKey {api_key}",
                 }
             )
+        elif username and password:
+            self.session.auth = (username, password)
+            self.session.headers.update({"kbn-xsrf": "true"})
 
         self.verify = verify
 
@@ -65,11 +78,12 @@ def __init__(self, cloud_id=None, kibana_url=None, api_key=None, verify=True, el
         self.elasticsearch = elasticsearch
 
         if not self.elasticsearch and self.elastic_url:
-            self.elasticsearch = Elasticsearch(
-                hosts=[self.elastic_url],
-                api_key=api_key,
-                verify_certs=self.verify,
-            )
+            es_kwargs = {"verify_certs": self.verify}
+            if api_key:
+                es_kwargs["api_key"] = api_key
+            elif username and password:
+                es_kwargs["basic_auth"] = (username, password)
+            self.elasticsearch = Elasticsearch(hosts=[self.elastic_url], **es_kwargs)
             self.elasticsearch.info()
 
         if not verify:

lib/kibana/pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "detection-rules-kibana"
-version = "0.4.4"
+version = "0.5.0"
 description = "Kibana API utilities for Elastic Detection Rules"
 license = {text = "Elastic License v2"}
 keywords = ["Elastic", "Kibana", "Detection Rules", "Security", "Elasticsearch"]