Add support for local file contents

Author: eric-forte-elastic

detection_rules/kbwrap.py

@@ -250,6 +250,12 @@ def _process_imported_items(
         '"alert.attributes.tags: \\"test\\"" to filter for rules that have the tag "test"'
     ),
 )
+@click.option(
+    "--load-rule-loading",
+    "-lr",
+    is_flag=True,
+    help="Enable arbitrary rule loading from the rules directory (Can be very slow!)",
+)
 @click.pass_context
 def kibana_export_rules(  # noqa: PLR0912, PLR0913, PLR0915
     ctx: click.Context,
@@ -268,6 +274,7 @@ def kibana_export_rules(  # noqa: PLR0912, PLR0913, PLR0915
     local_updated_date: bool = False,
     custom_rules_only: bool = False,
     export_query: str | None = None,
+    load_rule_loading: bool = False,
 ) -> list[TOMLRule]:
     """Export custom rules from Kibana."""
     kibana = ctx.obj["kibana"]
@@ -277,6 +284,10 @@ def kibana_export_rules(  # noqa: PLR0912, PLR0913, PLR0915
     if rule_name and rule_id:
         raise click.UsageError("Cannot use --rule-id and --rule-name together. Please choose one.")
 
+    rules = None
+    if load_rule_loading:
+        rules = RuleCollection.default()
+
     with kibana:
         # Look up rule IDs by name if --rule-name was provided
         if rule_name:
@@ -358,10 +369,16 @@ def kibana_export_rules(  # noqa: PLR0912, PLR0913, PLR0915
             tactic_name = first_tactic if not no_tactic_filename else None  # type: ignore[reportUnknownMemberType]
             rule_name = rulename_to_filename(rule_resource.get("name"), tactic_name=tactic_name)  # type: ignore[reportUnknownMemberType]
 
+            local_contents = None
             save_path = directory / f"{rule_name}"
+            if rules and params.get("rule_id") in rules.id_map:
+                save_path = rules.id_map[params["rule_id"]].path
+                local_contents = rules.id_map[params["rule_id"]].contents
             params.update(
                 update_metadata_from_file(
-                    save_path, {"creation_date": local_creation_date, "updated_date": local_updated_date}
+                    {"creation_date": local_creation_date, "updated_date": local_updated_date},
+                    save_path,
+                    local_contents,
                 )
             )
             contents = TOMLRuleContents.from_rule_resource(**params)  # type: ignore[reportArgumentType]

detection_rules/main.py

@@ -157,6 +157,12 @@ def generate_rules_index(
 @click.option("--strip-none-values", "-snv", is_flag=True, help="Strip None values from the rule")
 @click.option("--local-creation-date", "-lc", is_flag=True, help="Preserve the local creation date of the rule")
 @click.option("--local-updated-date", "-lu", is_flag=True, help="Preserve the local updated date of the rule")
+@click.option(
+    "--load-rule-loading",
+    "-lr",
+    is_flag=True,
+    help="Enable arbitrary rule loading from the rules directory (Can be very slow!)",
+)
 def import_rules_into_repo(  # noqa: PLR0912, PLR0913, PLR0915
     input_file: tuple[Path, ...] | None,
     required_only: bool,
@@ -171,6 +177,7 @@ def import_rules_into_repo(  # noqa: PLR0912, PLR0913, PLR0915
     strip_none_values: bool,
     local_creation_date: bool,
     local_updated_date: bool,
+    load_rule_loading: bool,
 ) -> None:
     """Import rules from json, toml, or yaml files containing Kibana exported rule(s)."""
     errors: list[str] = []
@@ -189,6 +196,10 @@ def import_rules_into_repo(  # noqa: PLR0912, PLR0913, PLR0915
     if not file_contents:
         click.echo("Must specify at least one file!")
 
+    rules = None
+    if load_rule_loading:
+        rules = RuleCollection.default()
+
     exceptions_containers = {}
     exceptions_items = {}
 
@@ -210,7 +221,14 @@ def import_rules_into_repo(  # noqa: PLR0912, PLR0913, PLR0915
         base_path = rulename_to_filename(base_path) if base_path else base_path
         if base_path is None:
             raise ValueError(f"Invalid rule file, please ensure the rule has a name field: {contents}")
-        rule_path = Path(os.path.join(str(save_directory) if save_directory else RULES_DIRS[0], base_path))  # noqa: PTH118
+
+        local_contents = None
+        rule_path = Path(
+            os.path.join(str(save_directory) if save_directory else RULES_DIRS[0], base_path)  # noqa: PTH118
+        )
+        if rules and contents.get("rule_id") in rules.id_map:
+            rule_path = rules.id_map[contents["rule_id"]].path
+            local_contents = rules.id_map[contents["rule_id"]].contents
 
         # handle both rule json formats loaded from kibana and toml
         data_view_id = contents.get("data_view_id") or contents.get("rule", {}).get("data_view_id")
@@ -226,7 +244,7 @@ def import_rules_into_repo(  # noqa: PLR0912, PLR0913, PLR0915
 
         contents.update(
             update_metadata_from_file(
-                Path(rule_path), {"creation_date": local_creation_date, "updated_date": local_updated_date}
+                {"creation_date": local_creation_date, "updated_date": local_updated_date}, rule_path, local_contents
             )
         )
 

detection_rules/rule_loader.py

@@ -139,14 +139,20 @@ def load_locks_from_tag(
     return commit_hash, version, deprecated
 
 
-def update_metadata_from_file(rule_path: Path, fields_to_update: dict[str, Any]) -> dict[str, Any]:
-    """Update metadata fields for a rule with local contents."""
+def update_metadata_from_file(
+    fields_to_update: dict[str, Any], rule_path: Path | None = None, local_contents: TOMLRuleContents | None = None
+) -> dict[str, Any]:
+    """Update metadata fields for a rule with local contents or provided contents."""
 
     contents: dict[str, Any] = {}
-    if not rule_path.exists():
-        return contents
-
-    rule_contents = RuleCollection().load_file(rule_path).contents
+    if not rule_path and not local_contents:
+        raise ValueError("Either 'rule_path' or 'local_contents' must be provided.")
+
+    rule_contents = None
+    if rule_path:
+        if not rule_path.exists():
+            return contents
+        rule_contents = local_contents or RuleCollection().load_file(rule_path).contents
 
     if not isinstance(rule_contents, TOMLRuleContents):
         raise TypeError("TOML rule expected")

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "detection_rules"
-version = "1.3.10"
+version = "1.3.11"
 description = "Detection Rules is the home for rules used by Elastic Security. This repository is used for the development, maintenance, testing, validation, and release of rules for Elastic Security’s Detection Engine."
 readme = "README.md"
 requires-python = ">=3.12"