fix(types): also expand IPv6 CIDRs to the uncompressed form written by Sysmon

sigma/types.py

@@ -4,7 +4,7 @@
 from abc import ABC, abstractmethod
 from dataclasses import InitVar, dataclass, field
 from enum import Enum, auto
-from ipaddress import IPv4Network, IPv6Network, ip_network
+from ipaddress import IPv4Network, IPv6Address, IPv6Network, ip_network
 from itertools import product
 from math import inf, isfinite
 from typing import (
@@ -1004,8 +1004,67 @@ def expand(
                     for other in patterns
                 )
             ]
+            # Some log sources, e.g. Sysmon network events, write IPv6 addresses uncompressed.
+            # Add patterns for this representation that are not already covered by the RFC 5952
+            # patterns above.
+            uncompressed: list[str] = []
+            for subnet_v6 in self.network.subnets(prefix_diff):
+                for pattern in self._expand_ipv6_subnet_uncompressed(subnet_v6, wildcard):
+                    if pattern not in uncompressed:
+                        uncompressed.append(pattern)
+            patterns += [
+                pattern
+                for pattern in uncompressed
+                if pattern not in patterns
+                and not any(
+                    other != pattern
+                    and other.endswith(wildcard)
+                    and pattern.startswith(other[: -len(wildcard)])
+                    for other in patterns + uncompressed
+                )
+            ]
         return patterns
 
+    @staticmethod
+    def _ipv6_groups(address: IPv6Address) -> list[str]:
+        """
+        Return the eight groups of an IPv6 address as four-digit hexadecimal strings.
+
+        IPv6Address.exploded can't be used for this, because current Python versions write the
+        last 32 bits of IPv4-mapped addresses in dotted notation.
+        """
+        digits = format(int(address), "032x")
+        return [digits[i : i + 4] for i in range(0, 32, 4)]
+
+    @staticmethod
+    def _expand_ipv6_subnet_uncompressed(subnet: IPv6Network, wildcard: str) -> list[str]:
+        """
+        Generate wildcard patterns that match the uncompressed textual representation written
+        by some log sources, e.g. Sysmon network events (all eight groups, lower case, no
+        leading zeros, no :: compression, e.g. 0:0:0:0:0:0:0:1 for ::1), of all addresses of an
+        IPv6 network whose prefix length is a multiple of 4. Groups are handled like in
+        _expand_ipv6_subnet.
+        """
+        fixed_groups, rem = divmod(subnet.prefixlen, 16)
+        fixed_nibbles = rem // 4
+        exploded = SigmaCIDRExpression._ipv6_groups(subnet.network_address)
+        fixed = [format(int(group, 16), "x") for group in exploded[:fixed_groups]]
+        if fixed_groups == 8:  # single address
+            return [":".join(fixed)]
+
+        if fixed_nibbles == 0:  # first group not covered by prefix
+            partial = [wildcard]
+        else:  # group partially covered by prefix
+            nibbles = exploded[fixed_groups][:fixed_nibbles]
+            free_digits = 4 - fixed_nibbles
+            if int(nibbles, 16) != 0:
+                partial = [nibbles.lstrip("0") + "?" * free_digits]
+            else:
+                partial = ["?" * n for n in range(1, free_digits + 1)]
+        # Everything after the first group not covered completely by the prefix is a wildcard.
+        rest = [wildcard] if fixed_nibbles > 0 and fixed_groups < 7 else []
+        return [":".join(fixed + [group] + rest) for group in partial]
+
     @staticmethod
     def _expand_ipv6_subnet(subnet: IPv6Network, wildcard: str) -> list[str]:
         """
@@ -1030,7 +1089,7 @@ def _expand_ipv6_subnet(subnet: IPv6Network, wildcard: str) -> list[str]:
 
         fixed_groups, rem = divmod(subnet.prefixlen, 16)
         fixed_nibbles = rem // 4
-        exploded = subnet.network_address.exploded.split(":")
+        exploded = SigmaCIDRExpression._ipv6_groups(subnet.network_address)
 
         # For each group: list of possible textual representations if the group is not part of
         # the compressed zero run and whether the group can be zero / non-zero.

tests/test_types.py

@@ -922,26 +922,62 @@ def test_cidr_expand_ipv6_64_compressed():
 
 
 def test_cidr_expand_ipv6_112():
-    assert SigmaCIDRExpression("2001:db8::/112").expand() == ["2001:db8::*"]
+    assert SigmaCIDRExpression("2001:db8::/112").expand() == [
+        "2001:db8::*",
+        "2001:db8:0:0:0:0:0:*",
+    ]
 
 
 def test_cidr_expand_ipv6_120():
     assert SigmaCIDRExpression("2001:db8::/120").expand() == [
         "2001:db8::?",
         "2001:db8::??",
         "2001:db8::",
+        "2001:db8:0:0:0:0:0:?",
+        "2001:db8:0:0:0:0:0:??",
     ]
 
 
 def test_cidr_expand_ipv6_128_compressed():
-    assert SigmaCIDRExpression("2001:db8::1/128").expand() == ["2001:db8::1"]
+    assert SigmaCIDRExpression("2001:db8::1/128").expand() == [
+        "2001:db8::1",
+        "2001:db8:0:0:0:0:0:1",
+    ]
+
+
+def test_cidr_expand_ipv6_128_not_compressed():
+    assert SigmaCIDRExpression("2001:db8:1:2:3:4:5:6/128").expand() == ["2001:db8:1:2:3:4:5:6"]
+
+
+@pytest.mark.parametrize(
+    "cidr,expected",
+    [
+        ("::1/128", ["::1", "0:0:0:0:0:0:0:1"]),
+        ("::ffff:0:0/96", ["::ffff:*:*", "0:0:0:0:0:ffff:*"]),
+        (
+            "::ffff:0:0/104",
+            ["::ffff:?:*", "::ffff:??:*", "0:0:0:0:0:ffff:?:*", "0:0:0:0:0:ffff:??:*"],
+        ),
+        ("64:ff9b::/96", ["64:ff9b::*", "64:ff9b:0:0:0:0:*"]),
+        ("100::/64", ["100::*", "100:0:0:0:*"]),
+        # uncompressed form already covered by the RFC 5952 patterns
+        ("2001:db8::/64", ["2001:db8:0:0:*", "2001:db8::*"]),
+        ("fe80::/10", ["fe8?:*", "fe9?:*", "fea?:*", "feb?:*"]),
+    ],
+)
+def test_cidr_expand_ipv6_uncompressed(cidr, expected):
+    """Sysmon network events write IPv6 addresses without :: compression."""
+    assert SigmaCIDRExpression(cidr).expand() == expected
 
 
 @pytest.mark.parametrize(
     "cidr,expected",
     [
         # zero groups compressed in network but not in all addresses
-        ("2001:db8:0:0:1::/80", ["2001:db8::1:*:*:*", "2001:db8:0:0:1::"]),
+        (
+            "2001:db8:0:0:1::/80",
+            ["2001:db8::1:*:*:*", "2001:db8:0:0:1::", "2001:db8:0:0:1:*"],
+        ),
         ("2001:0:0:1::/64", ["2001::1:*:*:*:*", "2001:0:0:1:*"]),
         # leading zeros are not written, partially covered groups need an exact digit count
         ("2001:db8:1000::/36", ["2001:db8:1???:*"]),
@@ -995,6 +1031,60 @@ def test_cidr_expand_ipv6_matches(cidr, inside, outside):
         assert not any(fnmatchcase(address, pattern) for pattern in patterns), address
 
 
+def test_cidr_expand_ipv6_128_ipv4_mapped_uncompressed():
+    """Current Python versions write the last groups of IPv4-mapped addresses as dotted quad."""
+    assert "0:0:0:0:0:ffff:102:304" in SigmaCIDRExpression("::ffff:102:304/128").expand()
+
+
+def _uncompressed(address):
+    """Uncompressed representation, as written by Sysmon: 0:0:0:0:0:0:0:1 for ::1"""
+    return ":".join(format(int(group, 16), "x") for group in address.exploded.split(":"))
+
+
+@pytest.mark.parametrize(
+    "cidr,inside,outside",
+    [
+        ("::1/128", ["::1"], ["::", "::2", "::1:0", "1::1"]),
+        ("2620:1ec:4::152/128", ["2620:1ec:4::152"], ["2620:1ec:4::153", "2620:1ec:4:0:1::152"]),
+        (
+            "64:ff9b::/96",
+            ["64:ff9b::", "64:ff9b::102:304", "64:ff9b::ffff:ffff"],
+            ["64:ff9b:1::", "64:ff9a::102:304", "::102:304"],
+        ),
+        (
+            "2001:db8:0:0:1::/80",
+            ["2001:db8::1:0:0:1", "2001:db8::1:a:b:c", "2001:db8:0:0:1::"],
+            ["2001:db8:ffff::1", "2001:db8::2:0:0:1", "2001:db8::1", "2001:db8:1::1"],
+        ),
+        (
+            "2001:db8::/120",
+            ["2001:db8::", "2001:db8::1", "2001:db8::ff"],
+            ["2001:db8::100", "2001:db8::1:0", "2001:db8:1::"],
+        ),
+        (
+            "2001:db8:1000::/36",
+            ["2001:db8:1000::", "2001:db8:1fff:ffff::1", "2001:db8:1abc:1:2:3:4:5"],
+            ["2001:db8:1::", "2001:db8:1:2:3:4:5:6", "2001:db8:160:1:2:3:4:5", "2001:db8:2000::"],
+        ),
+    ],
+)
+def test_cidr_expand_ipv6_matches_uncompressed(cidr, inside, outside):
+    """
+    Patterns match the RFC 5952 and the uncompressed representation of addresses inside the
+    network and reject both representations of the listed outside addresses.
+    """
+    patterns = SigmaCIDRExpression(cidr).expand()
+    network = IPv6Network(cidr)
+    for address in inside:
+        assert IPv6Address(address) in network
+        for text in (str(IPv6Address(address)), _uncompressed(IPv6Address(address))):
+            assert any(fnmatchcase(text, pattern) for pattern in patterns), text
+    for address in outside:
+        assert IPv6Address(address) not in network
+        for text in (str(IPv6Address(address)), _uncompressed(IPv6Address(address))):
+            assert not any(fnmatchcase(text, pattern) for pattern in patterns), text
+
+
 def test_cidr_expand_ipv6_without_wildcard():
     assert SigmaCIDRExpression("2001:db8::/64").expand(wildcard=None) == ["2001:db8::/64"]
 
@@ -1021,7 +1111,7 @@ def test_cidr_expand_ipv6_64():
 
 
 def test_cidr_expand_ipv6_128():
-    assert SigmaCIDRExpression("::1/128").expand() == ["::1"]
+    assert SigmaCIDRExpression("::1/128").expand() == ["::1", "0:0:0:0:0:0:0:1"]
 
 
 def test_cidr_invalid():