Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
111 changes: 108 additions & 3 deletions sigma/cli/check.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,7 @@
from collections import Counter
from sys import stderr
from textwrap import fill
import xml.etree.ElementTree as ET

from dataclasses import fields

Expand All @@ -16,6 +17,62 @@

severity_color = {"low": "green", "medium": "yellow", "high": "red"}

# JUnit-specific icons covering PySigma validation and parsing possibilities
SEVERITY_ICONS = {
"critical": "💥",
"high": "🔴",
"medium": "🟠",
"low": "🟡",
"informational": "ℹ️",
"parsing_error": "🚫",
"condition_error": "⚠️",
"error": "⚠️", # Fallback for other SigmaError subclasses
"ok": "✅"
}

def generate_junit_report(results, output_file):
"""Generates JUnit XML grouped by PySigma error/issue types."""
root = ET.Element("testsuites", name="Sigma Rule Validation")

suites = {}
for res in results:
suite_name = res.get('issue_type', 'Validation Success')
if suite_name not in suites:
suites[suite_name] = []
suites[suite_name].append(res)

for suite_name, tests in suites.items():
failures = len([t for t in tests if t['status'] == 'failed'])
test_suite = ET.SubElement(
root, "testsuite",
name=suite_name,
tests=str(len(tests)),
failures=str(failures)
)

for res in tests:
# Map the severity string to the icon mapping, fallback to error
severity = res.get('severity', 'ok').lower()
icon = SEVERITY_ICONS.get(severity, SEVERITY_ICONS['error'])

test_case = ET.SubElement(
test_suite, "testcase",
name=f"{icon} {res['rule_name']}",
classname=suite_name,
file=res.get('file_path', 'unknown')
)

if res['status'] == 'failed':
failure = ET.SubElement(
test_case, "failure",
message=f"{res.get('issue_type')}: {res.get('severity', '').upper()}"
)
failure.text = res.get('description', '')

tree = ET.ElementTree(root)
with open(output_file, "wb") as f:
tree.write(f, encoding="utf-8", xml_declaration=True)

# ==========================================
# Data Processing & Extraction Functions
# ==========================================
Expand Down Expand Up @@ -70,7 +127,7 @@ def validate_loaded_rules(check_rules, rule_validator):
return issues


def load_and_check_rules(input, file_pattern, rule_errors, cond_errors):
def load_and_check_rules(input, file_pattern, rule_errors, cond_errors, junit_results=None):
rule_collection = load_rules(input, file_pattern)
check_rules = list()
first_error = True
Expand All @@ -85,6 +142,16 @@ def load_and_check_rules(input, file_pattern, rule_errors, cond_errors):
for error in rule.errors:
click.echo(error)
rule_errors.update((error.__class__.__name__,))
if junit_results is not None:
# Extract error type dynamically (SigmaParsingError vs SigmaValueError etc.)
error_type = error.__class__.__name__
rule_name = rule.title or str(rule.id) or "Unknown Rule"
file_path = str(rule.source) if rule.source else "unknown"
sev = "parsing_error" if "Parse" in error_type else "error"
junit_results.append({
"rule_name": rule_name, "file_path": file_path, "status": "failed",
"issue_type": error_type, "severity": sev, "description": str(error)
})
elif isinstance(rule, SigmaRule): # rule has no errors, parse condition
try:
for condition in rule.detection.parsed_condition:
Expand All @@ -96,6 +163,13 @@ def load_and_check_rules(input, file_pattern, rule_errors, cond_errors):
f"Condition error in { str(condition.source) }:{ error }"
)
cond_errors.update((error,))
if junit_results is not None:
rule_name = rule.title or str(rule.id) or "Unknown Rule"
file_path = str(rule.source) if rule.source else "unknown"
junit_results.append({
"rule_name": rule_name, "file_path": file_path, "status": "failed",
"issue_type": e.__class__.__name__, "severity": "condition_error", "description": error
})
else:
check_rules.append(rule)
return check_rules
Expand Down Expand Up @@ -128,6 +202,11 @@ def load_and_check_rules(input, file_pattern, rule_errors, cond_errors):
show_default=True,
help="Fail on Sigma rule validation issues.",
)
@click.option(
"--junitxml",
type=click.Path(path_type=pathlib.Path),
help="Output results in JUnit XML format to the specified file."
)
Comment thread
thomaspatzke marked this conversation as resolved.
@click.option(
"--exclude",
"-x",
Expand All @@ -143,7 +222,7 @@ def load_and_check_rules(input, file_pattern, rule_errors, cond_errors):
type=click.Path(exists=True, allow_dash=True, path_type=pathlib.Path),
)
def check(
input, validation_config, file_pattern, fail_on_error, fail_on_issues, exclude
input, validation_config, file_pattern, fail_on_error, fail_on_issues, junitxml, exclude
):
"""Check Sigma rules for validity and best practices (not yet implemented)."""

Expand All @@ -152,7 +231,8 @@ def check(
try:
rule_errors = Counter()
cond_errors = Counter()
check_rules = load_and_check_rules(input, file_pattern, rule_errors, cond_errors)
junit_results = [] if junitxml else None
check_rules = load_and_check_rules(input, file_pattern, rule_errors, cond_errors, junit_results)

# TODO: From Python 3.10 the commented line below can be used.
rule_error_count = sum(rule_errors.values())
Expand All @@ -174,6 +254,17 @@ def check(
for rule_with_issue in issue.rules
]
)
if junitxml:
for rule in issue.rules:
junit_results.append({
"rule_name": rule.title or str(rule.path),
"file_path": str(rule.source) if rule.source else "unknown",
"status": "failed",
"issue_type": type(issue).__name__,
"severity": issue.severity.name.lower(),
"description": str(issue.description)
})

additional_fields = " ".join(
[
f"{field.name}={click.style(issue.__getattribute__(field.name) or '-', bold=True, fg='blue')}"
Expand Down Expand Up @@ -267,6 +358,10 @@ def check(
else:
click.echo("No validation issues found.")

if junitxml:
generate_junit_report(junit_results, junitxml)
click.echo(f"\nJUnit report saved to: {junitxml}")

if (
fail_on_error
and (rule_error_count > 0 or cond_error_count > 0)
Expand All @@ -276,4 +371,14 @@ def check(
click.echo("Check failure")
click.get_current_context().exit(1)
except SigmaError as e:
# Handles total collection parsing crashes (e.g., SigmaCollectionError)
if junitxml:
generate_junit_report([{
"rule_name": "Global/Collection Loading Error",
"file_path": str(input),
"status": "failed",
"issue_type": e.__class__.__name__,
"severity": "error",
"description": str(e)
}], junitxml)
raise click.ClickException("Check error: " + str(e))
61 changes: 61 additions & 0 deletions tests/test_check.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,9 @@
from click.testing import CliRunner

from sigma.cli.check import check
import xml.etree.ElementTree as ET
from sigma.cli.check import generate_junit_report
import pathlib


def test_check_help():
Expand Down Expand Up @@ -88,3 +91,61 @@ def test_check_exclude():
assert "Invalid validators name" in result.stdout
assert "myvalidator" in result.stdout
assert "Check failure" in result.stdout


def test_generate_junit_report_writes_file(tmp_path):
# Prepare sample results with one failed and one passed test
results = [
{
"rule_name": "rule_one",
"file_path": "tests/files/valid/sigma_rule.yml",
"status": "failed",
"issue_type": "TestIssue",
"severity": "high",
"description": "Something went wrong",
},
{
"rule_name": "rule_two",
"file_path": "tests/files/valid/sigma_rule.yml",
"status": "passed",
"issue_type": "Validation Success",
"severity": "ok",
"description": "All good",
},
]

out = tmp_path / "junit.xml"
generate_junit_report(results, str(out))

# Ensure file exists and is valid XML with expected structure
assert out.exists()
tree = ET.parse(str(out))
root = tree.getroot()
assert root.tag == "testsuites"

suites = {ts.get("name"): ts for ts in root.findall("testsuite")}
assert "TestIssue" in suites
assert "Validation Success" in suites

test_suite = suites["TestIssue"]
assert test_suite.get("tests") == "1"
assert test_suite.get("failures") == "1"

# Check testcase contains the icon for high severity (🔴)
testcase = test_suite.find("testcase")
assert testcase is not None
assert testcase.get("name").startswith("🔴 ")

failure = testcase.find("failure")
assert failure is not None
assert failure.get("message") == "TestIssue: HIGH"


def test_check_cli_generates_junitxml(tmp_path):
cli = CliRunner()
out = tmp_path / "report.xml"
result = cli.invoke(check, ["--junitxml", str(out), "tests/files/valid"])
assert result.exit_code == 0
assert out.exists()
tree = ET.parse(str(out))
assert tree.getroot().tag == "testsuites"
Loading