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
28 changes: 22 additions & 6 deletions sigma/cli/convert.py
Original file line number Diff line number Diff line change
Expand Up @@ -188,12 +188,23 @@ def write_callback(rule, output_format, index, cond, result):

return result

# Convert the entire collection with the callback
try:
backend.convert(rule_collection, format, correlation_method, callback=write_callback)
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.
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, callback=write_callback
)
else:
backend.convert_rule(rule, format, callback=write_callback)
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)

# Now write the collected results to files
for rule_id, results in rule_results.items():
if not results:
Expand Down Expand Up @@ -250,6 +261,11 @@ def write_callback(rule, output_format, index, cond, result):

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
53 changes: 53 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 @@ -461,3 +463,54 @@ def test_convert_output_dir_with_filter(tmp_path):
content = (output_dir / "sigma_rule.txt").read_text()
assert 'not User startswith "ADM_"' in content


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