Beta 0.9.1

.github/workflows/release.yml

@@ -8,7 +8,7 @@ on:
 
 jobs:
   build-and-publish:
-    runs-on: ubuntu-20.04
+    runs-on: ubuntu-24.04
     environment: release
     permissions:
       id-token: write

.github/workflows/test.yml

@@ -10,8 +10,8 @@ jobs:
   test:
     strategy:
       matrix:
-        os: ["ubuntu-20.04", "windows-2019", "macos-12"]
-        python-version: ["3.8", "3.9", "3.10", "3.11"]
+        os: ["ubuntu-24.04", "windows-2022", "macos-14"]
+        python-version: ["3.9", "3.10", "3.11", "3.12", "3.13"]
     runs-on: ${{ matrix.os }}
     steps:
       - uses: actions/checkout@v4

README.md

@@ -15,6 +15,8 @@ TBD
 
 All the data value are in the config.py
 
+To use a local json verison, you need to put them in a `validator_json` folder visible from the launch directory.
+
 # 📜 Maintainer
 
 This pipelines is currently maintained by:

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "pySigma-validators-sigmahq"
-version = "0.9.0"
+version = "0.9.1"
 description = "pySigma SigmaHQ validators"
 authors = ["François Hubaut <[email protected]>"]
 license = "LGPL-2.1-only"
@@ -11,7 +11,7 @@ packages = [
 ]
 
 [tool.poetry.dependencies]
-python = "^3.8"
+python = "^3.9"
 pysigma = "^0.11"
 
 [tool.poetry.group.dev.dependencies]

sigma/validators/sigmahq/config.py

@@ -4,14 +4,18 @@
 from sigma.rule import SigmaLogSource
 import json
 import os
+import pathlib
+import requests
 
 
 def load_remote_json(url: str, filename: str) -> dict:
-    # url to check
-    # else
     full_name = os.getcwd() + "/validator_json/" + filename
-    with open(full_name, "r", encoding="UTF-8") as file:
-        json_dict = json.load(file)
+    if pathlib.Path(full_name).exists():
+        with open(full_name, "r", encoding="UTF-8") as file:
+            json_dict = json.load(file)
+    else:
+        r = requests.get(url)
+        json_dict = json.loads(r.content.decode(encoding="UTF-8"))
     return json_dict
 
 
@@ -21,27 +25,34 @@ def core_logsource(source: SigmaLogSource) -> SigmaLogSource:
     )
 
 
-def load_taxonomy_json(json_name: str) -> dict:
-    field_info = {}
-    json_dict = load_remote_json("github", json_name)
+def load_taxonomy_json(url: str, json_name: str) -> dict:
+    json_dict = load_remote_json(url, json_name)
+    info = {}
     for value in json_dict["taxonomy"].values():
         logsource = core_logsource(SigmaLogSource.from_dict(value["logsource"]))
-        field_info[logsource] = value["field"]["natif"]
-        field_info[logsource].extend(value["field"]["custom"])
+        info[logsource] = value
+    return info
+
+
+def get_taxonomy_field(sigma: dict) -> dict:
+    field_info = {}
+    for key, value in sigma.items():
+        field_info[key] = value["field"]["natif"]
+        field_info[key].extend(value["field"]["custom"])
     return field_info
 
 
-def load_filepattern_json(json_name):
+def load_filepattern_json(url: str, json_name: str):
     prefix_info = {}
-    json_dict = load_remote_json("github", json_name)
+    json_dict = load_remote_json(url, json_name)
     for value in json_dict["pattern"].values():
         logsource = core_logsource(SigmaLogSource.from_dict(value["logsource"]))
         prefix_info[logsource] = value["prefix"]
     return prefix_info
 
 
-def load_windows_json(json_name):
-    json_dict = load_remote_json("github", json_name)
+def load_windows_json(url: str, json_name: str):
+    json_dict = load_remote_json(url, json_name)
     data = dict()
     for category in json_dict["category_provider_name"]:
         data[SigmaLogSource(product="windows", category=category, service=None)] = (
@@ -51,33 +62,30 @@ def load_windows_json(json_name):
 
 
 class ConfigHQ:
-    title_max_length = 120
-
-    sigma_taxonomy: Dict[SigmaLogSource, List[str]] = {}
-    sigma_taxonomy_unicast: Dict[SigmaLogSource, List[str]] = {}
+    sigma_taxonomy: Dict = {}
+    sigma_fieldsname: Dict[SigmaLogSource, List[str]] = {}
+    sigma_fieldsname_unicast: Dict[SigmaLogSource, List[str]] = {}
 
     sigmahq_logsource_filepattern: Dict[SigmaLogSource, str] = {}
 
     windows_no_eventid: List[str] = []
     windows_provider_name: Dict[SigmaLogSource, List[str]] = {}
 
-    sigmahq_unsupported_regex_group_constructs: Tuple[str] = (
-        "(?=",
-        "(?!",
-        "(?<=",
-        "(?<!",
-        "(?>",
-    )
-
     def __init__(self) -> None:
-        self.sigma_taxonomy = load_taxonomy_json("sigma.json")
-        self.sigma_taxonomy_unicast = {
-            k: [v.lower() for v in l] for k, l in self.sigma_taxonomy.items()
+        self.sigma_taxonomy = load_taxonomy_json(
+            url="https://github.com/SigmaHQ/pySigma-validators-sigmaHQ/raw/refs/heads/main/validator_json/sigma.json",
+            json_name="sigma.json",
+        )
+        self.sigma_fieldsname = get_taxonomy_field(self.sigma_taxonomy)
+        self.sigma_fieldsname_unicast = {
+            k: [v.lower() for v in l] for k, l in self.sigma_fieldsname.items()
         }
 
         self.sigmahq_logsource_filepattern = load_filepattern_json(
-            "sigmahq_filename.json"
+            url="https://github.com/SigmaHQ/pySigma-validators-sigmaHQ/raw/refs/heads/main/validator_json/sigmahq_filename.json",
+            json_name="sigmahq_filename.json",
         )
         self.windows_no_eventid, self.windows_provider_name = load_windows_json(
-            "sigmahq_windows_validator.json"
+            url="https://github.com/SigmaHQ/pySigma-validators-sigmaHQ/raw/refs/heads/main/validator_json/sigmahq_windows_validator.json",
+            json_name="sigmahq_windows_validator.json",
         )

sigma/validators/sigmahq/detection.py

@@ -104,16 +104,22 @@ class SigmahqUnsupportedRegexGroupConstructIssue(SigmaValidationIssue):
 class SigmahqUnsupportedRegexGroupConstructValidator(SigmaDetectionItemValidator):
     """Checks if a rule uses a an unsupported regular expression group constructs."""
 
+    regex_list: Tuple[str] = (
+        "(?=",
+        "(?!",
+        "(?<=",
+        "(?<!",
+        "(?>",
+    )
+
     def validate_detection_item(
         self, detection_item: SigmaDetectionItem
     ) -> List[SigmaValidationIssue]:
         unsupported_regexps: Set[str] = set()
 
         if SigmaRegularExpressionModifier in detection_item.modifiers:
             for value in detection_item.value:
-                for (
-                    unsupported_group_construct
-                ) in config.sigmahq_unsupported_regex_group_constructs:
+                for unsupported_group_construct in self.regex_list:
                     if unsupported_group_construct in value.regexp:
                         unsupported_regexps.add(value.regexp)
 

sigma/validators/sigmahq/field.py

@@ -64,11 +64,11 @@ def validate(self, rule: SigmaRule) -> List[SigmaValidationIssue]:
             rule.logsource.category, rule.logsource.product, rule.logsource.service
         )
         if (
-            core_logsource in config.sigma_taxonomy
-            and len(config.sigma_taxonomy[core_logsource]) > 0
+            core_logsource in config.sigma_fieldsname
+            and len(config.sigma_fieldsname[core_logsource]) > 0
         ):
-            self.fields = config.sigma_taxonomy[core_logsource]
-            self.unifields = config.sigma_taxonomy_unicast[core_logsource]
+            self.fields = config.sigma_fieldsname[core_logsource]
+            self.unifields = config.sigma_fieldsname_unicast[core_logsource]
             return super().validate(rule)
         return []
 
@@ -100,11 +100,11 @@ def validate(self, rule: SigmaRule) -> List[SigmaValidationIssue]:
             rule.logsource.category, rule.logsource.product, rule.logsource.service
         )
         if (
-            core_logsource in config.sigma_taxonomy
-            and len(config.sigma_taxonomy[core_logsource]) > 0
+            core_logsource in config.sigma_fieldsname
+            and len(config.sigma_fieldsname[core_logsource]) > 0
         ):
-            self.fields = config.sigma_taxonomy[core_logsource]
-            self.unifields = config.sigma_taxonomy_unicast[core_logsource]
+            self.fields = config.sigma_fieldsname[core_logsource]
+            self.unifields = config.sigma_fieldsname_unicast[core_logsource]
             return super().validate(rule)
         return []
 
@@ -244,51 +244,50 @@ def validate_detection_item(
 
 
 # Python 3.9 do not have the match
-# @dataclass
-# class SigmahqInvalidHashKvIssue(SigmaValidationIssue):
-#     description: ClassVar[str] = (
-#         "A Sysmon Hash search must be valid Hash_Type=Hash_Value"
-#     )
-#     severity: ClassVar[SigmaValidationIssueSeverity] = SigmaValidationIssueSeverity.HIGH
-#     value: str
-
-
-# class SigmahqInvalidHashKvValidator(SigmaDetectionItemValidator):
-#     """Check field Sysmon Hash Key-Value search is valid."""
-
-#     hash_field: Tuple[str] = ("Hashes", "Hash")
-#     hash_key: Tuple[str] = ("MD5", "SHA1", "SHA256", "IMPHASH")
-
-#     def validate_detection_item(
-#         self, detection_item: SigmaDetectionItem
-#     ) -> List[SigmaValidationIssue]:
-
-#         errors = []
-#         if detection_item.field is not None and detection_item.field in self.hash_field:
-#             for v in detection_item.value:
-#                 if isinstance(v, SigmaString):
-#                     # v.original is empty when use |contains
-#                     for s_value in v.s:
-#                         if isinstance(s_value, str):
-#                             try:
-#                                 hash_name, hash_data = s_value.split("=")
-#                                 if hash_name in self.hash_key:
-#                                     match hash_name:
-#                                         case "MD5":
-#                                             hash_regex = r"^[a-fA-F0-9]{32}$"
-#                                         case "SHA1":
-#                                             hash_regex = r"^[a-fA-F0-9]{40}$"
-#                                         case "SHA256":
-#                                             hash_regex = r"^[a-fA-F0-9]{64}$"
-#                                         case "IMPHASH":
-#                                             hash_regex = r"^[a-fA-F0-9]{32}$"
-#                                     if re.search(hash_regex, hash_data) is None:
-#                                         errors.append(hash_data)
-#                                 else:
-#                                     errors.append(hash_name)
-#                             except ValueError:
-#                                 errors.append(s_value)
-#                 else:
-#                     errors.append(v)
-
-#         return [SigmahqInvalidHashKvIssue(self.rule, v) for v in errors]
+@dataclass
+class SigmahqInvalidHashKvIssue(SigmaValidationIssue):
+    description: ClassVar[str] = (
+        "A Sysmon Hash search must be valid Hash_Type=Hash_Value"
+    )
+    severity: ClassVar[SigmaValidationIssueSeverity] = SigmaValidationIssueSeverity.HIGH
+    value: str
+
+
+class SigmahqInvalidHashKvValidator(SigmaDetectionItemValidator):
+    """Check field Sysmon Hash Key-Value search is valid."""
+
+    hash_field: Tuple[str] = ("Hashes", "Hash")
+    hash_key: Tuple[str] = ("MD5", "SHA1", "SHA256", "IMPHASH")
+
+    def validate_detection_item(
+        self, detection_item: SigmaDetectionItem
+    ) -> List[SigmaValidationIssue]:
+
+        errors = []
+        if detection_item.field is not None and detection_item.field in self.hash_field:
+            for v in detection_item.value:
+                if isinstance(v, SigmaString):
+                    # v.original is empty when use |contains
+                    for s_value in v.s:
+                        if isinstance(s_value, str):
+                            try:
+                                hash_name, hash_data = s_value.split("=")
+                                if hash_name in self.hash_key:
+                                    if hash_name == "MD5":
+                                        hash_regex = r"^[a-fA-F0-9]{32}$"
+                                    elif hash_name == "SHA1":
+                                        hash_regex = r"^[a-fA-F0-9]{40}$"
+                                    elif hash_name == "SHA256":
+                                        hash_regex = r"^[a-fA-F0-9]{64}$"
+                                    elif hash_name == "IMPHASH":
+                                        hash_regex = r"^[a-fA-F0-9]{32}$"
+                                    if re.search(hash_regex, hash_data) is None:
+                                        errors.append(hash_data)
+                                else:
+                                    errors.append(hash_name)
+                            except ValueError:
+                                errors.append(s_value)
+                else:
+                    errors.append(v)
+
+        return [SigmahqInvalidHashKvIssue(self.rule, v) for v in errors]

sigma/validators/sigmahq/logsource.py

@@ -27,7 +27,7 @@ def validate(self, rule: SigmaRule) -> List[SigmaValidationIssue]:
         core_logsource = SigmaLogSource(
             rule.logsource.category, rule.logsource.product, rule.logsource.service
         )
-        if not core_logsource in config.sigma_taxonomy:
+        if not core_logsource in config.sigma_fieldsname:
             return [SigmahqLogsourceUnknownIssue(rule, rule.logsource)]
         else:
             return []
@@ -57,3 +57,33 @@ def validate(self, rule: SigmaRule) -> List[SigmaValidationIssue]:
                 return [SigmahqSysmonMissingEventidIssue(rule)]
         else:
             return []
+
+
+@dataclass
+class SigmahqLogsourceDefinitionIssue(SigmaValidationIssue):
+    description: ClassVar[str] = "Rule uses an unknown logsource definition"
+    severity: ClassVar[SigmaValidationIssueSeverity] = (
+        SigmaValidationIssueSeverity.MEDIUM
+    )
+    logsource: SigmaLogSource
+
+
+class SigmahqLogsourceDefinitionValidator(SigmaRuleValidator):
+    """Checks if a rule uses the unknown logsource definition."""
+
+    def validate(self, rule: SigmaRule) -> List[SigmaValidationIssue]:
+        if rule.logsource.definition:
+            core_logsource = SigmaLogSource(
+                rule.logsource.category, rule.logsource.product, rule.logsource.service
+            )
+            if (
+                core_logsource in config.sigma_taxonomy
+                and config.sigma_taxonomy[core_logsource]["logsource"]["definition"]
+            ):
+                if (
+                    rule.logsource.definition
+                    != config.sigma_taxonomy[core_logsource]["logsource"]["definition"]
+                ):
+                    return [SigmahqLogsourceDefinitionIssue(rule, rule.logsource)]
+
+        return []

sigma/validators/sigmahq/title.py

@@ -14,19 +14,20 @@
 
 @dataclass
 class SigmahqTitleLengthIssue(SigmaValidationIssue):
-    description: ClassVar[str] = (
-        f"Rule has a title longer than {config.title_max_length} characters."
-    )
+    description: ClassVar[str] = "Rule has a title is too large."
     severity: ClassVar[SigmaValidationIssueSeverity] = (
         SigmaValidationIssueSeverity.MEDIUM
     )
 
 
+@dataclass(frozen=True)
 class SigmahqTitleLengthValidator(SigmaRuleValidator):
     """Checks if a rule has an excessively long title."""
 
+    max_length: int = 120
+
     def validate(self, rule: SigmaRule) -> List[SigmaValidationIssue]:
-        if len(rule.title) > config.title_max_length:
+        if len(rule.title) > self.max_length:
             return [SigmahqTitleLengthIssue([rule])]
         else:
             return []

tests/files/rule_filename_errors/Name.yml

@@ -9,7 +9,7 @@
         - attack.execution
         - attack.t1059
     author: Thomas Patzke
-    date: 2020/07/13
+    date: 2020-07-13
     logsource:
         category: process_creation
         product: windows