admin/olafhartong-sysmon-modular
1
0
Fork 0
mirror of https://github.com/olafhartong/sysmon-modular synced 2026-06-08 16:29:39 +00:00
Code Issues Packages Projects Releases Wiki Activity

add template, update merge python script

Browse Source
This commit is contained in:
Connor Shade
2023-04-13 06:02:51 -04:00
parent 16b1ca42cf
commit 4790ccfa6f
2 changed files with 132 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files
merge_sysmon_configs.py
+101 -33
View File
@@ -1,17 +1,11 @@
"""
merge_sysmon_configs.py
Disclaimer: The following script was generated using GPT-4 through chat.openai.com
The conversation can be found here: https://gist.github.com/cnnrshd/7d76c2956cf0ff3f46e04f61f582af60
Modifications past the initial commit are (likely) human - only modifications I've made are comments.
This script is used to merge multiple Sysmon configuration files based on their priority.
The highest priority is at the top.
This script is used to merge multiple Sysmon configuration files based on their priority - highest at top.
It reads a list of configuration file paths and priorities from an input TSV, CSV, or JSON file.
The merged Sysmon configuration is printed to the console.
Usage:
python merge_sysmon_configs.py <input_file> [-f/--format tsv/csv/json]
python merge_sysmon_configs.py <input_file> [-f/--format tsv/csv/json] -b <template_file> -o <output_file>
Arguments:
input_file: Path to the TSV, CSV, or JSON file containing filepaths and priorities.
@@ -24,18 +18,32 @@ Arguments:
-f/--format: Optional flag to override file format detection.
Supported formats are tsv, csv, and json.
-o/--outfile: Where to write the output to. Defaults to stdout.
-b/--base-config: A template config to insert compiled config into. See sysmon_template.xml.
This is also where banners can be modified.
The script will merge the Sysmon configurations based on their type, subtype, and priority.
Configurations with the same type and subtype will be merged within the same heading.
Type refers to the Event Type (ProcessCreate, FileCreate), subtype referes to include/exclude
Disclaimer: The following script was generated in part using GPT-4 through chat.openai.com
Disclaimer is provided in accordance with OpenAI's Usage Policy and Terms of User
The conversation can be found here: https://gist.github.com/cnnrshd/7d76c2956cf0ff3f46e04f61f582af60
"""
import argparse
import json
import csv
import json
import logging
import os
from pathlib import Path
from typing import Dict, List, Tuple, Union
from lxml import etree
from typing import List, Dict, Tuple, Union
from packaging.version import parse as vparse
logging.basicConfig(level=logging.INFO)
def detect_file_format(file_path: str) -> str:
"""
@@ -50,6 +58,7 @@ def detect_file_format(file_path: str) -> str:
_, ext = os.path.splitext(file_path)
file_format = ext[1:].lower()
if file_format not in ["tsv", "csv", "json"]:
logging.exception(f"Received file format {file_format}")
raise ValueError("Unsupported file format")
return file_format
@@ -66,16 +75,20 @@ def read_file_list(file_path: str, file_format: str) -> List[Dict[str, Union[str
"""
file_list = []
if file_format == "tsv":
logging.info(f"Reading file {file_path} as 'tsv'")
with open(file_path, "r") as file:
reader = csv.DictReader(file, delimiter="\t")
file_list = [row for row in reader]
elif file_format == "csv":
logging.info(f"Reading file {file_path} as 'csv'")
with open(file_path, "r") as file:
reader = csv.DictReader(file)
file_list = [row for row in reader]
elif file_format == "json":
logging.info(f"Reading file {file_path} as 'json'")
with open(file_path, "r") as file:
file_list = json.load(file)
logging.info(f"Detected {len(file_list)} items in {file_path}")
return file_list
def merge_sysmon_configs(file_list: List[Dict[str, Union[str, int]]]) -> etree.Element:
@@ -91,34 +104,84 @@ def merge_sysmon_configs(file_list: List[Dict[str, Union[str, int]]]) -> etree.E
merged_sysmon = etree.Element("Sysmon")
merged_event_filtering = etree.SubElement(merged_sysmon, "EventFiltering")
event_dict: Dict[Tuple[str, str], etree.Element] = {}
versions_set = set()
for file_info in sorted(file_list, key=lambda x: int(x["priority"])):
for file_info in sorted(file_list, key=lambda x: int(x["priority"]),reverse=True):
logging.debug(f"Working with {file_info}")
file_path = file_info["filepath"]
if Path(file_path).is_file():
with open(file_path, "r") as file:
tree = etree.parse(file)
rule_group = tree.find(".//RuleGroup")
for event in rule_group:
event_type = event.tag
onmatch = event.get("onmatch")
key = (event_type, onmatch)
tree = etree.parse(file_path,parser=etree.XMLParser(remove_blank_text=True))
# grab schema version
version = tree.getroot().get("schemaversion")
try:
versions_set.add(vparse(version))
except Exception as e:
logging.exception(f"Error parsing version {version} in file {file_path}, skipping file")
continue
rule_group = tree.find(".//RuleGroup")
for event in rule_group:
event_type = event.tag
onmatch = event.get("onmatch")
key = (event_type, onmatch)
if key not in event_dict:
event_dict[key] = event
merged_event_filtering.append(rule_group)
else:
for child in event:
event_dict[key].append(child)
if key not in event_dict:
event_dict[key] = event
merged_event_filtering.append(rule_group)
else:
for child in event:
event_dict[key].append(child)
else:
logging.warning(f"Provided invalid path {file_path}, will not be merged")
versions_list = list(versions_set)
versions_list.sort(reverse=True)
logging.debug(f"Versions found across all files: {[str(vers) for vers in versions_list]}")
merged_sysmon.set("schemaversion", str(versions_list[0]))
return merged_sysmon
def merge_with_base_config(merged_sysmon: etree.Element, base_config_file: str) -> etree.Element:
"""
Merge the base config with the merged Sysmon configurations.
Args:
merged_sysmon: Merged Sysmon configurations.
base_config_file: Path to the base config file.
Returns:
Merged Sysmon configurations with the base config.
"""
base_tree = etree.parse(base_config_file.name, parser=etree.XMLParser(remove_blank_text=True))
base_root = base_tree.getroot()
base_event_filtering = base_root.find("EventFiltering")
if base_event_filtering is not None:
base_root.remove(base_event_filtering)
new_event_filtering = etree.Element("EventFiltering")
for rule_group in merged_sysmon.findall("EventFiltering/RuleGroup"):
new_event_filtering.append(rule_group)
base_root.set("schemaversion",merged_sysmon.get("schemaversion"))
base_root.append(new_event_filtering)
return base_root
def main() -> None:
parser = argparse.ArgumentParser(description="Merge Sysmon config files")
parser.add_argument("file", help="Path to the TSV, CSV, or JSON file containing filepaths and priorities")
parser.add_argument("file", type=argparse.FileType("r"), help="Path to the TSV, CSV, or JSON file containing filepaths and priorities")
parser.add_argument("-f", "--format", choices=["tsv", "csv", "json"], help="Override file format detection")
parser.add_argument("-o", "--outfile", type=argparse.FileType("w"), default="-", help="File to output to, defaults to stdout")
parser.add_argument("-b", "--base-config", type=argparse.FileType("r"), help="Path to the base config file with top-level Sysmon elements")
parser.add_argument("--debug", action="store_true", default=False, help="Enable debug logging")
args = parser.parse_args()
file_path = args.file
if args.debug:
logging.getLogger().setLevel(logging.DEBUG)
file_path = args.file.name
try:
if not args.format:
@@ -126,22 +189,27 @@ def main() -> None:
else:
file_format = args.format
except ValueError as e:
print(f"Error: {e}")
return
logging.exception(f"Error with file type: {e}")
exit(1)
try:
file_list = read_file_list(file_path, file_format)
except Exception as e:
print(f"Error reading file list: {e}")
return
logging.exception(f"Error reading file list: {e}")
exit(1)
try:
merged_sysmon = merge_sysmon_configs(file_list)
if args.base_config:
merged_sysmon = merge_with_base_config(merged_sysmon, args.base_config)
except TypeError as e:
logging.exception(f"Likely missing a priority in config list: {e}")
exit(1)
except Exception as e:
print(f"Error merging Sysmon configs: {e}")
return
logging.exception(f"Error merging Sysmon configs: {e}")
exit(1)
print(etree.tostring(merged_sysmon, pretty_print=True).decode())
args.outfile.write(etree.tostring(merged_sysmon, pretty_print=True).decode())
if __name__ == "__main__":
main()
templates/sysmon_template.xml
+31
View File
@@ -0,0 +1,31 @@
<Sysmon schemaversion="4.21">
<!-- NOTICE : This is a modified version of Sysmon-Modular that is restructured to enhance -->
<!-- config layering and MITRE ATT&CK Technique mapping -->
<!-- for more information go to https://github.com/olafhartong/sysmon-modular/wiki -->
<!-- -->
<!-- //** ***// -->
<!-- ///#(** **%(/// -->
<!-- ((&&&** **&&&(( -->
<!-- (&&&** ,(((((((. **&&&( -->
<!-- ((&&**(((((//(((((((/**&&(( _____ __ __ -->
<!-- (&&///((////(((((((///&&( / ___/__ ___________ ___ ____ ____ ____ ___ ____ ____/ /_ __/ /___ ______ -->
<!-- &////(/////(((((/(////& \__ \/ / / / ___/ __ `__ \/ __ \/ __ \______/ __ `__ \/ __ \/ __ / / / / / __ `/ ___/ -->
<!-- ((// /////(///// /((( ___/ / /_/ (__ ) / / / / / /_/ / / / /_____/ / / / / / /_/ / /_/ / /_/ / / /_/ / / -->
<!-- &(((((#.///////// #(((((& /____/\__, /____/_/ /_/ /_/\____/_/ /_/ /_/ /_/ /_/\____/\__,_/\__,_/_/\__,_/_/ -->
<!-- &&&&((#///////((#((&&&& /____/ -->
<!-- &&&&(#/***//(#(&&&& -->
<!-- &&&&****///&&&& by Olaf Hartong -->
<!-- (& ,&. & friends -->
<!-- .*&&*. -->
<!-- -->
<HashAlgorithms>*</HashAlgorithms>
<!-- This now also determines the file names of the files preserved (String) -->
<CheckRevocation>False</CheckRevocation>
<!-- Setting this to true might impact performance -->
<DnsLookup>False</DnsLookup>
<!-- Disables lookup behavior, default is True (Boolean) -->
<ArchiveDirectory>Sysmon</ArchiveDirectory>
<!-- Sets the name of the directory in the C:\ root where preserved files will be saved (String)-->
<EventFiltering>
</EventFiltering>
</Sysmon>