Add OpenSearch AWS SigV4 authentication support (#673)

* Add OpenSearch AWS SigV4 authentication support

* Increase SigV4 coverage for auth validation and CLI config wiring

* Update parsedmarc/opensearch.py

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

* Update docs/source/usage.md

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

---------

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

docs/source/usage.md

@@ -281,6 +281,10 @@ The full set of configuration options are:
   - `user` - str: Basic auth username
   - `password` - str: Basic auth password
   - `api_key` - str: API key
+  - `auth_type` - str: Authentication type: `basic` (default) or `awssigv4` (the key `authentication_type` is accepted as an alias for this option)
+  - `aws_region` - str: AWS region for SigV4 authentication
+    (required when `auth_type = awssigv4`)
+  - `aws_service` - str: AWS service for SigV4 signing (Default: `es`)
   - `ssl` - bool: Use an encrypted SSL/TLS connection
     (Default: `True`)
   - `timeout` - float: Timeout in seconds (Default: 60)

parsedmarc/cli.py

@@ -671,6 +671,9 @@ def _main():
         opensearch_username=None,
         opensearch_password=None,
         opensearch_api_key=None,
+        opensearch_auth_type="basic",
+        opensearch_aws_region=None,
+        opensearch_aws_service="es",
         kafka_hosts=None,
         kafka_username=None,
         kafka_password=None,
@@ -1104,6 +1107,16 @@ def _main():
             # Since 8.20
             if "api_key" in opensearch_config:
                 opts.opensearch_api_key = opensearch_config["api_key"]
+            if "auth_type" in opensearch_config:
+                opts.opensearch_auth_type = opensearch_config["auth_type"].strip().lower()
+            elif "authentication_type" in opensearch_config:
+                opts.opensearch_auth_type = (
+                    opensearch_config["authentication_type"].strip().lower()
+                )
+            if "aws_region" in opensearch_config:
+                opts.opensearch_aws_region = opensearch_config["aws_region"].strip()
+            if "aws_service" in opensearch_config:
+                opts.opensearch_aws_service = opensearch_config["aws_service"].strip()
 
         if "splunk_hec" in config.sections():
             hec_config = config["splunk_hec"]
@@ -1450,6 +1463,9 @@ def _main():
                     password=opts.opensearch_password,
                     api_key=opts.opensearch_api_key,
                     timeout=opensearch_timeout_value,
+                    auth_type=opts.opensearch_auth_type,
+                    aws_region=opts.opensearch_aws_region,
+                    aws_service=opts.opensearch_aws_service,
                 )
                 opensearch.migrate_indexes(
                     aggregate_indexes=[os_aggregate_index],

parsedmarc/opensearch.py

@@ -4,7 +4,9 @@ from __future__ import annotations
 
 from typing import Any, Optional, Union
 
+import boto3
 from opensearchpy import (
+    AWSV4SignerAuth,
     Boolean,
     Date,
     Document,
@@ -15,6 +17,7 @@ from opensearchpy import (
     Nested,
     Object,
     Q,
+    RequestsHttpConnection,
     Search,
     Text,
     connections,
@@ -272,6 +275,9 @@ def set_hosts(
     password: Optional[str] = None,
     api_key: Optional[str] = None,
     timeout: Optional[float] = 60.0,
+    auth_type: str = "basic",
+    aws_region: Optional[str] = None,
+    aws_service: str = "es",
 ):
     """
     Sets the OpenSearch hosts to use
@@ -284,6 +290,9 @@ def set_hosts(
         password (str): The password to use for authentication
         api_key (str): The Base64 encoded API key to use for authentication
         timeout (float): Timeout in seconds
+        auth_type (str): OpenSearch auth mode: basic (default) or awssigv4
+        aws_region (str): AWS region for SigV4 auth (required for awssigv4)
+        aws_service (str): AWS service for SigV4 signing (default: es)
     """
     if not isinstance(hosts, list):
         hosts = [hosts]
@@ -295,10 +304,32 @@ def set_hosts(
             conn_params["ca_certs"] = ssl_cert_path
         else:
             conn_params["verify_certs"] = False
-    if username and password:
-        conn_params["http_auth"] = username + ":" + password
-    if api_key:
-        conn_params["api_key"] = api_key
+    normalized_auth_type = (auth_type or "basic").strip().lower()
+    if normalized_auth_type == "awssigv4":
+        if not aws_region:
+            raise OpenSearchError(
+                "OpenSearch AWS SigV4 auth requires 'aws_region' to be set"
+            )
+        session = boto3.Session()
+        credentials = session.get_credentials()
+        if credentials is None:
+            raise OpenSearchError(
+                "Unable to load AWS credentials for OpenSearch SigV4 authentication"
+            )
+        conn_params["http_auth"] = AWSV4SignerAuth(
+            credentials, aws_region, aws_service
+        )
+        conn_params["connection_class"] = RequestsHttpConnection
+    elif normalized_auth_type == "basic":
+        if username and password:
+            conn_params["http_auth"] = username + ":" + password
+        if api_key:
+            conn_params["api_key"] = api_key
+    else:
+        raise OpenSearchError(
+            f"Unsupported OpenSearch auth_type '{auth_type}'. "
+            "Expected 'basic' or 'awssigv4'."
+        )
     connections.create_connection(**conn_params)
 
 

tests.py

@@ -30,6 +30,7 @@ from parsedmarc.mail.imap import IMAPConnection
 import parsedmarc.mail.gmail as gmail_module
 import parsedmarc.mail.graph as graph_module
 import parsedmarc.mail.imap as imap_module
+import parsedmarc.opensearch as opensearch_module
 import parsedmarc.utils
 
 # Detect if running in GitHub Actions to skip DNS lookups
@@ -185,6 +186,99 @@ class Test(unittest.TestCase):
             parsedmarc.parsed_smtp_tls_reports_to_csv(parsed_report)
             print("Passed!")
 
+    def testOpenSearchSigV4RequiresRegion(self):
+        with self.assertRaises(opensearch_module.OpenSearchError):
+            opensearch_module.set_hosts(
+                "https://example.org:9200",
+                auth_type="awssigv4",
+            )
+
+    def testOpenSearchSigV4ConfiguresConnectionClass(self):
+        fake_credentials = object()
+        with patch.object(opensearch_module.boto3, "Session") as session_cls:
+            session_cls.return_value.get_credentials.return_value = fake_credentials
+            with patch.object(
+                opensearch_module, "AWSV4SignerAuth", return_value="auth"
+            ) as signer:
+                with patch.object(
+                    opensearch_module.connections, "create_connection"
+                ) as create_connection:
+                    opensearch_module.set_hosts(
+                        "https://example.org:9200",
+                        use_ssl=True,
+                        auth_type="awssigv4",
+                        aws_region="eu-west-1",
+                    )
+        signer.assert_called_once_with(fake_credentials, "eu-west-1", "es")
+        create_connection.assert_called_once()
+        self.assertEqual(
+            create_connection.call_args.kwargs.get("connection_class"),
+            opensearch_module.RequestsHttpConnection,
+        )
+        self.assertEqual(create_connection.call_args.kwargs.get("http_auth"), "auth")
+
+    def testOpenSearchSigV4RejectsUnknownAuthType(self):
+        with self.assertRaises(opensearch_module.OpenSearchError):
+            opensearch_module.set_hosts(
+                "https://example.org:9200",
+                auth_type="kerberos",
+            )
+
+    def testOpenSearchSigV4RequiresAwsCredentials(self):
+        with patch.object(opensearch_module.boto3, "Session") as session_cls:
+            session_cls.return_value.get_credentials.return_value = None
+            with self.assertRaises(opensearch_module.OpenSearchError):
+                opensearch_module.set_hosts(
+                    "https://example.org:9200",
+                    auth_type="awssigv4",
+                    aws_region="eu-west-1",
+                )
+
+    @patch("parsedmarc.cli.opensearch.migrate_indexes")
+    @patch("parsedmarc.cli.opensearch.set_hosts")
+    @patch("parsedmarc.cli.get_dmarc_reports_from_mailbox")
+    @patch("parsedmarc.cli.IMAPConnection")
+    def testCliPassesOpenSearchSigV4Settings(
+        self,
+        mock_imap_connection,
+        mock_get_reports,
+        mock_set_hosts,
+        _mock_migrate_indexes,
+    ):
+        mock_imap_connection.return_value = object()
+        mock_get_reports.return_value = {
+            "aggregate_reports": [],
+            "forensic_reports": [],
+            "smtp_tls_reports": [],
+        }
+
+        config = """[general]
+save_aggregate = true
+silent = true
+
+[imap]
+host = imap.example.com
+user = test-user
+password = test-password
+
+[opensearch]
+hosts = localhost
+authentication_type = awssigv4
+aws_region = eu-west-1
+aws_service = aoss
+"""
+        with tempfile.NamedTemporaryFile("w", suffix=".ini", delete=False) as config_file:
+            config_file.write(config)
+            config_path = config_file.name
+        self.addCleanup(lambda: os.path.exists(config_path) and os.remove(config_path))
+
+        with patch.object(sys, "argv", ["parsedmarc", "-c", config_path]):
+            parsedmarc.cli._main()
+
+        self.assertEqual(mock_set_hosts.call_args.kwargs.get("auth_type"), "awssigv4")
+        self.assertEqual(mock_set_hosts.call_args.kwargs.get("aws_region"), "eu-west-1")
+        self.assertEqual(mock_set_hosts.call_args.kwargs.get("aws_service"), "aoss")
+
 
 class _FakeGraphResponse:
     def __init__(self, status_code, payload=None, text=""):