add new field related_integrations to the post build (#2060)

* add new field `related_integrations` to the post build

* add exception for endpoint `integration`

* Skip rules without related integrations

* lint

* refactor related_integrations to TOMLRuleContents class

* update to reflect required_fields updates

* add todo

* add new line for linting

* related_integrations updates, get_packaged_integrations returns list of dictionaries, started work on integrations py

* build_integrations_manifest command completed

* initial test completed for post-building related_integrations

* removed get_integration_manifest method from rule, removed global integrations path

* moved integration related methods to integrations.py and fixed flake issues

* adjustments for PipedQuery from eql sequence rules and packages with no integration

* adjusted github client import for integrations.py

* Update detection_rules/devtools.py

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

* Update detection_rules/devtools.py

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

* added integration manifest schema, made adjustments

* Update detection_rules/integrations.py

* Update detection_rules/rule.py

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

* Update detection_rules/rule.py

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

* Update detection_rules/integrations.py

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

* Update detection_rules/integrations.py

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

* Update detection_rules/integrations.py

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

* Update detection_rules/rule.py

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

* removed get_integrations_package to consolidate code

* removed type list return

* adjusted import flake errors

* Update detection_rules/integrations.py

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

* Update detection_rules/integrations.py

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

* adjusted indentation error

* adjusted rule.get_packaged_integrations to account for kql.ast.OrExpr if event.dataset is not set

* Update detection_rules/devtools.py

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

* Update detection_rules/devtools.py

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

* Update detection_rules/integrations.py

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

* Update detection_rules/integrations.py

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

* adjusted find_least_compatible_version in integrations.py

* Update detection_rules/integrations.py

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

* fixed flake issues

* adjusted get_packaged_integrations

* iterate the ast for literal event.dataset values

* Update detection_rules/integrations.py

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

* Update detection_rules/integrations.py

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

* Update detection_rules/integrations.py

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

* Update detection_rules/integrations.py

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

* made small adjustments to address errors during build manifests command

* addressing integrations.find_least_compatible method to return None instead of raise error only

* Update detection_rules/integrations.py

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

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

Author: 3 people

detection_rules/devtools.py

@@ -40,6 +40,7 @@
 from .schemas import definitions, get_stack_versions
 from .semver import Version
 from .utils import dict_hash, get_path, get_etc_path, load_dump
+from .integrations import build_integrations_manifest
 
 RULES_DIR = get_path('rules')
 GH_CONFIG = Path.home() / ".config" / "gh" / "hosts.yml"
@@ -1088,3 +1089,17 @@ def get_branches(outfile: Path):
     branch_list = get_stack_versions(drop_patch=True)
     target_branches = json.dumps(branch_list[:-1]) + "\n"
     outfile.write_text(target_branches)
+
+
+@dev_group.group('integrations')
+def integrations_group():
+    """Commands for dev integrations methods."""
+
+
+@integrations_group.command('build-manifests')
+@click.option('--overwrite', '-o', is_flag=True, help="Overwrite the existing integrations-manifest.json.gz file")
+@click.option("--token", required=True, prompt=get_github_token() is None, default=get_github_token(),
+              help="GitHub token to use for the PR", hide_input=True)
+def build_integration_manifests(overwrite: bool, token: str):
+    """Builds consolidated integrations manifests file."""
+    build_integrations_manifest(token, overwrite)

detection_rules/etc/integration-manifests.json.gz

@@ -0,0 +1,122 @@
+# Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+# or more contributor license agreements. Licensed under the Elastic License
+# 2.0; you may not use this file except in compliance with the Elastic License
+# 2.0.
+
+"""Functions to support and interact with Kibana integrations."""
+import gzip
+import json
+import os
+import re
+from pathlib import Path
+from typing import Union
+
+import yaml
+from marshmallow import EXCLUDE, Schema, fields, post_load
+
+from .ghwrap import GithubClient
+from .semver import Version
+from .utils import INTEGRATION_RULE_DIR, cached, get_etc_path, read_gzip
+
+MANIFEST_FILE_PATH = Path(get_etc_path('integration-manifests.json.gz'))
+
+
+@cached
+def load_integrations_manifests() -> dict:
+    """Load the consolidated integrations manifest."""
+    return json.loads(read_gzip(get_etc_path('integration-manifests.json.gz')))
+
+
+class IntegrationManifestSchema(Schema):
+    name = fields.Str(required=True)
+    version = fields.Str(required=True)
+    release = fields.Str(required=True)
+    description = fields.Str(required=True)
+    conditions = fields.Dict(required=True)
+    policy_templates = fields.List(fields.Dict, required=True)
+    owner = fields.Dict(required=True)
+
+    @post_load
+    def transform_policy_template(self, data, **kwargs):
+        data["policy_templates"] = [policy["name"] for policy in data["policy_templates"]]
+        return data
+
+
+def build_integrations_manifest(token: str, overwrite: bool) -> None:
+    """Builds a new local copy of manifest.yaml from integrations Github."""
+    if overwrite:
+        if os.path.exists(MANIFEST_FILE_PATH):
+            os.remove(MANIFEST_FILE_PATH)
+    rule_integrations = [d.name for d in Path(INTEGRATION_RULE_DIR).glob('*') if d.is_dir()]
+    if "endpoint" in rule_integrations:
+        rule_integrations.remove("endpoint")
+
+    final_integration_manifests = {integration: {} for integration in rule_integrations}
+
+    # initialize github client and point to package-storage prod
+    github = GithubClient(token)
+    client = github.authenticated_client
+    organization = client.get_organization("elastic")
+    repository = organization.get_repo("package-storage")
+    pkg_storage_prod_branch = repository.get_branch("production")
+    pkg_storage_branch_sha = pkg_storage_prod_branch.commit.sha
+
+    for integration in rule_integrations:
+        integration_manifests = get_integration_manifests(repository, pkg_storage_branch_sha,
+                                                          pkg_path=f"packages/{integration}")
+        for manifest in integration_manifests:
+            validated_manifest = IntegrationManifestSchema(unknown=EXCLUDE).load(manifest)
+            package_version = validated_manifest.pop("version")
+            final_integration_manifests[integration][package_version] = validated_manifest
+
+    manifest_file = gzip.open(MANIFEST_FILE_PATH, "w+")
+    manifest_file_bytes = json.dumps(final_integration_manifests).encode("utf-8")
+    manifest_file.write(manifest_file_bytes)
+
+
+def find_least_compatible_version(package: str, integration: str,
+                                  current_stack_version: str, packages_manifest: dict) -> Union[str, None]:
+    """Finds least compatible version for specified integration based on stack version supplied."""
+    integration_manifests = {k: v for k, v in sorted(packages_manifest[package].items(), key=Version)}
+
+    def compare_versions(int_ver: str, pkg_ver: str) -> bool:
+        """Compares integration and package version"""
+        pkg_major, pkg_minor = Version(pkg_ver)
+        integration_major, integration_minor = Version(int_ver)[:2]
+
+        if int(integration_major) < int(pkg_major) or int(pkg_major) > int(integration_major):
+            return False
+
+        compatible = Version(int_ver) <= Version(pkg_ver)
+        return compatible
+
+    for version, manifest in integration_manifests.items():
+        for kibana_compat_vers in re.sub(r"\>|\<|\=|\^", "", manifest["conditions"]["kibana.version"]).split(" || "):
+            if compare_versions(kibana_compat_vers, current_stack_version):
+                return version
+    print(f"no compatible version for integration {package}:{integration}")
+    return None
+
+
+def get_integration_manifests(repository, sha: str, pkg_path: str) -> list:
+    """Iterates over specified integrations from package-storage and combines manifests per version."""
+    integration = pkg_path.split("/")[-1]
+    versioned_packages = repository.get_dir_contents(pkg_path, ref=sha)
+    versions = [p.path.split("/")[-1] for p in versioned_packages]
+
+    manifests = []
+    for version in versions:
+        contents = repository.get_dir_contents(f"{pkg_path}/{version}", ref=sha)
+        print(f"Processing {integration} - Version: {version}")
+
+        processing_version = contents[0].path.split("/")[2]
+        manifest_content = [c for c in contents if "manifest" in c.path]
+
+        if len(manifest_content) < 1:
+            raise Exception(f"manifest file does not exist for {integration}:{processing_version}")
+
+        path = manifest_content[0].path
+        manifest_content = yaml.safe_load(repository.get_contents(path, ref=sha).decoded_content.decode())
+        manifests.append(manifest_content)
+
+    return manifests

detection_rules/integrations.py

@@ -17,11 +17,14 @@
 
 import eql
 import kql
+from kql.ast import FieldComparison
 from marko.block import Document as MarkoDocument
 from marko.ext.gfm import gfm
 from marshmallow import ValidationError, validates_schema
 
 from . import beats, ecs, utils
+from .integrations import (find_least_compatible_version,
+                           load_integrations_manifests)
 from .misc import load_current_package_version
 from .mixins import MarshmallowDataclassMixin, StackCompatMixin
 from .rule_formatter import nested_normalize, toml_write
@@ -165,6 +168,12 @@ class RequiredFields:
         type: definitions.NonEmptyStr
         ecs: bool
 
+    @dataclass
+    class RelatedIntegrations:
+        package: definitions.NonEmptyStr
+        version: definitions.NonEmptyStr
+        integration: Optional[definitions.NonEmptyStr]
+
     actions: Optional[list]
     author: List[str]
     building_block_type: Optional[str]
@@ -186,7 +195,7 @@ class RequiredFields:
     # explicitly NOT allowed!
     # output_index: Optional[str]
     references: Optional[List[str]]
-    related_integrations: Optional[List[str]] = field(metadata=dict(metadata=dict(min_compat="8.3")))
+    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")))
     risk_score: definitions.RiskScore
     risk_score_mapping: Optional[List[RiskScoreMapping]]
@@ -665,7 +674,7 @@ def _post_dict_transform(self, obj: dict) -> dict:
         """Transform the converted API in place before sending to Kibana."""
         super()._post_dict_transform(obj)
 
-        self.add_related_integrations(obj)
+        self._add_related_integrations(obj)
         self._add_required_fields(obj)
         self._add_setup(obj)
 
@@ -675,10 +684,37 @@ def _post_dict_transform(self, obj: dict) -> dict:
         subclass.from_dict(obj)
         return obj
 
-    def add_related_integrations(self, obj: dict) -> None:
+    def _add_related_integrations(self, obj: dict) -> None:
         """Add restricted field related_integrations to the obj."""
-        # field_name = "related_integrations"
-        ...
+        field_name = "related_integrations"
+        package_integrations = obj.get(field_name, [])
+
+        if not package_integrations and self.metadata.integration:
+            packages_manifest = load_integrations_manifests()
+            current_stack_version = load_current_package_version()
+
+            if self.check_restricted_field_version(field_name):
+                if isinstance(self.data, QueryRuleData) and self.data.language != 'lucene':
+                    package_integrations = self._get_packaged_integrations(packages_manifest)
+
+                    if not package_integrations:
+                        return
+
+                    for package in package_integrations:
+                        package["version"] = find_least_compatible_version(
+                            package=package["package"],
+                            integration=package["integration"],
+                            current_stack_version=current_stack_version,
+                            packages_manifest=packages_manifest)
+
+                        # if integration is not a policy template remove
+                        if package["version"]:
+                            policy_templates = packages_manifest[
+                                package["package"]][package["version"]]["policy_templates"]
+                            if package["integration"] not in policy_templates:
+                                del package["integration"]
+
+            obj.setdefault("related_integrations", package_integrations)
 
     def _add_required_fields(self, obj: dict) -> None:
         """Add restricted field required_fields to the obj, derived from the query AST."""
@@ -689,7 +725,7 @@ def _add_required_fields(self, obj: dict) -> None:
             required_fields = []
 
         field_name = "required_fields"
-        if self.check_restricted_field_version(field_name=field_name):
+        if required_fields and self.check_restricted_field_version(field_name=field_name):
             obj.setdefault(field_name, required_fields)
 
     def _add_setup(self, obj: dict) -> None:
@@ -759,6 +795,31 @@ def compare_field_versions(min_stack: Version, max_stack: Version) -> bool:
         max_stack = max_stack or current_version
         return Version(min_stack) <= current_version >= Version(max_stack)
 
+    def _get_packaged_integrations(self, package_manifest: dict) -> Optional[List[dict]]:
+        packaged_integrations = []
+        datasets = set()
+
+        for node in self.data.get('ast', []):
+            if isinstance(node, eql.ast.Comparison) and str(node.left) == 'event.dataset':
+                datasets.update(set(n.value for n in node if isinstance(n, eql.ast.Literal)))
+            elif isinstance(node, FieldComparison) and str(node.field) == 'event.dataset':
+                datasets.update(set(str(n) for n in node if isinstance(n, kql.ast.Value)))
+
+        if not datasets:
+            return
+
+        for value in sorted(datasets):
+            integration = 'Unknown'
+            if '.' in value:
+                package, integration = value.split('.', 1)
+            else:
+                package = value
+
+            if package in list(package_manifest):
+                packaged_integrations.append({"package": package, "integration": integration})
+
+        return packaged_integrations
+
     @validates_schema
     def post_validation(self, value: dict, **kwargs):
         """Additional validations beyond base marshmallow schemas."""

detection_rules/rule.py

@@ -33,6 +33,7 @@
 CURR_DIR = os.path.dirname(os.path.abspath(__file__))
 ROOT_DIR = os.path.dirname(CURR_DIR)
 ETC_DIR = os.path.join(ROOT_DIR, "detection_rules", "etc")
+INTEGRATION_RULE_DIR = os.path.join(ROOT_DIR, "rules", "integrations")
 
 
 class NonelessDict(dict):