security-tools/sigma-rules
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

[Bug] resolves bug in Rule version methods (#2021)

Browse Source 
* [Bug] resolves bug in Rule version methods

* comment out unused code with notes

(cherry picked from commit 744f56d98e)
This commit is contained in:
Justin Ibarra
2022-06-07 15:40:46 -08:00
committed by github-actions[bot]
parent 57194b8e59
commit 8564185a7d
10 changed files with 123 additions and 114 deletions
Download Patch File Download Diff File Expand all files Collapse all files
detection_rules/version_lock.py
+49 -43
View File
@@ -20,7 +20,10 @@ ETC_VERSION_LOCK_FILE = "version.lock.json"
ETC_VERSION_LOCK_PATH = Path(get_etc_path()) / ETC_VERSION_LOCK_FILE
ETC_DEPRECATED_RULES_FILE = "deprecated_rules.json"
ETC_DEPRECATED_RULES_PATH = Path(get_etc_path()) / ETC_DEPRECATED_RULES_FILE
MIN_LOCK_VERSION_DEFAULT = Version("7.13.0")
# This was the original version the lock was created under. This constant has been replaced by
# schemas.get_min_supported_stack_version to dynamically determine the minimum
# MIN_LOCK_VERSION_DEFAULT = Version("7.13.0")
@dataclass(frozen=True)
@@ -34,8 +37,8 @@ class BaseEntry:
@dataclass(frozen=True)
class VersionLockFileEntry(MarshmallowDataclassMixin, BaseEntry):
"""Schema for a rule entry in the version lock."""
min_stack_version: Optional[definitions.SemVer]
previous: Optional[Dict[definitions.SemVer, BaseEntry]]
min_stack_version: Optional[definitions.SemVerMinorOnly]
previous: Optional[Dict[definitions.SemVerMinorOnly, BaseEntry]]
@dataclass(frozen=True)
@@ -82,10 +85,11 @@ class DeprecatedRulesFile(LockDataclassMixin):
def _convert_lock_version(stack_version: Optional[str]) -> Version:
"""Convert an optional stack version to the minimum for the lock."""
min_version = get_min_supported_stack_version()
min_version = get_min_supported_stack_version(drop_patch=True)
if stack_version is None:
return min_version
return max(Version(stack_version), min_version)
short_stack_version = Version(Version(stack_version)[:2])
return max(short_stack_version, min_version)
@cached
@@ -189,32 +193,26 @@ class VersionLock:
return list(changed_rules), list(new_rules), list(newly_deprecated)
verbose_echo('Rule changes detected!')
route = None
existing_rule_lock = {}
original_hash = None
changes = []
def add_changes(r, *msg):
if not original_hash or original_hash != current_rule_lock['sha256']:
new = [f' {route}: {r.id}, new version: {existing_rule_lock["version"]}']
new.extend([f' - {m}' for m in msg if m])
changes.extend(new)
def log_changes(r, route_taken, new_rule_version, *msg):
new = [f' {route_taken}: {r.id}, new version: {new_rule_version}']
new.extend([f' - {m}' for m in msg if m])
changes.extend(new)
for rule in rules:
if rule.contents.metadata.maturity == "production" or rule.id in newly_deprecated:
# assume that older stacks are always locked first
min_stack = _convert_lock_version(rule.contents.metadata.min_stack_version)
current_rule_lock = rule.contents.lock_info(bump=not exclude_version_update)
existing_rule_lock: dict = lock_file_contents.setdefault(rule.id, {})
original_hash = existing_rule_lock.get('sha256')
lock_from_rule = rule.contents.lock_info(bump=not exclude_version_update)
lock_from_file: dict = lock_file_contents.setdefault(rule.id, {})
# prevent rule type changes for already locked and released rules (#1854)
if existing_rule_lock:
name = current_rule_lock['rule_name']
existing_type = existing_rule_lock['type']
current_type = current_rule_lock['type']
if lock_from_file:
name = lock_from_rule['rule_name']
existing_type = lock_from_file['type']
current_type = lock_from_rule['type']
if existing_type != current_type:
err_msg = f'cannot change "type" in locked rule: {name} from {existing_type} to {current_type}'
raise ValueError(err_msg)
@@ -224,41 +222,41 @@ class VersionLock:
# 2) on the latest, after a breaking change has been locked
# 3) on the latest stack, locking in a breaking change
# 4) on an old stack, after a breaking change has been made
latest_locked_stack_version = _convert_lock_version(existing_rule_lock.get("min_stack_version"))
latest_locked_stack_version = _convert_lock_version(lock_from_file.get("min_stack_version"))
if not existing_rule_lock or min_stack == latest_locked_stack_version:
if not lock_from_file or min_stack == latest_locked_stack_version:
route = 'A'
# 1) no breaking changes ever made or the first time a rule is created
# 2) on the latest, after a breaking change has been locked
existing_rule_lock.update(current_rule_lock)
lock_from_file.update(lock_from_rule)
new_version = lock_from_rule['version']
# add the min_stack_version to the lock if it's explicitly set
log_msg = None
if rule.contents.metadata.min_stack_version is not None:
existing_rule_lock["min_stack_version"] = str(min_stack)
lock_from_file["min_stack_version"] = str(min_stack)
log_msg = f'min_stack_version added: {min_stack}'
add_changes(rule, log_msg)
log_changes(rule, route, new_version, log_msg)
elif min_stack > latest_locked_stack_version:
route = 'B'
# 3) on the latest stack, locking in a breaking change
previous_lock_info = {
"rule_name": existing_rule_lock["rule_name"],
"sha256": existing_rule_lock["sha256"],
"version": existing_rule_lock["version"],
"type": existing_rule_lock["type"]
"rule_name": lock_from_file["rule_name"],
"sha256": lock_from_file["sha256"],
"version": lock_from_file["version"],
"type": lock_from_file["type"]
}
existing_rule_lock.setdefault("previous", {})
lock_from_file.setdefault("previous", {})
# move the current locked info into the previous section
existing_rule_lock["previous"][str(latest_locked_stack_version)] = previous_lock_info
lock_from_file["previous"][str(latest_locked_stack_version)] = previous_lock_info
# overwrite the "latest" part of the lock at the top level
# TODO: would need to preserve space here as well if supporting forked version spacing
existing_rule_lock.update(current_rule_lock, min_stack_version=str(min_stack))
add_changes(
rule,
lock_from_file.update(lock_from_rule, min_stack_version=str(min_stack))
new_version = lock_from_rule['version']
log_changes(
rule, route, new_version,
f'previous {latest_locked_stack_version} saved as version: {previous_lock_info["version"]}',
f'current min_stack updated to {min_stack}'
)
@@ -266,23 +264,31 @@ class VersionLock:
elif min_stack < latest_locked_stack_version:
route = 'C'
# 4) on an old stack, after a breaking change has been made (updated fork)
assert str(min_stack) in existing_rule_lock.get("previous", {}), \
assert str(min_stack) in lock_from_file.get("previous", {}), \
f"Expected {rule.id} @ v{min_stack} in the rule lock"
# TODO: Figure out whether we support locking old versions and if we want to
# "leave room" by skipping versions when breaking changes are made.
# We can still inspect the version lock manually after locks are made,
# since it's a good summary of everything that happens
existing_rule_lock["previous"][str(min_stack)] = current_rule_lock
existing_rule_lock.update(current_rule_lock)
add_changes(rule, f'previous version {min_stack} updated version to {current_rule_lock["version"]}')
# if version bump collides with future bump, fail
# if space, change and log
info_from_rule = (lock_from_rule['sha256'], lock_from_rule['version'])
info_from_file = (lock_from_file["previous"][str(min_stack)]['sha256'],
lock_from_file["previous"][str(min_stack)]['version'])
if info_from_rule != info_from_file:
lock_from_file["previous"][str(min_stack)] = lock_from_rule
new_version = lock_from_rule["version"]
log_changes(rule, route, 'unchanged',
f'previous version {min_stack} updated version to {new_version}')
continue
else:
raise RuntimeError("Unreachable code")
if 'previous' in existing_rule_lock:
if 'previous' in lock_from_file:
current_rule_version = rule.contents.lock_info()['version']
for min_stack_version, versioned_lock in existing_rule_lock['previous'].items():
for min_stack_version, versioned_lock in lock_from_file['previous'].items():
existing_lock_version = versioned_lock['version']
if current_rule_version < existing_lock_version:
raise ValueError(f'{rule.id} - previous {min_stack_version=} {existing_lock_version=} '