Add export-rules command

CLI.md

@@ -167,6 +167,22 @@ Options:
   -h, --help  Show this message and exit.
 ```
 
+Alternatively, rules can be exported into a consolidated ndjson file which can be imported in the Kibana security app 
+directly.
+
+```console
+Usage: detection_rules export-rules [OPTIONS] [RULE_ID]...
+
+  Export rule(s) into an importable ndjson file.
+
+Options:
+  -f, --rule-file FILE       Export specified rule files
+  -d, --directory DIRECTORY  Recursively export rules from a directory
+  -o, --outfile FILE         Name of file for exported rules
+  -r, --randomize-id         Randomize rule IDs before export
+  -h, --help                 Show this message and exit.
+```
+
 _*To load a custom rule, the proper index must be setup first. The simplest way to do this is to click 
 the `Load prebuilt detection rules and timeline templates` button on the `detections` page in the Kibana security app._
 

detection_rules/main.py

@@ -7,6 +7,8 @@
 import json
 import os
 import re
+import time
+from pathlib import Path
 
 import click
 import jsonschema
@@ -130,7 +132,7 @@ def mass_update(ctx, query, metadata, language, field):
 @click.option('--rule-file', '-f', type=click.Path(dir_okay=False), help='Optionally view a rule from a specified file')
 @click.option('--api-format/--rule-format', default=True, help='Print the rule in final api or rule format')
 @click.pass_context
-def view_rule(ctx, rule_id, rule_file, api_format):
+def view_rule(ctx, rule_id, rule_file, api_format, verbose=True):
     """View an internal rule or specified rule file."""
     rule = None
 
@@ -149,12 +151,48 @@ def view_rule(ctx, rule_id, rule_file, api_format):
     if not rule:
         client_error('Unknown format!')
 
-    click.echo(toml_write(rule.rule_format()) if not api_format else
-               json.dumps(rule.contents, indent=2, sort_keys=True))
+    if verbose:
+        click.echo(toml_write(rule.rule_format()) if not api_format else
+                   json.dumps(rule.contents, indent=2, sort_keys=True))
 
     return rule
 
 
+@root.command('export-rules')
+@click.argument('rule-id', nargs=-1, required=False)
+@click.option('--rule-file', '-f', multiple=True, type=click.Path(dir_okay=False), help='Export specified rule files')
+@click.option('--directory', '-d', multiple=True, type=click.Path(file_okay=False),
+              help='Recursively export rules from a directory')
+@click.option('--outfile', '-o', default=get_path('exports', f'{time.strftime("%Y%m%dT%H%M%SL")}.ndjson'),
+              type=click.Path(dir_okay=False), help='Name of file for exported rules')
+@click.option('--randomize-id', '-r', is_flag=True, help='Randomize rule IDs before export')
+def export_rules(rule_id, rule_file, directory, outfile, randomize_id):
+    """Export rule(s) into an importable ndjson file."""
+    from .packaging import Package
+
+    if not (rule_id or rule_file or directory):
+        client_error('Must specify a rule_id, rule_file, or directory')
+
+    rules = [r for r in rule_loader.load_rules(verbose=False).values() if r.id in rule_id] if rule_id else []
+
+    rule_files = list(rule_file)
+    for _dir in directory:
+        rule_files.extend(list(Path(_dir).rglob('*.toml')))
+
+    file_lookup = rule_loader.load_rule_files(verbose=False, paths=rule_files)
+    rules.extend(rule_loader.load_rules(file_lookup=file_lookup).values())
+
+    if randomize_id:
+        from uuid import uuid4
+        for rule in rules:
+            rule.contents['rule_id'] = str(uuid4())
+
+    Path(outfile).parent.mkdir(exist_ok=True)
+    package = Package(rules, '_', verbose=False)
+    package.export(outfile)
+    return package.rules
+
+
 @root.command('validate-rule')
 @click.argument('rule-id', required=False)
 @click.option('--rule-name', '-n')

detection_rules/packaging.py

@@ -10,6 +10,7 @@
 import os
 import shutil
 from collections import defaultdict, OrderedDict
+from typing import List
 
 import click
 
@@ -137,24 +138,25 @@ class Package(object):
     """Packaging object for siem rules and releases."""
 
     def __init__(self, rules, name, deprecated_rules=None, release=False, current_versions=None, min_version=None,
-                 max_version=None, update_version_lock=False):
+                 max_version=None, update_version_lock=False, verbose=True):
         """Initialize a package."""
-        self.rules = [r.copy() for r in rules]  # type: list[Rule]
+        self.rules: List[Rule] = [r.copy() for r in rules]
         self.name = name
-        self.deprecated_rules = [r.copy() for r in deprecated_rules or []]  # type: list[Rule]
+        self.deprecated_rules: List[Rule] = [r.copy() for r in deprecated_rules or []]
         self.release = release
 
         self.changed_rule_ids, self.new_rules_ids, self.removed_rule_ids = self._add_versions(current_versions,
-                                                                                              update_version_lock)
+                                                                                              update_version_lock,
+                                                                                              verbose=verbose)
 
         if min_version or max_version:
             self.rules = [r for r in self.rules
                           if (min_version or 0) <= r.contents['version'] <= (max_version or r.contents['version'])]
 
-    def _add_versions(self, current_versions, update_versions_lock=False):
+    def _add_versions(self, current_versions, update_versions_lock=False, verbose=True):
         """Add versions to rules at load time."""
         return manage_versions(self.rules, deprecated_rules=self.deprecated_rules, current_versions=current_versions,
-                               save_changes=update_versions_lock)
+                               save_changes=update_versions_lock, verbose=verbose)
 
     @staticmethod
     def _package_notice_file(save_dir):
@@ -245,6 +247,17 @@ def save(self, verbose=True):
         if verbose:
             click.echo('Package saved to: {}'.format(save_dir))
 
+    def export(self, outfile, verbose=True):
+        """Export rules into a consolidated ndjson file."""
+        base, _ = os.path.splitext(outfile)
+        outfile = base + '.ndjson'
+
+        with open(outfile, 'w') as f:
+            f.write('\n'.join(json.dumps(r.contents, sort_keys=True) for r in self.rules))
+
+        if verbose:
+            click.echo(f'Exported {len(self.rules)} rules into {outfile}')
+
     def get_package_hash(self, as_api=True, verbose=True):
         """Get hash of package contents."""
         contents = base64.b64encode(self.get_consolidated(as_api=as_api).encode('utf-8'))