9.2.0

### Added

- OpenSearch AWS SigV4 authentication support (PR #673)
- IMAP move/delete compatibility fallbacks (PR #671)
- `fail_on_output_error` CLI option for sink failures (PR #672)
- Gmail service account auth mode for non-interactive runs (PR #676)
- Microsoft Graph certificate authentication support (PRs #692 and #693)
- Microsoft Graph well-known folder fallback for root listing failures (PR #618 and #684 close #609)

### Fixed

- Pass mailbox since filter through `watch_inbox` callback (PR #670 closes issue #581)
- `parsedmarc.mail.gmail.GmailConnection.delete_message` now properly calls the Gmail API (PR #668)
- Avoid extra mailbox fetch in batch and test mode (PR #691 closes #533)

.github/ISSUE_TEMPLATE/bug_report.yml

@@ -0,0 +1,72 @@
+name: Bug report
+description: Report a reproducible parsedmarc bug
+title: "[Bug]: "
+labels:
+  - bug
+body:
+  - type: input
+    id: version
+    attributes:
+      label: parsedmarc version
+      description: Include the parsedmarc version or commit if known.
+      placeholder: 9.x.x
+    validations:
+      required: true
+  - type: dropdown
+    id: input_backend
+    attributes:
+      label: Input backend
+      description: Which input path or mailbox backend is involved?
+      options:
+        - IMAP
+        - MS Graph
+        - Gmail API
+        - Maildir
+        - mbox
+        - Local file / direct parse
+        - Other
+    validations:
+      required: true
+  - type: textarea
+    id: environment
+    attributes:
+      label: Environment
+      description: Runtime, container image, OS, Python version, or deployment details.
+      placeholder: Docker on Debian, Python 3.12, parsedmarc installed from PyPI
+    validations:
+      required: true
+  - type: textarea
+    id: config
+    attributes:
+      label: Sanitized config
+      description: Include the relevant config fragment with secrets removed.
+      render: ini
+  - type: textarea
+    id: steps
+    attributes:
+      label: Steps to reproduce
+      description: Describe the smallest reproducible sequence you can.
+      placeholder: |
+        1. Configure parsedmarc with ...
+        2. Run ...
+        3. Observe ...
+    validations:
+      required: true
+  - type: textarea
+    id: expected_actual
+    attributes:
+      label: Expected vs actual behavior
+      description: What did you expect, and what happened instead?
+    validations:
+      required: true
+  - type: textarea
+    id: logs
+    attributes:
+      label: Logs or traceback
+      description: Paste sanitized logs or a traceback if available.
+      render: text
+  - type: textarea
+    id: samples
+    attributes:
+      label: Sample report availability
+      description: If you can share a sanitized sample report or message, note that here.

.github/ISSUE_TEMPLATE/config.yml

@@ -0,0 +1,5 @@
+blank_issues_enabled: true
+contact_links:
+  - name: Security issue
+    url: https://github.com/domainaware/parsedmarc/security/policy
+    about: Please use the security policy and avoid filing public issues for undisclosed vulnerabilities.

.github/ISSUE_TEMPLATE/feature_request.yml

@@ -0,0 +1,30 @@
+name: Feature request
+description: Suggest a new feature or behavior change
+title: "[Feature]: "
+labels:
+  - enhancement
+body:
+  - type: textarea
+    id: problem
+    attributes:
+      label: Problem statement
+      description: What workflow or limitation are you trying to solve?
+    validations:
+      required: true
+  - type: textarea
+    id: proposal
+    attributes:
+      label: Proposed behavior
+      description: Describe the feature or behavior you want.
+    validations:
+      required: true
+  - type: textarea
+    id: alternatives
+    attributes:
+      label: Alternatives considered
+      description: Describe workarounds or alternative approaches you considered.
+  - type: textarea
+    id: impact
+    attributes:
+      label: Compatibility or operational impact
+      description: Note config, output, performance, or deployment implications if relevant.

.github/pull_request_template.md

@@ -0,0 +1,24 @@
+## Summary
+
+- 
+
+## Why
+
+- 
+
+## Testing
+
+- 
+
+## Backward Compatibility / Risk
+
+- 
+
+## Related Issue
+
+- Closes #
+
+## Checklist
+
+- [ ] Tests added or updated if behavior changed
+- [ ] Docs updated if config or user-facing behavior changed

.github/workflows/python-tests.yml

@@ -10,7 +10,32 @@ on:
     branches: [ master ]
 
 jobs:
-  build:
+  lint-docs-build:
+    runs-on: ubuntu-latest
+
+    steps:
+    - uses: actions/checkout@v5
+    - name: Set up Python
+      uses: actions/setup-python@v6
+      with:
+        python-version: "3.13"
+    - name: Install Python dependencies
+      run: |
+        python -m pip install --upgrade pip
+        pip install .[build]
+    - name: Check code style
+      run: |
+        ruff check .
+    - name: Test building documentation
+      run: |
+        cd docs
+        make html
+    - name: Test building packages
+      run: |
+        hatch build
+
+  test:
+    needs: lint-docs-build
     runs-on: ubuntu-latest
 
     services:
@@ -46,13 +71,6 @@ jobs:
       run: |
         python -m pip install --upgrade pip
         pip install .[build]
-    - name: Test building documentation
-      run: |
-        cd docs
-        make html
-    - name: Check code style
-      run: |
-        ruff check .
     - name: Run unit tests
       run: |
         pytest --cov --cov-report=xml tests.py
@@ -61,9 +79,6 @@ jobs:
         pip install -e .
         parsedmarc --debug -c ci.ini samples/aggregate/*
         parsedmarc --debug -c ci.ini samples/forensic/*
-    - name: Test building packages
-      run: |
-        hatch build
     - name: Upload coverage to Codecov
       uses: codecov/codecov-action@v5
       with:

CHANGELOG.md

@@ -1,5 +1,22 @@
 # Changelog
 
+## 9.2.0
+
+### Added
+
+- OpenSearch AWS SigV4 authentication support (PR #673)
+- IMAP move/delete compatibility fallbacks (PR #671)
+- `fail_on_output_error` CLI option for sink failures (PR #672)
+- Gmail service account auth mode for non-interactive runs (PR #676)
+- Microsoft Graph certificate authentication support (PRs #692 and #693)
+- Microsoft Graph well-known folder fallback for root listing failures (PR #618 and #684 close #609)
+
+### Fixed
+
+- Pass mailbox since filter through `watch_inbox` callback (PR #670 closes issue #581)
+- `parsedmarc.mail.gmail.GmailConnection.delete_message` now properly calls the Gmail API (PR #668)
+- Avoid extra mailbox fetch in batch and test mode (PR #691 closes #533)
+
 ## 9.1.2
 
 ### Fixes

CONTRIBUTING.md

@@ -0,0 +1,78 @@
+# Contributing
+
+Thanks for contributing to parsedmarc.
+
+## Local setup
+
+Use a virtual environment for local development.
+
+```bash
+python3 -m venv .venv
+. .venv/bin/activate
+python -m pip install --upgrade pip
+pip install .[build]
+```
+
+## Before opening a pull request
+
+Run the checks that match your change:
+
+```bash
+ruff check .
+pytest --cov --cov-report=xml tests.py
+```
+
+If you changed documentation:
+
+```bash
+cd docs
+make html
+```
+
+If you changed CLI behavior or parsing logic, it is also useful to exercise the
+sample reports:
+
+```bash
+parsedmarc --debug -c ci.ini samples/aggregate/*
+parsedmarc --debug -c ci.ini samples/forensic/*
+```
+
+To skip DNS lookups during tests, set:
+
+```bash
+GITHUB_ACTIONS=true
+```
+
+## Pull request guidelines
+
+- Keep pull requests small and focused. Separate bug fixes, docs updates, and
+  repo-maintenance changes where practical.
+- Add or update tests when behavior changes.
+- Update docs when configuration or user-facing behavior changes.
+- Include a short summary, the reason for the change, and the testing you ran.
+- Link the related issue when there is one.
+
+## Branch maintenance
+
+Upstream `master` may move quickly. Before asking for review or after another PR
+lands, rebase your branch onto the current upstream branch and force-push with
+lease if needed:
+
+```bash
+git fetch upstream
+git rebase upstream/master
+git push --force-with-lease
+```
+
+## CI and coverage
+
+GitHub Actions is the source of truth for linting, docs, and test status.
+
+Codecov patch coverage is usually the most relevant signal for small PRs. Project
+coverage can be noisier when the base comparison is stale, so interpret it in
+the context of the actual diff.
+
+## Questions
+
+Use GitHub issues for bugs and feature requests. If you are not sure whether a
+change is wanted, opening an issue first is usually the safest path.

SECURITY.md

@@ -0,0 +1,29 @@
+# Security Policy
+
+## Reporting a vulnerability
+
+Please do not open a public GitHub issue for an undisclosed security
+vulnerability. Use GitHub private vulnerability reporting in the Security tab of this project instead.
+
+When reporting a vulnerability, include:
+
+- the affected parsedmarc version or commit
+- the component or integration involved
+- clear reproduction details if available
+- potential impact
+- any suggested mitigation or workaround
+
+## Supported versions
+
+Security fixes will be applied to the latest released version and
+the current `master` branch.
+
+Older versions will not receive backported fixes.
+
+## Disclosure process
+
+After a report is received, maintainers can validate the issue, assess impact,
+and coordinate a fix before public disclosure.
+
+Please avoid publishing proof-of-concept details until maintainers have had a
+reasonable opportunity to investigate and release a fix or mitigation.

codecov.yml

@@ -0,0 +1,11 @@
+codecov:
+  require_ci_to_pass: true
+
+coverage:
+  status:
+    project:
+      default:
+        informational: true
+    patch:
+      default:
+        informational: false

docs/source/usage.md

@@ -146,6 +146,9 @@ The full set of configuration options are:
   - `dns_timeout` - float: DNS timeout period
   - `debug` - bool: Print debugging messages
   - `silent` - bool: Only print errors (Default: `True`)
+  - `fail_on_output_error` - bool: Exit with a non-zero status code if
+      any configured output destination fails while saving/publishing
+      reports (Default: `False`)
   - `log_file` - str: Write log messages to a file at this path
   - `n_procs` - int: Number of process to run in parallel when
       parsing in CLI mode (Default: `1`)
@@ -200,7 +203,7 @@ The full set of configuration options are:
   - `password` - str: The IMAP password
 - `msgraph`
   - `auth_method` - str: Authentication method, valid types are
-      `UsernamePassword`, `DeviceCode`, or `ClientSecret`
+      `UsernamePassword`, `DeviceCode`, `ClientSecret`, or `Certificate`
       (Default: `UsernamePassword`).
   - `user` - str: The M365 user, required when the auth method is
       UsernamePassword
@@ -208,6 +211,11 @@ The full set of configuration options are:
       method is UsernamePassword
   - `client_id` - str: The app registration's client ID
   - `client_secret` - str: The app registration's secret
+  - `certificate_path` - str: Path to a PEM or PKCS12 certificate
+      including the private key. Required when the auth method is
+      `Certificate`
+  - `certificate_password` - str: Optional password for the
+      certificate file when using `Certificate` auth
   - `tenant_id` - str: The Azure AD tenant ID. This is required
       for all auth methods except UsernamePassword.
   - `mailbox` - str: The mailbox name. This defaults to the
@@ -245,6 +253,9 @@ The full set of configuration options are:
     -Description "Restrict access to dmarc reports mailbox."
     ```
 
+    The same application permission and mailbox scoping guidance
+    applies to the `Certificate` auth method.
+
     :::
 - `elasticsearch`
   - `hosts` - str: A comma separated list of hostnames and ports
@@ -281,6 +292,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)
@@ -394,10 +409,19 @@ The full set of configuration options are:
       credentials, None to disable (Default: `None`)
   - `token_file` - str: Path to save the token file
       (Default: `.token`)
+  - `auth_mode` - str: Authentication mode, `installed_app` (default)
+      or `service_account`
+  - `service_account_user` - str: Delegated mailbox user for Gmail
+      service account auth (required for domain-wide delegation). Also
+      accepted as `delegated_user` for backward compatibility.
 
     :::{note}
     credentials_file and token_file can be got with [quickstart](https://developers.google.com/gmail/api/quickstart/python).Please change the scope to `https://www.googleapis.com/auth/gmail.modify`.
     :::
+    :::{note}
+    When `auth_mode = service_account`, `credentials_file` must point to a
+    Google service account key JSON file, and `token_file` is not used.
+    :::
   - `include_spam_trash` - bool: Include messages in Spam and
       Trash when searching reports (Default: `False`)
   - `scopes` - str: Comma separated list of scopes to use when
@@ -502,6 +526,33 @@ PUT _cluster/settings
 Increasing this value increases resource usage.
 :::
 
+## Performance tuning
+
+For large mailbox imports or backfills, parsedmarc can consume a noticeable amount
+of memory, especially when it runs on the same host as Elasticsearch or
+OpenSearch. The following settings can reduce peak memory usage and make long
+imports more predictable:
+
+- Reduce `mailbox.batch_size` to smaller values such as `100-500` instead of
+  processing a very large message set at once. Smaller batches trade throughput
+  for lower peak memory use and less sink pressure.
+- Keep `n_procs` low for mailbox-heavy runs. In practice, `1-2` workers is often
+  a safer starting point for large backfills than aggressive parallelism.
+- Use `mailbox.since` to process reports in smaller time windows such as `1d`,
+  `7d`, or another interval that fits the backlog. This makes it easier to catch
+  up incrementally instead of loading an entire mailbox history in one run.
+- Set `strip_attachment_payloads = True` when forensic reports contain large
+  attachments and you do not need to retain the raw payloads in the parsed
+  output.
+- Prefer running parsedmarc separately from Elasticsearch or OpenSearch, or
+  reserve enough RAM for both services if they must share a host.
+- For very large imports, prefer incremental supervised runs, such as a
+  scheduler or systemd service, over infrequent massive backfills.
+
+These are operational tuning recommendations rather than hard requirements, but
+they are often enough to avoid memory pressure and reduce failures during
+high-volume mailbox processing.
+
 ## Multi-tenant support
 
 Starting in `8.19.0`, ParseDMARC provides multi-tenant support by placing data into separate OpenSearch or Elasticsearch index prefixes. To set this up, create a YAML file that is formatted where each key is a tenant name, and the value is a list of domains related to that tenant, not including subdomains, like this:

parsedmarc/__init__.py

@@ -962,10 +962,12 @@ def extract_report(content: Union[bytes, str, BinaryIO]) -> str:
     return report
 
 
-def extract_report_from_file_path(file_path: str):
+def extract_report_from_file_path(
+    file_path: Union[str, bytes, os.PathLike[str], os.PathLike[bytes]],
+) -> str:
     """Extracts report from a file at the given file_path"""
     try:
-        with open(file_path, "rb") as report_file:
+        with open(os.fspath(file_path), "rb") as report_file:
             return extract_report(report_file.read())
     except FileNotFoundError:
         raise ParserError("File was not found")
@@ -1660,7 +1662,7 @@ def parse_report_email(
 
 
 def parse_report_file(
-    input_: Union[bytes, str, BinaryIO],
+    input_: Union[bytes, str, os.PathLike[str], os.PathLike[bytes], BinaryIO],
     *,
     nameservers: Optional[list[str]] = None,
     dns_timeout: float = 2.0,
@@ -1677,7 +1679,8 @@ def parse_report_file(
     file-like object. or bytes
 
     Args:
-        input_ (str | bytes | BinaryIO): A path to a file, a file like object, or bytes
+        input_ (str | os.PathLike | bytes | BinaryIO): A path to a file,
+            a file-like object, or bytes
         nameservers (list): A list of one or more nameservers to use
             (Cloudflare's public DNS resolvers by default)
         dns_timeout (float): Sets the DNS timeout in seconds
@@ -1694,9 +1697,10 @@ def parse_report_file(
         dict: The parsed DMARC report
     """
     file_object: BinaryIO
-    if isinstance(input_, str):
-        logger.debug("Parsing {0}".format(input_))
-        file_object = open(input_, "rb")
+    if isinstance(input_, (str, os.PathLike)):
+        file_path = os.fspath(input_)
+        logger.debug("Parsing {0}".format(file_path))
+        file_object = open(file_path, "rb")
     elif isinstance(input_, (bytes, bytearray, memoryview)):
         file_object = BytesIO(bytes(input_))
     else:
@@ -2137,14 +2141,17 @@ def get_dmarc_reports_from_mailbox(
         "smtp_tls_reports": smtp_tls_reports,
     }
 
-    if current_time:
-        total_messages = len(
-            connection.fetch_messages(reports_folder, since=current_time)
-        )
+    if not test and not batch_size:
+        if current_time:
+            total_messages = len(
+                connection.fetch_messages(reports_folder, since=current_time)
+            )
+        else:
+            total_messages = len(connection.fetch_messages(reports_folder))
     else:
-        total_messages = len(connection.fetch_messages(reports_folder))
+        total_messages = 0
 
-    if not test and not batch_size and total_messages > 0:
+    if total_messages > 0:
         # Process emails that came in during the last run
         results = get_dmarc_reports_from_mailbox(
             connection=connection,
@@ -2186,6 +2193,7 @@ def watch_inbox(
     dns_timeout: float = 6.0,
     strip_attachment_payloads: bool = False,
     batch_size: int = 10,
+    since: Optional[Union[datetime, date, str]] = None,
     normalize_timespan_threshold_hours: float = 24,
 ):
     """
@@ -2212,6 +2220,7 @@ def watch_inbox(
         strip_attachment_payloads (bool): Replace attachment payloads in
             forensic report samples with None
         batch_size (int): Number of messages to read and process before saving
+        since: Search for messages since certain time
         normalize_timespan_threshold_hours (float): Normalize timespans beyond this
     """
 
@@ -2231,6 +2240,7 @@ def watch_inbox(
             dns_timeout=dns_timeout,
             strip_attachment_payloads=strip_attachment_payloads,
             batch_size=batch_size,
+            since=since,
             create_folders=False,
             normalize_timespan_threshold_hours=normalize_timespan_threshold_hours,
         )

parsedmarc/cli.py

@@ -194,6 +194,13 @@ def _main():
         return None
 
     def process_reports(reports_):
+        output_errors = []
+
+        def log_output_error(destination, error):
+            message = f"{destination} Error: {error}"
+            logger.error(message)
+            output_errors.append(message)
+
         indent_value = 2 if opts.prettify_json else None
         output_str = "{0}\n".format(
             json.dumps(reports_, ensure_ascii=False, indent=indent_value)
@@ -230,11 +237,9 @@ def _main():
                 except elastic.AlreadySaved as warning:
                     logger.warning(warning.__str__())
                 except elastic.ElasticsearchError as error_:
-                    logger.error("Elasticsearch Error: {0}".format(error_.__str__()))
+                    log_output_error("Elasticsearch", error_.__str__())
                 except Exception as error_:
-                    logger.error(
-                        "Elasticsearch exception error: {}".format(error_.__str__())
-                    )
+                    log_output_error("Elasticsearch exception", error_.__str__())
 
                 try:
                     if opts.opensearch_hosts:
@@ -252,11 +257,9 @@ def _main():
                 except opensearch.AlreadySaved as warning:
                     logger.warning(warning.__str__())
                 except opensearch.OpenSearchError as error_:
-                    logger.error("OpenSearch Error: {0}".format(error_.__str__()))
+                    log_output_error("OpenSearch", error_.__str__())
                 except Exception as error_:
-                    logger.error(
-                        "OpenSearch exception error: {}".format(error_.__str__())
-                    )
+                    log_output_error("OpenSearch exception", error_.__str__())
 
                 try:
                     if opts.kafka_hosts:
@@ -264,25 +267,25 @@ def _main():
                             report, kafka_aggregate_topic
                         )
                 except Exception as error_:
-                    logger.error("Kafka Error: {0}".format(error_.__str__()))
+                    log_output_error("Kafka", error_.__str__())
 
                 try:
                     if opts.s3_bucket:
                         s3_client.save_aggregate_report_to_s3(report)
                 except Exception as error_:
-                    logger.error("S3 Error: {0}".format(error_.__str__()))
+                    log_output_error("S3", error_.__str__())
 
                 try:
                     if opts.syslog_server:
                         syslog_client.save_aggregate_report_to_syslog(report)
                 except Exception as error_:
-                    logger.error("Syslog Error: {0}".format(error_.__str__()))
+                    log_output_error("Syslog", error_.__str__())
 
                 try:
                     if opts.gelf_host:
                         gelf_client.save_aggregate_report_to_gelf(report)
                 except Exception as error_:
-                    logger.error("GELF Error: {0}".format(error_.__str__()))
+                    log_output_error("GELF", error_.__str__())
 
                 try:
                     if opts.webhook_aggregate_url:
@@ -291,7 +294,7 @@ def _main():
                             json.dumps(report, ensure_ascii=False, indent=indent_value)
                         )
                 except Exception as error_:
-                    logger.error("Webhook Error: {0}".format(error_.__str__()))
+                    log_output_error("Webhook", error_.__str__())
 
             if opts.hec:
                 try:
@@ -299,7 +302,7 @@ def _main():
                     if len(aggregate_reports_) > 0:
                         hec_client.save_aggregate_reports_to_splunk(aggregate_reports_)
                 except splunk.SplunkError as e:
-                    logger.error("Splunk HEC error: {0}".format(e.__str__()))
+                    log_output_error("Splunk HEC", e.__str__())
 
         if opts.save_forensic:
             for report in reports_["forensic_reports"]:
@@ -319,9 +322,9 @@ def _main():
                 except elastic.AlreadySaved as warning:
                     logger.warning(warning.__str__())
                 except elastic.ElasticsearchError as error_:
-                    logger.error("Elasticsearch Error: {0}".format(error_.__str__()))
+                    log_output_error("Elasticsearch", error_.__str__())
                 except InvalidDMARCReport as error_:
-                    logger.error(error_.__str__())
+                    log_output_error("Invalid DMARC report", error_.__str__())
 
                 try:
                     shards = opts.opensearch_number_of_shards
@@ -339,9 +342,9 @@ def _main():
                 except opensearch.AlreadySaved as warning:
                     logger.warning(warning.__str__())
                 except opensearch.OpenSearchError as error_:
-                    logger.error("OpenSearch Error: {0}".format(error_.__str__()))
+                    log_output_error("OpenSearch", error_.__str__())
                 except InvalidDMARCReport as error_:
-                    logger.error(error_.__str__())
+                    log_output_error("Invalid DMARC report", error_.__str__())
 
                 try:
                     if opts.kafka_hosts:
@@ -349,25 +352,25 @@ def _main():
                             report, kafka_forensic_topic
                         )
                 except Exception as error_:
-                    logger.error("Kafka Error: {0}".format(error_.__str__()))
+                    log_output_error("Kafka", error_.__str__())
 
                 try:
                     if opts.s3_bucket:
                         s3_client.save_forensic_report_to_s3(report)
                 except Exception as error_:
-                    logger.error("S3 Error: {0}".format(error_.__str__()))
+                    log_output_error("S3", error_.__str__())
 
                 try:
                     if opts.syslog_server:
                         syslog_client.save_forensic_report_to_syslog(report)
                 except Exception as error_:
-                    logger.error("Syslog Error: {0}".format(error_.__str__()))
+                    log_output_error("Syslog", error_.__str__())
 
                 try:
                     if opts.gelf_host:
                         gelf_client.save_forensic_report_to_gelf(report)
                 except Exception as error_:
-                    logger.error("GELF Error: {0}".format(error_.__str__()))
+                    log_output_error("GELF", error_.__str__())
 
                 try:
                     if opts.webhook_forensic_url:
@@ -376,7 +379,7 @@ def _main():
                             json.dumps(report, ensure_ascii=False, indent=indent_value)
                         )
                 except Exception as error_:
-                    logger.error("Webhook Error: {0}".format(error_.__str__()))
+                    log_output_error("Webhook", error_.__str__())
 
             if opts.hec:
                 try:
@@ -384,7 +387,7 @@ def _main():
                     if len(forensic_reports_) > 0:
                         hec_client.save_forensic_reports_to_splunk(forensic_reports_)
                 except splunk.SplunkError as e:
-                    logger.error("Splunk HEC error: {0}".format(e.__str__()))
+                    log_output_error("Splunk HEC", e.__str__())
 
         if opts.save_smtp_tls:
             for report in reports_["smtp_tls_reports"]:
@@ -404,9 +407,9 @@ def _main():
                 except elastic.AlreadySaved as warning:
                     logger.warning(warning.__str__())
                 except elastic.ElasticsearchError as error_:
-                    logger.error("Elasticsearch Error: {0}".format(error_.__str__()))
+                    log_output_error("Elasticsearch", error_.__str__())
                 except InvalidDMARCReport as error_:
-                    logger.error(error_.__str__())
+                    log_output_error("Invalid DMARC report", error_.__str__())
 
                 try:
                     shards = opts.opensearch_number_of_shards
@@ -424,9 +427,9 @@ def _main():
                 except opensearch.AlreadySaved as warning:
                     logger.warning(warning.__str__())
                 except opensearch.OpenSearchError as error_:
-                    logger.error("OpenSearch Error: {0}".format(error_.__str__()))
+                    log_output_error("OpenSearch", error_.__str__())
                 except InvalidDMARCReport as error_:
-                    logger.error(error_.__str__())
+                    log_output_error("Invalid DMARC report", error_.__str__())
 
                 try:
                     if opts.kafka_hosts:
@@ -434,25 +437,25 @@ def _main():
                             smtp_tls_reports, kafka_smtp_tls_topic
                         )
                 except Exception as error_:
-                    logger.error("Kafka Error: {0}".format(error_.__str__()))
+                    log_output_error("Kafka", error_.__str__())
 
                 try:
                     if opts.s3_bucket:
                         s3_client.save_smtp_tls_report_to_s3(report)
                 except Exception as error_:
-                    logger.error("S3 Error: {0}".format(error_.__str__()))
+                    log_output_error("S3", error_.__str__())
 
                 try:
                     if opts.syslog_server:
                         syslog_client.save_smtp_tls_report_to_syslog(report)
                 except Exception as error_:
-                    logger.error("Syslog Error: {0}".format(error_.__str__()))
+                    log_output_error("Syslog", error_.__str__())
 
                 try:
                     if opts.gelf_host:
                         gelf_client.save_smtp_tls_report_to_gelf(report)
                 except Exception as error_:
-                    logger.error("GELF Error: {0}".format(error_.__str__()))
+                    log_output_error("GELF", error_.__str__())
 
                 try:
                     if opts.webhook_smtp_tls_url:
@@ -461,7 +464,7 @@ def _main():
                             json.dumps(report, ensure_ascii=False, indent=indent_value)
                         )
                 except Exception as error_:
-                    logger.error("Webhook Error: {0}".format(error_.__str__()))
+                    log_output_error("Webhook", error_.__str__())
 
             if opts.hec:
                 try:
@@ -469,7 +472,7 @@ def _main():
                     if len(smtp_tls_reports_) > 0:
                         hec_client.save_smtp_tls_reports_to_splunk(smtp_tls_reports_)
                 except splunk.SplunkError as e:
-                    logger.error("Splunk HEC error: {0}".format(e.__str__()))
+                    log_output_error("Splunk HEC", e.__str__())
 
         if opts.la_dce:
             try:
@@ -490,14 +493,16 @@ def _main():
                     opts.save_smtp_tls,
                 )
             except loganalytics.LogAnalyticsException as e:
-                logger.error("Log Analytics error: {0}".format(e.__str__()))
+                log_output_error("Log Analytics", e.__str__())
             except Exception as e:
-                logger.error(
-                    "Unknown error occurred"
-                    + " during the publishing"
-                    + " to Log Analytics: "
-                    + e.__str__()
+                log_output_error("Log Analytics", f"Unknown publishing error: {e}")
+
+        if opts.fail_on_output_error and output_errors:
+            raise ParserError(
+                "Output destination failures detected: {0}".format(
+                    " | ".join(output_errors)
                 )
+            )
 
     arg_parser = ArgumentParser(description="Parses DMARC reports")
     arg_parser.add_argument(
@@ -639,6 +644,8 @@ def _main():
         graph_password=None,
         graph_client_id=None,
         graph_client_secret=None,
+        graph_certificate_path=None,
+        graph_certificate_password=None,
         graph_tenant_id=None,
         graph_mailbox=None,
         graph_allow_unencrypted_storage=False,
@@ -671,6 +678,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,
@@ -710,6 +720,8 @@ def _main():
         gmail_api_paginate_messages=True,
         gmail_api_scopes=[],
         gmail_api_oauth2_port=8080,
+        gmail_api_auth_mode="installed_app",
+        gmail_api_service_account_user=None,
         maildir_path=None,
         maildir_create=False,
         log_file=args.log_file,
@@ -734,6 +746,7 @@ def _main():
         webhook_smtp_tls_url=None,
         webhook_timeout=60,
         normalize_timespan_threshold_hours=24.0,
+        fail_on_output_error=False,
     )
     args = arg_parser.parse_args()
 
@@ -816,6 +829,10 @@ def _main():
                 opts.silent = bool(general_config.getboolean("silent"))
             if "warnings" in general_config:
                 opts.warnings = bool(general_config.getboolean("warnings"))
+            if "fail_on_output_error" in general_config:
+                opts.fail_on_output_error = bool(
+                    general_config.getboolean("fail_on_output_error")
+                )
             if "log_file" in general_config:
                 opts.log_file = general_config["log_file"]
             if "n_procs" in general_config:
@@ -997,6 +1014,19 @@ def _main():
                     )
                     exit(-1)
 
+            if opts.graph_auth_method == AuthMethod.Certificate.name:
+                if "certificate_path" in graph_config:
+                    opts.graph_certificate_path = graph_config["certificate_path"]
+                else:
+                    logger.critical(
+                        "certificate_path setting missing from the msgraph config section"
+                    )
+                    exit(-1)
+                if "certificate_password" in graph_config:
+                    opts.graph_certificate_password = graph_config[
+                        "certificate_password"
+                    ]
+
             if "client_id" in graph_config:
                 opts.graph_client_id = graph_config["client_id"]
             else:
@@ -1102,6 +1132,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"]
@@ -1287,6 +1327,16 @@ def _main():
                 opts.gmail_api_oauth2_port = gmail_api_config.getint(
                     "oauth2_port", 8080
                 )
+            if "auth_mode" in gmail_api_config:
+                opts.gmail_api_auth_mode = gmail_api_config.get("auth_mode").strip()
+            if "service_account_user" in gmail_api_config:
+                opts.gmail_api_service_account_user = gmail_api_config.get(
+                    "service_account_user"
+                ).strip()
+            elif "delegated_user" in gmail_api_config:
+                opts.gmail_api_service_account_user = gmail_api_config.get(
+                    "delegated_user"
+                ).strip()
 
         if "maildir" in config.sections():
             maildir_api_config = config["maildir"]
@@ -1438,6 +1488,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],
@@ -1710,6 +1763,8 @@ def _main():
                 tenant_id=opts.graph_tenant_id,
                 client_id=opts.graph_client_id,
                 client_secret=opts.graph_client_secret,
+                certificate_path=opts.graph_certificate_path,
+                certificate_password=opts.graph_certificate_password,
                 username=opts.graph_user,
                 password=opts.graph_password,
                 token_file=opts.graph_token_file,
@@ -1741,6 +1796,8 @@ def _main():
                 paginate_messages=opts.gmail_api_paginate_messages,
                 reports_folder=opts.mailbox_reports_folder,
                 oauth2_port=opts.gmail_api_oauth2_port,
+                auth_mode=opts.gmail_api_auth_mode,
+                service_account_user=opts.gmail_api_service_account_user,
             )
 
         except Exception:
@@ -1804,7 +1861,11 @@ def _main():
         "smtp_tls_reports": smtp_tls_reports,
     }
 
-    process_reports(parsing_results)
+    try:
+        process_reports(parsing_results)
+    except ParserError as error:
+        logger.error(error.__str__())
+        exit(1)
 
     if opts.smtp_host:
         try:
@@ -1849,6 +1910,7 @@ def _main():
                 dns_timeout=opts.dns_timeout,
                 strip_attachment_payloads=opts.strip_attachment_payloads,
                 batch_size=mailbox_batch_size_value,
+                since=opts.mailbox_since,
                 ip_db_path=opts.ip_db_path,
                 always_use_local_files=opts.always_use_local_files,
                 reverse_dns_map_path=opts.reverse_dns_map_path,
@@ -1859,6 +1921,9 @@ def _main():
         except FileExistsError as error:
             logger.error("{0}".format(error.__str__()))
             exit(1)
+        except ParserError as error:
+            logger.error(error.__str__())
+            exit(1)
 
 
 if __name__ == "__main__":

parsedmarc/constants.py

@@ -1,3 +1,3 @@
-__version__ = "9.1.2"
+__version__ = "9.2.0"
 
 USER_AGENT = f"parsedmarc/{__version__}"

parsedmarc/mail/gmail.py

@@ -10,6 +10,7 @@ from typing import List
 
 from google.auth.transport.requests import Request
 from google.oauth2.credentials import Credentials
+from google.oauth2 import service_account
 from google_auth_oauthlib.flow import InstalledAppFlow
 from googleapiclient.discovery import build
 from googleapiclient.errors import HttpError
@@ -18,7 +19,29 @@ from parsedmarc.log import logger
 from parsedmarc.mail.mailbox_connection import MailboxConnection
 
 
-def _get_creds(token_file, credentials_file, scopes, oauth2_port):
+def _get_creds(
+    token_file,
+    credentials_file,
+    scopes,
+    oauth2_port,
+    auth_mode="installed_app",
+    service_account_user=None,
+):
+    normalized_auth_mode = (auth_mode or "installed_app").strip().lower()
+    if normalized_auth_mode == "service_account":
+        creds = service_account.Credentials.from_service_account_file(
+            credentials_file,
+            scopes=scopes,
+        )
+        if service_account_user:
+            creds = creds.with_subject(service_account_user)
+        return creds
+    if normalized_auth_mode != "installed_app":
+        raise ValueError(
+            f"Unsupported Gmail auth_mode '{auth_mode}'. "
+            "Expected 'installed_app' or 'service_account'."
+        )
+
     creds = None
 
     if Path(token_file).exists():
@@ -47,8 +70,17 @@ class GmailConnection(MailboxConnection):
         reports_folder: str,
         oauth2_port: int,
         paginate_messages: bool,
+        auth_mode: str = "installed_app",
+        service_account_user: str | None = None,
     ):
-        creds = _get_creds(token_file, credentials_file, scopes, oauth2_port)
+        creds = _get_creds(
+            token_file,
+            credentials_file,
+            scopes,
+            oauth2_port,
+            auth_mode=auth_mode,
+            service_account_user=service_account_user,
+        )
         self.service = build("gmail", "v1", credentials=creds)
         self.include_spam_trash = include_spam_trash
         self.reports_label_id = self._find_label_id_for_label(reports_folder)
@@ -126,7 +158,7 @@ class GmailConnection(MailboxConnection):
         return urlsafe_b64decode(msg["raw"]).decode(errors="replace")
 
     def delete_message(self, message_id: str):
-        self.service.users().messages().delete(userId="me", id=message_id)
+        self.service.users().messages().delete(userId="me", id=message_id).execute()
 
     def move_message(self, message_id: str, folder_name: str):
         label_id = self._find_label_id_for_label(folder_name)

parsedmarc/mail/graph.py

@@ -12,19 +12,25 @@ from azure.identity import (
     UsernamePasswordCredential,
     DeviceCodeCredential,
     ClientSecretCredential,
+    CertificateCredential,
     TokenCachePersistenceOptions,
     AuthenticationRecord,
 )
 from msgraph.core import GraphClient
+from requests.exceptions import RequestException
 
 from parsedmarc.log import logger
 from parsedmarc.mail.mailbox_connection import MailboxConnection
 
+GRAPH_REQUEST_RETRY_ATTEMPTS = 3
+GRAPH_REQUEST_RETRY_DELAY_SECONDS = 5
+
 
 class AuthMethod(Enum):
     DeviceCode = 1
     UsernamePassword = 2
     ClientSecret = 3
+    Certificate = 4
 
 
 def _get_cache_args(token_path: Path, allow_unencrypted_storage):
@@ -83,30 +89,55 @@ def _generate_credential(auth_method: str, token_path: Path, **kwargs):
             tenant_id=kwargs["tenant_id"],
             client_secret=kwargs["client_secret"],
         )
+    elif auth_method == AuthMethod.Certificate.name:
+        cert_path = kwargs.get("certificate_path")
+        if not cert_path:
+            raise ValueError(
+                "certificate_path is required when auth_method is 'Certificate'"
+            )
+        credential = CertificateCredential(
+            client_id=kwargs["client_id"],
+            tenant_id=kwargs["tenant_id"],
+            certificate_path=cert_path,
+            password=kwargs.get("certificate_password"),
+        )
     else:
         raise RuntimeError(f"Auth method {auth_method} not found")
     return credential
 
 
 class MSGraphConnection(MailboxConnection):
+    _WELL_KNOWN_FOLDERS = {
+        "inbox": "inbox",
+        "archive": "archive",
+        "drafts": "drafts",
+        "sentitems": "sentitems",
+        "deleteditems": "deleteditems",
+        "junkemail": "junkemail",
+    }
+
     def __init__(
         self,
         auth_method: str,
         mailbox: str,
         graph_url: str,
         client_id: str,
-        client_secret: str,
-        username: str,
-        password: str,
+        client_secret: Optional[str],
+        username: Optional[str],
+        password: Optional[str],
         tenant_id: str,
         token_file: str,
         allow_unencrypted_storage: bool,
+        certificate_path: Optional[str] = None,
+        certificate_password: Optional[Union[str, bytes]] = None,
     ):
         token_path = Path(token_file)
         credential = _generate_credential(
             auth_method,
             client_id=client_id,
             client_secret=client_secret,
+            certificate_path=certificate_path,
+            certificate_password=certificate_password,
             username=username,
             password=password,
             tenant_id=tenant_id,
@@ -117,10 +148,10 @@ class MSGraphConnection(MailboxConnection):
             "credential": credential,
             "cloud": graph_url,
         }
-        if not isinstance(credential, ClientSecretCredential):
+        if not isinstance(credential, (ClientSecretCredential, CertificateCredential)):
             scopes = ["Mail.ReadWrite"]
             # Detect if mailbox is shared
-            if mailbox and username != mailbox:
+            if mailbox and username and username != mailbox:
                 scopes = ["Mail.ReadWrite.Shared"]
             auth_record = credential.authenticate(scopes=scopes)
             _cache_auth_record(auth_record, token_path)
@@ -129,6 +160,23 @@ class MSGraphConnection(MailboxConnection):
         self._client = GraphClient(**client_params)
         self.mailbox_name = mailbox
 
+    def _request_with_retries(self, method_name: str, *args, **kwargs):
+        for attempt in range(1, GRAPH_REQUEST_RETRY_ATTEMPTS + 1):
+            try:
+                return getattr(self._client, method_name)(*args, **kwargs)
+            except RequestException as error:
+                if attempt == GRAPH_REQUEST_RETRY_ATTEMPTS:
+                    raise
+                logger.warning(
+                    "Transient MS Graph %s error on attempt %s/%s: %s",
+                    method_name.upper(),
+                    attempt,
+                    GRAPH_REQUEST_RETRY_ATTEMPTS,
+                    error,
+                )
+                sleep(GRAPH_REQUEST_RETRY_DELAY_SECONDS)
+        raise RuntimeError("no retry attempts configured")
+
     def create_folder(self, folder_name: str):
         sub_url = ""
         path_parts = folder_name.split("/")
@@ -143,7 +191,7 @@ class MSGraphConnection(MailboxConnection):
 
         request_body = {"displayName": folder_name}
         request_url = f"/users/{self.mailbox_name}/mailFolders{sub_url}"
-        resp = self._client.post(request_url, json=request_body)
+        resp = self._request_with_retries("post", request_url, json=request_body)
         if resp.status_code == 409:
             logger.debug(f"Folder {folder_name} already exists, skipping creation")
         elif resp.status_code == 201:
@@ -173,7 +221,7 @@ class MSGraphConnection(MailboxConnection):
             params["$top"] = batch_size
         else:
             params["$top"] = 100
-        result = self._client.get(url, params=params)
+        result = self._request_with_retries("get", url, params=params)
         if result.status_code != 200:
             raise RuntimeError(f"Failed to fetch messages {result.text}")
         messages = result.json()["value"]
@@ -181,7 +229,7 @@ class MSGraphConnection(MailboxConnection):
         while "@odata.nextLink" in result.json() and (
             since is not None or (batch_size == 0 or batch_size - len(messages) > 0)
         ):
-            result = self._client.get(result.json()["@odata.nextLink"])
+            result = self._request_with_retries("get", result.json()["@odata.nextLink"])
             if result.status_code != 200:
                 raise RuntimeError(f"Failed to fetch messages {result.text}")
             messages.extend(result.json()["value"])
@@ -190,7 +238,7 @@ class MSGraphConnection(MailboxConnection):
     def mark_message_read(self, message_id: str):
         """Marks a message as read"""
         url = f"/users/{self.mailbox_name}/messages/{message_id}"
-        resp = self._client.patch(url, json={"isRead": "true"})
+        resp = self._request_with_retries("patch", url, json={"isRead": "true"})
         if resp.status_code != 200:
             raise RuntimeWarning(
                 f"Failed to mark message read{resp.status_code}: {resp.json()}"
@@ -198,7 +246,7 @@ class MSGraphConnection(MailboxConnection):
 
     def fetch_message(self, message_id: str, **kwargs):
         url = f"/users/{self.mailbox_name}/messages/{message_id}/$value"
-        result = self._client.get(url)
+        result = self._request_with_retries("get", url)
         if result.status_code != 200:
             raise RuntimeWarning(
                 f"Failed to fetch message{result.status_code}: {result.json()}"
@@ -210,7 +258,7 @@ class MSGraphConnection(MailboxConnection):
 
     def delete_message(self, message_id: str):
         url = f"/users/{self.mailbox_name}/messages/{message_id}"
-        resp = self._client.delete(url)
+        resp = self._request_with_retries("delete", url)
         if resp.status_code != 204:
             raise RuntimeWarning(
                 f"Failed to delete message {resp.status_code}: {resp.json()}"
@@ -220,7 +268,7 @@ class MSGraphConnection(MailboxConnection):
         folder_id = self._find_folder_id_from_folder_path(folder_name)
         request_body = {"destinationId": folder_id}
         url = f"/users/{self.mailbox_name}/messages/{message_id}/move"
-        resp = self._client.post(url, json=request_body)
+        resp = self._request_with_retries("post", url, json=request_body)
         if resp.status_code != 201:
             raise RuntimeWarning(
                 f"Failed to move message {resp.status_code}: {resp.json()}"
@@ -248,6 +296,19 @@ class MSGraphConnection(MailboxConnection):
         else:
             return self._find_folder_id_with_parent(folder_name, None)
 
+    def _get_well_known_folder_id(self, folder_name: str) -> Optional[str]:
+        folder_key = folder_name.lower().replace(" ", "").replace("-", "")
+        alias = self._WELL_KNOWN_FOLDERS.get(folder_key)
+        if alias is None:
+            return None
+
+        url = f"/users/{self.mailbox_name}/mailFolders/{alias}?$select=id,displayName"
+        folder_resp = self._request_with_retries("get", url)
+        if folder_resp.status_code != 200:
+            return None
+        payload = folder_resp.json()
+        return payload.get("id")
+
     def _find_folder_id_with_parent(
         self, folder_name: str, parent_folder_id: Optional[str]
     ):
@@ -256,8 +317,12 @@ class MSGraphConnection(MailboxConnection):
             sub_url = f"/{parent_folder_id}/childFolders"
         url = f"/users/{self.mailbox_name}/mailFolders{sub_url}"
         filter = f"?$filter=displayName eq '{folder_name}'"
-        folders_resp = self._client.get(url + filter)
+        folders_resp = self._request_with_retries("get", url + filter)
         if folders_resp.status_code != 200:
+            if parent_folder_id is None:
+                well_known_folder_id = self._get_well_known_folder_id(folder_name)
+                if well_known_folder_id:
+                    return well_known_folder_id
             raise RuntimeWarning(f"Failed to list folders.{folders_resp.json()}")
         folders: list = folders_resp.json()["value"]
         matched_folders = [

parsedmarc/mail/imap.py

@@ -55,10 +55,28 @@ class IMAPConnection(MailboxConnection):
         return cast(str, self._client.fetch_message(message_id, parse=False))
 
     def delete_message(self, message_id: int):
-        self._client.delete_messages([message_id])
+        try:
+            self._client.delete_messages([message_id])
+        except IMAPClientError as error:
+            logger.warning(
+                "IMAP delete fallback for message %s due to server error: %s",
+                message_id,
+                error,
+            )
+            self._client.add_flags([message_id], [r"\Deleted"], silent=True)
+            self._client.expunge()
 
     def move_message(self, message_id: int, folder_name: str):
-        self._client.move_messages([message_id], folder_name)
+        try:
+            self._client.move_messages([message_id], folder_name)
+        except IMAPClientError as error:
+            logger.warning(
+                "IMAP move fallback for message %s due to server error: %s",
+                message_id,
+                error,
+            )
+            self._client.copy([message_id], folder_name)
+            self.delete_message(message_id)
 
     def keepalive(self):
         self._client.noop()

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)