feat(content-manager): add non-mutating rules plan

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

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."
 )

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,

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])