[Bug] Strip Non-Public Fields Prior to Uploading Rules (#2986)

detection_rules/kbwrap.py

@@ -5,16 +5,16 @@
 
 """Kibana cli commands."""
 import sys
-import uuid
 
 import click
 
+
 import kql
 from kibana import Signal, RuleResource
 from .cli_utils import multi_collection
 from .main import root
 from .misc import add_params, client_error, kibana_options, get_kibana_client, nested_set
-from .schemas import downgrade
+from .rule import downgrade_contents_from_rule
 from .utils import format_command_options
 
 
@@ -45,14 +45,7 @@ def upload_rule(ctx, rules, replace_id):
 
     for rule in rules:
         try:
-            payload = rule.contents.to_api_format()
-            payload.setdefault("meta", {}).update(rule.contents.metadata.to_dict())
-
-            if replace_id:
-                payload["rule_id"] = str(uuid.uuid4())
-
-            payload = downgrade(payload, target_version=kibana.version)
-
+            payload = downgrade_contents_from_rule(rule, kibana.version, replace_id=replace_id)
         except ValueError as e:
             client_error(f'{e} in version:{kibana.version}, for rule: {rule.name}', e, ctx=ctx)
 

detection_rules/rule.py

@@ -31,7 +31,8 @@ from .misc import load_current_package_version
 from .mixins import MarshmallowDataclassMixin, StackCompatMixin
 from .rule_formatter import nested_normalize, toml_write
 from .schemas import (SCHEMA_DIR, definitions, downgrade,
-                      get_min_supported_stack_version, get_stack_schemas)
+                      get_min_supported_stack_version, get_stack_schemas,
+                      strip_non_public_fields)
 from .schemas.stack_compat import get_restricted_fields
 from .utils import cached, convert_time_span, PatchedTemplate
 
@@ -1300,13 +1301,23 @@ class DeprecatedRule(dict):
         return self.contents.name
 
 
-def downgrade_contents_from_rule(rule: TOMLRule, target_version: str) -> dict:
+def downgrade_contents_from_rule(rule: TOMLRule, target_version: str, replace_id: bool = True) -> dict:
     """Generate the downgraded contents from a rule."""
-    payload = rule.contents.to_api_format()
-    meta = payload.setdefault("meta", {})
-    meta["original"] = dict(id=rule.id, **rule.contents.metadata.to_dict())
-    payload["rule_id"] = str(uuid4())
-    payload = downgrade(payload, target_version)
+    rule_dict = rule.contents.to_dict()["rule"]
+    min_stack_version = target_version or rule.contents.metadata.min_stack_version or "8.3.0"
+    min_stack_version = Version.parse(min_stack_version,
+                                      optional_minor_and_patch=True)
+    rule_dict.setdefault("meta", {}).update(rule.contents.metadata.to_dict())
+
+    if replace_id:
+        rule_dict["rule_id"] = str(uuid4())
+
+    rule_dict = downgrade(rule_dict, target_version=str(min_stack_version))
+    meta = rule_dict.pop("meta")
+    rule_contents = TOMLRuleContents.from_dict({"rule": rule_dict, "metadata": meta,
+                                                "transform": rule.contents.transform})
+    payload = rule_contents.to_api_format()
+    payload = strip_non_public_fields(min_stack_version, payload)
     return payload
 
 

detection_rules/schemas/__init__.py

@@ -65,9 +65,6 @@ def get_schema_file(version: Version, rule_type: str) -> dict:
 def strip_additional_properties(version: Version, api_contents: dict) -> dict:
     """Remove all fields that the target schema doesn't recognize."""
 
-    if Version.parse(version, optional_minor_and_patch=True) >= Version.parse("8.3.0"):
-        api_contents = strip_build_time_fields(api_contents)
-
     stripped = {}
     target_schema = get_schema_file(version, api_contents["type"])
 
@@ -80,14 +77,13 @@ def strip_additional_properties(version: Version, api_contents: dict) -> dict:
     return stripped
 
 
-def strip_build_time_fields(api_contents: dict) -> dict:
-    """Remove all fields that are only used at build time."""
-    contents = api_contents.copy()
-    if "related_integrations" in contents:
-        del contents["related_integrations"]
-    if "required_fields" in contents:
-        del contents["required_fields"]
-    return contents
+def strip_non_public_fields(min_stack_version: Version, data_dict: dict) -> dict:
+    """Remove all non public fields."""
+    for field, version_range in definitions.NON_PUBLIC_FIELDS.items():
+        if version_range[0] <= min_stack_version <= (version_range[1] or min_stack_version):
+            if field in data_dict:
+                del data_dict[field]
+    return data_dict
 
 
 @migrate("7.8")

detection_rules/schemas/definitions.py

@@ -9,6 +9,7 @@ from typing import List, Literal, Final
 
 from marshmallow import validate
 from marshmallow_dataclass import NewType
+from semver import Version
 
 ASSET_TYPE = "security_rule"
 SAVED_OBJECT_TYPE = "security-rule"
@@ -28,6 +29,11 @@ MINOR_SEMVER = r'^\d+\.\d+$'
 BRANCH_PATTERN = f'{VERSION_PATTERN}|^master$'
 
 NON_DATASET_PACKAGES = ['apm', 'endpoint', 'system', 'windows', 'cloud_defend', 'network_traffic']
+NON_PUBLIC_FIELDS = {
+    "related_integrations": (Version.parse('8.3.0'), None),
+    "required_fields": (Version.parse('8.3.0'), None),
+    "setup": (Version.parse('8.3.0'), None)
+}
 INTERVAL_PATTERN = r'^\d+[mshd]$'
 TACTIC_URL = r'^https://attack.mitre.org/tactics/TA[0-9]+/$'
 TECHNIQUE_URL = r'^https://attack.mitre.org/techniques/T[0-9]+/$'