Add tests to ensure rules are properly deprecated (#1050)

* Add tests to ensure rules are properly deprecated
* add deprecate-rule command
This commit is contained in:
Justin Ibarra
2021-03-17 00:31:33 -05:00
committed by GitHub
parent 93f8f2dd94
commit d4cc4432ce
8 changed files with 181 additions and 12 deletions
+26
View File
@@ -298,6 +298,32 @@ def search_rule_prs(ctx, no_loop, query, columns, language, token, threads):
ctx.invoke(search_rules, query=query, columns=columns, language=language, rules=all_rules, pager=True)
@dev_group.command('deprecate-rule')
@click.argument('rule-file', type=click.Path(dir_okay=False))
@click.pass_context
def deprecate_rule(ctx: click.Context, rule_file: str):
"""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 = Rule(path=rule_file, contents=contents)
if rule.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')
rule.metadata.update(updated_date=today, deprecation_date=today, maturity='deprecated')
deprecated_path = get_path('rules', '_deprecated', rule_file.name)
rule.save(new_path=deprecated_path, as_rule=True)
rule_file.unlink()
click.echo(f'Rule moved to {deprecated_path} - remember to git add this file')
@dev_group.group('test')
def test_group():
"""Commands for testing against stack resources."""
+10 -7
View File
@@ -20,6 +20,7 @@ import yaml
from . import rule_loader
from .misc import JS_LICENSE, cached
from .rule import Rule, downgrade_contents_from_rule # noqa: F401
from .schemas import CurrentSchema
from .utils import Ndjson, get_path, get_etc_path, load_etc_dump, save_etc_dump
RELEASE_DIR = get_path("releases")
@@ -28,7 +29,7 @@ NOTICE_FILE = get_path('NOTICE.txt')
# CHANGELOG_FILE = Path(get_etc_path('rules-changelog.json'))
def filter_rule(rule: Rule, config_filter: dict, exclude_fields: dict) -> bool:
def filter_rule(rule: Rule, config_filter: dict, exclude_fields: Optional[dict] = None) -> bool:
"""Filter a rule based off metadata and a package configuration."""
flat_rule = rule.flattened_contents
for key, values in config_filter.items():
@@ -46,6 +47,7 @@ def filter_rule(rule: Rule, config_filter: dict, exclude_fields: dict) -> bool:
if len(rule_values & values) == 0:
return False
exclude_fields = exclude_fields or {}
for index, fields in exclude_fields.items():
if rule.unique_fields and (rule.contents['index'] == index or index == 'any'):
if set(rule.unique_fields) & set(fields):
@@ -66,8 +68,9 @@ def load_versions(current_versions: dict = None):
return current_versions or load_etc_dump('version.lock.json')
def manage_versions(rules: list, deprecated_rules: list = None, current_versions: dict = None,
exclude_version_update=False, add_new=True, save_changes=False, verbose=True) -> (list, list, list):
def manage_versions(rules: List[Rule], deprecated_rules: list = None, current_versions: dict = None,
exclude_version_update=False, add_new=True, save_changes=False,
verbose=True) -> (List[str], List[str], List[str]):
"""Update the contents of the version.lock file and optionally save changes."""
new_rules = {}
changed_rules = []
@@ -103,13 +106,12 @@ def manage_versions(rules: list, deprecated_rules: list = None, current_versions
if deprecated_rules:
rule_deprecations = load_etc_dump('deprecated_rules.json')
deprecation_date = str(datetime.date.today())
for rule in deprecated_rules:
if rule.id not in rule_deprecations:
rule_deprecations[rule.id] = {
'rule_name': rule.name,
'deprecation_date': deprecation_date
'deprecation_date': rule.metadata['deprecation_date'],
'stack_version': CurrentSchema.STACK_VERSION
}
newly_deprecated.append(rule.id)
@@ -129,7 +131,8 @@ def manage_versions(rules: list, deprecated_rules: list = None, current_versions
click.echo('Updated version.lock.json file')
if newly_deprecated:
save_etc_dump(sorted(OrderedDict(rule_deprecations)), 'deprecated_rules.json')
save_etc_dump(OrderedDict(sorted(rule_deprecations.items(), key=lambda e: e[1]['rule_name'])),
'deprecated_rules.json')
if verbose:
click.echo('Updated deprecated_rules.json file')
+7 -2
View File
@@ -234,9 +234,14 @@ def filter_rules(rules, metadata_field, value):
return [rule for rule in rules if rule.metadata.get(metadata_field, '') == value]
def get_production_rules(verbose=False):
def get_production_rules(verbose=False, include_deprecated=False) -> List[Rule]:
"""Get rules with a maturity of production."""
return filter_rules(load_rules(verbose=verbose).values(), 'maturity', 'production')
from .packaging import filter_rule
maturity = ['production']
if include_deprecated:
maturity.append('deprecated')
return [rule for rule in load_rules(verbose=verbose).values() if filter_rule(rule, {'maturity': maturity})]
@cached
+1
View File
@@ -72,6 +72,7 @@ class TomlMetadata(GenericSchema):
# rule validated against each ecs schema contained
beats_version = jsl.StringField(pattern=VERSION_PATTERN, required=False)
comments = jsl.StringField(required=False)
deprecation_date = jsl.StringField(required=False, pattern=DATE_PATTERN, default=time.strftime('%Y/%m/%d'))
ecs_versions = jsl.ArrayField(jsl.StringField(pattern=BRANCH_PATTERN, required=True), required=False)
maturity = jsl.StringField(enum=MATURITY_LEVELS, default='development', required=True)
+23
View File
@@ -0,0 +1,23 @@
# Deprecating rules
Rules that have been version locked (added to [version.lock.json](../etc/version.lock.json)), which also means they
have been added to the detection engine in Kibana, must be properly [deprecated](#steps-to-properly-deprecate-a-rule).
If a rule was never version locked (not yet pushed to Kibana or still in non-`production` `maturity`), the rule can
simply be removed with no additional changes, or updated the `maturity = "development"`, which will leave it out of the
release package to Kibana.
## Steps to properly deprecate a rule
1. Update the `maturity` to `deprecated`
2. Move the rule file to [rules/_deprecated](../rules/_deprecated)
3. Add `deprecation_date` and update `updated_date` to match
Next time the versions are locked, the rule will be added to the [deprecated_rules.json](../etc/deprecated_rules.json)
file.
### Using the deprecate-rule command
Alternatively, you can run `python -m detection_rules dev deprecate-rule <rule-file>`, which will perform all the steps
+7 -1
View File
@@ -1 +1,7 @@
{}
{
"3a86e085-094c-412d-97ff-2439731e59cb": {
"deprecation_date": "2021-03-03",
"rule_name": "Setgid Bit Set via chmod",
"stack_version": "7.13"
}
}
@@ -0,0 +1,56 @@
[metadata]
creation_date = "2020/04/23"
deprecation_date = "2021/03/16"
maturity = "deprecated"
updated_date = "2021/03/16"
[rule]
author = ["Elastic"]
description = """
An adversary may add the setgid bit to a file or directory in order to run a file with the privileges of the owning
group. An adversary can take advantage of this to either do a shell escape or exploit a vulnerability in an application
with the setgid bit to get code running in a different users context. Additionally, adversaries can use this mechanism
on their own malware to make sure they're able to execute in elevated contexts in the future.
"""
from = "now-9m"
index = ["auditbeat-*", "logs-endpoint.events.*"]
language = "lucene"
license = "Elastic License"
max_signals = 33
name = "Setgid Bit Set via chmod"
risk_score = 21
rule_id = "3a86e085-094c-412d-97ff-2439731e59cb"
severity = "low"
tags = ["Elastic", "Host", "Linux", "Threat Detection", "Privilege Escalation"]
timestamp_override = "event.ingested"
type = "query"
query = '''
event.category:process AND event.type:(start or process_started) AND process.name:chmod AND process.args:(g+s OR /2[0-9]{3}/) AND NOT user.name:root
'''
[[rule.threat]]
framework = "MITRE ATT&CK"
[[rule.threat.technique]]
id = "T1548"
name = "Abuse Elevation Control Mechanism"
reference = "https://attack.mitre.org/techniques/T1548/"
[[rule.threat.technique.subtechnique]]
id = "T1548.001"
name = "Setuid and Setgid"
reference = "https://attack.mitre.org/techniques/T1548/001/"
[rule.threat.tactic]
id = "TA0004"
name = "Privilege Escalation"
reference = "https://attack.mitre.org/tactics/TA0004/"
[[rule.threat]]
framework = "MITRE ATT&CK"
[rule.threat.tactic]
id = "TA0003"
name = "Persistence"
reference = "https://attack.mitre.org/tactics/TA0003/"
+51 -2
View File
@@ -19,8 +19,9 @@ import pytoml
from rta import get_ttp_names
from detection_rules import attack, beats, ecs
from detection_rules.packaging import load_versions
from detection_rules.rule_loader import FILE_PATTERN, find_unneeded_defaults_from_rule
from detection_rules.utils import load_etc_dump
from detection_rules.utils import get_path, load_etc_dump
from detection_rules.rule import Rule
from .base import BaseRuleTest
@@ -60,9 +61,20 @@ class TestValidRules(BaseRuleTest):
def test_all_rules_as_rule_schema(self):
"""Ensure that every rule file validates against the rule schema."""
rules_path = get_path('rules')
for file_name, contents in self.rule_files.items():
rule = Rule(file_name, contents)
rule.validate(as_rule=True)
if rule.metadata['maturity'] == 'deprecated':
continue
try:
rule.validate(as_rule=True)
except jsonschema.ValidationError as exc:
rule_path = Path(rule.path).relative_to(rules_path)
exc.message = f'{rule_path} -> {exc}'
raise exc
def test_all_rule_queries_optimized(self):
"""Ensure that every rule query is in optimized form."""
@@ -430,6 +442,43 @@ class TestRuleMetadata(BaseRuleTest):
err_msg = f'The following rules have an updated_date older than the creation_date\n {rules_str}'
self.fail(err_msg)
def test_deprecated_rules(self):
"""Test that deprecated rules are properly handled."""
versions = load_versions()
deprecations = load_etc_dump('deprecated_rules.json')
deprecated_rules = {}
for rule in self.rules:
maturity = rule.metadata['maturity']
if maturity == 'deprecated':
deprecated_rules[rule.id] = rule
err_msg = f'{self.rule_str(rule)} cannot be deprecated if it has not been version locked. ' \
f'Convert to `development` or delete the rule file instead'
self.assertIn(rule.id, versions, err_msg)
rule_path = Path(rule.path).relative_to(get_path('rules'))
err_msg = f'{self.rule_str(rule)} deprecated rules should be stored in ' \
f'"{get_path("rules", "_deprecated")}" folder'
self.assertEqual('_deprecated', rule_path.parts[0], err_msg)
err_msg = f'{self.rule_str(rule)} missing deprecation date'
self.assertIn('deprecation_date', rule.metadata, err_msg)
err_msg = f'{self.rule_str(rule)} deprecation_date and updated_date should match'
self.assertEqual(rule.metadata['deprecation_date'], rule.metadata['updated_date'], err_msg)
missing_rules = sorted(set(versions).difference(set(self.rule_lookup)))
missing_rule_strings = '\n '.join(f'{r} - {versions[r]["rule_name"]}' for r in missing_rules)
err_msg = f'Deprecated rules should not be removed, but moved to the rules/_deprecated folder instead. ' \
f'The following rules have been version locked and are missing. ' \
f'Re-add to the deprecated folder and update maturity to "deprecated": \n {missing_rule_strings}'
self.assertEqual([], missing_rules, err_msg)
for rule_id, entry in deprecations.items():
rule_str = f'{rule_id} - {entry["rule_name"]} ->'
self.assertIn(rule_id, deprecated_rules, f'{rule_str} is logged in "deprecated_rules.json" but is missing')
class TestTuleTiming(BaseRuleTest):
"""Test rule timing and timestamps."""