[FR] [DAC] further decouple reliance on default rule dir locations (#3654)

* Improve dac custom init

* Fix path naming

* patch for ci runs

* add doc strings and rename test config name

* expand how unit test are selected

* Updated to support list of dirs

* raise unlink to CLI

* Fix unit test config post assertion

* Add a custom method to generate the test config

* add explicit checks for package.yml fields

* newline

* raise SystemExit instead of sys.exit

* Collapsing missing config message and exit

* flake8

* update base config

* typo

* Updated config parsing

* Update detection_rules/config.py

Co-authored-by: Justin Ibarra <[email protected]>

* simplify package requirements

* remove import

* add dataclass to validate rules config file and create default setup-config cli

* add kibana_version cli param

* update doc string

* rename delete cli option to overwrite, and small edits to exceptions

* Typo in config

* Add resolve

* Added TODO

* Cleanup

* Update path to config path

* Added get_rules_dir_path function

* revert config change

* Updated config

* Minor Cleanup

* Cleanup get_base_rule_dir

* Updated to remove try except

* update test cli command

* Updated config generation

* Add default in config

* Update test CLI

* update readme

* readme updates

* Add support for multiple rules dirs

* Updated import-rules-to-repo readme

* Update unit test

* Remove redundant check

* Remove additional parse_rules_config from mappings

---------

Co-authored-by: Mika Ayenson <[email protected]>
Co-authored-by: Mika Ayenson <[email protected]>
Co-authored-by: Justin Ibarra <[email protected]>

Author: 3 people

CLI.md

@@ -77,9 +77,10 @@ Usage: detection_rules import-rules-to-repo [OPTIONS] [INPUT_FILE]...
   Import rules from json, toml, yaml, or Kibana exported rule file(s).
 
 Options:
-  --required-only            Only prompt for required fields
-  -d, --directory DIRECTORY  Load files from a directory
-  -h, --help                 Show this message and exit.
+  --required-only                 Only prompt for required fields
+  -d, --directory DIRECTORY       Load files from a directory
+  -s, --save-directory DIRECTORY  Save imported rules to a directory
+  -h, --help                      Show this message and exit.
 ```
 
 The primary advantage of using this command is the ability to import multiple rules at once. Multiple rule paths can be
@@ -89,6 +90,8 @@ a combination of both.
 In addition to the formats mentioned using `create-rule`, this will also accept an `.ndjson`/`jsonl` file
 containing multiple rules (as would be the case with a bulk export).
 
+The `-s/--save-directory` is an optional parameter to specify a non default directory to place imported rules. If it is not specified, the first directory specified in the rules config will be used.
+
 This will also strip additional fields and prompt for missing required fields.
 
 <a id="note-3">\* Note</a>: This will attempt to parse ALL files recursively within a specified directory.
@@ -286,6 +289,10 @@ _*To load a custom rule, the proper index must be setup first. The simplest way
 the `Load prebuilt detection rules and timeline templates` button on the `detections` page in the Kibana security app._
 
 
+_*To load a custom rule, the proper index must be setup first. The simplest way to do this is to click
+the `Load prebuilt detection rules and timeline templates` button on the `detections` page in the Kibana security app._
+
+
 ### Using `import-rules`
 
 This is a better option than `upload-rule` as it is built on refreshed APIs

detection_rules/cli_utils.py

@@ -18,13 +18,11 @@
 from .attack import matrix, tactics, build_threat_map_entry
 from .rule import TOMLRule, TOMLRuleContents
 from .rule_loader import (RuleCollection,
-                          DEFAULT_PREBUILT_RULES_DIR,
-                          DEFAULT_PREBUILT_BBR_DIR,
+                          DEFAULT_PREBUILT_RULES_DIRS,
+                          DEFAULT_PREBUILT_BBR_DIRS,
                           dict_filter)
 from .schemas import definitions
-from .utils import clear_caches, get_path
-
-RULES_DIR = get_path("rules")
+from .utils import clear_caches
 
 
 def single_collection(f):
@@ -49,7 +47,7 @@ def get_collection(*args, **kwargs):
         rules.load_directories(Path(d) for d in directories)
 
         if rule_id:
-            rules.load_directories((DEFAULT_PREBUILT_RULES_DIR, DEFAULT_PREBUILT_BBR_DIR),
+            rules.load_directories(DEFAULT_PREBUILT_RULES_DIRS + DEFAULT_PREBUILT_BBR_DIRS,
                                    obj_filter=dict_filter(rule__rule_id=rule_id))
             if len(rules) != 1:
                 client_error(f"Could not find rule with ID {rule_id}")
@@ -83,7 +81,7 @@ def get_collection(*args, **kwargs):
         rules.load_directories(Path(d) for d in directories)
 
         if rule_id:
-            rules.load_directories((DEFAULT_PREBUILT_RULES_DIR, DEFAULT_PREBUILT_BBR_DIR),
+            rules.load_directories(DEFAULT_PREBUILT_RULES_DIRS + DEFAULT_PREBUILT_BBR_DIRS,
                                    obj_filter=dict_filter(rule__rule_id=rule_id))
             found_ids = {rule.id for rule in rules}
             missing = set(rule_id).difference(found_ids)
@@ -187,7 +185,8 @@ def rule_prompt(path=None, rule_type=None, required_only=True, save=True, verbos
 
             contents[name] = result
 
-    suggested_path = os.path.join(RULES_DIR, contents['name'])  # TODO: UPDATE BASED ON RULE STRUCTURE
+    # DEFAULT_PREBUILT_RULES_DIRS[0] is a required directory just as a suggestion
+    suggested_path = os.path.join(DEFAULT_PREBUILT_RULES_DIRS[0], contents['name'])
     path = os.path.realpath(path or input('File path for rule [{}]: '.format(suggested_path)) or suggested_path)
     meta = {'creation_date': creation_date, 'updated_date': creation_date, 'maturity': 'development'}
 

detection_rules/config.py

@@ -6,7 +6,7 @@
 """Configuration support for custom components."""
 import fnmatch
 import os
-from dataclasses import dataclass
+from dataclasses import dataclass, field
 from pathlib import Path
 from functools import cached_property
 from typing import Dict, List, Optional
@@ -186,6 +186,7 @@ class RulesConfig:
     version_lock: Dict[str, dict]
 
     action_dir: Optional[Path] = None
+    bbr_rules_dirs: Optional[List[Path]] = field(default_factory=list)
     exception_dir: Optional[Path] = None
 
     def __post_init__(self):
@@ -248,18 +249,23 @@ def parse_rules_config(path: Optional[Path] = None) -> RulesConfig:
     # files
     # paths are relative
     files = {f'{k}_file': base_dir.joinpath(v) for k, v in loaded['files'].items()}
-    contents = {k: load_dump(str(base_dir.joinpath(v))) for k, v in loaded['files'].items()}
+    contents = {k: load_dump(str(base_dir.joinpath(v).resolve())) for k, v in loaded['files'].items()}
 
     contents.update(**files)
 
     # directories
     # paths are relative
     if loaded.get('directories'):
-        contents.update({k: base_dir.joinpath(v) for k, v in loaded['directories'].items()})
+        contents.update({k: base_dir.joinpath(v).resolve() for k, v in loaded['directories'].items()})
 
     # rule_dirs
     # paths are relative
-    contents['rule_dirs'] = [base_dir.joinpath(d) for d in loaded.get('rule_dirs')]
+    contents['rule_dirs'] = [base_dir.joinpath(d).resolve() for d in loaded.get('rule_dirs')]
+
+    # bbr_rules_dirs
+    # paths are relative
+    if loaded.get('bbr_rules_dirs'):
+        contents['bbr_rules_dirs'] = [base_dir.joinpath(d).resolve() for d in loaded.get('bbr_rules_dirs', [])]
 
     try:
         rules_config = RulesConfig(test_config=test_config, **contents)

detection_rules/custom_rules.py

@@ -27,7 +27,8 @@ def create_config_content() -> str:
     """Create the content for the _config.yaml file."""
     # Base structure of the configuration
     config_content = {
-        'rule_dirs': ['rules', 'rules_building_block'],
+        'rule_dirs': ['rules'],
+        'bbr_rules_dirs': [],
         'files': {
             'deprecated_rules': 'etc/deprecated_rules.json',
             'packages': 'etc/packages.yml',

detection_rules/devtools.py

@@ -33,7 +33,6 @@
 from .beats import (download_beats_schema, download_latest_beats_schema,
                     refresh_main_schema)
 from .cli_utils import single_collection
-from .config import parse_rules_config
 from .docs import IntegrationSecurityDocs, IntegrationSecurityDocsMDX
 from .ecs import download_endpoint_schemas, download_schemas
 from .endgame import EndgameSchemaManager
@@ -51,13 +50,11 @@
                         Package)
 from .rule import (AnyRuleData, BaseRuleData, DeprecatedRule, QueryRuleData,
                    RuleTransform, ThreatMapping, TOMLRule, TOMLRuleContents)
-from .rule_loader import RuleCollection, production_filter
+from .rule_loader import RULES_CONFIG, RuleCollection, production_filter
 from .schemas import definitions, get_stack_versions
 from .utils import dict_hash, get_etc_path, get_path, load_dump
 from .version_lock import VersionLockFile, loaded_version_lock
 
-RULES_CONFIG = parse_rules_config()
-RULES_DIR = get_path('rules')
 GH_CONFIG = Path.home() / ".config" / "gh" / "hosts.yml"
 NAVIGATOR_GIST_ID = '1a3f65224822a30a8228a8ed20289a89'
 NAVIGATOR_URL = 'https://ela.st/detection-rules-navigator'
@@ -315,15 +312,17 @@ def prune_staging_area(target_stack_version: str, dry_run: bool, exception_list:
             continue
 
         # it's a change to a rule file, load it and check the version
-        if str(change.path.absolute()).startswith(RULES_DIR) and change.path.suffix == ".toml":
-            # bypass TOML validation in case there were schema changes
-            dict_contents = RuleCollection.deserialize_toml_string(change.read())
-            min_stack_version: Optional[str] = dict_contents.get("metadata", {}).get("min_stack_version")
-
-            if min_stack_version is not None and \
-                    (target_stack_version < Version.parse(min_stack_version, optional_minor_and_patch=True)):
-                # rule is incompatible, add to the list of reversions to make later
-                reversions.append(change)
+        for rules_dir in RULES_CONFIG.rule_dirs:
+            if str(change.path.absolute()).startswith(str(rules_dir)) and change.path.suffix == ".toml":
+                # bypass TOML validation in case there were schema changes
+                dict_contents = RuleCollection.deserialize_toml_string(change.read())
+                min_stack_version: Optional[str] = dict_contents.get("metadata", {}).get("min_stack_version")
+
+                if min_stack_version is not None and \
+                        (target_stack_version < Version.parse(min_stack_version, optional_minor_and_patch=True)):
+                    # rule is incompatible, add to the list of reversions to make later
+                    reversions.append(change)
+                break
 
     if len(reversions) == 0:
         click.echo("No files restored from staging area")
@@ -733,7 +732,7 @@ def deprecate_rule(ctx: click.Context, rule_file: Path):
         ctx.exit()
 
     today = time.strftime('%Y/%m/%d')
-    deprecated_path = get_path('rules', '_deprecated', rule_file.name)
+    deprecated_path = rule.get_base_rule_dir() / '_deprecated' / rule_file.name
 
     # create the new rule and save it
     new_meta = dataclasses.replace(rule.contents.metadata,

detection_rules/etc/_config.yaml

@@ -1,8 +1,8 @@
 # detection-rules config file
-
+bbr_rules_dirs:
+  - ../../rules_building_block
 rule_dirs:
-  - rules
-  - rules_building_block
+  - ../../rules
 files:
   deprecated_rules: deprecated_rules.json
   packages: packages.yml

detection_rules/main.py

@@ -33,10 +33,8 @@
 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
 
-
-RULES_DIR = get_path('rules')
-ROOT_DIR = Path(RULES_DIR).parent
 RULES_CONFIG = parse_rules_config()
+RULES_DIRS = RULES_CONFIG.rule_dirs
 
 
 @click.group('detection-rules', context_settings={'help_option_names': ['-h', '--help']})
@@ -101,7 +99,10 @@ def generate_rules_index(ctx: click.Context, query, overwrite, save_files=True):
 @click.argument('input-file', type=click.Path(dir_okay=False, exists=True), nargs=-1, required=False)
 @click.option('--required-only', is_flag=True, help='Only prompt for required fields')
 @click.option('--directory', '-d', type=click.Path(file_okay=False, exists=True), help='Load files from a directory')
-def import_rules_into_repo(input_file, required_only, directory):
+@click.option('--save-directory', '-s', type=click.Path(file_okay=False, exists=True),
+              help='Save imported rules to a directory')
+def import_rules_into_repo(input_file: click.Path, required_only: bool, directory: click.Path,
+                           save_directory: click.Path):
     """Import rules from json, toml, yaml, or Kibana exported rule file(s)."""
     rule_files = glob.glob(os.path.join(directory, '**', '*.*'), recursive=True) if directory else []
     rule_files = sorted(set(rule_files + list(input_file)))
@@ -116,7 +117,7 @@ def import_rules_into_repo(input_file, required_only, directory):
     for contents in rule_contents:
         base_path = contents.get('name') or contents.get('rule', {}).get('name')
         base_path = rulename_to_filename(base_path) if base_path else base_path
-        rule_path = os.path.join(RULES_DIR, base_path) if base_path else None
+        rule_path = os.path.join(save_directory if save_directory is not None else RULES_DIRS[0], base_path)
         additional = ['index'] if not contents.get('data_view_id') else ['data_view_id']
         rule_prompt(rule_path, required_only=required_only, save=True, verbose=True,
                     additional_required=additional, **contents)

detection_rules/navigator.py

@@ -17,7 +17,6 @@
 from .attack import CURRENT_ATTACK_VERSION
 from .mixins import MarshmallowDataclassMixin
 from .rule import TOMLRule
-from .rule_loader import DEFAULT_PREBUILT_RULES_DIR, DEFAULT_PREBUILT_BBR_DIR
 from .schemas import definitions
 
 
@@ -162,11 +161,13 @@ def links_dict(label: str, url: any) -> dict:
         return links
 
     def rule_links_dict(self, rule: TOMLRule) -> dict:
+        """Create a links dictionary for a rule."""
         base_url = 'https://github.com/elastic/detection-rules/blob/main/rules/'
-        try:
-            base_path = str(rule.path.resolve().relative_to(DEFAULT_PREBUILT_RULES_DIR))
-        except ValueError:
-            base_path = str(rule.path.resolve().relative_to(DEFAULT_PREBUILT_BBR_DIR))
+        base_path = str(rule.get_base_rule_dir())
+
+        if base_path is None:
+            raise ValueError("Could not find a valid base path for the rule")
+
         url = f'{base_url}{base_path}'
         return self.links_dict(rule.name, url)
 

detection_rules/packaging.py

@@ -23,7 +23,7 @@
 from .misc import JS_LICENSE, cached
 from .navigator import NavigatorBuilder, Navigator
 from .rule import TOMLRule, QueryRuleData, ThreatMapping
-from .rule_loader import DeprecatedCollection, RuleCollection, DEFAULT_PREBUILT_RULES_DIR, DEFAULT_PREBUILT_BBR_DIR
+from .rule_loader import DeprecatedCollection, RuleCollection
 from .schemas import definitions
 from .utils import Ndjson, get_path, get_etc_path
 from .version_lock import loaded_version_lock
@@ -479,10 +479,10 @@ def create_bulk_index_body(self) -> Tuple[Ndjson, Ndjson]:
 
             bulk_upload_docs.append(create)
 
-            try:
-                relative_path = str(rule.path.resolve().relative_to(DEFAULT_PREBUILT_RULES_DIR))
-            except ValueError:
-                relative_path = str(rule.path.resolve().relative_to(DEFAULT_PREBUILT_BBR_DIR))
+            relative_path = str(rule.get_base_rule_dir())
+
+            if relative_path is None:
+                raise ValueError(f"Could not find a valid relative path for the rule: {rule.id}")
 
             rule_doc = dict(hash=rule.contents.sha256(),
                             source='repo',

detection_rules/rule.py

@@ -43,6 +43,8 @@
 MIN_FLEET_PACKAGE_VERSION = '7.13.0'
 TIME_NOW = time.strftime('%Y/%m/%d')
 RULES_CONFIG = parse_rules_config()
+DEFAULT_PREBUILT_RULES_DIRS = RULES_CONFIG.rule_dirs
+DEFAULT_PREBUILT_BBR_DIRS = RULES_CONFIG.bbr_rules_dirs
 
 
 BUILD_FIELD_VERSIONS = {
@@ -1321,6 +1323,14 @@ def get_asset(self) -> dict:
         """Generate the relevant fleet compatible asset."""
         return {"id": self.id, "attributes": self.contents.to_api_format(), "type": definitions.SAVED_OBJECT_TYPE}
 
+    def get_base_rule_dir(self) -> Path | None:
+        """Get the base rule directory for the rule."""
+        rule_path = self.path.resolve()
+        for rules_dir in DEFAULT_PREBUILT_RULES_DIRS + DEFAULT_PREBUILT_BBR_DIRS:
+            if rule_path.is_relative_to(rules_dir):
+                return rule_path.relative_to(rules_dir)
+        return None
+
     def save_toml(self):
         assert self.path is not None, f"Can't save rule {self.name} (self.id) without a path"
         converted = dict(metadata=self.contents.metadata.to_dict(), rule=self.contents.data.to_dict())

detection_rules/rule_loader.py

@@ -17,17 +17,17 @@
 from marshmallow.exceptions import ValidationError
 
 from . import utils
+from .config import parse_rules_config
 from .mappings import RtaMappings
 from .rule import (
     DeprecatedRule, DeprecatedRuleContents, DictRule, TOMLRule, TOMLRuleContents
 )
 from .schemas import definitions
 from .utils import cached, get_path
 
-DEFAULT_PREBUILT_RULES_DIR = Path(get_path("rules"))
-DEFAULT_PREBUILT_BBR_DIR = Path(get_path("rules_building_block"))
-DEFAULT_PREBUILT_DEPRECATED_DIR = DEFAULT_PREBUILT_RULES_DIR / '_deprecated'
-DEFAULT_PREBUILT_RTA_DIR = get_path("rta")
+RULES_CONFIG = parse_rules_config()
+DEFAULT_PREBUILT_RULES_DIRS = RULES_CONFIG.rule_dirs
+DEFAULT_PREBUILT_BBR_DIRS = RULES_CONFIG.bbr_rules_dirs
 FILE_PATTERN = r'^([a-z0-9_])+\.(json|toml)$'
 
 
@@ -294,8 +294,8 @@ def default(cls) -> 'RawRuleCollection':
         """Return the default rule collection, which retrieves from rules/."""
         if cls.__default is None:
             collection = RawRuleCollection()
-            collection.load_directory(DEFAULT_PREBUILT_RULES_DIR)
-            collection.load_directory(DEFAULT_PREBUILT_BBR_DIR)
+            collection.load_directories(DEFAULT_PREBUILT_RULES_DIRS)
+            collection.load_directories(DEFAULT_PREBUILT_BBR_DIRS)
             collection.freeze()
             cls.__default = collection
 
@@ -306,7 +306,7 @@ def default_bbr(cls) -> 'RawRuleCollection':
         """Return the default BBR collection, which retrieves from building_block_rules/."""
         if cls.__default_bbr is None:
             collection = RawRuleCollection()
-            collection.load_directory(DEFAULT_PREBUILT_BBR_DIR)
+            collection.load_directories(DEFAULT_PREBUILT_BBR_DIRS)
             collection.freeze()
             cls.__default_bbr = collection
 
@@ -443,8 +443,10 @@ def load_git_tag(self, branch: str, remote: Optional[str] = None, skip_query_val
         from .version_lock import VersionLock, add_rule_types_to_lock
 
         git = utils.make_git()
-        rules_dir = DEFAULT_PREBUILT_RULES_DIR.relative_to(get_path("."))
-        paths = git("ls-tree", "-r", "--name-only", branch, rules_dir).splitlines()
+        paths = []
+        for rules_dir in DEFAULT_PREBUILT_RULES_DIRS:
+            rules_dir = rules_dir.relative_to(get_path("."))
+            paths.extend(git("ls-tree", "-r", "--name-only", branch, rules_dir).splitlines())
 
         rule_contents = []
         rule_map = {}
@@ -508,8 +510,8 @@ def default(cls) -> 'RuleCollection':
         """Return the default rule collection, which retrieves from rules/."""
         if cls.__default is None:
             collection = RuleCollection()
-            collection.load_directory(DEFAULT_PREBUILT_RULES_DIR)
-            collection.load_directory(DEFAULT_PREBUILT_BBR_DIR)
+            collection.load_directories(DEFAULT_PREBUILT_RULES_DIRS)
+            collection.load_directories(DEFAULT_PREBUILT_BBR_DIRS)
             collection.freeze()
             cls.__default = collection
 
@@ -520,7 +522,7 @@ def default_bbr(cls) -> 'RuleCollection':
         """Return the default BBR collection, which retrieves from building_block_rules/."""
         if cls.__default_bbr is None:
             collection = RuleCollection()
-            collection.load_directory(DEFAULT_PREBUILT_BBR_DIR)
+            collection.load_directories(DEFAULT_PREBUILT_BBR_DIRS)
             collection.freeze()
             cls.__default_bbr = collection
 
@@ -630,8 +632,8 @@ def download_worker(pr_info):
 
 __all__ = (
     "FILE_PATTERN",
-    "DEFAULT_PREBUILT_RULES_DIR",
-    "DEFAULT_PREBUILT_BBR_DIR",
+    "DEFAULT_PREBUILT_RULES_DIRS",
+    "DEFAULT_PREBUILT_BBR_DIRS",
     "load_github_pr_rules",
     "DeprecatedCollection",
     "DeprecatedRule",