v1alpha sample to bulk update alerts

PiperOrigin-RevId: 651751014

detect/v1alpha/bulk_update_alerts.py

@@ -0,0 +1,215 @@
+#!/usr/bin/env python3
+
+# Copyright 2024 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+r"""Executable and reusable sample for bulk updating alerts.
+
+Usage:
+  python -m alerts.v1alpha.bulk_update_alerts \
+    --project_id=<PROJECT_ID>   \
+    --project_instance=<PROJECT_INSTANCE> \
+    --alert_ids_file=<PATH_TO_FILE> \
+    --confidence_score=<CONFIDENCE_SCORE> \
+    --priority=<PRIORITY> \
+    --reason=<REASON> \
+    --reputation=<REPUTATION> \
+    --priority=<PRIORITY> \
+    --status=<STATUS> \
+    --verdict=<VERDICT> \
+    --risk_score=<RISK_SCORE> \
+    --disregarded=<DISREGARDED> \
+    --severity=<SEVERITY> \
+    --comment=<COMMENT> \
+    --root_cause=<ROOT_CAUSE> \
+    --severity_display=<SEVERITY_DISPLAY>
+
+# pylint: disable=line-too-long
+API reference:
+  https://cloud.google.com/chronicle/docs/reference/rest/v1alpha/projects.locations.instances.legacy/legacyUpdateAlert
+  https://cloud.google.com/chronicle/docs/reference/rest/v1alpha/Noun#Priority
+  https://cloud.google.com/chronicle/docs/reference/rest/v1alpha/Noun#Reason
+  https://cloud.google.com/chronicle/docs/reference/rest/v1alpha/Noun#Reputation
+  https://cloud.google.com/chronicle/docs/reference/rest/v1alpha/Noun#Priority
+  https://cloud.google.com/chronicle/docs/reference/rest/v1alpha/Noun#Status
+  https://cloud.google.com/chronicle/docs/reference/rest/v1alpha/Noun#Verdict
+"""
+# pylint: enable=line-too-long
+
+import argparse
+import json
+
+from common import chronicle_auth
+from common import project_id
+from common import project_instance
+from common import regions
+
+from . import update_alert
+
+
+CHRONICLE_API_BASE_URL = "https://chronicle.googleapis.com"
+SCOPES = [
+    "https://www.googleapis.com/auth/cloud-platform",
+]
+DEFAULT_FEEDBACK = {
+    "reason": "REASON_MAINTENANCE",
+    "reputation": "REPUTATION_UNSPECIFIED",
+    "status": "CLOSED",
+    "verdict": "VERDICT_UNSPECIFIED",
+    "comment": "automated cleanup",
+    "rootCause": "Other",
+}
+
+
+if __name__ == "__main__":
+  parser = argparse.ArgumentParser()
+  chronicle_auth.add_argument_credentials_file(parser)
+  project_instance.add_argument_project_instance(parser)
+  project_id.add_argument_project_id(parser)
+  regions.add_argument_region(parser)
+  parser.add_argument(
+      "--alert_ids_file", type=str, required=True,
+      help="file with one alert ID per line"
+  )
+  parser.add_argument(
+      "--confidence_score",
+      type=int,
+      required=False,
+      default=None,
+      help="confidence score [1-100] of the finding",
+  )
+  parser.add_argument(
+      "--priority",
+      choices=update_alert.PRIORITY_ENUM,
+      required=False,
+      default=None,
+      help="alert priority.",
+  )
+  parser.add_argument(
+      "--reason",
+      choices=update_alert.REASON_ENUM,
+      required=False,
+      default=DEFAULT_FEEDBACK["reason"],
+      help="reason for closing an Alert",
+  )
+  parser.add_argument(
+      "--reputation",
+      choices=update_alert.REPUTATION_ENUM,
+      required=False,
+      default=DEFAULT_FEEDBACK["reputation"],
+      help="A categorization of the finding as useful or not useful",
+  )
+  parser.add_argument(
+      "--status",
+      choices=update_alert.STATUS_ENUM,
+      required=False,
+      default=DEFAULT_FEEDBACK["status"],
+      help="alert status",
+  )
+  parser.add_argument(
+      "--verdict",
+      choices=update_alert.VERDICT_ENUM,
+      required=False,
+      default=DEFAULT_FEEDBACK["verdict"],
+      help="a verdict on whether the finding reflects a security incident",
+  )
+  parser.add_argument(
+      "--risk_score",
+      type=int,
+      required=False,
+      default=None,
+      help="risk score [0-100] of the finding",
+  )
+  parser.add_argument(
+      "--disregarded",
+      type=bool,
+      required=False,
+      default=None,
+      help="Analyst disregard (or un-disregard) the event",
+  )
+  parser.add_argument(
+      "--severity",
+      type=int,
+      required=False,
+      default=None,
+      help="severity score [0-100] of the finding",
+  )
+  parser.add_argument(
+      "--comment",
+      type=str,
+      required=False,
+      default=DEFAULT_FEEDBACK["comment"],
+      help="Analyst comment.",
+  )
+  parser.add_argument(
+      "--root_cause",
+      type=str,
+      required=False,
+      default=DEFAULT_FEEDBACK["rootCause"],
+      help="Alert root cause.",
+  )
+  args = parser.parse_args()
+  # Check if at least one of the specific arguments is provided
+  if not any(
+      [
+          args.comment or args.comment == "",  # pylint: disable=g-explicit-bool-comparison
+          args.disregarded,
+          args.priority,
+          args.reason,
+          args.reputation,
+          args.risk_score or args.risk_score == 0,
+          args.root_cause or args.root_cause == "",  # pylint: disable=g-explicit-bool-comparison
+          args.severity or args.severity == 0,
+          args.status,
+          args.verdict,
+      ]
+  ):
+    parser.error("At least one of the arguments "
+                 "--comment, "
+                 "--disregarded, "
+                 "--priority, "
+                 "--reason, "
+                 "--reputation, "
+                 "--risk_score, "
+                 "--root_cause, "
+                 "--severity, "
+                 "--status, "
+                 "or --verdict "
+                 "is required.")
+
+  auth_session = chronicle_auth.initialize_http_session(
+      args.credentials_file,
+      SCOPES,
+  )
+  with open(args.alert_ids_file) as fh:
+    for alert_id in fh:
+      a_list = update_alert.update_alert(
+          auth_session,
+          args.project_id,
+          args.project_instance,
+          args.region,
+          alert_id.strip(),
+          args.confidence_score,
+          args.reason,
+          args.reputation,
+          args.priority,
+          args.status,
+          args.verdict,
+          args.risk_score,
+          args.disregarded,
+          args.severity,
+          args.comment,
+          args.root_cause,
+      )
+      print(json.dumps(a_list, indent=2))