diff --git a/sigma/cli/convert.py b/sigma/cli/convert.py index e4c41b0..9824765 100644 --- a/sigma/cli/convert.py +++ b/sigma/cli/convert.py @@ -1,5 +1,4 @@ import json -import hashlib import os import pathlib import textwrap @@ -151,7 +150,7 @@ 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 (result, rule) tuples + rule_results = {} # Maps id() of the rule object to list of (result, rule) tuples # Convert the entire collection try: @@ -167,79 +166,89 @@ def write_separate_files( # 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 - elif hasattr(rule, 'title') and rule.title: - rule_id = rule.title - else: - # This should rarely happen - Sigma rules should have id or title - # Use stable hash of rule string representation for reproducibility - 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 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: - continue - + # Group results by rule object: ids and titles are not guaranteed to be unique (e.g. two + # rules without id sharing a title) and would merge the results of different rules. + results = [(result, rule) for result in results if result is not None] + if results: + rule_results[id(rule)] = results + + # Determine the output file of each result. Several results can share a file name, e.g. the + # queries of a rule with multiple conditions or of multiple rules contained in one YAML file + # if the template doesn't contain {index}. + outputs = {} # Maps normalized output path to list of (output_path, result, rule) tuples + for results in rule_results.values(): # Get the rule from the first result _, rule = results[0] - + # Get rule source path if rule.source and hasattr(rule.source, 'path'): rule_source_path = pathlib.Path(rule.source.path) else: # If no source path, use rule ID or title as filename rule_source_path = pathlib.Path(f"{rule.id or rule.title}.yml") - - # Write results - if len(results) == 1: - # Single result, no index needed - result, _ = results[0] - output_path = output_dir / render_output_filename(filename_template, rule_source_path, base_dir, None) - output_path.parent.mkdir(parents=True, exist_ok=True) - - if isinstance(result, str): - output_path.write_bytes(bytes(result, encoding)) - files_written += 1 - elif isinstance(result, bytes): - output_path.write_bytes(result) - files_written += 1 - elif isinstance(result, dict): - output_path.write_bytes(bytes(json.dumps(result, indent=json_indent), encoding)) - files_written += 1 - else: - 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) + + # Multiple results get a sequential index (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): + index = file_idx if len(results) > 1 else None + if not isinstance(result, (str, bytes, dict)): + click.echo(f"Warning: Backend returned unexpected format '{type(result).__name__}' for rule '{rule.title}' result {file_idx}/{len(results)} (source: {rule.source}). Expected str, bytes, or dict. This result will not be written to file.", err=True) + continue + output_path = output_dir / render_output_filename(filename_template, rule_source_path, base_dir, index) + outputs.setdefault(os.path.normcase(os.path.abspath(output_path)), []).append( + (output_path, result, rule) + ) + + def to_bytes(result): + if isinstance(result, str): + return bytes(result, encoding) + elif isinstance(result, bytes): + return result else: - # Multiple results, add sequential index to filename - # 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) - - if isinstance(result, str): - output_path.write_bytes(bytes(result, encoding)) - files_written += 1 - elif isinstance(result, bytes): - output_path.write_bytes(result) - files_written += 1 - elif isinstance(result, dict): - output_path.write_bytes(bytes(json.dumps(result, indent=json_indent), encoding)) - files_written += 1 - else: - click.echo(f"Warning: Backend returned unexpected format '{type(result).__name__}' for rule '{rule.title}' result {file_idx}/{len(results)} (source: {rule.source}). Expected str, bytes, or dict. This result will not be written to file.", err=True) - + return bytes(json.dumps(result, indent=json_indent), encoding) + + # Write each output file once. Results of the same source file are joined like in single + # file output, results of different source files (or binary results) collide: only the + # first one is written, the others are reported and not silently overwritten. + collisions = 0 + for items in outputs.values(): + output_path, first_result, first_rule = items[0] + joined, colliding = [items[0]], [] + for item in items[1:]: + _, result, rule = item + if ( + str(rule.source) != str(first_rule.source) + or isinstance(result, bytes) + or isinstance(first_result, bytes) + ): + colliding.append(item) + else: + joined.append(item) + if colliding: + collisions += len(colliding) + click.echo( + f"Error: Output file '{output_path}' is already used for rule '{first_rule.title}' " + f"(source: {first_rule.source}). Not written: " + + ", ".join(f"rule '{rule.title}' (source: {rule.source})" for _, _, rule in colliding), + err=True, + ) + + separator = "\n" if all(isinstance(result, dict) for _, result, _ in joined) else "\n\n" + output_path.parent.mkdir(parents=True, exist_ok=True) + output_path.write_bytes( + bytes(separator, encoding).join(to_bytes(result) for _, result, _ in joined) + ) + files_written += 1 + click.echo(f"Wrote {files_written} file(s) to {output_dir}", err=True) + if collisions: + raise click.ClickException( + f"{collisions} result(s) not written because their output file names collide. " + "Use {path} and/or {index} in --output-filename-template to get distinct file names." + ) + @click.command() @click.option( diff --git a/tests/test_convert.py b/tests/test_convert.py index b02d51f..ec5ba34 100644 --- a/tests/test_convert.py +++ b/tests/test_convert.py @@ -536,6 +536,106 @@ def test_convert_output_dir_skips_non_output_correlation_base_rules(tmp_path): ) assert result.exit_code == 0 # 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 + # correlation rules are output (the same three queries that -o prints). The + # template has no {index}, so they share one output file. + assert "Wrote 1 file(s)" in result.stderr + stdout_result = cli.invoke( + convert, + ["-t", "text_query_test", "-c", "test", "tests/files/sigma_correlation_rules.yml"], + ) + content = (output_dir / "sigma_correlation_rules.txt").read_text() + assert content.strip() == stdout_result.stdout.strip() + + +def _rule_yaml(title, value, rule_id=None): + id_line = f"id: {rule_id}\n" if rule_id else "" + return ( + f"title: {title}\n{id_line}logsource:\n category: test\n" + f"detection:\n sel:\n fieldA: {value}\n condition: sel\n" + ) + + +def test_convert_output_dir_multiple_queries_without_index(tmp_path): + """All queries of a multi-condition rule end up in its file if the template has no {index}.""" + cli = CliRunner() + output_dir = tmp_path / "output" + result = cli.invoke( + convert, + [ + "-t", + "text_query_test", + "--output-dir", + str(output_dir), + "tests/files/multiple_rules/multi_condition.yml", + ], + ) + assert result.exit_code == 0 + assert "Wrote 1 file(s)" in result.stderr + content = (output_dir / "multi_condition.txt").read_text() + for exe in ("first", "second", "third"): + assert f"\\{exe}.exe" in content + + +def test_convert_output_dir_same_stem_different_sources(tmp_path): + """Rules from different source files mapping to the same output file are not overwritten.""" + input_dir = tmp_path / "rules" + (input_dir / "a").mkdir(parents=True) + (input_dir / "b").mkdir() + (input_dir / "a" / "proc.yml").write_text( + _rule_yaml("Proc A", "value_a", "11111111-0000-4000-8000-000000000001") + ) + (input_dir / "b" / "proc.yml").write_text( + _rule_yaml("Proc B", "value_b", "11111111-0000-4000-8000-000000000002") + ) + 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 "Wrote 1 file(s)" in result.stderr + assert "1 result(s) not written" in result.stderr + assert [p.name for p in output_dir.iterdir()] == ["proc.txt"] + content = (output_dir / "proc.txt").read_text() + assert ("value_a" in content) != ("value_b" in content) + + # With {path} in the template both are written. + output_dir = tmp_path / "output_path" + result = cli.invoke( + convert, + [ + "-t", + "text_query_test", + "--output-dir", + str(output_dir), + "--output-filename-template", + "{path}/{stem}.txt", + str(input_dir), + ], + ) + assert result.exit_code == 0 + assert "Wrote 2 file(s)" in result.stderr + assert "value_a" in (output_dir / "a" / "proc.txt").read_text() + assert "value_b" in (output_dir / "b" / "proc.txt").read_text() + + +def test_convert_output_dir_rules_without_id_same_title(tmp_path): + """Rules without id sharing a title are distinct rules and written to their own files.""" + input_dir = tmp_path / "rules" + input_dir.mkdir() + (input_dir / "one.yml").write_text(_rule_yaml("Same title", "value_one")) + (input_dir / "two.yml").write_text(_rule_yaml("Same title", "value_two")) + 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 == 0 + assert "Wrote 2 file(s)" in result.stderr + one = (output_dir / "one.txt").read_text() + two = (output_dir / "two.txt").read_text() + assert "value_one" in one and "value_two" not in one + assert "value_two" in two and "value_one" not in two