# Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one # or more contributor license agreements. Licensed under the Elastic License # 2.0; you may not use this file except in compliance with the Elastic License # 2.0. """Helper utilities to manage the version lock.""" from copy import deepcopy from typing import List, Optional import click from .rule_loader import RuleCollection from .semver import Version from .utils import dict_hash, load_etc_dump, save_etc_dump, cached ETC_VERSION_LOCK_FILE = "version.lock.json" ETC_DEPRECATED_RULES_FILE = "deprecated_rules.json" MIN_LOCK_VERSION_DEFAULT = Version("7.13.0") def _convert_lock_version(stack_version: Optional[str]) -> Version: """Convert an optional stack version to the minimum for the lock.""" if stack_version is None: return MIN_LOCK_VERSION_DEFAULT return max(Version(stack_version), MIN_LOCK_VERSION_DEFAULT) @cached def load_versions(): """Load the versions file.""" return load_etc_dump(ETC_VERSION_LOCK_FILE) class VersionLock: """Version handling for rule files and collections.""" def __init__(self, version_lock_file: Optional[str] = None, deprecated_lock_file: Optional[str] = None, version_lock: Optional[dict] = None, deprecated_lock: Optional[dict] = None, name: Optional[str] = None): assert (version_lock_file or version_lock), 'Must provide version lock file or contents' assert (deprecated_lock_file or deprecated_lock), 'Must provide deprecated lock file or contents' self.name = name self.version_lock_file = version_lock_file self.deprecated_lock_file = deprecated_lock_file self.version_lock = load_etc_dump(self.version_lock_file) if version_lock_file else version_lock self.deprecated_lock = load_etc_dump(self.deprecated_lock_file) if deprecated_lock_file else deprecated_lock def save_versions(self, current_versions: dict): save_etc_dump(current_versions, self.version_lock_file) print('Updated version.lock.json file') def get_locked_version(self, rule_id: str, min_stack_version: Optional[str] = None) -> Optional[int]: if rule_id in self.version_lock: latest_version_info = self.version_lock[rule_id] stack_version_info = latest_version_info.get("previous", {}).get(min_stack_version, latest_version_info) return stack_version_info['version'] def get_locked_hash(self, rule_id: str, min_stack_version: Optional[str] = None) -> Optional[str]: """Get the version info matching the min_stack_version if present.""" if rule_id in self.version_lock: latest_version_info = self.version_lock[rule_id] stack_version_info = latest_version_info.get("previous", {}).get(min_stack_version, latest_version_info) existing_sha256: str = stack_version_info['sha256'] return existing_sha256 def manage_versions(self, rules: RuleCollection, exclude_version_update=False, save_changes=False, verbose=True) -> (List[str], List[str], List[str]): """Update the contents of the version.lock file and optionally save changes.""" from .packaging import current_stack_version lock_file_contents = deepcopy(self.version_lock) version_lock_hash = dict_hash(lock_file_contents) current_deprecated_lock = deepcopy(self.deprecated_lock) verbose_echo = click.echo if verbose else (lambda x: None) already_deprecated = set(current_deprecated_lock) deprecated_rules = set(rules.deprecated.id_map) new_rules = set(rule.id for rule in rules if rule.contents.latest_version is None) - deprecated_rules changed_rules = set(rule.id for rule in rules if rule.contents.is_dirty) - deprecated_rules # manage deprecated rules newly_deprecated = deprecated_rules - already_deprecated if not (new_rules or changed_rules or newly_deprecated): 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) 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') # 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 existing_type != current_type: err_msg = f'cannot change "type" in locked rule: {name} from {existing_type} to {current_type}' raise ValueError(err_msg) # scenarios to handle, assuming older stacks are always locked first: # 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 # 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")) if not existing_rule_lock 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) # 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) log_msg = f'min_stack_version added: {min_stack}' add_changes(rule, 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"], } existing_rule_lock.setdefault("previous", {}) # move the current locked info into the previous section existing_rule_lock["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, f'previous {latest_locked_stack_version} saved as version: {previous_lock_info["version"]}', f'current min_stack updated to {min_stack}' ) 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", {}), \ 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"]}') continue else: raise RuntimeError("Unreachable code") if 'previous' in existing_rule_lock: current_rule_version = rule.contents.lock_info()['version'] for min_stack_version, versioned_lock in existing_rule_lock['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=} ' f'has a higher version than {current_rule_version=}') for rule in rules.deprecated: if rule.id in newly_deprecated: current_deprecated_lock[rule.id] = { "rule_name": rule.name, "stack_version": current_stack_version(), "deprecation_date": rule.contents.metadata['deprecation_date'] } if save_changes or verbose: click.echo(f' - {len(changed_rules)} changed rules') click.echo(f' - {len(new_rules)} new rules') click.echo(f' - {len(newly_deprecated)} newly deprecated rules') if save_changes: click.echo('Detailed changes: \n' + '\n'.join(changes)) if not save_changes: verbose_echo( 'run `build-release --update-version-lock` to update version.lock.json and deprecated_rules.json') return list(changed_rules), list(new_rules), list(newly_deprecated) new_hash = dict_hash(lock_file_contents) if version_lock_hash != new_hash: save_etc_dump(lock_file_contents, ETC_VERSION_LOCK_FILE) click.echo('Updated version.lock.json file') # reset local version lock self.version_lock = lock_file_contents if newly_deprecated: save_etc_dump(current_deprecated_lock, ETC_DEPRECATED_RULES_FILE) click.echo('Updated deprecated_rules.json file') # reset local version lock self.deprecated_lock = current_deprecated_lock return changed_rules, list(new_rules), newly_deprecated default_version_lock = VersionLock(ETC_VERSION_LOCK_FILE, ETC_DEPRECATED_RULES_FILE, name='default')