diff --git a/sigma/cli/check.py b/sigma/cli/check.py index b2565cd..1a6359b 100644 --- a/sigma/cli/check.py +++ b/sigma/cli/check.py @@ -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 @@ -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 # ========================================== @@ -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 @@ -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: @@ -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 @@ -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." +) @click.option( "--exclude", "-x", @@ -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).""" @@ -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()) @@ -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')}" @@ -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) @@ -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)) diff --git a/tests/test_check.py b/tests/test_check.py index 5fa0c39..006da08 100644 --- a/tests/test_check.py +++ b/tests/test_check.py @@ -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(): @@ -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"