Add support for URL shortening on Kibana 7.16

CHANGELOG.md

@@ -31,6 +31,7 @@
 - sphinx 4.3.1 to 4.3.2 - [#618](https://github.com/jertel/elastalert2/pull/618) - @nsano-rururu
 - Remove unused parameter boto-profile - [#622](https://github.com/jertel/elastalert2/pull/622) - @nsano-rururu
 - [Docs] Include Docker example; add additional FAQs - [#623](https://github.com/jertel/elastalert2/pull/623) - @nsano-rururu
+- Add support for URL shortening with Kibana 7.16+ - [#633](https://github.com/jertel/elastalert2/pull/633) - @jertel
 
 # 2.2.3
 

elastalert/kibana_external_url_formatter.py

@@ -1,4 +1,5 @@
 import boto3
+from datetime import datetime
 import os
 from urllib.parse import parse_qsl, urlencode, urljoin, urlparse, urlsplit, urlunsplit
 
@@ -46,12 +47,19 @@ def format(self, relative_url: str) -> str:
 class ShortKibanaExternalUrlFormatter(KibanaExternalUrlFormatter):
     '''Formats external urls using the Kibana Shorten URL API'''
 
-    def __init__(self, base_url: str, auth: AuthBase, security_tenant: str) -> None:
+    def __init__(self, base_url: str, auth: AuthBase, security_tenant: str, new_shortener: bool, id: str) -> None:
         self.auth = auth
         self.security_tenant = security_tenant
         self.goto_url = urljoin(base_url, 'goto/')
+        self.use_new_shortener = new_shortener
+        self.id = id
 
-        shorten_url = urljoin(base_url, 'api/shorten_url')
+        if self.use_new_shortener:
+            path = 'api/short_url'
+        else:
+            path = 'api/shorten_url'
+
+        shorten_url = urljoin(base_url, path)
         if security_tenant:
             shorten_url = append_security_tenant(shorten_url, security_tenant)
         self.shorten_url = shorten_url
@@ -62,6 +70,13 @@ def format(self, relative_url: str) -> str:
         if self.security_tenant:
             long_url = append_security_tenant(long_url, self.security_tenant)
 
+        if self.use_new_shortener:
+            json = { 'locatorId': "elastalert-" + self.id, 'params': { 'url': long_url } }
+            response_param = 'id'
+        else:
+            json = { 'url': long_url }
+            response_param = 'urlId'
+
         try:
             response = requests.post(
                 url=self.shorten_url,
@@ -70,16 +85,14 @@ def format(self, relative_url: str) -> str:
                     'kbn-xsrf': 'elastalert',
                     'osd-xsrf': 'elastalert'
                 },
-                json={
-                    'url': long_url
-                }
+                json=json
             )
             response.raise_for_status()
         except RequestException as e:
             raise EAException("Failed to invoke Kibana Shorten URL API: %s" % e)
 
         response_body = response.json()
-        url_id = response_body.get('urlId')
+        url_id = response_body.get(response_param)
 
         goto_url = urljoin(self.goto_url, url_id)
         if self.security_tenant:
@@ -121,18 +134,29 @@ def create_kibana_auth(kibana_url, rule) -> AuthBase:
     # Unauthenticated
     return None
 
+def is_kibana_atleastsevensixteen(version: str):
+    """
+    Returns True when the Kibana server version >= 7.16
+    """
+    major, minor = list(map(int, version.split(".")[:2]))
+    return major > 7 or (major == 7 and minor >= 16)
 
 def create_kibana_external_url_formatter(
     rule,
     shorten: bool,
-    security_tenant: str
+    security_tenant: str,
 ) -> KibanaExternalUrlFormatter:
     '''Creates a Kibana external url formatter'''
 
     base_url = rule.get('kibana_url')
+    new_shortener = is_kibana_atleastsevensixteen(rule.get('kibana_discover_version', '0.0'))
 
     if shorten:
         auth = create_kibana_auth(base_url, rule)
-        return ShortKibanaExternalUrlFormatter(base_url, auth, security_tenant)
+
+        # id must be unique, can be used to delete the shortened url, currently unused except for creation
+        id = datetime.now().isoformat()
+
+        return ShortKibanaExternalUrlFormatter(base_url, auth, security_tenant, new_shortener, id)
 
     return AbsoluteKibanaExternalUrlFormatter(base_url, security_tenant)

tests/kibana_external_url_formatter_test.py

@@ -97,6 +97,30 @@ def raise_for_status(self):
         return MockResponse(400)
 
 
+def mock_7_16_kibana_shorten_url_api(*args, **kwargs):
+    class MockResponse:
+        def __init__(self, status_code):
+            self.status_code = status_code
+
+        def json(self):
+            return {
+                'id': 'a1f77a80-6847-11ec-9b91-e5d43d1e9ca2'
+            }
+
+        def raise_for_status(self):
+            if self.status_code == 400:
+                raise requests.exceptions.HTTPError()
+
+    json = kwargs['json']
+    params = json['params']
+    url = params['url']
+
+    if url.startswith('/app/'):
+        return MockResponse(200)
+    else:
+        return MockResponse(400)
+
+
 class ShortenUrlTestCase:
     def __init__(
          self,
@@ -200,6 +224,115 @@ def test_short_kinbana_external_url_formatter(
         base_url=test_case.base_url,
         auth=test_case.authorization,
         security_tenant=test_case.security_tenant,
+        new_shortener=False,
+        id='unused',
+    )
+
+    actualUrl = formatter.format(test_case.relative_url)
+    assert actualUrl == test_case.expected_url
+
+    mock_post.assert_called_once_with(**test_case.expected_api_request)
+
+
+@mock.patch('requests.post', side_effect=mock_7_16_kibana_shorten_url_api)
+@pytest.mark.parametrize("test_case", [
+
+    # Relative to kibana plugin
+    ShortenUrlTestCase(
+        base_url='http://elasticsearch.test.org/_plugin/kibana/',
+        relative_url='app/dev_tools#/console',
+        expected_api_request={
+            'url': 'http://elasticsearch.test.org/_plugin/kibana/api/short_url',
+            'auth': None,
+            'headers': {
+                'kbn-xsrf': 'elastalert',
+                'osd-xsrf': 'elastalert'
+            },
+            'json': {
+                'locatorId': 'elastalert-some_unique_id',
+                'params': {
+                    'url': '/app/dev_tools#/console'
+                }
+            }
+        },
+        expected_url='http://elasticsearch.test.org/_plugin/kibana/goto/a1f77a80-6847-11ec-9b91-e5d43d1e9ca2'
+    ),
+
+    # Relative to root of dedicated Kibana domain
+    ShortenUrlTestCase(
+        base_url='http://kibana.test.org/',
+        relative_url='/app/dev_tools#/console',
+        expected_api_request={
+            'url': 'http://kibana.test.org/api/short_url',
+            'auth': None,
+            'headers': {
+                'kbn-xsrf': 'elastalert',
+                'osd-xsrf': 'elastalert'
+            },
+            'json': {
+                'locatorId': 'elastalert-some_unique_id',
+                'params': {
+                    'url': '/app/dev_tools#/console'
+                }
+            }
+        },
+        expected_url='http://kibana.test.org/goto/a1f77a80-6847-11ec-9b91-e5d43d1e9ca2'
+    ),
+
+    # With authentication
+    ShortenUrlTestCase(
+        base_url='http://kibana.test.org/',
+        auth=HTTPBasicAuth('john', 'doe'),
+        relative_url='/app/dev_tools#/console',
+        expected_api_request={
+            'url': 'http://kibana.test.org/api/short_url',
+            'auth': HTTPBasicAuth('john', 'doe'),
+            'headers': {
+                'kbn-xsrf': 'elastalert',
+                'osd-xsrf': 'elastalert'
+            },
+            'json': {
+                'locatorId': 'elastalert-some_unique_id',
+                'params': {
+                    'url': '/app/dev_tools#/console'
+                }
+            }
+        },
+        expected_url='http://kibana.test.org/goto/a1f77a80-6847-11ec-9b91-e5d43d1e9ca2'
+    ),
+
+    # With security tenant
+    ShortenUrlTestCase(
+        base_url='http://kibana.test.org/',
+        security_tenant='global',
+        relative_url='/app/dev_tools#/console',
+        expected_api_request={
+            'url': 'http://kibana.test.org/api/short_url?security_tenant=global',
+            'auth': None,
+            'headers': {
+                'kbn-xsrf': 'elastalert',
+                'osd-xsrf': 'elastalert'
+            },
+            'json': {
+                'locatorId': 'elastalert-some_unique_id',
+                'params': {
+                    'url': '/app/dev_tools?security_tenant=global#/console'
+                }
+            }
+        },
+        expected_url='http://kibana.test.org/goto/a1f77a80-6847-11ec-9b91-e5d43d1e9ca2?security_tenant=global'
+    )
+])
+def test_7_16_short_kibana_external_url_formatter(
+    mock_post: mock.MagicMock,
+    test_case: ShortenUrlTestCase
+):
+    formatter = ShortKibanaExternalUrlFormatter(
+        base_url=test_case.base_url,
+        auth=test_case.authorization,
+        security_tenant=test_case.security_tenant,
+        new_shortener=True,
+        id='some_unique_id',
     )
 
     actualUrl = formatter.format(test_case.relative_url)
@@ -214,6 +347,8 @@ def test_short_kinbana_external_url_formatter_request_exception(mock_post: mock.
         base_url='http://kibana.test.org',
         auth=None,
         security_tenant=None,
+        new_shortener=False,
+        id='unused',
     )
     with pytest.raises(EAException, match="Failed to invoke Kibana Shorten URL API"):
         formatter.format('http://wacky.org')