Ross Wolf
150 lines
6.3 KiB
Python
150 lines
6.3 KiB
Python
# Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
|
|
# or more contributor license agreements. Licensed under the Elastic License;
|
|
# you may not use this file except in compliance with the Elastic License.
|
|
|
|
"""Test that the packages are built correctly."""
|
|
import unittest
|
|
import uuid
|
|
import yaml
|
|
|
|
from detection_rules import rule_loader
|
|
from detection_rules.packaging import Package, PACKAGE_FILE
|
|
|
|
|
|
class TestPackages(unittest.TestCase):
|
|
"""Test package building and saving."""
|
|
|
|
@staticmethod
|
|
def get_test_rule(version=1, count=1):
|
|
def get_rule_contents():
|
|
contents = {
|
|
"author": ["Elastic"],
|
|
"description": "test description",
|
|
"language": "kuery",
|
|
"license": "Elastic License",
|
|
"name": "test rule",
|
|
"query": "process.name:test.query",
|
|
"risk_score": 21,
|
|
"rule_id": str(uuid.uuid4()),
|
|
"severity": "low",
|
|
"type": "query"
|
|
}
|
|
return contents
|
|
|
|
rules = [rule_loader.Rule('test.toml', get_rule_contents()) for i in range(count)]
|
|
version_info = {
|
|
rule.id: {
|
|
'rule_name': rule.name,
|
|
'sha256': rule.get_hash(),
|
|
'version': version
|
|
} for rule in rules
|
|
}
|
|
|
|
return rules, version_info
|
|
|
|
def test_package_loader_production_config(self):
|
|
"""Test that packages are loading correctly."""
|
|
|
|
def test_package_loader_default_configs(self):
|
|
"""Test configs in etc/packages.yml."""
|
|
with open(PACKAGE_FILE) as f:
|
|
configs = yaml.safe_load(f)['package']
|
|
|
|
package = Package.from_config(configs)
|
|
for rule in package.rules:
|
|
rule.contents.pop('version')
|
|
rule.validate(as_rule=True)
|
|
|
|
@rule_loader.mock_loader
|
|
def test_package_summary(self):
|
|
"""Test the generation of the package summary."""
|
|
rules = list(rule_loader.load_rules().values())
|
|
package = Package(rules, 'test-package')
|
|
changed_rules, new_rules = package.bump_versions(save_changes=False)
|
|
package.generate_summary(changed_rules, new_rules)
|
|
|
|
def test_versioning_diffs(self):
|
|
"""Test that versioning is detecting diffs as expected."""
|
|
rules, version_info = self.get_test_rule()
|
|
package = Package(rules, 'test', current_versions=version_info)
|
|
|
|
# test versioning doesn't falsely detect changes
|
|
changed_rules, new_rules = package.changed_rules, package.new_rules
|
|
|
|
self.assertEqual(0, len(changed_rules), 'Package version bumping is improperly detecting changed rules')
|
|
self.assertEqual(0, len(new_rules), 'Package version bumping is improperly detecting new rules')
|
|
self.assertEqual(1, package.rules[0].contents['version'], 'Package version bumping unexpectedly')
|
|
|
|
# test versioning detects a new rule
|
|
package.rules[0].contents.pop('version')
|
|
changed_rules, new_rules = package.bump_versions(current_versions={})
|
|
|
|
self.assertEqual(0, len(changed_rules), 'Package version bumping is improperly detecting changed rules')
|
|
self.assertEqual(1, len(new_rules), 'Package version bumping is not detecting new rules')
|
|
self.assertEqual(1, package.rules[0].contents['version'],
|
|
'Package version bumping not setting version to 1 for new rules')
|
|
|
|
# test versioning detects a hash changes
|
|
package.rules[0].contents.pop('version')
|
|
package.rules[0].contents['query'] = 'process.name:changed.test.query'
|
|
changed_rules, new_rules = package.bump_versions(current_versions=version_info)
|
|
|
|
self.assertEqual(1, len(changed_rules), 'Package version bumping is not detecting changed rules')
|
|
self.assertEqual(0, len(new_rules), 'Package version bumping is improperly detecting new rules')
|
|
self.assertEqual(2, package.rules[0].contents['version'], 'Package version not bumping on changes')
|
|
|
|
@rule_loader.mock_loader
|
|
def test_rule_versioning(self):
|
|
"""Test that all rules are properly versioned and tracked"""
|
|
self.maxDiff = None
|
|
rules = rule_loader.load_rules().values()
|
|
original_hashes = []
|
|
post_bump_hashes = []
|
|
|
|
# test that no rules have versions defined
|
|
for rule in rules:
|
|
self.assertIsNone(rule.contents.get('version'), '{} - {}: explicitly sets a version in the rule file')
|
|
original_hashes.append(rule.get_hash())
|
|
|
|
package = Package(rules, 'test-package')
|
|
|
|
# test that all rules have versions defined
|
|
# package.bump_versions(save_changes=False)
|
|
for rule in package.rules:
|
|
self.assertGreaterEqual(rule.contents.get('version'), 1, '{} - {}: version is not being set in package')
|
|
|
|
# test that rules validate with version
|
|
for rule in package.rules:
|
|
rule.validate(versioned=True)
|
|
rule.contents.pop('version')
|
|
post_bump_hashes.append(rule.get_hash())
|
|
|
|
# test that no hashes changed as a result of the version bumps
|
|
self.assertListEqual(original_hashes, post_bump_hashes, 'Version bumping modified the hash of a rule')
|
|
|
|
def test_version_filter(self):
|
|
"""Test that version filtering is working as expected."""
|
|
msg = 'Package version filter failing'
|
|
|
|
rules, version_info = self.get_test_rule(version=1, count=3)
|
|
package = Package(rules, 'test', current_versions=version_info, min_version=2)
|
|
self.assertEqual(0, len(package.rules), msg)
|
|
|
|
rules, version_info = self.get_test_rule(version=5, count=3)
|
|
package = Package(rules, 'test', current_versions=version_info, max_version=2)
|
|
self.assertEqual(0, len(package.rules), msg)
|
|
|
|
rules, version_info = self.get_test_rule(version=2, count=3)
|
|
package = Package(rules, 'test', current_versions=version_info, min_version=1, max_version=3)
|
|
self.assertEqual(3, len(package.rules), msg)
|
|
|
|
rules, version_info = self.get_test_rule(version=1, count=3)
|
|
|
|
version = 1
|
|
for rule_id, vinfo in version_info.items():
|
|
vinfo['version'] = version
|
|
version += 1
|
|
|
|
package = Package(rules, 'test', current_versions=version_info, min_version=2, max_version=2)
|
|
self.assertEqual(1, len(package.rules), msg)
|