mirror of
https://github.com/domainaware/parsedmarc.git
synced 2026-07-30 20:25:57 +00:00
Add Google SecOps (Chronicle) output via the v1 events.import API
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 <noreply@anthropic.com>
This commit is contained in:
Sean Whalen
co-authored by
Claude Fable 5
Claude Fable 5
parent
95e881bdb3
commit
fba083f351
+15
-6
@@ -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
|
||||
|
||||
|
||||
@@ -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
|
||||
|
||||
|
||||
@@ -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) |
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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")
|
||||
@@ -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",
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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)
|
||||
Reference in New Issue
Block a user