diff --git a/.github/workflows/commit_checks.yaml b/.github/workflows/commit_checks.yaml index c60d665..7dfb73e 100644 --- a/.github/workflows/commit_checks.yaml +++ b/.github/workflows/commit_checks.yaml @@ -17,7 +17,7 @@ jobs: - uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7 - uses: actions/setup-python@ece7cb06caefa5fff74198d8649806c4678c61a1 # v6 with: - python-version: '3.13' # Specify a Python version explicitly + python-version: '3.14' # Specify a Python version explicitly - uses: pre-commit/action@2c7b3805fd2a0fd8c1884dcaebf91fc102a13ecd # v3.0.1 test: @@ -30,7 +30,7 @@ jobs: fail-fast: false matrix: os: ["ubuntu-latest", "macos-latest", "windows-latest"] - python-version: ["3.10", "3.11", "3.12", "3.13", "3.14"] + python-version: ["3.11", "3.12", "3.13", "3.14"] env: APIKEY: ${{ secrets.APIKEY }} DOMAIN: ${{ secrets.DOMAIN }} diff --git a/.github/workflows/pr_validation.yml b/.github/workflows/pr_validation.yml index 9c5b2cb..eda5e4f 100644 --- a/.github/workflows/pr_validation.yml +++ b/.github/workflows/pr_validation.yml @@ -18,7 +18,7 @@ jobs: - name: Set up Python uses: actions/setup-python@ece7cb06caefa5fff74198d8649806c4678c61a1 # v6 with: - python-version: '3.13' + python-version: '3.14' - name: Build package run: | diff --git a/.github/workflows/publish.yml b/.github/workflows/publish.yml index 2d4e78f..39fef52 100644 --- a/.github/workflows/publish.yml +++ b/.github/workflows/publish.yml @@ -25,7 +25,7 @@ jobs: - name: Set up Python uses: actions/setup-python@ece7cb06caefa5fff74198d8649806c4678c61a1 # v6 with: - python-version: '3.13' + python-version: '3.14' - name: Install build tools run: pip install --upgrade build setuptools wheel setuptools-scm twine diff --git a/.github/workflows/security.yml b/.github/workflows/security.yml index d3fd468..e93c786 100644 --- a/.github/workflows/security.yml +++ b/.github/workflows/security.yml @@ -18,7 +18,7 @@ jobs: - uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v4 - uses: actions/setup-python@ece7cb06caefa5fff74198d8649806c4678c61a1 # v5 with: - python-version: "3.13" + python-version: "3.14" cache: 'pip' - run: python -m pip install --upgrade pip - run: pip install ruff bandit mypy pip-audit @@ -51,7 +51,7 @@ jobs: steps: - uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v4 - uses: actions/setup-python@ece7cb06caefa5fff74198d8649806c4678c61a1 # v5 - with: { python-version: "3.13" } + with: { python-version: "3.14" } - run: python -m pip install --upgrade pip - run: pip install pip-audit - run: pip-audit --strict diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index 4583608..0ac9108 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -37,6 +37,9 @@ ci: skip: [] submodules: false +# Run hooks on staged files by default, ignoring the commit-msg phase +default_stages: [pre-commit] + # .pre-commit-config.yaml repos: - repo: https://github.com/pre-commit/pre-commit-hooks diff --git a/CHANGELOG.md b/CHANGELOG.md index 7007d89..9dd3b29 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,7 +2,28 @@ We [keep a changelog.](http://keepachangelog.com/) -## [Unreleased] +## [Unreleased] (v1.9.0) + +### Added + +- **IdempotencyGuard**: Implemented exactly-once email delivery mechanisms (SHA-256 fingerprinting) to prevent duplicate sends during network partitions. +- **RetryPolicy**: Introduced a flexible retry configuration with stateless exponential backoff and jitter to mitigate the "Thundering Herd" effect. +- **`httpx2` Compatibility**: Added native support for the modern `httpx2` engine via the `mailgun._httpx_compat.py` bridge, with graceful fallbacks. +- **`mailgun.ext` Ecosystem**: Introduced strict Pydantic v2 payload schemas (e.g., `SendMessageSchema`). +- **ChunkedStreamer**: Added memory-safe lazy streaming for large file attachments (up to 25MB) using 512KB partitions. +- **SpamGuard**: Added a zero-network static HTML analyzer to preemptively flag deliverability risks (XSS, missing alt tags) before dispatching to Mailgun. +- **IDN Routing**: Implemented `normalize_domain` to natively convert Internationalized Domain Names to safe RFC 3490 Punycode. +- Added `DeliverabilityError` exception to handle SpamGuard rule violations. + +### Changed + +- **[BREAKING CHANGE]** Dropped support for Python 3.10. The SDK now strictly requires Python 3.11 or higher. +- Purged the `typing-extensions` dependency from the package, replacing it with standard library `typing` equivalents (`Self`, `TypedDict`, `NotRequired`). + +### Fixed + +- Fixed strict typing issues and circular import risks by routing response contracts (`MockResponse`) strictly through `mailgun/types.py`. +- Updated GitHub Actions CI workflows to explicitly test against Python 3.14. ## v1.8.0 - 2026-07-20 diff --git a/README.md b/README.md index abb9364..b428275 100644 --- a/README.md +++ b/README.md @@ -29,16 +29,18 @@ Check out all the resources and Python code examples in the official - [Usage](#usage) - [Logging Debugging and Secure Redaction](#logging-debugging-and-secure-redaction) - [Timeout Configuration](#timeout-configuration) + - [Exactly-Once Delivery & Retry Policies](#exactly-once-delivery--retry-policies) - [API Response Codes](#api-response-codes) - [IDE Autocompletion & DX](#ide-autocompletion--dx) - - [Zero-Leak Sandbox Mode](#zero-leak-sandbox-mode) - - [API Response Codes](#api-response-codes) - - [Context Managers (Safe Resource Teardown)](#context-managers-safe-resource-teardown) + - [Zero-Leak Development Mode](#zero-leak-development-mode) + - [Strict Payload Schemas](#strict-payload-schemas) + - [Strict Typed Schemas (mailgun.ext)](#strict-typed-schemas-mailgunext) + - [Memory-Safe Attachments (ChunkedStreamer)](#memory-safe-attachments-chunkedstreamer) - [Fluent Message Builder](#fluent-message-builder) - [Streaming Pagination](#streaming-pagination) - - [Strict Payload Schemas](#strict-payload-schemas) - - [API Reference](#request-examples) - - [Full list of supported endpoints](#full-list-of-supported-endpoints) + - [Readiness Probe](#readiness-probe) + - [API Reference](#api-reference) + - [Full list of examples](#full-list-of-examples) - [Messages](#messages) - [Send an email](#send-an-email) - [Send an email with advanced parameters (Tags, Testmode, STO)](#send-an-email-with-advanced-parameters-tags-testmode-sto) @@ -51,9 +53,9 @@ Check out all the resources and Python code examples in the official - [Create a domain](#create-a-domain) - [Update a domain](#update-a-domain) - [Domain connections](#domain-connections) - - [Domain keys](#domain-keys) - - [List keys for all domains](#list-keys-for-all-domains) - - [Create a domain key](#create-a-domain-key) + - [Domain keys](#domain-keys) + - [List keys for all domains](#list-keys-for-all-domains) + - [Create a domain key](#create-a-domain-key) - [Update DKIM authority](#update-dkim-authority) - [Domain Tracking](#domain-tracking) - [Get tracking settings](#get-tracking-settings) @@ -126,15 +128,13 @@ Check out all the resources and Python code examples in the official - [License](#license) - [Contribute](#contribute) - [Security](#security) + - [Enterprise Security Audit Hooks (PEP 578)](#enterprise-security-audit-hooks-pep-578) + - [Pre-Flight Delivery Validation (SpamGuard)](#pre-flight-delivery-validation-spamguard) - [Contributors](#contributors) ## Compatibility -This library `mailgun` officially supports the following Python versions: - -- python >=3.10,\<3.15 - -It's tested up to 3.14 (including). +This SDK is compatible with Python **3.11+**. It is tested up to 3.14 (including). It guarantees cross-platform compatibility across Linux, macOS, and Windows. ## Requirements @@ -145,7 +145,8 @@ To build the `mailgun` package from the sources you need `setuptools` (as a buil ### Runtime dependencies -At runtime the package requires `requests >=2.33.0`. For async support, it uses `httpx >=0.24` and `typing-extensions >=4.7.1` (for pre-3.11 backward compatibility). +At runtime the package requires `requests >=2.33.0`. For async support, it uses `httpx2 >=2.7.0`. +Async client automatically detects and uses `httpx2` if available, falling back seamlessly to legacy `httpx`. ### Test dependencies @@ -168,9 +169,7 @@ Use the below code to install it locally by cloning this repository: ```bash git clone https://github.com/mailgun/mailgun-python cd mailgun-python -``` -```bash pip install . ``` @@ -257,9 +256,6 @@ Synchronous and Asynchronous Clients. Initialize your [Mailgun](http://www.mailgun.com/) client. -> [!TIP] -> **New in v1.7.0:** The SDK now utilizes connection pooling (`requests.Session`) under the hood to dramatically improve performance by reusing TLS connections. - **The Simple Variant (Backward Compatible)** For simple scripts, lambdas, or single-request apps, you can initialize and use the client directly. Python's garbage collector will eventually clean up the connection. @@ -268,7 +264,7 @@ import os from mailgun.client import Client client = Client(auth=("api", os.environ["APIKEY"])) -client.messages.create(data={"to": "user@example.com"}) +client.messages.create(domain="your-domain.com", data={"to": "user@example.com"}) ``` > [!WARNING] @@ -277,13 +273,25 @@ client.messages.create(data={"to": "user@example.com"}) **The Recommended Variant (Context Manager)** +Initialize your Mailgun client using the with context manager to ensure connection pooling (`requests.Session)` and underlying socket descriptors are gracefully torn down: + ```python import os from mailgun.client import Client # Sockets are safely managed and closed automatically with Client(auth=("api", os.environ["APIKEY"])) as client: - client.messages.create(data={"to": "user@example.com"}) + response = client.messages.create( + domain=os.environ["DOMAIN"], + data={ + "from": os.environ["MESSAGES_FROM"], + "to": [os.environ["MESSAGES_TO"]], + "subject": "Hello from Mailgun Python SDK", + "text": "Testing some Mailgun awesomeness!", + }, + ) + print(response.status_code) + print(response.json()) ``` ### AsyncClient @@ -301,7 +309,7 @@ async def main(): # and automatic socket teardown. async with AsyncClient(auth=("api", "your-api-key")) as client: response = await client.messages.create( - domain="YOUR_DOMAIN_NAME", + domain=os.environ["DOMAIN"], data={ "from": "Excited User ", "to": ["bar@example.com"], @@ -318,42 +326,6 @@ if __name__ == "__main__": ## Usage -Send a message with a Synchronous Client safely inside a context manager. - -```python -import os -from mailgun import Client - -# Send an email using context manager -with Client(auth=("api", os.environ["APIKEY"])) as client: - response = client.messages.create( - data={ - "from": "Excited User ", - "to": ["recipient@example.com"], - "subject": "Hello from Mailgun Python SDK", - "text": "Testing some Mailgun awesomeness!", - } - ) - - print(response.status_code) - print(response.json()) -``` - -The `AsyncClient` provides async equivalents for all methods available in the sync `Client`. The method signatures and parameters are identical - simply add `await` when calling methods: - -```python -import os -from mailgun import Client, AsyncClient - -# Sync version -with Client(auth=("api", os.environ["APIKEY"])) as client: - result = client.domainlist.get() - -# Async version -async with AsyncClient(auth=("api", os.environ["APIKEY"])) as client: - result = await client.domainlist.get() -``` - For detailed examples of all available methods, parameters, and use cases, refer to the [mailgun/examples](mailgun/examples) section. All examples can be adapted to async by using `AsyncClient` and adding `await` to method calls. ### Logging, Debugging, and Secure Redaction @@ -367,6 +339,7 @@ To enable detailed logging in your application, configure the logger before init ```python import logging +import os from mailgun import Client # Enable DEBUG level for the Mailgun SDK logger @@ -379,7 +352,7 @@ logging.basicConfig(format="%(levelname)s - %(name)s - %(message)s") with Client(auth=("api", "key-super-secret-12345")) as client: # API keys will be redacted: # "Sending request to https://api.mailgun.net/v3/messages with auth ('api', 'key-[REDACTED]')" - client.domains.get() + client.domains.get(domain=os.environ["DOMAIN"]) ``` ### Timeout Configuration @@ -389,59 +362,41 @@ By default, the SDK relies on the underlying HTTP client's standard timeouts. To Timeouts can be passed as a single `float` (seconds for both connect and read) or a tuple (connect_timeout, read_timeout): ```python +import os from mailgun import Client # 3.5 seconds to connect, 15 seconds to wait for the server response with Client(auth=("api", "your-key"), timeout=(3.5, 15.0)) as client: # Execute safely timed API calls here - pass + client.domains.get(domain=os.environ["DOMAIN"]) ``` -### IDE Autocompletion & DX - -The `Client` utilizes a dynamic routing engine but is heavily optimized for modern Developer Experience (DX). +### Exactly-Once Delivery & Retry Policies -- **Introspection**: Calling `dir(client)` or using autocomplete in IDEs like VS Code or PyCharm will automatically expose all available API endpoints (e.g., `client.messages`, `client.domains`, `client.bounces`). -- **Security Guardrails**: If you accidentally print the client instance or an exception traceback occurs in your CI/CD logs, your API key is strictly redacted from memory dumps: (`'api', '***REDACTED***'`). -- **Performance**: JSON payloads are automatically minified before transit to save bandwidth on large batch requests, and internal route resolution is heavily cached in memory. - -### Zero-Leak Sandbox Mode - -For local development and CI/CD pipelines, the Mailgun SDK offers a native **Zero-Leak Sandbox Mode**. By initializing the client with `dry_run=True`, the SDK will safely intercept all network traffic locally. - -This allows you to fully validate your SDK initialization, dynamic routing, and payload building without dispatching real HTTP requests to Mailgun servers. This prevents accidental spam, list mutations, or billing charges during testing. +Configure resilient retries using exponential backoff and jitter alongside `IdempotencyGuard` to prevent duplicate billing during transient partitions: ```python +import os +import uuid from mailgun.client import Client +from mailgun.config import RetryPolicy -# 1. Initialize the client in strict Sandbox Mode -with Client(auth=("api", "your-api-key"), dry_run=True) as client: - # 2. Execute a state-changing API call - response = client.messages.create( - domain="yourdomain.com", - data={ - "from": "sender@example.com", - "to": "test@example.com", - "subject": "Testing Sandbox", - "text": "This will not actually send!", - }, - ) +# Configure 3 retries, a 1.0s base delay, a 10.0s max cap, and respect 429 Retry-After headers +custom_retry = RetryPolicy(max_retries=3, base_delay=1.0, max_delay=10.0, respect_retry_after=True) - # 3. The SDK intercepts the I/O layer and returns a mock 200 OK response - print(response.status_code) - # Outputs: 200 +with Client(auth=("api", "your-api-key"), retry_policy=custom_retry) as client: + # Generate a unique idempotency key for this specific transaction + headers = {"Idempotency-Key": str(uuid.uuid4())} - print(response.json()) - # Outputs: {"message": "Dry run successful - request intercepted", "id": ""} + # If the network fails, the SDK will safely back off and retry. + # IdempotencyGuard ensures retries won't result in duplicate emails to the user + client.messages.create( + domain=os.environ["DOMAIN"], + data={"to": "user@example.com", "subject": "Payment Receipt", "text": "Hello World!"}, + headers=headers, + ) ``` -Key Behaviors in `dry_run` Mode: - -- Local payload checks (like strict minification and JSON serialization) still execute. -- Security sanitization and path segment rules still execute. -- Deprecation warnings will still be raised if you use an outdated endpoint. -- `sys.audit` events and standard `logging` messages are still emitted, clearly marked with `DRY RUN: Intercepting request...`. - ### API Response Codes All of Mailgun's HTTP response codes follow standard HTTP definitions. For some additional information and @@ -462,35 +417,148 @@ request, such as a non-existing endpoint. **500/502/503** - Internal Error on the Mailgun side. The SDK automatically retries these using Exponential Backoff. If the issue persists, please reach out to our support team. -### Context Managers (Safe Resource Teardown) +### IDE Autocompletion & DX + +The `Client`/`AsyncClient` utilize a dynamic routing engine but is heavily optimized for modern Developer Experience (DX). + +- **Introspection**: Calling `dir(client)` or using autocomplete in IDEs like VS Code or PyCharm will automatically expose all available API endpoints (e.g., `client.messages`, `client.domains`, `client.bounces`). +- **Security Guardrails**: If you accidentally print the client instance or an exception traceback occurs in your CI/CD logs, your API key is strictly redacted from memory dumps: (`'api', '***REDACTED***'`). +- **Performance**: JSON payloads are automatically minified before transit to save bandwidth on large batch requests, and internal route resolution is heavily cached in memory. + +### Zero-Leak Development Mode + +During local development and automated CI/CD test runs, you can instantiate the client in dry-run mode to completely intercept and mock outbound network requests: + +```python +import os +from mailgun.client import Client + +# dry_run=True intercepts the network call and prevents actual delivery. +# Network requests will be skipped, returning a synthetic 200 OK MockResponse +with Client(auth=("api", "API_KEY"), dry_run=True) as client: + client.messages.create( + domain=os.environ["DOMAIN"], + data={"from": "test@test.com", "to": "user@test.com"}, + ) +``` + +Key Behaviors in `dry_run` Mode: + +- Local payload checks (like strict minification and JSON serialization) still execute. +- Security sanitization and path segment rules still execute. +- Deprecation warnings will still be raised if you use an outdated endpoint. +- `sys.audit` events and standard `logging` messages are still emitted, clearly marked with `DRY RUN: Intercepting request...`. -Always use the `Client` or `AsyncClient` inside a `with` statement. This ensures that underlying TCP connection pools are safely closed and sensitive API keys are immediately purged from memory once the block exits, preventing resource leaks. +### Strict Payload Schemas -**Synchronous:** +If you prefer to build your own dictionaries instead of using the builder, you can opt in to `TypedDict` schemas for full IDE autocomplete and `mypy` compile-time safety. ```python from mailgun import Client +from mailgun.types import SendMessagePayload -with Client(auth=("api", "your-api-key")) as client: - response = client.domains.get() - print(response.json()) -# Connection pool is closed and credentials are wiped from memory here. +my_data: SendMessagePayload = { + "from": "admin@domain.com", + "to": ["user@example.com"], + "subject": "Strictly Typed Request", +} + +with Client(auth=("api", "key")) as client: + client.messages.create(domain="domain.com", data=my_data) ``` -**Asynchronous:** +### Strict Typed Schemas (mailgun.ext) + +For enterprise applications using frameworks like FastAPI or Django, you can import strict typed dictionaries and Pydantic models from `mailgun.ext.pydantic` to validate input payloads at system boundaries: ```python -import asyncio -from mailgun import AsyncClient +from mailgun.ext.pydantic.models import SendMessageSchema +from pydantic import ValidationError +try: + valid_payload = SendMessageSchema( + from_="admin@company.com", + to=["user@example.com"], + subject="Weekly Report", + text="Here is your report.", + ) + print("✅ Valid payload passed validation!") + print(f"Data: {valid_payload.to_mailgun_payload()}") +except ValidationError as e: + print(f"❌ Valid payload failed: {e}") +``` -async def main(): +**Strict Payload Validation (Pydantic & FastAPI)**: + +This enables "Fail-Fast" local validation and perfectly integrates with frameworks like FastAPI. + +First, ensure you have installed the optional dependencies: `pip install mailgun[fastapi]` + +```python +from fastapi import FastAPI, Depends +from mailgun.client import AsyncClient +from mailgun.ext.pydantic.models import SendMessageSchema + +app = FastAPI() + + +# Dependency to manage the connection pool safely +async def get_mailgun_client(): async with AsyncClient(auth=("api", "your-api-key")) as client: - response = await client.domains.get() - print(response.json()) + yield client + +@app.post("/send-email") +async def send_email( + payload: SendMessageSchema, # Pydantic instantly validates incoming JSON + mailgun_client: AsyncClient = Depends(get_mailgun_client), +): + # .model_dump(by_alias=True) ensures keys like 'from_' map safely to 'from' + clean_data = payload.model_dump(by_alias=True, exclude_none=True) -asyncio.run(main()) + response = await mailgun_client.messages.create(domain="your-domain.com", data=clean_data) + return response.json() +``` + +### Memory-Safe Attachments (ChunkedStreamer) + +When sending massive attachments or processing large file exports, use the `ChunkedStreamer` utility to stream data in 512KB chunks, preventing out-of-memory (OOM) errors in resource-constrained environments or serverless functions: + +```python +import os + +from mailgun.builders import MailgunMessageBuilder +from mailgun.client import AsyncClient, Client + +API_KEY: str = os.environ.get("APIKEY", "") +DOMAIN: str = os.environ.get("DOMAIN", "") +MESSAGES_TO = os.environ.get("MESSAGES_TO") or f"success@{DOMAIN}" + +test_file = "large_report.pdf" +with open(test_file, "wb") as f: + f.write(os.urandom(20 * 1024 * 1024)) + +try: + payload, files = ( + MailgunMessageBuilder(f"mailgun@{DOMAIN}") + .add_recipient(MESSAGES_TO) + .set_subject("Monthly Enterprise Report") + .set_text("Here is the 20MB data export.") + .attach_stream(test_file) + .build() + ) + + # If the network is slow, increase read/write timeout to 300 sec, + # but keep 10 sec for connection timeout. + custom_timeout = (10.0, 300.0) + + with Client(auth=("api", API_KEY), timeout=custom_timeout) as client: + req = client.messages.create(domain=DOMAIN, data=payload, files=files) + print("Success:", req.json()) + +finally: + if os.path.exists(test_file): + os.remove(test_file) ``` ### Fluent Message Builder @@ -501,6 +569,7 @@ Constructing complex multipart emails with custom variables (`v:`), custom heade from mailgun import Client from mailgun.builders import MailgunMessageBuilder +# Construct a complex email using the fluent interface with Client(auth=("api", "your-api-key")) as client: payload, files = ( MailgunMessageBuilder("support@yourdomain.com") @@ -510,6 +579,20 @@ with Client(auth=("api", "your-api-key")) as client: .add_custom_variable("invoice_id", 1234) # Translates to "v:invoice_id" .add_custom_header("Reply-To", "billing@...") # Translates to "h:Reply-To" .attach_file("/tmp/invoice_1234.pdf", safe_base_dir="/tmp/") # Path Traversal guardrail + # Define short, human-readable aliases for complex local file paths + .attach_inline("assets/logos/logo_v2_final.png", cid="company_logo") + .attach_inline("assets/signatures/ceo_sign.png", cid="ceo_signature") + .set_html( + """ + + + Company Logo
+

Hello! Thank you for choosing us.


+ CEO Signature + + + """ + ) .build() ) @@ -518,7 +601,7 @@ with Client(auth=("api", "your-api-key")) as client: ### Streaming Pagination -For endpoints that return massive datasets (like Events, Bounces, or Suppressions), loading all pages into memory can crash your application. +For endpoints that return massive datasets (like Events, Bounces, or Suppressions), loading all pages into memory can cause latency spikes or Out-of-Memory crashes. The `.stream()` method handles cursor-based pagination invisibly under the hood, yielding one item at a time. ```python @@ -530,30 +613,30 @@ with Client(auth=("api", "key")) as client: print(f"Bounced: {event['recipient']}") ``` -### Strict Payload Schemas - -If you prefer to build your own dictionaries instead of using the builder, you can opt in to `TypedDict` schemas for full IDE autocomplete and `mypy` compile-time safety. +### Readiness Probe ```python +import sys +import os from mailgun import Client -from mailgun.types import SendMessagePayload -my_data: SendMessagePayload = { - "from": "admin@domain.com", - "to": ["user@example.com"], - "subject": "Strictly Typed Request", -} +api_key = os.environ.get("MAILGUN_API_KEY") -with Client(auth=("api", "key")) as client: - client.messages.create(domain="domain.com", data=my_data) +with Client(auth=("api", api_key)) as client: + if client.ping(): + print("Status: Healthy") + sys.exit(0) # Exit code 0 indicates success + else: + print("Status: Unhealthy") + sys.exit(1) # Exit code 1 triggers container restart/unready state ``` -## Request examples +## API Reference -### Full list of supported endpoints +### Full list of examples > [!IMPORTANT] -> This is a full list of supported endpoints this SDK provides [mailgun/examples](mailgun/examples) +> This is a full list of [mailgun/examples](mailgun/examples). ### Messages @@ -575,7 +658,7 @@ data = { } with Client(auth=("api", os.environ["APIKEY"])) as client: - req = client.messages.create(data=data) + req = client.messages.create(domain=os.environ["DOMAIN"], data=data) ``` #### Send an email with advanced parameters (Tags, Testmode, STO) @@ -598,7 +681,7 @@ data = { } with Client(auth=("api", os.environ["APIKEY"])) as client: - req = client.messages.create(data=data) + req = client.messages.create(domain=os.environ["DOMAIN"], data=data) ``` #### Send an email with attachments @@ -613,7 +696,7 @@ from mailgun import Client with Client(auth=("api", os.environ["APIKEY"])) as client: files = [("attachment", ("report.pdf", Path("report.pdf").read_bytes()))] # Assuming `data` is predefined like in the previous example - req = client.messages.create(data=data, files=files) + req = client.messages.create(domain=os.environ["DOMAIN"], data=data, files=files) ``` #### Send a scheduled message @@ -701,7 +784,7 @@ from mailgun import Client domain_name = "python.test.com" with Client(auth=("api", os.environ["APIKEY"])) as client: - data = client.domains.get(domain_name=domain_name) + data = client.domains.get(domain=domain_name) print(data.json()) ``` @@ -772,15 +855,15 @@ def get_dkim_keys() -> None: GET /v1/dkim/keys :return: """ - data = { + query = { "page": "string", "limit": "0", - "signing_domain": "python.test.domain5", + "signing_domain": os.environ["DOMAIN"], "selector": "smtp", } with Client(auth=("api", os.environ["APIKEY"])) as client: - request = client.dkim_keys.get(data=data) + request = client.dkim_keys.get(filters=query) print(request.json()) ``` @@ -831,10 +914,8 @@ def post_dkim_keys() -> None: "pem": files, } - headers = {"Content-Type": "multipart/form-data"} - with Client(auth=("api", os.environ["APIKEY"])) as client: - request = client.dkim_keys.create(data=data, headers=headers, files=files) + request = client.dkim_keys.create(data=data, files=files) print(request.json()) ``` @@ -896,7 +977,7 @@ data = { } with Client(auth=("api", os.environ["APIKEY"])) as client: - req = client.domains_webhooks.create(data=data) + client.domains_webhooks.create(domain=os.environ["DOMAIN"], data=data) ``` #### Get all webhooks @@ -906,7 +987,7 @@ import os from mailgun import Client with Client(auth=("api", os.environ["APIKEY"])) as client: - req = client.domains_webhooks.get() + client.domains_webhooks.get(domain=os.environ["DOMAIN"]) ``` #### Create Account-Level Webhooks (v1) @@ -961,6 +1042,7 @@ with Client(auth=("api", os.environ["APIKEY"])) as client: Items that have no bounces and no delays (`classified_failures_count==0`) are not returned. ```python +import json import os from mailgun import Client @@ -1000,7 +1082,7 @@ payload = { headers = {"Content-Type": "application/json"} with Client(auth=("api", os.environ["APIKEY"])) as client: - req = client.bounceclassification_metrics.create(data=payload, headers=headers) + req = client.bounceclassification_metrics.create(data=json.dumps(payload), headers=headers) print(req.json()) ``` @@ -1096,6 +1178,7 @@ filtered to provide insights into the health of your email infrastructure Gets customer event logs for an account. ```python +import json import os from mailgun import Client @@ -1108,7 +1191,7 @@ def post_analytics_logs() -> None: """ domain: str = os.environ["DOMAIN"] - data = { + nested_dict = { "start": "Wed, 24 Sep 2025 00:00:00 +0000", "end": "Thu, 25 Sep 2025 00:00:00 +0000", "filter": { @@ -1127,8 +1210,10 @@ def post_analytics_logs() -> None: }, } + headers = {"Content-Type": "application/json"} + with Client(auth=("api", os.environ["APIKEY"])) as client: - req = client.analytics_logs.create(data=data) + req = client.analytics_logs.create(data=json.dumps(nested_dict), headers=headers) print(req.json()) ``` @@ -1342,7 +1427,7 @@ data = { } with Client(auth=("api", os.environ["APIKEY"])) as client: - req = client.routes.create(domain=domain, data=data) + req = client.routes.create(data=data) print(req.json()) ``` @@ -1398,7 +1483,7 @@ from mailgun import Client domain: str = os.environ["DOMAIN"] with Client(auth=("api", os.environ["APIKEY"])) as client: - req = client.lists.delete(domain=domain, address=f"python_sdk2@{domain}") + req = client.lists.delete(address=f"python_sdk2@{domain}") print(req.json()) ``` @@ -1669,12 +1754,11 @@ Thanks to the dynamic routing engine, the SDK natively supports Mailgun's supple import os from mailgun import Client -domain: str = os.environ["DOMAIN"] data = {"address": "test2@gmail.com"} params = {"provider_lookup": "false"} with Client(auth=("api", os.environ["APIKEY"])) as client: - req = client.addressvalidate.create(domain=domain, data=data, filters=params) + req = client.addressvalidate.create(data=data, filters=params) print(req.json()) ``` @@ -1724,7 +1808,7 @@ It will successfully execute the request but will emit a non-breaking Python `De ## Type Hinting -This SDK is fully type-hinted and compatible with static type checkers like `mypy` and `pyright`. +This SDK is fully type-hinted and complies with PEP 561 (`py.typed` included). Static type checkers (`mypy`, `pyright`) are enforced during CI checks. Because of the dynamic URL dispatch engine (`__getattr__`), IDEs may flag endpoints like `client.messages.create` as `Any`. If you enforce strict typing in your application, you may safely ignore these specific dynamically dispatched calls. @@ -1770,6 +1854,32 @@ with Client(auth=("api", os.environ.get("APIKEY", "your-api-key"))) as client: response = client.domains.get() ``` +### Pre-Flight Delivery Validation (SpamGuard) + +The SDK includes a zero-network static analyzer called **SpamGuard**. It evaluates your payload *before* making an HTTP request. If your HTML contains known spam triggers (like invalid tags) or if you attempt to send to malformed Internationalized Domain Names (IDN), the SDK fails fast locally. + +```python +import os +from mailgun.client import Client +from mailgun.handlers.error_handler import DeliverabilityError + +domain = os.environ["DOMAIN"] + +with Client(auth=("api", "YOUR_API_KEY")) as client: + try: + client.messages.create( + domain=domain, + data={ + "from": "sender@YOUR_DOMAIN_NAME", + "to": ["test@example.com"], + "subject": "Hello", + "html": "", # Will trigger SpamGuard + }, + ) + except DeliverabilityError as e: + print(f"Pre-flight check failed! Risk score: {e.score}. Issues: {e.issues}") +``` + ## Contributors - [@diskovod](https://github.com/diskovod) diff --git a/SECURITY.md b/SECURITY.md index 3f9ead5..979fb7e 100644 --- a/SECURITY.md +++ b/SECURITY.md @@ -4,8 +4,8 @@ | Version | Supported | | ------- | ------------------ | -| 1.8.x | :white_check_mark: | -| < 1.8.0 | :x: | +| 1.9.x | :white_check_mark: | +| < 1.9.0 | :x: | # Vulnerability Disclosure diff --git a/conda.recipe/meta.yaml b/conda.recipe/meta.yaml index 5661147..c52c022 100644 --- a/conda.recipe/meta.yaml +++ b/conda.recipe/meta.yaml @@ -16,7 +16,7 @@ source: build: number: 0 - skip: True # [py<310] + skip: True # [py<311] script: {{ PYTHON }} -m pip install . --no-deps --no-build-isolation -vv requirements: @@ -28,6 +28,7 @@ requirements: {% endfor %} run: - python + - httpx2 >=2.7.0 - httpx >=0.24 - requests >=2.33.0 - typing-extensions >=4.7.1 # [py<311] diff --git a/environment-dev.yaml b/environment-dev.yaml index ae1411a..7739b47 100644 --- a/environment-dev.yaml +++ b/environment-dev.yaml @@ -3,7 +3,7 @@ name: mailgun-dev channels: - defaults dependencies: - - python >=3.10 + - python >=3.11 # build & host deps - pip - setuptools-scm @@ -11,8 +11,11 @@ dependencies: - python-build # runtime deps - requests >=2.32.5 + - conda-forge::httpx2 >=2.7.0 - httpx >=0.24.0 - - typing_extensions >=4.7.1 # [py<311] + # extras + - fastapi + - pydantic >=2.0.0 # tests - conda-forge::pyfakefs - coverage >=4.5.4 diff --git a/environment.yaml b/environment.yaml index 35c2c34..6a7b826 100644 --- a/environment.yaml +++ b/environment.yaml @@ -3,13 +3,13 @@ name: mailgun channels: - defaults dependencies: - - python >=3.10 + - python >=3.11 # build & host deps - pip # runtime deps - requests >=2.32.5 + - conda-forge::httpx2 >=2.7.0 - httpx >=0.24.0 - - typing_extensions >=4.7.1 # [py<311] # tests - pytest >=9.0.3 # other diff --git a/mailgun/_httpx_compat.py b/mailgun/_httpx_compat.py new file mode 100644 index 0000000..eb9f532 --- /dev/null +++ b/mailgun/_httpx_compat.py @@ -0,0 +1,28 @@ +"""HTTPX compatibility layer for the Mailgun SDK. + +Prefers httpx2 (the actively maintained continuation), but gracefully +falls back to the legacy httpx library if httpx2 is unavailable. +""" + +from typing import TYPE_CHECKING + + +if TYPE_CHECKING: + # For static analysis (Mypy/Pyright), we import from httpx since the APIs are identical + import httpx + from httpx import AsyncClient, HTTPError, Response, Timeout + + HAS_HTTPX2: bool +else: + try: + import httpx2 as httpx + from httpx2 import AsyncClient, HTTPError, Response, Timeout + + HAS_HTTPX2 = True + except ImportError: + import httpx + from httpx import AsyncClient, HTTPError, Response, Timeout + + HAS_HTTPX2 = False + +__all__ = ["HAS_HTTPX2", "AsyncClient", "HTTPError", "Response", "Timeout", "httpx"] diff --git a/mailgun/builders.py b/mailgun/builders.py index 1c5953c..681890e 100644 --- a/mailgun/builders.py +++ b/mailgun/builders.py @@ -2,18 +2,139 @@ from __future__ import annotations +import asyncio import json -import sys +import mimetypes +from contextlib import suppress from pathlib import Path -from typing import Any +from typing import IO, TYPE_CHECKING, Any, Self, Union +from mailgun.security import IdempotencyGuard, SecurityGuard, SpamGuard, SpamReport -if sys.version_info >= (3, 11): - from typing import Self -else: - from typing_extensions import Self -from mailgun.security import SecurityGuard +if TYPE_CHECKING: + from collections.abc import AsyncGenerator, Generator + + +CHUNK_SIZE: int = 512 * 1024 # 512KB + +# Defines the 3-tuple structure: (filename, payload, content_type) +FileContent = Union[bytes, "ChunkedStreamer", IO[bytes]] +FileTuple = tuple[str, FileContent, str] + + +class ChunkedStreamer: + """Generator-based stream for safe attachment processing (CWE-400 Defense). + + Lazily reads files in chunks to prevent Out-Of-Memory (OOM) crashes + when processing large attachments in memory-constrained environments + (like Serverless functions). + """ + + __slots__ = ("_file", "_file_path", "chunk_size") + + def __init__( + self, + file_path: str | Path, + safe_base_dir: str | Path | None = None, + chunk_size: int = CHUNK_SIZE, + ) -> None: + """Init chunked streamer.""" + # Provide a secure default base directory (e.g., current working directory) if None is passed + resolved_base_dir = safe_base_dir if safe_base_dir is not None else Path.cwd() + safe_path = SecurityGuard.validate_attachment_path(file_path, resolved_base_dir) + + self._file_path = str(safe_path) + self.chunk_size = chunk_size + self._file: IO[bytes] | None = None + + def read(self, size: int) -> bytes: + """File-like read method required by requests/httpx multipart encoders. + + Args: + size: The maximum number of bytes to read. + + Returns: + A byte string containing the read data. + """ + if self._file is None: + self._file = Path(self._file_path).open("rb") # noqa: SIM115 + + chunk = self._file.read(size) + + # Auto-close the file descriptor as soon as EOF is reached. + # This guarantees teardown even if the HTTP library forgets to call .close(). + if not chunk: + self.close() + + return chunk + + def __iter__(self) -> Generator[bytes, None, None]: + """Stream the file natively in chunks. + + Yields: + Sequential byte chunks of the file payload. + """ + try: + # Sync the iterator with the class-level _file descriptor + if self._file is None: + self._file = Path(self._file_path).open("rb") # noqa: SIM115 + + while True: + chunk = self._file.read(self.chunk_size) + if not chunk: + break + yield chunk + finally: + # The finally block executes if the generator exhausts + # naturally OR if a network error causes a GeneratorExit early. + self.close() + + async def __aiter__(self) -> AsyncGenerator[bytes, None]: + """Safely stream chunks in an async context without blocking the event loop. + + Yields: + Sequential byte chunks of the file payload. + """ + try: + if self._file is None: + # Offload the blocking open() call to a thread pool + self._file = await asyncio.to_thread(Path(self._file_path).open, "rb") + + while True: + # Offload the blocking read() call to a thread pool + chunk = await asyncio.to_thread(self._file.read, self.chunk_size) + if not chunk: + break + yield chunk + finally: + self.close() + + def close(self) -> None: + """Explicitly close the underlying file descriptor to prevent leaks. + + This method is automatically called by requests/httpx after the + multipart payload has been fully transmitted. + """ + file_obj = getattr(self, "_file", None) + if file_obj is not None: + with suppress(Exception): + file_obj.close() + self._file = None + + def __del__(self) -> None: + """Safety net to prevent FD leaks if the object is garbage collected before being explicitly closed.""" + with suppress(Exception): + self.close() + + @property + def name(self) -> str: + """The file name to satisfy some HTTP library introspection checks. + + Returns: + The base name of the file. + """ + return Path(self._file_path).name class MailgunMessageBuilder: @@ -26,7 +147,9 @@ class MailgunMessageBuilder: def __init__(self, from_email: str) -> None: """Initialize the builder with a sender email.""" self._payload: dict[str, Any] = {"from": from_email, "to": []} - self._files: list[tuple[str, tuple[str, bytes]]] = [] + self._files: list[tuple[str, FileTuple]] = [] + self._idempotency_safe: bool = True # Enabled dy default + self._domain: str = from_email.rsplit("@", maxsplit=1)[-1] if "@" in from_email else "" def add_recipient(self, email: str, recipient_type: str = "to") -> Self: """Add a recipient (to, cc, bcc). @@ -57,6 +180,7 @@ def set_subject(self, subject: str) -> Self: Returns: The builder instance. """ + SecurityGuard.validate_no_control_characters(subject, "Subject") self._payload["subject"] = subject return self @@ -118,6 +242,8 @@ def add_custom_header(self, key: str, value: str) -> Self: Returns: The builder instance. """ + SecurityGuard.validate_no_control_characters(key, "Custom Header Key") + SecurityGuard.validate_no_control_characters(value, "Custom Header Value") self._payload[f"h:{key}"] = value return self @@ -134,6 +260,9 @@ def add_option(self, key: str, *, value: bool | str) -> Self: def attach_file(self, file_path: str | Path, safe_base_dir: str | Path | None = None) -> Self: """Safely attach a file to the email, protected against Path Traversal and OOM. + Standard attachment upload (Reads entire file into memory). + Useful for small files (logos, receipts). + Returns: The builder instance. """ @@ -146,14 +275,31 @@ def attach_file(self, file_path: str | Path, safe_base_dir: str | Path | None = # 2. Apply CWE-400 Memory Guardrail (Fail-fast if > 25MB) SecurityGuard.check_file_size(path) + content_type, _ = mimetypes.guess_type(str(path)) + if not content_type: + content_type = "application/octet-stream" + # 3. Read into memory for the multipart payload - file_data = path.read_bytes() - self._files.append(("attachment", (path.name, file_data))) + file_bytes = path.read_bytes() + self._files.append(("attachment", (path.name, file_bytes, content_type))) return self - def attach_inline(self, file_path: str | Path, safe_base_dir: str | Path | None = None) -> Self: - """Safely attach an inline image/file, protected against Path Traversal and OOM. + def attach_stream( + self, + file_path: str | Path, + safe_base_dir: str | Path | None = None, + chunk_size: int = CHUNK_SIZE, + ) -> Self: + """Memory-safe streamed attachment upload (CWE-400 protection). + + Uses ChunkedStreamer to lazily read the file. Highly recommended + for large PDFs, videos, or datasets (up to 25MB). + + Args: + file_path: Path to the target file. + safe_base_dir: Guardrail directory to prevent Path Traversal. + chunk_size: Bytes to read per iteration (default 512KB). Returns: The builder instance. @@ -161,10 +307,53 @@ def attach_inline(self, file_path: str | Path, safe_base_dir: str | Path | None path = Path(file_path) if safe_base_dir: - path = SecurityGuard.validate_attachment_path(path, safe_base_dir) + SecurityGuard.validate_attachment_path(path, safe_base_dir) + + SecurityGuard.check_file_size(path) + + content_type, _ = mimetypes.guess_type(str(path)) + if not content_type: + content_type = "application/octet-stream" + + streamer = ChunkedStreamer(path, chunk_size=chunk_size) + + self._files.append(("attachment", (path.name, streamer, content_type))) + + return self + + def attach_inline( + self, file_path: str | Path, cid: str | None = None, safe_base_dir: str | Path | None = None + ) -> Self: + """Safely prepare and map an inline image attachment with an explicit Content-ID. + + This method instantly reads the file context into memory and terminates + the file handler to prevent descriptor leaks in high-concurrency runtimes. + + Args: + file_path: The local absolute or relative path to the target image/file. + cid: Optional custom Content-ID alias. If omitted, the filename is used as the CID. + safe_base_dir: Guardrail path directory to insulate against Path Traversal (CWE-22). + + Returns: + The builder instance for fluent call chaining. + """ + path = Path(file_path) + + if safe_base_dir: + SecurityGuard.validate_attachment_path(path, safe_base_dir) + SecurityGuard.check_file_size(path) - self._files.append(("inline", (path.name, path.read_bytes()))) + target_cid = cid or path.name + + content_type, _ = mimetypes.guess_type(str(path)) + if not content_type: + content_type = "application/octet-stream" + + file_bytes = path.read_bytes() + + self._files.append(("inline", (target_cid, file_bytes, content_type))) + return self def set_template_version(self, version: str) -> Self: @@ -206,7 +395,28 @@ def set_recipient_variables(self, variables: dict[str, dict[str, Any]]) -> Self: self._payload["recipient-variables"] = json.dumps(variables, separators=(",", ":")) return self - def build(self) -> tuple[dict[str, Any], list[tuple[str, tuple[str, bytes]]] | None]: + def check_deliverability(self) -> dict[str, float | list[str] | bool] | SpamReport: + """Performs a local, zero-network static analysis of the current HTML payload to detect common structural spam triggers. + + Returns: + A dictionary containing a deliverability score and a list of identified issues. + """ + html_payload = self._payload.get("html") + if not html_payload: + return {"score": 100.0, "issues": ["No HTML content to analyze."], "is_safe": True} + + return SpamGuard.check_html(html_payload) + + def set_idempotency_safe(self, *, enabled: bool) -> Self: + """Allows you to force-disable the automatic generation of the idempotency key. + + Returns: + The builder instance. + """ + self._idempotency_safe = enabled + return self + + def build(self) -> tuple[dict[str, Any], list[tuple[str, FileTuple]] | None]: """Finalize the payload for the sync and async clients. Returns: @@ -214,6 +424,12 @@ def build(self) -> tuple[dict[str, Any], list[tuple[str, tuple[str, bytes]]] | N """ final_payload = self._payload.copy() + if self._idempotency_safe and "h:X-Idempotency-Key" not in final_payload: + idempotency_key = IdempotencyGuard.generate_key( + self._domain, final_payload, self._files + ) + final_payload["h:X-Idempotency-Key"] = idempotency_key + for key in ["to", "cc", "bcc"]: if key in final_payload and isinstance(final_payload[key], list): # Only collapse into a string if the list actually has items diff --git a/mailgun/client.py b/mailgun/client.py index c1340a1..97054e7 100644 --- a/mailgun/client.py +++ b/mailgun/client.py @@ -16,27 +16,21 @@ from __future__ import annotations +import contextlib import ssl -import sys import warnings -from typing import TYPE_CHECKING, Any, Final +from http import HTTPStatus +from typing import TYPE_CHECKING, Any, Final, Self -import httpx import requests # pyright: ignore[reportMissingModuleSource] -from urllib3.util.retry import Retry +from mailgun._httpx_compat import httpx from mailgun.config import Config from mailgun.endpoints import AsyncEndpoint, BaseEndpoint, Endpoint from mailgun.filters import RedactingFilter from mailgun.security import SecretAuth, SecureHTTPAdapter, SecurityGuard -if sys.version_info >= (3, 11): - from typing import Self -else: - from typing_extensions import Self - - if TYPE_CHECKING: import types @@ -88,7 +82,11 @@ def __init__( **kwargs: Additional configuration parameters, such as 'api_url'. """ self.auth = SecurityGuard.validate_auth(auth) - self.config = Config(api_url=kwargs.get("api_url")) + self.config = Config( + api_url=kwargs.get("api_url"), + dry_run=kwargs.get("dry_run", False), + retry_policy=kwargs.get("retry_policy"), + ) # DX Guardrail: Constructor Deprecation Interceptions if "api_version" in kwargs: @@ -168,15 +166,8 @@ def _build_resilient_session() -> requests.Session: A configured requests.Session instance. """ session = requests.Session() - retry_strategy = Retry( - total=3, - backoff_factor=1, - status_forcelist=[429, 500, 502, 503, 504], - allowed_methods=["GET", "OPTIONS", "HEAD"], - ) - adapter = SecureHTTPAdapter( - max_retries=retry_strategy, pool_connections=100, pool_maxsize=100 - ) + + adapter = SecureHTTPAdapter(pool_connections=100, pool_maxsize=100) session.mount("https://", adapter) session.mount("http://", adapter) return session @@ -202,26 +193,33 @@ def __getattr__(self, name: str) -> Any: try: url, headers = self.config[name] - return Endpoint( + endpoint = Endpoint( url=url, headers=headers, auth=self.auth, session=self._session, timeout=self.timeout, + dry_run=self.config.dry_run, ) + except KeyError: # __getattr__ must return AttributeError msg = f"'{self.__class__.__name__}' object has no attribute '{name}'" raise AttributeError(msg) from None + endpoint.retry_policy = getattr(self.config, "retry_policy", None) + return endpoint + def close(self) -> None: """Close the underlying requests.Session connection pool and purge memory.""" - if self._session: + # Safely fetch without triggering AttributeError on unbound slots + session = getattr(self, "_session", None) + if session: try: # CWE-316: Clear session resources - self._session.auth = None - self._session.headers.clear() - self._session.close() + session.auth = None + session.headers.clear() + session.close() finally: self._session = None self.auth = None @@ -243,6 +241,39 @@ def __exit__( """Exit the synchronous context manager, ensuring connection pools are closed.""" self.close() + def __del__(self) -> None: + """Emit a ResourceWarning if the client is garbage-collected without being closed.""" + if getattr(self, "_session", None) is not None: + warnings.warn( + "Unclosed Client detected. Please use the client as a context manager or call client.close() explicitly.", + ResourceWarning, + stacklevel=2, + ) + with contextlib.suppress(Exception): + self.close() + + def ping(self) -> bool: + """Perform a fast, low-overhead health check to verify API credentials. + + This checks network connectivity to the Mailgun infrastructure. + This method is fail-safe: it will never raise network exceptions or + authentication errors to the application layer. Instead, it returns a + clean boolean value, making it ideal for container readiness probes. + + Returns: + bool: True if the connection succeeds and credentials are valid (HTTP 200), + False on network timeouts, DNS drops, or invalid API keys. + """ + try: + # Query the domains endpoint with a strict limit of 1 + response = self.domains.get(filters={"limit": 1}) + except Exception: # noqa: BLE001 - Explicitly failing closed on readiness probe + return False + else: + if hasattr(response, "status_code"): + return bool(response.status_code == HTTPStatus.OK) + return False + # ============================================================================== # 2. ASYNC IMPLEMENTATION @@ -291,17 +322,23 @@ def __getattr__(self, name: str) -> Any: try: url, headers = self.config[name] - return AsyncEndpoint( + + endpoint = AsyncEndpoint( url=url, headers=headers, auth=self.auth, client=self._client, timeout=self.timeout, + dry_run=self.config.dry_run, ) + except KeyError: msg = f"'{self.__class__.__name__}' object has no attribute '{name}'" raise AttributeError(msg) from None + endpoint.retry_policy = getattr(self.config, "retry_policy", None) + return endpoint + @property def _client(self) -> httpx.AsyncClient: """Provide lazy initialization for the underlying httpx.AsyncClient. @@ -364,3 +401,36 @@ async def __aexit__( exc_tb: The traceback associated with the exception. """ await self.aclose() + + def __del__(self) -> None: + """Safety net for unclosed sockets (CWE-400) if context managers are skipped.""" + if self._httpx_client is not None and not self._httpx_client.is_closed: + warnings.warn( + f"Unclosed {self.__class__.__name__} detected. You must explicitly " + "call '.aclose()' or use the 'async with' context manager to prevent " + "socket and memory leaks.", + ResourceWarning, + stacklevel=2, + ) + + async def ping(self) -> bool: + """Perform a fast, low-overhead health check to verify API credentials. + + This checks network connectivity to the Mailgun infrastructure. + This method is fail-safe: it will never raise network exceptions or + authentication errors to the application layer. Instead, it returns a + clean boolean value, making it ideal for container readiness probes. + + Returns: + bool: True if the connection succeeds and credentials are valid (HTTP 200), + False on network timeouts, DNS drops, or invalid API keys. + """ + try: + # Query the domains endpoint with a strict limit of 1 + response = await self.domains.get(filters={"limit": 1}) + except Exception: # noqa: BLE001 - Explicitly failing closed on readiness probe + return False + else: + if hasattr(response, "status_code"): + return bool(response.status_code == HTTPStatus.OK) + return False diff --git a/mailgun/config.py b/mailgun/config.py index 718fc6d..f4c4482 100644 --- a/mailgun/config.py +++ b/mailgun/config.py @@ -1,7 +1,8 @@ from __future__ import annotations +import random import sys -from enum import Enum +from enum import StrEnum from functools import lru_cache from types import MappingProxyType from typing import TYPE_CHECKING, Any, Final @@ -25,7 +26,7 @@ logger = get_logger(__name__) -@lru_cache +@lru_cache(maxsize=256) def _get_cached_route_data(clean_key: str) -> dict[str, Any]: """Apply internal cached routing logic. @@ -60,7 +61,7 @@ def _get_cached_route_data(clean_key: str) -> dict[str, Any]: return {"version": APIVersion.V3.value, "keys": tuple(route_parts)} -class APIVersion(str, Enum): +class APIVersion(StrEnum): """Constants for Mailgun API versions.""" V1 = "v1" @@ -70,13 +71,52 @@ class APIVersion(str, Enum): V5 = "v5" +class RetryPolicy: + """Deterministic exponential backoff engine.""" + + __slots__ = ("base_delay", "max_delay", "max_retries", "respect_retry_after") + + def __init__( + self, + max_retries: int = 3, + base_delay: float = 1.0, + max_delay: float = 10.0, + *, + respect_retry_after: bool = True, + ) -> None: + """Initialize the RetryPolicy engine. + + Args: + max_retries: The maximum number of retry attempts. + base_delay: The base multiplier for exponential backoff in seconds. + max_delay: The absolute maximum delay cap in seconds. + respect_retry_after: Whether to parse and respect 429 Retry-After headers. + """ + self.max_retries: Final = max_retries + self.base_delay: Final = base_delay + self.max_delay: Final = max_delay + self.respect_retry_after: Final = respect_retry_after + + def calculate_delay(self, attempt: int) -> float: + """Calculates exponential backoff with random jitter to prevent collisions. + + Args: + attempt: The current retry attempt number (0-indexed). + + Returns: + A float representing the sleep delay in seconds before the next attempt. + """ + backoff = min(self.max_delay, self.base_delay * (2**attempt)) + return random.uniform(0, backoff) # noqa: S311 - Randomness used for network jitter, not crypto. + + class Config: """Configuration engine for the Mailgun API client. Using a data-driven routing approach. """ - __slots__ = ("_baked_urls", "api_url", "dry_run", "ex_handler") + __slots__ = ("_baked_urls", "api_url", "dry_run", "ex_handler", "retry_policy") DEFAULT_API_URL: Final[str] = "https://api.mailgun.net" USER_AGENT: Final[str] = f"mailgun-api-python/{__version__}" @@ -102,7 +142,15 @@ class Config: _V3_ENDPOINTS: Final[frozenset[str]] = frozenset(routes.DOMAIN_ENDPOINTS["v3"]) _V4_ENDPOINTS: Final[frozenset[str]] = frozenset(routes.DOMAIN_ENDPOINTS.get("v4", [])) - def __init__(self, api_url: str | None = None, *, dry_run: bool = False) -> None: + _audit_hook_enabled: bool = False + + def __init__( + self, + api_url: str | None = None, + *, + dry_run: bool = False, + retry_policy: RetryPolicy | None = None, + ) -> None: """Initialize the configuration engine. Args: @@ -110,9 +158,11 @@ def __init__(self, api_url: str | None = None, *, dry_run: bool = False) -> None dry_run: Prevents network execution and intercepts requests locally. """ self.ex_handler: bool = True + self.dry_run: bool = dry_run - base_url_input: str = api_url or self.DEFAULT_API_URL + self.retry_policy: RetryPolicy = retry_policy or RetryPolicy() + base_url_input: str = api_url or self.DEFAULT_API_URL self.api_url: str = self._normalize_api_url(base_url_input) self._baked_urls: Final[dict[str, str]] = { @@ -274,6 +324,8 @@ def enable_security_audit(cls) -> None: Enterprise security teams can enable this during SDK boot to gain instant visibility into API requests sent via the SDK without altering standard logs. """ + if cls._audit_hook_enabled: + return def audit_hook(event: str, args: tuple[Any, ...]) -> None: if event == "mailgun.api.request": @@ -281,4 +333,5 @@ def audit_hook(event: str, args: tuple[Any, ...]) -> None: logger.info("SECURITY AUDIT: Outbound API call tracked - %s %s", method, url) sys.addaudithook(audit_hook) + cls._audit_hook_enabled = True logger.info("Mailgun Security Audit Hooks Enabled.") diff --git a/mailgun/endpoints.py b/mailgun/endpoints.py index b39484f..15595b3 100644 --- a/mailgun/endpoints.py +++ b/mailgun/endpoints.py @@ -1,20 +1,21 @@ from __future__ import annotations +import asyncio import json import sys +import time import warnings from functools import lru_cache +from http import HTTPStatus from typing import TYPE_CHECKING, Any, Final from urllib.parse import parse_qs, urlparse -import httpx -import requests -from requests.exceptions import ( - ConnectionError as RequestsConnectionError, # pyright: ignore[reportMissingModuleSource] -) +import requests # pyright: ignore[reportMissingModuleSource] from requests.models import Response # pyright: ignore[reportMissingModuleSource] from mailgun import routes +from mailgun._httpx_compat import httpx +from mailgun.config import RetryPolicy from mailgun.handlers.error_handler import ApiError, MailgunTimeoutError from mailgun.logger import get_logger from mailgun.security import SecurityGuard @@ -23,9 +24,7 @@ if TYPE_CHECKING: from collections.abc import Callable, Iterable, Mapping - from httpx import Response as HttpxResponse - - from mailgun.types import TimeoutType + from mailgun.types import APIResponseType, AsyncAPIResponseType, TimeoutType logger = get_logger(__name__) @@ -45,7 +44,12 @@ def build_path_from_keys(keys: Iterable[str]) -> str: if not keys: return "" keys_seq = keys if isinstance(keys, (list, tuple)) else list(keys) - return "".join(f"/{SecurityGuard.sanitize_path_segment(k)}" for k in keys_seq if k) + # Safely evaluate truthiness to prevent dropping `0` integer IDs + return "".join( + f"/{SecurityGuard.sanitize_path_segment(str(k))}" + for k in keys_seq + if k is not None and str(k).strip() + ) @lru_cache(maxsize=32) @@ -165,7 +169,7 @@ def _load_handler(endpoint_key: str) -> Callable[..., str]: # noqa: PLR0911, PL class BaseEndpoint: """Base class for endpoints. Contains methods common for Endpoint and AsyncEndpoint.""" - __slots__ = ("_auth", "_timeout", "_url", "dry_run", "headers") + __slots__ = ("_auth", "_timeout", "_url", "dry_run", "headers", "retry_policy") def __init__( self, @@ -190,6 +194,7 @@ def __init__( self._auth = auth self._timeout = timeout self.dry_run = dry_run + self.retry_policy = None @staticmethod def _warn_if_deprecated(method: str, target_url: str) -> None: @@ -211,6 +216,21 @@ def _warn_if_deprecated(method: str, target_url: str) -> None: logger.warning(warning_message) break + @staticmethod + def _reset_stream_pointers(files: Any) -> None: + """Ensure the idempotency of file generators and buffers during retries.""" + if not isinstance(files, list): + return + for _, file_tuple in files: + if isinstance(file_tuple, tuple) and len(file_tuple) >= 2: # noqa: PLR2004 + file_obj = file_tuple[1] + # If it's our ChunkedStreamer, close current FD, so __iter__ open it again + if hasattr(file_obj, "close") and hasattr(file_obj, "chunk_size"): + file_obj.close() + # If it's BytesIO or an opened file + elif hasattr(file_obj, "seek"): + file_obj.seek(0) + def __repr__(self) -> str: """DX: Show the actual resolved target route instead of memory address. @@ -266,7 +286,9 @@ def _merge_headers(self, kwargs: dict[str, Any]) -> dict[str, str]: if custom_headers and isinstance(custom_headers, dict): req_headers.update(custom_headers) - return req_headers + # CWE-400 / Crash Prevention: Enforce string keys and values to + # prevent HTTP protocol serialization crashes in requests/httpx. + return {str(k): str(v) for k, v in req_headers.items()} def _prepare_request( self, @@ -294,11 +316,14 @@ def _prepare_request( safe_kwargs = SecurityGuard.filter_safe_kwargs(kwargs) safe_headers = SecurityGuard.sanitize_headers(headers) or {} target_domain = SecurityGuard.sanitize_domain(domain) + target_domain_normalized = SecurityGuard.normalize_domain(target_domain) actual_timeout = timeout if timeout is not None else self._timeout safe_timeout = SecurityGuard.sanitize_timeout(actual_timeout) - target_url = self.build_url(url, domain=target_domain, method=safe_method, **kwargs) + target_url = self.build_url( + url, domain=target_domain_normalized, method=safe_method, **kwargs + ) self._warn_if_deprecated(safe_method, target_url) # PEP 578 and protection against Log Forging (CWE-117) @@ -335,7 +360,7 @@ def __init__( super().__init__(url, headers, auth, timeout=timeout, dry_run=dry_run) self._session = session or requests.Session() - def api_call( + def api_call( # noqa: PLR0914, PLR0915 self, auth: tuple[str, str] | None, method: str, @@ -347,7 +372,7 @@ def api_call( files: Any | None = None, domain: str | None = None, **kwargs: Any, - ) -> Response | Any: + ) -> APIResponseType: # noqa: PLR0914, PLR0915 - Core request loop contains complex retry/sandbox logic """Execute the HTTP request to the Mailgun API. Args: @@ -375,17 +400,24 @@ def api_call( SecurityGuard.validate_no_control_characters(target_url, context="Endpoint URL") - # Zero-Leak Sandbox Mode Interception + # --- DRY RUN INTERCEPTOR (SYNC) --- if self.dry_run: logger.info( - "DRY RUN: Intercepting %s request to %s", safe_method.upper(), safe_url_for_log + "DRY RUN: Intercepting sync %s request to %s", + safe_method.upper(), + safe_url_for_log, ) mock_resp = Response() - mock_resp.status_code = 200 - mock_resp.encoding = "utf-8" + mock_resp.status_code = HTTPStatus.OK mock_resp._content = b'{"message": "Dry run successful - request intercepted", "id": ""}' # noqa: SLF001 + mock_resp.encoding = "utf-8" + mock_resp.url = target_url return mock_resp + # Ensure protocol consistency: HTTP libraries MUST generate their own multipart boundaries + if files and safe_headers: + safe_headers = {k: v for k, v in safe_headers.items() if k.lower() != "content-type"} + # Case-insensitive validation for Content-Type to conform with RFC 7230 is_json_request = any( k.lower() == "content-type" and "application/json" in str(v).lower() @@ -397,57 +429,102 @@ def api_call( req_method = getattr(self._session, safe_method.lower()) + policy = getattr(self, "retry_policy", None) or RetryPolicy() + max_attempts = policy.max_retries + 1 + sys.audit("mailgun.api.request", safe_method.upper(), safe_url_for_log) logger.debug("Sending Request: %s %s", safe_method.upper(), safe_url_for_log) - try: - response = req_method( - target_url, - data=data, - params=filters, - headers=safe_headers, - auth=auth, - timeout=safe_timeout, - files=files, - verify=True, - stream=False, - allow_redirects=False, - **safe_kwargs, - ) - - status_code = getattr(response, "status_code", 200) - is_error = isinstance(status_code, int) and status_code >= _HTTP_ERROR_THRESHOLD - if is_error: - logger.error( - "API Error %s | %s %s", status_code, safe_method.upper(), safe_url_for_log - ) - else: - logger.debug( - "API Success %s | %s %s", - getattr(response, "status_code", 200), - safe_method.upper(), + for attempt in range(max_attempts): + try: + response = req_method( target_url, + data=data, + params=filters, + headers=safe_headers, + auth=auth, + timeout=safe_timeout, + files=files, + verify=True, + stream=False, + allow_redirects=False, + **safe_kwargs, ) - except requests.exceptions.Timeout as e: - logger.exception("Timeout Error: %s %s", safe_method.upper(), safe_url_for_log) - raise MailgunTimeoutError("Request timed out") from e - except RequestsConnectionError as e: - logger.critical("Connection Failed (DNS/Network): %s | URL: %s", e, safe_url_for_log) - msg = f"Network routing failed: {e}" - raise ApiError(msg) from e - except requests.RequestException as e: - logger.critical("Request Exception: %s | URL: %s", e, safe_url_for_log) - raise ApiError(e) from e - else: + status_code = getattr(response, "status_code", 200) + is_transient_error = status_code in {429, 500, 502, 503, 504} + + # Логіка Retry Policy + if is_transient_error and attempt < max_attempts - 1: + delay = policy.calculate_delay(attempt) + + if status_code == HTTPStatus.TOO_MANY_REQUESTS and policy.respect_retry_after: + retry_after = response.headers.get("Retry-After") + if retry_after and retry_after.isdigit(): + # Clamp the delay to prevent infinite sleeping (CWE-400) + delay = min(float(retry_after), policy.max_delay) + + logger.warning( + "API Transient Error %s | Retrying in %.2fs (Attempt %d/%d) | URL: %s", + status_code, + delay, + attempt + 1, + policy.max_retries, + safe_url_for_log, + ) + self._reset_stream_pointers(files) + time.sleep(delay) + continue + + # Фінальна обробка після виходу з циклу ретраїв + is_error = isinstance(status_code, int) and status_code >= _HTTP_ERROR_THRESHOLD + if is_error: + logger.error( + "API Error %s | %s %s", status_code, safe_method.upper(), safe_url_for_log + ) + else: + logger.debug( + "API Success %s | %s %s", status_code, safe_method.upper(), target_url + ) + + except requests.RequestException as e: + if attempt < max_attempts - 1: + delay = policy.calculate_delay(attempt) + + logger.warning( + "Network Error: %s | Retrying in %.2fs (Attempt %d/%d) | URL: %s", + e, + delay, + attempt + 1, + policy.max_retries, + safe_url_for_log, + ) + + self._reset_stream_pointers(files) + + time.sleep(delay) + + continue + + if isinstance(e, requests.exceptions.Timeout): + logger.exception("Timeout Error: %s %s", safe_method.upper(), safe_url_for_log) + raise MailgunTimeoutError("Request timed out") from e + + logger.critical( + "Connection Failed (DNS/Network): %s | URL: %s", e, safe_url_for_log + ) + msg = f"Network routing failed: {e}" + raise ApiError(msg) from e return response + return None + def get( self, filters: Mapping[str, str | Any] | None = None, domain: str | None = None, **kwargs: Any, - ) -> Response: + ) -> APIResponseType: """Send a GET request to retrieve resources. Args: @@ -477,7 +554,7 @@ def create( headers: Any = None, files: Any | None = None, **kwargs: Any, - ) -> Response: + ) -> APIResponseType: """Send a POST request to create a new resource or execute an action. Args: @@ -509,7 +586,7 @@ def create( def put( self, data: Any | None = None, filters: Mapping[str, str | Any] | None = None, **kwargs: Any - ) -> Response: + ) -> APIResponseType: """Send a PUT request to update or replace a resource. Args: @@ -533,7 +610,7 @@ def put( def patch( self, data: Any | None = None, filters: Mapping[str, str | Any] | None = None, **kwargs: Any - ) -> Response: + ) -> APIResponseType: """Send a PATCH request to partially update a resource. Args: @@ -557,7 +634,7 @@ def patch( def update( self, data: Any | None, filters: Mapping[str, str | Any] | None = None, **kwargs: Any - ) -> Response: + ) -> APIResponseType: """Send a PUT request specifically structured for updating resources with dynamic headers. Args: @@ -579,7 +656,7 @@ def update( **kwargs, ) - def delete(self, domain: str | None = None, **kwargs: Any) -> Response: + def delete(self, domain: str | None = None, **kwargs: Any) -> APIResponseType: """Send a DELETE request to remove a resource. Args: @@ -633,7 +710,11 @@ def stream( # Mailgun returns a full URL. Parse it to extract just the new pagination parameters # (like 'page' or 'url') so the next self.get() call works correctly. query_params = parse_qs(urlparse(next_url).query) - current_filters.update({k: v[0] for k, v in query_params.items()}) + for k, v in query_params.items(): + if not v: + continue + # If Mailgun returned multiple values (e.g., multiple tags), preserve the list + current_filters[k] = v[0] if len(v) == 1 else v # ============================================================================== @@ -669,7 +750,7 @@ def __init__( super().__init__(url, headers, auth, timeout=timeout, dry_run=dry_run) self._client = client or httpx.AsyncClient() - async def api_call( + async def api_call( # noqa: PLR0914, PLR0915 self, auth: tuple[str, str] | None, method: str, @@ -677,11 +758,11 @@ async def api_call( headers: dict[str, str], data: Any | None = None, filters: Mapping[str, str | Any] | None = None, - timeout: TimeoutType = None, + timeout: TimeoutType = None, # noqa: ASYNC109 files: Any | None = None, domain: str | None = None, **kwargs: Any, - ) -> HttpxResponse: + ) -> AsyncAPIResponseType: # noqa: PLR0914, PLR0915 """Execute the asynchronous HTTP request to the Mailgun API. Args: @@ -709,7 +790,7 @@ async def api_call( SecurityGuard.validate_no_control_characters(target_url, context="Endpoint URL") - # Zero-Leak Sandbox Mode Interception + # --- DRY RUN INTERCEPTOR (ASYNC) --- if self.dry_run: logger.info( "DRY RUN: Intercepting async %s request to %s", @@ -756,48 +837,82 @@ async def api_call( else: request_kwargs["data"] = data + policy = getattr(self, "retry_policy", None) or RetryPolicy() + max_attempts = policy.max_retries + 1 + # PEP 578 and protection against Log Forging (CWE-117) sys.audit("mailgun.api.request", safe_method.upper(), safe_url_for_log) logger.debug("Sending Async Request: %s %s", safe_method.upper(), safe_url_for_log) - try: - response = await self._client.request(**request_kwargs) - - status_code = getattr(response, "status_code", 200) - is_error = isinstance(status_code, int) and status_code >= _HTTP_ERROR_THRESHOLD - if is_error: - logger.error( - "API Error %s | %s %s", status_code, safe_method.upper(), safe_url_for_log - ) - else: - logger.debug( - "API Success %s | %s %s", - getattr(response, "status_code", 200), - safe_method.upper(), - target_url, - ) + for attempt in range(max_attempts): + try: + response = await self._client.request(**request_kwargs) + + status_code = getattr(response, "status_code", 200) + is_transient_error = status_code in {429, 500, 502, 503, 504} + + if is_transient_error and attempt < max_attempts - 1: + delay = policy.calculate_delay(attempt) + + if status_code == HTTPStatus.TOO_MANY_REQUESTS and policy.respect_retry_after: + retry_after = response.headers.get("Retry-After") + if retry_after and retry_after.isdigit(): + # Clamp the delay to prevent infinite sleeping (CWE-400) + delay = min(float(retry_after), policy.max_delay) + + logger.warning( + "API Transient Error %s | Async Retrying in %.2fs (Attempt %d/%d)", + status_code, + delay, + attempt + 1, + policy.max_retries, + ) + self._reset_stream_pointers(files) + await asyncio.sleep(delay) + continue + + is_error = isinstance(status_code, int) and status_code >= _HTTP_ERROR_THRESHOLD + if is_error: + logger.error( + "API Error %s | %s %s", status_code, safe_method.upper(), safe_url_for_log + ) + else: + logger.debug( + "API Success %s | %s %s", status_code, safe_method.upper(), target_url + ) + + except (httpx.TimeoutException, httpx.ConnectError, httpx.RequestError) as e: + if attempt < max_attempts - 1: + delay = policy.calculate_delay(attempt) + logger.warning( + "Async Network Error: %s | Retrying in %.2fs (Attempt %d/%d)", + e, + delay, + attempt + 1, + policy.max_retries, + ) + self._reset_stream_pointers(files) + await asyncio.sleep(delay) + continue + + if isinstance(e, httpx.TimeoutException): + logger.exception("Timeout Error: %s %s", safe_method.upper(), safe_url_for_log) + raise MailgunTimeoutError("Request timed out") from e + + logger.critical("Async Connection Failed: %s | URL: %s", e, safe_url_for_log) + msg = f"Network routing failed: {e}" + raise ApiError(msg) from e - except httpx.TimeoutException as e: - logger.exception("Timeout Error: %s %s", safe_method.upper(), safe_url_for_log) - raise MailgunTimeoutError("Request timed out") from e - except httpx.ConnectError as e: - logger.critical( - "Async Connection Failed (DNS/Network): %s | URL: %s", e, safe_url_for_log - ) - msg = f"Network routing failed: {e}" - raise ApiError(msg) from e - except httpx.RequestError as e: - logger.critical("Request Exception: %s | URL: %s", e, safe_url_for_log) - raise ApiError(e) from e - else: return response + return None + async def get( self, filters: Mapping[str, str | Any] | None = None, domain: str | None = None, **kwargs: Any, - ) -> HttpxResponse: + ) -> AsyncAPIResponseType: """Send an asynchronous GET request to retrieve resources. Args: @@ -827,7 +942,7 @@ async def create( headers: Any = None, files: Any | None = None, **kwargs: Any, - ) -> HttpxResponse: + ) -> AsyncAPIResponseType: """Send an asynchronous POST request to create a new resource or execute an action. Args: @@ -859,7 +974,7 @@ async def create( async def put( self, data: Any | None = None, filters: Mapping[str, str | Any] | None = None, **kwargs: Any - ) -> HttpxResponse: + ) -> AsyncAPIResponseType: """Send an asynchronous PUT request to update or replace a resource. Args: @@ -883,7 +998,7 @@ async def put( async def patch( self, data: Any | None = None, filters: Mapping[str, str | Any] | None = None, **kwargs: Any - ) -> HttpxResponse: + ) -> AsyncAPIResponseType: """Send an asynchronous PATCH request to partially update a resource. Args: @@ -907,7 +1022,7 @@ async def patch( async def update( self, data: Any | None, filters: Mapping[str, str | Any] | None = None, **kwargs: Any - ) -> HttpxResponse: + ) -> AsyncAPIResponseType: """Send an asynchronous PUT request specifically structured for updating resources with dynamic headers. Args: @@ -930,7 +1045,7 @@ async def update( **kwargs, ) - async def delete(self, domain: str | None = None, **kwargs: Any) -> httpx.Response: + async def delete(self, domain: str | None = None, **kwargs: Any) -> AsyncAPIResponseType: """Send an asynchronous DELETE request to remove a resource. Args: @@ -974,4 +1089,7 @@ async def stream( break query_params = parse_qs(urlparse(next_url).query) - current_filters.update({k: v[0] for k, v in query_params.items()}) + for k, v in query_params.items(): + if not v: + continue + current_filters[k] = v[0] if len(v) == 1 else v diff --git a/mailgun/examples/bounce_classification_examples.py b/mailgun/examples/bounce_classification_examples.py index 6619ee0..50ae212 100644 --- a/mailgun/examples/bounce_classification_examples.py +++ b/mailgun/examples/bounce_classification_examples.py @@ -1,6 +1,7 @@ """Examples for Mailgun Bounce Classification API.""" import asyncio +import json import os from typing import Any @@ -47,7 +48,7 @@ def post_list_statistic_v2_sync(api_key: str, domain: str) -> None: headers: dict[str, str] = {"Content-Type": "application/json"} with Client(auth=("api", api_key)) as client: - req = client.bounce_classification.create(data=payload, headers=headers) + req = client.bounce_classification.create(data=json.dumps(payload), headers=headers) print(req.json()) diff --git a/mailgun/examples/builder_examples.py b/mailgun/examples/builder_examples.py index 7f24905..6ce9bfd 100644 --- a/mailgun/examples/builder_examples.py +++ b/mailgun/examples/builder_examples.py @@ -1,10 +1,16 @@ """Examples for Mailgun Message Builders and Clients.""" import asyncio +import logging import os from mailgun.builders import MailgunMessageBuilder from mailgun.client import AsyncClient, Client +from mailgun.handlers.error_handler import DeliverabilityError + + +logging.basicConfig(level=logging.INFO, format="%(levelname)s - %(message)s") +logger = logging.getLogger(__name__) def send_standard_email_sync(api_key: str, domain: str) -> None: @@ -13,9 +19,10 @@ def send_standard_email_sync(api_key: str, domain: str) -> None: (Synchronous Execution) """ print("\n--- Sending Standard Email (Sync) ---") + payload, files = ( MailgunMessageBuilder(f"support@{domain}") - .add_recipient("user1@example.com") + .add_recipient(MESSAGES_TO) .set_subject("Your Monthly Invoice") .set_text("Please find your invoice attached.") .set_html("

Please find your invoice attached.

") @@ -40,7 +47,7 @@ async def send_template_email_async(api_key: str, domain: str) -> None: print("\n--- Sending Template Email (Async) ---") payload, files = ( MailgunMessageBuilder(f"marketing@{domain}") - .add_recipient("user2@example.com") + .add_recipient(MESSAGES_TO) .set_subject("Special Offer Inside!") .set_template("promo-template") .set_template_version("v2") @@ -63,7 +70,7 @@ def send_batch_email_sync(api_key: str, domain: str) -> None: print("\n--- Sending Batch Email (Sync) ---") payload, files = ( MailgunMessageBuilder(f"newsletter@{domain}") - .add_recipient("alice@example.com") + .add_recipient(MESSAGES_TO) .add_recipient("bob@example.com") .set_subject("Hey %recipient.name%, your weekly update!") .set_text("Hi %recipient.name%, your user ID is %recipient.id%.") @@ -98,7 +105,7 @@ async def send_amp_and_inline_images_async(api_key: str, domain: str) -> None: payload, files = ( MailgunMessageBuilder(f"hello@{domain}") - .add_recipient("user3@example.com") + .add_recipient(MESSAGES_TO) .set_subject("Interactive Email") .set_html('') .set_amp_html("AMP Content") @@ -118,9 +125,182 @@ async def send_amp_and_inline_images_async(api_key: str, domain: str) -> None: os.remove(dummy_image_path) +def send_marketing_campaign(api_key: str, domain: str): + html_content = """ + + +

Welcome!

+ + + + + + """ + builder = MailgunMessageBuilder(f"support@{domain}").set_html(html_content) + + report = builder.check_deliverability() + + if not report["is_safe"]: + raise DeliverabilityError(score=report["score"], issues=report["issues"]) + + logger.info("Template is safe. Proceeding to send...") + + payload, files = ( + builder.add_recipient(MESSAGES_TO) + .set_subject("Your Monthly Invoice") + .set_text("Please find your invoice attached.") + .build() + ) + print(f"Payload: {payload}") + + with Client(auth=("api", api_key)) as client: + req = client.messages.create(domain=domain, data=payload, files=files) + print(req.json()) + + +def send_large_report_sync(api_key: str, domain: str) -> None: + """ + Example: Sending a massive 20MB monthly report safely without spiking RAM. + """ + print("\n--- Sending Large Report Safely ---") + + test_file = "large_report.pdf" + with open(test_file, "wb") as f: + f.write(os.urandom(20 * 1024 * 1024)) + + try: + payload, files = ( + MailgunMessageBuilder(f"mailgun@{domain}") + .add_recipient(MESSAGES_TO) + .set_subject("Monthly Enterprise Report") + .set_text("Here is the 20MB data export.") + .attach_stream(test_file) + .build() + ) + + # Increase read/write timeout to 300 sec, + # but keep 10 sec for connection timeout. + custom_timeout = (10.0, 300.0) + + with Client(auth=("api", api_key), timeout=custom_timeout) as client: + req = client.messages.create(domain=domain, data=payload, files=files) + print("Success:", req.json()) + + finally: + if os.path.exists(test_file): + os.remove(test_file) + + +async def send_large_report_async(api_key: str, domain: str) -> None: + """ + Example: Asynchronously sending a massive 20MB monthly report safely + without spiking RAM or blocking the event loop (CWE-400 Defense). + """ + print("\n--- Sending Large Report Safely (Async) ---") + + test_file = "large_report_async.pdf" + + # Generate a dummy 20MB file synchronously for setup + with open(test_file, "wb") as f: + f.write(os.urandom(20 * 1024 * 1024)) + + try: + # 1. Build the payload and attach the streamer + payload, files = ( + MailgunMessageBuilder(f"mailgun@{domain}") + .add_recipient(MESSAGES_TO) + .set_subject("Monthly Enterprise Report (Async)") + .set_text("Here is the 20MB data export sent securely via asyncio.") + .attach_stream(test_file) + .build() + ) + + # Increase read/write timeout to 300 sec, + # but keep 10 sec for connection timeout. + custom_timeout = (10.0, 300.0) + + # 2. Use the AsyncClient context manager + async with AsyncClient(auth=("api", api_key), timeout=custom_timeout) as client: + # 3. Await the request. Under the hood, httpx will detect the + # ChunkedStreamer and iterate over __aiter__ automatically. + req = await client.messages.create(domain=domain, data=payload, files=files) + print("Success:", req.json()) + + finally: + # Clean up the dummy file + if os.path.exists(test_file): + os.remove(test_file) + + +def test_idempotency_guard_in_action(domain: str) -> None: + """ + Demonstration of the automatic generation of the idempotency key (IdempotencyGuard). + Proves the determinism of SHA-256 hashing when building the payload. + """ + print("\n--- 🛡️ Testing IdempotencyGuard (Client-Side Exactly-Once) ---") + + # Scenario 1: Build the original transactional email + builder1 = ( + MailgunMessageBuilder(f"mailgun@{domain}") + .add_recipient(MESSAGES_TO) + .set_subject("Invoice Payment #1024") + .set_text("Your invoice for $50.00 has been successfully paid.") + ) + payload1, _ = builder1.build() + key1 = payload1.get("h:X-Idempotency-Key") + print(f"👉 Payload 1 (Original): {key1}") + + # Scenario 2: Build an identical email (simulate a retry after network drop) + builder2 = ( + MailgunMessageBuilder(f"mailgun@{domain}") + .add_recipient(MESSAGES_TO) + .set_subject("Invoice Payment #1024") + .set_text("Your invoice for $50.00 has been successfully paid.") + ) + payload2, _ = builder2.build() + key2 = payload2.get("h:X-Idempotency-Key") + print(f"👉 Payload 2 (Duplicate): {key2}") + + # Scenario 3: Change at least one character (different invoice number) + builder3 = ( + MailgunMessageBuilder(f"mailgun@{domain}") + .add_recipient(MESSAGES_TO) + .set_subject("Invoice Payment #1025") # CHANGED! + .set_text("Your invoice for $50.00 has been successfully paid.") + ) + payload3, _ = builder3.build() + key3 = payload3.get("h:X-Idempotency-Key") + print(f"👉 Payload 3 (New email): {key3}") + + # Scenario 4: Developer explicitly disables protection + builder4 = ( + MailgunMessageBuilder(f"mailgun@{domain}") + .set_idempotency_safe(False) # DISABLED! + .add_recipient("customer@example.com") + .set_subject("Invoice Payment #1024") + ) + payload4, _ = builder4.build() + + # --- CONCLUSIONS (ASSERTIONS) --- + print("\n--- 📊 Validation Results ---") + if key1 == key2: + print( + "✅ SUCCESS: Keys 1 and 2 are identical. Mailgun will reject the duplicate upon network retry." + ) + else: + print("❌ ERROR: Duplicate keys differ!") + + if key1 != key3: + print("✅ SUCCESS: Key 3 is unique. The new email will pass safely.") + + if "h:X-Idempotency-Key" not in payload4: + print("✅ SUCCESS: Protection manually disabled. Idempotency header is missing.") + + if __name__ == "__main__": API_KEY: str = os.environ.get("APIKEY", "") DOMAIN: str = os.environ.get("DOMAIN", "") + MESSAGES_TO = os.environ.get("MESSAGES_TO") or f"success@{DOMAIN}" if not API_KEY or not DOMAIN: print("Please set the 'APIKEY' and 'DOMAIN' environment variables to run examples.") @@ -129,6 +309,18 @@ async def send_amp_and_inline_images_async(api_key: str, domain: str) -> None: send_standard_email_sync(api_key=API_KEY, domain=DOMAIN) send_batch_email_sync(api_key=API_KEY, domain=DOMAIN) + send_large_report_sync(API_KEY, DOMAIN) + + test_idempotency_guard_in_action(DOMAIN) + + try: + send_marketing_campaign(api_key=API_KEY, domain=DOMAIN) + except DeliverabilityError as e: + # The user gracefully catches the error and sees a clean, actionable message + # without a terrifying system traceback. + logger.error(f"Campaign aborted by SpamGuard:\n{e}") + # 2. Run Asynchronous Examples asyncio.run(send_template_email_async(api_key=API_KEY, domain=DOMAIN)) asyncio.run(send_amp_and_inline_images_async(api_key=API_KEY, domain=DOMAIN)) + asyncio.run(send_large_report_async(API_KEY, DOMAIN)) diff --git a/mailgun/examples/domain_examples.py b/mailgun/examples/domain_examples.py index 9da29b0..e7c8cf7 100644 --- a/mailgun/examples/domain_examples.py +++ b/mailgun/examples/domain_examples.py @@ -50,7 +50,7 @@ def get_simple_domain_sync(api_key: str, domain_name: str) -> None: :return: None """ with Client(auth=("api", api_key)) as client: - response = client.domains.get(domain_name=domain_name) + response = client.domains.get(domain=domain_name) print("GET Simple Domain:", response.json()) @@ -221,14 +221,14 @@ def get_dkim_keys_sync(api_key: str, domain_name: str) -> None: GET /v1/dkim/keys :return: None """ - data: dict[str, str] = { + params = { "page": "string", "limit": "0", "signing_domain": domain_name, "selector": "smtp", } with Client(auth=("api", api_key)) as client: - response = client.dkim_keys.get(data=data) + response = client.dkim_keys.get(filters=params) print("GET DKIM Keys:", response.json()) diff --git a/mailgun/examples/dry_run_examples.py b/mailgun/examples/dry_run_examples.py new file mode 100644 index 0000000..756021b --- /dev/null +++ b/mailgun/examples/dry_run_examples.py @@ -0,0 +1,21 @@ +import logging +from mailgun.client import Client + +logging.basicConfig(level=logging.INFO, format="%(message)s") + + +def run_standard_route_mock() -> None: + """ + Scenario: Core Network Mocking (dry_run). + If you query an endpoint with dry_run=True, the SDK safely + returns a mock JSON response without making an HTTP request. + """ + print("\n--- 🧪 Dry Run Execution ---") + with Client(auth=("api", "fake-key"), dry_run=True) as client: + response = client.domains.get() + print("\nSystem response (Intercepted):") + print(response.json()) + + +if __name__ == "__main__": + run_standard_route_mock() diff --git a/mailgun/examples/email_validation_examples.py b/mailgun/examples/email_validation_examples.py index d2f1464..53ee477 100644 --- a/mailgun/examples/email_validation_examples.py +++ b/mailgun/examples/email_validation_examples.py @@ -213,11 +213,11 @@ def post_preview_sync(api_key: str, csv_filepath: Path) -> None: get_bulk_validate_sync(api_key=API_KEY) post_bulk_list_validate_sync(api_key=API_KEY, csv_filepath=VALIDATION_CSV) get_bulk_list_validate_sync(api_key=API_KEY) - # delete_bulk_list_validate_sync(api_key=API_KEY, domain=DOMAIN) + # delete_bulk_list_validate_sync(api_key=API_KEY) get_preview_sync(api_key=API_KEY) post_preview_sync(api_key=API_KEY, csv_filepath=PREVIEW_CSV) - # delete_preview_sync(api_key=API_KEY, domain=DOMAIN) + # delete_preview_sync(api_key=API_KEY) print("\n--- Running Asynchronous Examples ---") asyncio.run(post_single_validate_async(api_key=API_KEY)) diff --git a/mailgun/examples/ext_examples.py b/mailgun/examples/ext_examples.py new file mode 100644 index 0000000..c77b091 --- /dev/null +++ b/mailgun/examples/ext_examples.py @@ -0,0 +1,94 @@ +from fastapi import FastAPI, HTTPException, Depends +from mailgun.client import AsyncClient +from mailgun.handlers.error_handler import ApiError +from mailgun.ext.pydantic.models import SendMessageSchema +from pydantic import ValidationError + +app = FastAPI() + + +# 1. Dependency Injection for the Client Lifecycle +async def get_mailgun_client(): + # 2. Enable dry_run=True so it mocks the network locally! + async with AsyncClient(auth=("api", "my-key"), dry_run=True) as client: + yield client + + +# JSON example to test in http://127.0.0.1:8000/docs#/default/send_email_send_email_post +# { +# "to": ["user@example.com"], +# "from": "admin@company.com", +# "subject": "Weekly Report", +# "text": "Here is your report.", +# "custom_params": { +# "v:invoice_id": "99824", +# "h:X-Priority": "High", +# "o:tracking": "yes" +# } +# } +@app.post("/send-email") +async def send_email( + payload: SendMessageSchema, mailgun_client: AsyncClient = Depends(get_mailgun_client) +): + # Use a serializer to flatten custom_params and exclude None values + clean_data = payload.to_mailgun_payload() + + try: + response = await mailgun_client.messages.create(domain="my-domain.com", data=clean_data) + return response.json() + + except ApiError as e: + # 3. Gracefully handle actual Mailgun network/auth errors + raise HTTPException(status_code=400, detail=str(e)) + + +def test_validation(): + print("--- 1. Testing Valid Payload ---") + try: + valid_payload = SendMessageSchema( + from_="admin@company.com", + to=["user@example.com"], + subject="Weekly Report", + text="Here is your report.", + ) + print("✅ Valid payload passed validation!") + print(f"Data: {valid_payload.to_mailgun_payload()}") + except ValidationError as e: + print(f"❌ Valid payload failed: {e}") + + print("\n--- 2. Testing Invalid Email ---") + try: + SendMessageSchema( + from_="admin@company.com", + to=["bad-email-format"], # Missing @ + subject="Test", + text="Content", + ) + except ValidationError as e: + print(f"✅ Caught expected error (Invalid email):") + print(e.json()) + + print("\n--- 3. Testing Missing Content ---") + try: + SendMessageSchema(from_="admin@company.com", to=["user@example.com"], subject="Empty body") + except ValidationError as e: + print(f"✅ Caught expected error (Missing content):") + print(e.json()) + + print("\n--- 4. Testing Custom Variables (v: and h:) ---") + try: + # Use custom_params instead of **kwargs to ensure security and validation + var_payload = SendMessageSchema( + from_="admin@company.com", + to=["user@example.com"], + text="Variables test", + custom_params={"v:my_var": "123", "h:X-Custom-Header": "Value"}, + ) + print("✅ Custom variables/headers accepted!") + print(f"Flattened payload: {var_payload.to_mailgun_payload()}") + except ValidationError as e: + print(f"❌ Custom variables failed: {e}") + + +if __name__ == "__main__": + test_validation() diff --git a/mailgun/examples/ip_pools_examples.py b/mailgun/examples/ip_pools_examples.py index c60afbd..83dc8ab 100644 --- a/mailgun/examples/ip_pools_examples.py +++ b/mailgun/examples/ip_pools_examples.py @@ -14,17 +14,17 @@ # ============================================================================== -def get_ippools_sync(api_key: str, domain: str) -> None: +def get_ippools_sync(api_key: str) -> None: """ GET /v1/ip_pools :return: None """ with Client(auth=("api", api_key)) as client: - response = client.ippools.get(domain=domain) + response = client.ippools.get() print("GET IP Pools (Sync):", response.json()) -def create_ippool_sync(api_key: str, domain: str) -> None: +def create_ippool_sync(api_key: str) -> None: """ POST /v1/ip_pools :return: None @@ -35,11 +35,11 @@ def create_ippool_sync(api_key: str, domain: str) -> None: "ips": ["1.2.3.4"], } with Client(auth=("api", api_key)) as client: - response = client.ippools.create(domain=domain, data=post_data) + response = client.ippools.create(data=post_data) print("POST Create IP Pool (Sync):", response.json()) -def update_ippool_sync(api_key: str, domain: str, pool_id: str) -> None: +def update_ippool_sync(api_key: str, pool_id: str) -> None: """ PATCH /v1/ip_pools/{pool_id} :return: None @@ -49,17 +49,17 @@ def update_ippool_sync(api_key: str, domain: str, pool_id: str) -> None: "description": "Test3", } with Client(auth=("api", api_key)) as client: - response = client.ippools.patch(domain=domain, data=data, pool_id=pool_id) + response = client.ippools.patch(data=data, pool_id=pool_id) print("PATCH Update IP Pool (Sync):", response.json()) -def delete_ippool_sync(api_key: str, domain: str, pool_id: str) -> None: +def delete_ippool_sync(api_key: str, pool_id: str) -> None: """ DELETE /v1/ip_pools/{pool_id} :return: None """ with Client(auth=("api", api_key)) as client: - response = client.ippools.delete(domain=domain, pool_id=pool_id) + response = client.ippools.delete(pool_id=pool_id) print("DELETE IP Pool (Sync):", response.json()) diff --git a/mailgun/examples/ips_examples.py b/mailgun/examples/ips_examples.py index 06f4ccf..fa48db7 100644 --- a/mailgun/examples/ips_examples.py +++ b/mailgun/examples/ips_examples.py @@ -13,24 +13,24 @@ # ============================================================================== -def get_ips_sync(api_key: str, domain: str) -> None: +def get_ips_sync(api_key: str) -> None: """ GET /ips :return: None """ filters: dict[str, str] = {"dedicated": "true"} with Client(auth=("api", api_key)) as client: - response = client.ips.get(domain=domain, filters=filters) + response = client.ips.get(filters=filters) print("GET IPs (Sync):", response.json()) -def get_single_ip_sync(api_key: str, domain: str, target_ip: str) -> None: +def get_single_ip_sync(api_key: str, target_ip: str) -> None: """ GET /ips/ :return: None """ with Client(auth=("api", api_key)) as client: - response = client.ips.get(domain=domain, ip=target_ip) + response = client.ips.get(ip=target_ip) print("GET Single IP (Sync):", response.json()) @@ -75,24 +75,24 @@ def delete_domain_ip_sync(api_key: str, domain: str, target_ip: str) -> None: # ============================================================================== -async def get_ips_async(api_key: str, domain: str) -> None: +async def get_ips_async(api_key: str) -> None: """ GET /ips (Asynchronous) :return: None """ filters: dict[str, str] = {"dedicated": "true"} async with AsyncClient(auth=("api", api_key)) as client: - response = await client.ips.get(domain=domain, filters=filters) + response = await client.ips.get(filters=filters) print("GET IPs (Async):", response.json()) -async def get_single_ip_async(api_key: str, domain: str, target_ip: str) -> None: +async def get_single_ip_async(api_key: str, target_ip: str) -> None: """ GET /ips/ (Asynchronous) :return: None """ async with AsyncClient(auth=("api", api_key)) as client: - response = await client.ips.get(domain=domain, ip=target_ip) + response = await client.ips.get(ip=target_ip) print("GET Single IP (Async):", response.json()) @@ -148,15 +148,15 @@ async def delete_domain_ip_async(api_key: str, domain: str, target_ip: str) -> N print("Please set the 'APIKEY' and 'DOMAIN' environment variables to run examples.") else: print("--- Running Synchronous Examples ---") - get_ips_sync(api_key=API_KEY, domain=DOMAIN) - # get_single_ip_sync(api_key=API_KEY, domain=DOMAIN, target_ip=TARGET_IP) + get_ips_sync(api_key=API_KEY) + # get_single_ip_sync(api_key=API_KEY, target_ip=TARGET_IP) # get_domain_ips_sync(api_key=API_KEY, domain=DOMAIN) # post_domains_ip_sync(api_key=API_KEY, domain=DOMAIN, target_ip=TARGET_IP) # delete_domain_ip_sync(api_key=API_KEY, domain=DOMAIN, target_ip=TARGET_IP) print("\n--- Running Asynchronous Examples ---") - asyncio.run(get_ips_async(api_key=API_KEY, domain=DOMAIN)) - # asyncio.run(get_single_ip_async(api_key=API_KEY, domain=DOMAIN, target_ip=TARGET_IP)) + asyncio.run(get_ips_async(api_key=API_KEY)) + # asyncio.run(get_single_ip_async(api_key=API_KEY, target_ip=TARGET_IP)) # asyncio.run(get_domain_ips_async(api_key=API_KEY, domain=DOMAIN)) # asyncio.run(post_domains_ip_async(api_key=API_KEY, domain=DOMAIN, target_ip=TARGET_IP)) # asyncio.run(delete_domain_ip_async(api_key=API_KEY, domain=DOMAIN, target_ip=TARGET_IP)) diff --git a/mailgun/examples/logs_examples.py b/mailgun/examples/logs_examples.py index 0a4a586..9998c8c 100644 --- a/mailgun/examples/logs_examples.py +++ b/mailgun/examples/logs_examples.py @@ -3,6 +3,7 @@ from __future__ import annotations import asyncio +import json import os from typing import Any @@ -40,7 +41,8 @@ def post_analytics_logs_sync(api_key: str, domain: str) -> None: } with Client(auth=("api", api_key)) as client: - response = client.analytics_logs.create(data=data) + headers = {"Content-Type": "application/json"} + response = client.analytics_logs.create(data=json.dumps(data), headers=headers) print("POST Analytics Logs (Sync):", response.json()) @@ -75,7 +77,8 @@ async def post_analytics_logs_async(api_key: str, domain: str) -> None: } async with AsyncClient(auth=("api", api_key)) as client: - response = await client.analytics_logs.create(data=data) + headers = {"Content-Type": "application/json"} + response = await client.analytics_logs.create(data=json.dumps(data), headers=headers) print("POST Analytics Logs (Async):", response.json()) diff --git a/mailgun/examples/mailing_lists_examples.py b/mailgun/examples/mailing_lists_examples.py index d542579..6e783a7 100644 --- a/mailgun/examples/mailing_lists_examples.py +++ b/mailgun/examples/mailing_lists_examples.py @@ -19,7 +19,7 @@ def delete_list_sync(api_key: str, domain: str, list_address: str) -> None: :return: None """ with Client(auth=("api", api_key)) as client: - response = client.lists.delete(domain=domain, address=list_address) + response = client.lists.delete(address=list_address) print("DELETE List (Sync):", response.json()) @@ -29,7 +29,7 @@ def get_list_pages_sync(api_key: str, domain: str) -> None: :return: None """ with Client(auth=("api", api_key)) as client: - response = client.lists_pages.get(domain=domain) + response = client.lists_pages.get() print("GET List Pages (Sync):", response.json()) diff --git a/mailgun/examples/metrics_examples.py b/mailgun/examples/metrics_examples.py index 0e51e8f..69a9ab6 100644 --- a/mailgun/examples/metrics_examples.py +++ b/mailgun/examples/metrics_examples.py @@ -3,6 +3,7 @@ from __future__ import annotations import asyncio +import json import os from typing import Any @@ -39,9 +40,9 @@ def post_analytics_metrics_sync(api_key: str, domain: str) -> None: "include_subaccounts": True, "include_aggregates": True, } - with Client(auth=("api", api_key)) as client: - response = client.analytics_metrics.create(data=data) + headers = {"Content-Type": "application/json"} + response = client.analytics_metrics.create(data=json.dumps(data), headers=headers) print("POST Analytics Metrics (Sync):", response.json()) @@ -120,7 +121,8 @@ async def post_analytics_metrics_async(api_key: str, domain: str) -> None: } async with AsyncClient(auth=("api", api_key)) as client: - response = await client.analytics_metrics.create(data=data) + headers = {"Content-Type": "application/json"} + response = await client.analytics_metrics.create(data=json.dumps(data), headers=headers) print("POST Analytics Metrics (Async):", response.json()) diff --git a/mailgun/examples/routes_examples.py b/mailgun/examples/routes_examples.py index 42b9c82..754eb9f 100644 --- a/mailgun/examples/routes_examples.py +++ b/mailgun/examples/routes_examples.py @@ -13,45 +13,45 @@ # ============================================================================== -def delete_route_sync(api_key: str, domain: str, route_id: str) -> None: +def delete_route_sync(api_key: str, route_id: str) -> None: """ DELETE /routes/ :return: None """ with Client(auth=("api", api_key)) as client: - response = client.routes.delete(domain=domain, route_id=route_id) + response = client.routes.delete(route_id=route_id) print("DELETE Route (Sync):", response.json()) -def get_route_by_id_sync(api_key: str, domain: str, route_id: str) -> None: +def get_route_by_id_sync(api_key: str, route_id: str) -> None: """ GET /routes/ :return: None """ with Client(auth=("api", api_key)) as client: - response = client.routes.get(domain=domain, route_id=route_id) + response = client.routes.get(route_id=route_id) print("GET Route By ID (Sync):", response.json()) -def get_routes_match_sync(api_key: str, domain: str, sender: str) -> None: +def get_routes_match_sync(api_key: str, sender: str) -> None: """ GET /routes/match :return: None """ filters: dict[str, str] = {"address": sender} with Client(auth=("api", api_key)) as client: - response = client.routes_match.get(domain=domain, filters=filters) + response = client.routes_match.get(filters=filters) print("GET Routes Match (Sync):", response.json()) -def get_routes_sync(api_key: str, domain: str) -> None: +def get_routes_sync(api_key: str) -> None: """ GET /routes :return: None """ filters: dict[str, int] = {"skip": 0, "limit": 1} with Client(auth=("api", api_key)) as client: - response = client.routes.get(domain=domain, filters=filters) + response = client.routes.get(filters=filters) print("GET Routes (Sync):", response.json()) @@ -67,7 +67,7 @@ def post_routes_sync(api_key: str, domain: str) -> None: "action": ["forward('http://myhost.com/messages/')", "stop()"], } with Client(auth=("api", api_key)) as client: - response = client.routes.create(domain=domain, data=data) + response = client.routes.create(data=data) print("POST Routes (Sync):", response.json()) @@ -83,7 +83,7 @@ def put_route_sync(api_key: str, domain: str, route_id: str) -> None: "action": ["forward('http://myhost.com/messages/')", "stop()"], } with Client(auth=("api", api_key)) as client: - response = client.routes.put(domain=domain, data=data, route_id=route_id) + response = client.routes.put(data=data, route_id=route_id) print("PUT Route (Sync):", response.json()) @@ -92,45 +92,45 @@ def put_route_sync(api_key: str, domain: str, route_id: str) -> None: # ============================================================================== -async def delete_route_async(api_key: str, domain: str, route_id: str) -> None: +async def delete_route_async(api_key: str, route_id: str) -> None: """ DELETE /routes/ (Asynchronous) :return: None """ async with AsyncClient(auth=("api", api_key)) as client: - response = await client.routes.delete(domain=domain, route_id=route_id) + response = await client.routes.delete(route_id=route_id) print("DELETE Route (Async):", response.json()) -async def get_route_by_id_async(api_key: str, domain: str, route_id: str) -> None: +async def get_route_by_id_async(api_key: str, route_id: str) -> None: """ GET /routes/ (Asynchronous) :return: None """ async with AsyncClient(auth=("api", api_key)) as client: - response = await client.routes.get(domain=domain, route_id=route_id) + response = await client.routes.get(route_id=route_id) print("GET Route By ID (Async):", response.json()) -async def get_routes_async(api_key: str, domain: str) -> None: +async def get_routes_async(api_key: str) -> None: """ GET /routes (Asynchronous) :return: None """ filters: dict[str, int] = {"skip": 0, "limit": 1} async with AsyncClient(auth=("api", api_key)) as client: - response = await client.routes.get(domain=domain, filters=filters) + response = await client.routes.get(filters=filters) print("GET Routes (Async):", response.json()) -async def get_routes_match_async(api_key: str, domain: str, sender: str) -> None: +async def get_routes_match_async(api_key: str, sender: str) -> None: """ GET /routes/match (Asynchronous) :return: None """ filters: dict[str, str] = {"address": sender} async with AsyncClient(auth=("api", api_key)) as client: - response = await client.routes_match.get(domain=domain, filters=filters) + response = await client.routes_match.get(filters=filters) print("GET Routes Match (Async):", response.json()) @@ -146,7 +146,7 @@ async def post_routes_async(api_key: str, domain: str) -> None: "action": ["forward('http://myhost.com/messages/')", "stop()"], } async with AsyncClient(auth=("api", api_key)) as client: - response = await client.routes.create(domain=domain, data=data) + response = await client.routes.create(data=data) print("POST Routes (Async):", response.json()) @@ -162,7 +162,7 @@ async def put_route_async(api_key: str, domain: str, route_id: str) -> None: "action": ["forward('http://myhost.com/messages/')", "stop()"], } async with AsyncClient(auth=("api", api_key)) as client: - response = await client.routes.put(domain=domain, data=data, route_id=route_id) + response = await client.routes.put(data=data, route_id=route_id) print("PUT Route (Async):", response.json()) @@ -183,6 +183,6 @@ async def put_route_async(api_key: str, domain: str, route_id: str) -> None: print("Please set the 'APIKEY' and 'DOMAIN' environment variables to run examples.") else: print("--- Running Synchronous Examples ---") - get_routes_match_sync(api_key=API_KEY, domain=DOMAIN, sender=SENDER) - # get_routes_sync(api_key=API_KEY, domain=DOMAIN) - # get_route_by_id_sync(api_key + get_routes_match_sync(api_key=API_KEY, sender=SENDER) + # get_routes_sync(api_key=API_KEY) + # get_route_by_id_sync(api_key=API_KEY) diff --git a/mailgun/examples/tags_new_examples.py b/mailgun/examples/tags_new_examples.py index 478a852..ab5954b 100644 --- a/mailgun/examples/tags_new_examples.py +++ b/mailgun/examples/tags_new_examples.py @@ -22,7 +22,7 @@ def delete_analytics_tags_sync(api_key: str, tag_name: str) -> None: """ data: dict[str, str] = {"tag": tag_name} with Client(auth=("api", api_key)) as client: - response = client.analytics_tags.delete(data=data) + response = client.analytics_tags.delete(filters=data) print("DELETE Analytics Tags (Sync):", response.json()) diff --git a/mailgun/examples/webhooks_examples.py b/mailgun/examples/webhooks_examples.py index 45970e5..993224c 100644 --- a/mailgun/examples/webhooks_examples.py +++ b/mailgun/examples/webhooks_examples.py @@ -60,9 +60,7 @@ def put_webhook_sync(api_key: str, domain: str) -> None: PUT /v3/domains//webhooks/ :return: None """ - data: dict[str, Any] = { - "url": ["https://facebook.com", "https://google.com"], - } + data = [("url", "https://facebook.com"), ("url", "https://google.com")] with Client(auth=("api", api_key)) as client: response = client.domains_webhooks.put(domain=domain, webhook_name="clicked", data=data) print("PUT Webhook (Sync):", response.json()) @@ -119,9 +117,7 @@ async def put_webhook_async(api_key: str, domain: str) -> None: PUT /v3/domains//webhooks/ (Asynchronous) :return: None """ - data: dict[str, Any] = { - "url": ["https://facebook.com", "https://google.com"], - } + data = [("url", "https://facebook.com"), ("url", "https://google.com")] async with AsyncClient(auth=("api", api_key)) as client: response = await client.domains_webhooks.put( domain=domain, webhook_name="clicked", data=data diff --git a/mailgun/ext/__init__.py b/mailgun/ext/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/mailgun/ext/pydantic/__init__.py b/mailgun/ext/pydantic/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/mailgun/ext/pydantic/models.py b/mailgun/ext/pydantic/models.py new file mode 100644 index 0000000..137727c --- /dev/null +++ b/mailgun/ext/pydantic/models.py @@ -0,0 +1,153 @@ +# mypy: disable-error-code="untyped-decorator" + +import re +from typing import Any + +from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator + + +# Lightweight regex for email validation without depending on `pydantic[email]` +_EMAIL_REGEX = re.compile(r"^[^@]+@[^@]+\.[^@]+$") +# CWE-113: Strict detection of Carriage Return and Line Feed characters +_CRLF_REGEX = re.compile(r"[\r\n]") + + +def _validate_emails(value: str | list[str]) -> str | list[str]: + """Internal validator for email formats. + + Args: + value: The email or list of emails to validate. + + Returns: + The validated email or list of emails. + + Raises: + ValueError: If an email format is invalid or contains injection vectors. + """ + if not value: + raise ValueError("Email fields cannot be empty.") + + emails = [value] if isinstance(value, str) else value + for email in emails: + # 1. Poka-yoke: Prevent HTTP Header Injection (CWE-113) + if _CRLF_REGEX.search(email): + msg = f"Security Alert (CWE-113): CRLF injection detected in email: '{email}'" + raise ValueError(msg) + + # Quick format check. Ignore names (e.g., "John Doe ") + raw_email = email.split("<")[-1].replace(">", "").strip() + if not _EMAIL_REGEX.match(raw_email): + msg = f"Invalid email format detected: '{email}'" + raise ValueError(msg) + return value + + +class SendMessageSchema(BaseModel): + """Pydantic v2 Strict Schema for the Mailgun V3 Send Message endpoint. + + Provides compile-time safety, runtime validation, and auto-completion. + """ + + model_config = ConfigDict( + populate_by_name=True, + # 'allow' is risky. We switch to 'forbid' for top-level fields + # and handle dynamic keys explicitly in the model validator. + extra="forbid", + str_strip_whitespace=True, + strict=True, # Prevents type coercion (e.g., bool -> int) + ) + + # Required fields + to: str | list[str] = Field(..., description="Email address(es) of the recipient(s)") + from_: str = Field(..., alias="from", description="Email address of the sender") + + # Optional recipients + cc: str | list[str] | None = Field(default=None) + bcc: str | list[str] | None = Field(default=None) + + # Subject and content (CWE-400: Strict memory bounding set to 25MB max) + subject: str | None = Field(default=None, max_length=998) # RFC 2822 limit + text: str | None = Field(default=None, max_length=25_000_000) + html: str | None = Field(default=None, max_length=25_000_000) + amp_html: str | None = Field(default=None, max_length=25_000_000) + template: str | None = Field(default=None, max_length=255) + + # The strict container for dynamic parameters + # This prevents Mass Assignment while supporting Mailgun's dynamic schema + custom_params: dict[str, str] = Field(default_factory=dict) + + @field_validator("custom_params") + @classmethod + def validate_prefixes(cls, v: dict[str, str]) -> dict[str, str]: + """Validates that custom parameter keys start with allowed Mailgun prefixes and contain no CRLFs. + + Args: + v: The dictionary of custom parameters to validate. + + Returns: + The validated dictionary of custom parameters. + + Raises: + ValueError: If a key does not start with 'v:', 'h:', 'o:', or contains CRLFs. + """ + for key, val in v.items(): + if not key.startswith(("v:", "h:", "o:")): + msg = ( + f"Unknown custom parameter '{key}'. " + "Mailgun specific options must start with 'v:', 'h:', or 'o:'" + ) + raise ValueError(msg) + + # CWE-113: Block CRLF injection in custom headers and variables + if _CRLF_REGEX.search(key) or _CRLF_REGEX.search(str(val)): + msg_0 = f"Security Alert (CWE-113): CRLF injection detected in custom parameter: '{key}'" + raise ValueError(msg_0) + + return v + + @field_validator("to", "from_", "cc", "bcc", mode="after") + @classmethod + def check_email_formats(cls, v: Any) -> Any: + """Validates the correct format of email addresses. + + Returns: + The validated input value. + """ + if v is not None: + _validate_emails(v) + return v + + @model_validator(mode="after") + def validate_body(self) -> "SendMessageSchema": + """Cross-validation of body content. + + Returns: + The validated schema instance. + + Raises: + ValueError: If no body parts are provided or invalid prefixes are used. + """ + # Ensure the presence of the email body + if not any([self.text, self.html, self.template, self.amp_html]): + raise ValueError( + "A Mailgun message must contain at least one body part: " + "'text', 'html', 'amp_html', or 'template'." + ) + + return self + + def to_mailgun_payload(self) -> dict[str, Any]: + """SERIALIZER: Flattens custom_params into the top-level payload. + + This is the method the SDK should call before sending. + + Returns: + Standard fields as a dict + """ + # Get standard fields as a dict + data: dict[str, Any] = self.model_dump( + by_alias=True, exclude_none=True, exclude={"custom_params"} + ) + # Flatten custom_params into the root + data.update(self.custom_params) + return data diff --git a/mailgun/filters.py b/mailgun/filters.py index 09ef753..acdf75a 100644 --- a/mailgun/filters.py +++ b/mailgun/filters.py @@ -1,5 +1,6 @@ import logging import re +from typing import Any, Final class RedactingFilter(logging.Filter): @@ -9,6 +10,69 @@ class RedactingFilter(logging.Filter): """ SECRET_PATTERN = re.compile(r"(key-|pubkey-)[\w\-]+") + MAX_REDACTION_DEPTH: Final[int] = 4 + + # Standard LogRecord attributes to ignore for maximum performance + _STANDARD_ATTRS = frozenset( + { + "args", + "asctime", + "created", + "exc_info", + "exc_text", + "filename", + "funcName", + "levelname", + "levelno", + "lineno", + "message", + "module", + "msecs", + "msg", + "name", + "pathname", + "process", + "processName", + "relativeCreated", + "stack_info", + "thread", + "threadName", + "taskName", + } + ) + + def _deep_redact(self, data: Any, depth: int = 0) -> Any: # noqa: PLR0911 + """Recursively sanitize strings, dictionaries, and iterables. + + Returns: + A safely sanitized copy of the input data with secrets redacted. + """ + # Prevent stack overflow and CPU spikes on complex/circular objects + if depth > self.MAX_REDACTION_DEPTH: + return "" + + if isinstance(data, dict): + return {k: self._deep_redact(v, depth + 1) for k, v in data.items()} + if isinstance(data, list): + return [self._deep_redact(item, depth + 1) for item in data] + if isinstance(data, tuple): + if hasattr(data, "_fields"): + return type(data)(*(self._deep_redact(item, depth + 1) for item in data)) + return tuple(self._deep_redact(item, depth + 1) for item in data) + if isinstance(data, str): + return self.SECRET_PATTERN.sub(r"\1[REDACTED]", data) + if isinstance(data, (int, float, bool, type(None))): + return data + + # CWE-316: Prevent "Late Stringification" bypass on custom objects + if hasattr(data, "model_dump") and callable(data.model_dump): + return self._deep_redact(data.model_dump(), depth + 1) + if hasattr(data, "__dict__"): + return self._deep_redact(vars(data), depth + 1) + + # Catch-all for Pydantic, Dataclasses, and custom objects + # Force stringification to prevent "Late Stringification" bypass + return self.SECRET_PATTERN.sub(r"\1[REDACTED]", str(data)) def filter(self, record: logging.LogRecord) -> bool: """Filter out sensitive secrets from log records. @@ -16,19 +80,17 @@ def filter(self, record: logging.LogRecord) -> bool: Returns: True to allow the record to be logged. """ - # Redact simple string messages + # 1. Redact primary message if isinstance(record.msg, str): record.msg = self.SECRET_PATTERN.sub(r"\1[REDACTED]", record.msg) - # Redact formatting arguments if present - if isinstance(record.args, dict): - record.args = { - k: self.SECRET_PATTERN.sub(r"\1[REDACTED]", str(v)) if isinstance(v, str) else v - for k, v in record.args.items() - } - elif isinstance(record.args, tuple): - record.args = tuple( - self.SECRET_PATTERN.sub(r"\1[REDACTED]", str(v)) if isinstance(v, str) else v - for v in record.args - ) + # 2. Redact tuple/dict args WITHOUT changing their types + if isinstance(record.args, (dict, tuple)): + record.args = self._deep_redact(record.args) + + # 3. Redact dynamically injected 'extra' attributes + for attr_name, attr_value in record.__dict__.items(): + if attr_name not in self._STANDARD_ATTRS: + record.__dict__[attr_name] = self._deep_redact(attr_value) + return True diff --git a/mailgun/handlers/domains_handler.py b/mailgun/handlers/domains_handler.py index 0946224..7540970 100644 --- a/mailgun/handlers/domains_handler.py +++ b/mailgun/handlers/domains_handler.py @@ -120,9 +120,14 @@ def handle_sending_queues( """ keys = url.get("keys", []) if "sending_queues" in keys or "sendingqueues" in keys: - base_clean = str(url["base"]).replace("domains/", "").replace("domains", "").rstrip("/") + # Safely strip the trailing suffix without mangling custom proxy hosts + base_clean = str(url["base"]).rstrip("/") + if base_clean.endswith("/domains"): + base_clean = base_clean.removesuffix("/domains") + safe_domain = SecurityGuard.sanitize_path_segment(domain) if domain else "" return f"{base_clean}/{safe_domain}/sending_queues" + return str(url["base"]) @@ -199,6 +204,8 @@ def handle_webhooks( # noqa: PLR0914 url: dict[str, Any], domain: str | None, method: str | None, + data: dict[str, Any] | None = None, + filters: dict[str, Any] | None = None, **kwargs: Any, ) -> str: """Dynamically route webhooks to v1, v3, or v4 based on domain and payload. @@ -232,12 +239,12 @@ def handle_webhooks( # noqa: PLR0914 webhook_name = webhook_name or keys[1] keys = [keys[0]] - data = kwargs.get("data") or {} - filters = kwargs.get("filters") or {} + data_dict = data or {} + filters_dict = filters or {} # Payload Detection (Content-Based Routing) - has_event_types = isinstance(data, dict) and "event_types" in data - has_url_query = isinstance(filters, dict) and "url" in filters + has_event_types = isinstance(data_dict, dict) and "event_types" in data_dict + has_url_query = isinstance(filters_dict, dict) and "url" in filters_dict method_lower = (method or "").lower() is_v4 = False diff --git a/mailgun/handlers/error_handler.py b/mailgun/handlers/error_handler.py index d1e2c5f..cdcc288 100644 --- a/mailgun/handlers/error_handler.py +++ b/mailgun/handlers/error_handler.py @@ -37,3 +37,20 @@ class RouteNotFoundError(ApiError): class UploadError(ApiError): """Raised when the maximum message size is greater than 25 MB.""" + + +class DeliverabilityError(ApiError): + """Raised when SpamGuard detects critical structural flaws in the HTML payload that would severely penalize domain reputation or trigger spam filters.""" + + def __init__(self, score: float, issues: list[str]) -> None: + self.score = score + self.issues = issues + + # Format a highly readable, bulleted message for the console + formatted_issues = "\n - ".join(issues) + message = ( + f"HTML Deliverability Check Failed (Score: {score}/100).\n" + f"The payload was blocked to protect your domain reputation. " + f"Please fix the following issues:\n - {formatted_issues}" + ) + super().__init__(message) diff --git a/mailgun/handlers/suppressions_handler.py b/mailgun/handlers/suppressions_handler.py index 99999ed..4c0cb2d 100644 --- a/mailgun/handlers/suppressions_handler.py +++ b/mailgun/handlers/suppressions_handler.py @@ -16,7 +16,7 @@ def handle_bounces( domain: str | None, _method: str | None, **kwargs: Any, -) -> Any: +) -> str: """Handle Bounces URL construction. Args: @@ -46,7 +46,7 @@ def handle_unsubscribes( domain: str | None, _method: str | None, **kwargs: Any, -) -> Any: +) -> str: """Handle Unsubscribes URL construction. Args: @@ -76,7 +76,7 @@ def handle_complaints( domain: str | None, _method: str | None, **kwargs: Any, -) -> Any: +) -> str: """Handle Complaints URL construction. Args: diff --git a/mailgun/handlers/tags_handler.py b/mailgun/handlers/tags_handler.py index 3a6b743..0be6cec 100644 --- a/mailgun/handlers/tags_handler.py +++ b/mailgun/handlers/tags_handler.py @@ -12,7 +12,7 @@ def handle_tags( - url: Any, + url: dict[str, Any], domain: str | None, _method: str | None, **kwargs: Any, diff --git a/mailgun/handlers/templates_handler.py b/mailgun/handlers/templates_handler.py index c3e94ac..d489a97 100644 --- a/mailgun/handlers/templates_handler.py +++ b/mailgun/handlers/templates_handler.py @@ -36,15 +36,21 @@ def handle_templates( base_url_str = str(url["base"]) if domain: - if "/v4/" in base_url_str: - base_url_str = base_url_str.replace("/v4/", "/v3/") + # Safely downgrade version targeting ONLY the suffix + if base_url_str.endswith("/v4/"): + base_url_str = base_url_str[:-4] + "/v3/" + elif base_url_str.endswith("/v4"): + base_url_str = base_url_str[:-3] + "/v3/" base_url_str = base_url_str if base_url_str.endswith("/") else f"{base_url_str}/" safe_domain = SecurityGuard.sanitize_path_segment(domain) domain_url = f"{base_url_str}{safe_domain}{final_keys}" else: - if "/v3/" in base_url_str: - base_url_str = base_url_str.replace("/v3/", "/v4/") + # Safely upgrade version targeting ONLY the suffix + if base_url_str.endswith("/v3/"): + base_url_str = base_url_str[:-4] + "/v4/" + elif base_url_str.endswith("/v3"): + base_url_str = base_url_str[:-3] + "/v4" base_url_str = base_url_str.rstrip("/") domain_url = f"{base_url_str}{final_keys}" diff --git a/mailgun/security.py b/mailgun/security.py index 78b2ced..f7a0a43 100644 --- a/mailgun/security.py +++ b/mailgun/security.py @@ -1,13 +1,15 @@ import hashlib import hmac +import json import math import re import ssl import sys +import time import unicodedata -import warnings +from html.parser import HTMLParser from pathlib import Path -from typing import Any, Final +from typing import Any, Final, TypedDict from urllib.parse import quote, unquote, urlparse from requests.adapters import HTTPAdapter @@ -40,14 +42,34 @@ class SecureHTTPAdapter(HTTPAdapter): Mitigates CWE-319. """ - def init_poolmanager(self, *args: Any, **kwargs: Any) -> None: - """Initialize the pool manager with a secure TLS context.""" + @staticmethod + def _get_secure_ssl_context() -> ssl.SSLContext: + """Create and return a hardened SSL context enforcing TLS 1.2+. + + Returns: + ssl.SSLContext: A hardened SSL context. + """ context = ssl.create_default_context() context.minimum_version = ssl.TLSVersion.TLSv1_2 - kwargs["ssl_context"] = context + return context + + def init_poolmanager(self, *args: Any, **kwargs: Any) -> None: + """Initialize the pool manager with a secure TLS context.""" + kwargs["ssl_context"] = self._get_secure_ssl_context() # HTTPAdapter lacks strict static types for this internal method. super().init_poolmanager(*args, **kwargs) + def proxy_manager_for(self, proxy: str, **proxy_kwargs: Any) -> Any: + """Ensure proxy connections also strictly enforce TLS 1.2+. + + Returns: + Any: The proxy manager instance. + """ + # Inject our hardened SSL context into the proxy kwargs + proxy_kwargs["ssl_context"] = self._get_secure_ssl_context() + # Pass it up to the parent class to actually construct the ProxyManager + return super().proxy_manager_for(proxy, **proxy_kwargs) + class SecretAuth(tuple): # type: ignore[type-arg] """OWASP: Obfuscate credentials in memory dumps and tracebacks.""" @@ -214,6 +236,7 @@ def sanitize_timeout(cls, timeout: TimeoutType) -> TimeoutType: Strict Creation-Time Timeout Constraints & Float Validation. Prevents thread pool exhaustion from infinite blocking (CWE-400). + Enforces a strict maximum boundary of 300 seconds. Args: timeout: The requested timeout value. @@ -222,18 +245,18 @@ def sanitize_timeout(cls, timeout: TimeoutType) -> TimeoutType: The safely verified timeout value. Raises: - ValueError: If the timeout is a negative number, zero, non-finite, - or a tuple with an incorrect number of elements. + ValueError: If the timeout is None, negative, zero, non-finite, + exceeds 300 seconds, or a tuple with an incorrect number of elements. """ if timeout is None: - # Soft Deprecation - warnings.warn( - "Passing 'timeout=None' allows infinite socket blocking (CWE-400). " - "This will be removed in a future major release. Please provide an explicit timeout.", - DeprecationWarning, - stacklevel=3, + msg = ( + "Security Alert (CWE-400): Infinite timeouts are forbidden. Provide a finite value." ) - return None + raise ValueError(msg) + + # Extract values from httpx.Timeout object cleanly + if hasattr(timeout, "read") and hasattr(timeout, "connect"): + timeout = (getattr(timeout, "connect", 60.0), getattr(timeout, "read", 60.0)) def _validate_float(val: Any) -> float: """Validate float value. @@ -246,7 +269,7 @@ def _validate_float(val: Any) -> float: Raises: TypeError: If the timeout is not a numeric type. - ValueError: If the timeout is NaN, Infinity, or less than or equal to zero. + ValueError: If the timeout is NaN, Infinity, less than/equal to zero, or exceeds 300. """ if isinstance(val, bool) or not isinstance(val, (int, float)): msg = f"Timeout must be a numeric value, got {type(val).__name__}" @@ -258,6 +281,11 @@ def _validate_float(val: Any) -> float: raise ValueError("Timeout must be a finite number.") if f_val <= 0: raise ValueError("Timeout must be a strictly positive finite number.") + if f_val > 300.0: # noqa: PLR2004 + raise ValueError( + "Security Alert: Timeout exceeds maximum allowed boundary of 300 seconds." + ) + return f_val if isinstance(timeout, tuple): @@ -266,7 +294,7 @@ def _validate_float(val: Any) -> float: raise ValueError( "Timeout must be a tuple containing exactly two elements: (connect, read)." ) - return (_validate_float(timeout[0]), _validate_float(timeout[1])) + return _validate_float(timeout[0]), _validate_float(timeout[1]) return _validate_float(timeout) @@ -424,24 +452,32 @@ def validate_attachment_path(file_path: str | Path, safe_base_dir: str | Path) - Raises: ValueError: If the resolved path escapes the safe base directory. - FileNotFoundError: If the file does not exist. """ - target = Path(file_path).resolve() - base = Path(safe_base_dir).resolve() + original_path = str(file_path) + target_path = Path(file_path).resolve() - if not target.is_relative_to(base): - sys.audit("mailgun.security.path_traversal_attempt", str(target)) - msg = ( - f"Security Alert (CWE-22): Path traversal blocked. " - f"File {target} is outside of safe directory {base}." - ) + if not target_path.exists() or not target_path.is_file(): + msg = f"Security Alert: Invalid attachment path or not a file: {file_path}" raise ValueError(msg) - if not target.exists() or not target.is_file(): - msg = f"Attachment not found or is not a file: {target}" - raise FileNotFoundError(msg) + if safe_base_dir: + base_path = Path(safe_base_dir).resolve() + if not target_path.is_relative_to(base_path): + raise ValueError("Security Alert (CWE-22): Path traversal attempt detected.") + else: + # Fallback zero-trust checks if no specific sandbox is provided + if ".." in original_path: + raise ValueError( + "Security Alert (CWE-22): Path traversal tokens ('..') are explicitly forbidden." + ) - return target + forbidden_roots = ("/etc", "/var", "/root", "/boot", "C:\\Windows", "C:\\System32") + if any(str(target_path).lower().startswith(root.lower()) for root in forbidden_roots): + raise ValueError( + "Security Alert: Access to sensitive OS system directories is explicitly forbidden." + ) + + return target_path @staticmethod def check_file_size(file_path: str | Path, max_size_mb: int = 25) -> None: @@ -454,7 +490,13 @@ def check_file_size(file_path: str | Path, max_size_mb: int = 25) -> None: ValueError: If the file exceeds the maximum allowed size. """ path = Path(file_path) - size_bytes = Path(path).stat().st_size + + # MUST assert it's a regular file to reject infinite /dev/zero or FIFOs + if not path.is_file(): + msg = f"Security Alert (CWE-400): Path is not a regular file: {path}" + raise ValueError(msg) + + size_bytes = path.stat().st_size max_bytes = max_size_mb * 1024 * 1024 if size_bytes > max_bytes: @@ -477,45 +519,233 @@ def sanitize_log_trace(value: Any) -> str: return _PATH_CONTROL_CHAR_RE.sub("_", safe_str) @staticmethod - def verify_webhook(signing_key: str, token: str, timestamp: str, signature: str) -> bool: + def verify_webhook( + signing_key: str | bytes, + token: str, + timestamp: str | int, + signature: str, + max_age_seconds: int = 300, + ) -> bool: """Cryptographically verify a Mailgun webhook signature. - Protects against CWE-347 (Improper Verification) and CWE-208 (Timing Attacks). + Protects against CWE-347 (Improper Verification), CWE-208 (Timing Attacks), + and CWE-294 (Capture-Replay Attacks). Args: signing_key: The Mailgun webhook signing key from the dashboard. token: The token provided in the webhook payload. timestamp: The timestamp provided in the webhook payload. signature: The signature provided in the webhook payload. + max_age_seconds: Maximum allowed age of the webhook in seconds. Returns: - True if the signature mathematically matches the payload, False otherwise. + True if the signature mathematically matches and is within TTL, False otherwise. Raises: - TypeError: If any of the signature components are not strictly strings. - ValueError: If the cryptographic payload is malformed or undecodable. + TypeError: If the signature components are invalid types. """ - # 1. Type Guard: Prevent AttributeError if a developer or attacker - # passes None, an int, or a list instead of a string. - if ( - not isinstance(signing_key, str) - or not isinstance(token, str) - or not isinstance(timestamp, str) - or not isinstance(signature, str) - ): - raise TypeError("Webhook signature components must be strictly strings.") + # 1. Type Guard: Prevent AttributeError and Type Confusion + if not isinstance(token, str) or not isinstance(signature, str): + raise TypeError("Security Alert: Webhook token and signature must be strings.") + + if not isinstance(signing_key, (str, bytes)): + raise TypeError("Security Alert: Signing key must be a string or bytes.") + # 2. Extract integer for math, but DO NOT mutate the raw timestamp string try: - # 2. Canonicalization: Encode strings to bytes safely - msg = f"{timestamp}{token}".encode() - key = signing_key.encode("utf-8") + ts_math = int(timestamp) + except (ValueError, TypeError) as e: + raise TypeError("Security Alert: Webhook timestamp must be a valid integer.") from e - # 3. Cryptographic Hashing - expected_mac = hmac.new(key, msg, hashlib.sha256).hexdigest() + # 3. TTL/Replay Attack Prevention (CWE-294) + if abs(time.time() - ts_math) > max_age_seconds: + logger.warning("Security Alert (CWE-294): Webhook timestamp expired.") + return False - # 4. Timing Attack Prevention: NEVER use '==' for crypto comparisons. - return hmac.compare_digest(expected_mac, signature) + # 4. Canonicalization: Encode securely + if isinstance(signing_key, str): + signing_key = signing_key.encode("utf-8") + + # Hash the exact raw string representation, not the integer cast. + raw_timestamp = str(timestamp) + msg = f"{raw_timestamp}{token}".encode() + + # 5. Cryptographic Hashing + expected_mac = hmac.new(key=signing_key, msg=msg, digestmod=hashlib.sha256).hexdigest() + + # 6. Timing Attack Prevention (CWE-208): NEVER use '==' for crypto comparisons. + return hmac.compare_digest(expected_mac, signature) + + @staticmethod + def normalize_domain(domain: str | None) -> str: + """Natively convert internationalized domain names (IDN) to Punycode (RFC 3490). + + This prevents UnicodeEncodeError when HTTP clients (like requests/httpx) + attempt to route to or build URLs with non-ASCII domains (e.g., Cyrillic). + + Args: + domain: The target domain name + + Returns: + The ASCII-safe Punycode string - except AttributeError as e: - # Fail-closed if underlying C-extensions reject malformed encodings - raise ValueError("Malformed cryptographic payload.") from e + Raises: + ValueError: If invalid domain name encoding. + """ + if not domain: + return "" + + try: + # Encode the Unicode string to IDNA bytes, then decode to an ASCII string. + # If the domain is already ASCII (e.g., 'example.com'), it remains unchanged. + return domain.encode("idna").decode("ascii") + except UnicodeError as e: + # Fallback or raise a clear validation error if the domain is completely malformed + msg = f"Invalid domain name encoding: {domain}" + raise ValueError(msg) from e + + +class SpamReport(TypedDict): + """Schema for the local deliverability check report.""" + + score: float + issues: list[str] + is_safe: bool + + +class _SpamGuardParser(HTMLParser): + """Internal lightning-fast HTML parser for detecting structural spam triggers.""" + + def __init__(self) -> None: + super().__init__() + self.issues: list[str] = [] + self.has_alt_tags = True + self.image_count = 0 + self.has_scripts = False + + def handle_starttag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None: + attr_dict = dict(attrs) + + if tag == "img": + self.image_count += 1 + if "alt" not in attr_dict or not attr_dict["alt"]: + self.has_alt_tags = False + + if tag == "script": + self.has_scripts = True + self.issues.append("CRITICAL: " + result = SpamGuard.check_html(bad_html) + + assert result["is_safe"] is False + assert result["score"] < 100.0 + + def test_analyze_html_flags_missing_alt_attributes(self) -> None: + """Deliverability check: Ensure missing alt tags trigger a warning penalty.""" + html_without_alt = "" + result = SpamGuard.check_html(html_without_alt) + + assert any("Missing 'alt' attributes" in issue for issue in result["issues"])