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
25 changes: 20 additions & 5 deletions sigma/cli/convert.py
Original file line number Diff line number Diff line change
Expand Up @@ -153,11 +153,21 @@ def write_separate_files(
files_written = 0
rule_results = {} # Maps rule ID to list of (result, rule) tuples

# Convert the entire collection
try:
backend.convert(rule_collection, format, correlation_method)
except Exception as e:
click.echo(f"Warning: Failed to convert rules: {e}", err=True)
# Convert rule by rule (mirroring Backend.convert) so that one failing rule neither
# aborts the conversion of the remaining rules nor gets silently ignored. Each
# successful conversion stores the finalized result on the rule, collected below.
failed_rules = []
backend.init_processing_pipeline(format)
rule_collection.resolve_rule_references()
for rule in rule_collection.rules:
try:
if isinstance(rule, SigmaCorrelationRule):
backend.convert_correlation_rule(rule, format, correlation_method)
else:
backend.convert_rule(rule, format)
except (SigmaError, NotImplementedError) as e:
failed_rules.append((rule, e))
click.echo(f"Error: Failed to convert rule {rule.source or rule.title}: {e}", err=True)

# Collect the finalized conversion results of all rules that are output
for rule in rule_collection.get_output_rules():
Expand Down Expand Up @@ -240,6 +250,11 @@ def write_separate_files(

click.echo(f"Wrote {files_written} file(s) to {output_dir}", err=True)

if failed_rules:
raise click.ClickException(
f"{len(failed_rules)} rule(s) failed to convert, see errors above."
)


@click.command()
@click.option(
Expand Down
52 changes: 52 additions & 0 deletions tests/test_convert.py
Original file line number Diff line number Diff line change
@@ -1,3 +1,5 @@
import pathlib

from click.testing import CliRunner
import pytest
from sigma.cli.convert import convert
Expand Down Expand Up @@ -539,3 +541,53 @@ def test_convert_output_dir_skips_non_output_correlation_base_rules(tmp_path):
# correlation rules are output (the same three queries that -o prints).
assert "Wrote 3 file(s)" in result.stderr


UNCONVERTIBLE_RULE = """title: Unconvertible
id: 9f1b8f4a-0000-4000-8000-000000000001
logsource:
category: test
detection:
sel:
fieldA|expand: "%var%"
condition: sel
"""


def _write_rules_with_unconvertible(tmp_path):
input_dir = tmp_path / "rules"
input_dir.mkdir()
(input_dir / "a_unconvertible.yml").write_text(UNCONVERTIBLE_RULE)
(input_dir / "b_rule.yml").write_text(
pathlib.Path("tests/files/valid/sigma_rule.yml").read_text()
)
return input_dir


def test_convert_output_dir_conversion_error_fails_and_continues(tmp_path):
"""A rule that fails to convert makes --output-dir exit non-zero, but later rules are still written."""
input_dir = _write_rules_with_unconvertible(tmp_path)
output_dir = tmp_path / "output"
cli = CliRunner()
result = cli.invoke(
convert,
["-t", "text_query_test", "--output-dir", str(output_dir), str(input_dir)],
)
assert result.exit_code == 1
assert "a_unconvertible.yml" in result.stderr
assert "1 rule(s) failed to convert" in result.stderr
assert not (output_dir / "a_unconvertible.txt").exists()
assert (output_dir / "b_rule.txt").exists()


def test_convert_output_dir_conversion_error_skip_unsupported(tmp_path):
"""With --skip-unsupported the failing rule is only reported as ignored error."""
input_dir = _write_rules_with_unconvertible(tmp_path)
output_dir = tmp_path / "output"
cli = CliRunner()
result = cli.invoke(
convert,
["-t", "text_query_test", "-s", "--output-dir", str(output_dir), str(input_dir)],
)
assert result.exit_code == 0
assert "Ignored errors" in result.output
assert (output_dir / "b_rule.txt").exists()
Loading