1"""The `suricata_check._checkers` module contains functionality for selecting checkers."""
2
3import logging
4import logging.handlers
5import pkgutil
6from collections.abc import Iterable, Sequence
7from functools import lru_cache
8from typing import TypeVar
9
10from suricata_check.checkers.interface import CheckerInterface
11from suricata_check.checkers.interface._dummy import DummyChecker
12from suricata_check.utils.regex_provider import get_regex_provider
13
14_regex_provider = get_regex_provider()
15
16_logger = logging.getLogger(__name__)
17
18# Global variable to check if extensions have already been imported in case get_checkers() is called multiple times.
19suricata_check_extensions_imported = False
20
21
22def _import_extensions() -> None:
23 global suricata_check_extensions_imported # noqa: PLW0603
24 if suricata_check_extensions_imported is True:
25 return
26
27 for module in pkgutil.iter_modules():
28 if module.name.startswith("suricata_check_"):
29 try:
30 imported_module = __import__(module.name)
31 _logger.info(
32 "Detected and successfully imported suricata-check extension %s with version %s.",
33 module.name.replace("_", "-"),
34 getattr(imported_module, "__version__"),
35 )
36 except ImportError:
37 _logger.warning(
38 "Detected potential suricata-check extension %s but failed to import it.",
39 module.name.replace("_", "-"),
40 )
41 suricata_check_extensions_imported = True
42
43
[docs]
44@lru_cache(maxsize=1)
45def get_checkers(
46 include: Sequence[str] = (".*",),
47 exclude: Sequence[str] = (),
48 issue_severity: int = logging.INFO,
49) -> Sequence[CheckerInterface]:
50 """Auto discovers all available checkers that implement the CheckerInterface.
51
52 Returns:
53 A list of available checkers that implement the CheckerInterface.
54
55 """
56 # Check for extensions and try to import them
57 _import_extensions()
58
59 checkers: list[CheckerInterface] = []
60 for checker in __get_all_subclasses(CheckerInterface):
61 if checker.__name__ == DummyChecker.__name__:
62 continue
63
64 # Initialize DummyCheckers to retrieve error messages.
65 if issubclass(checker, DummyChecker):
66 checker()
67
68 enabled, relevant_codes = __get_checker_enabled(
69 checker,
70 include,
71 exclude,
72 issue_severity,
73 )
74
75 if enabled:
76 checkers.append(checker(include=relevant_codes))
77
78 else:
79 _logger.info("Checker %s is disabled.", checker.__name__)
80
81 _logger.info(
82 "Discovered and enabled checkers: [%s]",
83 ", ".join([c.__class__.__name__ for c in checkers]),
84 )
85 if len(checkers) == 0:
86 _logger.warning(
87 "No checkers were enabled. Check the include and exclude arguments.",
88 )
89
90 # Perform a uniqueness check on the codes emmitted by the checkers
91 for checker1 in checkers:
92 for checker2 in checkers:
93 if checker1 == checker2:
94 continue
95 if not set(checker1.codes).isdisjoint(checker2.codes):
96 msg = f"Checker {checker1.__class__.__name__} and {checker2.__class__.__name__} have overlapping codes."
97 _logger.error(msg)
98
99 return sorted(checkers, key=lambda x: x.__class__.__name__)
100
101
102def __get_checker_enabled(
103 checker: type[CheckerInterface],
104 include: Sequence[str],
105 exclude: Sequence[str],
106 issue_severity: int,
107) -> tuple[bool, set[str]]:
108 enabled = checker.enabled_by_default
109
110 # If no include regexes are provided, include all by default
111 if len(include) == 0:
112 relevant_codes = set(checker.codes.keys())
113 else:
114 # If include regexes are provided, include all codes that match any of these regexes
115 relevant_codes = set()
116
117 for regex in include:
118 relevant_codes.update(
119 set(
120 filter(
121 lambda code: _regex_provider.compile("^" + regex + "$").match(
122 code,
123 )
124 is not None,
125 checker.codes.keys(),
126 ),
127 ),
128 )
129
130 if len(relevant_codes) > 0:
131 enabled = True
132
133 # Now remove the codes that are excluded according to any of the provided exclude regexes
134 for regex in exclude:
135 relevant_codes = set(
136 filter(
137 lambda code: _regex_provider.compile("^" + regex + "$").match(code)
138 is None,
139 relevant_codes,
140 ),
141 )
142
143 # Now filter out irrelevant codes based on severity
144 relevant_codes = set(
145 filter(
146 lambda code: checker.codes[code]["severity"] >= issue_severity,
147 relevant_codes,
148 ),
149 )
150
151 if len(relevant_codes) == 0:
152 enabled = False
153
154 return enabled, relevant_codes
155
156
157Cls = TypeVar("Cls")
158
159
160def __get_all_subclasses(cls: type[Cls]) -> Iterable[type[Cls]]:
161 """Returns all class types that subclass the provided type."""
162 all_subclasses = []
163
164 for subclass in cls.__subclasses__():
165 all_subclasses.append(subclass)
166 all_subclasses.extend(__get_all_subclasses(subclass))
167
168 return all_subclasses