Add domain-name support for stix patterning and translation (#64)

stix_shifter/src/modules/qradar/aql_query_constructor.py

@@ -93,8 +93,7 @@ def _format_equality(value) -> str:
 
     @staticmethod
     def _format_like(value) -> str:
-        value = value.replace('%', '*')
-        value = value.replace('_', '?')
+        value = "'%{value}%'".format(value=value)
         return AqlQueryStringPatternTranslator._escape_value(value)
 
     @staticmethod
@@ -151,8 +150,14 @@ def _parse_expression(self, expression, qualifier=None) -> str:
                 if expression.comparator == ComparisonComparators.IsSubSet:
                     comparison_string += comparator + "(" + "'" + value + "'," + mapped_field + ")"
                 else:
+                    # There's no aql field for domain-name. using Like operator to find domian name from the url
+                    if mapped_field == 'domainname' and comparator != ComparisonComparators.Like:
+                        comparator = self.comparator_lookup[ComparisonComparators.Like]
+                        value = self._format_like(expression.value)
+
                     comparison_string += "{mapped_field} {comparator} {value}".format(
                         mapped_field=mapped_field, comparator=comparator, value=value)
+
                 if (mapped_fields_count > 1):
                     comparison_string += " OR "
                     mapped_fields_count -= 1

stix_shifter/src/modules/qradar/json/aql_event_fields.json

@@ -26,17 +26,7 @@
     "BASE64(payload) as payload",
     "URL as url",
     "magnitude as magnitude",
-    "Filename as filename"
-  ],
-  "domain-name": [
-    "QIDNAME(qid) as qidname",
-    "CATEGORYNAME(highlevelcategory) as high_level_category_name",
-    "devicetime as devicetime",
-    "sourceip as sourceip",
-    "sourceport as sourceport",
-    "sourcemac as sourcemac",
-    "destinationip as destinationip",
-    "destinationport as destinationport",
-    "destinationmac as destinationmac"
+    "Filename as filename",
+    "URL as domainname"
   ]
 }

stix_shifter/src/modules/qradar/json/from_stix_map.json

@@ -42,5 +42,10 @@
     "fields": {
       "payload_bin": ["payload"]
     }
+  },
+  "domain-name": {
+    "fields": {
+      "value": ["domainname"]
+    }
   }
 }

stix_shifter/src/modules/qradar/json/to_stix_map.json

@@ -112,5 +112,9 @@
     "key": "network-traffic.protocols",
     "object": "nt",
     "transformer": "ToLowercaseArray"
+  },
+  "domainname": {
+    "key": "domain-name.value",
+    "transformer": "ToDomainName"
   }
 }

stix_shifter/src/transformers.py

@@ -2,6 +2,7 @@
 from datetime import datetime, timezone
 import base64
 import re
+from urllib.parse import urlparse
 
 
 class ValueTransformer():
@@ -116,6 +117,18 @@ def transform(obj):
         except ValueError:
             print("Cannot convert input to file name")
 
+class ToDomainName(ValueTransformer):
+    """A value transformer for expected domain name"""
+
+    @staticmethod
+    def transform(url):
+        try:
+            parsed_url = urlparse(url)
+            domain_name = parsed_url.netloc
+            return domain_name
+        except ValueError:
+            print("Cannot convert input to file name")
+
 def get_all_transformers():
     return {"SplunkToTimestamp": SplunkToTimestamp, "EpochToTimestamp": EpochToTimestamp, "ToInteger": ToInteger, "ToString": ToString, "ToLowercaseArray": ToLowercaseArray,
-            "ToBase64": ToBase64, "ToFilePath": ToFilePath, "ToFileName": ToFileName, "StringToBool": StringToBool}
+            "ToBase64": ToBase64, "ToFilePath": ToFilePath, "ToFileName": ToFileName, "StringToBool": StringToBool, "ToDomainName": ToDomainName}

tests/qradar_json_to_stix/test_class.py

@@ -63,7 +63,7 @@ def test_cybox_observables(self):
         source_ip = "fd80:655e:171d:30d4:fd80:655e:171d:30d4"
         destination_ip = "255.255.255.1"
         file_name = "somefile.exe"
-        data = {"sourceip": source_ip, "destinationip": destination_ip, "url": url, "payload": payload, "username": user_id, "protocol": 'TCP', "sourceport": 3000, "destinationport": 2000, "filename": file_name}
+        data = {"sourceip": source_ip, "destinationip": destination_ip, "url": url, "payload": payload, "username": user_id, "protocol": 'TCP', "sourceport": 3000, "destinationport": 2000, "filename": file_name, "domainname": url}
 
         result_bundle = json_to_stix_translator.convert_to_stix(
             data_source, map_data, [data], transformers.get_all_transformers(), options)
@@ -118,7 +118,12 @@ def test_cybox_observables(self):
         assert(curr_obj.keys() == {'type', 'name'})
         assert(curr_obj['name'] == file_name)
 
-        assert(objects.keys() == set(map(str, range(0, 7))))
+        curr_obj = TestTransform.get_first_of_type(objects.values(), 'domain-name')
+        assert(curr_obj is not None), 'domain-name object type not found'
+        assert(curr_obj.keys() == {'type', 'value'})
+        assert(curr_obj['value'] == 'example.com')
+
+        assert(objects.keys() == set(map(str, range(0, 8))))
 
     def test_custom_props(self):
         data = {"logsourceid": 126, "qid": 55500004,

tests/qradar_stix_to_aql/test_class.py

@@ -9,7 +9,7 @@
 endtime as endtime, devicetime as devicetime, sourceip as sourceip, sourceport as sourceport, sourcemac as sourcemac, \
 destinationip as destinationip, destinationport as destinationport, destinationmac as destinationmac, \
 username as username, eventdirection as direction, identityip as identityip, identityhostname as identity_host_name, \
-eventcount as eventcount, PROTOCOLNAME(protocolid) as protocol, BASE64(payload) as payload, URL as url, magnitude as magnitude, Filename as filename"
+eventcount as eventcount, PROTOCOLNAME(protocolid) as protocol, BASE64(payload) as payload, URL as url, magnitude as magnitude, Filename as filename, URL as domainname"
 
 from_statement = " FROM events "
 
@@ -208,3 +208,10 @@ def test_custom_time_limit_and_result_count(self):
         where_statement = "WHERE (sourceip = '192.168.122.83' OR destinationip = '192.168.122.83' OR identityip = '192.168.122.83') limit {} last {} minutes".format(result_limit, timerange)
         parsed_stix = [{'value': '192.168.122.83', 'comparison_operator': '=', 'attribute': 'ipv4-addr:value'}]
         assert query == {'queries': [selections + from_statement + where_statement], 'parsed_stix': parsed_stix}
+
+    def test_domainname_query(self):
+        stix_pattern = "[domain-name:value = 'example.com']"
+        query = shifter.translate('qradar', 'query', '{}', stix_pattern)
+        where_statement = "WHERE domainname LIKE '%example.com%' {} {}".format(default_limit, default_time)
+        parsed_stix = [{'attribute': 'domain-name:value', 'comparison_operator': '=', 'value': 'example.com'}]
+        assert query == {'queries': [selections + from_statement + where_statement], 'parsed_stix': parsed_stix}