[FR][DAC] Import Rules Verbose Message (#4093)

* Draft Verbose Message

* Fix Linting

* Made more descriptive

* Updated for readability
This commit is contained in:
Eric Forte
2024-10-09 17:19:59 -04:00
committed by GitHub
parent 281926052c
commit 4edef2ea80
+43 -32
View File
@@ -94,6 +94,45 @@ def kibana_import_rules(ctx: click.Context, rules: RuleCollection, overwrite: Op
overwrite_exceptions: Optional[bool] = False,
overwrite_action_connectors: Optional[bool] = False) -> (dict, List[RuleResource]):
"""Import custom rules into Kibana."""
def _handle_response_errors(response: dict):
"""Handle errors from the import response."""
def _parse_list_id(s: str):
"""Parse the list ID from the error message."""
match = re.search(r'list_id: "(.*?)"', s)
return match.group(1) if match else None
# Re-try to address known Kibana issue: https://github.com/elastic/kibana/issues/143864
workaround_errors = []
flattened_exceptions = [e for sublist in exception_dicts for e in sublist]
all_exception_list_ids = {exception["list_id"] for exception in flattened_exceptions}
click.echo(f'{len(response["errors"])} rule(s) failed to import!')
for error in response['errors']:
click.echo(f' - {error["rule_id"]}: ({error["error"]["status_code"]}) {error["error"]["message"]}')
if "references a non existent exception list" in error["error"]["message"]:
list_id = _parse_list_id(error["error"]["message"])
if list_id in all_exception_list_ids:
workaround_errors.append(error["rule_id"])
if workaround_errors:
workaround_errors = list(set(workaround_errors))
click.echo(f'Missing exception list errors detected for {len(workaround_errors)} rules. '
'Try re-importing using the following command and rule IDs:\n')
click.echo('python -m detection_rules kibana import-rules -o ', nl=False)
click.echo(' '.join(f'-id {rule_id}' for rule_id in workaround_errors))
click.echo()
def _process_imported_items(imported_items_list, item_type_description, item_key):
"""Displays appropriately formatted success message that all items imported successfully."""
all_ids = {item[item_key] for sublist in imported_items_list for item in sublist}
if all_ids:
click.echo(f'{len(all_ids)} {item_type_description} successfully imported')
ids_str = '\n - '.join(all_ids)
click.echo(f' - {ids_str}')
kibana = ctx.obj['kibana']
rule_dicts = [r.contents.to_api_format() for r in rules]
with kibana:
@@ -113,43 +152,15 @@ def kibana_import_rules(ctx: click.Context, rules: RuleCollection, overwrite: Op
overwrite_action_connectors=overwrite_action_connectors
)
def handle_response_errors(response: dict):
"""Handle errors from the import response."""
def parse_list_id(s: str):
"""Parse the list ID from the error message."""
match = re.search(r'list_id: "(.*?)"', s)
return match.group(1) if match else None
# Re-try to address known Kibana issue: https://github.com/elastic/kibana/issues/143864
workaround_errors = []
flattened_exceptions = [e for sublist in exception_dicts for e in sublist]
all_exception_list_ids = {exception["list_id"] for exception in flattened_exceptions}
click.echo(f'{len(response["errors"])} rule(s) failed to import!')
for error in response['errors']:
click.echo(f' - {error["rule_id"]}: ({error["error"]["status_code"]}) {error["error"]["message"]}')
if "references a non existent exception list" in error["error"]["message"]:
list_id = parse_list_id(error["error"]["message"])
if list_id in all_exception_list_ids:
workaround_errors.append(error["rule_id"])
if workaround_errors:
workaround_errors = list(set(workaround_errors))
click.echo(f'Missing exception list errors detected for {len(workaround_errors)} rules. '
'Try re-importing using the following command and rule IDs:\n')
click.echo('python -m detection_rules kibana import-rules -o ', nl=False)
click.echo(' '.join(f'-id {rule_id}' for rule_id in workaround_errors))
click.echo()
if successful_rule_ids:
click.echo(f'{len(successful_rule_ids)} rule(s) successfully imported')
rule_str = '\n - '.join(successful_rule_ids)
click.echo(f' - {rule_str}')
if response['errors']:
handle_response_errors(response)
_handle_response_errors(response)
else:
_process_imported_items(exception_dicts, 'exception list(s)', 'list_id')
_process_imported_items(action_connectors_dicts, 'action connector(s)', 'id')
return response, results