Fix: Added more connect tests for ipv6 and cidr modifier

SigmaHQ · Jun 20, 2024 · 23267b2 · 23267b2
1 parent 36d1d41
commit 23267b2
Showing 1 changed file with 37 additions and 1 deletion.
diff --git a/tests/test_backend_elasticsearch_lucene_connect.py b/tests/test_backend_elasticsearch_lucene_connect.py
@@ -141,6 +141,20 @@ def fixture_prepare_es_data():
             verify=False,
             auth=pytest.es_creds,
         )
+        requests.post(
+            f"{pytest.es_url}/test-index/_doc/",
+            json={"ipfield": "fe80:0000:0000:0000:0000:0000:0000:beef"},
+            timeout=120,
+            verify=False,
+            auth=pytest.es_creds,
+        )
+        requests.post(
+            f"{pytest.es_url}/test-index/_doc/",
+            json={"ipfield": "2603:1080:beef::1"},
+            timeout=120,
+            verify=False,
+            auth=pytest.es_creds,
+        )
         requests.post(
             f"{pytest.es_url}/test-index/_doc/",
             json={"ipfield": "10.5.5.5"},
@@ -461,7 +475,7 @@ def test_connect_lucene_regex_query(
         result_dsl = lucene_backend.convert(rule, output_format="dsl_lucene")[0]
         self.query_backend_hits(result_dsl, num_wanted=1)
 
-    def test_connect_lucene_cidr_query(
+    def test_connect_lucene_cidr_v4_query(
         self, prepare_es_data, lucene_backend: LuceneBackend
     ):
         rule = SigmaCollection.from_yaml(
@@ -481,6 +495,28 @@ def test_connect_lucene_cidr_query(
         result_dsl = lucene_backend.convert(rule, output_format="dsl_lucene")[0]
         self.query_backend_hits(result_dsl, num_wanted=1)
 
+    def test_connect_lucene_cidr_v6_query(
+        self, prepare_es_data, lucene_backend: LuceneBackend
+    ):
+        rule = SigmaCollection.from_yaml(
+            """
+                title: Test
+                status: test
+                logsource:
+                    category: test_category
+                    product: test_product
+                detection:
+                    sel:
+                        ipfield|cidr:
+                            - 'fe80::/10'
+                            - '2603:1080::/25'
+                    condition: sel
+            """
+        )
+
+        result_dsl = lucene_backend.convert(rule, output_format="dsl_lucene")[0]
+        self.query_backend_hits(result_dsl, num_wanted=2)
+
     def test_connect_lucene_ip_query(
         self, prepare_es_data, lucene_backend: LuceneBackend
     ):