| # Licensed to the Apache Software Foundation (ASF) under one or more | |
| # contributor license agreements. See the NOTICE file distributed with | |
| # this work for additional information regarding copyright ownership. | |
| # The ASF licenses this file to You 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. | |
| import json | |
| import ssl | |
| import yaml | |
| import logging | |
| import smtplib | |
| import os | |
| from dataclasses import dataclass | |
| from datetime import datetime, timedelta, timezone | |
| from google.cloud import logging_v2 | |
| from google.cloud import storage | |
| from typing import List, Dict, Any | |
| import argparse | |
| REPORT_SUBJECT = "Weekly IAM Security Events Report" | |
| REPORT_BODY_TEMPLATE = """ | |
| Hello Team, | |
| Please find below the summary of IAM security events for the past week: | |
| {event_summary} | |
| Best Regards, | |
| Automated GitHub Action | |
| """ | |
| @dataclass | |
| class SinkCls: | |
| name: str | |
| description: str | |
| filter_methods: List[str] | |
| excluded_principals: List[str] | |
| class LogAnalyzer(): | |
| def __init__(self, project_id: str, gcp_bucket: str, logger: logging.Logger, sinks: List[SinkCls]): | |
| self.project_id = project_id | |
| self.bucket = gcp_bucket | |
| self.logger = logger | |
| self.sinks = sinks | |
| def _construct_filter(self, sink: SinkCls) -> str: | |
| """ | |
| Constructs a filter string for a given sink. | |
| Args: | |
| sink (Sink): The sink object containing filter information. | |
| Returns: | |
| str: The constructed filter string. | |
| """ | |
| method_filters = [] | |
| for method in sink.filter_methods: | |
| method_filters.append(f'protoPayload.methodName="{method}"') | |
| exclusion_filters = [] | |
| for principal in sink.excluded_principals: | |
| exclusion_filters.append(f'protoPayload.authenticationInfo.principalEmail != "{principal}"') | |
| if method_filters and exclusion_filters: | |
| filter_ = f"({' OR '.join(method_filters)}) AND ({' AND '.join(exclusion_filters)})" | |
| elif method_filters: | |
| filter_ = f"({' OR '.join(method_filters)})" | |
| elif exclusion_filters: | |
| filter_ = f"({' AND '.join(exclusion_filters)})" | |
| else: | |
| filter_ = "" | |
| return filter_ | |
| def _create_log_sink(self, sink: SinkCls) -> None: | |
| """ | |
| Creates a log sink in GCP if it doesn't already exist. | |
| If it already exists, it updates the sink with the new filter in case the filter has changed. | |
| Args: | |
| sink (Sink): The sink object to create. | |
| """ | |
| logging_client = logging_v2.Client(project=self.project_id) | |
| filter_ = self._construct_filter(sink) | |
| destination = "storage.googleapis.com/{bucket}".format(bucket=self.bucket) | |
| sink_client = logging_client.sink(sink.name, filter_=filter_, destination=destination) | |
| if sink_client.exists(): | |
| self.logger.debug(f"Sink {sink.name} already exists.") | |
| sink_client.reload() | |
| if sink_client.filter_ != filter_: | |
| sink_client.filter_ = filter_ | |
| sink_client.update() | |
| self.logger.info(f"Updated sink {sink.name}'s filter.") | |
| else: | |
| sink_client.create() | |
| self.logger.info(f"Created sink {sink.name}.") | |
| # Reload the sink to get the writer_identity, this may take a few moments | |
| sink_client.reload() | |
| self._grant_bucket_permissions(sink_client) | |
| logging_client.close() | |
| def _grant_bucket_permissions(self, sink: logging_v2.Sink) -> None: | |
| """ | |
| Grants a log sink's writer identity permissions to write to the bucket. | |
| """ | |
| logging_client = logging_v2.Client(project=self.project_id) | |
| storage_client = storage.Client(project=self.project_id) | |
| sink.reload() | |
| writer_identity = sink.writer_identity | |
| if not writer_identity: | |
| self.logger.warning(f"Could not retrieve writer identity for sink {sink.name}. " | |
| f"Manual permission granting might be required.") | |
| return | |
| bucket = storage_client.get_bucket(self.bucket) | |
| policy = bucket.get_iam_policy(requested_policy_version=3) | |
| iam_role = "roles/storage.objectCreator" | |
| # Workaround for projects where the writer_identity is not a valid service account. | |
| if writer_identity == "serviceAccount:cloud-logs@system.gserviceaccount.com": | |
| member = "group:cloud-logs@google.com" | |
| else: | |
| member = f"serviceAccount:{writer_identity}" | |
| # Check if the policy is already configured | |
| if any(member in b.get("members", []) and b.get("role") == iam_role for b in policy.bindings): | |
| self.logger.debug(f"Sink {sink.name} already has the necessary permissions.") | |
| return | |
| policy.bindings.append({ | |
| "role": iam_role, | |
| "members": {member} | |
| }) | |
| bucket.set_iam_policy(policy) | |
| self.logger.info(f"Granted {iam_role} to {member} on bucket {self.bucket} for sink {sink.name}.") | |
| def initialize_sinks(self) -> None: | |
| for sink in self.sinks: | |
| self._create_log_sink(sink) | |
| self.logger.info(f"Initialized sink: {sink.name}") | |
| def get_event_logs(self, days: int = 7) -> List[Dict[str, Any]]: | |
| """ | |
| Reads and retrieves log events from the specified time range from the GCP Cloud Storage bucket. | |
| Args: | |
| days (int): The number of days to look back for log analysis. | |
| Returns: | |
| List[Dict[str, Any]]: A list of log entries that match the specified time range. | |
| """ | |
| found_events = [] | |
| storage_client = storage.Client(project=self.project_id) | |
| now = datetime.now(timezone.utc) | |
| end_time = now.replace(minute=0, second=0, microsecond=0) - timedelta(minutes=30) | |
| start_time = end_time - timedelta(days=days) | |
| blobs = storage_client.list_blobs(self.bucket) | |
| for blob in blobs: | |
| if not (start_time <= blob.time_created < end_time): | |
| continue | |
| self.logger.debug(f"Processing blob: {blob.name}") | |
| content = blob.download_as_string().decode("utf-8") | |
| for num, line in enumerate(content.splitlines(), 1): | |
| try: | |
| log_entry = json.loads(line) | |
| payload = log_entry.get("protoPayload") | |
| if not payload: | |
| self.logger.warning(f"Skipping log in blob {blob.name}, line {num}: no protoPayload found.") | |
| continue | |
| event_details = { | |
| "timestamp": log_entry.get("timestamp", "N/A"), | |
| "principal": payload.get("authenticationInfo", {}).get("principalEmail", "N/A"), | |
| "method": payload.get("methodName", "N/A"), | |
| "resource": payload.get("resourceName", "N/A"), | |
| "project_id": log_entry.get("resource", {}).get("labels", {}).get("project_id", "N/A"), | |
| "file_name": blob.name | |
| } | |
| found_events.append(event_details) | |
| except json.JSONDecodeError: | |
| self.logger.warning(f"Skipping invalid JSON log in blob {blob.name}, line {num}.") | |
| continue | |
| storage_client.close() | |
| return found_events | |
| def create_weekly_email_report(self, dry_run: bool = False) -> None: | |
| """ | |
| Creates an email report based on the events found this week. | |
| If `dry_run` is True, it will print the report to the console instead of sending it. | |
| """ | |
| events = self.get_event_logs(days=7) | |
| if not events: | |
| self.logger.info("No events found for the weekly report.") | |
| return | |
| events.sort(key=lambda x: x['timestamp'], reverse=True) | |
| event_summary = "\n".join( | |
| f"Timestamp: {event['timestamp']}, Principal: {event['principal']}, Method: {event['method']}, Resource: {event['resource']}, Project ID: {event['project_id']}, File: {event['file_name']}" | |
| for event in events | |
| ) | |
| report_subject = REPORT_SUBJECT | |
| report_body = REPORT_BODY_TEMPLATE.format(event_summary=event_summary) | |
| if dry_run: | |
| self.logger.info("Dry run: printing email report to console.") | |
| print(f"Subject: {report_subject}\n") | |
| print(f"Body:\n{report_body}") | |
| return | |
| self.send_email(report_subject, report_body) | |
| def send_email(self, subject: str, body: str) -> None: | |
| """ | |
| Sends an email with the specified subject and body. | |
| If email configuration is not fully set, it prints the email instead. | |
| Args: | |
| subject (str): The subject of the email. | |
| body (str): The body of the email. | |
| """ | |
| smtp_server = os.getenv("SMTP_SERVER") | |
| smtp_port_str = os.getenv("SMTP_PORT") | |
| recipient = os.getenv("EMAIL_RECIPIENT") | |
| email = os.getenv("EMAIL_ADDRESS") | |
| password = os.getenv("EMAIL_PASSWORD") | |
| if not all([smtp_server, smtp_port_str, recipient, email, password]): | |
| self.logger.warning("Email configuration is not fully set. Printing email instead.") | |
| print(f"Subject: {subject}\n") | |
| print(f"Body:\n{body}") | |
| return | |
| assert smtp_server is not None | |
| assert smtp_port_str is not None | |
| assert recipient is not None | |
| assert email is not None | |
| assert password is not None | |
| message = f"Subject: {subject}\n\n{body}" | |
| context = ssl.create_default_context() | |
| try: | |
| smtp_port = int(smtp_port_str) | |
| with smtplib.SMTP_SSL(smtp_server, smtp_port, context=context) as server: | |
| server.login(email, password) | |
| server.sendmail(email, recipient, message) | |
| self.logger.info(f"Successfully sent email report to {recipient}") | |
| except Exception as e: | |
| self.logger.error(f"Failed to send email report: {e}") | |
| def load_config_from_yaml(config_path: str) -> Dict[str, Any]: | |
| with open(config_path, 'r') as file: | |
| config = yaml.safe_load(file) | |
| c = { | |
| "project_id": config.get("project_id"), | |
| "gcp_bucket": config.get("bucket_name"), | |
| "sinks": [], | |
| "logger": logging.getLogger(__name__) | |
| } | |
| for sink_config in config.get("sinks", []): | |
| sink = SinkCls( | |
| name=sink_config["name"], | |
| description=sink_config["description"], | |
| filter_methods=sink_config.get("filter_methods", []), | |
| excluded_principals=sink_config.get("excluded_principals", []) | |
| ) | |
| c["sinks"].append(sink) | |
| logging_config = config.get("logging", {}) | |
| log_level = logging_config.get("level", "INFO") | |
| log_format = logging_config.get("format", "[%(asctime)s] %(levelname)s: %(message)s") | |
| c["logger"].setLevel(log_level) | |
| logging.basicConfig(level=log_level, format=log_format) | |
| return c | |
| def main(): | |
| """ | |
| Main entry point for the script. | |
| """ | |
| parser = argparse.ArgumentParser(description="GCP IAM Log Analyzer") | |
| parser.add_argument("--config", required=True, help="Path to the configuration YAML file.") | |
| subparsers = parser.add_subparsers(dest="command", required=True) | |
| subparsers.add_parser("initialize", help="Initialize/update log sinks in GCP.") | |
| report_parser = subparsers.add_parser("generate-report", help="Generate and send the weekly IAM security report.") | |
| report_parser.add_argument("--dry-run", action="store_true", help="Do not send email, print report to console.") | |
| args = parser.parse_args() | |
| config = load_config_from_yaml(args.config) | |
| log_analyzer = LogAnalyzer( | |
| project_id=config["project_id"], | |
| gcp_bucket=config["gcp_bucket"], | |
| logger=config["logger"], | |
| sinks=config["sinks"] | |
| ) | |
| if args.command == "initialize": | |
| log_analyzer.initialize_sinks() | |
| log_analyzer.logger.info("Sinks initialized successfully.") | |
| elif args.command == "generate-report": | |
| log_analyzer.create_weekly_email_report(dry_run=args.dry_run) | |
| log_analyzer.logger.info("Weekly report generation process completed.") | |
| if __name__ == "__main__": | |
| main() |