Add Google SecOps output module implementation

Co-authored-by: seanthegeek <44679+seanthegeek@users.noreply.github.com>

parsedmarc/cli.py

@@ -27,6 +27,7 @@ from parsedmarc import (
     gelf,
     get_dmarc_reports_from_mailbox,
     get_dmarc_reports_from_mbox,
+    google_secops,
     kafkaclient,
     loganalytics,
     opensearch,
@@ -284,6 +285,14 @@ def _main():
                 except Exception as error_:
                     logger.error("GELF Error: {0}".format(error_.__str__()))
 
+                try:
+                    if opts.google_secops:
+                        events = google_secops_client.save_aggregate_report_to_google_secops(report)
+                        for event in events:
+                            print(event)
+                except Exception as error_:
+                    logger.error("Google SecOps Error: {0}".format(error_.__str__()))
+
                 try:
                     if opts.webhook_aggregate_url:
                         indent_value = 2 if opts.prettify_json else None
@@ -369,6 +378,14 @@ def _main():
                 except Exception as error_:
                     logger.error("GELF Error: {0}".format(error_.__str__()))
 
+                try:
+                    if opts.google_secops:
+                        events = google_secops_client.save_forensic_report_to_google_secops(report)
+                        for event in events:
+                            print(event)
+                except Exception as error_:
+                    logger.error("Google SecOps Error: {0}".format(error_.__str__()))
+
                 try:
                     if opts.webhook_forensic_url:
                         indent_value = 2 if opts.prettify_json else None
@@ -454,6 +471,14 @@ def _main():
                 except Exception as error_:
                     logger.error("GELF Error: {0}".format(error_.__str__()))
 
+                try:
+                    if opts.google_secops:
+                        events = google_secops_client.save_smtp_tls_report_to_google_secops(report)
+                        for event in events:
+                            print(event)
+                except Exception as error_:
+                    logger.error("Google SecOps Error: {0}".format(error_.__str__()))
+
                 try:
                     if opts.webhook_smtp_tls_url:
                         indent_value = 2 if opts.prettify_json else None
@@ -722,6 +747,12 @@ def _main():
         gelf_host=None,
         gelf_port=None,
         gelf_mode=None,
+        google_secops=False,
+        google_secops_include_ruf_payload=False,
+        google_secops_ruf_payload_max_bytes=4096,
+        google_secops_static_observer_name=None,
+        google_secops_static_observer_vendor="parsedmarc",
+        google_secops_static_environment=None,
         webhook_aggregate_url=None,
         webhook_forensic_url=None,
         webhook_smtp_tls_url=None,
@@ -1301,6 +1332,30 @@ def _main():
                 logger.critical("mode setting missing from the gelf config section")
                 exit(-1)
 
+        if "google_secops" in config.sections():
+            google_secops_config = config["google_secops"]
+            opts.google_secops = True
+            if "include_ruf_payload" in google_secops_config:
+                opts.google_secops_include_ruf_payload = bool(
+                    google_secops_config.getboolean("include_ruf_payload")
+                )
+            if "ruf_payload_max_bytes" in google_secops_config:
+                opts.google_secops_ruf_payload_max_bytes = google_secops_config.getint(
+                    "ruf_payload_max_bytes"
+                )
+            if "static_observer_name" in google_secops_config:
+                opts.google_secops_static_observer_name = google_secops_config[
+                    "static_observer_name"
+                ]
+            if "static_observer_vendor" in google_secops_config:
+                opts.google_secops_static_observer_vendor = google_secops_config[
+                    "static_observer_vendor"
+                ]
+            if "static_environment" in google_secops_config:
+                opts.google_secops_static_environment = google_secops_config[
+                    "static_environment"
+                ]
+
         if "webhook" in config.sections():
             webhook_config = config["webhook"]
             if "aggregate_url" in webhook_config:
@@ -1479,6 +1534,18 @@ def _main():
         except Exception as error_:
             logger.error("GELF Error: {0}".format(error_.__str__()))
 
+    if opts.google_secops:
+        try:
+            google_secops_client = google_secops.GoogleSecOpsClient(
+                include_ruf_payload=opts.google_secops_include_ruf_payload,
+                ruf_payload_max_bytes=opts.google_secops_ruf_payload_max_bytes,
+                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,
+            )
+        except Exception as error_:
+            logger.error("Google SecOps Error: {0}".format(error_.__str__()))
+
     if (
         opts.webhook_aggregate_url
         or opts.webhook_forensic_url

parsedmarc/google_secops.py

@@ -0,0 +1,516 @@
+# -*- coding: utf-8 -*-
+
+"""Google SecOps (Chronicle) output module for parsedmarc"""
+
+from __future__ import annotations
+
+import json
+from datetime import datetime, timezone
+from typing import Any, Optional
+
+from parsedmarc.log import logger
+from parsedmarc.utils import human_timestamp_to_datetime
+
+
+class GoogleSecOpsClient:
+    """A client for Google SecOps (Chronicle) UDM output"""
+
+    def __init__(
+        self,
+        include_ruf_payload: bool = False,
+        ruf_payload_max_bytes: int = 4096,
+        static_observer_name: Optional[str] = None,
+        static_observer_vendor: str = "parsedmarc",
+        static_environment: Optional[str] = None,
+    ):
+        """
+        Initializes the GoogleSecOpsClient
+
+        Args:
+            include_ruf_payload: Include RUF message payload in output
+            ruf_payload_max_bytes: Maximum bytes of RUF payload to include
+            static_observer_name: Static observer name for telemetry
+            static_observer_vendor: Static observer vendor (default: parsedmarc)
+            static_environment: Static environment (prod/dev/custom string)
+        """
+        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
+
+    def _get_severity(self, disposition: str, spf_aligned: bool, dkim_aligned: bool) -> str:
+        """
+        Derive severity from DMARC disposition and alignment
+
+        Args:
+            disposition: DMARC policy disposition
+            spf_aligned: Whether SPF is aligned
+            dkim_aligned: Whether DKIM is aligned
+
+        Returns:
+            Severity level: HIGH, MEDIUM, or LOW
+        """
+        if disposition == "reject":
+            return "HIGH"
+        elif disposition == "quarantine" and not (spf_aligned or dkim_aligned):
+            return "MEDIUM"
+        else:
+            return "LOW"
+
+    def _get_description(
+        self,
+        dmarc_pass: bool,
+        spf_result: Optional[str],
+        dkim_result: Optional[str],
+        spf_aligned: bool,
+        dkim_aligned: bool,
+        disposition: str,
+    ) -> str:
+        """
+        Generate description for the event
+
+        Args:
+            dmarc_pass: Whether DMARC passed
+            spf_result: SPF result
+            dkim_result: DKIM result
+            spf_aligned: Whether SPF is aligned
+            dkim_aligned: Whether DKIM is aligned
+            disposition: DMARC disposition
+
+        Returns:
+            Human-readable description
+        """
+        parts = []
+        
+        if dmarc_pass:
+            parts.append("DMARC pass")
+        else:
+            parts.append("DMARC fail")
+        
+        if spf_result:
+            parts.append(f"SPF={spf_result}")
+        if dkim_result:
+            parts.append(f"DKIM={dkim_result}")
+        
+        if spf_aligned:
+            parts.append("SPF aligned")
+        elif spf_result:
+            parts.append("SPF not aligned")
+        
+        if dkim_aligned:
+            parts.append("DKIM aligned")
+        elif dkim_result:
+            parts.append("DKIM not aligned")
+        
+        parts.append(f"disposition={disposition}")
+        
+        return "; ".join(parts)
+
+    def _format_timestamp(self, timestamp_str: str) -> str:
+        """
+        Convert parsedmarc timestamp to RFC 3339 format
+
+        Args:
+            timestamp_str: Timestamp string from parsedmarc
+
+        Returns:
+            RFC 3339 formatted timestamp
+        """
+        try:
+            dt = human_timestamp_to_datetime(timestamp_str)
+            # Ensure timezone-aware datetime
+            if dt.tzinfo is None:
+                dt = dt.replace(tzinfo=timezone.utc)
+            return dt.isoformat()
+        except Exception:
+            # Fallback to current time if parsing fails
+            return datetime.now(timezone.utc).isoformat()
+
+    def save_aggregate_report_to_google_secops(
+        self, aggregate_report: dict[str, Any]
+    ) -> list[str]:
+        """
+        Convert aggregate DMARC report to Google SecOps UDM format (NDJSON)
+
+        Args:
+            aggregate_report: Aggregate report dictionary from parsedmarc
+
+        Returns:
+            List of NDJSON event strings
+        """
+        logger.debug("Converting aggregate report to Google SecOps UDM format")
+        events = []
+
+        try:
+            report_metadata = aggregate_report["report_metadata"]
+            policy_published = aggregate_report["policy_published"]
+
+            for record in aggregate_report["records"]:
+                # Extract values
+                source_ip = record["source"]["ip_address"]
+                source_country = record["source"].get("country")
+                source_reverse_dns = record["source"].get("reverse_dns")
+                source_base_domain = record["source"].get("base_domain")
+                source_name = record["source"].get("name")
+                
+                header_from = record["identifiers"]["header_from"]
+                envelope_from = record["identifiers"]["envelope_from"]
+                
+                disposition = record["policy_evaluated"]["disposition"]
+                spf_aligned = record["alignment"]["spf"]
+                dkim_aligned = record["alignment"]["dkim"]
+                dmarc_pass = record["alignment"]["dmarc"]
+                
+                count = record["count"]
+                interval_begin = record["interval_begin"]
+                interval_end = record["interval_end"]
+                
+                # Get auth results
+                spf_results = record["auth_results"].get("spf", [])
+                dkim_results = record["auth_results"].get("dkim", [])
+                
+                spf_result = spf_results[0]["result"] if spf_results else None
+                dkim_result = dkim_results[0]["result"] if dkim_results else None
+                
+                # Build UDM event
+                event: dict[str, Any] = {
+                    "event_type": "DMARC_AGGREGATE",
+                    "metadata": {
+                        "event_timestamp": self._format_timestamp(interval_begin),
+                        "event_type": "GENERIC_EVENT",
+                        "product_name": "parsedmarc",
+                        "vendor_name": self.static_observer_vendor,
+                    },
+                    "principal": {
+                        "ip": [source_ip],
+                    },
+                    "target": {
+                        "domain": {
+                            "name": header_from,
+                        }
+                    },
+                    "security_result": [
+                        {
+                            "severity": self._get_severity(
+                                disposition, spf_aligned, dkim_aligned
+                            ),
+                            "description": self._get_description(
+                                dmarc_pass,
+                                spf_result,
+                                dkim_result,
+                                spf_aligned,
+                                dkim_aligned,
+                                disposition,
+                            ),
+                            "detection_fields": [
+                                {"key": "dmarc_disposition", "value": disposition},
+                                {"key": "dmarc_policy", "value": policy_published["p"]},
+                                {"key": "dmarc_pass", "value": str(dmarc_pass).lower()},
+                                {"key": "spf_aligned", "value": str(spf_aligned).lower()},
+                                {"key": "dkim_aligned", "value": str(dkim_aligned).lower()},
+                            ],
+                        }
+                    ],
+                    "additional": {
+                        "fields": [
+                            {"key": "report_org", "value": report_metadata["org_name"]},
+                            {"key": "report_id", "value": report_metadata["report_id"]},
+                            {"key": "report_begin", "value": report_metadata["begin_date"]},
+                            {"key": "report_end", "value": report_metadata["end_date"]},
+                            {"key": "message_count", "value": str(count)},
+                            {"key": "interval_begin", "value": interval_begin},
+                            {"key": "interval_end", "value": interval_end},
+                            {"key": "envelope_from", "value": envelope_from},
+                        ]
+                    },
+                }
+                
+                # Add optional fields
+                if source_country:
+                    event["principal"]["location"] = {"country_or_region": source_country}
+                
+                if source_reverse_dns:
+                    event["principal"]["hostname"] = source_reverse_dns
+                
+                if source_base_domain:
+                    event["additional"]["fields"].append(
+                        {"key": "source_base_domain", "value": source_base_domain}
+                    )
+                
+                if source_name:
+                    event["additional"]["fields"].append(
+                        {"key": "source_name", "value": source_name}
+                    )
+                
+                if self.static_observer_name:
+                    event["metadata"]["product_deployment_id"] = self.static_observer_name
+                
+                if self.static_environment:
+                    event["additional"]["fields"].append(
+                        {"key": "environment", "value": self.static_environment}
+                    )
+                
+                # Add SPF results
+                if spf_results:
+                    for idx, spf in enumerate(spf_results):
+                        event["additional"]["fields"].append(
+                            {"key": f"spf_{idx}_domain", "value": spf.get("domain", "")}
+                        )
+                        event["additional"]["fields"].append(
+                            {"key": f"spf_{idx}_result", "value": spf.get("result", "")}
+                        )
+                
+                # Add DKIM results
+                if dkim_results:
+                    for idx, dkim in enumerate(dkim_results):
+                        event["additional"]["fields"].append(
+                            {"key": f"dkim_{idx}_domain", "value": dkim.get("domain", "")}
+                        )
+                        event["additional"]["fields"].append(
+                            {"key": f"dkim_{idx}_result", "value": dkim.get("result", "")}
+                        )
+                
+                events.append(json.dumps(event, ensure_ascii=False))
+        
+        except Exception as e:
+            logger.error(f"Error converting aggregate report to Google SecOps format: {e}")
+            # Generate error event
+            error_event: dict[str, Any] = {
+                "event_type": "DMARC_PARSE_ERROR",
+                "metadata": {
+                    "event_timestamp": datetime.now(timezone.utc).isoformat(),
+                    "event_type": "GENERIC_EVENT",
+                    "product_name": "parsedmarc",
+                    "vendor_name": self.static_observer_vendor,
+                },
+                "security_result": [
+                    {
+                        "severity": "ERROR",
+                        "description": f"Failed to parse DMARC aggregate report: {str(e)}",
+                    }
+                ],
+            }
+            events.append(json.dumps(error_event, ensure_ascii=False))
+        
+        return events
+
+    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)
+
+        Args:
+            forensic_report: Forensic report dictionary from parsedmarc
+
+        Returns:
+            List of NDJSON event strings
+        """
+        logger.debug("Converting forensic report to Google SecOps UDM format")
+        events = []
+
+        try:
+            source_ip = forensic_report["source"]["ip_address"]
+            source_country = forensic_report["source"].get("country")
+            source_reverse_dns = forensic_report["source"].get("reverse_dns")
+            
+            reported_domain = forensic_report["reported_domain"]
+            arrival_date = forensic_report["arrival_date_utc"]
+            auth_failure = forensic_report.get("auth_failure", [])
+            
+            # Determine severity - forensic reports indicate failures
+            severity = "MEDIUM"
+            if "dmarc" in auth_failure:
+                severity = "MEDIUM"
+            
+            # Build description
+            auth_failure_str = ", ".join(auth_failure) if auth_failure else "unknown"
+            description = f"DMARC forensic report: authentication failure ({auth_failure_str})"
+            
+            # Build UDM event
+            event: dict[str, Any] = {
+                "event_type": "DMARC_FORENSIC",
+                "metadata": {
+                    "event_timestamp": self._format_timestamp(arrival_date),
+                    "event_type": "GENERIC_EVENT",
+                    "product_name": "parsedmarc",
+                    "vendor_name": self.static_observer_vendor,
+                },
+                "principal": {
+                    "ip": [source_ip],
+                },
+                "target": {
+                    "domain": {
+                        "name": reported_domain,
+                    }
+                },
+                "security_result": [
+                    {
+                        "severity": severity,
+                        "description": description,
+                        "detection_fields": [
+                            {"key": "auth_failure", "value": auth_failure_str},
+                        ],
+                    }
+                ],
+                "additional": {
+                    "fields": [
+                        {"key": "arrival_date", "value": arrival_date},
+                        {"key": "feedback_type", "value": forensic_report.get("feedback_type", "")},
+                    ]
+                },
+            }
+            
+            # Add optional fields
+            if source_country:
+                event["principal"]["location"] = {"country_or_region": source_country}
+            
+            if source_reverse_dns:
+                event["principal"]["hostname"] = source_reverse_dns
+            
+            if forensic_report.get("message_id"):
+                event["additional"]["fields"].append(
+                    {"key": "message_id", "value": forensic_report["message_id"]}
+                )
+            
+            if forensic_report.get("authentication_results"):
+                event["additional"]["fields"].append(
+                    {"key": "authentication_results", "value": forensic_report["authentication_results"]}
+                )
+            
+            if forensic_report.get("delivery_result"):
+                event["additional"]["fields"].append(
+                    {"key": "delivery_result", "value": forensic_report["delivery_result"]}
+                )
+            
+            if self.static_observer_name:
+                event["metadata"]["product_deployment_id"] = self.static_observer_name
+            
+            if self.static_environment:
+                event["additional"]["fields"].append(
+                    {"key": "environment", "value": self.static_environment}
+                )
+            
+            # Add payload excerpt if enabled
+            if self.include_ruf_payload and forensic_report.get("sample"):
+                sample = forensic_report["sample"]
+                if len(sample) > self.ruf_payload_max_bytes:
+                    sample = sample[:self.ruf_payload_max_bytes] + "... [truncated]"
+                event["additional"]["fields"].append(
+                    {"key": "message_sample", "value": sample}
+                )
+            
+            events.append(json.dumps(event, ensure_ascii=False))
+        
+        except Exception as e:
+            logger.error(f"Error converting forensic report to Google SecOps format: {e}")
+            # Generate error event
+            error_event: dict[str, Any] = {
+                "event_type": "DMARC_PARSE_ERROR",
+                "metadata": {
+                    "event_timestamp": datetime.now(timezone.utc).isoformat(),
+                    "event_type": "GENERIC_EVENT",
+                    "product_name": "parsedmarc",
+                    "vendor_name": self.static_observer_vendor,
+                },
+                "security_result": [
+                    {
+                        "severity": "ERROR",
+                        "description": f"Failed to parse DMARC forensic report: {str(e)}",
+                    }
+                ],
+            }
+            events.append(json.dumps(error_event, ensure_ascii=False))
+        
+        return events
+
+    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)
+
+        Args:
+            smtp_tls_report: SMTP TLS report dictionary from parsedmarc
+
+        Returns:
+            List of NDJSON event strings
+        """
+        logger.debug("Converting SMTP TLS report to Google SecOps UDM format")
+        events = []
+
+        try:
+            organization_name = smtp_tls_report.get("organization_name", "")
+            begin_date = smtp_tls_report["begin_date"]
+            end_date = smtp_tls_report["end_date"]
+            
+            for policy in smtp_tls_report.get("policies", []):
+                policy_domain = policy["policy"]["policy_domain"]
+                
+                for failure in policy.get("failure_details", []):
+                    # Build UDM event for each failure
+                    event: dict[str, Any] = {
+                        "event_type": "SMTP_TLS_REPORT",
+                        "metadata": {
+                            "event_timestamp": self._format_timestamp(begin_date),
+                            "event_type": "GENERIC_EVENT",
+                            "product_name": "parsedmarc",
+                            "vendor_name": self.static_observer_vendor,
+                        },
+                        "target": {
+                            "domain": {
+                                "name": policy_domain,
+                            }
+                        },
+                        "security_result": [
+                            {
+                                "severity": "LOW",
+                                "description": f"SMTP TLS failure: {failure.get('result_type', 'unknown')}",
+                            }
+                        ],
+                        "additional": {
+                            "fields": [
+                                {"key": "organization_name", "value": organization_name},
+                                {"key": "report_begin", "value": begin_date},
+                                {"key": "report_end", "value": end_date},
+                                {"key": "result_type", "value": failure.get("result_type", "")},
+                                {"key": "failed_session_count", "value": str(failure.get("failed_session_count", 0))},
+                            ]
+                        },
+                    }
+                    
+                    if failure.get("sending_mta_ip"):
+                        event["principal"] = {"ip": [failure["sending_mta_ip"]]}
+                    
+                    if self.static_observer_name:
+                        event["metadata"]["product_deployment_id"] = self.static_observer_name
+                    
+                    if self.static_environment:
+                        event["additional"]["fields"].append(
+                            {"key": "environment", "value": self.static_environment}
+                        )
+                    
+                    events.append(json.dumps(event, ensure_ascii=False))
+        
+        except Exception as e:
+            logger.error(f"Error converting SMTP TLS report to Google SecOps format: {e}")
+            # Generate error event
+            error_event: dict[str, Any] = {
+                "event_type": "DMARC_PARSE_ERROR",
+                "metadata": {
+                    "event_timestamp": datetime.now(timezone.utc).isoformat(),
+                    "event_type": "GENERIC_EVENT",
+                    "product_name": "parsedmarc",
+                    "vendor_name": self.static_observer_vendor,
+                },
+                "security_result": [
+                    {
+                        "severity": "ERROR",
+                        "description": f"Failed to parse SMTP TLS report: {str(e)}",
+                    }
+                ],
+            }
+            events.append(json.dumps(error_event, ensure_ascii=False))
+        
+        return events

tests.py

@@ -3,6 +3,7 @@
 
 from __future__ import absolute_import, print_function, unicode_literals
 
+import json
 import os
 import unittest
 from glob import glob
@@ -156,6 +157,127 @@ class Test(unittest.TestCase):
             parsedmarc.parsed_smtp_tls_reports_to_csv(parsed_report)
             print("Passed!")
 
+    def testGoogleSecOpsAggregateReport(self):
+        """Test Google SecOps aggregate report conversion"""
+        print()
+        from parsedmarc.google_secops import GoogleSecOpsClient
+        
+        client = GoogleSecOpsClient()
+        sample_path = "samples/aggregate/example.net!example.com!1529366400!1529452799.xml"
+        print("Testing Google SecOps aggregate conversion for {0}: ".format(sample_path), end="")
+        
+        parsed_file = parsedmarc.parse_report_file(sample_path, always_use_local_files=True)
+        parsed_report = parsed_file["report"]
+        
+        events = client.save_aggregate_report_to_google_secops(parsed_report)
+        
+        # Verify we got events
+        assert len(events) > 0, "Expected at least one event"
+        
+        # Verify each event is valid JSON
+        for event in events:
+            event_dict = json.loads(event)
+            assert "event_type" in event_dict
+            assert event_dict["event_type"] == "DMARC_AGGREGATE"
+            assert "metadata" in event_dict
+            assert "principal" in event_dict
+            assert "target" in event_dict
+            assert "security_result" in event_dict
+        
+        print("Passed!")
+
+    def testGoogleSecOpsForensicReport(self):
+        """Test Google SecOps forensic report conversion"""
+        print()
+        from parsedmarc.google_secops import GoogleSecOpsClient
+        
+        # Test without payload
+        client = GoogleSecOpsClient(include_ruf_payload=False)
+        sample_path = "samples/forensic/dmarc_ruf_report_linkedin.eml"
+        print("Testing Google SecOps forensic conversion (no payload) for {0}: ".format(sample_path), end="")
+        
+        parsed_file = parsedmarc.parse_report_file(sample_path)
+        parsed_report = parsed_file["report"]
+        
+        events = client.save_forensic_report_to_google_secops(parsed_report)
+        
+        # Verify we got events
+        assert len(events) > 0, "Expected at least one event"
+        
+        # Verify each event is valid JSON
+        for event in events:
+            event_dict = json.loads(event)
+            assert "event_type" in event_dict
+            assert event_dict["event_type"] == "DMARC_FORENSIC"
+            
+            # Verify no payload in additional fields
+            if "additional" in event_dict and "fields" in event_dict["additional"]:
+                for field in event_dict["additional"]["fields"]:
+                    assert field["key"] != "message_sample", "Payload should not be included when disabled"
+        
+        print("Passed!")
+        
+        # Test with payload
+        client_with_payload = GoogleSecOpsClient(
+            include_ruf_payload=True,
+            ruf_payload_max_bytes=100
+        )
+        print("Testing Google SecOps forensic conversion (with payload) for {0}: ".format(sample_path), end="")
+        
+        events_with_payload = client_with_payload.save_forensic_report_to_google_secops(parsed_report)
+        
+        # Verify we got events
+        assert len(events_with_payload) > 0, "Expected at least one event"
+        
+        # Verify payload is included
+        for event in events_with_payload:
+            event_dict = json.loads(event)
+            
+            # Check if message_sample is in additional fields
+            has_sample = False
+            if "additional" in event_dict and "fields" in event_dict["additional"]:
+                for field in event_dict["additional"]["fields"]:
+                    if field["key"] == "message_sample":
+                        has_sample = True
+                        # Verify truncation (100 bytes + "... [truncated]" suffix)
+                        assert len(field["value"]) <= 120, f"Payload should be truncated, got {len(field['value'])} bytes"
+                        break
+            
+            assert has_sample, "Payload should be included when enabled"
+        
+        print("Passed!")
+
+    def testGoogleSecOpsConfiguration(self):
+        """Test Google SecOps client configuration"""
+        print()
+        from parsedmarc.google_secops import GoogleSecOpsClient
+        
+        print("Testing Google SecOps client configuration: ", end="")
+        
+        # Test default configuration
+        client1 = GoogleSecOpsClient()
+        assert client1.include_ruf_payload is False
+        assert client1.ruf_payload_max_bytes == 4096
+        assert client1.static_observer_vendor == "parsedmarc"
+        assert client1.static_observer_name is None
+        assert client1.static_environment is None
+        
+        # Test custom configuration
+        client2 = GoogleSecOpsClient(
+            include_ruf_payload=True,
+            ruf_payload_max_bytes=8192,
+            static_observer_name="test-observer",
+            static_observer_vendor="test-vendor",
+            static_environment="prod"
+        )
+        assert client2.include_ruf_payload is True
+        assert client2.ruf_payload_max_bytes == 8192
+        assert client2.static_observer_name == "test-observer"
+        assert client2.static_observer_vendor == "test-vendor"
+        assert client2.static_environment == "prod"
+        
+        print("Passed!")
+
 
 if __name__ == "__main__":
     unittest.main(verbosity=2)