Merge branch 'main' into deprecatemac

Author: DefSecSentinel

.github/ISSUE_TEMPLATE/new_meta.yaml

@@ -37,7 +37,7 @@ body:
   - type: textarea
     attributes:
       label: Tasking
-      value: "```[tasklist]\n### Meta Tasks\n- [ ] Provide Week 1 Update Comment\n- [ ] Provide Week 2 Update or Closeout Comment\n```"
+      value: "\n### Meta Tasks\n- [ ] Provide Week 1 Update Comment\n- [ ] Provide Week 2 Update or Closeout Comment\n"
       render:
 
   - type: textarea

.github/workflows/kibana-mitre-update.yml

@@ -16,7 +16,7 @@ jobs:
 
       - name: Get MITRE Attack changed files
         id: changed-attack-files
-        uses: tj-actions/changed-files@v46
+        uses: tj-actions/changed-files@2f7c5bfce28377bc069a65ba478de0a74aa0ca32 # v46.0.1
         with:
           files: detection_rules/etc/attack-v*.json.gz
 

CLI.md

@@ -265,6 +265,7 @@ Options:
   -e, --overwrite-exceptions      Overwrite exceptions in existing rules
   -ac, --overwrite-action-connectors
                                   Overwrite action connectors in existing rules
+  -nt, --no-tactic-filename       Allow rule filenames without tactic prefix. Use this if rules have been exported with this flag.
   -h, --help                      Show this message and exit.
 ```
 
@@ -515,11 +516,13 @@ Options:
                                   Directory to export exceptions to
   -da, --default-author TEXT      Default author for rules missing one
   -r, --rule-id TEXT              Optional Rule IDs to restrict export to
+  -rn, --rule-name TEXT           Optional Rule name to restrict export to (KQL, case-insensitive, supports wildcards)
   -ac, --export-action-connectors
                                   Include action connectors in export
   -e, --export-exceptions         Include exceptions in export
   -s, --skip-errors               Skip errors when exporting rules
   -sv, --strip-version            Strip the version fields from all rules
+  -nt, --no-tactic-filename       Exclude tactic prefix in exported filenames for rules. Use same flag for import-rules to prevent warnings and disable its unit test.
   -h, --help                      Show this message and exit.
 
 ```

detection_rules/cli_utils.py

@@ -23,6 +23,9 @@
                           dict_filter)
 from .schemas import definitions
 from .utils import clear_caches, rulename_to_filename
+from .config import parse_rules_config
+
+RULES_CONFIG = parse_rules_config()
 
 
 def single_collection(f):
@@ -66,11 +69,15 @@ def multi_collection(f):
     @click.option("--directory", "-d", multiple=True, type=click.Path(file_okay=False), required=False,
                   help="Recursively load rules from a directory")
     @click.option("--rule-id", "-id", multiple=True, required=False)
+    @click.option("--no-tactic-filename", "-nt", is_flag=True, required=False,
+                  help="Allow rule filenames without tactic prefix. "
+                  "Use this if rules have been exported with this flag.")
     @functools.wraps(f)
     def get_collection(*args, **kwargs):
         rule_id: List[str] = kwargs.pop("rule_id", [])
         rule_files: List[str] = kwargs.pop("rule_file")
         directories: List[str] = kwargs.pop("directory")
+        no_tactic_filename: bool = kwargs.pop("no_tactic_filename", False)
 
         rules = RuleCollection()
 
@@ -99,7 +106,10 @@ def get_collection(*args, **kwargs):
         for rule in rules:
             threat = rule.contents.data.get("threat")
             first_tactic = threat[0].tactic.name if threat else ""
-            rule_name = rulename_to_filename(rule.contents.data.name, tactic_name=first_tactic)
+            # Check if flag or config is set to not include tactic in the filename
+            no_tactic_filename = no_tactic_filename or RULES_CONFIG.no_tactic_filename
+            tactic_name = None if no_tactic_filename else first_tactic
+            rule_name = rulename_to_filename(rule.contents.data.name, tactic_name=tactic_name)
             if rule.path.name != rule_name:
                 click.secho(
                     f"WARNING: Rule path does not match required path: {rule.path.name} != {rule_name}", fg="yellow"
@@ -210,18 +220,11 @@ def rule_prompt(path=None, rule_type=None, required_only=True, save=True, verbos
     # DEFAULT_PREBUILT_RULES_DIRS[0] is a required directory just as a suggestion
     suggested_path = Path(DEFAULT_PREBUILT_RULES_DIRS[0]) / contents['name']
     path = Path(path or input(f'File path for rule [{suggested_path}]: ') or suggested_path).resolve()
-    # Inherit maturity from the rule already exists
-    maturity = "development"
-    if path.exists():
-        rules = RuleCollection()
-        rules.load_file(path)
-        if rules:
-            maturity = rules.rules[0].contents.metadata.maturity
-
+    # Inherit maturity and optionally local dates from the rule if it already exists
     meta = {
-        "creation_date": creation_date,
-        "updated_date": creation_date,
-        "maturity": maturity,
+        "creation_date": kwargs.get("creation_date") or creation_date,
+        "updated_date": kwargs.get("updated_date") or creation_date,
+        "maturity": "development" or kwargs.get("maturity"),
     }
 
     try:

detection_rules/config.py

@@ -193,6 +193,7 @@ class RulesConfig:
     exception_dir: Optional[Path] = None
     normalize_kql_keywords: bool = True
     bypass_optional_elastic_validation: bool = False
+    no_tactic_filename: bool = False
 
     def __post_init__(self):
         """Perform post validation on packages.yaml file."""
@@ -311,6 +312,10 @@ def parse_rules_config(path: Optional[Path] = None) -> RulesConfig:
     if contents['bypass_optional_elastic_validation']:
         set_all_validation_bypass(contents['bypass_optional_elastic_validation'])
 
+    # no_tactic_filename
+    contents['no_tactic_filename'] = loaded.get('no_tactic_filename', False)
+
+    # return the config
     try:
         rules_config = RulesConfig(test_config=test_config, **contents)
     except (ValueError, TypeError) as e:

detection_rules/etc/_config.yaml

@@ -72,3 +72,8 @@ normalize_kql_keywords: False
 # If set in this file, the path should be relative to the location of this config. If passed as an environment variable,
 #    it should be the full path
 # Note: Using the `custom-rules setup-config <name>` command will generate a config called `test_config.yaml`
+
+# To prevent the tactic prefix from being added to the rule filename, set the line below to True
+# This config line can be used instead of specifying the `--no-tactic-filename` flag in the CLI
+# Mind that for unit tests, you also want to disable the filename test in the test_config.yaml
+# no_tactic_filename: True

detection_rules/etc/non-ecs-schema.json

@@ -172,7 +172,9 @@
   },
   "logs-azure.signinlogs-*": {
     "azure.signinlogs.properties.conditional_access_audiences.application_id": "keyword",
-    "azure.signinlogs.properties.original_transfer_method": "keyword"
+    "azure.signinlogs.properties.original_transfer_method": "keyword",
+    "azure.auditlogs.properties.target_resources.0.display_name": "keyword",
+    "azure.signinlogs.properties.authentication_details.authentication_method": "keyword"
   },
   "logs-azure.activitylogs-*": {
     "azure.activitylogs.properties.authentication_protocol": "keyword",

detection_rules/kbwrap.py

@@ -24,7 +24,7 @@
 from .main import root
 from .misc import add_params, client_error, kibana_options, get_kibana_client, nested_set
 from .rule import downgrade_contents_from_rule, TOMLRuleContents, TOMLRule
-from .rule_loader import RuleCollection
+from .rule_loader import RuleCollection, update_metadata_from_file
 from .utils import format_command_options, rulename_to_filename
 
 RULES_CONFIG = parse_rules_config()
@@ -108,27 +108,44 @@ def _parse_list_id(s: str):
 
         # Re-try to address known Kibana issue: https://github.com/elastic/kibana/issues/143864
         workaround_errors = []
+        workaround_error_types = set()
 
         flattened_exceptions = [e for sublist in exception_dicts for e in sublist]
         all_exception_list_ids = {exception["list_id"] for exception in flattened_exceptions}
 
         click.echo(f'{len(response["errors"])} rule(s) failed to import!')
 
+        action_connector_validation_error = "Error validating create data"
+        action_connector_type_error = "expected value of type [string] but got [undefined]"
         for error in response['errors']:
-            click.echo(f' - {error["rule_id"]}: ({error["error"]["status_code"]}) {error["error"]["message"]}')
+            error_message = error["error"]["message"]
+            click.echo(f' - {error["rule_id"]}: ({error["error"]["status_code"]}) {error_message}')
 
-            if "references a non existent exception list" in error["error"]["message"]:
-                list_id = _parse_list_id(error["error"]["message"])
+            if "references a non existent exception list" in error_message:
+                list_id = _parse_list_id(error_message)
                 if list_id in all_exception_list_ids:
                     workaround_errors.append(error["rule_id"])
+                    workaround_error_types.add("non existent exception list")
+
+            if action_connector_validation_error in error_message and action_connector_type_error in error_message:
+                workaround_error_types.add("connector still being built")
 
         if workaround_errors:
             workaround_errors = list(set(workaround_errors))
-            click.echo(f'Missing exception list errors detected for {len(workaround_errors)} rules. '
-                       'Try re-importing using the following command and rule IDs:\n')
-            click.echo('python -m detection_rules kibana import-rules -o ', nl=False)
-            click.echo(' '.join(f'-id {rule_id}' for rule_id in workaround_errors))
-            click.echo()
+            if "non existent exception list" in workaround_error_types:
+                click.echo(
+                    f"Missing exception list errors detected for {len(workaround_errors)} rules. "
+                    "Try re-importing using the following command and rule IDs:\n"
+                )
+                click.echo("python -m detection_rules kibana import-rules -o ", nl=False)
+                click.echo(" ".join(f"-id {rule_id}" for rule_id in workaround_errors))
+                click.echo()
+            if "connector still being built" in workaround_error_types:
+                click.echo(
+                    f"Connector still being built errors detected for {len(workaround_errors)} rules. "
+                    "Please try re-importing the rules again."
+                )
+                click.echo()
 
     def _process_imported_items(imported_items_list, item_type_description, item_key):
         """Displays appropriately formatted success message that all items imported successfully."""
@@ -178,20 +195,38 @@ def _process_imported_items(imported_items_list, item_type_description, item_key
 @click.option("--exceptions-directory", "-ed", required=False, type=Path, help="Directory to export exceptions to")
 @click.option("--default-author", "-da", type=str, required=False, help="Default author for rules missing one")
 @click.option("--rule-id", "-r", multiple=True, help="Optional Rule IDs to restrict export to")
+@click.option("--rule-name", "-rn", required=False, help="Optional Rule name to restrict export to "
+              "(KQL, case-insensitive, supports wildcards)")
 @click.option("--export-action-connectors", "-ac", is_flag=True, help="Include action connectors in export")
 @click.option("--export-exceptions", "-e", is_flag=True, help="Include exceptions in export")
 @click.option("--skip-errors", "-s", is_flag=True, help="Skip errors when exporting rules")
 @click.option("--strip-version", "-sv", is_flag=True, help="Strip the version fields from all rules")
+@click.option("--no-tactic-filename", "-nt", is_flag=True,
+              help="Exclude tactic prefix in exported filenames for rules. "
+              "Use same flag for import-rules to prevent warnings and disable its unit test.")
+@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.pass_context
 def kibana_export_rules(ctx: click.Context, directory: Path, action_connectors_directory: Optional[Path],
                         exceptions_directory: Optional[Path], default_author: str,
-                        rule_id: Optional[Iterable[str]] = None, export_action_connectors: bool = False,
-                        export_exceptions: bool = False, skip_errors: bool = False, strip_version: bool = False
-                        ) -> List[TOMLRule]:
+                        rule_id: Optional[Iterable[str]] = None, rule_name: Optional[str] = None,
+                        export_action_connectors: bool = False,
+                        export_exceptions: bool = False, skip_errors: bool = False, strip_version: bool = False,
+                        no_tactic_filename: bool = False, local_creation_date: bool = False,
+                        local_updated_date: bool = False) -> List[TOMLRule]:
     """Export custom rules from Kibana."""
     kibana = ctx.obj["kibana"]
     kibana_include_details = export_exceptions or export_action_connectors
+
+    # Only allow one of rule_id or rule_name
+    if rule_name and rule_id:
+        raise click.UsageError("Cannot use --rule-id and --rule-name together. Please choose one.")
+
     with kibana:
+        # Look up rule IDs by name if --rule-name was provided
+        if rule_name:
+            found = RuleResource.find(filter=f"alert.attributes.name:{rule_name}")
+            rule_id = [r["rule_id"] for r in found]
         results = RuleResource.export_rules(list(rule_id), exclude_export_details=not kibana_include_details)
 
     # Handle Exceptions Directory Location
@@ -215,6 +250,8 @@ def kibana_export_rules(ctx: click.Context, directory: Path, action_connectors_d
         return []
 
     rules_results = results
+    action_connector_results = []
+    exception_results = []
     if kibana_include_details:
         # Assign counts to variables
         rules_count = results[-1]["exported_rules_count"]
@@ -242,22 +279,27 @@ def kibana_export_rules(ctx: click.Context, directory: Path, action_connectors_d
             rule_resource["author"] = rule_resource.get("author") or default_author or [rule_resource.get("created_by")]
             if isinstance(rule_resource["author"], str):
                 rule_resource["author"] = [rule_resource["author"]]
-            # Inherit maturity from the rule already exists
-            maturity = "development"
+            # Inherit maturity and optionally local dates from the rule if it already exists
+            params = {
+                "rule": rule_resource,
+                "maturity": "development",
+            }
             threat = rule_resource.get("threat")
             first_tactic = threat[0].get("tactic").get("name") if threat else ""
-            rule_name = rulename_to_filename(rule_resource.get("name"), tactic_name=first_tactic)
-            # check if directory / f"{rule_name}" exists
-            if (directory / f"{rule_name}").exists():
-                rules = RuleCollection()
-                rules.load_file(directory / f"{rule_name}")
-                if rules:
-                    maturity = rules.rules[0].contents.metadata.maturity
-
-            contents = TOMLRuleContents.from_rule_resource(
-                rule_resource, maturity=maturity
+            # Check if flag or config is set to not include tactic in the filename
+            no_tactic_filename = no_tactic_filename or RULES_CONFIG.no_tactic_filename
+            # Check if the flag is set to not include tactic in the filename
+            tactic_name = first_tactic if not no_tactic_filename else None
+            rule_name = rulename_to_filename(rule_resource.get("name"), tactic_name=tactic_name)
+
+            save_path = directory / f"{rule_name}"
+            params.update(
+                update_metadata_from_file(
+                    save_path, {"creation_date": local_creation_date, "updated_date": local_updated_date}
+                )
             )
-            rule = TOMLRule(contents=contents, path=directory / f"{rule_name}")
+            contents = TOMLRuleContents.from_rule_resource(**params)
+            rule = TOMLRule(contents=contents, path=save_path)
         except Exception as e:
             if skip_errors:
                 print(f'- skipping {rule_resource.get("name")} - {type(e).__name__}')

detection_rules/main.py

@@ -32,7 +32,7 @@
 )
 from .rule import TOMLRule, TOMLRuleContents, QueryRuleData
 from .rule_formatter import toml_write
-from .rule_loader import RuleCollection
+from .rule_loader import RuleCollection, update_metadata_from_file
 from .schemas import all_versions, definitions, get_incompatible_fields, get_schema_file
 from .utils import Ndjson, get_path, get_etc_path, clear_caches, load_dump, load_rule_contents, rulename_to_filename
 
@@ -128,10 +128,13 @@ def generate_rules_index(ctx: click.Context, query, overwrite, save_files=True):
 @click.option("--skip-errors", "-ske", is_flag=True, help="Skip rule import errors")
 @click.option("--default-author", "-da", type=str, required=False, help="Default author for rules missing one")
 @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")
 def import_rules_into_repo(input_file: click.Path, required_only: bool, action_connector_import: bool,
                            exceptions_import: bool, directory: click.Path, save_directory: click.Path,
                            action_connectors_directory: click.Path, exceptions_directory: click.Path,
-                           skip_errors: bool, default_author: str, strip_none_values: bool):
+                           skip_errors: bool, default_author: str, strip_none_values: bool, local_creation_date: bool,
+                           local_updated_date: bool):
     """Import rules from json, toml, or yaml files containing Kibana exported rule(s)."""
     errors = []
     rule_files = glob.glob(os.path.join(directory, "**", "*.*"), recursive=True) if directory else []
@@ -179,6 +182,12 @@ def import_rules_into_repo(input_file: click.Path, required_only: bool, action_c
         if isinstance(contents["author"], str):
             contents["author"] = [contents["author"]]
 
+        contents.update(
+            update_metadata_from_file(
+                Path(rule_path), {"creation_date": local_creation_date, "updated_date": local_updated_date}
+            )
+        )
+
         output = rule_prompt(
             rule_path,
             required_only=required_only,

detection_rules/rule_loader.py

@@ -18,7 +18,8 @@
 from . import utils
 from .config import parse_rules_config
 from .rule import (
-    DeprecatedRule, DeprecatedRuleContents, DictRule, TOMLRule, TOMLRuleContents
+    DeprecatedRule, DeprecatedRuleContents, DictRule, TOMLRule,
+    TOMLRuleContents
 )
 from .schemas import definitions
 from .utils import cached, get_path
@@ -116,6 +117,20 @@ def load_locks_from_tag(remote: str, tag: str, version_lock: str = 'detection_ru
     return commit_hash, version, deprecated
 
 
+def update_metadata_from_file(rule_path: Path, fields_to_update: dict) -> dict:
+    """Update metadata fields for a rule with local contents."""
+    contents = {}
+    if not rule_path.exists():
+        return contents
+    local_metadata = RuleCollection().load_file(rule_path).contents.metadata.to_dict()
+    if local_metadata:
+        contents["maturity"] = local_metadata.get("maturity", "development")
+        for field_name, should_update in fields_to_update.items():
+            if should_update and field_name in local_metadata:
+                contents[field_name] = local_metadata[field_name]
+    return contents
+
+
 @dataclass
 class BaseCollection:
     """Base class for collections."""