B1157: Multi-type registry events

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "pySigma-backend-uberAgent"
-version = "0.3.69"
+version = "0.3.70"
 description = "pySigma uAQL backend"
 authors = ["vast limits GmbH <info@vastlimits.com>"]
 license = "MIT"

sigma/backends/uberagent/rule.py

@@ -178,7 +178,10 @@ def __str__(self):
             result += "RuleId = {}\n".format(self.id)
 
         result += "RuleName = {}\n".format(self.name)
-        result += "EventType = {}\n".format(self.event_type)
+        # split the event types by comma
+        event_types = [et.strip() for et in self.event_type.split(",")]
+        for et in event_types:
+            result += "EventType = {}\n".format(et)
         result += "Tag = {}\n".format(self._prefixed_tag())
 
         # The RiskScore is optional.
@@ -192,7 +195,11 @@ def __str__(self):
 
         result += "Query = {}\n".format(self.query)
 
-        if self.event_type == "Reg.Any":
+        # check if any of the event types is a registry event
+        if any(et.startswith("Reg.") for et in event_types):
+            if not all(et.startswith("Reg.") for et in event_types):
+                raise MalformedRuleException("Mixed event types in rule ID={}".format(self.id))
+
             result += "Hive = HKLM,HKU\n"
 
         # uberagent supports generic properties to be added to an activity rule since version 6.1

sigma/backends/uberagent/uberagent.py

@@ -286,6 +286,11 @@ def __init__(self,
         # Regular expressions are handled differntly in current development version.
         if ua_backend_version.is_version_7_2_or_newer():
             self.re_expression = '{field} regex "{regex}"'
+
+        if ua_backend_version.is_version_7_6_or_newer():
+            self.startswith_expression = "istartswith({field}, {value})"
+            self.endswith_expression = "iendswith({field}, {value})"
+            self.contains_expression = "icontains({field}, {value})"
 
         super().__init__(processing_pipeline, collect_errors)
 

sigma/pipelines/uberagent/__init__.py

@@ -1,4 +1,4 @@
-from .uberagent import uberagent, uberagent600, uberagent610, uberagent620, uberagent700, uberagent710, uberagent720, uberagent730, uberagent740, uberagent750, uberagent_develop
+from .uberagent import uberagent, uberagent600, uberagent610, uberagent620, uberagent700, uberagent710, uberagent720, uberagent730, uberagent740, uberagent750, uberagent760, uberagent_develop
 
 pipelines = {
     "uberagent": uberagent,
@@ -11,5 +11,6 @@
     "uberagent-7.3.0": uberagent730,
     "uberagent-7.4.0": uberagent740,
     "uberagent-7.5.0": uberagent750,
+    "uberagent-7.6.0": uberagent760,
     "uberagent-develop": uberagent_develop
 }

sigma/pipelines/uberagent/uberagent.py

@@ -16,7 +16,7 @@
 from sigma.pipelines.uberagent.transformation import ChangeLogsourceCategoryTransformation, FieldMappingTransformationLowercase, \
     FieldDetectionItemFailureTransformation, ReferencedFieldTransformation
 from sigma.pipelines.uberagent.version import UA_VERSION_6_0, UA_VERSION_6_1, UA_VERSION_6_2, UA_VERSION_7_0, \
-    UA_VERSION_7_1, UA_VERSION_7_2, UA_VERSION_7_3, UA_VERSION_7_4, UA_VERSION_7_5, UA_VERSION_DEVELOP, UA_VERSION_CURRENT_RELEASE, Version
+    UA_VERSION_7_1, UA_VERSION_7_2, UA_VERSION_7_3, UA_VERSION_7_4, UA_VERSION_7_5, UA_VERSION_7_6, UA_VERSION_DEVELOP, UA_VERSION_CURRENT_RELEASE, Version
 
 # Maps all known Sigma fields to uberAgent Process Event Properties
 # Note: The process properties are re-usable for all event types as all events are linked to a process.
@@ -333,7 +333,32 @@ def logsource_common_firewall() -> LogsourceCondition:
                  logsource_windows_registry_add(),
                  logsource_windows_registry_delete(),
                  logsource_windows_registry_set()
-             ],
+              ],
+              fields=ua_registry_event_mapping),
+
+    Logsource(UA_VERSION_7_6,
+              "Reg.Key.Create,Reg.Value.Write,Reg.Key.Delete,Reg.Value.Delete,Reg.Key.Rename,Reg.Key.Load,Reg.Key.Unload,Reg.Key.Save,Reg.Key.Restore,Reg.Key.Replace",
+              conditions=[
+                 logsource_windows_registry_event(),
+              ],
+              fields=ua_registry_event_mapping),
+
+    Logsource(UA_VERSION_7_6, "Reg.Key.Create,Reg.Value.Write,Reg.Key.Rename,Reg.Key.Restore,Reg.Key.Replace",
+              conditions=[
+                 logsource_windows_registry_add(),
+              ],
+              fields=ua_registry_event_mapping),
+
+    Logsource(UA_VERSION_7_6, "Reg.Delete",
+              conditions=[
+                 logsource_windows_registry_delete(),
+              ],
+              fields=ua_registry_event_mapping),
+
+    Logsource(UA_VERSION_7_6, "Reg.Value.Write,Reg.Key.Rename,Reg.Key.Restore,Reg.Key.Replace",
+              conditions=[
+                 logsource_windows_registry_set()
+              ],
               fields=ua_registry_event_mapping),
 
     #
@@ -493,8 +518,13 @@ def make_pipeline(uaVersion: Version):
     # A list to hold all processing items for the version-specific pipeline.
     items: List[ProcessingItem] = []
 
+    # Sort categories by version in descending order to prioritize newer versions.
+    # Conflicting/duplicate categories with different version will be dropped.
+    ua_categoriesByVersion = list(ua_categories)
+    ua_categoriesByVersion.sort(key=lambda c: c.version, reverse=True)
+
     # Iterate over the defined categories for uberAgent.
-    for category in ua_categories:
+    for category in ua_categoriesByVersion:
 
         # If the current version of uberAgent doesn't support the event type,
         # skip the rest of the loop.
@@ -635,6 +665,14 @@ def uberagent750() -> ProcessingPipeline:
     """
     return make_pipeline(Version(UA_VERSION_7_5))
 
+def uberagent760() -> ProcessingPipeline:
+    """
+    Create a processing pipeline for version 7.6 of uberAgent.
+
+    Returns:
+    - ProcessingPipeline: The assembled processing pipeline for version 7.6.
+    """
+    return make_pipeline(Version(UA_VERSION_7_6))
 
 def uberagent_develop() -> ProcessingPipeline:
     """

sigma/pipelines/uberagent/version.py

@@ -13,6 +13,7 @@
 UA_VERSION_7_3 = "7.3.0"
 UA_VERSION_7_4 = "7.4.0"
 UA_VERSION_7_5 = "7.5.0"
+UA_VERSION_7_6 = "7.6.0"
 
 # Represents the next upcoming version
 UA_VERSION_DEVELOP = "develop"
@@ -75,6 +76,9 @@ def is_version_7_4_or_newer(self) -> bool:
 
     def is_version_7_5_or_newer(self) -> bool:
         return self.is_version_develop() or self._version() >= self._version_tuple(UA_VERSION_7_5)
+
+    def is_version_7_6_or_newer(self) -> bool:
+        return self.is_version_develop() or self._version() >= self._version_tuple(UA_VERSION_7_6)
 
     def is_version_develop(self) -> bool:
         return self._outputVersion == UA_VERSION_DEVELOP
@@ -155,6 +159,12 @@ def is_field_supported(self, field: Field) -> bool:
         if self.is_version_7_4_or_newer() and field.version == UA_VERSION_7_4:
             return True
 
+        if self.is_version_7_5_or_newer() and field.version == UA_VERSION_7_5:
+            return True
+
+        if self.is_version_7_6_or_newer() and field.version == UA_VERSION_7_6:
+            return True
+
         if self.is_version_develop() and field.version == UA_VERSION_DEVELOP:
             return True
 
@@ -196,6 +206,12 @@ def is_logsource_supported(self, logsource: Logsource) -> bool:
         if self.is_version_7_4_or_newer() and logsource.version == UA_VERSION_7_4:
             return True
 
+        if self.is_version_7_5_or_newer() and logsource.version == UA_VERSION_7_5:
+            return True
+
+        if self.is_version_7_6_or_newer() and logsource.version == UA_VERSION_7_6:
+            return True
+
         if self.is_version_develop() and logsource.version == UA_VERSION_DEVELOP:
             return True
 

tests/test_backend_uberAgent.py

@@ -7,6 +7,10 @@
 def uberAgent_backend():
     return uberagent()
 
+@pytest.fixture
+def uberAgent76_backend():
+    return uberagent(backend_version="7.6.0")
+
 
 def test_uberAgent_and_expression(uberAgent_backend: uberagent):
     assert uberAgent_backend.convert(
@@ -105,6 +109,25 @@ def test_uberAgent_in_expression(uberAgent_backend: uberagent):
     ) == ['fieldA == "valueA" or fieldA == "valueB" or fieldA like r"valueC%"']
 
 
+def test_uberAgent76_in_expression(uberAgent76_backend: uberagent):
+    assert uberAgent76_backend.convert(
+        SigmaCollection.from_yaml("""
+            title: Test
+            status: test
+            logsource:
+                category: test_category
+                product: test_product
+            detection:
+                sel:
+                    fieldA:
+                        - valueA
+                        - valueB
+                        - valueC*
+                condition: sel
+        """)
+    ) == ['fieldA == "valueA" or fieldA == "valueB" or istartswith(fieldA, "valueC")']
+
+
 def test_uberAgent_cidr_query(uberAgent_backend: uberagent):
     assert uberAgent_backend.convert(
         SigmaCollection.from_yaml("""
@@ -121,6 +144,22 @@ def test_uberAgent_cidr_query(uberAgent_backend: uberagent):
     ) == ['field like r"192.168.%"']
 
 
+def test_uberAgent76_cidr_query(uberAgent76_backend: uberagent):
+    assert uberAgent76_backend.convert(
+        SigmaCollection.from_yaml("""
+            title: Test
+            status: test
+            logsource:
+                category: test_category
+                product: test_product
+            detection:
+                sel:
+                    field|cidr: 192.168.0.0/16
+                condition: sel
+        """)
+    ) == ['istartswith(field, "192.168.")']
+
+
 def test_uberAgent_null_query(uberAgent_backend: uberagent):
     assert uberAgent_backend.convert(
         SigmaCollection.from_yaml("""

tests/test_pipelines_uberAgent.py

@@ -4,7 +4,7 @@
 
 from sigma.backends.uberagent import uberagent
 from sigma.backends.uberagent.exceptions import MissingPropertyException, MissingFunctionException
-from sigma.pipelines.uberagent import uberagent as uberagent_pipeline, uberagent600, uberagent610, uberagent620, uberagent700, uberagent710, uberagent720, uberagent730, uberagent740, uberagent_develop
+from sigma.pipelines.uberagent import uberagent as uberagent_pipeline, uberagent600, uberagent610, uberagent620, uberagent700, uberagent710, uberagent720, uberagent730, uberagent740, uberagent750, uberagent760, uberagent_develop
 
 
 def test_ua_windows():
@@ -569,3 +569,67 @@ def test_uberagent_registry_unsupported_renamevalue():
                         EventType: RenameValue
                     condition: sel
             """))
+
+
+def test_reg_any_uA75():
+    expected = \
+        '[ThreatDetectionRule platform=Windows]\n' \
+        '# This is a test rule.\n' \
+        'RuleId = 01234567-1234-5678-1234-567890123456\n' \
+        'RuleName = Test\n' \
+        'EventType = Reg.Any\n' \
+        'Tag = test\n' \
+        'Query = Process.Path == "test" and Process.CommandLine == "test"\n' \
+        'Hive = HKLM,HKU\n'
+
+    assert uberagent(processing_pipeline=uberagent750()).convert(
+        SigmaCollection.from_yaml("""
+            id: 01234567-1234-5678-1234-567890123456
+            title: Test
+            description: This is a test rule.
+            status: test
+            logsource:
+                product: windows
+                category: registry_event
+            detection:
+                sel:
+                    Image: test
+                    CommandLine: test
+                condition: sel
+        """), "conf") == [expected]
+
+def test_reg_event_uA76():
+    expected = \
+        '[ThreatDetectionRule platform=Windows]\n' \
+        '# This is a test rule.\n' \
+        'RuleId = 01234567-1234-5678-1234-567890123456\n' \
+        'RuleName = Test\n' \
+        'EventType = Reg.Key.Create\n' \
+        'EventType = Reg.Value.Write\n' \
+        'EventType = Reg.Key.Delete\n' \
+        'EventType = Reg.Value.Delete\n' \
+        'EventType = Reg.Key.Rename\n' \
+        'EventType = Reg.Key.Load\n' \
+        'EventType = Reg.Key.Unload\n' \
+        'EventType = Reg.Key.Save\n' \
+        'EventType = Reg.Key.Restore\n' \
+        'EventType = Reg.Key.Replace\n' \
+        'Tag = test\n' \
+        'Query = Process.Path == "test" and Process.CommandLine == "test"\n' \
+        'Hive = HKLM,HKU\n'
+
+    assert uberagent(processing_pipeline=uberagent760()).convert(
+        SigmaCollection.from_yaml("""
+            id: 01234567-1234-5678-1234-567890123456
+            title: Test
+            description: This is a test rule.
+            status: test
+            logsource:
+                product: windows
+                category: registry_event
+            detection:
+                sel:
+                    Image: test
+                    CommandLine: test
+                condition: sel
+        """), "conf") == [expected]