update readme and make minor improvements

PiperOrigin-RevId: 600839301

tools/rule_manager/README.md

@@ -23,9 +23,12 @@ repo.
 
 ## Setup
 
+Use Python 3.10 or above.
+
 ```console
 # Create and activate a Python virtual environment after cloning this directory into a location of your choosing
-$ python3.10 -m virtualenv venv
+$ pip3 install virtualenv
+$ python3 -m virtualenv venv
 $ source venv/bin/activate
 
 # Install the project's dependencies
@@ -58,12 +61,12 @@ refrain from including any sensitive information such as service account keys or
 * Used to configure the [logging level](https://docs.python.org/3/library/logging.html#levels) for this project. The 
   recommendation is to set this to `INFO` or `DEBUG` for more verbose logging.
 
-### `CHRONICLE_BASE_URL`
+#### `CHRONICLE_BASE_URL`
 
 * Set the `CHRONICLE_BASE_URL` variable to your regional service endpoint for the Chronicle API.
 * For example, the base URL for the US regional service endpoint is https://us-chronicle.googleapis.com/v1alpha
 
-### `CHRONICLE_INSTANCE`
+#### `CHRONICLE_INSTANCE`
 
 * Set the `CHRONICLE_INSTANCE` variable as follows: `projects/{google-cloud-project-id}/locations/{chronicle-instance-location}/instances/{chronicle-instance-id}`
   * Replace the `{google-cloud-project-id}` placeholder with your Google Cloud project ID that is linked to your 
@@ -73,13 +76,13 @@ refrain from including any sensitive information such as service account keys or
   * Replace the `chronicle-instance-id` placeholder with the `Customer ID` for your Chronicle instance. You can find 
     this under `Settings` - `SIEM Settings` - `Profile` in Chronicle's UI.
 
-### `AUTHORIZATION_SCOPES`
+#### `AUTHORIZATION_SCOPES`
 
 * Set the `AUTHORIZATION_SCOPES` variable to `AUTHORIZATION_SCOPES={"CHRONICLE_API":["https://www.googleapis.com/auth/cloud-platform"]}`
 * Refer to the [Authentication methods at Google](https://cloud.google.com/docs/authentication/) documentation for 
   information on OAuth 2.0 scopes.
 
-### `CHRONICLE_API_CREDENTIALS`
+#### `CHRONICLE_API_CREDENTIALS`
 
 * For the purposes of authenticating to and managing detection rules via Chronicle's API, you can create a [service account](https://cloud.google.com/iam/docs/service-account-overview)
   in the Google Cloud project that's linked to your Chronicle instance.
@@ -108,8 +111,9 @@ refrain from including any sensitive information such as service account keys or
   value for the `CHRONICLE_API_CREDENTIALS` variable. Enter the variable's value in JSON format, on a single line as 
   shown in above example `.env` file).
 
+### Executing the CLI
+
 ```console
-# Verify that the CLI executes successfully
 (venv) $ python -m rule_cli -h
 16-Jan-24 16:14:00 MST | INFO | <module> | Rule CLI started
 usage: __main__.py [-h] [--pull-latest-rules] [--update-remote-rules] [--verify-rules] {verify-rule} ...
@@ -128,7 +132,7 @@ subcommands:
     verify-rule         Verify that a rule is a valid YARA-L 2.0 rule.
 ```
 
-To run the tests.
+### Running the tests
 
 ```console
 (venv) $ pip install -r requirements_dev.txt
@@ -146,7 +150,7 @@ The pull latest rules command retrieves the latest version of all rules from Chr
 files in the `rules` directory.
 
 The rule state is written to the `rule_config.yaml` file. The rule state contains metadata about the state of each rule
-such as whether it is live enabled/disabled, the rule ID, the rule version ID, etc.
+such as whether it is enabled/disabled/archived, the rule ID, the rule's revision ID, etc.
 
 Example output from pull latest rules command:
 
@@ -161,7 +165,7 @@ Example output from pull latest rules command:
 
 ## Verify rule(s)
 
-The `--verify-rule` and `--verify-rules` commands use Chronicle's API to verify that YARA-L 2.0 rules are valid without
+The `verify-rule` and `--verify-rules` commands use Chronicle's API to verify that YARA-L 2.0 rules are valid without
 creating a new rule or evaluating it over data.
 
 Example output from verify rule command:
@@ -173,8 +177,13 @@ Example output from verify rule command:
 16-Jan-24 16:16:11 MST | INFO | verify_rule_text | Rule verified successfully (rules/dns_query_to_recently_created_domain.yaral). Response: {'success': True}
 
 python -m rule_cli --verify-rules
-16-Jan-24 16:17:08 MST | INFO | <module> | Rule CLI started
-16-Jan-24 16:17:08 MST | INFO | <module> | Attempting to verify all local rules
+19-Jan-24 11:13:06 MST | INFO | <module> | Rule CLI started
+19-Jan-24 11:13:06 MST | INFO | <module> | Attempting to verify all local rules
+19-Jan-24 11:13:07 MST | INFO | verify_rules | Rule verification succeeded for rule (/Users/x/Documents/projects/detection-engineering/rules/google_workspace_multiple_files_sent_as_email_attachment_from_google_drive.yaral). Response: {'success': True}
+...
+19-Jan-24 11:13:10 MST | INFO | verify_rules | Rule verification succeeded for 36 rules
+19-Jan-24 11:17:32 MST | ERROR | verify_rules | Rule verification failed for 2 rules
+19-Jan-24 11:13:10 MST | ERROR | verify_rules | Rule verification failed for rule (/Users/x/Documents/projects/detection-engineering/rules/okta_new_api_token_created.yaral). Response: {...}
 ...
 ```
 

tools/rule_manager/chronicle_api/rules/list_rules.py

@@ -34,14 +34,15 @@ def list_rules(
 
   Args:
     http_session: Authorized session for HTTP requests.
-    page_size: Maximum number of rules to return. Must be non-negative, and is
-      capped at a server-side limit of 1000. Optional - a server-side default
-      of 100 is used if the size is 0 or a None value.
-    page_token: Page token from a previous ListRules call used for pagination.
-      Optional - the first page is retrieved if the token is the empty string
-      or a None value.
-    view: The scope of fields to populate for the Rule being returned.
-      Optional. Reference:
+    page_size (optional): Maximum number of rules to return.
+      Must be non-negative, and is capped at a server-side limit of 1000.
+      A server-side default of 100 is used if the size is 0 or a None value.
+    page_token (optional): Page token from a previous ListRules call used for
+      pagination.
+      The first page is retrieved if the token is the empty string or a None
+      value.
+    view (optional): The scope of fields to populate for the Rule being
+      returned. Reference:
       https://cloud.google.com/chronicle/docs/reference/rest/v1alpha/RuleView
 
   Returns:

tools/rule_manager/rule_cli/__main__.py

@@ -22,6 +22,7 @@
 import os
 import pathlib
 import sys
+import time
 
 from chronicle_api import chronicle_auth
 from chronicle_api.rules.verify_rule import verify_rule
@@ -58,7 +59,7 @@ def pull_latest_rules():
 
   remote_rules = Rules.get_remote_rules(http_session=http_session)
 
-  if not remote_rules.rules:  # pylint: disable="g-explicit-length-test"
+  if len(remote_rules.rules) == 0:  # pylint: disable="g-explicit-length-test"
     return
 
   # Delete existing local rule files before writing a fresh copy of all rules
@@ -117,22 +118,46 @@ def verify_rules():
   """Verify that all detection rules are valid YARA-L 2.0 rules using Chronicle's API."""
   http_session = initialize_http_session()
 
+  # Maintain lists of successful and failed YARA-L 2.0 verification responses.
+  verify_rule_successes = []
+  verify_rule_errors = []
+
   for rule_file in list(RULES_DIR.glob("*.yaral")):
     with open(rule_file, "r", encoding="utf-8") as f:
       rule_text = f.read()
 
     response = verify_rule(http_session=http_session, rule_text=rule_text)
+    time.sleep(0.6)  # Sleep to avoid exceeding API rate limit
 
     if response.get("success") is True:
-      LOGGER.debug(
-          "Rule verified successfully (%s). Response: %s", rule_file, response
+      LOGGER.info(
+          "Rule verification succeeded for rule (%s). Response: %s",
+          rule_file,
+          response,
       )
+      verify_rule_successes.append(rule_file)
 
     else:
-      raise RuleVerificationError(
-          f"Rule verification error ({rule_file}). Response:"
-          f" {json.dumps(response, indent=4)}"
+      verify_rule_errors.append({"rule": rule_file, "response": response})
+
+  LOGGER.info(
+      "Rule verification succeeded for %s rules", len(verify_rule_successes)
+  )
+
+  if verify_rule_errors:
+    LOGGER.error(
+        "Rule verification failed for %s rules", len(verify_rule_errors)
+    )
+    # Log each rule verification error before raising an exception
+    for error in verify_rule_errors:
+      LOGGER.error(
+          "Rule verification failed for rule (%s). Response: %s",
+          error["rule"],
+          json.dumps(error["response"], indent=4),
       )
+    raise RuleVerificationError(
+        f"Rule verification failed for {len(verify_rule_errors)} rules"
+    )
 
 
 if __name__ == "__main__":

tools/rule_manager/rule_cli/rules.py

@@ -22,6 +22,7 @@
 import logging
 import pathlib
 import re
+import time
 from typing import Any, List, Mapping, Optional, Sequence, Tuple
 
 from chronicle_api.rules.create_rule import create_rule
@@ -257,6 +258,7 @@ def get_remote_rules(
           page_token=next_page_token,
           view="FULL",
       )
+      time.sleep(0.6)  # Sleep to avoid exceeding API rate limit
 
       if retrieved_rules is not None:
         LOGGER.info("Retrieved %s rules", len(retrieved_rules))
@@ -282,6 +284,7 @@ def get_remote_rules(
       rule["deployment_state"] = get_rule_deployment(
           http_session=http_session, resource_name=rule["name"]
       )
+      time.sleep(0.6)  # Sleep to avoid exceeding API rate limit
 
     parsed_rules = Rules.parse_rules(rules=raw_rules)
 
@@ -343,11 +346,11 @@ def extract_rule_name(
   ) -> str:
     """Extract the rule name from the YARA-L rule."""
     rule_name_match = re.search(
-        pattern=r"rule ([A-Za-z0-9_]+)[\r\n\s]?{", string=rule_text
+        pattern=r"rule(\s+)([A-Za-z0-9_]+)[\r\n\s]*", string=rule_text
     )
 
     if rule_name_match:
-      rule_name = rule_name_match.group(1)
+      rule_name = rule_name_match.group(2)
     else:
       raise RuleError(
           f"Unable to extract rule name from YARA-L rule in {rule_file_path}"
@@ -412,7 +415,8 @@ def check_rule_settings(cls, rule: Rule):
           f"{rule.name} - alerting (true/false) option is missing."
       )
 
-    # Check that enabled or alerting are not set to True if archived is True.
+    # Check that enabled or alerting are not set to True if archived is set to
+    # True.
     if rule.archived is True and (
         rule.enabled is True or rule.alerting is True
     ):
@@ -501,6 +505,7 @@ def update_remote_rules(
               update_mask=["text"],
               updates={"text": local_rule.text},
           )
+          time.sleep(0.6)  # Sleep to avoid exceeding API rate limit
           update_summary["new_version_created"].append((rule_id, rule_name))
         LOGGER.debug(
             "Rule %s (%s) - No changes found in rule text", rule_name, rule_id
@@ -521,9 +526,11 @@ def update_remote_rules(
           new_rule = create_rule(
               http_session=http_session, rule_text=local_rule.text
           )
+          time.sleep(0.6)  # Sleep to avoid exceeding API rate limit
           new_rule["deployment_state"] = get_rule_deployment(
               http_session=http_session, resource_name=new_rule["name"]
           )
+          time.sleep(0.6)  # Sleep to avoid exceeding API rate limit
           remote_rule = Rules.parse_rule(new_rule)
           LOGGER.info(
               "Created new rule %s (%s)", remote_rule.name, remote_rule.id
@@ -549,6 +556,7 @@ def update_remote_rules(
               update_mask=["text"],
               updates={"text": local_rule.text},
           )
+          time.sleep(0.6)  # Sleep to avoid exceeding API rate limit
           remote_rule = Rules.parse_rule(new_rule_version)
           update_summary["new_version_created"].append((rule_id, rule_name))
 
@@ -599,6 +607,7 @@ def update_remote_rule_state(
           update_mask=["archived"],
           updates={"archived": False},
       )
+      time.sleep(0.6)  # Sleep to avoid exceeding API rate limit
       rule_updates["unarchived"] = True
 
     LOGGER.debug(
@@ -613,6 +622,7 @@ def update_remote_rule_state(
           update_mask=["enabled"],
           updates={"enabled": True},
       )
+      time.sleep(0.6)  # Sleep to avoid exceeding API rate limit
       rule_updates["enabled"] = True
 
     # Disable the rule if required.
@@ -624,6 +634,7 @@ def update_remote_rule_state(
           update_mask=["enabled"],
           updates={"enabled": False},
       )
+      time.sleep(0.6)  # Sleep to avoid exceeding API rate limit
       rule_updates["disabled"] = True
 
     LOGGER.debug(
@@ -638,6 +649,7 @@ def update_remote_rule_state(
           update_mask=["alerting"],
           updates={"alerting": True},
       )
+      time.sleep(0.6)  # Sleep to avoid exceeding API rate limit
       rule_updates["alerting_enabled"] = True
 
     # Disable alerting for the rule if required.
@@ -649,6 +661,7 @@ def update_remote_rule_state(
           update_mask=["alerting"],
           updates={"alerting": False},
       )
+      time.sleep(0.6)  # Sleep to avoid exceeding API rate limit
       rule_updates["alerting_disabled"] = True
 
     LOGGER.debug("%s - Checking if the rule should be archived", log_msg_prefix)
@@ -661,6 +674,7 @@ def update_remote_rule_state(
           update_mask=["archived"],
           updates={"archived": True},
       )
+      time.sleep(0.6)  # Sleep to avoid exceeding API rate limit
       rule_updates["archived"] = True
 
     return rule_updates