From 2753cb4369e53003b9661320d833f8ef65d5862d Mon Sep 17 00:00:00 2001 From: aah20 Date: Mon, 17 Aug 2026 12:01:26 +0300 Subject: [PATCH] feat(content-manager): add non-mutating rules plan Signed-off-by: aah20 --- tools/content_manager/README.md | 16 +++ .../content_manager/__main__.py | 32 +++++ .../content_manager/content_manager/rules.py | 119 ++++++++++++++++++ .../content_manager/test_rules.py | 100 +++++++++++++++ 4 files changed, 267 insertions(+) diff --git a/tools/content_manager/README.md b/tools/content_manager/README.md index 48d2980..dcc84dc 100644 --- a/tools/content_manager/README.md +++ b/tools/content_manager/README.md @@ -313,6 +313,22 @@ Example output from `rules verify-all` command: ... ``` +### Plan rule updates + +The `rules plan` command previews the changes that `rules update` would need to +make without calling rule creation, revision, or deployment update APIs. The +plan is emitted as deterministic JSON and includes a SHA-256 digest so it can be +retained as change-review evidence. + +```shell +python -m content_manager rules plan +python -m content_manager rules plan --output-file rule-plan.json +``` + +The plan lists rule creation, revision creation, and deployment-state changes. +It includes remote revision identifiers for existing rules, but excludes rule +text and Google Cloud resource names. + ### Update rules in Google SecOps The `rules update` command updates detection rules in Google SecOps based diff --git a/tools/content_manager/content_manager/__main__.py b/tools/content_manager/content_manager/__main__.py index 5c8c7c3..196b5e7 100644 --- a/tools/content_manager/content_manager/__main__.py +++ b/tools/content_manager/content_manager/__main__.py @@ -107,6 +107,21 @@ def update(cls): # update the local rule files. RuleOperations.get() + @classmethod + def plan(cls, output_file: pathlib.Path | None = None): + """Plan rule updates without modifying Google SecOps.""" + http_session = initialize_http_session() + plan = Rules.plan_remote_rule_updates(http_session=http_session) + if plan is None: + return + + plan_json = json.dumps(plan, indent=2) + if output_file: + output_file.write_text(f"{plan_json}\n", encoding="utf-8") + LOGGER.info("Wrote rule update plan to %s", output_file) + else: + click.echo(plan_json) + @classmethod def verify(cls, rule_file_path: pathlib.Path): """Verify that a rule is a valid YARA-L rule using Google SecOps' API.""" @@ -535,6 +550,23 @@ def update_rules(): RuleOperations.update() +@rules.command( + "plan", + short_help="Preview rule changes without modifying Google SecOps.", +) +@click.option( + "--output-file", + "-o", + required=False, + type=click.Path(file_okay=True, dir_okay=False, path_type=pathlib.Path), + help="Optional path for the machine-readable JSON plan.", +) +def plan_rules(output_file: pathlib.Path | None): + """Preview rule changes without modifying Google SecOps.""" + LOGGER.info("Planning rule changes without modifying Google SecOps") + RuleOperations.plan(output_file=output_file) + + @rules.command( "verify", short_help="Verify a single rule file is a valid YARA-L rule." ) diff --git a/tools/content_manager/content_manager/rules.py b/tools/content_manager/content_manager/rules.py index a5afb60..79fa5a8 100644 --- a/tools/content_manager/content_manager/rules.py +++ b/tools/content_manager/content_manager/rules.py @@ -571,6 +571,125 @@ def check_rule_settings(cls, rule: Rule): " be enabled or have alerting enabled." ) + @classmethod + def build_rule_update_plan( + cls, local_rules: "Rules", remote_rules: "Rules" + ) -> Mapping[str, Any]: + """Build a deterministic, non-mutating plan for local rule changes.""" + remote_by_name = {rule.name: rule for rule in remote_rules.rules} + remote_by_id = {rule.id: rule for rule in remote_rules.rules if rule.id} + changes = [] + unchanged = 0 + + for local_rule in sorted(local_rules.rules, key=lambda rule: rule.name): + remote_rule = remote_by_name.get(local_rule.name) + matched_by_name = remote_rule is not None + rule_changes = [] + + if remote_rule is None and local_rule.id is None: + rule_changes.append({ + "action": "create", + "rule_name": local_rule.name, + "rule_id": None, + "remote_revision_id": None, + "local_rule_digest": hashlib.sha256( + local_rule.text.encode("utf-8") + ).hexdigest(), + "desired_state": { + "enabled": local_rule.enabled, + "alerting": local_rule.alerting, + "archived": local_rule.archived, + }, + }) + else: + if remote_rule is None: + remote_rule = remote_by_id.get(local_rule.id) + + if not matched_by_name or cls.compare_rule_text( + rule_text_1=local_rule.text, + rule_text_2=remote_rule.text, + ): + rule_changes.append({ + "action": "create_revision", + "rule_name": local_rule.name, + "rule_id": local_rule.id, + "remote_revision_id": ( + remote_rule.revision_id if remote_rule else None + ), + "local_rule_digest": hashlib.sha256( + local_rule.text.encode("utf-8") + ).hexdigest(), + }) + + if remote_rule is not None: + state_pairs = ( + ("enabled", local_rule.enabled, remote_rule.enabled), + ("alerting", local_rule.alerting, remote_rule.alerting), + ("archived", local_rule.archived, remote_rule.archived), + ) + for field, desired, current in state_pairs: + if desired is not None and desired != current: + rule_changes.append({ + "action": f"set_{field}", + "rule_name": local_rule.name, + "rule_id": local_rule.id or remote_rule.id, + "remote_revision_id": remote_rule.revision_id, + "desired_value": desired, + }) + + if rule_changes: + changes.extend(rule_changes) + else: + unchanged += 1 + + action_names = ( + "create", + "create_revision", + "set_enabled", + "set_alerting", + "set_archived", + ) + summary = { + action: sum(change["action"] == action for change in changes) + for action in action_names + } + plan_without_digest = { + "schema_version": "1", + "summary": summary, + "unchanged_rules": unchanged, + "changes": changes, + } + canonical_plan = json.dumps( + plan_without_digest, sort_keys=True, separators=(",", ":") + ) + return { + **plan_without_digest, + "plan_digest": hashlib.sha256( + canonical_plan.encode("utf-8") + ).hexdigest(), + } + + @classmethod + def plan_remote_rule_updates( + cls, + http_session: requests.AuthorizedSession, + rules_dir: pathlib.Path = RULES_DIR, + rule_config_file: pathlib.Path = RULE_CONFIG_FILE, + ) -> Mapping[str, Any] | None: + """Plan local rule changes without calling any mutating API.""" + LOGGER.info("Loading local files from %s", rules_dir) + local_rules = cls.load_rules( + rules_dir=rules_dir, rule_config_file=rule_config_file + ) + if not local_rules.rules: + LOGGER.info("No local rule files found") + return None + + remote_rules = cls.get_remote_rules(http_session=http_session) + return cls.build_rule_update_plan( + local_rules=local_rules, remote_rules=remote_rules + ) + @classmethod def update_remote_rules( cls, diff --git a/tools/content_manager/content_manager/test_rules.py b/tools/content_manager/content_manager/test_rules.py index f7066a4..991cc75 100644 --- a/tools/content_manager/content_manager/test_rules.py +++ b/tools/content_manager/content_manager/test_rules.py @@ -217,6 +217,106 @@ def test_check_for_duplicate_rule_ids(parsed_test_rules): assert "Duplicate rule IDs found" in str(excinfo.value) +def test_build_rule_update_plan_unchanged(parsed_test_rules): + """An unchanged ruleset produces a stable, empty plan.""" + remote_rules = copy.deepcopy(parsed_test_rules) + + plan = Rules.build_rule_update_plan(parsed_test_rules, remote_rules) + reversed_plan = Rules.build_rule_update_plan( + Rules(rules=list(reversed(parsed_test_rules.rules))), remote_rules + ) + + assert plan["changes"] == [] + assert plan["unchanged_rules"] == len(parsed_test_rules.rules) + assert plan["plan_digest"] == reversed_plan["plan_digest"] + + +def test_build_rule_update_plan_revision_and_state(parsed_test_rules): + """Text and deployment differences are represented as separate changes.""" + remote_rules = copy.deepcopy(parsed_test_rules) + remote_rule = remote_rules.rules[0] + remote_rule.text = f"{remote_rule.text}\n// remote-only change" + remote_rule.enabled = not parsed_test_rules.rules[0].enabled + remote_rule.alerting = not parsed_test_rules.rules[0].alerting + + plan = Rules.build_rule_update_plan(parsed_test_rules, remote_rules) + rule_changes = [ + change + for change in plan["changes"] + if change["rule_name"] == parsed_test_rules.rules[0].name + ] + + assert [change["action"] for change in rule_changes] == [ + "create_revision", + "set_enabled", + "set_alerting", + ] + assert rule_changes[0]["remote_revision_id"] == remote_rule.revision_id + assert "text" not in json.dumps(plan) + assert "projects/" not in json.dumps(plan) + + +def test_build_rule_update_plan_create_and_rename(parsed_test_rules): + """New and renamed rules retain state and revision evidence.""" + local_rules = copy.deepcopy(parsed_test_rules) + remote_rules = copy.deepcopy(parsed_test_rules) + new_rule = local_rules.rules[0].model_copy(update={ + "name": "new_rule", + "id": None, + "resource_name": None, + "revision_id": None, + }) + renamed_rule = local_rules.rules[1].model_copy( + update={"name": "renamed_rule"} + ) + local_rules.rules = [new_rule, renamed_rule] + + plan = Rules.build_rule_update_plan(local_rules, remote_rules) + + assert plan["changes"][0]["action"] == "create" + assert plan["changes"][0]["desired_state"] == { + "enabled": new_rule.enabled, + "alerting": new_rule.alerting, + "archived": new_rule.archived, + } + assert plan["changes"][1]["action"] == "create_revision" + assert ( + plan["changes"][1]["remote_revision_id"] + == remote_rules.rules[1].revision_id + ) + + +def test_plan_remote_rule_updates_never_mutates(monkeypatch, parsed_test_rules): + """Planning may read remote state but must never call a mutating API.""" + remote_rules = copy.deepcopy(parsed_test_rules) + mutation_calls = [] + + monkeypatch.setattr( + Rules, + "load_rules", + classmethod(lambda cls, rules_dir, rule_config_file: parsed_test_rules), + ) + monkeypatch.setattr( + Rules, + "get_remote_rules", + classmethod(lambda cls, http_session: remote_rules), + ) + + def record_mutation(*args, **kwargs): + mutation_calls.append((args, kwargs)) + + monkeypatch.setattr("content_manager.rules.create_rule", record_mutation) + monkeypatch.setattr("content_manager.rules.update_rule", record_mutation) + monkeypatch.setattr( + "content_manager.rules.update_rule_deployment", record_mutation + ) + + plan = Rules.plan_remote_rule_updates(http_session=object()) + + assert plan is not None + assert mutation_calls == [] + + def test_extract_rule_name(parsed_test_rules: Rules): """Tests for rules.Rules.extract_rule_name.""" rule = copy.deepcopy(parsed_test_rules.rules[0])