diff --git a/README.md b/README.md index 57a8f2d5..ab709f54 100644 --- a/README.md +++ b/README.md @@ -44,7 +44,7 @@ Thanks to all - Optionally send the results to Elasticsearch, Opensearch, and/or Splunk, for use with premade dashboards - Optionally send reports to Apache Kafka -- Optionally send reports to Google SecOps (Chronicle) in UDM format +- Optionally send reports to Google SecOps (Chronicle) in UDM format via API or stdout ## Python Compatibility diff --git a/docs/source/google_secops.md b/docs/source/google_secops.md index 7ae1c134..5451de82 100644 --- a/docs/source/google_secops.md +++ b/docs/source/google_secops.md @@ -6,12 +6,29 @@ To enable Google SecOps output, add a `[google_secops]` section to your configuration file: +### Primary Method: Chronicle Ingestion API + +The recommended approach is to send events directly to Chronicle via the Ingestion API: + ```ini [general] save_aggregate = True save_forensic = True [google_secops] +# Required: Path to Google service account JSON credentials file +api_credentials_file = /path/to/service-account-credentials.json + +# Required: Chronicle customer ID +api_customer_id = your-customer-id-here + +# Optional: Chronicle region (default: us) +# Options: us, europe, asia-southeast1, me-central2, australia-southeast1 +api_region = us + +# Optional: Log type for Chronicle ingestion (default: DMARC) +api_log_type = DMARC + # Optional: Include forensic report message payload (default: False) # For privacy, message bodies are excluded by default include_ruf_payload = False @@ -29,6 +46,23 @@ static_observer_vendor = parsedmarc static_environment = prod ``` +### Alternative Method: stdout Output + +If you prefer to use an external log shipper (Fluentd, Logstash, Chronicle forwarder), set `use_stdout = True`: + +```ini +[google_secops] +# Output to stdout instead of Chronicle API +use_stdout = True + +# Other optional configuration options (as above) +include_ruf_payload = False +ruf_payload_max_bytes = 4096 +static_observer_name = my-instance +static_observer_vendor = parsedmarc +static_environment = prod +``` + ## Output Format The Google SecOps output produces newline-delimited JSON (NDJSON) in Chronicle UDM format, which can be ingested into Google SecOps for hunting and dashboarding. @@ -338,9 +372,25 @@ By default, forensic report message bodies are **excluded** from the output to p The Google SecOps output works with all parsedmarc input methods, including file processing and mailbox monitoring. -### Processing Files +### Primary Method: Direct API Ingestion -To output DMARC reports from files to Google SecOps, redirect stdout or use the output in your ingestion pipeline: +With Chronicle Ingestion API configured, events are sent directly to Chronicle: + +```bash +# Process files - events are sent to Chronicle API automatically +parsedmarc -c config.ini samples/aggregate/*.xml + +# Monitor mailbox - events are sent to Chronicle API in real-time +parsedmarc -c config.ini +``` + +No additional log shippers or pipelines are needed. The Google SecOps client handles authentication and batching automatically. + +### Alternative Method: stdout Output with Log Shipper + +If using `use_stdout = True` in your configuration, output DMARC reports to an external log shipper: + +#### Processing Files ```bash # Output to stdout @@ -349,11 +399,11 @@ parsedmarc -c config.ini samples/aggregate/*.xml > dmarc_events.ndjson # Stream to file parsedmarc -c config.ini samples/aggregate/*.xml >> /var/log/dmarc/events.ndjson -# Use with a log shipper (e.g., Fluentd, Logstash) -parsedmarc -c config.ini samples/aggregate/*.xml | your-log-shipper +# Pipe to log shipper (e.g., Fluentd, Logstash, Chronicle forwarder) +parsedmarc -c config.ini samples/aggregate/*.xml | fluentd ``` -### Monitoring Mailboxes +#### Monitoring Mailboxes The Google SecOps output automatically works when monitoring mailboxes via IMAP, Microsoft Graph, or Gmail API. Configure your mailbox connection and enable watching: @@ -373,15 +423,17 @@ user = dmarc@example.com password = yourpassword [google_secops] +# Use stdout mode for log shipper integration +use_stdout = True include_ruf_payload = False static_observer_name = mailbox-monitor static_environment = prod ``` -When watching a mailbox, parsedmarc will continuously output UDM events to stdout as new reports arrive. Pipe this to your log shipper for real-time ingestion: +When watching a mailbox with stdout mode, parsedmarc continuously outputs UDM events as new reports arrive: ```bash parsedmarc -c config.ini | fluentd ``` -The output is in newline-delimited JSON format, with one UDM event per line, ready for ingestion into Google SecOps. +The output is in newline-delimited JSON format, with one UDM event per line, ready for collection by your log shipper. diff --git a/parsedmarc/cli.py b/parsedmarc/cli.py index 89dabd81..8570e370 100644 --- a/parsedmarc/cli.py +++ b/parsedmarc/cli.py @@ -753,6 +753,11 @@ def _main(): google_secops_static_observer_name=None, google_secops_static_observer_vendor="parsedmarc", google_secops_static_environment=None, + google_secops_api_credentials_file=None, + google_secops_api_customer_id=None, + google_secops_api_region="us", + google_secops_api_log_type="DMARC", + google_secops_use_stdout=False, webhook_aggregate_url=None, webhook_forensic_url=None, webhook_smtp_tls_url=None, @@ -1355,6 +1360,26 @@ def _main(): opts.google_secops_static_environment = google_secops_config[ "static_environment" ] + if "api_credentials_file" in google_secops_config: + opts.google_secops_api_credentials_file = google_secops_config[ + "api_credentials_file" + ] + if "api_customer_id" in google_secops_config: + opts.google_secops_api_customer_id = google_secops_config[ + "api_customer_id" + ] + if "api_region" in google_secops_config: + opts.google_secops_api_region = google_secops_config[ + "api_region" + ] + if "api_log_type" in google_secops_config: + opts.google_secops_api_log_type = google_secops_config[ + "api_log_type" + ] + if "use_stdout" in google_secops_config: + opts.google_secops_use_stdout = google_secops_config.getboolean( + "use_stdout" + ) if "webhook" in config.sections(): webhook_config = config["webhook"] @@ -1542,6 +1567,11 @@ def _main(): static_observer_name=opts.google_secops_static_observer_name, static_observer_vendor=opts.google_secops_static_observer_vendor, static_environment=opts.google_secops_static_environment, + api_credentials_file=opts.google_secops_api_credentials_file, + api_customer_id=opts.google_secops_api_customer_id, + api_region=opts.google_secops_api_region, + api_log_type=opts.google_secops_api_log_type, + use_stdout=opts.google_secops_use_stdout, ) except Exception as error_: logger.error("Google SecOps Error: {0}".format(error_.__str__())) diff --git a/parsedmarc/google_secops.py b/parsedmarc/google_secops.py index 03f42913..aa3ec6bf 100644 --- a/parsedmarc/google_secops.py +++ b/parsedmarc/google_secops.py @@ -8,10 +8,19 @@ import json from datetime import datetime, timezone from typing import Any, Optional +import requests +from google.auth.transport.requests import Request +from google.oauth2 import service_account + +from parsedmarc.constants import USER_AGENT from parsedmarc.log import logger from parsedmarc.utils import human_timestamp_to_datetime +class GoogleSecOpsError(RuntimeError): + """Raised when a Google SecOps API error occurs""" + + class GoogleSecOpsClient: """A client for Google SecOps (Chronicle) UDM output""" @@ -22,6 +31,11 @@ class GoogleSecOpsClient: static_observer_name: Optional[str] = None, static_observer_vendor: str = "parsedmarc", static_environment: Optional[str] = None, + api_credentials_file: Optional[str] = None, + api_customer_id: Optional[str] = None, + api_region: str = "us", + api_log_type: str = "DMARC", + use_stdout: bool = False, ): """ Initializes the GoogleSecOpsClient @@ -32,12 +46,124 @@ class GoogleSecOpsClient: static_observer_name: Static observer name for telemetry static_observer_vendor: Static observer vendor (default: parsedmarc) static_environment: Static environment (prod/dev/custom string) + api_credentials_file: Path to Google service account JSON credentials + api_customer_id: Chronicle customer ID (required for API) + api_region: Chronicle region (us, europe, asia-southeast1, etc.) + api_log_type: Log type for Chronicle ingestion (default: DMARC) + use_stdout: Output to stdout instead of API (default: False) """ self.include_ruf_payload = include_ruf_payload self.ruf_payload_max_bytes = ruf_payload_max_bytes self.static_observer_name = static_observer_name self.static_observer_vendor = static_observer_vendor self.static_environment = static_environment + self.use_stdout = use_stdout + + # API configuration + self.api_credentials_file = api_credentials_file + self.api_customer_id = api_customer_id + self.api_region = api_region + self.api_log_type = api_log_type + self.credentials = None + self.session = None + + # Initialize API client if not using stdout + if not self.use_stdout: + if not self.api_credentials_file or not self.api_customer_id: + raise GoogleSecOpsError( + "api_credentials_file and api_customer_id are required when not using stdout. " + "Set use_stdout=True to output to stdout instead." + ) + self._initialize_api_client() + + def _initialize_api_client(self): + """Initialize the Chronicle API client with authentication""" + try: + logger.debug("Initializing Chronicle API client") + + # Load service account credentials + self.credentials = service_account.Credentials.from_service_account_file( + self.api_credentials_file, + scopes=["https://www.googleapis.com/auth/cloud-platform"], + ) + + # Create session with authentication + self.session = requests.Session() + self.session.headers.update({"User-Agent": USER_AGENT}) + + logger.info("Chronicle API client initialized successfully") + except Exception as e: + raise GoogleSecOpsError(f"Failed to initialize Chronicle API client: {e}") + + def _get_api_endpoint(self) -> str: + """Get the Chronicle Ingestion API endpoint based on region""" + return f"https://{self.api_region}-chronicle.googleapis.com/v1alpha/projects/{self.api_customer_id}/locations/{self.api_region}/instances/default/logTypes/{self.api_log_type}/logs:import" + + def _send_events_to_api(self, events: list[str]) -> None: + """ + Send UDM events to Chronicle Ingestion API + + Args: + events: List of NDJSON event strings + """ + if not events: + return + + try: + # Refresh credentials if needed + if not self.credentials.valid: + self.credentials.refresh(Request()) + + # Prepare request + endpoint = self._get_api_endpoint() + headers = { + "Authorization": f"Bearer {self.credentials.token}", + "Content-Type": "application/json", + } + + # Chronicle expects events in inline_source format + # Each event should be a separate log entry + payload = { + "inline_source": { + "logs": [json.loads(event) for event in events] + } + } + + logger.debug(f"Sending {len(events)} events to Chronicle API") + + response = self.session.post( + endpoint, + headers=headers, + json=payload, + timeout=60, + ) + + if response.status_code == 200: + logger.info(f"Successfully sent {len(events)} events to Chronicle") + else: + error_msg = f"Chronicle API error: {response.status_code} - {response.text}" + logger.error(error_msg) + raise GoogleSecOpsError(error_msg) + + except GoogleSecOpsError: + raise + except Exception as e: + raise GoogleSecOpsError(f"Failed to send events to Chronicle API: {e}") + + def _output_events(self, events: list[str]) -> None: + """ + Output events either to stdout or Chronicle API + + Args: + events: List of NDJSON event strings + """ + if self.use_stdout: + # Output to stdout for collection by external log shippers + for event in events: + print(event) + else: + # Send directly to Chronicle API + self._send_events_to_api(events) def _get_severity(self, disposition: str, spf_aligned: bool, dkim_aligned: bool) -> str: """ @@ -131,13 +257,16 @@ class GoogleSecOpsClient: self, aggregate_report: dict[str, Any] ) -> list[str]: """ - Convert aggregate DMARC report to Google SecOps UDM format (NDJSON) + Convert aggregate DMARC report to Google SecOps UDM format and send to Chronicle + + When use_stdout=False: Events are sent to Chronicle API, returns empty list + When use_stdout=True: Returns list of NDJSON event strings for stdout Args: aggregate_report: Aggregate report dictionary from parsedmarc Returns: - List of NDJSON event strings + List of NDJSON event strings (empty if sent to API) """ logger.debug("Converting aggregate report to Google SecOps UDM format") events = [] @@ -308,19 +437,26 @@ class GoogleSecOpsClient: } events.append(json.dumps(error_event, ensure_ascii=False)) - return events + # Output events (to stdout or API) + self._output_events(events) + + # Return events only if using stdout (for CLI to print) + return events if self.use_stdout else [] def save_forensic_report_to_google_secops( self, forensic_report: dict[str, Any] ) -> list[str]: """ - Convert forensic DMARC report to Google SecOps UDM format (NDJSON) + Convert forensic DMARC report to Google SecOps UDM format and send to Chronicle + + When use_stdout=False: Events are sent to Chronicle API, returns empty list + When use_stdout=True: Returns list of NDJSON event strings for stdout Args: forensic_report: Forensic report dictionary from parsedmarc Returns: - List of NDJSON event strings + List of NDJSON event strings (empty if sent to API) """ logger.debug("Converting forensic report to Google SecOps UDM format") events = [] @@ -456,19 +592,26 @@ class GoogleSecOpsClient: } events.append(json.dumps(error_event, ensure_ascii=False)) - return events + # Output events (to stdout or API) + self._output_events(events) + + # Return events only if using stdout (for CLI to print) + return events if self.use_stdout else [] def save_smtp_tls_report_to_google_secops( self, smtp_tls_report: dict[str, Any] ) -> list[str]: """ - Convert SMTP TLS report to Google SecOps UDM format (NDJSON) + Convert SMTP TLS report to Google SecOps UDM format and send to Chronicle + + When use_stdout=False: Events are sent to Chronicle API, returns empty list + When use_stdout=True: Returns list of NDJSON event strings for stdout Args: smtp_tls_report: SMTP TLS report dictionary from parsedmarc Returns: - List of NDJSON event strings + List of NDJSON event strings (empty if sent to API) """ logger.debug("Converting SMTP TLS report to Google SecOps UDM format") events = [] @@ -553,4 +696,8 @@ class GoogleSecOpsClient: } events.append(json.dumps(error_event, ensure_ascii=False)) - return events + # Output events (to stdout or API) + self._output_events(events) + + # Return events only if using stdout (for CLI to print) + return events if self.use_stdout else []