[high] Write finalized, postprocessed queries with --output-dir

sigma/cli/check.py

@@ -172,28 +172,6 @@ def load_and_check_rules(input, file_pattern, rule_errors, cond_errors, junit_re
                     })
         else:
             check_rules.append(rule)
-
-    # Errors that are not attached to any loaded rule, e.g. an unknown collection action or
-    # an invalid filter (filters are kept in rule_collection.filters, not in .rules).
-    rule_error_ids = {
-        id(error) for rule in rule_collection.rules for error in rule.errors
-    }
-    for error in rule_collection.errors:
-        if id(error) in rule_error_ids:
-            continue
-        if first_error:
-            click.echo("=== Sigma Rule Errors ===")
-            first_error = False
-        click.echo(error)
-        rule_errors.update((error.__class__.__name__,))
-        if junit_results is not None:
-            error_type = error.__class__.__name__
-            source = getattr(error, "source", None)
-            file_path = str(source) if source else "unknown"
-            junit_results.append({
-                "rule_name": file_path, "file_path": file_path, "status": "failed",
-                "issue_type": error_type, "severity": "error", "description": str(error)
-            })
     return check_rules
 
 @click.command()

sigma/cli/convert.py

@@ -12,6 +12,7 @@
 from sigma.correlations import SigmaCorrelationRule
 from sigma.conversion.base import Backend
 from sigma.exceptions import (
+    SigmaConversionError,
     SigmaError,
     SigmaPipelineNotAllowedForBackendError,
     SigmaPipelineNotFoundError,
@@ -125,13 +126,16 @@ def write_separate_files(
     base_dir: pathlib.Path,
 ):
     """
-    Convert rules and write each to a separate file using callback mechanism.
-
-    This function uses the callback parameter in backend.convert() to write each
-    converted condition to a separate file. This approach works for both regular
-    rules and correlation rules, as the callback receives the rule source information
-    for each converted condition.
-
+    Convert rules and write each to a separate file.
+
+    The whole collection is converted with backend.convert() so that rule references
+    (e.g. of correlation rules) are resolved. Afterwards, the finalized conversion result
+    of each rule that is meant to be output is written to its own file. The per-condition
+    callback of backend.convert() is not used for this, because it is invoked before the
+    query is finalized for the output format and before query postprocessing of the
+    processing pipeline, and it is also invoked for rules that are not output (e.g. base
+    rules of correlation rules).
+
     Args:
         rule_collection: Collection of Sigma rules to convert
         backend: Backend instance for conversion
@@ -147,28 +151,22 @@ def write_separate_files(
 
     # Track number of files written and conversion results per rule
     files_written = 0
-    rule_results = {}  # Maps rule ID to list of (index, result) tuples
-
-    def write_callback(rule, output_format, index, cond, result):
-        """
-        Callback function called for each converted condition.
-
-        Args:
-            rule: The Sigma rule being converted (SigmaRule or SigmaCorrelationRule)
-            output_format: The output format
-            index: Index of the condition being converted
-            cond: The condition being converted
-            result: The conversion result
-
-        Returns:
-            The result unchanged (we just write it to file as a side effect)
-        """
-        nonlocal files_written
-
-        # Skip None results
-        if result is None:
-            return result
-
+    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)
+
+    # Collect the finalized conversion results of all rules that are output
+    for rule in rule_collection.get_output_rules():
+        try:
+            results = rule.get_conversion_result()
+        except SigmaConversionError:
+            # Rule was not converted (conversion failed or was aborted)
+            continue
+
         # Get rule ID for grouping results - Sigma rules should always have an id or title
         if hasattr(rule, 'id') and rule.id:
             rule_id = rule.id
@@ -180,31 +178,12 @@ def write_callback(rule, output_format, index, cond, result):
             rule_hash = hashlib.sha256(str(rule).encode()).hexdigest()[:16]
             rule_id = f"unknown_{rule_hash}"
             click.echo(f"Warning: Rule has no ID or title, using generated identifier: {rule_id}", err=True)
-
-        # Store result for this rule
-        if rule_id not in rule_results:
-            rule_results[rule_id] = []
-        rule_results[rule_id].append((result, rule))
-
-        return result
-
-    # 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)
 
+        # Store results for this rule
+        rule_results.setdefault(rule_id, []).extend(
+            (result, rule) for result in results if result is not None
+        )
+
     # Now write the collected results to files
     for rule_id, results in rule_results.items():
         if not results:
@@ -240,9 +219,9 @@ def write_callback(rule, output_format, index, cond, result):
                 click.echo(f"Warning: Backend returned unexpected format '{type(result).__name__}' for rule '{rule.title}' (source: {rule.source}). Expected str, bytes, or dict. Result will not be written to file.", err=True)
         else:
             # Multiple results, add sequential index to filename
-            # We use enumerate for sequential numbering (1, 2, 3...) instead of the callback index
-            # because the callback index represents the condition number within the rule, which may
-            # not be sequential or may have gaps. We want consistent, predictable filenames.
+            # We use enumerate for sequential numbering (1, 2, 3...) instead of the condition index
+            # because conditions that produce no query leave gaps. We want consistent, predictable
+            # filenames.
             for file_idx, (result, _) in enumerate(results, start=1):
                 output_path = output_dir / render_output_filename(filename_template, rule_source_path, base_dir, file_idx)
                 output_path.parent.mkdir(parents=True, exist_ok=True)
@@ -261,11 +240,6 @@ 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(

tests/test_check.py

@@ -149,54 +149,3 @@ def test_check_cli_generates_junitxml(tmp_path):
     assert out.exists()
     tree = ET.parse(str(out))
     assert tree.getroot().tag == "testsuites"
-
-
-def test_check_unknown_collection_action(tmp_path):
-    """Collection-level errors (not attached to any rule) must be reported and fail the check."""
-    (tmp_path / "typo.yml").write_text("action: repaet\ntitle: typo in action keyword\n")
-    cli = CliRunner()
-    result = cli.invoke(check, [str(tmp_path)])
-    assert result.exit_code == 1
-    assert "Unknown Sigma collection action 'repaet'" in result.stdout
-    assert "Found 1 errors" in result.stdout
-
-
-def test_check_invalid_filter(tmp_path):
-    """Errors in filters (kept in SigmaCollection.filters, not .rules) must fail the check."""
-    (tmp_path / "filter.yml").write_text(
-        "title: filter with invalid id\n"
-        "id: not-a-uuid\n"
-        "logsource:\n"
-        "  category: test\n"
-        "filter:\n"
-        "  rules: []\n"
-        "  selection:\n"
-        "    User: x\n"
-        "  condition: not selection\n"
-    )
-    cli = CliRunner()
-    result = cli.invoke(check, [str(tmp_path)])
-    assert result.exit_code == 1
-    assert "SigmaIdentifierError" in result.stdout
-    assert "Found 1 errors" in result.stdout
-
-
-def test_check_rule_errors_not_double_counted():
-    """Per-rule errors are also contained in SigmaCollection.errors and must be counted once."""
-    cli = CliRunner()
-    result = cli.invoke(check, ["tests/files/invalid"])
-    assert result.exit_code == 1
-    assert "Found 6 errors, 1 condition errors" in result.stdout
-
-
-def test_check_collection_error_junitxml(tmp_path):
-    rules = tmp_path / "rules"
-    rules.mkdir()
-    (rules / "typo.yml").write_text("action: repaet\ntitle: typo in action keyword\n")
-    out = tmp_path / "report.xml"
-    cli = CliRunner()
-    result = cli.invoke(check, ["--junitxml", str(out), str(rules)])
-    assert result.exit_code == 1
-    tree = ET.parse(str(out))
-    failures = tree.getroot().findall(".//failure")
-    assert any("repaet" in (f.text or "") + (f.get("message") or "") for f in failures)

tests/test_convert.py

@@ -1,5 +1,3 @@
-import pathlib
-
 from click.testing import CliRunner
 import pytest
 from sigma.cli.convert import convert
@@ -412,7 +410,7 @@ def test_convert_output_dir_with_index(tmp_path):
 
 
 def test_convert_output_dir_with_correlation_rules(tmp_path):
-    """Test that correlation rules are supported with --output-dir using callback mechanism."""
+    """Test that correlation rules are supported with --output-dir."""
     cli = CliRunner()
     output_dir = tmp_path / "output"
     result = cli.invoke(
@@ -433,8 +431,8 @@ def test_convert_output_dir_with_correlation_rules(tmp_path):
     # Verify that output files were created
     assert output_dir.exists()
     output_files = list(output_dir.glob("*.txt"))
-    # We should have files for base rules and correlation rules
-    # The exact number depends on how the backend handles correlation rules
+    # Files are written for the output rules (the correlation rules and base rules with
+    # generate: true), the same rules that are converted without --output-dir
     assert len(output_files) > 0, f"Expected output files in {output_dir}, but found none"
 
 
@@ -464,53 +462,80 @@ def test_convert_output_dir_with_filter(tmp_path):
     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()
+def test_convert_output_dir_applies_output_format(tmp_path):
+    """--output-dir must write the query as finalized for the chosen output format."""
+    cli = CliRunner()
+    single = cli.invoke(
+        convert, ["-t", "text_query_test", "-f", "test", "tests/files/valid/sigma_rule.yml"]
     )
-    return input_dir
+    assert single.exit_code == 0
+    output_dir = tmp_path / "output"
+    result = cli.invoke(
+        convert,
+        [
+            "-t",
+            "text_query_test",
+            "-f",
+            "test",
+            "--output-dir",
+            str(output_dir),
+            "tests/files/valid/sigma_rule.yml",
+        ],
+    )
+    assert result.exit_code == 0
+    content = (output_dir / "sigma_rule.txt").read_text()
+    assert content.startswith("[ ") and content.endswith(" ]")
+    assert content == single.stdout.strip()
 
 
-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)
+def test_convert_output_dir_applies_pipeline_postprocessing(tmp_path):
+    """--output-dir must apply query postprocessing items of processing pipelines."""
+    pipeline = tmp_path / "embed.yml"
+    pipeline.write_text(
+        "name: embed\n"
+        "priority: 100\n"
+        "postprocessing:\n"
+        "  - type: embed\n"
+        "    prefix: 'index=prod ('\n"
+        "    suffix: ')'\n"
+    )
     output_dir = tmp_path / "output"
     cli = CliRunner()
     result = cli.invoke(
         convert,
-        ["-t", "text_query_test", "--output-dir", str(output_dir), str(input_dir)],
+        [
+            "-t",
+            "text_query_test",
+            "-p",
+            str(pipeline),
+            "--output-dir",
+            str(output_dir),
+            "tests/files/valid/sigma_rule.yml",
+        ],
     )
-    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()
+    assert result.exit_code == 0
+    content = (output_dir / "sigma_rule.txt").read_text()
+    assert content.startswith("index=prod (") and content.endswith(")")
 
 
-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"
+def test_convert_output_dir_skips_non_output_correlation_base_rules(tmp_path):
+    """Base rules that are not output (generate: false) must not be written with --output-dir."""
     cli = CliRunner()
+    output_dir = tmp_path / "output"
     result = cli.invoke(
         convert,
-        ["-t", "text_query_test", "-s", "--output-dir", str(output_dir), str(input_dir)],
+        [
+            "-t",
+            "text_query_test",
+            "-c",
+            "test",
+            "--output-dir",
+            str(output_dir),
+            "tests/files/sigma_correlation_rules.yml",
+        ],
     )
     assert result.exit_code == 0
-    assert "Ignored errors" in result.output
-    assert (output_dir / "b_rule.txt").exists()
+    # The file holds three correlation rules and three base rules; only the three
+    # correlation rules are output (the same three queries that -o prints).
+    assert "Wrote 3 file(s)" in result.stderr