test_rules.py

#!/usr/bin/env python3
"""
Checks for noncompliance or common errors on all rules

Run using the command
# python test_rules.py
"""

import os
import unittest
import yaml
import re
import string

# from attackcti import attack_client
from colorama import init
from colorama import Fore
import collections


# Old Tests cover by pySigma 0.10.9 and simgma-cli 0.7.10
# Use sigma check --fail-on-error --fail-on-issues --validation-config tests/sigma_cli_conf.yml rules*
#


class TestRules(unittest.TestCase):
    # @classmethod
    # def setUpClass(cls):
    #     print("Calling get_mitre_data()")
    #     # Get Current Data from MITRE ATT&CK®
    #     cls.MITRE_ALL = get_mitre_data()
    #     print("Catched data - starting tests...")

    # MITRE_TECHNIQUE_NAMES = [
    #     "process_injection",
    #     "signed_binary_proxy_execution",
    #     "process_injection",
    # ]  # incomplete list
    # MITRE_TACTICS = [
    #     "initial_access",
    #     "execution",
    #     "persistence",
    #     "privilege_escalation",
    #     "defense_evasion",
    #     "credential_access",
    #     "discovery",
    #     "lateral_movement",
    #     "collection",
    #     "exfiltration",
    #     "command_and_control",
    #     "impact",
    #     "launch",
    # ]
    # # Don't use trademarks in rules - they require non-ASCII characters to be used on we don't want them in our rules
    TRADE_MARKS = {"MITRE ATT&CK", "ATT&CK"}

    path_to_rules = [
        "rules",
        "rules-emerging-threats",
        "rules-placeholder",
        "rules-threat-hunting",
        "rules-compliance",
    ]

    # Helper functions
    def yield_next_rule_file_path(self, path_to_rules: list) -> str:
        for path_ in path_to_rules:
            for root, _, files in os.walk(path_):
                for file in files:
                    if file.endswith(".yml"):
                        yield os.path.join(root, file)

    def get_rule_part(self, file_path: str, part_name: str):
        yaml_dicts = self.get_rule_yaml(file_path)
        for yaml_part in yaml_dicts:
            if part_name in yaml_part.keys():
                return yaml_part[part_name]

        return None

    def get_rule_yaml(self, file_path: str) -> dict:
        data = []

        with open(file_path, encoding="utf-8") as f:
            yaml_parts = yaml.safe_load_all(f)
            for part in yaml_parts:
                data.append(part)

        return data

    # Tests
    def test_legal_trademark_violations(self):
        # See Issue # https://github.com/SigmaHQ/sigma/issues/1028
        files_with_legal_issues = []

        for file in self.yield_next_rule_file_path(self.path_to_rules):
            with open(file, "r", encoding="utf-8") as fh:
                file_data = fh.read()
                for tm in self.TRADE_MARKS:
                    if tm in file_data:
                        files_with_legal_issues.append(file)

        self.assertEqual(
            files_with_legal_issues,
            [],
            Fore.RED
            + "There are rule files which contains a trademark or reference that doesn't comply with the respective trademark requirements - please remove the trademark to avoid legal issues",
        )

    # sigma cli SigmahqFieldDuplicateValueIssue
    # def test_look_for_duplicate_filters(self):
    #     def check_list_or_recurse_on_dict(item, depth: int, special: bool) -> None:
    #         if type(item) == list:
    #             check_if_list_contain_duplicates(item, depth, special)
    #         elif type(item) == dict and depth <= MAX_DEPTH:
    #             for keys, sub_item in item.items():
    #                 if (
    #                     "|base64" in keys or "|re" in keys
    #                 ):  # Covers both "base64" and "base64offset" modifiers, and "re" modifier
    #                     check_list_or_recurse_on_dict(sub_item, depth + 1, True)
    #                 else:
    #                     check_list_or_recurse_on_dict(sub_item, depth + 1, special)

    #     def check_if_list_contain_duplicates(
    #         item: list, depth: int, special: bool
    #     ) -> None:
    #         try:
    #             # We use a list comprehension to convert all the element to lowercase. Since we don't care about casing in SIGMA except for the following modifiers
    #             #   - "base64offset"
    #             #   - "base64"
    #             #   - "re"
    #             if special:
    #                 item_ = item
    #             else:
    #                 item_ = [i.lower() for i in item]
    #             if len(item_) != len(set(item_)):
    #                 # We find the duplicates and then print them to the user
    #                 duplicates = [
    #                     i
    #                     for i, count in collections.Counter(item_).items()
    #                     if count > 1
    #                 ]
    #                 print(
    #                     Fore.RED
    #                     + "Rule {} has duplicate filters {}".format(file, duplicates)
    #                 )
    #                 files_with_duplicate_filters.append(file)
    #         except:
    #             # unhashable types like dictionaries
    #             for sub_item in item:
    #                 if type(sub_item) == dict and depth <= MAX_DEPTH:
    #                     check_list_or_recurse_on_dict(sub_item, depth + 1, special)

    #     MAX_DEPTH = 3
    #     files_with_duplicate_filters = []

    #     for file in self.yield_next_rule_file_path(self.path_to_rules):
    #         detection = self.get_rule_part(file_path=file, part_name="detection")
    #         check_list_or_recurse_on_dict(detection, 1, False)

    #     self.assertEqual(
    #         files_with_duplicate_filters,
    #         [],
    #         Fore.RED + "There are rules with duplicate filters",
    #     )

    #sigma cli SigmahqFieldWithSpaceIssue
    # def test_field_name_with_space(self):
    #     def key_iterator(fields, faulty):
    #         for key, value in fields.items():
    #             if " " in key:
    #                 faulty.append(key)
    #                 print(
    #                     Fore.YELLOW
    #                     + "Rule {} has a space in field name ({}).".format(file, key)
    #                 )
    #             if type(value) == dict:
    #                 key_iterator(value, faulty)

    #     faulty_fieldnames = []
    #     for file in self.yield_next_rule_file_path(self.path_to_rules):
    #         detection = self.get_rule_part(file_path=file, part_name="detection")
    #         key_iterator(detection, faulty_fieldnames)

    #     self.assertEqual(
    #         faulty_fieldnames,
    #         [],
    #         Fore.RED
    #         + "There are rules with an unsupported field name. Spaces are not allowed. (Replace space with an underscore character '_' )",
    #     )

    #sigma cli AllOfThemConditionIssue
    # def test_single_named_condition_with_x_of_them(self):
    #     faulty_detections = []

    #     for file in self.yield_next_rule_file_path(self.path_to_rules):
    #         yaml = self.get_rule_yaml(file_path=file)
    #         detection = self.get_rule_part(file_path=file, part_name="detection")

    #         has_them_in_condition = "them" in detection["condition"]
    #         has_only_one_named_condition = len(detection) == 2
    #         not_multipart_yaml_file = len(yaml) == 1

    #         if (
    #             has_them_in_condition
    #             and has_only_one_named_condition
    #             and not_multipart_yaml_file
    #         ):
    #             faulty_detections.append(file)

    #     self.assertEqual(
    #         faulty_detections,
    #         [],
    #         Fore.RED
    #         + "There are rules using '1/all of them' style conditions but only have one condition",
    #     )

    def test_duplicate_detections(self):
        def compare_detections(detection1: dict, detection2: dict) -> bool:
            # If they have different log sources. They can't be the same
            # We first remove any definitions fields (if there are any) in the logsource to avoid typos
            detection1["logsource"].pop("definition", None)
            detection2["logsource"].pop("definition", None)

            if detection1["logsource"] != detection2["logsource"]:
                return False

            # detections not the same count can't be the same
            if len(detection1) != len(detection2):
                return False

            for named_condition in detection1:
                # don't check timeframes
                if named_condition == "timeframe":
                    continue

                # condition clause must be the same too
                if named_condition == "condition":
                    if detection1["condition"] != detection2["condition"]:
                        return False
                    else:
                        continue

                # Named condition must exist in both rule files
                if named_condition not in detection2:
                    return False

                # can not be the same  if len is not equal
                if len(detection1[named_condition]) != len(detection2[named_condition]):
                    return False

                for condition in detection1[named_condition]:
                    if type(condition) != str:
                        return False

                    if condition not in detection2[named_condition]:
                        return False

                    # We add this check in case of keyword rules. Where no field is used. The parser returns a list instead of a dict
                    # If the 2 list are different that means they aren't the same
                    if (type(detection1[named_condition]) == list) or (
                        type(detection2[named_condition]) == list
                    ):
                        condition_value1 = detection1[named_condition]
                        condition_value2 = detection2[named_condition]
                    else:
                        condition_value1 = detection1[named_condition][condition]
                        condition_value2 = detection2[named_condition][condition]

                    if condition_value1 != condition_value2:
                        return False

            return True

        faulty_detections = []
        files_and_their_detections = {}

        for file in self.yield_next_rule_file_path(self.path_to_rules):
            detection = self.get_rule_part(file_path=file, part_name="detection")
            logsource = self.get_rule_part(file_path=file, part_name="logsource")
            detection["logsource"] = {}
            detection["logsource"].update(logsource)
            yaml = self.get_rule_yaml(file_path=file)

            is_multipart_yaml_file = len(yaml) != 1
            if is_multipart_yaml_file:
                continue

            for key in files_and_their_detections:
                if compare_detections(detection, files_and_their_detections[key]):
                    faulty_detections.append((key, file))

            files_and_their_detections[file] = detection

        self.assertEqual(
            faulty_detections,
            [],
            Fore.YELLOW + "There are rule files with exactly the same detection logic.",
        )

    def test_source_eventlog(self):
        faulty_detections = []

        for file in self.yield_next_rule_file_path(self.path_to_rules):
            detection = self.get_rule_part(file_path=file, part_name="detection")
            detection_str = str(detection).lower()
            if "'source': 'eventlog'" in detection_str:
                faulty_detections.append(file)

        self.assertEqual(
            faulty_detections,
            [],
            Fore.YELLOW
            + "There are detections with 'Source: Eventlog'. This does not add value to the detection.",
        )

    def test_event_id_instead_of_process_creation(self):
        faulty_detections = []
        for file in self.yield_next_rule_file_path(self.path_to_rules):
            with open(file, encoding="utf-8") as f:
                for line in f:
                    if (
                        re.search(r".*EventID: (?:1|4688)\s*$", line)
                        and file not in faulty_detections
                    ):
                        detection = self.get_rule_part(
                            file_path=file, part_name="detection"
                        )
                        if detection:
                            for search_identifier in detection:
                                if isinstance(detection[search_identifier], dict):
                                    for field in detection[search_identifier]:
                                        if "Provider_Name" in field:
                                            if isinstance(
                                                detection[search_identifier][
                                                    "Provider_Name"
                                                ],
                                                list,
                                            ):
                                                for value in detection[
                                                    search_identifier
                                                ]["Provider_Name"]:
                                                    if (
                                                        "Microsoft-Windows-Security-Auditing"
                                                        in value
                                                        or "Microsoft-Windows-Sysmon"
                                                        in value
                                                    ):
                                                        if (
                                                            file
                                                            not in faulty_detections
                                                        ):
                                                            faulty_detections.append(
                                                                file
                                                            )
                                            else:
                                                if (
                                                    "Microsoft-Windows-Security-Auditing"
                                                    in detection[search_identifier][
                                                        "Provider_Name"
                                                    ]
                                                    or "Microsoft-Windows-Sysmon"
                                                    in detection[search_identifier][
                                                        "Provider_Name"
                                                    ]
                                                ):
                                                    if file not in faulty_detections:
                                                        faulty_detections.append(file)

        self.assertEqual(
            faulty_detections,
            [],
            Fore.YELLOW
            + "There are rules still using Sysmon 1 or Event ID 4688. Please migrate to the process_creation category.",
        )

    def test_sysmon_rule_without_eventid(self):
        faulty_rules = []
        for file in self.yield_next_rule_file_path(self.path_to_rules):
            logsource = self.get_rule_part(file_path=file, part_name="logsource")
            if logsource:
                service = logsource.get("service", "")
                if service.lower() == "sysmon":
                    with open(file, encoding="utf-8") as f:
                        found = False
                        for line in f:
                            # might be on a single line or in multiple lines
                            if re.search(r".*EventID:.*$", line):
                                found = True
                                break
                        if not found:
                            faulty_rules.append(file)

        self.assertEqual(
            faulty_rules,
            [],
            Fore.RED
            + "There are rules using sysmon events but with no EventID specified",
        )

    # sigma cli SigmahqFalsepositivesCapitalIssue
    # def test_optional_falsepositives_capital(self):
    #     faulty_rules = []
    #     for file in self.yield_next_rule_file_path(self.path_to_rules):
    #         fps = self.get_rule_part(file_path=file, part_name="falsepositives")
    #         if fps:
    #             for fp in fps:
    #                 # first letter should be capital
    #                 try:
    #                     if fp[0].upper() != fp[0]:
    #                         print(
    #                             Fore.YELLOW
    #                             + "Rule {} defines a falsepositive that does not start with a capital letter: '{}'.".format(
    #                                 file, fp
    #                             )
    #                         )
    #                         faulty_rules.append(file)
    #                 except TypeError as err:
    #                     print("TypeError Exception for rule {}".format(file))
    #                     print("Error: {}".format(err))
    #                     print("Maybe you created an empty falsepositive item?")

    #     self.assertEqual(
    #         faulty_rules,
    #         [],
    #         Fore.RED
    #         + "There are rules with false positives that don't start with a capital letter (e.g. 'unknown' should be 'Unknown')",
    #     )

    # sigma cli SigmahqFalsepositivesBannedWordIssue
    # def test_optional_falsepositives_blocked_content(self):
    #     faulty_rules = []
    #     banned_words = ["none", "pentest", "penetration test"]
    #     common_typos = ["unkown", "ligitimate", "legitim ", "legitimeate"]
    #     for file in self.yield_next_rule_file_path(self.path_to_rules):
    #         fps = self.get_rule_part(file_path=file, part_name="falsepositives")
    #         if fps:
    #             for fp in fps:
    #                 for typo in common_typos:
    #                     if fp == "Unknow" or typo in fp.lower():
    #                         print(
    #                             Fore.YELLOW
    #                             + "Rule {} defines a falsepositive with a common typo: '{}'.".format(
    #                                 file, typo
    #                             )
    #                         )
    #                         faulty_rules.append(file)
    #                 for banned_word in banned_words:
    #                     if banned_word in fp.lower():
    #                         print(
    #                             Fore.YELLOW
    #                             + "Rule {} defines a falsepositive with an invalid reason: '{}'.".format(
    #                                 file, banned_word
    #                             )
    #                         )
    #                         faulty_rules.append(file)

    #     self.assertEqual(
    #         faulty_rules,
    #         [],
    #         Fore.RED
    #         + "There are rules with invalid false positive definitions (e.g. Pentest, None or common typos)",
    #     )

    def test_optional_license(self):
        faulty_rules = []
        for file in self.yield_next_rule_file_path(self.path_to_rules):
            license_str = self.get_rule_part(file_path=file, part_name="license")
            if license_str:
                if not isinstance(license_str, str):
                    print(
                        Fore.YELLOW
                        + "Rule {} has a malformed 'license' (has to be a string).".format(
                            file
                        )
                    )
                    faulty_rules.append(file)

        self.assertEqual(
            faulty_rules,
            [],
            Fore.RED
            + "There are rules with malformed 'license' fields. (has to be a string )",
        )

    # sigma cli SigmaReferencesError
    # def test_references(self):
    #     faulty_rules = []
    #     for file in self.yield_next_rule_file_path(self.path_to_rules):
    #         references = self.get_rule_part(file_path=file, part_name="references")
    #         # Reference field doesn't exist
    #         # if not references:
    #         # print(Fore.YELLOW + "Rule {} has no field 'references'.".format(file))
    #         # faulty_rules.append(file)
    #         if references:
    #             # it exists but isn't a list
    #             if not isinstance(references, list):
    #                 print(
    #                     Fore.YELLOW
    #                     + "Rule {} has a references field that isn't a list.".format(
    #                         file
    #                     )
    #                 )
    #                 faulty_rules.append(file)

    #     self.assertEqual(
    #         faulty_rules,
    #         [],
    #         Fore.RED
    #         + "There are rules with malformed 'references' fields. (has to be a list of values even if it contains only a single value)",
    #     )

    # sigme cli SigmahqLinkDescriptionIssue
    # def test_references_in_description(self):
    #     # This test checks for the presence of a links and special keywords in the "description" field while there is no "references" field.
    #     faulty_rules = []
    #     for file in self.yield_next_rule_file_path(self.path_to_rules):
    #         references = self.get_rule_part(file_path=file, part_name="references")
    #         # Reference field doesn't exist
    #         if not references:
    #             descriptionfield = self.get_rule_part(
    #                 file_path=file, part_name="description"
    #             )
    #             if descriptionfield:
    #                 for i in [
    #                     "http://",
    #                     "https://",
    #                     "internal research",
    #                 ]:  # Extends the list with other common references starters
    #                     if i in descriptionfield.lower():
    #                         print(
    #                             Fore.RED
    #                             + "Rule {} has a field that contains references to external links but no references set. Add a 'references' key and add URLs as list items.".format(
    #                                 file
    #                             )
    #                         )
    #                         faulty_rules.append(file)

    #     self.assertEqual(
    #         faulty_rules,
    #         [],
    #         Fore.RED
    #         + "There are rules with malformed 'description' fields. (links and external references have to be in a seperate field named 'references'. see specification https://github.com/SigmaHQ/sigma-specification)",
    #     )

    def test_file_names(self):
        faulty_rules = []
        name_lst = []
        filename_pattern = re.compile(r"[a-z0-9_]{10,90}\.yml")
        for file in self.yield_next_rule_file_path(self.path_to_rules):
            filename = os.path.basename(file)
            if filename in name_lst:
                print(Fore.YELLOW + "Rule {} is a duplicate file name.".format(file))
                faulty_rules.append(file)
            elif filename[-4:] != ".yml":
                print(
                    Fore.YELLOW + "Rule {} has a invalid extension (.yml).".format(file)
                )
                faulty_rules.append(file)
            elif len(filename) > 90:
                print(
                    Fore.YELLOW + "Rule {} has a file name too long >90.".format(file)
                )
                faulty_rules.append(file)
            elif len(filename) < 14:
                print(
                    Fore.YELLOW + "Rule {} has a file name too short <14.".format(file)
                )
                faulty_rules.append(file)
            elif filename_pattern.match(filename) == None or not "_" in filename:
                print(
                    Fore.YELLOW
                    + "Rule {} has a file name that doesn't match our standard.".format(
                        file
                    )
                )
                faulty_rules.append(file)
            else:
                # This test make sure that every rules has a filename that corresponds to
                # It's specific logsource.
                # Fix Issue #1381 (https://github.com/SigmaHQ/sigma/issues/1381)
                logsource = self.get_rule_part(file_path=file, part_name="logsource")
                if logsource:
                    pattern_prefix = ""
                    os_infix = ""
                    os_bool = False
                    for key, value in logsource.items():
                        if key == "definition":
                            pass
                        else:
                            if key == "product":
                                # This is to get the OS for certain categories
                                if value == "windows":
                                    os_infix = "win_"
                                elif value == "macos":
                                    os_infix = "macos_"
                                elif value == "linux":
                                    os_infix = "lnx_"
                                # For other stuff
                                elif value == "aws":
                                    pattern_prefix = "aws_"
                                elif value == "azure":
                                    pattern_prefix = "azure_"
                                elif value == "gcp":
                                    pattern_prefix = "gcp_"
                                elif value == "m365":
                                    pattern_prefix = "microsoft365_"
                                elif value == "okta":
                                    pattern_prefix = "okta_"
                                elif value == "onelogin":
                                    pattern_prefix = "onelogin_"
                                elif value == "github":
                                    pattern_prefix = "github_"
                            elif key == "category":
                                if value == "process_creation":
                                    pattern_prefix = "proc_creation_"
                                    os_bool = True
                                elif value == "image_load":
                                    pattern_prefix = "image_load_"
                                elif value == "file_event":
                                    pattern_prefix = "file_event_"
                                    os_bool = True
                                elif value == "registry_set":
                                    pattern_prefix = "registry_set_"
                                elif value == "registry_add":
                                    pattern_prefix = "registry_add_"
                                elif value == "registry_event":
                                    pattern_prefix = "registry_event_"
                                elif value == "registry_delete":
                                    pattern_prefix = "registry_delete_"
                                elif value == "registry_rename":
                                    pattern_prefix = "registry_rename_"
                                elif value == "process_access":
                                    pattern_prefix = "proc_access_"
                                    os_bool = True
                                elif value == "driver_load":
                                    pattern_prefix = "driver_load_"
                                    os_bool = True
                                elif value == "dns_query":
                                    pattern_prefix = "dns_query_"
                                    os_bool = True
                                elif value == "ps_script":
                                    pattern_prefix = "posh_ps_"
                                elif value == "ps_module":
                                    pattern_prefix = "posh_pm_"
                                elif value == "ps_classic_start":
                                    pattern_prefix = "posh_pc_"
                                elif value == "pipe_created":
                                    pattern_prefix = "pipe_created_"
                                elif value == "network_connection":
                                    pattern_prefix = "net_connection_"
                                    os_bool = True
                                elif value == "file_rename":
                                    pattern_prefix = "file_rename_"
                                    os_bool = True
                                elif value == "file_delete":
                                    pattern_prefix = "file_delete_"
                                    os_bool = True
                                elif value == "file_change":
                                    pattern_prefix = "file_change_"
                                    os_bool = True
                                elif value == "file_access":
                                    pattern_prefix = "file_access_"
                                    os_bool = True
                                elif value == "create_stream_hash":
                                    pattern_prefix = "create_stream_hash_"
                                elif value == "create_remote_thread":
                                    pattern_prefix = "create_remote_thread_win_"
                                elif value == "dns":
                                    pattern_prefix = "net_dns_"
                                elif value == "firewall":
                                    pattern_prefix = "net_firewall_"
                                elif value == "webserver":
                                    pattern_prefix = "web_"
                            elif key == "service":
                                if value == "auditd":
                                    pattern_prefix = "lnx_auditd_"
                                elif value == "modsecurity":
                                    pattern_prefix = "modsec_"
                                elif value == "diagnosis-scripted":
                                    pattern_prefix = "win_diagnosis_scripted_"
                                elif value == "firewall-as":
                                    pattern_prefix = "win_firewall_as_"
                                elif value == "msexchange-management":
                                    pattern_prefix = "win_exchange_"
                                elif value == "security":
                                    pattern_prefix = "win_security_"
                                elif value == "system":
                                    pattern_prefix = "win_system_"
                                elif value == "taskscheduler":
                                    pattern_prefix = "win_taskscheduler_"
                                elif value == "terminalservices-localsessionmanager":
                                    pattern_prefix = "win_terminalservices_"
                                elif value == "windefend":
                                    pattern_prefix = "win_defender_"
                                elif value == "wmi":
                                    pattern_prefix = "win_wmi_"
                                elif value == "codeintegrity-operational":
                                    pattern_prefix = "win_codeintegrity_"
                                elif value == "bits-client":
                                    pattern_prefix = "win_bits_client_"
                                elif value == "applocker":
                                    pattern_prefix = "win_applocker_"
                                elif value == "dns-server-analytic":
                                    pattern_prefix = "win_dns_analytic_"
                                elif value == "bitlocker":
                                    pattern_prefix = "win_bitlocker_"
                                elif value == "capi2":
                                    pattern_prefix = "win_capi2_"
                                elif (
                                    value
                                    == "certificateservicesclient-lifecycle-system"
                                ):
                                    pattern_prefix = "win_certificateservicesclient_lifecycle_system_"
                                elif value == "pim":
                                    pattern_prefix = "azure_pim_"

                    # This value is used to test if we should add the OS infix for certain categories
                    if os_bool:
                        pattern_prefix += os_infix
                    if pattern_prefix != "":
                        if not filename.startswith(pattern_prefix):
                            print(
                                Fore.YELLOW
                                + "Rule {} has a file name that doesn't match our standard naming convention.".format(
                                    file
                                )
                            )
                            faulty_rules.append(file)
            name_lst.append(filename)

        self.assertEqual(
            faulty_rules,
            [],
            Fore.RED
            + r"There are rules with malformed file names (too short, too long, uppercase letters, a minus sign etc.). Please see the file names used in our repository and adjust your file names accordingly. The pattern for a valid file name is \'[a-z0-9_]{10,90}\.yml\' and it has to contain at least an underline character. It also has to follow the following naming convention https://github.com/SigmaHQ/sigma-specification/blob/main/sigmahq/Sigmahq_filename_rule.md",
        )

    # sigma cli sigmahq_title_caseIssue, sigmahq_title_endIssue, sigmahq_title_lengthIssue, sigmahq_title_startIssue
    # def test_title(self):
    #     faulty_rules = []
    #     allowed_lowercase_words = [
    #         "the",
    #         "for",
    #         "in",
    #         "with",
    #         "via",
    #         "on",
    #         "to",
    #         "without",
    #         "of",
    #         "through",
    #         "from",
    #         "by",
    #         "as",
    #         "a",
    #         "or",
    #         "at",
    #         "and",
    #         "an",
    #         "over",
    #         "new",
    #     ]
    #     for file in self.yield_next_rule_file_path(self.path_to_rules):
    #         title = self.get_rule_part(file_path=file, part_name="title")
    #         if not title:
    #             print(Fore.RED + "Rule {} has no field 'title'.".format(file))
    #             faulty_rules.append(file)
    #             continue
    #         elif len(title) > 110:
    #             print(
    #                 Fore.YELLOW
    #                 + "Rule {} has a title field with too many characters (>110)".format(
    #                     file
    #                 )
    #             )
    #             faulty_rules.append(file)
    #         if title.startswith("Detects "):
    #             print(
    #                 Fore.RED
    #                 + "Rule {} has a title that starts with 'Detects'".format(file)
    #             )
    #             faulty_rules.append(file)
    #         if title.endswith("."):
    #             print(Fore.RED + "Rule {} has a title that ends with '.'".format(file))
    #             faulty_rules.append(file)
    #         wrong_casing = []
    #         for word in title.split(" "):
    #             if (
    #                 word.islower()
    #                 and not word.lower() in allowed_lowercase_words
    #                 and not "." in word
    #                 and not "/" in word
    #                 and not "_" in word
    #                 and not word[0].isdigit()
    #             ):
    #                 wrong_casing.append(word)
    #         if len(wrong_casing) > 0:
    #             print(
    #                 Fore.RED
    #                 + "Rule {} has a title that has not title capitalization. Words: '{}'".format(
    #                     file, ", ".join(wrong_casing)
    #                 )
    #             )
    #             faulty_rules.append(file)

    #     self.assertEqual(
    #         faulty_rules,
    #         [],
    #         Fore.RED
    #         + "There are rules with non-conform 'title' fields. Please check: https://github.com/SigmaHQ/sigma/wiki/Rule-Creation-Guide#title",
    #     )

    def test_title_in_first_line(self):
        faulty_rules = []
        for file in self.yield_next_rule_file_path(self.path_to_rules):
            yaml = self.get_rule_yaml(file)

            # skip multi-part yaml
            if len(yaml) > 1:
                continue

            # this probably is not the best way to check whether
            # title is the attribute given in the 1st line
            # (also assumes dict keeps the order from the input file)
            if list(yaml[0].keys())[0] != "title":
                print(
                    Fore.RED
                    + "Rule {} does not have its 'title' attribute in the first line".format(
                        file
                    )
                )
                faulty_rules.append(file)

        self.assertEqual(
            faulty_rules,
            [],
            Fore.RED
            + "There are rules without the 'title' attribute in their first line.",
        )

    def test_selection_list_one_value(self):
        def treat_list(file, values, valid_, selection_name):
            # rule with only list of Keywords term
            if len(values) == 1 and not isinstance(values[0], str):
                print(
                    Fore.RED
                    + "Rule {} has the selection ({}) with a list of only 1 element in detection".format(
                        file, selection_name
                    )
                )
                valid_ = False
            elif isinstance(values[0], dict):
                valid_ = treat_dict(file, values, valid_, selection_name)
            return valid_

        def treat_dict(file, values, valid_, selection_name):
            if isinstance(values, list):
                for dict_ in values:
                    for key_ in dict_.keys():
                        if isinstance(dict_[key_], list):
                            if len(dict_[key_]) == 1:
                                print(
                                    Fore.RED
                                    + "Rule {} has the selection ({}/{}) with a list of only 1 value in detection".format(
                                        file, selection_name, key_
                                    )
                                )
                                valid_ = False
            else:
                dict_ = values
                for key_ in dict_.keys():
                    if isinstance(dict_[key_], list):
                        if len(dict_[key_]) == 1:
                            print(
                                Fore.RED
                                + "Rule {} has the selection ({}/{}) with a list of only 1 value in detection".format(
                                    file, selection_name, key_
                                )
                            )
                            valid_ = False
            return valid_

        faulty_rules = []
        for file in self.yield_next_rule_file_path(self.path_to_rules):
            detection = self.get_rule_part(file_path=file, part_name="detection")
            if detection:
                valid = True
                for key in detection:
                    values = detection[key]
                    if isinstance(detection[key], list):
                        valid = treat_list(file, values, valid, key)

                    if isinstance(detection[key], dict):
                        valid = treat_dict(file, values, valid, key)

                    if not valid:
                        faulty_rules.append(file)

        self.assertEqual(
            faulty_rules,
            [],
            Fore.RED + "There are rules using list with only 1 element",
        )

    # simga cli SigmahqSigmacIssue
    # def test_selection_start_or_and(self):
    #     faulty_rules = []
    #     for file in self.yield_next_rule_file_path(self.path_to_rules):
    #         detection = self.get_rule_part(file_path=file, part_name="detection")
    #         if detection:
    #             # This test is a best effort to avoid breaking SIGMAC parser. You could do more testing and try to fix this once and for all by modifiying the token regular expressions https://github.com/SigmaHQ/sigma/blob/b9ae5303f12cda8eb6b5b90a32fd7f11ad65645d/tools/sigma/parser/condition.py#L107-L127
    #             for key in detection:
    #                 if key[:3].lower() == "sel":
    #                     continue
    #                 elif key[:2].lower() == "or":
    #                     print(
    #                         Fore.RED
    #                         + "Rule {} has a selection '{}' that starts with the string 'or'".format(
    #                             file, key
    #                         )
    #                     )
    #                     faulty_rules.append(file)
    #                 elif key[:3].lower() == "and":
    #                     print(
    #                         Fore.RED
    #                         + "Rule {} has a selection '{}' that starts with the string 'and'".format(
    #                             file, key
    #                         )
    #                     )
    #                     faulty_rules.append(file)
    #                 elif key[:3].lower() == "not":
    #                     print(
    #                         Fore.RED
    #                         + "Rule {} has a selection '{}' that starts with the string 'not'".format(
    #                             file, key
    #                         )
    #                     )
    #                     faulty_rules.append(file)

    #     self.assertEqual(
    #         faulty_rules,
    #         [],
    #         Fore.RED
    #         + "There are rules with bad selection names. Can't start a selection name with an 'or*' or an 'and*' or a 'not*' ",
    #     )

    # sigma validator dangling_detection
    # def test_unused_selection(self):
    #     faulty_rules = []
    #     for file in self.yield_next_rule_file_path(self.path_to_rules):
    #         detection = self.get_rule_part(file_path=file, part_name="detection")
    #         condition = detection["condition"]
    #         wildcard_selections = re.compile(r"\sof\s([\w\*]+)(?:$|\s|\))")

    #         # skip rules containing aggregations
    #         if type(condition) == list:
    #             continue

    #         for selection in detection:
    #             if selection == "condition":
    #                 continue
    #             if selection == "timeframe":
    #                 continue

    #             # remove special keywords
    #             condition_list = (
    #                 condition.replace("not ", "")
    #                 .replace("1 of ", "")
    #                 .replace("all of ", "")
    #                 .replace(" or ", " ")
    #                 .replace(" and ", " ")
    #                 .replace("(", "")
    #                 .replace(")", "")
    #                 .split(" ")
    #             )
    #             if selection in condition_list:
    #                 continue

    #             # find all wildcards in condition
    #             found = False
    #             for wildcard_selection in wildcard_selections.findall(condition):
    #                 # wildcard matches selection
    #                 if (
    #                     re.search(wildcard_selection.replace(r"*", r".*"), selection)
    #                     is not None
    #                 ):
    #                     found = True
    #                     break
    #             # selection was not found in condition
    #             if not found:
    #                 print(
    #                     Fore.RED
    #                     + "Rule {} has an unused selection '{}'".format(file, selection)
    #                 )
    #                 faulty_rules.append(file)

    #     self.assertEqual(
    #         faulty_rules, [], Fore.RED + "There are rules with unused selections"
    #     )

    # sigma cli SigmahqInvalidAllModifierIssue
    # def test_all_value_modifier_single_item(self):
    #     faulty_rules = []
    #     for file in self.yield_next_rule_file_path(self.path_to_rules):
    #         detection = self.get_rule_part(file_path=file, part_name="detection")
    #         if detection:
    #             for search_identifier in detection:
    #                 if isinstance(detection[search_identifier], dict):
    #                     for field in detection[search_identifier]:
    #                         if "|all" in field and not isinstance(
    #                             detection[search_identifier][field], list
    #                         ):
    #                             print(
    #                                 Fore.RED
    #                                 + "Rule {} uses the 'all' modifier on a single item in selection ({}/{})".format(
    #                                     file, search_identifier, field
    #                                 )
    #                             )
    #                             faulty_rules.append(file)

    #     self.assertEqual(
    #         faulty_rules,
    #         [],
    #         Fore.RED
    #         + "There are rules with |all modifier only having one item. "
    #         + "Single item values are not allowed to have an all modifier as some back-ends cannot support it. "
    #         + "If you use it as a workaround to duplicate a field in a selection, use a new selection instead.",
    #     )

    # sigma cli SigmahqFieldUserIssue
    # def test_field_user_localization(self):
    #     def checkUser(faulty_rules, dict):
    #         for key, value in dict.items():
    #             if "User" in key:
    #                 if type(value) == str:
    #                     if "AUTORI" in value or "AUTHORI" in value:
    #                         print("Localized user name '{}'.".format(value))
    #                         faulty_rules.append(file)

    #     faulty_rules = []
    #     for file in self.yield_next_rule_file_path(self.path_to_rules):
    #         detection = self.get_rule_part(file_path=file, part_name="detection")
    #         for sel_key, sel_value in detection.items():
    #             if sel_key == "condition" or sel_key == "timeframe":
    #                 continue
    #             # single item selection
    #             if type(sel_value) == dict:
    #                 checkUser(faulty_rules, sel_value)
    #             if type(sel_value) == list:
    #                 # skip keyword selection
    #                 if type(sel_value[0]) != dict:
    #                     continue
    #                 # multiple item selection
    #                 for item in sel_value:
    #                     checkUser(faulty_rules, item)

    #     self.assertEqual(
    #         faulty_rules,
    #         [],
    #         Fore.RED
    #         + "There are rules that match using localized user accounts. Better employ a generic version such as:\n"
    #         + "User|contains: # covers many language settings\n"
    #         + "    - 'AUTHORI'\n"
    #         + "    - 'AUTORI'",
    #     )

    # sigma condition error
    # def test_condition_operator_casesensitive(self):
    #     faulty_rules = []
    #     for file in self.yield_next_rule_file_path(self.path_to_rules):
    #         detection = self.get_rule_part(file_path=file, part_name="detection")
    #         if detection:
    #             valid = True
    #             if isinstance(detection["condition"], str):
    #                 param = detection["condition"].split(" ")
    #                 for item in param:
    #                     if item.lower() == "or" and not item == "or":
    #                         valid = False
    #                     elif item.lower() == "and" and not item == "and":
    #                         valid = False
    #                     elif item.lower() == "not" and not item == "not":
    #                         valid = False
    #                     elif item.lower() == "of" and not item == "of":
    #                         valid = False
    #                 if not valid:
    #                     print(
    #                         Fore.RED
    #                         + "Rule {} has a invalid condition '{}' : 'or','and','not','of' are lowercase".format(
    #                             file, detection["condition"]
    #                         )
    #                     )
    #                     faulty_rules.append(file)

    #     self.assertEqual(
    #         faulty_rules,
    #         [],
    #         Fore.RED + "There are rules using condition without lowercase operator",
    #     )

    def test_broken_thor_logsource_config(self):
        faulty_config = False

        # This test check of the "thor.yml" config file has a missing "WinEventLog:" prefix in Windows log sources
        path_to_thor_config = "tests/thor.yml"
        try:
            path_to_thor_config = os.path.join(
                os.path.dirname(os.path.realpath(__file__)), path_to_thor_config
            )
            thor_logsources = self.get_rule_yaml(path_to_thor_config)[0]["logsources"]

            for key, value in thor_logsources.items():
                try:
                    if value["product"] == "windows":
                        sources_list = value["sources"]
                        for i in sources_list:
                            if not i.startswith("WinEventLog:"):
                                faulty_config = True
                                print(
                                    Fore.RED
                                    + "/tests/thor.yml config file has a broken source. Windows Eventlog sources must start with the keyword 'WinEventLog:'"
                                )
                except:
                    pass
            self.assertEqual(
                faulty_config,
                False,
                Fore.RED
                + "thor.yml configuration file located in 'tests/thor.yml' has a borken log source definition",
            )
        except:
            self.assertEqual(
                faulty_config,
                False,
                Fore.RED
                + "thor.yml configuration file was not found. Please make sure to run the script from the root of the sigma folder",
            )

    def test_re_invalid_escapes(self):
        faulty_rules = []
        MAX_DEPTH = 3

        def create_escape_allow_list():
            """
            Create a list of characters that are allowed to be escaped.
            1. Based on string.punctuation chars that would already be escaped by re.escape()
            2. Followed by special chars like '\n', '\t', '\[0-9]' etc.
            3. Followed by Double- or Single Quote to escape string literals.
            """
            allowed_2_be_escaped = []
            index = 0
            l = tuple(re.escape(string.punctuation))
            for c in l:
                if c == "\\":
                    allowed_2_be_escaped.append(l[index + 1])
                index += 1

            re_specials = [
                "A",
                "b",
                "B",
                "d",
                "D",
                "f",
                "n",
                "r",
                "s",
                "S",
                "t",
                "v",
                "w",
                "W",
                "Z",
                # Match Groups
                "0",
                "1",
                "2",
                "3",
                "4",
                "5",
                "6",
                "7",
                "8",
                "9",
            ]
            allowed_2_be_escaped.extend(re_specials)

            allowed_2_be_escaped.extend(
                [
                    '"',
                    "'",
                ]
            )

            return allowed_2_be_escaped

        def check_list_or_recurse_on_dict(item, depth: int, special: bool) -> None:
            """
            Recursive walk through the detection to find "|re" occurance.
            Jump to check_item_for_bad_escapes with lists or strings found.
            """
            if type(item) == list:
                pass
                # check_item_for_bad_escapes(item)
            elif type(item) == dict and depth <= MAX_DEPTH:
                for keys, sub_item in item.items():
                    if (
                        "|re" in keys
                    ):  # Covers both "base64" and "base64offset" modifiers
                        if type(sub_item) == str or type(sub_item) == list:
                            check_item_for_bad_escapes(sub_item)
                        else:
                            check_list_or_recurse_on_dict(sub_item, depth + 1, True)
                    else:
                        check_list_or_recurse_on_dict(sub_item, depth + 1, special)

        def check_item_for_bad_escapes(item):
            """
            Check item against bad escaped characters
            """
            found_bad_escapes = []
            to_check = []
            if type(item) == str:
                to_check.append(item)
            else:
                to_check = item
            for str_item in to_check:
                l = tuple(str_item)
                index = 0
                for c in l:
                    if c == "\\":
                        # 'l[index-1] != "\\"' ---> Allows "\\\\"
                        # Check if character after \ is not in escape_allow_list and also not already found
                        if (
                            l[index - 1] != "\\"
                            and l[index + 1] not in escape_allow_list
                            and l[index + 1] not in found_bad_escapes
                        ):
                            # Only for debugging:
                            # print(f"Illegal escape found {c}{l[index+1]}")
                            found_bad_escapes.append(f"{l[index+1]}")
                    index += 1

            if len(found_bad_escapes) > 0:
                print(
                    Fore.RED
                    + "Rule {} has forbidden escapes in |re '{}'".format(
                        file, ",".join(found_bad_escapes)
                    )
                )
                faulty_rules.append(file)

        # Create escape_allow_list for this test
        escape_allow_list = create_escape_allow_list()

        # For each rule file, extract detection and dive into recursion
        for file in self.yield_next_rule_file_path(self.path_to_rules):
            detection = self.get_rule_part(file_path=file, part_name="detection")
            if detection:
                check_list_or_recurse_on_dict(detection, 1, False)

        self.assertEqual(
            faulty_rules, [], Fore.RED + "There are rules using illegal re-escapes"
        )

    # def test_confirm_extension_is_yml(self):
    # files_with_incorrect_extensions = []

    # for file in self.yield_next_rule_file_path(self.path_to_rules):
    # file_name_and_extension = os.path.splitext(file)
    # if len(file_name_and_extension) == 2:
    # extension = file_name_and_extension[1]
    # if extension != ".yml":
    # files_with_incorrect_extensions.append(file)

    # self.assertEqual(files_with_incorrect_extensions, [], Fore.RED +
    # "There are rule files with extensions other than .yml")
    # sigma-cli validators attacktag
    # def test_confirm_correct_mitre_tags(self):
    #     files_with_incorrect_mitre_tags = []

    #     for file in self.yield_next_rule_file_path(self.path_to_rules):
    #         tags = self.get_rule_part(file_path=file, part_name="tags")
    #         if tags:
    #             for tag in tags:
    #                 if tag.startswith("attack.") and tag not in self.MITRE_ALL:
    #                     print(
    #                         Fore.RED
    #                         + "Rule {} has the following incorrect MITRE tag {}".format(
    #                             file, tag
    #                         )
    #                     )
    #                     files_with_incorrect_mitre_tags.append(file)

    #     self.assertEqual(
    #         files_with_incorrect_mitre_tags,
    #         [],
    #         Fore.RED
    #         + "There are rules with incorrect/unknown MITRE Tags. (please inform us about new tags that are not yet supported in our tests) and check the correct tags here: https://attack.mitre.org/ ",
    #     )

    # sigma validators duplicate_tag
    # def test_duplicate_tags(self):
    #     files_with_incorrect_mitre_tags = []

    #     for file in self.yield_next_rule_file_path(self.path_to_rules):
    #         tags = self.get_rule_part(file_path=file, part_name="tags")
    #         if tags:
    #             known_tags = []
    #             for tag in tags:
    #                 if tag in known_tags:
    #                     print(
    #                         Fore.RED
    #                         + "Rule {} has the duplicate tag {}".format(file, tag)
    #                     )
    #                     files_with_incorrect_mitre_tags.append(file)
    #                 else:
    #                     known_tags.append(tag)

    #     self.assertEqual(
    #         files_with_incorrect_mitre_tags,
    #         [],
    #         Fore.RED + "There are rules with duplicate tags",
    #     )

    #  sigma validators duplicate_references
    # def test_duplicate_references(self):
    #     files_with_duplicate_references = []

    #     for file in self.yield_next_rule_file_path(self.path_to_rules):
    #         references = self.get_rule_part(file_path=file, part_name="references")
    #         if references:
    #             known_references = []
    #             for reference in references:
    #                 if reference in known_references:
    #                     print(
    #                         Fore.RED
    #                         + "Rule {} has the duplicate reference {}".format(
    #                             file, reference
    #                         )
    #                     )
    #                     files_with_duplicate_references.append(file)
    #                 else:
    #                     known_references.append(reference)

    #     self.assertEqual(
    #         files_with_duplicate_references,
    #         [],
    #         Fore.RED + "There are rules with duplicate references",
    #     )

    # sigma validator identifier_existence identifier_uniqueness
    # def test_missing_id(self):
    #     faulty_rules = []
    #     dict_id = {}
    #     for file in self.yield_next_rule_file_path(self.path_to_rules):
    #         id = self.get_rule_part(file_path=file, part_name="id")
    #         if not id:
    #             print(Fore.YELLOW + "Rule {} has no field 'id'.".format(file))
    #             faulty_rules.append(file)
    #         elif len(id) != 36:
    #             print(
    #                 Fore.YELLOW
    #                 + "Rule {} has a malformed 'id' (not 36 chars).".format(file)
    #             )
    #             faulty_rules.append(file)
    #         elif id.lower() in dict_id.keys():
    #             print(
    #                 Fore.YELLOW
    #                 + "Rule {} has the same 'id' as {}. Ids have to be unique.".format(
    #                     file, dict_id[id]
    #                 )
    #             )
    #             faulty_rules.append(file)
    #         else:
    #             dict_id[id.lower()] = file

    #     self.assertEqual(
    #         faulty_rules,
    #         [],
    #         Fore.RED
    #         + "There are rules with missing or malformed 'id' fields. Generate an id (e.g. here: https://www.uuidgenerator.net/version4) and add it to the reported rule(s).",
    #     )
    # sigma-cli error
    # def test_optional_date_modified(self):
    #     faulty_rules = []
    #     for file in self.yield_next_rule_file_path(self.path_to_rules):
    #         modifiedfield = self.get_rule_part(file_path=file, part_name="modified")
    #         if modifiedfield:
    #             if not isinstance(modifiedfield, str):
    #                 print(
    #                     Fore.YELLOW
    #                     + "Rule {} has a malformed 'modified' (should be YYYY/MM/DD).".format(
    #                         file
    #                     )
    #                 )
    #                 faulty_rules.append(file)
    #             elif len(modifiedfield) != 10:
    #                 print(
    #                     Fore.YELLOW
    #                     + "Rule {} has a malformed 'modified' (not 10 chars, should be YYYY/MM/DD).".format(
    #                         file
    #                     )
    #                 )
    #                 faulty_rules.append(file)
    #             elif modifiedfield[4] != "/" or modifiedfield[7] != "/":
    #                 print(
    #                     Fore.YELLOW
    #                     + "Rule {} has a malformed 'modified' (should be YYYY/MM/DD).".format(
    #                         file
    #                     )
    #                 )
    #                 faulty_rules.append(file)

    #     self.assertEqual(
    #         faulty_rules,
    #         [],
    #         Fore.RED
    #         + "There are rules with malformed 'modified' fields. (create one, e.g. date: 2019-01-14)",
    #     )

    # sigma-cli error and validator status_existence status_unsupported
    # def test_optional_status(self):
    #     faulty_rules = []
    #     valid_status = ["stable", "test", "experimental", "deprecated", "unsupported"]
    #     for file in self.yield_next_rule_file_path(self.path_to_rules):
    #         status_str = self.get_rule_part(file_path=file, part_name="status")
    #         if status_str:
    #             if not status_str in valid_status:
    #                 print(
    #                     Fore.YELLOW
    #                     + "Rule {} has a invalid 'status' (check wiki).".format(file)
    #                 )
    #                 faulty_rules.append(file)
    #             elif status_str == "unsupported":
    #                 print(
    #                     Fore.YELLOW
    #                     + "Rule {} has the unsupported 'status', can not be in rules directory".format(
    #                         file
    #                     )
    #                 )
    #                 faulty_rules.append(file)
    #         else:
    #             print(
    #                 Fore.YELLOW + "Rule {} is missing the 'status' field".format(file)
    #             )
    #             faulty_rules.append(file)

    #     self.assertEqual(
    #         faulty_rules,
    #         [],
    #         Fore.RED
    #         + "There are rules with malformed or missing 'status' fields. (check https://github.com/SigmaHQ/sigma-specification)",
    #     )

    # Sigma validator all_of_them_condition
    # def test_all_of_them_condition(self):
    #     faulty_detections = []

    #     for file in self.yield_next_rule_file_path(self.path_to_rules):
    #         detection = self.get_rule_part(file_path=file, part_name="detection")

    #         if "all of them" in detection["condition"]:
    #             faulty_detections.append(file)

    #     self.assertEqual(
    #         faulty_detections,
    #         [],
    #         Fore.RED
    #         + "There are rules using 'all of them'. Better use e.g. 'all of selection*' instead (and use the 'selection_' prefix as search-identifier).",
    #     )

    # sigma-cli validators tlptag
    # def test_optional_tlp(self):
    #     faulty_rules = []
    #     valid_tlp = [
    #         "WHITE",
    #         "GREEN",
    #         "AMBER",
    #         "RED",
    #     ]
    #     for file in self.yield_next_rule_file_path(self.path_to_rules):
    #         tlp_str = self.get_rule_part(file_path=file, part_name="tlp")
    #         if tlp_str:
    #             # it exists but isn't a string
    #             if not isinstance(tlp_str, str):
    #                 print(
    #                     Fore.YELLOW
    #                     + "Rule {} has a 'tlp' field that isn't a string.".format(file)
    #                 )
    #                 faulty_rules.append(file)
    #             elif not tlp_str.upper() in valid_tlp:
    #                 print(
    #                     Fore.YELLOW
    #                     + "Rule {} has a 'tlp' field with not valid value.".format(file)
    #                 )
    #                 faulty_rules.append(file)

    #     self.assertEqual(
    #         faulty_rules,
    #         [],
    #         Fore.RED
    #         + "There are rules with malformed optional 'tlp' fields. (https://www.cisa.gov/tlp)",
    #     )

    # Not in the specification
    # def test_optional_target(self):
    #     faulty_rules = []
    #     for file in self.yield_next_rule_file_path(self.path_to_rules):
    #         target = self.get_rule_part(file_path=file, part_name="target")
    #         if target:
    #             # it exists but isn't a list
    #             if not isinstance(target, list):
    #                 print(
    #                     Fore.YELLOW
    #                     + "Rule {} has a 'target' field that isn't a list.".format(file)
    #                 )
    #                 faulty_rules.append(file)

    #     self.assertEqual(
    #         faulty_rules,
    #         [],
    #         Fore.RED
    #         + "There are rules with malformed 'target' fields. (has to be a list of values even if it contains only a single value)",
    #     )
    # sigma validators duplicate_title
    # def test_duplicate_titles(self):
    #     # This test ensure that every rule has a unique title
    #     faulty_rules = []
    #     titles_dict = {}
    #     for file in self.yield_next_rule_file_path(self.path_to_rules):
    #         title = (
    #             self.get_rule_part(file_path=file, part_name="title").lower().rstrip()
    #         )
    #         duplicate = False
    #         for rule, title_ in titles_dict.items():
    #             if title == title_:
    #                 print(
    #                     Fore.RED
    #                     + "Rule {} has an already used title in {}.".format(file, rule)
    #                 )
    #                 duplicate = True
    #                 faulty_rules.append(file)
    #                 continue
    #         if not duplicate:
    #             titles_dict[file] = title

    #     self.assertEqual(
    #         faulty_rules,
    #         [],
    #         Fore.RED
    #         + "There are rules that share the same 'title'. Please check: https://github.com/SigmaHQ/sigma/wiki/Rule-Creation-Guide#title",
    #     )

    # def test_invalid_logsource_attributes(self):
    #     faulty_rules = []
    #     valid_logsource = [
    #         'category',
    #         'product',
    #         'service',
    #         'definition',
    #     ]
    #     for file in self.yield_next_rule_file_path(self.path_to_rules):
    #         logsource = self.get_rule_part(
    #             file_path=file, part_name="logsource")
    #         if not logsource:
    #             print(Fore.RED + "Rule {} has no 'logsource'.".format(file))
    #             faulty_rules.append(file)
    #             continue
    #         valid = True
    #         for key in logsource:
    #             if key.lower() not in valid_logsource:
    #                 print(
    #                     Fore.RED + "Rule {} has a logsource with an invalid field ({})".format(file, key))
    #                 valid = False
    #             elif not isinstance(logsource[key], str):
    #                 print(
    #                     Fore.RED + "Rule {} has a logsource with an invalid field type ({})".format(file, key))
    #                 valid = False
    #         if not valid:
    #             faulty_rules.append(file)

    #     self.assertEqual(faulty_rules, [], Fore.RED +
    #                      "There are rules with non-conform 'logsource' fields. Please check: https://github.com/SigmaHQ/sigma/wiki/Rule-Creation-Guide#log-source")
    # def test_field_name_typo(self):
    #     # add "OriginalFilename" after Aurora switched to SourceFilename
    #     # add "ProviderName" after special case powershell classic is resolved
    #     faulty_rules = []
    #     for file in self.yield_next_rule_file_path(self.path_to_rules):
    #         # typos is a list of tuples where each tuple contains ("The typo", "The correct version")
    #         typos = [("ServiceFilename", "ServiceFileName"), ("TargetFileName", "TargetFilename"), ("SourceFileName", "OriginalFileName"), ("Commandline", "CommandLine"), ("Targetobject", "TargetObject"), ("OriginalName", "OriginalFileName"), ("ImageFileName", "OriginalFileName"), ("details", "Details")]
    #         # Some fields exists in certain log sources in different forms than other log sources. We need to handle these as special cases
    #         # We check first the logsource to handle special cases
    #         logsource = self.get_rule_part(file_path=file, part_name="logsource").values()
    #         # add more typos in specific logsources below
    #         if "windefend" in logsource:
    #             typos += [("New_Value", "NewValue"), ("Old_Value", "OldValue"), ('Source_Name', 'SourceName'), ("Newvalue", "NewValue"), ("Oldvalue", "OldValue"), ('Sourcename', 'SourceName')]
    #         elif "registry_set" in logsource or "registry_add" in logsource or "registry_event" in logsource:
    #             typos += [("Targetobject", "TargetObject"), ("Eventtype", "EventType"), ("Newname", "NewName")]
    #         elif "process_creation" in logsource:
    #             typos += [("Parentimage", "ParentImage"), ("Integritylevel", "IntegrityLevel"), ("IntegritiLevel", "IntegrityLevel")]
    #         elif "file_access" in logsource:
    #             del(typos[typos.index(("TargetFileName", "TargetFilename"))]) # We remove the entry to "TargetFileName" to avoid confusion
    #             typos += [("TargetFileName", "FileName"), ("TargetFilename","FileName")]
    #         detection = self.get_rule_part(file_path=file, part_name="detection")
    #         if detection:
    #             for search_identifier in detection:
    #                 if isinstance(detection[search_identifier], dict):
    #                     for field in detection[search_identifier]:
    #                         for typo in typos:
    #                             if typo[0] in field:
    #                                 print(Fore.RED + "Rule {} has a common typo ({}) which should be ({}) in selection ({}/{})".format(file, typo[0], typo[1], search_identifier, field))
    #                                 faulty_rules.append(file)

    #     self.assertEqual(faulty_rules, [], Fore.RED + "There are rules with common typos in field names.")

    # Sigma error SigmaModifierError
    # def test_unknown_value_modifier(self):
    #     known_modifiers = [
    #         "contains",
    #         "startswith",
    #         "endswith",
    #         "all",
    #         "base64offset",
    #         "base64",
    #         "utf16le",
    #         "utf16be",
    #         "wide",
    #         "utf16",
    #         "windash",
    #         "re",
    #         "cidr",
    #     ]
    #     faulty_rules = []
    #     for file in self.yield_next_rule_file_path(self.path_to_rules):
    #         detection = self.get_rule_part(file_path=file, part_name="detection")
    #         if detection:
    #             for search_identifier in detection:
    #                 if isinstance(detection[search_identifier], dict):
    #                     for field in detection[search_identifier]:
    #                         if "|" in field:
    #                             for current_modifier in field.split("|")[1:]:
    #                                 found = False
    #                                 for target_modifier in known_modifiers:
    #                                     if current_modifier == target_modifier:
    #                                         found = True
    #                                 if not found:
    #                                     print(
    #                                         Fore.RED
    #                                         + "Rule {} uses an unknown field modifier ({}/{})".format(
    #                                             file, search_identifier, field
    #                                         )
    #                                     )
    #                                     faulty_rules.append(file)

    #     self.assertEqual(
    #         faulty_rules,
    #         [],
    #         Fore.RED
    #         + "There are rules with unknown value modifiers. Most often it is just a typo.",
    #     )

    # sigma error and validator attacktag,cartag,cvetag,detection_tag,stptag,tlptag
    # def test_optional_tags(self):
    #     files_with_incorrect_tags = []
    #     tags_pattern = re.compile(
    #         r"cve\.\d+\.\d+|attack\.(t\d{4}\.\d{3}|[gts]\d{4})$|attack\.[a-z_]+|car\.\d{4}-\d{2}-\d{3}|detection\.\w+"
    #     )
    #     for file in self.yield_next_rule_file_path(self.path_to_rules):
    #         tags = self.get_rule_part(file_path=file, part_name="tags")
    #         if tags:
    #             for tag in tags:
    #                 if tags_pattern.match(tag) == None:
    #                     print(
    #                         Fore.RED
    #                         + "Rule {} has the invalid tag <{}>".format(file, tag)
    #                     )
    #                     files_with_incorrect_tags.append(file)

    #     self.assertEqual(
    #         files_with_incorrect_tags,
    #         [],
    #         Fore.RED
    #         + "There are rules with incorrect/unknown Tags. (please inform us about new tags that are not yet supported in our tests) and check the correct tags here: https://github.com/SigmaHQ/sigma-specification/blob/main/Tags_specification.md ",
    #     )

    # sigma error and validator custom_attributes
    # def test_optional_related(self):
    #     faulty_rules = []
    #     valid_type = ["derived", "obsoletes", "merged", "renamed", "similar"]
    #     for file in self.yield_next_rule_file_path(self.path_to_rules):
    #         related_lst = self.get_rule_part(file_path=file, part_name="related")
    #         if related_lst:
    #             # it exists but isn't a list
    #             if not isinstance(related_lst, list):
    #                 print(
    #                     Fore.YELLOW
    #                     + "Rule {} has a 'related' field that isn't a list.".format(
    #                         file
    #                     )
    #                 )
    #                 faulty_rules.append(file)
    #             else:
    #                 type_ok = True
    #                 for ref in related_lst:
    #                     try:
    #                         id_str = ref["id"]
    #                         type_str = ref["type"]
    #                     except KeyError:
    #                         print(
    #                             Fore.YELLOW
    #                             + "Rule {} has an invalid form of 'related/type' value.".format(
    #                                 file
    #                             )
    #                         )
    #                         faulty_rules.append(file)
    #                         continue
    #                     if not type_str in valid_type:
    #                         type_ok = False
    #                 # Only add one time if many bad type in the same file
    #                 if type_ok == False:
    #                     print(
    #                         Fore.YELLOW
    #                         + "Rule {} has a 'related/type' invalid value.".format(file)
    #                     )
    #                     faulty_rules.append(file)
    #         else:
    #             typo_list = []
    #             # Add more typos
    #             typo_list.append(
    #                 self.get_rule_part(file_path=file, part_name="realted")
    #             )
    #             typo_list.append(
    #                 self.get_rule_part(file_path=file, part_name="relatde")
    #             )
    #             typo_list.append(self.get_rule_part(file_path=file, part_name="relted"))
    #             typo_list.append(self.get_rule_part(file_path=file, part_name="rlated"))

    #             for i in typo_list:
    #                 if i != None:
    #                     print(
    #                         Fore.YELLOW
    #                         + "Rule {} has a typo in it's 'related' field.".format(file)
    #                     )
    #                     faulty_rules.append(file)

    #     self.assertEqual(
    #         faulty_rules,
    #         [],
    #         Fore.RED
    #         + "There are rules with malformed optional 'related' fields. (check https://github.com/SigmaHQ/sigma-specification)",
    #     )

    # sigma error validators date_existence
    # def test_missing_date(self):
    #     faulty_rules = []
    #     for file in self.yield_next_rule_file_path(self.path_to_rules):
    #         datefield = self.get_rule_part(file_path=file, part_name="date")
    #         if not datefield:
    #             print(Fore.YELLOW + "Rule {} has no field 'date'.".format(file))
    #             faulty_rules.append(file)
    #         elif not isinstance(datefield, str):
    #             print(
    #                 Fore.YELLOW
    #                 + "Rule {} has a malformed 'date' (should be YYYY/MM/DD).".format(
    #                     file
    #                 )
    #             )
    #             faulty_rules.append(file)
    #         elif len(datefield) != 10:
    #             print(
    #                 Fore.YELLOW
    #                 + "Rule {} has a malformed 'date' (not 10 chars, should be YYYY/MM/DD).".format(
    #                     file
    #                 )
    #             )
    #             faulty_rules.append(file)
    #         elif datefield[4] != "/" or datefield[7] != "/":
    #             print(
    #                 Fore.YELLOW
    #                 + "Rule {} has a malformed 'date' (should be YYYY/MM/DD).".format(
    #                     file
    #                 )
    #             )
    #             faulty_rules.append(file)

    #     self.assertEqual(
    #         faulty_rules,
    #         [],
    #         Fore.RED
    #         + "There are rules with missing or malformed 'date' fields. (create one, e.g. date: 2019-01-14)",
    #     )

    # sigma validators description_existence description_length
    # def test_missing_description(self):
    #     faulty_rules = []
    #     for file in self.yield_next_rule_file_path(self.path_to_rules):
    #         descriptionfield = self.get_rule_part(
    #             file_path=file, part_name="description"
    #         )
    #         if not descriptionfield:
    #             print(Fore.YELLOW + "Rule {} has no field 'description'.".format(file))
    #             faulty_rules.append(file)
    #         elif not isinstance(descriptionfield, str):
    #             print(
    #                 Fore.YELLOW
    #                 + "Rule {} has a 'description' field that isn't a string.".format(
    #                     file
    #                 )
    #             )
    #             faulty_rules.append(file)
    #         elif len(descriptionfield) < 16:
    #             print(
    #                 Fore.YELLOW
    #                 + "Rule {} has a really short description. Please elaborate.".format(
    #                     file
    #                 )
    #             )
    #             faulty_rules.append(file)

    #     self.assertEqual(
    #         faulty_rules,
    #         [],
    #         Fore.RED
    #         + "There are rules with missing or malformed 'description' field. (create one, e.g. description: Detects the suspicious behaviour of process XY doing YZ)",
    #     )

    # sigma error validators level_existence
    # def test_level(self):
    #     faulty_rules = []
    #     valid_level = [
    #         "informational",
    #         "low",
    #         "medium",
    #         "high",
    #         "critical",
    #     ]
    #     for file in self.yield_next_rule_file_path(self.path_to_rules):
    #         level_str = self.get_rule_part(file_path=file, part_name="level")
    #         if not level_str:
    #             print(Fore.YELLOW + "Rule {} has no field 'level'.".format(file))
    #             faulty_rules.append(file)
    #         elif not level_str in valid_level:
    #             print(
    #                 Fore.YELLOW
    #                 + "Rule {} has a invalid 'level' (check wiki).".format(file)
    #             )
    #             faulty_rules.append(file)

    #     self.assertEqual(
    #         faulty_rules,
    #         [],
    #         Fore.RED
    #         + "There are rules with missing or malformed 'level' fields. (check https://github.com/SigmaHQ/sigma-specification)",
    #     )

    # sigma error
    # def test_optional_fields(self):
    #     faulty_rules = []
    #     for file in self.yield_next_rule_file_path(self.path_to_rules):
    #         fields_str = self.get_rule_part(file_path=file, part_name="fields")
    #         if fields_str:
    #             # it exists but isn't a list
    #             if not isinstance(fields_str, list):
    #                 print(
    #                     Fore.YELLOW
    #                     + "Rule {} has a 'fields' field that isn't a list.".format(file)
    #                 )
    #                 faulty_rules.append(file)

    #     self.assertEqual(
    #         faulty_rules,
    #         [],
    #         Fore.RED
    #         + "There are rules with malformed optional 'fields' fields. (has to be a list of values even if it contains only a single value)",
    #     )

    # sigma error
    # def test_optional_falsepositives_listtype(self):
    #     faulty_rules = []
    #     for file in self.yield_next_rule_file_path(self.path_to_rules):
    #         falsepositives_str = self.get_rule_part(
    #             file_path=file, part_name="falsepositives"
    #         )
    #         if falsepositives_str:
    #             # it exists but isn't a list
    #             if not isinstance(falsepositives_str, list):
    #                 print(
    #                     Fore.YELLOW
    #                     + "Rule {} has a 'falsepositives' field that isn't a list.".format(
    #                         file
    #                     )
    #                 )
    #                 faulty_rules.append(file)

    #     self.assertEqual(
    #         faulty_rules,
    #         [],
    #         Fore.RED
    #         + "There are rules with malformed optional 'falsepositives' fields. (has to be a list of values even if it contains only a single value)",
    #     )

    # sigma error
    # # Upgrade Detection Rule License  1.1
    # def test_optional_author(self):
    #     faulty_rules = []
    #     for file in self.yield_next_rule_file_path(self.path_to_rules):
    #         author_str = self.get_rule_part(file_path=file, part_name="author")
    #         if author_str:
    #             # it exists but isn't a string
    #             if not isinstance(author_str, str):
    #                 print(
    #                     Fore.YELLOW
    #                     + "Rule {} has a 'author' field that isn't a string.".format(
    #                         file
    #                     )
    #                 )
    #                 faulty_rules.append(file)

    #     self.assertEqual(
    #         faulty_rules,
    #         [],
    #         Fore.RED
    #         + "There are rules with malformed 'author' fields. (has to be a string even if it contains many author)",
    #     )

    # sigma validator custom_attributes
    # def test_references_plural(self):
    #     faulty_rules = []
    #     for file in self.yield_next_rule_file_path(self.path_to_rules):
    #         reference = self.get_rule_part(file_path=file, part_name="reference")
    #         if reference:
    #             # it exists but in singular form
    #             faulty_rules.append(file)

    #     self.assertEqual(
    #         faulty_rules,
    #         [],
    #         Fore.RED
    #         + "There are rules with malformed 'references' fields. (has to be 'references' in plural form, not singular)",
    #     )


# sigma-cli validators attacktag
# def get_mitre_data():
#     """
#     Use Tags from CTI subrepo to get consitant data
#     """
#     cti_path = "cti/"
#     cti_path = os.path.join(os.path.dirname(os.path.realpath(__file__)), cti_path)

#     # Get ATT&CK information
#     lift = attack_client(local_path=cti_path)
#     # Techniques
#     MITRE_TECHNIQUES = []
#     MITRE_TECHNIQUE_NAMES = []
#     MITRE_PHASE_NAMES = set()
#     MITRE_TOOLS = []
#     MITRE_GROUPS = []
#     # Techniques
#     enterprise_techniques = lift.get_enterprise_techniques()
#     for t in enterprise_techniques:
#         MITRE_TECHNIQUE_NAMES.append(
#             t["name"].lower().replace(" ", "_").replace("-", "_")
#         )
#         for r in t.external_references:
#             if "external_id" in r:
#                 MITRE_TECHNIQUES.append(r["external_id"].lower())
#         if "kill_chain_phases" in t:
#             for kc in t["kill_chain_phases"]:
#                 if "phase_name" in kc:
#                     MITRE_PHASE_NAMES.add(kc["phase_name"].replace("-", "_"))
#     # Tools / Malware
#     enterprise_tools = lift.get_enterprise_tools()
#     for t in enterprise_tools:
#         for r in t.external_references:
#             if "external_id" in r:
#                 MITRE_TOOLS.append(r["external_id"].lower())
#     enterprise_malware = lift.get_enterprise_malware()
#     for m in enterprise_malware:
#         for r in m.external_references:
#             if "external_id" in r:
#                 MITRE_TOOLS.append(r["external_id"].lower())
#     # Groups
#     enterprise_groups = lift.get_enterprise_groups()
#     for g in enterprise_groups:
#         for r in g.external_references:
#             if "external_id" in r:
#                 MITRE_GROUPS.append(r["external_id"].lower())

#     # Debugging
#     print(
#         "MITRE ATT&CK LIST LENGTHS: %d %d %d %d %d"
#         % (
#             len(MITRE_TECHNIQUES),
#             len(MITRE_TECHNIQUE_NAMES),
#             len(list(MITRE_PHASE_NAMES)),
#             len(MITRE_GROUPS),
#             len(MITRE_TOOLS),
#         )
#     )

#     # Combine all IDs to a big tag list
#     return [
#         "attack." + item
#         for item in MITRE_TECHNIQUES
#         + MITRE_TECHNIQUE_NAMES
#         + list(MITRE_PHASE_NAMES)
#         + MITRE_GROUPS
#         + MITRE_TOOLS
#     ]


if __name__ == "__main__":
    init(autoreset=True)
    # Run the tests
    unittest.main()