From fba083f351c01aae70edc9d1df755039f04cb43c Mon Sep 17 00:00:00 2001 From: Sean Whalen <44679+seanthegeek@users.noreply.github.com> Date: Sat, 11 Jul 2026 15:08:31 -0400 Subject: [PATCH] Add Google SecOps (Chronicle) output via the v1 events.import API MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Sends parsed reports to a Google SecOps instance as pre-normalized UDM events through the GA v1 Chronicle API: POST https://chronicle.{region}.rep.googleapis.com/v1/{parent}/events:import Pre-normalized events bypass SecOps's server-side (CBN) parsing layer, so no parser needs to be installed in the tenant; the CBN parser in google_secops_parser/ remains the collector-based alternative for deployments that want raw-log retention, and both paths emit the same UDM shape and additional-field keys so searches port between them. Google's ingestion docs recommend the UDM-events path when possible. Implementation notes, all grounded in the v1 reference docs: - Auth is standard Google Cloud IAM (Chronicle API Editor role / chronicle.events.import permission): a service account key file when [gsecops] credentials_file is set, Application Default Credentials otherwise. google-auth was already a transitive requirement via mailsuite[gmail]; it is now declared directly. - Batches follow the documented best practices (1,000 events per request, 60 s timeout). events.import is all-or-nothing — one invalid event rejects the whole request — so a rejected batch is bisected to isolate invalid events; valid events are still delivered and the drop count is raised as an output error at the end. - to and subject are repeated fields in the UDM Email message; SecurityResult action and category are repeated enums. - additional is a protobuf Struct, so counts and ASNs stay numbers and alignment flags stay booleans — range-queryable without the CBN string-hop. New [gsecops] config section (project_id, instance_id, region, credentials_file) is wired through the INI schema, _parse_config, Namespace defaults, PARSEDMARC_GSECOPS_* env vars, the usage docs, and per-batch client construction (SIGHUP-reload safe by construction). Tests build events from real sample reports and assert on the payloads sent through a mocked AuthorizedSession, including batching and the 400-bisect path. Co-Authored-By: Claude Fable 5 --- CHANGELOG.md | 21 +- docs/source/usage.md | 22 ++ google_secops_parser/README.md | 17 +- parsedmarc/cli.py | 47 ++++ parsedmarc/gsecops.py | 445 +++++++++++++++++++++++++++++++++ pyproject.toml | 1 + tests/test_cli.py | 58 +++++ tests/test_gsecops.py | 317 +++++++++++++++++++++++ 8 files changed, 918 insertions(+), 10 deletions(-) create mode 100644 parsedmarc/gsecops.py create mode 100644 tests/test_gsecops.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 15cb3a37..12195c81 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,14 +4,23 @@ ### Features +- **Google SecOps (Chronicle) output** (`[gsecops]` config section). Sends + reports to a Google SecOps instance as pre-normalized Unified Data Model + (UDM) events via the GA v1 Chronicle API `events.import` method — DMARC + aggregate and failure reports as `EMAIL_TRANSACTION` events, SMTP TLS + reports as `GENERIC_EVENT` events. Uses standard Google Cloud authentication + (a service account key file, or Application Default Credentials), batches + per the documented API best practices, and isolates invalid events by + bisecting rejected batches so one bad event cannot discard a whole run. + Because the events arrive already normalized, no parser needs to be + installed in the SecOps tenant. - **Google SecOps (Chronicle) UDM parser** (`google_secops_parser/`). A configuration-based normalizer (CBN) that maps the JSON events parsedmarc - emits through the `[syslog]` output to the Unified Data Model: DMARC - aggregate and failure reports become `EMAIL_TRANSACTION` events, SMTP TLS - reports become `GENERIC_EVENT` events. Ships with real sample events for the - SecOps parser-validation tool; see `google_secops_parser/README.md` for - installation, field mappings, and caveats (not yet validated against a live - tenant). + emits through its `[syslog]` output to the same UDM shape as the `[gsecops]` + output, for deployments that prefer collector-based ingestion with raw-log + retention. Ships with real sample events for the SecOps parser-validation + tool; see `google_secops_parser/README.md` for installation, field mappings, + and caveats (not yet validated against a live tenant). ### Bug fixes diff --git a/docs/source/usage.md b/docs/source/usage.md index 395a67e4..90a60429 100644 --- a/docs/source/usage.md +++ b/docs/source/usage.md @@ -565,6 +565,27 @@ The full set of configuration options are: - `smtp_tls_url` - str: URL of the webhook which should receive the smtp_tls reports - `timeout` - int: Interval in which the webhook call should timeout +- `gsecops` - Send reports to [Google SecOps](https://cloud.google.com/security/products/security-operations) + (Chronicle) as Unified Data Model (UDM) events via the v1 Chronicle API + [`events.import`](https://docs.cloud.google.com/chronicle/docs/reference/ingestion-methods) + method. Pre-normalized UDM events bypass SecOps's server-side parsing, so no + custom parser needs to be installed in the tenant (the alternative, + raw-log-preserving path is the syslog output plus the parser in the + [google_secops_parser](https://github.com/domainaware/parsedmarc/tree/master/google_secops_parser) + directory). + - `project_id` - str: The Google Cloud project ID linked to the SecOps + instance at onboarding. A correctly-permissioned account in any other + project will fail to authenticate. + - `instance_id` - str: The SecOps instance (customer) GUID, shown under + **SIEM Settings > Profile** in the SecOps console + - `region` - str: The SecOps instance region, e.g. `us` or `europe` + (Default: `us`) + - `credentials_file` - str: Path to a Google service account JSON key file. + When not set, [Application Default Credentials](https://cloud.google.com/docs/authentication/application-default-credentials) + are used. Either way, the account must hold the **Chronicle API Editor** + IAM role (the `chronicle.events.import` permission) in the linked + project. + :::{warning} It is **strongly recommended** to **not** use the `nameservers` setting. By default, `parsedmarc` uses @@ -766,6 +787,7 @@ For sections with underscores in the name, the full section name is used: | `log_analytics` | `PARSEDMARC_LOG_ANALYTICS_` | | `gelf` | `PARSEDMARC_GELF_` | | `webhook` | `PARSEDMARC_WEBHOOK_` | +| `gsecops` | `PARSEDMARC_GSECOPS_` | ## Performance tuning diff --git a/google_secops_parser/README.md b/google_secops_parser/README.md index 27998dab..01edf77a 100644 --- a/google_secops_parser/README.md +++ b/google_secops_parser/README.md @@ -5,6 +5,15 @@ custom parser (configuration-based normalizer / CBN) that maps the JSON events parsedmarc emits through its built-in `[syslog]` output to the Unified Data Model (UDM). +> **Prefer an API-based setup?** parsedmarc also ships a `[gsecops]` output +> that sends the same UDM events directly to the Chronicle API +> (`events.import`), with no collector or custom parser to install — see the +> `gsecops` section of the parsedmarc usage documentation. Use this parser +> instead when you want collector-based ingestion with raw-log retention, or +> when you already run a Bindplane pipeline. Both paths use the same UDM +> mapping and the same `additional` field keys, so searches and dashboards +> port between them. + This is a **SecOps-side parser** — parsedmarc already ships structured JSON over syslog, and the DMARC→UDM mapping lives here so that a downstream UDM schema change is a parser edit rather than a parsedmarc release. One paired @@ -39,7 +48,7 @@ parsedmarc emits three flat JSON shapes (one object per syslog line). The parser detects them by a field unique to each and maps them as follows: | parsedmarc report | Detected by | UDM `metadata.event_type` | -|---|---|---| +| --- | --- | --- | | DMARC aggregate | `xml_schema` | `EMAIL_TRANSACTION` | | DMARC failure | `feedback_type` or `arrival_date_utc` | `EMAIL_TRANSACTION` | | SMTP TLS (RFC 8460) | `policy_type` or `result_type` | `GENERIC_EVENT` | @@ -106,7 +115,7 @@ and [SecurityResult reference](https://docs.cloud.google.com/chronicle/docs/refe ### DMARC aggregate → `EMAIL_TRANSACTION` | parsedmarc field | UDM field | -|---|---| +| --- | --- | | `begin_date` | `metadata.event_timestamp` (via `date{}`) | | `report_id` | `metadata.product_log_id` | | `source_ip_address` | `principal.ip` | @@ -121,7 +130,7 @@ and [SecurityResult reference](https://docs.cloud.google.com/chronicle/docs/refe ### DMARC failure → `EMAIL_TRANSACTION` | parsedmarc field | UDM field | -|---|---| +| --- | --- | | `arrival_date_utc` | `metadata.event_timestamp` (via `date{}`) | | `message_id` | `metadata.product_log_id`, `network.email.mail_id` | | `source_ip_address` | `principal.ip` | @@ -137,7 +146,7 @@ and [SecurityResult reference](https://docs.cloud.google.com/chronicle/docs/refe ### SMTP TLS → `GENERIC_EVENT` | parsedmarc field | UDM field | -|---|---| +| --- | --- | | `begin_date` | `metadata.event_timestamp` (ISO 8601, via `date{}`) | | `report_id` | `metadata.product_log_id` | | `policy_domain` | `target.hostname` (the noun; falls back to `receiving_mx_hostname` when absent) | diff --git a/parsedmarc/cli.py b/parsedmarc/cli.py index 578a3e66..0adba0a2 100644 --- a/parsedmarc/cli.py +++ b/parsedmarc/cli.py @@ -31,6 +31,7 @@ from parsedmarc import ( gelf, get_dmarc_reports_from_mailbox, get_dmarc_reports_from_mbox, + gsecops, kafkaclient, loganalytics, opensearch, @@ -155,6 +156,7 @@ _KNOWN_SECTIONS = frozenset( "log_analytics", "gelf", "webhook", + "gsecops", } ) @@ -175,6 +177,7 @@ _DIRECT_FILE_KEYS = frozenset( "MSGRAPH_TOKEN_FILE", "GMAIL_API_CREDENTIALS_FILE", "GMAIL_API_TOKEN_FILE", + "GSECOPS_CREDENTIALS_FILE", ] ) @@ -1169,6 +1172,27 @@ def _parse_config(config: ConfigParser, opts): if "timeout" in webhook_config: opts.webhook_timeout = webhook_config.getint("timeout") + if "gsecops" in config.sections(): + gsecops_config = config["gsecops"] + if "project_id" in gsecops_config: + opts.gsecops_project_id = gsecops_config["project_id"] + else: + raise ConfigurationError( + "project_id setting missing from the gsecops config section" + ) + if "instance_id" in gsecops_config: + opts.gsecops_instance_id = gsecops_config["instance_id"] + else: + raise ConfigurationError( + "instance_id setting missing from the gsecops config section" + ) + if "region" in gsecops_config: + opts.gsecops_region = gsecops_config["region"] + if "credentials_file" in gsecops_config: + opts.gsecops_credentials_file = _expand_path( + gsecops_config["credentials_file"] + ) + return index_prefix_domain_map @@ -1841,6 +1865,25 @@ def _main(): except Exception as e: log_output_error("Log Analytics", f"Unknown publishing error: {e}") + if opts.gsecops_project_id: + try: + gsecops_client = gsecops.GoogleSecOpsClient( + project_id=opts.gsecops_project_id, + instance_id=opts.gsecops_instance_id, + region=opts.gsecops_region, + credentials_file=opts.gsecops_credentials_file, + ) + gsecops_client.publish_results( + reports_, + opts.save_aggregate, + opts.save_failure, + opts.save_smtp_tls, + ) + except gsecops.GoogleSecOpsError as e: + log_output_error("Google SecOps", e.__str__()) + except Exception as e: + log_output_error("Google SecOps", f"Unknown publishing error: {e}") + if opts.fail_on_output_error and output_errors: raise ParserError( "Output destination failures detected: {0}".format( @@ -2104,6 +2147,10 @@ def _main(): webhook_failure_url=None, webhook_smtp_tls_url=None, webhook_timeout=60, + gsecops_project_id=None, + gsecops_instance_id=None, + gsecops_region="us", + gsecops_credentials_file=None, normalize_timespan_threshold_hours=24.0, postgresql_host=None, postgresql_port=5432, diff --git a/parsedmarc/gsecops.py b/parsedmarc/gsecops.py new file mode 100644 index 00000000..8b973abf --- /dev/null +++ b/parsedmarc/gsecops.py @@ -0,0 +1,445 @@ +# -*- coding: utf-8 -*- + +"""Google SecOps (Chronicle) output. + +Sends parsed reports to a Google SecOps instance as Unified Data Model (UDM) +events via the GA v1 Chronicle API ``events.import`` method: + + POST https://chronicle.{region}.rep.googleapis.com/v1/{parent}/events:import + parent = projects/{project}/locations/{region}/instances/{instance} + +Because the events are already normalized, they bypass SecOps's server-side +parsing (CBN) layer entirely. The UDM mapping mirrors the CBN parser shipped +in ``google_secops_parser/`` — the same ``additional`` keys are used by both +delivery paths so searches and dashboards port between them: + + * DMARC aggregate report record -> EMAIL_TRANSACTION + * DMARC failure report record -> EMAIL_TRANSACTION + * SMTP TLS report record -> GENERIC_EVENT + +API references: + https://docs.cloud.google.com/chronicle/docs/reference/ingestion-methods + https://docs.cloud.google.com/chronicle/docs/reference/rest/v1/projects.locations.instances.events/import + +Authentication uses standard Google Cloud credentials (a service account key +file, or Application Default Credentials when no file is configured). The +account needs the Chronicle API Editor IAM role — specifically the +``chronicle.events.import`` permission — in the Google Cloud project that was +linked to the SecOps instance at onboarding; a correctly-permissioned account +in any other project will fail to authenticate. +""" + +from __future__ import annotations + +from typing import Any + +import google.auth +from google.auth.transport.requests import AuthorizedSession +from google.oauth2 import service_account + +from parsedmarc import ( + parsed_aggregate_reports_to_csv_rows, + parsed_failure_reports_to_csv_rows, + parsed_smtp_tls_reports_to_csv_rows, +) +from parsedmarc.log import logger +from parsedmarc.types import AggregateReport, FailureReport, SMTPTLSReport + +# https://docs.cloud.google.com/chronicle/docs/reference/rest/v1/projects.locations.instances.events/import +# also accepts the narrower https://www.googleapis.com/auth/chronicle scope; +# cloud-platform is what Google's own secops SDK requests. +_SCOPES = ["https://www.googleapis.com/auth/cloud-platform"] + +# Documented API behavior and best practices: 4 MB maximum request size, +# ~1,000 log lines and ~2 MB per batch optimal, 60 second request timeout. +# https://docs.cloud.google.com/chronicle/docs/reference/ingestion-methods +_MAX_EVENTS_PER_BATCH = 1000 +_REQUEST_TIMEOUT = 60 + +_VENDOR = "parsedmarc" + +_DISPOSITION_TO_ACTION = { + "none": "ALLOW", + "quarantine": "QUARANTINE", + "reject": "BLOCK", +} + +_DELIVERY_RESULT_TO_ACTION = { + "delivered": "ALLOW", + "quarantine": "QUARANTINE", + "reject": "BLOCK", +} + + +class GoogleSecOpsError(Exception): + """Raised when a Google SecOps API error occurs""" + + +def _rfc3339(timestamp: str) -> str: + """Converts parsedmarc's ``YYYY-MM-DD HH:MM:SS`` UTC wall-clock strings to + RFC 3339 (proto Timestamp JSON). SMTP TLS timestamps are already RFC 3339 + and pass through unchanged.""" + timestamp = timestamp.replace(" ", "T") + if not timestamp.endswith("Z"): + timestamp += "Z" + return timestamp + + +def _metadata( + event_type: str, product_event_type: str, timestamp: str, log_id: Any +) -> dict[str, Any]: + metadata: dict[str, Any] = { + "eventTimestamp": _rfc3339(timestamp), + "eventType": event_type, + "vendorName": _VENDOR, + "productName": _VENDOR, + "productEventType": product_event_type, + } + if log_id: + metadata["productLogId"] = log_id + return metadata + + +def _source_noun(row: dict[str, Any]) -> dict[str, Any]: + """principal: the sending source (machine details only).""" + noun: dict[str, Any] = {} + if row.get("source_ip_address"): + noun["ip"] = [row["source_ip_address"]] + if row.get("source_reverse_dns"): + noun["hostname"] = row["source_reverse_dns"] + if row.get("source_country"): + noun["location"] = {"countryOrRegion": row["source_country"]} + return noun + + +def _additional(row: dict[str, Any], keys: dict[str, str]) -> dict[str, Any]: + """Builds the ``additional`` Struct from row values. + + ``keys`` maps row keys to their ``additional`` key names (matching the + CBN parser's key names so searches port between the two delivery paths). + Because ``additional`` is a protobuf Struct, native JSON types survive: + booleans stay booleans and counts stay numbers, so SecOps can filter and + range-query them directly. Empty strings and ``None`` (absent in the + source report) are dropped. + """ + additional: dict[str, Any] = {} + for row_key, additional_key in keys.items(): + value = row.get(row_key) + if value is None or value == "": + continue + # parsedmarc writes the literal "none" when there are no overrides + if row_key.startswith("policy_override_") and value == "none": + continue + additional[additional_key] = value + return additional + + +_AGGREGATE_ADDITIONAL_KEYS = { + "org_name": "org_name", + "org_email": "org_email", + "org_extra_contact_info": "org_extra_contact_info", + "errors": "errors", + "begin_date": "begin_date", + "end_date": "end_date", + "count": "count", + "p": "dmarc_policy", + "sp": "dmarc_subdomain_policy", + "np": "dmarc_np_policy", + "pct": "dmarc_pct", + "fo": "dmarc_fo", + "adkim": "dkim_alignment_mode", + "aspf": "spf_alignment_mode", + "testing": "dmarc_testing", + "discovery_method": "discovery_method", + "normalized_timespan": "normalized_timespan", + "dmarc_aligned": "dmarc_aligned", + "spf_aligned": "spf_aligned", + "dkim_aligned": "dkim_aligned", + "disposition": "disposition", + "dkim_domains": "dkim_domains", + "dkim_selectors": "dkim_selectors", + "dkim_results": "dkim_results", + "spf_domains": "spf_domains", + "spf_scopes": "spf_scopes", + "spf_results": "spf_results", + "policy_override_reasons": "policy_override_reasons", + "policy_override_comments": "policy_override_comments", + "source_base_domain": "source_base_domain", + "source_name": "source_name", + "source_type": "source_type", + "source_asn": "source_asn", + "source_as_name": "source_as_name", + "source_as_domain": "source_as_domain", + "envelope_from": "envelope_from", + "envelope_to": "envelope_to", +} + +_FAILURE_ADDITIONAL_KEYS = { + "feedback_type": "feedback_type", + "auth_failure": "auth_failure", + "delivery_result": "delivery_result", + "authentication_results": "authentication_results", + "authentication_mechanisms": "authentication_mechanisms", + "user_agent": "user_agent", + "dkim_domain": "dkim_domain", + "arrival_date": "arrival_date", + "source_base_domain": "source_base_domain", + "source_name": "source_name", + "source_type": "source_type", + "source_asn": "source_asn", + "source_as_name": "source_as_name", + "source_as_domain": "source_as_domain", +} + +_SMTP_TLS_ADDITIONAL_KEYS = { + "organization_name": "organization_name", + "begin_date": "begin_date", + "end_date": "end_date", + "policy_domain": "policy_domain", + "policy_type": "policy_type", + "policy_strings": "policy_strings", + "mx_host_patterns": "mx_host_patterns", + "successful_session_count": "successful_session_count", + "failed_session_count": "failed_session_count", + "result_type": "result_type", + "failure_reason_code": "failure_reason_code", + "receiving_mx_hostname": "receiving_mx_hostname", + "receiving_mx_helo": "receiving_mx_helo", + "additional_info_uri": "additional_info_uri", +} + + +def aggregate_report_to_udm_events( + report: AggregateReport, +) -> list[dict[str, Any]]: + """Converts a parsed aggregate report to UDM EMAIL_TRANSACTION events, + one per record row""" + events = [] + for row in parsed_aggregate_reports_to_csv_rows(report): + udm: dict[str, Any] = { + "metadata": _metadata( + "EMAIL_TRANSACTION", "aggregate", row["begin_date"], row["report_id"] + ) + } + principal = _source_noun(row) + if principal: + udm["principal"] = principal + if row.get("domain"): + udm["target"] = {"hostname": row["domain"]} + if row.get("header_from"): + # aggregate reports only carry the From domain, so this is a bare + # domain rather than a full address (documented in the CBN + # parser's README as caveat 5) + udm["network"] = {"email": {"from": row["header_from"]}} + security_result: dict[str, Any] = { + "summary": "DMARC aggregate report", + "action": [ + _DISPOSITION_TO_ACTION.get(row["disposition"], "UNKNOWN_ACTION") + ], + } + if row["dmarc_aligned"] is False: + security_result["category"] = ["AUTH_VIOLATION"] + udm["securityResult"] = [security_result] + udm["additional"] = _additional(row, _AGGREGATE_ADDITIONAL_KEYS) + events.append({"udm": udm}) + return events + + +def failure_report_to_udm_events(report: FailureReport) -> list[dict[str, Any]]: + """Converts a parsed failure report to UDM EMAIL_TRANSACTION events""" + events = [] + for row in parsed_failure_reports_to_csv_rows(report): + udm: dict[str, Any] = { + "metadata": _metadata( + "EMAIL_TRANSACTION", + "failure", + row["arrival_date_utc"], + row.get("message_id"), + ) + } + principal = _source_noun(row) + if principal: + udm["principal"] = principal + if row.get("reported_domain"): + udm["target"] = {"hostname": row["reported_domain"]} + email: dict[str, Any] = {} + if row.get("original_mail_from"): + email["from"] = row["original_mail_from"] + if row.get("original_rcpt_to"): + # to and subject are repeated fields in the UDM Email message + email["to"] = [row["original_rcpt_to"]] + if row.get("subject"): + email["subject"] = [row["subject"]] + if row.get("message_id"): + email["mailId"] = row["message_id"] + if email: + udm["network"] = {"email": email} + udm["securityResult"] = [ + { + "summary": "DMARC failure report", + "category": ["AUTH_VIOLATION"], + "action": [ + _DELIVERY_RESULT_TO_ACTION.get( + row.get("delivery_result") or "", "UNKNOWN_ACTION" + ) + ], + } + ] + udm["additional"] = _additional(row, _FAILURE_ADDITIONAL_KEYS) + events.append({"udm": udm}) + return events + + +def smtp_tls_report_to_udm_events(report: SMTPTLSReport) -> list[dict[str, Any]]: + """Converts a parsed SMTP TLS report to UDM GENERIC_EVENT events, one per + policy summary row and one per failure-detail row""" + events = [] + for row in parsed_smtp_tls_reports_to_csv_rows(report): + udm: dict[str, Any] = { + "metadata": _metadata( + "GENERIC_EVENT", "smtp_tls", row["begin_date"], row["report_id"] + ) + } + target: dict[str, Any] = {} + if row.get("policy_domain"): + target["hostname"] = row["policy_domain"] + elif row.get("receiving_mx_hostname"): + target["hostname"] = row["receiving_mx_hostname"] + if row.get("receiving_ip"): + target["ip"] = [row["receiving_ip"]] + if target: + udm["target"] = target + if row.get("sending_mta_ip"): + udm["principal"] = {"ip": [row["sending_mta_ip"]]} + if row.get("result_type"): + udm["securityResult"] = [ + { + "summary": "SMTP TLS report failure", + "category": ["POLICY_VIOLATION"], + "action": ["FAIL"], + } + ] + udm["additional"] = _additional(row, _SMTP_TLS_ADDITIONAL_KEYS) + events.append({"udm": udm}) + return events + + +class GoogleSecOpsClient(object): + """A client for the Google SecOps (Chronicle) v1 events.import method""" + + def __init__( + self, + project_id: str, + instance_id: str, + region: str = "us", + credentials_file: str | None = None, + ): + """ + Initializes the GoogleSecOpsClient + + Args: + project_id (str): The Google Cloud project linked to the SecOps + instance at onboarding + instance_id (str): The SecOps instance (customer) GUID + region (str): The SecOps instance region, e.g. ``us`` or + ``europe`` (Default: ``us``) + credentials_file (str): Path to a service account JSON key file. + When not set, Application Default Credentials are used. + """ + if not project_id or not instance_id: + raise GoogleSecOpsError( + "Invalid configuration. project_id and instance_id are required." + ) + parent = "projects/{0}/locations/{1}/instances/{2}".format( + project_id, region, instance_id + ) + self.url = ( + "https://chronicle.{0}.rep.googleapis.com/v1/{1}/events:import".format( + region, parent + ) + ) + if credentials_file: + credentials = service_account.Credentials.from_service_account_file( + credentials_file, scopes=_SCOPES + ) + else: + credentials, _ = google.auth.default(scopes=_SCOPES) + self.session = AuthorizedSession(credentials) + self._dropped = 0 + + def _import_events(self, events: list[dict[str, Any]]) -> None: + """Imports a batch of UDM events, bisecting on HTTP 400. + + The v1 ``events.import`` method is all-or-nothing: one invalid event + rejects the entire request. Splitting a rejected batch isolates the + invalid events so the valid remainder is still delivered; each + invalid event is logged and counted rather than aborting the batch. + """ + response = self.session.post( + self.url, + json={"inlineSource": {"events": events}}, + timeout=_REQUEST_TIMEOUT, + ) + if response.status_code == 200: + return + if response.status_code == 400 and len(events) > 1: + middle = len(events) // 2 + self._import_events(events[:middle]) + self._import_events(events[middle:]) + return + if response.status_code == 400: + logger.error( + "Google SecOps rejected event {0}: {1}".format(events[0], response.text) + ) + self._dropped += 1 + return + raise GoogleSecOpsError( + "Import failed with HTTP {0}: {1}".format( + response.status_code, response.text + ) + ) + + def save_events(self, events: list[dict[str, Any]]) -> None: + """Imports UDM events in documented-best-practice batch sizes""" + self._dropped = 0 + for start in range(0, len(events), _MAX_EVENTS_PER_BATCH): + self._import_events(events[start : start + _MAX_EVENTS_PER_BATCH]) + if self._dropped: + raise GoogleSecOpsError( + "{0} of {1} events were rejected by Google SecOps " + "(see error log for details)".format(self._dropped, len(events)) + ) + + def publish_results( + self, + results: dict[str, Any], + save_aggregate: bool, + save_failure: bool, + save_smtp_tls: bool, + ) -> None: + """ + Publishes DMARC and/or SMTP TLS reports to Google SecOps as UDM + events + + Args: + results (dict): Parsing results from parsedmarc + save_aggregate (bool): Whether to save aggregate reports + save_failure (bool): Whether to save failure reports + save_smtp_tls (bool): Whether to save SMTP TLS reports + """ + events: list[dict[str, Any]] = [] + if save_aggregate: + for report in results["aggregate_reports"]: + events += aggregate_report_to_udm_events(report) + if save_failure: + for report in results["failure_reports"]: + events += failure_report_to_udm_events(report) + if save_smtp_tls: + for report in results["smtp_tls_reports"]: + events += smtp_tls_report_to_udm_events(report) + if len(events) > 0: + logger.info( + "Publishing {0} UDM events to Google SecOps".format(len(events)) + ) + self.save_events(events) + logger.info("Successfully published UDM events to Google SecOps") diff --git a/pyproject.toml b/pyproject.toml index 2c6e30cb..342c8e5e 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -45,6 +45,7 @@ dependencies = [ "elasticsearch-dsl==7.4.0", "elasticsearch<7.14.0", "expiringdict>=1.1.4", + "google-auth>=2.0.0", "kafka-python>=2.3.2", "lxml>=4.4.0", "mailsuite[gmail,msgraph]>=2.2.2", diff --git a/tests/test_cli.py b/tests/test_cli.py index b3789bad..5156b687 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -3752,6 +3752,64 @@ class TestParseConfigWebhook(unittest.TestCase): self.assertEqual(opts.webhook_failure_url, "https://old.example.com/fail") +class TestParseConfigGsecops(unittest.TestCase): + def test_gsecops_complete(self): + from parsedmarc.cli import _parse_config + + cp = _config_with( + "gsecops", + { + "project_id": "my-project", + "instance_id": "my-instance", + "region": "europe", + "credentials_file": "~/gsecops-key.json", + }, + ) + opts = _opts() + _parse_config(cp, opts) + self.assertEqual(opts.gsecops_project_id, "my-project") + self.assertEqual(opts.gsecops_instance_id, "my-instance") + self.assertEqual(opts.gsecops_region, "europe") + # file path config values must be wrapped with _expand_path + self.assertEqual( + opts.gsecops_credentials_file, + os.path.expanduser("~/gsecops-key.json"), + ) + + def test_gsecops_missing_project_id_raises(self): + from parsedmarc.cli import ConfigurationError, _parse_config + + cp = _config_with("gsecops", {"instance_id": "my-instance"}) + with self.assertRaises(ConfigurationError): + _parse_config(cp, _opts()) + + def test_gsecops_missing_instance_id_raises(self): + from parsedmarc.cli import ConfigurationError, _parse_config + + cp = _config_with("gsecops", {"project_id": "my-project"}) + with self.assertRaises(ConfigurationError): + _parse_config(cp, _opts()) + + def test_gsecops_env_vars_resolve_to_section(self): + """PARSEDMARC_GSECOPS_* env vars round-trip into the gsecops + section via longest-prefix section matching.""" + from argparse import Namespace + from parsedmarc.cli import _load_config, _parse_config + + env = { + "PARSEDMARC_GSECOPS_PROJECT_ID": "env-project", + "PARSEDMARC_GSECOPS_INSTANCE_ID": "env-instance", + "PARSEDMARC_GSECOPS_REGION": "asia-southeast1", + } + with patch.dict(os.environ, env, clear=False): + config = _load_config(None) + opts = Namespace() + _parse_config(config, opts) + self.assertEqual(opts.gsecops_project_id, "env-project") + self.assertEqual(opts.gsecops_instance_id, "env-instance") + self.assertEqual(opts.gsecops_region, "asia-southeast1") + + class TestConfigureLogging(unittest.TestCase): """_configure_logging is called in every child process for parallel parsing — if it stops attaching a handler, log output goes dark in diff --git a/tests/test_gsecops.py b/tests/test_gsecops.py new file mode 100644 index 00000000..bd75364f --- /dev/null +++ b/tests/test_gsecops.py @@ -0,0 +1,317 @@ +"""Tests for parsedmarc.gsecops""" + +import unittest +from typing import Any, cast +from unittest.mock import MagicMock, patch + +import parsedmarc +from parsedmarc.gsecops import ( + GoogleSecOpsClient, + GoogleSecOpsError, + aggregate_report_to_udm_events, + failure_report_to_udm_events, + smtp_tls_report_to_udm_events, +) +from parsedmarc.types import AggregateReport, FailureReport, SMTPTLSReport + + +def _parse_sample(path: str) -> Any: + result = parsedmarc.parse_report_file( + path, always_use_local_files=True, offline=True + ) + return result["report"] + + +def _aggregate_report() -> AggregateReport: + return cast( + AggregateReport, + _parse_sample("samples/aggregate/!example.com!1538204542!1538463818.xml"), + ) + + +def _failure_report() -> FailureReport: + return cast( + FailureReport, + _parse_sample( + "samples/failure/DMARC Failure Report for domain.de " + "(mail-from=sharepoint@domain.de, ip=10.10.10.10).eml" + ), + ) + + +def _smtp_tls_report() -> SMTPTLSReport: + return cast(SMTPTLSReport, _parse_sample("samples/smtp_tls/mail.ru.json")) + + +class TestAggregateUdmEvents(unittest.TestCase): + """Aggregate reports map to EMAIL_TRANSACTION events whose fields a + UDM search or dashboard would actually filter on.""" + + def test_event_shape(self): + events = aggregate_report_to_udm_events(_aggregate_report()) + self.assertEqual(len(events), 1) + udm = events[0]["udm"] + metadata = udm["metadata"] + self.assertEqual(metadata["eventType"], "EMAIL_TRANSACTION") + # RFC 3339: the record interval start, not ingest time + self.assertEqual(metadata["eventTimestamp"], "2018-10-01T17:07:12Z") + self.assertEqual(metadata["vendorName"], "parsedmarc") + self.assertEqual(metadata["productName"], "parsedmarc") + self.assertEqual(metadata["productEventType"], "aggregate") + self.assertEqual(metadata["productLogId"], "example.com:1538463741") + self.assertEqual(udm["principal"]["ip"], ["12.20.127.122"]) + self.assertEqual(udm["principal"]["location"]["countryOrRegion"], "US") + self.assertEqual(udm["target"]["hostname"], "example.com") + # aggregate reports only carry the From domain + self.assertEqual(udm["network"]["email"]["from"], "example.com") + + def test_security_result(self): + events = aggregate_report_to_udm_events(_aggregate_report()) + (security_result,) = events[0]["udm"]["securityResult"] + # disposition "none" -> ALLOW; action and category are repeated + # fields in the UDM SecurityResult message + self.assertEqual(security_result["action"], ["ALLOW"]) + # dmarc_aligned is False in this sample -> AUTH_VIOLATION + self.assertEqual(security_result["category"], ["AUTH_VIOLATION"]) + + def test_additional_preserves_native_json_types(self): + """additional is a protobuf Struct, so numbers and booleans must + arrive as native JSON types for range queries and boolean filters — + parsedmarc's "store numbers as numbers" rule.""" + events = aggregate_report_to_udm_events(_aggregate_report()) + additional = events[0]["udm"]["additional"] + self.assertIs(additional["count"], 1) + self.assertEqual(additional["source_asn"], 7018) + self.assertIs(additional["dmarc_aligned"], False) + self.assertIs(additional["spf_aligned"], False) + # renamed keys match the CBN parser in google_secops_parser/ so + # searches port between the two delivery paths + self.assertEqual(additional["dmarc_policy"], "none") + self.assertEqual(additional["dmarc_subdomain_policy"], "reject") + self.assertEqual(additional["dmarc_pct"], "100") + self.assertEqual(additional["source_name"], "AT&T") + + def test_additional_drops_empty_values(self): + """Absent report fields are empty strings in the flat rows; they + must not appear as empty additional entries.""" + events = aggregate_report_to_udm_events(_aggregate_report()) + additional = events[0]["udm"]["additional"] + for key, value in additional.items(): + self.assertNotEqual(value, "", f"empty value for {key}") + # this sample publishes no np/fo/testing and no DKIM results + self.assertNotIn("dmarc_np_policy", additional) + self.assertNotIn("dmarc_fo", additional) + self.assertNotIn("dmarc_testing", additional) + self.assertNotIn("dkim_domains", additional) + + +class TestFailureUdmEvents(unittest.TestCase): + def test_event_shape(self): + events = failure_report_to_udm_events(_failure_report()) + self.assertEqual(len(events), 1) + udm = events[0]["udm"] + metadata = udm["metadata"] + self.assertEqual(metadata["eventType"], "EMAIL_TRANSACTION") + self.assertEqual(metadata["eventTimestamp"], "2018-10-01T09:20:27Z") + self.assertEqual(metadata["productEventType"], "failure") + self.assertEqual(udm["principal"]["ip"], ["10.10.10.10"]) + self.assertEqual(udm["target"]["hostname"], "domain.de") + email = udm["network"]["email"] + self.assertEqual(email["from"], "sharepoint@domain.de") + # to and subject are repeated fields in the UDM Email message + self.assertEqual(email["to"], ["peter.pan@domain.de"]) + self.assertEqual(email["subject"], ["Subject"]) + self.assertEqual(email["mailId"], metadata["productLogId"]) + + def test_security_result(self): + events = failure_report_to_udm_events(_failure_report()) + (security_result,) = events[0]["udm"]["securityResult"] + self.assertEqual(security_result["category"], ["AUTH_VIOLATION"]) + # delivery_result "policy" has no direct action mapping + self.assertEqual(security_result["action"], ["UNKNOWN_ACTION"]) + + def test_additional_drops_null_enrichment(self): + """Failure rows carry None for offline enrichment fields + (source_name, dkim_domain, ...); None must never reach the API.""" + events = failure_report_to_udm_events(_failure_report()) + additional = events[0]["udm"]["additional"] + for key, value in additional.items(): + self.assertIsNotNone(value, f"null value for {key}") + self.assertNotEqual(value, "", f"empty value for {key}") + self.assertEqual(additional["auth_failure"], "dmarc") + self.assertEqual(additional["delivery_result"], "policy") + self.assertNotIn("source_name", additional) + self.assertNotIn("dkim_domain", additional) + + +class TestSmtpTlsUdmEvents(unittest.TestCase): + def test_success_and_failure_rows(self): + events = smtp_tls_report_to_udm_events(_smtp_tls_report()) + # mail.ru sample: one policy summary row + two failure details + self.assertEqual(len(events), 3) + for event in events: + udm = event["udm"] + self.assertEqual(udm["metadata"]["eventType"], "GENERIC_EVENT") + self.assertEqual(udm["metadata"]["eventTimestamp"], "2024-02-22T00:00:00Z") + # the policy domain is the noun on every row, including + # failure details (requires the paired serializer fix) + self.assertEqual(udm["target"]["hostname"], "example.com") + self.assertEqual(udm["additional"]["policy_type"], "sts") + success, failure = events[0]["udm"], events[1]["udm"] + # summary rows carry counts (as numbers) and no security_result + self.assertNotIn("securityResult", success) + self.assertEqual(success["additional"]["successful_session_count"], 0) + self.assertEqual(success["additional"]["failed_session_count"], 1) + # failure rows carry the result_type and a FAIL security_result + (security_result,) = failure["securityResult"] + self.assertEqual(security_result["action"], ["FAIL"]) + self.assertEqual(security_result["category"], ["POLICY_VIOLATION"]) + self.assertEqual(failure["additional"]["result_type"], "sts-policy-fetch-error") + self.assertEqual( + failure["additional"]["failure_reason_code"], + "bad https response code: 404", + ) + + +def _client() -> tuple[GoogleSecOpsClient, MagicMock]: + """Builds a client with mocked Google credentials and returns it along + with the mocked AuthorizedSession instance.""" + with ( + patch( + "parsedmarc.gsecops.google.auth.default", + return_value=(MagicMock(), "some-project"), + ), + patch("parsedmarc.gsecops.AuthorizedSession") as mock_session_cls, + ): + client = GoogleSecOpsClient( + project_id="my-project", instance_id="my-instance", region="europe" + ) + return client, mock_session_cls.return_value + + +def _response(status_code: int, text: str = "") -> MagicMock: + return MagicMock(status_code=status_code, text=text) + + +def _event(name: str) -> dict: + return {"udm": {"metadata": {"productLogId": name}}} + + +class TestGoogleSecOpsClient(unittest.TestCase): + def test_missing_required_settings_raises(self): + with self.assertRaises(GoogleSecOpsError): + GoogleSecOpsClient(project_id="", instance_id="my-instance") + with self.assertRaises(GoogleSecOpsError): + GoogleSecOpsClient(project_id="my-project", instance_id="") + + def test_url_uses_region_and_parent(self): + client, _ = _client() + self.assertEqual( + client.url, + "https://chronicle.europe.rep.googleapis.com/v1/" + "projects/my-project/locations/europe/instances/my-instance" + "/events:import", + ) + + def test_service_account_file_credentials(self): + with ( + patch( + "parsedmarc.gsecops.service_account.Credentials" + ".from_service_account_file" + ) as mock_from_file, + patch("parsedmarc.gsecops.AuthorizedSession"), + ): + GoogleSecOpsClient( + project_id="my-project", + instance_id="my-instance", + credentials_file="/path/to/key.json", + ) + self.assertEqual(mock_from_file.call_args.args[0], "/path/to/key.json") + + def test_save_events_posts_inline_source_envelope(self): + client, session = _client() + session.post.return_value = _response(200) + events = [_event("a"), _event("b")] + client.save_events(events) + session.post.assert_called_once() + call = session.post.call_args + self.assertEqual(call.args[0], client.url) + self.assertEqual(call.kwargs["timeout"], 60) + self.assertEqual(call.kwargs["json"], {"inlineSource": {"events": events}}) + + def test_save_events_batches_at_1000(self): + client, session = _client() + session.post.return_value = _response(200) + client.save_events([_event(str(i)) for i in range(1500)]) + self.assertEqual(session.post.call_count, 2) + first, second = session.post.call_args_list + self.assertEqual(len(first.kwargs["json"]["inlineSource"]["events"]), 1000) + self.assertEqual(len(second.kwargs["json"]["inlineSource"]["events"]), 500) + + def test_http_400_bisects_and_delivers_valid_events(self): + """events.import is all-or-nothing: one invalid event rejects the + whole request. The client must bisect a rejected batch so every + valid event is still delivered, and report the drop.""" + client, session = _client() + bad_event = _event("bad") + + def post(url, *, json, timeout): + batch = json["inlineSource"]["events"] + if bad_event in batch: + return _response(400, "invalid event") + return _response(200) + + session.post.side_effect = post + events = [_event("a"), bad_event, _event("c")] + with self.assertRaises(GoogleSecOpsError) as raised: + client.save_events(events) + self.assertIn("1 of 3", str(raised.exception)) + delivered = [] + for call in session.post.call_args_list: + batch = call.kwargs["json"]["inlineSource"]["events"] + if bad_event not in batch: + delivered += batch + self.assertIn(_event("a"), delivered) + self.assertIn(_event("c"), delivered) + + def test_http_500_raises(self): + client, session = _client() + session.post.return_value = _response(500, "boom") + with self.assertRaises(GoogleSecOpsError) as raised: + client.save_events([_event("a"), _event("b")]) + self.assertIn("500", str(raised.exception)) + # a non-400 error must not trigger the bisect fallback + self.assertEqual(session.post.call_count, 1) + + def test_publish_results_honors_save_flags(self): + client, session = _client() + session.post.return_value = _response(200) + results = { + "aggregate_reports": [_aggregate_report()], + "failure_reports": [_failure_report()], + "smtp_tls_reports": [_smtp_tls_report()], + } + client.publish_results(results, True, True, True) + (call,) = session.post.call_args_list + batch = call.kwargs["json"]["inlineSource"]["events"] + # 1 aggregate row + 1 failure row + 3 SMTP TLS rows + self.assertEqual(len(batch), 5) + event_types = [e["udm"]["metadata"]["productEventType"] for e in batch] + self.assertEqual( + event_types, ["aggregate", "failure", "smtp_tls", "smtp_tls", "smtp_tls"] + ) + + def test_publish_results_skips_disabled_types(self): + client, session = _client() + session.post.return_value = _response(200) + results = { + "aggregate_reports": [_aggregate_report()], + "failure_reports": [_failure_report()], + "smtp_tls_reports": [_smtp_tls_report()], + } + client.publish_results(results, False, False, False) + session.post.assert_not_called() + + +if __name__ == "__main__": + unittest.main(verbosity=2)