[Bug] CLI Fixes

detection_rules/devtools.py

@@ -31,6 +31,7 @@
 from .packaging import PACKAGE_FILE, Package, RELEASE_DIR, current_stack_version, manage_versions
 from .rule import AnyRuleData, BaseRuleData, QueryRuleData, TOMLRule
 from .rule_loader import RuleCollection, production_filter
+from .schemas import definitions
 from .semver import Version
 from .utils import dict_hash, get_path, load_dump
 
@@ -582,29 +583,36 @@ def search_rule_prs(ctx, no_loop, query, columns, language, token, threads):
     all_rules = {}
     new, modified, errors = rule_loader.load_github_pr_rules(token=token, threads=threads)
 
-    def add_github_meta(this_rule, status, original_rule_id=None):
+    def add_github_meta(this_rule: TOMLRule, status: str, original_rule_id: Optional[definitions.UUIDString] = None):
         pr = this_rule.gh_pr
-        rule.metadata['status'] = status
-        rule.metadata['github'] = {
-            'base': pr.base.label,
-            'comments': [c.body for c in pr.get_comments()],
-            'commits': pr.commits,
-            'created_at': str(pr.created_at),
-            'head': pr.head.label,
-            'is_draft': pr.draft,
-            'labels': [lbl.name for lbl in pr.get_labels()],
-            'last_modified': str(pr.last_modified),
-            'title': pr.title,
-            'url': pr.html_url,
-            'user': pr.user.login
+        new_data = {}
+        new_meta = {
+            'status': status,
+            'github': {
+                'base': pr.base.label,
+                'comments': [c.body for c in pr.get_comments()],
+                'commits': pr.commits,
+                'created_at': str(pr.created_at),
+                'head': pr.head.label,
+                'is_draft': pr.draft,
+                'labels': [lbl.name for lbl in pr.get_labels()],
+                'last_modified': str(pr.last_modified),
+                'title': pr.title,
+                'url': pr.html_url,
+                'user': pr.user.login
+            }
         }
 
         if original_rule_id:
-            rule.metadata['original_rule_id'] = original_rule_id
-            rule.contents['rule_id'] = str(uuid4())
+            new_meta['original_rule_id'] = original_rule_id
+            new_data = {'rule_id': str(uuid4())}
 
-        rule_path = f'pr-{pr.number}-{rule.path}'
-        all_rules[rule_path] = rule.rule_format()
+        rule_path = Path(f'pr-{pr.number}-{rule.path}')
+
+        new_rule = this_rule.new(path=rule_path, data=dict(meta=new_meta, **new_data))
+
+        rule_dict = {'metadata': new_rule.contents.metadata.to_dict(), 'rule': new_rule.contents.to_api_format()}
+        all_rules[rule_path] = rule_dict
 
     for rule_id, rule in new.items():
         add_github_meta(rule, 'new')
@@ -623,34 +631,33 @@ def add_github_meta(this_rule, status, original_rule_id=None):
 
 
 @dev_group.command('deprecate-rule')
-@click.argument('rule-file', type=click.Path(dir_okay=False))
+@click.argument('rule-file', type=Path)
 @click.pass_context
-def deprecate_rule(ctx: click.Context, rule_file: str):
+def deprecate_rule(ctx: click.Context, rule_file: Path):
     """Deprecate a rule."""
-    import pytoml
     from .packaging import load_versions
 
     version_info = load_versions()
-    rule_file = Path(rule_file)
-    contents = pytoml.loads(rule_file.read_text())
+    rule_collection = RuleCollection()
+    contents = rule_collection.load_file(rule_file).contents
     rule = TOMLRule(path=rule_file, contents=contents)
 
-    if rule.id not in version_info:
+    if rule.contents.id not in version_info:
         click.echo('Rule has not been version locked and so does not need to be deprecated. '
                    'Delete the file or update the maturity to `development` instead')
         ctx.exit()
 
     today = time.strftime('%Y/%m/%d')
 
-    new_meta = dataclasses.replace(rule.contents.metadata,
-                                   updated_date=today,
-                                   deprecation_date=today,
-                                   maturity='deprecated')
-    contents = dataclasses.replace(rule.contents, metadata=new_meta)
+    new_meta = {
+        'updated_date': today,
+        'deprecation_date': today,
+        'maturity': 'deprecated'
+    }
     deprecated_path = get_path('rules', '_deprecated', rule_file.name)
 
     # create the new rule and save it
-    new_rule = TOMLRule(contents=contents, path=Path(deprecated_path))
+    new_rule = rule.new(path=Path(deprecated_path), metadata=new_meta)
     new_rule.save_toml()
 
     # remove the old rule

detection_rules/eswrap.py

@@ -36,6 +36,7 @@ def get_elasticsearch_client(cloud_id=None, elasticsearch_url=None, es_user=None
     es_password = es_password or click.prompt("es_password", hide_input=True)
     hosts = [elasticsearch_url] if elasticsearch_url else None
     timeout = kwargs.pop('timeout', 60)
+    kwargs['verify_certs'] = not kwargs.pop('ignore_ssl_errors', False)
 
     try:
         client = Elasticsearch(hosts=hosts, cloud_id=cloud_id, http_auth=(es_user, es_password), timeout=timeout,

detection_rules/kbwrap.py

@@ -17,8 +17,11 @@
 from .utils import format_command_options
 
 
-def get_kibana_client(cloud_id, kibana_url, kibana_user, kibana_password, kibana_cookie, **kwargs):
+def get_kibana_client(cloud_id, kibana_url, kibana_user, kibana_password, kibana_cookie, space, ignore_ssl_errors,
+                      provider_type, provider_name, **kwargs):
     """Get an authenticated Kibana client."""
+    from requests import HTTPError
+
     if not (cloud_id or kibana_url):
         client_error("Missing required --cloud-id or --kibana-url")
 
@@ -27,11 +30,22 @@ def get_kibana_client(cloud_id, kibana_url, kibana_user, kibana_password, kibana
         kibana_user = kibana_user or click.prompt("kibana_user")
         kibana_password = kibana_password or click.prompt("kibana_password", hide_input=True)
 
-    with Kibana(cloud_id=cloud_id, kibana_url=kibana_url, **kwargs) as kibana:
+    verify = not ignore_ssl_errors
+
+    with Kibana(cloud_id=cloud_id, kibana_url=kibana_url, space=space, verify=verify, **kwargs) as kibana:
         if kibana_cookie:
             kibana.add_cookie(kibana_cookie)
-        else:
-            kibana.login(kibana_user, kibana_password)
+            return kibana
+
+        try:
+            kibana.login(kibana_user, kibana_password, provider_type=provider_type, provider_name=provider_name)
+        except HTTPError as exc:
+            if exc.response.status_code == 401:
+                err_msg = f'Authentication failed for {kibana_url}. If credentials are valid, check --provider-name'
+                client_error(err_msg, exc, err=True)
+            else:
+                raise
+
         return kibana
 
 
@@ -57,7 +71,6 @@ def kibana_group(ctx: click.Context, **kibana_kwargs):
 @click.pass_context
 def upload_rule(ctx, rules, replace_id):
     """Upload a list of rule .toml files to Kibana."""
-
     kibana = ctx.obj['kibana']
     api_payloads = []
 
@@ -78,8 +91,22 @@ def upload_rule(ctx, rules, replace_id):
         api_payloads.append(rule)
 
     with kibana:
-        rules = RuleResource.bulk_create(api_payloads)
-        click.echo(f"Successfully uploaded {len(rules)} rules")
+        results = RuleResource.bulk_create(api_payloads)
+
+    success = []
+    errors = []
+    for result in results:
+        if 'error' in result:
+            errors.append(f'{result["rule_id"]} - {result["error"]["message"]}')
+        else:
+            success.append(result['rule_id'])
+
+    if success:
+        click.echo('Successful uploads:\n  - ' + '\n  - '.join(success))
+    if errors:
+        click.echo('Failed uploads:\n  - ' + '\n  - '.join(errors))
+
+    return results
 
 
 @kibana_group.command('search-alerts')

detection_rules/main.py

@@ -4,14 +4,13 @@
 # 2.0.
 
 """CLI commands for detection_rules."""
-import dataclasses
 import glob
 import json
 import os
 import re
 import time
 from pathlib import Path
-from typing import Dict
+from typing import Dict, List, Optional
 from uuid import uuid4
 
 import click
@@ -21,7 +20,7 @@
 from .rule import TOMLRule, TOMLRuleContents
 from .rule_formatter import toml_write
 from .rule_loader import RuleCollection
-from .schemas import all_versions
+from .schemas import all_versions, definitions
 from .utils import get_path, clear_caches, load_rule_contents
 
 RULES_DIR = get_path('rules')
@@ -40,7 +39,7 @@ def root(ctx, debug):
 
 
 @root.command('create-rule')
-@click.argument('path', type=click.Path(dir_okay=False))
+@click.argument('path', type=Path)
 @click.option('--config', '-c', type=click.Path(exists=True, dir_okay=False), help='Rule or config file')
 @click.option('--required-only', is_flag=True, help='Only prompt for required fields')
 @click.option('--rule-type', '-t', type=click.Choice(sorted(TOMLRuleContents.all_rule_types())),
@@ -94,7 +93,7 @@ def import_rules(input_file, directory):
 
     rule_contents = []
     for rule_file in rule_files:
-        rule_contents.extend(load_rule_contents(rule_file))
+        rule_contents.extend(load_rule_contents(Path(rule_file)))
 
     if not rule_contents:
         click.echo('Must specify at least one file!')
@@ -155,7 +154,7 @@ def mass_update(ctx, query, metadata, language, field):
 
 
 @root.command('view-rule')
-@click.argument('rule-file')
+@click.argument('rule-file', type=Path)
 @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_file, api_format):
@@ -167,6 +166,44 @@ def view_rule(ctx, rule_file, api_format):
     else:
         click.echo(toml_write(rule.contents.to_dict()))
 
+    return rule
+
+
+def _export_rules(rules: List[TOMLRule], outfile: Path, downgrade_version: Optional[definitions.SemVer] = None,
+                  verbose=True, skip_unsupported=False):
+    """Export rules into a consolidated ndjson file."""
+    from .rule import downgrade_contents_from_rule
+
+    outfile = outfile.with_suffix('.ndjson')
+    unsupported = []
+
+    if downgrade_version:
+        if skip_unsupported:
+            output_lines = []
+
+            for rule in rules:
+                try:
+                    output_lines.append(json.dumps(downgrade_contents_from_rule(rule, downgrade_version),
+                                                   sort_keys=True))
+                except ValueError as e:
+                    unsupported.append(f'{e}: {rule.id} - {rule.name}')
+                    continue
+
+        else:
+            output_lines = [json.dumps(downgrade_contents_from_rule(r, downgrade_version), sort_keys=True)
+                            for r in rules]
+    else:
+        output_lines = [json.dumps(r.contents.to_api_format(), sort_keys=True) for r in rules]
+
+    outfile.write_text('\n'.join(output_lines) + '\n')
+
+    if verbose:
+        click.echo(f'Exported {len(rules) - len(unsupported)} rules into {outfile}')
+
+        if skip_unsupported and unsupported:
+            unsupported_str = '\n- '.join(unsupported)
+            click.echo(f'Skipped {len(unsupported)} unsupported rules: \n- {unsupported_str}')
+
 
 @root.command('export-rules')
 @multi_collection
@@ -180,25 +217,20 @@ def view_rule(ctx, rule_file, api_format):
                    '(an error will be raised otherwise)')
 def export_rules(rules, outfile, replace_id, stack_version, skip_unsupported):
     """Export rule(s) into an importable ndjson file."""
-    from .packaging import Package
-
     assert len(rules) > 0, "No rules found"
 
     if replace_id:
-        # if we need to replace the id, take each rule object and create a copy
-        # of it, with only the rule_id field changed
-        old_rules = rules
-        rules = RuleCollection()
+        new_rules = []
+        for rule in rules:
+            new_rules.append(rule.new(data=dict(rule_id=str(uuid4()))))
+
+        rules = new_rules
 
-        for rule in old_rules:
-            new_data = dataclasses.replace(rule.contents.data, rule_id=str(uuid4()))
-            new_contents = dataclasses.replace(rule.contents, data=new_data)
-            rules.add_rule(TOMLRule(contents=new_contents))
+    _export_rules(rules=rules, outfile=Path(outfile), downgrade_version=stack_version,
+                  skip_unsupported=skip_unsupported)
 
     Path(outfile).parent.mkdir(exist_ok=True)
-    package = Package(rules, '_', verbose=False)
-    package.export(outfile, downgrade_version=stack_version, skip_unsupported=skip_unsupported)
-    return package.rules
+    return rules
 
 
 @root.command('validate-rule')
@@ -230,9 +262,10 @@ def search_rules(query, columns, language, count, verbose=True, rules: Dict[str,
     from eql.build import get_engine
     from eql import parse_query
     from eql.pipes import CountPipe
+    from .rule import get_unique_query_fields
 
     flattened_rules = []
-    rules = rules or {str(rule.path): rule for rule in RuleCollection.default()}
+    rules = rules or {str(rule.path): rule.contents.to_dict() for rule in RuleCollection.default()}
 
     for file_name, rule_doc in rules.items():
         flat: dict = {"file": os.path.relpath(file_name)}
@@ -253,8 +286,8 @@ def search_rules(query, columns, language, count, verbose=True, rules: Dict[str,
             technique_ids.extend([t['id'] for t in techniques])
             subtechnique_ids.extend([st['id'] for t in techniques for st in t.get('subtechnique', [])])
 
-        flat.update(techniques=technique_ids, tactics=tactic_names, subtechniques=subtechnique_ids)
-        #           unique_fields=TOMLRule.get_unique_query_fields(rule_doc['rule']))
+        flat.update(techniques=technique_ids, tactics=tactic_names, subtechniques=subtechnique_ids,
+                    unique_fields=get_unique_query_fields(rule_doc['rule']))
         flattened_rules.append(flat)
 
     flattened_rules.sort(key=lambda dct: dct["name"])

detection_rules/misc.py

@@ -279,14 +279,18 @@ def getdefault(name):
         'kibana_password': click.Option(['--kibana-password', '-kp'], default=getdefault('kibana_password')),
         'kibana_url': click.Option(['--kibana-url'], default=getdefault('kibana_url')),
         'kibana_user': click.Option(['--kibana-user', '-ku'], default=getdefault('kibana_user')),
-        'space': click.Option(['--space'], default=None, help='Kibana space')
+        'provider_type': click.Option(['--provider-type'], default=getdefault('provider_type')),
+        'provider_name': click.Option(['--provider-name'], default=getdefault('provider_name')),
+        'space': click.Option(['--space'], default=None, help='Kibana space'),
+        'ignore_ssl_errors': click.Option(['--ignore-ssl-errors'], default=getdefault('ignore_ssl_errors'))
     },
     'elasticsearch': {
         'cloud_id': click.Option(['--cloud-id'], default=getdefault("cloud_id")),
         'elasticsearch_url': click.Option(['--elasticsearch-url'], default=getdefault("elasticsearch_url")),
         'es_user': click.Option(['--es-user', '-eu'], default=getdefault("es_user")),
         'es_password': click.Option(['--es-password', '-ep'], default=getdefault("es_password")),
-        'timeout': click.Option(['--timeout', '-et'], default=60, help='Timeout for elasticsearch client')
+        'timeout': click.Option(['--timeout', '-et'], default=60, help='Timeout for elasticsearch client'),
+        'ignore_ssl_errors': click.Option(['--ignore-ssl-errors'], default=getdefault('ignore_ssl_errors'))
     }
 }
 kibana_options = list(client_options['kibana'].values())