[FR] Add limited support and validation for version and revision

CLI.md

@@ -475,6 +475,8 @@ python -m detection_rules kibana import-rules -d test-export-rules -o
 
 ### Exporting rules
 
+This command should be run with the `CUSTOM_RULES_DIR` envvar set, that way proper validation is applied to versioning when the rules are downloaded. See the [custom rules docs](docs/custom-rules.md) for more information.
+
 Example of a rule exporting, with errors skipped
 
 ```

detection_rules/cli_utils.py

@@ -5,6 +5,7 @@
 
 import copy
 import datetime
+import functools
 import os
 import typing
 from pathlib import Path
@@ -13,13 +14,12 @@
 import click
 
 import kql
-import functools
+
 from . import ecs
-from .attack import matrix, tactics, build_threat_map_entry
-from .rule import TOMLRule, TOMLRuleContents
-from .rule_loader import (RuleCollection,
-                          DEFAULT_PREBUILT_RULES_DIRS,
-                          DEFAULT_PREBUILT_BBR_DIRS,
+from .attack import build_threat_map_entry, matrix, tactics
+from .rule import BYPASS_VERSION_LOCK, TOMLRule, TOMLRuleContents
+from .rule_loader import (DEFAULT_PREBUILT_BBR_DIRS,
+                          DEFAULT_PREBUILT_RULES_DIRS, RuleCollection,
                           dict_filter)
 from .schemas import definitions
 from .utils import clear_caches
@@ -132,8 +132,8 @@ def rule_prompt(path=None, rule_type=None, required_only=True, save=True, verbos
             contents[name] = rule_type
             continue
 
-        # these are set at package release time
-        if name == 'version':
+        # these are set at package release time depending on the version strategy
+        if (name == 'version' or name == 'revision') and not BYPASS_VERSION_LOCK:
             continue
 
         if required_only and name not in required_fields:

detection_rules/config.py

@@ -187,6 +187,7 @@ class RulesConfig:
 
     action_dir: Optional[Path] = None
     bbr_rules_dirs: Optional[List[Path]] = field(default_factory=list)
+    bypass_version_lock: bool = False
     exception_dir: Optional[Path] = None
 
     def __post_init__(self):
@@ -262,6 +263,9 @@ def parse_rules_config(path: Optional[Path] = None) -> RulesConfig:
     # paths are relative
     contents['rule_dirs'] = [base_dir.joinpath(d).resolve() for d in loaded.get('rule_dirs')]
 
+    # version strategy
+    contents['bypass_version_lock'] = loaded.get('bypass_version_lock', False)
+
     # bbr_rules_dirs
     # paths are relative
     if loaded.get('bbr_rules_dirs'):

detection_rules/devtools.py

@@ -85,10 +85,21 @@ def dev_group():
 @click.option('--generate-navigator', is_flag=True, help='Generate ATT&CK navigator files')
 @click.option('--generate-docs', is_flag=True, default=False, help='Generate markdown documentation')
 @click.option('--update-message', type=str, help='Update message for new package')
-def build_release(config_file, update_version_lock: bool, generate_navigator: bool, generate_docs: str,
-                  update_message: str, release=None, verbose=True):
+@click.pass_context
+def build_release(ctx: click.Context, config_file, update_version_lock: bool, generate_navigator: bool,
+                  generate_docs: str, update_message: str, release=None, verbose=True):
     """Assemble all the rules into Kibana-ready release files."""
-    config = load_dump(str(config_file))['package']
+    if RULES_CONFIG.bypass_version_lock:
+        click.echo('WARNING: You cannot run this command when the versioning strategy is configured to bypass the '
+                   'version lock. Set `bypass_version_lock` to `False` in the rules config to use the version lock.')
+        ctx.exit()
+
+    config = load_dump(config_file)['package']
+
+    err_msg = f'No `registry_data` in package config. Please see the {get_etc_path("package.yaml")} file for an' \
+              f' example on how to supply this field in {PACKAGE_FILE}.'
+    assert 'registry_data' in config, err_msg
+
     registry_data = config['registry_data']
 
     if generate_navigator:
@@ -104,6 +115,7 @@ def build_release(config_file, update_version_lock: bool, generate_navigator: bo
 
     if update_version_lock:
         loaded_version_lock.manage_versions(package.rules, save_changes=True, verbose=verbose)
+
     package.save(verbose=verbose)
 
     previous_pkg_version = find_latest_integration_version("security_detection_engine", "ga",
@@ -335,8 +347,9 @@ def prune_staging_area(target_stack_version: str, dry_run: bool, exception_list:
 
 @dev_group.command('update-lock-versions')
 @click.argument('rule-ids', nargs=-1, required=False)
+@click.pass_context
 @click.option('--force', is_flag=True, help='Force update without confirmation')
-def update_lock_versions(rule_ids: Tuple[str, ...], force: bool):
+def update_lock_versions(ctx: click.Context, rule_ids: Tuple[str, ...], force: bool):
     """Update rule hashes in version.lock.json file without bumping version."""
     rules = RuleCollection.default()
 
@@ -350,6 +363,11 @@ def update_lock_versions(rule_ids: Tuple[str, ...], force: bool):
     ):
         return
 
+    if RULES_CONFIG.bypass_version_lock:
+        click.echo('WARNING: You cannot run this command when the versioning strategy is configured to bypass the '
+                   'version lock. Set `bypass_version_lock` to `False` in the rules config to use the version lock.')
+        ctx.exit()
+
     # this command may not function as expected anymore due to previous changes eliminating the use of add_new=False
     changed, new, _ = loaded_version_lock.manage_versions(rules, exclude_version_update=True, save_changes=True)
 
@@ -721,21 +739,23 @@ def add_github_meta(this_rule: TOMLRule, status: str, original_rule_id: Optional
 
 @dev_group.command('deprecate-rule')
 @click.argument('rule-file', type=Path)
+@click.option('--deprecation-folder', '-d', type=Path, required=True,
+              help='Location to move the deprecated rule file to')
 @click.pass_context
-def deprecate_rule(ctx: click.Context, rule_file: Path):
+def deprecate_rule(ctx: click.Context, rule_file: Path, deprecation_folder: Path):
     """Deprecate a rule."""
     version_info = loaded_version_lock.version_lock
     rule_collection = RuleCollection()
     contents = rule_collection.load_file(rule_file).contents
     rule = TOMLRule(path=rule_file, contents=contents)
 
-    if rule.contents.id not in version_info:
+    if rule.contents.id not in version_info and not RULES_CONFIG.bypass_version_lock:
         click.echo('Rule has not been version locked and so does not need to be deprecated. '
-                   'Delete the file or update the maturity to `development` instead')
+                   'Delete the file or update the maturity to `development` instead.')
         ctx.exit()
 
     today = time.strftime('%Y/%m/%d')
-    deprecated_path = rule.get_base_rule_dir() / '_deprecated' / rule_file.name
+    deprecated_path = deprecation_folder / rule_file.name
 
     # create the new rule and save it
     new_meta = dataclasses.replace(rule.contents.metadata,
@@ -744,6 +764,7 @@ def deprecate_rule(ctx: click.Context, rule_file: Path):
                                    maturity='deprecated')
     contents = dataclasses.replace(rule.contents, metadata=new_meta)
     new_rule = TOMLRule(contents=contents, path=deprecated_path)
+    deprecated_path.parent.mkdir(parents=True, exist_ok=True)
     new_rule.save_toml()
 
     # remove the old rule
@@ -817,13 +838,19 @@ def raw_permalink(raw_link):
 @click.argument('stack_version')
 @click.option('--skip-rule-updates', is_flag=True, help='Skip updating the rules')
 @click.option('--dry-run', is_flag=True, help='Print the changes rather than saving the file')
-def trim_version_lock(stack_version: str, skip_rule_updates: bool, dry_run: bool):
+@click.pass_context
+def trim_version_lock(ctx: click.Context, stack_version: str, skip_rule_updates: bool, dry_run: bool):
     """Trim all previous entries within the version lock file which are lower than the min_version."""
     stack_versions = get_stack_versions()
     assert stack_version in stack_versions, \
         f'Unknown min_version ({stack_version}), expected: {", ".join(stack_versions)}'
 
     min_version = Version.parse(stack_version)
+
+    if RULES_CONFIG.bypass_version_lock:
+        click.echo('WARNING: Cannot trim the version lock when the versioning strategy is configured to bypass the '
+                   'version lock. Set `bypass_version_lock` to `false` in the rules config to use the version lock.')
+        ctx.exit()
     version_lock_dict = loaded_version_lock.version_lock.to_dict()
     removed = defaultdict(list)
     rule_msv_drops = []

detection_rules/etc/_config.yaml

@@ -8,6 +8,14 @@ files:
   packages: packages.yaml
   stack_schema_map: stack-schema-map.yaml
   version_lock: version.lock.json
+
+# Set the versioning strategy.
+# 1. Set to False to use version.lock.json file
+# 2. Set to True to either:
+#   - Explicitly set within rule.version in the TOML file
+#   - Defer to kibana versions (never manually set)
+# bypass_version_lock: false
+
 # directories:
   # actions_dir: actions
   # exceptions_dir: exceptions

detection_rules/kbwrap.py

@@ -111,9 +111,10 @@ def kibana_import_rules(ctx: click.Context, rules: RuleCollection, overwrite: Op
 @click.option('--directory', '-d', required=True, type=Path, help='Directory to export rules to')
 @click.option('--rule-id', '-r', multiple=True, help='Optional Rule IDs to restrict export to')
 @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.pass_context
-def kibana_export_rules(ctx: click.Context, directory: Path,
-                        rule_id: Optional[Iterable[str]] = None, skip_errors: bool = False) -> List[TOMLRule]:
+def kibana_export_rules(ctx: click.Context, directory: Path, rule_id: Optional[Iterable[str]] = None,
+                        skip_errors: bool = False, strip_version: bool = False) -> List[TOMLRule]:
     """Export custom rules from Kibana."""
     kibana = ctx.obj['kibana']
     with kibana:
@@ -126,6 +127,10 @@ def kibana_export_rules(ctx: click.Context, directory: Path,
     exported = []
     for rule_resource in results:
         try:
+            if strip_version:
+                rule_resource.pop('revision', None)
+                rule_resource.pop('version', None)
+
             contents = TOMLRuleContents.from_rule_resource(rule_resource, maturity='production')
             threat = contents.data.get('threat')
             first_tactic = threat[0].tactic.name if threat else ''

detection_rules/packaging.py

@@ -92,11 +92,12 @@ def __init__(self, rules: RuleCollection, name: str, release: Optional[bool] = F
         self.historical = historical
 
         if min_version is not None:
-            self.rules = self.rules.filter(lambda r: min_version <= r.contents.latest_version)
+            self.rules = self.rules.filter(lambda r: min_version <= r.contents.saved_version)
 
         if max_version is not None:
-            self.rules = self.rules.filter(lambda r: max_version >= r.contents.latest_version)
+            self.rules = self.rules.filter(lambda r: max_version >= r.contents.saved_version)
 
+        assert not RULES_CONFIG.bypass_version_lock, "Packaging can not be used when version locking is bypassed."
         self.changed_ids, self.new_ids, self.removed_ids = \
             loaded_version_lock.manage_versions(self.rules, verbose=verbose, save_changes=False)
 

detection_rules/rule.py

@@ -45,6 +45,7 @@
 RULES_CONFIG = parse_rules_config()
 DEFAULT_PREBUILT_RULES_DIRS = RULES_CONFIG.rule_dirs
 DEFAULT_PREBUILT_BBR_DIRS = RULES_CONFIG.bbr_rules_dirs
+BYPASS_VERSION_LOCK = RULES_CONFIG.bypass_version_lock
 
 
 BUILD_FIELD_VERSIONS = {
@@ -363,6 +364,7 @@ class RelatedIntegrations:
     references: Optional[List[str]]
     related_integrations: Optional[List[RelatedIntegrations]] = field(metadata=dict(metadata=dict(min_compat="8.3")))
     required_fields: Optional[List[RequiredFields]] = field(metadata=dict(metadata=dict(min_compat="8.3")))
+    revision: Optional[int] = field(metadata=dict(metadata=dict(min_compat="8.8")))
     risk_score: definitions.RiskScore
     risk_score_mapping: Optional[List[RiskScoreMapping]]
     rule_id: definitions.UUIDString
@@ -378,6 +380,7 @@ class RelatedIntegrations:
     to: Optional[str]
     type: definitions.RuleType
     threat: Optional[List[ThreatMapping]]
+    version: Optional[definitions.PositiveInteger]
 
     @classmethod
     def save_schema(cls):
@@ -457,6 +460,22 @@ def process_note_plugins():
 
         return obj
 
+    @validates_schema
+    def validates_data(self, data, **kwargs):
+        """Validate fields and data for marshmallow schemas."""
+
+        # Validate version and revision fields not supplied for Elastic authored rules.
+        disallowed_fields = [field for field in ['version', 'revision'] if data.get(field) is not None]
+        if not disallowed_fields:
+            return
+
+        error_message = " and ".join(disallowed_fields)
+
+        if BYPASS_VERSION_LOCK is not True:
+            msg = (f"Configuration error: Rule {data['name']} - {data['rule_id']} "
+                   f"should not contain rules with `{error_message}` set.")
+            raise ValidationError(msg)
+
 
 class DataValidator:
     """Additional validation beyond base marshmallow schema validation."""
@@ -918,7 +937,7 @@ def type(self):
         pass
 
     def lock_info(self, bump=True) -> dict:
-        version = self.autobumped_version if bump else (self.latest_version or 1)
+        version = self.autobumped_version if bump else (self.saved_version or 1)
         contents = {"rule_name": self.name, "sha256": self.sha256(), "version": version, "type": self.type}
 
         return contents
@@ -965,20 +984,43 @@ def get_version_space(self) -> Optional[int]:
             return max_allowable_version - current_version - 1
 
     @property
-    def latest_version(self) -> Optional[int]:
-        """Retrieve the latest known version of the rule."""
-        min_stack = self.get_supported_version()
-        return self.version_lock.get_locked_version(self.id, min_stack)
+    def saved_version(self) -> Optional[int]:
+        """Retrieve the version from the version.lock or from the file if version locking is bypassed."""
+        toml_version = self.data.get("version")
+
+        if BYPASS_VERSION_LOCK:
+            return toml_version
+
+        if toml_version:
+            print(f"WARNING: Rule {self.name} - {self.id} has a version set in the rule TOML."
+                  " This `version` will be ignored and defaulted to the version.lock.json file."
+                  " Set `bypass_version_lock` to `True` in the rules config to use the TOML version.")
+
+        return self.version_lock.get_locked_version(self.id, self.get_supported_version())
 
     @property
     def autobumped_version(self) -> Optional[int]:
         """Retrieve the current version of the rule, accounting for automatic increments."""
-        version = self.latest_version
+        version = self.saved_version
+
+        if BYPASS_VERSION_LOCK:
+            raise NotImplementedError("This method is not implemented when version locking is not in use.")
+
+        # Default to version 1 if no version is set yet
         if version is None:
             return 1
 
+        # Auto-increment version if the rule is 'dirty' and not bypassing version lock
         return version + 1 if self.is_dirty else version
 
+    def get_synthetic_version(self, use_default: bool) -> Optional[int]:
+        """
+        Get the latest actual representation of a rule's version, where changes are accounted for automatically when
+        version locking is used, otherwise, return the version defined in the rule toml if present else optionally
+        default to 1.
+        """
+        return self.autobumped_version or self.saved_version or (1 if use_default else None)
+
     @classmethod
     def convert_supported_version(cls, stack_version: Optional[str]) -> Version:
         """Convert an optional stack version to the minimum for the lock in the form major.minor."""
@@ -1028,11 +1070,20 @@ def version_lock(self):
         # VersionLock
         from .version_lock import loaded_version_lock
 
+        if RULES_CONFIG.bypass_version_lock is True:
+            err_msg = "Cannot access the version lock when the versioning strategy is configured to bypass the" \
+                      " version lock. Set `bypass_version_lock` to `false` in the rules config to use the version lock."
+            raise ValueError(err_msg)
+
         return getattr(self, '_version_lock', None) or loaded_version_lock
 
     def set_version_lock(self, value):
         from .version_lock import VersionLock
 
+        err_msg = "Cannot set the version lock when the versioning strategy is configured to bypass the version lock." \
+                  " Set `bypass_version_lock` to `false` in the rules config to use the version lock."
+        assert not RULES_CONFIG.bypass_version_lock, err_msg
+
         if value and not isinstance(value, VersionLock):
             raise TypeError(f'version lock property must be set with VersionLock objects only. Got {type(value)}')
 
@@ -1272,7 +1323,7 @@ def flattened_dict(self) -> dict:
         flattened.update(self.metadata.to_dict())
         return flattened
 
-    def to_api_format(self, include_version: bool = True, include_metadata: bool = False) -> dict:
+    def to_api_format(self, include_version: bool = not BYPASS_VERSION_LOCK, include_metadata: bool = False) -> dict:
         """Convert the TOML rule to the API format."""
         rule_dict = self.to_dict()
         converted_data = rule_dict['rule']
@@ -1361,6 +1412,10 @@ def version_lock(self):
     def set_version_lock(self, value):
         from .version_lock import VersionLock
 
+        err_msg = "Cannot set the version lock when the versioning strategy is configured to bypass the version lock." \
+                  " Set `bypass_version_lock` to `false` in the rules config to use the version lock."
+        assert not RULES_CONFIG.bypass_version_lock, err_msg
+
         if value and not isinstance(value, VersionLock):
             raise TypeError(f'version lock property must be set with VersionLock objects only. Got {type(value)}')
 
@@ -1385,7 +1440,7 @@ def from_dict(cls, obj: dict):
         kwargs['transform'] = obj['transform'] if 'transform' in obj else None
         return cls(**kwargs)
 
-    def to_api_format(self, include_version=True) -> dict:
+    def to_api_format(self, include_version: bool = not BYPASS_VERSION_LOCK) -> dict:
         """Convert the TOML rule to the API format."""
         data = copy.deepcopy(self.data)
         if self.transform: