4020719131

Build # 4020719131

Build Type

push

github

Committed by GitHub

Commit Message

Merge pull request #42 from ninoseki/sigma

Run Details

300 of 421 new or added lines in 19 files covered. (71.26%)

7 existing lines in 4 files now uncovered.

608 of 791 relevant lines covered (76.86%)

1.54 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

50.0

/abuse_whois/pysigma/sigma_scan.py

import fnmatch
from typing import Any

import regex as re

from . import schemas


def match_search_id(rule: schemas.Rule, event: dict[Any, Any], search_id: str):
    search_fields = rule.get_search_fields(search_id)
    if search_fields:
        return find_matches(event, search_fields)

    raise ValueError()


def check_pair(event: dict[Any, Any], key: str, value: schemas.Query) -> bool:
    """
    Checks to see if a given key and value from the rule are also in the event.
    Takes into consideration any value modifiers.

    :param event: dict, a single event from the event log
    :param key: str, given dict key
    :param value: str, given key value
    :return: bool, whether or not the match exists in the event
    """
    # Before we can apply modifiers and search properly, we need to check if there
    # is even a value to modify, so do the null checks first
    if value is None:
        return event.get(key) is None

    if key not in event:
        return False

    if isinstance(value, re.Pattern):
        return bool(value.match(str(event[key])))

    # Because by default sigma string matching is case insensitive, lower the event
    # string before comparing it. The value string is already lowercase.
    # TODO potential optimization by caching lowercased event fields
    return str(event[key]).lower() == value


def find_matches(
    event: dict, search: schemas.DetectionField, match_all: bool = False
) -> bool:
    """
    Matches the items in the rule to the event. Iterates through the sections and if there's a list it iterates
    through that. Uses checkPair to see if the items in the list/dictionary match items in the event log.

    :param event: dict, event read from the Sysmon log
    :param search: An object describing what sort of search to run
    :param match_all: A bool indicating if we want all fields in list to hit
    :return: bool, whether or not we found a match
    """
    if search.list_search:
        check = all if match_all else any
        return check(
            any(check_pair(event, event_key, field) for event_key in event)
            for field in search.list_search
        )

    for field in search.map_search:
        if find_matches_by_map(event, field):
            return True

    return False


def find_matches_by_map(event: dict[Any, Any], search: schemas.DetectionMap) -> bool:
    """
    :param event:
    :param search: a dict of fields to search. All must be satisfied.
    :return:
    """

    for field_name, (value, modifiers) in search:
        if not find_matches_by_map_entry(event, field_name, value, modifiers):
            return False

    return True


def find_matches_by_map_entry(
    event: dict,
    field_name: str,
    field_values: list[schemas.Query],
    modifiers: list[str],
) -> bool:
    """
    :param event: the event to search in
    :param field_name: A field in the event we want to search
    :param field_values: valid values or patterns for the field in question
    :return:
    """

    # Normally any of the values in field_values is acceptable, but the all modifier inverts that
    if "all" in modifiers:
        for permitted_value in field_values:
            if not check_pair(event, field_name, permitted_value):
                return False

        return True

    for permitted_value in field_values:
        if check_pair(event, field_name, permitted_value):
            return True

    return False


def analyze_x_of(
    rule: schemas.Rule,
    event: dict[Any, Any],
    count: int | None = None,
    selector: str | None = None,
) -> bool:
    """
    Analyzes the truth value of an 'x of' condition specified within the condition string of the rule.

    :param rule: Rule currently being applied
    :param event: event currently being scanned
    :param count: left side of the x of statement, either 1 or None (for all)
    :param selector: right side of the x of statement, a pattern or None (for all)
    :return: bool, truth value of 'x of' condition
    """

    # First we need to choose our set of fields based on our selector.
    matches: dict[str, schemas.DetectionField] = {}
    all_searches = rule.get_all_searches()

    if selector is None:  # None indicates all.
        matches = all_searches
    else:
        for search_id, search_fields in all_searches.items():
            if fnmatch.fnmatch(search_id, selector):
                matches[search_id] = search_fields

    match_all = False
    if count is None:
        match_all = True
        count = len(matches)

    permitted_misses = len(matches) - count

    # Now that we have our searches to check, run them
    search_hits = 0
    search_misses = 0
    for search_id, search_fields in matches.items():
        if find_matches(event, search_fields, match_all):
            search_hits += 1
        else:
            search_misses += 1

        # Short circuit if we found the matches, or if we can't find the number anymore

        if search_hits >= count:
            return True

        if search_misses > permitted_misses:
            return False

    return False

1	import fnmatch	2✔
2	from typing import Any	2✔
3
4	import regex as re	2✔
5
6	from . import schemas	2✔
7
8
9	def match_search_id(rule: schemas.Rule, event: dict[Any, Any], search_id: str):	2✔
10	search_fields = rule.get_search_fields(search_id)	2✔
11	if search_fields:	2✔
12	return find_matches(event, search_fields)	2✔
13
NEW 14	raise ValueError()	×
15
16
17	def check_pair(event: dict[Any, Any], key: str, value: schemas.Query) -> bool:	2✔
18	"""
19	Checks to see if a given key and value from the rule are also in the event.
20	Takes into consideration any value modifiers.
21
22	:param event: dict, a single event from the event log
23	:param key: str, given dict key
24	:param value: str, given key value
25	:return: bool, whether or not the match exists in the event
26	"""
27	# Before we can apply modifiers and search properly, we need to check if there
28	# is even a value to modify, so do the null checks first
29	if value is None:	2✔
NEW 30	return event.get(key) is None	×
31
32	if key not in event:	2✔
33	return False	2✔
34
35	if isinstance(value, re.Pattern):	2✔
36	return bool(value.match(str(event[key])))	2✔
37
38	# Because by default sigma string matching is case insensitive, lower the event
39	# string before comparing it. The value string is already lowercase.
40	# TODO potential optimization by caching lowercased event fields
NEW 41	return str(event[key]).lower() == value	×
42
43
44	def find_matches(	2✔
45	event: dict, search: schemas.DetectionField, match_all: bool = False
46	) -> bool:
47	"""
48	Matches the items in the rule to the event. Iterates through the sections and if there's a list it iterates
49	through that. Uses checkPair to see if the items in the list/dictionary match items in the event log.
50
51	:param event: dict, event read from the Sysmon log
52	:param search: An object describing what sort of search to run
53	:param match_all: A bool indicating if we want all fields in list to hit
54	:return: bool, whether or not we found a match
55	"""
56	if search.list_search:	2✔
NEW 57	check = all if match_all else any	×
NEW 58	return check(	×
59	any(check_pair(event, event_key, field) for event_key in event)
60	for field in search.list_search
61	)
62
63	for field in search.map_search:	2✔
64	if find_matches_by_map(event, field):	2✔
65	return True	2✔
66
67	return False	2✔
68
69
70	def find_matches_by_map(event: dict[Any, Any], search: schemas.DetectionMap) -> bool:	2✔
71	"""
72	:param event:
73	:param search: a dict of fields to search. All must be satisfied.
74	:return:
75	"""
76
77	for field_name, (value, modifiers) in search:	2✔
78	if not find_matches_by_map_entry(event, field_name, value, modifiers):	2✔
79	return False	2✔
80
81	return True	2✔
82
83
84	def find_matches_by_map_entry(	2✔
85	event: dict,
86	field_name: str,
87	field_values: list[schemas.Query],
88	modifiers: list[str],
89	) -> bool:
90	"""
91	:param event: the event to search in
92	:param field_name: A field in the event we want to search
93	:param field_values: valid values or patterns for the field in question
94	:return:
95	"""
96
97	# Normally any of the values in field_values is acceptable, but the all modifier inverts that
98	if "all" in modifiers:	2✔
NEW 99	for permitted_value in field_values:	×
NEW 100	if not check_pair(event, field_name, permitted_value):	×
NEW 101	return False	×
102
NEW 103	return True	×
104
105	for permitted_value in field_values:	2✔
106	if check_pair(event, field_name, permitted_value):	2✔
107	return True	2✔
108
109	return False	2✔
110
111
112	def analyze_x_of(	2✔
113	rule: schemas.Rule,
114	event: dict[Any, Any],
115	count: int \| None = None,
116	selector: str \| None = None,
117	) -> bool:
118	"""
119	Analyzes the truth value of an 'x of' condition specified within the condition string of the rule.
120
121	:param rule: Rule currently being applied
122	:param event: event currently being scanned
123	:param count: left side of the x of statement, either 1 or None (for all)
124	:param selector: right side of the x of statement, a pattern or None (for all)
125	:return: bool, truth value of 'x of' condition
126	"""
127
128	# First we need to choose our set of fields based on our selector.
NEW 129	matches: dict[str, schemas.DetectionField] = {}	×
NEW 130	all_searches = rule.get_all_searches()	×
131
NEW 132	if selector is None: # None indicates all.	×
NEW 133	matches = all_searches	×
134	else:
NEW 135	for search_id, search_fields in all_searches.items():	×
NEW 136	if fnmatch.fnmatch(search_id, selector):	×
NEW 137	matches[search_id] = search_fields	×
138
NEW 139	match_all = False	×
NEW 140	if count is None:	×
NEW 141	match_all = True	×
NEW 142	count = len(matches)	×
143
NEW 144	permitted_misses = len(matches) - count	×
145
146	# Now that we have our searches to check, run them
NEW 147	search_hits = 0	×
NEW 148	search_misses = 0	×
NEW 149	for search_id, search_fields in matches.items():	×
NEW 150	if find_matches(event, search_fields, match_all):	×
NEW 151	search_hits += 1	×
152	else:
NEW 153	search_misses += 1	×
154
155	# Short circuit if we found the matches, or if we can't find the number anymore
156
NEW 157	if search_hits >= count:	×
NEW 158	return True	×
159
NEW 160	if search_misses > permitted_misses:	×
NEW 161	return False	×
162
NEW 163	return False	×

ninoseki / abuse_whois / 4020719131

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous