Downgrade rule version before uploading to Kibana

detection_rules/eswrap.py

@@ -233,6 +233,7 @@ def kibana_upload(toml_files, url, cloud_id, user, password):
     """Upload a list of rule .toml files to Kibana."""
     from uuid import uuid4
     from .packaging import manage_versions
+    from .schemas import downgrade
 
     with Kibana(cloud_id=cloud_id, url=url) as kibana:
         kibana.login(user, password)
@@ -252,7 +253,9 @@ def kibana_upload(toml_files, url, cloud_id, user, password):
             meta = payload.setdefault("meta", {})
             meta["original"] = dict(id=rule.id, **rule.metadata)
             payload["rule_id"] = str(uuid4())
-            api_payloads.append(RuleResource.from_dict(payload))
+            payload = downgrade(payload, kibana.version)
+            rule = RuleResource.from_dict(payload)
+            api_payloads.append(rule)
 
         rules = RuleResource.bulk_create(api_payloads)
         click.echo(f"Successfully uploaded {len(rules)} rules")

detection_rules/misc.py

@@ -197,7 +197,7 @@ def set_param_values(ctx, param, value):
     """Get value for defined key."""
     key = param.name
     config = parse_config()
-    env_key = 'DR_' + key
+    env_key = 'DR_' + key.upper()
     prompt = True if param.hide_input is not False else False
 
     if value:

detection_rules/schemas/__init__.py

@@ -26,17 +26,18 @@
 CurrentSchema = max(all_schemas, key=lambda cls: Version(cls.STACK_VERSION))
 
 
-def downgrade(api_contents, target_version):
+def downgrade(api_contents: dict, target_version: str):
     """Downgrade a rule to a target stack version."""
     # truncate to (major, minor)
+    target_version_str = target_version
     target_version = Version(target_version)[:2]
     versions = set(Version(schema_cls.STACK_VERSION) for schema_cls in all_schemas)
     role = api_contents.get("type")
 
     check_versioned = "version" in api_contents
 
     if target_version not in versions:
-        raise ValueError(f"Unable to downgrade from {CurrentSchema.STACK_VERSION} to {target_version}")
+        raise ValueError(f"Unable to downgrade from {CurrentSchema.STACK_VERSION} to {target_version_str}")
 
     current_schema = None
 

kibana/connector.py

@@ -29,6 +29,7 @@ def __init__(self, cloud_id=None, url=None, verify=True, elasticsearch=None):
         self.cloud_id = cloud_id
         self.kibana_url = url
         self.elastic_url = None
+        self.status = None
 
         if self.cloud_id:
             self.cluster_name, cloud_info = self.cloud_id.split(":")
@@ -45,6 +46,12 @@ def __init__(self, cloud_id=None, url=None, verify=True, elasticsearch=None):
             from requests.packages.urllib3.exceptions import InsecureRequestWarning
             requests.packages.urllib3.disable_warnings(InsecureRequestWarning)
 
+    @property
+    def version(self):
+        """Get the semantic version."""
+        if self.status:
+            return self.status.get("version", {}).get("number")
+
     def url(self, uri):
         """Get the full URL given a URI."""
         assert self.kibana_url is not None
@@ -99,6 +106,7 @@ def login(self, username, password):
 
         self.post(path, data=payload, error=True)
         self.authenticated = True
+        self.status = self.get("/api/status")
 
         # create ES and force authentication
         if self.elasticsearch is None and self.elastic_url is not None: