[FR] Limit historical rules to the latest 2

detection_rules/devtools.py

@@ -112,15 +112,15 @@ def build_release(config_file, update_version_lock: bool, generate_navigator: bo
                                                            registry_data['conditions']['kibana.version'].strip("^"))
     sde = SecurityDetectionEngine()
     historical_rules = sde.load_integration_assets(previous_pkg_version)
-    historical_rules = sde.transform_legacy_assets(historical_rules)
-    package.add_historical_rules(historical_rules, registry_data['version'])
+    limited_historical_rules = sde.keep_latest_versions(historical_rules)
+    package.add_historical_rules(limited_historical_rules, registry_data['version'])
     click.echo(f'[+] Adding historical rules from {previous_pkg_version} package')
 
     # NOTE: stopgap solution until security doc migration
     if generate_docs:
         click.echo(f'[+] Generating security docs for {registry_data["version"]} package')
         docs = IntegrationSecurityDocsMDX(registry_data['version'], Path(f'releases/{config["name"]}-docs'),
-                                          True, historical_rules, package, note=update_message)
+                                          True, limited_historical_rules, package, note=update_message)
         docs.generate()
 
     if verbose:

detection_rules/integrations.py

@@ -8,7 +8,7 @@
 import gzip
 import json
 import re
-from collections import OrderedDict
+from collections import defaultdict, OrderedDict
 from pathlib import Path
 from typing import Generator, List, Tuple, Union, Optional
 
@@ -26,6 +26,7 @@
 from .schemas import definitions
 
 MANIFEST_FILE_PATH = get_etc_path('integration-manifests.json.gz')
+NUM_LATEST_RULE_VERSIONS = 2
 SCHEMA_FILE_PATH = get_etc_path('integration-schemas.json.gz')
 _notified_integrations = set()
 
@@ -412,14 +413,27 @@ def load_integration_assets(self, package_version: Version) -> dict:
                       for x in asset_file_names}
         return assets
 
-    def transform_legacy_assets(self, assets: dict) -> dict:
-        """Transforms legacy rule assets to historical rules."""
-        # this code can be removed after the 8.8 minor release
-        # epr prebuilt rule packages should have appropriate file names
-
-        assets_transformed = {}
-        for asset_id, contents in assets.items():
-            new_asset_id = f"{contents['attributes']['rule_id']}_{contents['attributes']['version']}"
-            contents["id"] = new_asset_id
-            assets_transformed[new_asset_id] = contents
-        return assets_transformed
+    def keep_latest_versions(self, assets: dict, num_versions: int = NUM_LATEST_RULE_VERSIONS) -> dict:
+        """Keeps only the latest N versions of each rule to limit historical rule versions in our release package."""
+
+        # Dictionary to hold the sorted list of versions for each base rule ID
+        rule_versions = defaultdict(list)
+
+        # Separate rule ID and version, and group by base rule ID
+        for key in assets:
+            base_id, version = key.rsplit('_', 1)
+            version = int(version)  # Convert version to an integer for sorting
+            rule_versions[base_id].append((version, key))
+
+        # Dictionary to hold the final assets with only the specified number of latest versions
+        filtered_assets = {}
+
+        # Keep only the last/latest num_versions versions for each rule
+        # Sort versions and take the last num_versions
+        # Add the latest versions of the rule to the filtered assets
+        for base_id, versions in rule_versions.items():
+            latest_versions = sorted(versions, key=lambda x: x[0], reverse=True)[:num_versions]
+            for _, key in latest_versions:
+                filtered_assets[key] = assets[key]
+
+        return filtered_assets