Fix duplicate detection for normalized aggregate reports in Elasticsearch/OpenSearch

Change date_begin/date_end queries from exact match to range queries
(gte/lte) so that previously saved normalized time buckets are correctly
detected as duplicates within the original report's date range.

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

.claude/settings.json

@@ -1,16 +0,0 @@
-{
-  "permissions": {
-    "allow": [
-      "Bash(python -c \"import py_compile; py_compile.compile\\(''parsedmarc/cli.py'', doraise=True\\)\")",
-      "Bash(ruff check:*)",
-      "Bash(ruff format:*)",
-      "Bash(GITHUB_ACTIONS=true pytest --cov tests.py)",
-      "Bash(ls tests*)",
-      "Bash(GITHUB_ACTIONS=true python -m pytest --cov tests.py -x)",
-      "Bash(GITHUB_ACTIONS=true python -m pytest tests.py -x -v)"
-    ],
-    "additionalDirectories": [
-      "/tmp"
-    ]
-  }
-}

.github/ISSUE_TEMPLATE/bug_report.yml

@@ -1,72 +0,0 @@
-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

@@ -1,5 +0,0 @@
-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

@@ -1,30 +0,0 @@
-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

@@ -1,24 +0,0 @@
-## 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,32 +10,7 @@ on:
     branches: [ master ]
 
 jobs:
-  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
+  build:
     runs-on: ubuntu-latest
 
     services:
@@ -71,6 +46,13 @@ 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
@@ -79,6 +61,9 @@ 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:

.gitignore

@@ -137,7 +137,7 @@ samples/private
 *.html
 *.sqlite-journal
 
-parsedmarc*.ini
+parsedmarc.ini
 scratch.py
 
 parsedmarc/resources/maps/base_reverse_dns.csv

CHANGELOG.md

@@ -1,52 +1,5 @@
 # Changelog
 
-## 9.3.0
-
-### Added
-
-- SIGHUP-based configuration reload for watch mode — update output
-  destinations, DNS/GeoIP settings, processing flags, and log level
-  without restarting the service or interrupting in-progress report
-  processing. Use `systemctl reload parsedmarc` when running under
-  systemd.
-- Extracted `_parse_config_file()` and `_init_output_clients()` from
-  `_main()` in `cli.py` to support config reload and reduce code
-  duplication.
-
-## 9.2.1
-
-### Added
-
-- Better checking of `msconfig` configuration (PR #695)
-
-### Changed
-
-- Updated `dbip-country-lite` database to version `2026-03`
-- DNS query error logging level from `warning` to `debug`
-
-## 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
-
-- Fix duplicate detection for normalized aggregate reports in Elasticsearch/OpenSearch (PR #666 fixes issue #665)
-
 ## 9.1.1
 
 ### Fixes

CONTRIBUTING.md

@@ -1,78 +0,0 @@
-# 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

@@ -1,29 +0,0 @@
-# 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

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

docker-compose.dashboard-dev.yml

@@ -1,45 +0,0 @@
-name: parsedmarc-dashboards
-
-include:
-  - docker-compose.yml
-
-services:
-  kibana:
-    image: docker.elastic.co/kibana/kibana:8.19.7
-    environment:
-      - ELASTICSEARCH_HOSTS=http://elasticsearch:9200
-    ports:
-      - "127.0.0.1:5601:5601"
-    depends_on:
-      elasticsearch:
-        condition: service_healthy
-
-  opensearch-dashboards:
-    image: opensearchproject/opensearch-dashboards:2
-    environment:
-      - OPENSEARCH_HOSTS=["https://opensearch:9200"]
-    ports:
-      - "127.0.0.1:5602:5601"
-    depends_on:
-      opensearch:
-        condition: service_healthy
-
-  grafana:
-    image: grafana/grafana:latest
-    environment:
-      - GF_SECURITY_ADMIN_PASSWORD=${GRAFANA_PASSWORD}
-      - GF_INSTALL_PLUGINS=grafana-piechart-panel,grafana-worldmap-panel
-    ports:
-      - "127.0.0.1:3000:3000"
-    depends_on:
-      elasticsearch:
-        condition: service_healthy
-
-  splunk:
-    image: splunk/splunk:latest
-    environment:
-      - SPLUNK_START_ARGS=--accept-license
-      - "SPLUNK_GENERAL_TERMS=--accept-sgt-current-at-splunk-com"
-      - SPLUNK_PASSWORD=${SPLUNK_PASSWORD}
-    ports:
-      - "127.0.0.1:8000:8000"

docker-compose.yml

@@ -48,7 +48,7 @@ services:
       test:
         [
           "CMD-SHELL",
-          "curl -sk -u admin:${OPENSEARCH_INITIAL_ADMIN_PASSWORD} -XGET https://localhost:9200/_cluster/health?pretty | grep status | grep -q '\\(green\\|yellow\\)'"
+          "curl -s -XGET http://localhost:9201/_cluster/health?pretty | grep status | grep -q '\\(green\\|yellow\\)'"
         ]
       interval: 10s
       timeout: 10s

docs/source/usage.md

@@ -146,9 +146,6 @@ 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`)
@@ -203,7 +200,7 @@ The full set of configuration options are:
   - `password` - str: The IMAP password
 - `msgraph`
   - `auth_method` - str: Authentication method, valid types are
-      `UsernamePassword`, `DeviceCode`, `ClientSecret`, or `Certificate`
+      `UsernamePassword`, `DeviceCode`, or `ClientSecret`
       (Default: `UsernamePassword`).
   - `user` - str: The M365 user, required when the auth method is
       UsernamePassword
@@ -211,11 +208,6 @@ 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
@@ -253,9 +245,6 @@ 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
@@ -292,10 +281,6 @@ 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)
@@ -404,25 +389,15 @@ The full set of configuration options are:
   retry_attempts = 3
   retry_delay = 5
   ```
-
 - `gmail_api`
   - `credentials_file` - str: Path to file containing the
       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
@@ -443,7 +418,7 @@ The full set of configuration options are:
   - `dcr_smtp_tls_stream` - str: The stream name for the SMTP TLS reports in the DCR
 
   :::{note}
-    Information regarding the setup of the Data Collection Rule can be found [in the Azure documentation](https://learn.microsoft.com/en-us/azure/azure-monitor/logs/tutorial-logs-ingestion-portal).
+    Information regarding the setup of the Data Collection Rule can be found [here](https://learn.microsoft.com/en-us/azure/azure-monitor/logs/tutorial-logs-ingestion-portal).
     :::
 - `gelf`
   - `host` - str: The GELF server name or IP address
@@ -527,33 +502,6 @@ 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:
@@ -603,7 +551,6 @@ After=network.target network-online.target elasticsearch.service
 
 [Service]
 ExecStart=/opt/parsedmarc/venv/bin/parsedmarc -c /etc/parsedmarc.ini
-ExecReload=/bin/kill -HUP $MAINPID
 User=parsedmarc
 Group=parsedmarc
 Restart=always
@@ -636,44 +583,6 @@ sudo service parsedmarc restart
 
 :::
 
-### Reloading configuration without restarting
-
-When running in watch mode, `parsedmarc` supports reloading its
-configuration file without restarting the service or interrupting
-report processing that is already in progress. Send a `SIGHUP` signal
-to the process, or use `systemctl reload` if the unit file includes
-the `ExecReload` line shown above:
-
-```bash
-sudo systemctl reload parsedmarc
-```
-
-The reload takes effect after the current batch of reports finishes
-processing and all output operations (Elasticsearch, Kafka, S3, etc.)
-for that batch have completed. The following settings are reloaded:
-
-- All output destinations (Elasticsearch, OpenSearch, Kafka, S3,
-  Splunk, syslog, GELF, webhooks, Log Analytics)
-- Multi-tenant index prefix domain map (`index_prefix_domain_map` —
-  the referenced YAML file is re-read on reload)
-- DNS and GeoIP settings (`nameservers`, `dns_timeout`, `ip_db_path`,
-  `offline`, etc.)
-- Processing flags (`strip_attachment_payloads`, `batch_size`,
-  `check_timeout`, etc.)
-- Log level (`debug`, `verbose`, `warnings`, `silent`)
-
-Mailbox connection settings (IMAP host/credentials, Microsoft Graph,
-Gmail API, Maildir path) are **not** reloaded — changing those still
-requires a full restart.
-
-If the new configuration file contains errors, the reload is aborted
-and the previous configuration remains active. Check the logs for
-details:
-
-```bash
-journalctl -u parsedmarc.service -r
-```
-
 To check the status of the service, run:
 
 ```bash

parsedmarc/__init__.py

@@ -962,12 +962,10 @@ def extract_report(content: Union[bytes, str, BinaryIO]) -> str:
     return report
 
 
-def extract_report_from_file_path(
-    file_path: Union[str, bytes, os.PathLike[str], os.PathLike[bytes]],
-) -> str:
+def extract_report_from_file_path(file_path: str):
     """Extracts report from a file at the given file_path"""
     try:
-        with open(os.fspath(file_path), "rb") as report_file:
+        with open(file_path, "rb") as report_file:
             return extract_report(report_file.read())
     except FileNotFoundError:
         raise ParserError("File was not found")
@@ -1662,7 +1660,7 @@ def parse_report_email(
 
 
 def parse_report_file(
-    input_: Union[bytes, str, os.PathLike[str], os.PathLike[bytes], BinaryIO],
+    input_: Union[bytes, str, BinaryIO],
     *,
     nameservers: Optional[list[str]] = None,
     dns_timeout: float = 2.0,
@@ -1679,8 +1677,7 @@ def parse_report_file(
     file-like object. or bytes
 
     Args:
-        input_ (str | os.PathLike | bytes | BinaryIO): A path to a file,
-            a file-like object, or bytes
+        input_ (str | 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
@@ -1697,10 +1694,9 @@ def parse_report_file(
         dict: The parsed DMARC report
     """
     file_object: BinaryIO
-    if isinstance(input_, (str, os.PathLike)):
-        file_path = os.fspath(input_)
-        logger.debug("Parsing {0}".format(file_path))
-        file_object = open(file_path, "rb")
+    if isinstance(input_, str):
+        logger.debug("Parsing {0}".format(input_))
+        file_object = open(input_, "rb")
     elif isinstance(input_, (bytes, bytearray, memoryview)):
         file_object = BytesIO(bytes(input_))
     else:
@@ -2141,17 +2137,14 @@ def get_dmarc_reports_from_mailbox(
         "smtp_tls_reports": smtp_tls_reports,
     }
 
-    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))
+    if current_time:
+        total_messages = len(
+            connection.fetch_messages(reports_folder, since=current_time)
+        )
     else:
-        total_messages = 0
+        total_messages = len(connection.fetch_messages(reports_folder))
 
-    if total_messages > 0:
+    if not test and not batch_size and total_messages > 0:
         # Process emails that came in during the last run
         results = get_dmarc_reports_from_mailbox(
             connection=connection,
@@ -2193,9 +2186,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,
-    should_reload: Optional[Callable] = None,
 ):
     """
     Watches the mailbox for new messages and
@@ -2221,10 +2212,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
-        should_reload: Optional callable that returns True when a config
-            reload has been requested (e.g. via SIGHUP)
     """
 
     def check_callback(connection):
@@ -2243,17 +2231,12 @@ 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,
         )
         callback(res)
 
-    mailbox_connection.watch(
-        check_callback=check_callback,
-        check_timeout=check_timeout,
-        should_reload=should_reload,
-    )
+    mailbox_connection.watch(check_callback=check_callback, check_timeout=check_timeout)
 
 
 def append_json(

parsedmarc/constants.py

@@ -1,3 +1,3 @@
-__version__ = "9.2.1"
+__version__ = "9.1.1"
 
 USER_AGENT = f"parsedmarc/{__version__}"

parsedmarc/gelf.py

@@ -69,8 +69,3 @@ class GelfClient(object):
         for row in rows:
             log_context_data.parsedmarc = row
             self.logger.info("parsedmarc smtptls report")
-
-    def close(self):
-        """Remove and close the GELF handler, releasing its connection."""
-        self.logger.removeHandler(self.handler)
-        self.handler.close()

parsedmarc/kafkaclient.py

@@ -62,10 +62,6 @@ class KafkaClient(object):
         except NoBrokersAvailable:
             raise KafkaError("No Kafka brokers available")
 
-    def close(self):
-        """Close the Kafka producer, releasing background threads and sockets."""
-        self.producer.close()
-
     @staticmethod
     def strip_metadata(report: dict[str, Any]):
         """

parsedmarc/mail/gmail.py

@@ -10,7 +10,6 @@ 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
@@ -19,29 +18,7 @@ from parsedmarc.log import logger
 from parsedmarc.mail.mailbox_connection import MailboxConnection
 
 
-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'."
-        )
-
+def _get_creds(token_file, credentials_file, scopes, oauth2_port):
     creds = None
 
     if Path(token_file).exists():
@@ -70,17 +47,8 @@ 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,
-            auth_mode=auth_mode,
-            service_account_user=service_account_user,
-        )
+        creds = _get_creds(token_file, credentials_file, scopes, oauth2_port)
         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)
@@ -158,7 +126,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).execute()
+        self.service.users().messages().delete(userId="me", id=message_id)
 
     def move_message(self, message_id: str, folder_name: str):
         label_id = self._find_label_id_for_label(folder_name)
@@ -175,13 +143,11 @@ class GmailConnection(MailboxConnection):
         # Not needed
         pass
 
-    def watch(self, check_callback, check_timeout, should_reload=None):
+    def watch(self, check_callback, check_timeout):
         """Checks the mailbox for new messages every n seconds"""
         while True:
             sleep(check_timeout)
             check_callback(self)
-            if should_reload and should_reload():
-                return
 
     @lru_cache(maxsize=10)
     def _find_label_id_for_label(self, label_name: str) -> str:

parsedmarc/mail/graph.py

@@ -12,25 +12,19 @@ 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):
@@ -89,55 +83,30 @@ 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: Optional[str],
-        username: Optional[str],
-        password: Optional[str],
+        client_secret: str,
+        username: str,
+        password: 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,
@@ -148,10 +117,10 @@ class MSGraphConnection(MailboxConnection):
             "credential": credential,
             "cloud": graph_url,
         }
-        if not isinstance(credential, (ClientSecretCredential, CertificateCredential)):
+        if not isinstance(credential, ClientSecretCredential):
             scopes = ["Mail.ReadWrite"]
             # Detect if mailbox is shared
-            if mailbox and username and username != mailbox:
+            if mailbox and username != mailbox:
                 scopes = ["Mail.ReadWrite.Shared"]
             auth_record = credential.authenticate(scopes=scopes)
             _cache_auth_record(auth_record, token_path)
@@ -160,23 +129,6 @@ 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("/")
@@ -191,7 +143,7 @@ class MSGraphConnection(MailboxConnection):
 
         request_body = {"displayName": folder_name}
         request_url = f"/users/{self.mailbox_name}/mailFolders{sub_url}"
-        resp = self._request_with_retries("post", request_url, json=request_body)
+        resp = self._client.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:
@@ -221,7 +173,7 @@ class MSGraphConnection(MailboxConnection):
             params["$top"] = batch_size
         else:
             params["$top"] = 100
-        result = self._request_with_retries("get", url, params=params)
+        result = self._client.get(url, params=params)
         if result.status_code != 200:
             raise RuntimeError(f"Failed to fetch messages {result.text}")
         messages = result.json()["value"]
@@ -229,7 +181,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._request_with_retries("get", result.json()["@odata.nextLink"])
+            result = self._client.get(result.json()["@odata.nextLink"])
             if result.status_code != 200:
                 raise RuntimeError(f"Failed to fetch messages {result.text}")
             messages.extend(result.json()["value"])
@@ -238,7 +190,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._request_with_retries("patch", url, json={"isRead": "true"})
+        resp = self._client.patch(url, json={"isRead": "true"})
         if resp.status_code != 200:
             raise RuntimeWarning(
                 f"Failed to mark message read{resp.status_code}: {resp.json()}"
@@ -246,7 +198,7 @@ class MSGraphConnection(MailboxConnection):
 
     def fetch_message(self, message_id: str, **kwargs):
         url = f"/users/{self.mailbox_name}/messages/{message_id}/$value"
-        result = self._request_with_retries("get", url)
+        result = self._client.get(url)
         if result.status_code != 200:
             raise RuntimeWarning(
                 f"Failed to fetch message{result.status_code}: {result.json()}"
@@ -258,7 +210,7 @@ class MSGraphConnection(MailboxConnection):
 
     def delete_message(self, message_id: str):
         url = f"/users/{self.mailbox_name}/messages/{message_id}"
-        resp = self._request_with_retries("delete", url)
+        resp = self._client.delete(url)
         if resp.status_code != 204:
             raise RuntimeWarning(
                 f"Failed to delete message {resp.status_code}: {resp.json()}"
@@ -268,7 +220,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._request_with_retries("post", url, json=request_body)
+        resp = self._client.post(url, json=request_body)
         if resp.status_code != 201:
             raise RuntimeWarning(
                 f"Failed to move message {resp.status_code}: {resp.json()}"
@@ -278,13 +230,11 @@ class MSGraphConnection(MailboxConnection):
         # Not needed
         pass
 
-    def watch(self, check_callback, check_timeout, should_reload=None):
+    def watch(self, check_callback, check_timeout):
         """Checks the mailbox for new messages every n seconds"""
         while True:
             sleep(check_timeout)
             check_callback(self)
-            if should_reload and should_reload():
-                return
 
     @lru_cache(maxsize=10)
     def _find_folder_id_from_folder_path(self, folder_name: str) -> str:
@@ -298,19 +248,6 @@ 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]
     ):
@@ -319,12 +256,8 @@ 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._request_with_retries("get", url + filter)
+        folders_resp = self._client.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,33 +55,15 @@ class IMAPConnection(MailboxConnection):
         return cast(str, self._client.fetch_message(message_id, parse=False))
 
     def delete_message(self, message_id: int):
-        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()
+        self._client.delete_messages([message_id])
 
     def move_message(self, message_id: int, folder_name: str):
-        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)
+        self._client.move_messages([message_id], folder_name)
 
     def keepalive(self):
         self._client.noop()
 
-    def watch(self, check_callback, check_timeout, should_reload=None):
+    def watch(self, check_callback, check_timeout):
         """
         Use an IDLE IMAP connection to parse incoming emails,
         and pass the results to a callback function
@@ -111,5 +93,3 @@ class IMAPConnection(MailboxConnection):
             except Exception as e:
                 logger.warning("IMAP connection error. {0}. Reconnecting...".format(e))
                 sleep(check_timeout)
-            if should_reload and should_reload():
-                return

parsedmarc/mail/mailbox_connection.py

@@ -28,5 +28,5 @@ class MailboxConnection(ABC):
     def keepalive(self):
         raise NotImplementedError
 
-    def watch(self, check_callback, check_timeout, should_reload=None):
+    def watch(self, check_callback, check_timeout):
         raise NotImplementedError

parsedmarc/mail/maildir.py

@@ -63,12 +63,10 @@ class MaildirConnection(MailboxConnection):
     def keepalive(self):
         return
 
-    def watch(self, check_callback, check_timeout, should_reload=None):
+    def watch(self, check_callback, check_timeout):
         while True:
             try:
                 check_callback(self)
             except Exception as e:
                 logger.warning("Maildir init error. {0}".format(e))
-            if should_reload and should_reload():
-                return
             sleep(check_timeout)

parsedmarc/opensearch.py

@@ -4,9 +4,7 @@ from __future__ import annotations
 
 from typing import Any, Optional, Union
 
-import boto3
 from opensearchpy import (
-    AWSV4SignerAuth,
     Boolean,
     Date,
     Document,
@@ -17,7 +15,6 @@ from opensearchpy import (
     Nested,
     Object,
     Q,
-    RequestsHttpConnection,
     Search,
     Text,
     connections,
@@ -275,9 +272,6 @@ 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
@@ -290,9 +284,6 @@ 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]
@@ -304,30 +295,10 @@ def set_hosts(
             conn_params["ca_certs"] = ssl_cert_path
         else:
             conn_params["verify_certs"] = False
-    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'."
-        )
+    if username and password:
+        conn_params["http_auth"] = username + ":" + password
+    if api_key:
+        conn_params["api_key"] = api_key
     connections.create_connection(**conn_params)
 
 

parsedmarc/syslog.py

@@ -57,7 +57,7 @@ class SyslogClient(object):
         self.logger.setLevel(logging.INFO)
 
         # Create the appropriate syslog handler based on protocol
-        self.log_handler = self._create_syslog_handler(
+        log_handler = self._create_syslog_handler(
             server_name,
             server_port,
             self.protocol,
@@ -69,7 +69,7 @@ class SyslogClient(object):
             retry_delay,
         )
 
-        self.logger.addHandler(self.log_handler)
+        self.logger.addHandler(log_handler)
 
     def _create_syslog_handler(
         self,
@@ -179,8 +179,3 @@ class SyslogClient(object):
         rows = parsed_smtp_tls_reports_to_csv_rows(smtp_tls_reports)
         for row in rows:
             self.logger.info(json.dumps(row))
-
-    def close(self):
-        """Remove and close the syslog handler, releasing its socket."""
-        self.logger.removeHandler(self.log_handler)
-        self.log_handler.close()

parsedmarc/utils.py

@@ -205,7 +205,8 @@ def get_reverse_dns(
         )[0]
 
     except dns.exception.DNSException as e:
-        logger.debug(f"get_reverse_dns({ip_address}) exception: {e}")
+        logger.warning(f"get_reverse_dns({ip_address}) exception: {e}")
+        pass
 
     return hostname
 

parsedmarc/webhook.py

@@ -63,7 +63,3 @@ class WebhookClient(object):
             self.session.post(webhook_url, data=payload, timeout=self.timeout)
         except Exception as error_:
             logger.error("Webhook Error: {0}".format(error_.__str__()))
-
-    def close(self):
-        """Close the underlying HTTP session."""
-        self.session.close()