diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index 6b656997..f7a2c97a 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -164,8 +164,9 @@ repos: - id: slotscheck name: "🔍 check · slotscheck" additional_dependencies: - - requests>=2.32.5 + - requests>=2.33.0 - typing-extensions>=4.7.1 + - httpx2 >=2.7.0 - httpx>=0.24 - pytest>=9.0.3 - responses diff --git a/CHANGELOG.md b/CHANGELOG.md index 328fed92..9113bf16 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,7 +2,40 @@ We [keep a changelog.](http://keepachangelog.com/) -## [Unreleased] +## [Unreleased] (1.9.1) + +### Security + +- **SSRF and Scheme Whitelisting (CWE-918):** Enforced `ALLOWED_SCHEMES` validation (`https`, `http`) in `SecurityGuard.validate_mailgun_url()` and extended trusted hosts to include `.mailgun.com`. +- **Multi-Pass URL Encoding and Path Traversal (CWE-116 / CWE-22):** Hardened `SecurityGuard.sanitize_domain()` to iterate recursive `unquote()` checks up to 3 passes, apply NFKC Unicode normalization, and strip CRLF/slash sequences before evaluating `..` traversal sequences. +- **Replay Attack Window Verification (CWE-294):** Updated default webhook timestamp TTL in `SecurityGuard.verify_webhook()` to 900 seconds (15 minutes), rejecting expired requests while permitting `<= 0` to selectively bypass clock checks in test environments. +- **Pre-Flight Deliverability & XSS Detection (CWE-79 / CWE-400):** Extended `SpamGuard` with `_BLOCKED_TAGS` (`iframe`, `object`, `embed`, `applet`) and automated regex detection for inline event handlers (`on*`), and added pre-parsing length boundary checks. +- **Log Redaction Hardening (CWE-316 / CWE-117):** Increased `MAX_REDACTION_DEPTH` to 5 in `RedactingFilter` and added `frozenset` support. Preserved original `record.args` types (tuple, dict, list) to prevent string formatting crashes, wrapped object inspection in defensive guards, and added fallback handling for sets containing unhashable elements. +- **Payload Cycle & Recursion Protection:** Added `_deep_sanitize()` with a 50-level depth limit to `IdempotencyGuard` to prevent recursion overflow and handle cyclic data references. +- **Header Injection Boundaries (CWE-113 / RFC 9110):** Enforced ASCII control character checks across header keys and values in `SecurityGuard.sanitize_headers()` and guarded runtime telemetry calls with `sys in sys.modules`. + +### Fixed + +- **Stream Seek Pointer Displacement in Idempotency Checks:** Fixed `IdempotencyGuard.generate_key()` to record `stream.tell()` before reading file objects and restore the original pointer offset via `seek()` after computing hashes. +- **ChunkedStreamer Seeking & Pointer Handling:** Implemented `seek()` and `tell()` methods on `ChunkedStreamer`, strictly validated positive `chunk_size` values, and introduced explicit `_eof` state tracking to prevent duplicate reads. +- **Stream Pagination Query Parameter Type Drift:** Added `_cast_query_param()` in `BaseEndpoint` to preserve developer filter types (`int`, `float`, `bool`, `list`, `tuple`, `set`) during cursor pagination, and added guardrails for missing next cursors or non-dict payloads. +- **Safe Serialization in Builders:** Added `default=str` to `json.dumps()` across `MailgunMessageBuilder`, `MailgunTemplateBuilder`, and `BaseEndpoint` to safely serialize custom objects (e.g., UUID, datetime) without throwing `TypeError`. +- **Message Builder File Payload Mutation:** Returned a shallow copy of `self._files` in `MailgunMessageBuilder.build()` to prevent external consumers from mutating internal builder state. +- **Protected Attribute Routing Collisions:** Blocked routing fallbacks for `config` and `auth` in `Client.__getattr__()`, explicitly raising `AttributeError` instead of constructing invalid endpoints. +- **Domain Route Alias Canonicalization:** Normalized domain route aliases (`DOMAIN_ALIASES`) directly in `handle_domains()` and preserved literal `@` separators during credentials route generation. +- **Timeout Float Capacity Overflow:** Added `OverflowError` trapping during float conversion in `SecurityGuard.sanitize_timeout()`, mapping out-of-capacity numbers to standard `ValueError` exceptions. +- **Punycode Email Address Normalization:** Updated `SecurityGuard.normalize_domain()` to partition email addresses and apply IDNA Punycode encoding specifically to the host domain. + +### Changed + +- **Route Registry Definitions:** Registered `routes_match` under the v3 endpoint mapping, added `reputationanalytics_v2` under v2 prefix routes, and mapped `v1/spamtraps` deprecation warnings. +- **Dependency Specifications:** Synchronized `requests >=2.33.0` across environments, removed direct `conda-forge::` channel pinning for `httpx2` in `environment.yaml` and `environment-dev.yaml`, and aligned `.pre-commit-config.yaml` dependency constraints. +- **Linter Rule Alignment:** Removed the deprecated `missing-trailing-comma` selector from `pyproject.toml` to maintain compatibility with modern Ruff formatters. +- **Manage Script Permissions:** Updated `manage.sh` file mode to executable (`100755`). + +### Pull Requests Merged + +- PR #65: Hardening security and stability. ## v1.9.0 - 2026-08-04 diff --git a/PERFORMANCE.md b/PERFORMANCE.md index 3e3448c9..c97d58c8 100644 --- a/PERFORMANCE.md +++ b/PERFORMANCE.md @@ -29,6 +29,17 @@ String manipulation, dynamic imports (`importlib`), and sequential regex evaluat ______________________________________________________________________ +## Benchmarks (v1.9.0 vs. v1.9.1) + +| Metric | v1.9.0 (Baseline) | v1.9.1 (Current) | Delta / Notes | +| :-------------------------- | :---------------- | :--------------- | :------------------------------------- | +| **Cold Boot Time** | ~0.220 s | **~0.131 s** | **~40.5% Faster** (Optimized I/O read) | +| **Routing Speed (Mean)** | ~0.94 µs | **~0.95 µs** | **+9.5 ns** (Statistical parity) | +| **Async Throughput (Mean)** | ~3.06 ms | **~3.16 ms** | **+0.10 ms** (Type-cast validation) | +| **Sync Throughput (Mean)** | ~10.47 ms | **~10.15 ms** | **~3.1% Faster** (Tighter variance) | + +*Note: Tests were executed on CPython 3.13 (Apple M4 Pro, Darwin ARM64-bit) in an isolated environment.* + ## Benchmarks (v1.8.0 vs. v1.9.0) | Metric | v1.8.0 (Baseline) | v1.9.0 (Current) | Delta / Notes | diff --git a/README.md b/README.md index b4282759..0d855391 100644 --- a/README.md +++ b/README.md @@ -359,7 +359,8 @@ with Client(auth=("api", "key-super-secret-12345")) as client: By default, the SDK relies on the underlying HTTP client's standard timeouts. To prevent uncontrolled resource consumption (CWE-400) in high-throughput production environments, you can enforce strict global timeouts. -Timeouts can be passed as a single `float` (seconds for both connect and read) or a tuple (connect_timeout, read_timeout): +Timeouts can be passed as a single `float` (seconds for both connect and read) or a tuple (connect_timeout, read_timeout). +Timeouts are strictly validated against `float` overflow and capped at 300 seconds to prevent thread starvation. ```python import os @@ -421,7 +422,7 @@ If the issue persists, please reach out to our support team. 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`). +- **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`). Internal configuration and state properties (`client.config`, `client.auth`) are strictly isolated from dynamic endpoint dispatching. - **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. @@ -1806,6 +1807,8 @@ The SDK includes an active Interceptor engine that protects your application fro If you attempt to call a legacy or deprecated Mailgun endpoint (such as the old `v3` address validation or `v1` bounce classification), the SDK will **not** break your code. It will successfully execute the request but will emit a non-breaking Python `DeprecationWarning` and print a logger warning with instructions on which modern API endpoint to migrate to. +In `mailgun/routes.py`, `_DEPRECATED_ROUTES_PATTERNS` added explicit deprecation warnings for the legacy `/v1/spamtraps` family (`/v1/spamtraps`, `/v1/spamtraps/totals`, `/v1/spamtraps/filters`), directing users to the `v2` endpoint (`GET /v2/spamtraps`). + ## Type Hinting This SDK is fully type-hinted and complies with PEP 561 (`py.typed` included). Static type checkers (`mypy`, `pyright`) are enforced during CI checks. diff --git a/environment-dev.yaml b/environment-dev.yaml index 7739b473..ce5eb38c 100644 --- a/environment-dev.yaml +++ b/environment-dev.yaml @@ -10,8 +10,8 @@ dependencies: # PyPi only - python-build # runtime deps - - requests >=2.32.5 - - conda-forge::httpx2 >=2.7.0 + - requests >=2.33.0 + - httpx2 >=2.7.0 - httpx >=0.24.0 # extras - fastapi diff --git a/environment.yaml b/environment.yaml index 6a7b8266..5e49b0c1 100644 --- a/environment.yaml +++ b/environment.yaml @@ -7,8 +7,8 @@ dependencies: # build & host deps - pip # runtime deps - - requests >=2.32.5 - - conda-forge::httpx2 >=2.7.0 + - requests >=2.33.0 + - httpx2 >=2.7.0 - httpx >=0.24.0 # tests - pytest >=9.0.3 diff --git a/mailgun/builders.py b/mailgun/builders.py index f26bcb22..fdd6b394 100644 --- a/mailgun/builders.py +++ b/mailgun/builders.py @@ -31,7 +31,7 @@ class ChunkedStreamer: (like Serverless functions). """ - __slots__ = ("_file", "_file_path", "chunk_size") + __slots__ = ("_eof", "_file", "_file_path", "chunk_size") def __init__( self, @@ -40,7 +40,19 @@ def __init__( safe_base_dir: str | Path | None = None, chunk_size: int = CHUNK_SIZE, ) -> None: - """Init chunked streamer.""" + """Init chunked streamer. + + Args: + file_path: Path to the target attachment file. + safe_base_dir: Sandbox base directory for path validation. + chunk_size: Positive integer size in bytes per chunk. + + Raises: + ValueError: If chunk_size is not a strictly positive integer. + """ + if chunk_size <= 0: + raise ValueError("chunk_size must be a strictly positive integer.") + # 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) @@ -48,6 +60,7 @@ def __init__( self._file_path = str(safe_path) self.chunk_size = chunk_size self._file: IO[bytes] | None = None + self._eof = False def read(self, size: int) -> bytes: """File-like read method required by requests/httpx multipart encoders. @@ -58,6 +71,8 @@ def read(self, size: int) -> bytes: Returns: A byte string containing the read data. """ + if self._eof: + return b"" if self._file is None: self._file = Path(self._file_path).open("rb") # ruff: ignore[open-file-with-context-handler] @@ -66,10 +81,37 @@ def read(self, size: int) -> bytes: # 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._eof = True self.close() + return b"" return chunk + def seek(self, offset: int, whence: int = 0) -> int: + """Change the stream position to the given byte offset. + + Args: + offset: The position to seek to relative to whence. + whence: Reference point (0 for start, 1 for current, 2 for end). + + Returns: + The new absolute stream position in bytes. + """ + self._eof = False + if self._file is None: + self._file = Path(self._file_path).open("rb") # ruff: ignore[open-file-with-context-handler] + return self._file.seek(offset, whence) + + def tell(self) -> int: + """Return the current stream position in bytes. + + Returns: + The current file pointer position. + """ + if self._file is None: + return 0 + return self._file.tell() + def __iter__(self) -> Generator[bytes, None, None]: """Stream the file natively in chunks. @@ -149,7 +191,7 @@ 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, FileTuple]] = [] - self._idempotency_safe: bool = True # Enabled dy default + self._idempotency_safe: bool = True # Enabled by 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: @@ -231,7 +273,7 @@ def add_custom_variable(self, key: str, value: Any) -> Self: """ # Complex types must be serialized if isinstance(value, (dict, list)): - safe_val = json.dumps(value, separators=(",", ":")) + safe_val = json.dumps(value, separators=(",", ":"), default=str) else: safe_val = str(value) self._payload[f"v:{key}"] = safe_val @@ -323,7 +365,10 @@ def attach_stream( return self def attach_inline( - self, file_path: str | Path, cid: str | None = None, safe_base_dir: str | Path | None = None + 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. @@ -381,7 +426,7 @@ def set_template_variables(self, variables: dict[str, Any]) -> Self: Returns: The builder instance. """ - self._payload["t:variables"] = json.dumps(variables, separators=(",", ":")) + self._payload["t:variables"] = json.dumps(variables, separators=(",", ":"), default=str) return self def set_recipient_variables(self, variables: dict[str, dict[str, Any]]) -> Self: @@ -393,7 +438,11 @@ def set_recipient_variables(self, variables: dict[str, dict[str, Any]]) -> Self: Returns: The builder instance. """ - self._payload["recipient-variables"] = json.dumps(variables, separators=(",", ":")) + self._payload["recipient-variables"] = json.dumps( + variables, + separators=(",", ":"), + default=str, + ) return self def check_deliverability(self) -> dict[str, float | list[str] | bool] | SpamReport: @@ -430,7 +479,9 @@ def build(self) -> tuple[dict[str, Any], list[tuple[str, FileTuple]] | None]: if self._idempotency_safe and "h:X-Idempotency-Key" not in final_payload: idempotency_key = IdempotencyGuard.generate_key( - self._domain, final_payload, self._files + self._domain, + final_payload, + self._files, ) final_payload["h:X-Idempotency-Key"] = idempotency_key @@ -442,7 +493,8 @@ def build(self) -> tuple[dict[str, Any], list[tuple[str, FileTuple]] | None]: else: del final_payload[key] - return final_payload, self._files or None + files_copy = list(self._files) if self._files else None + return final_payload, files_copy class MailgunTemplateBuilder: @@ -534,7 +586,7 @@ def set_headers(self, headers: dict[str, str]) -> Self: Returns: The builder instance. """ - self._payload["headers"] = json.dumps(headers, separators=(",", ":")) + self._payload["headers"] = json.dumps(headers, separators=(",", ":"), default=str) return self def set_copy_requests(self, requests_list: list[dict[str, str]]) -> Self: diff --git a/mailgun/client.py b/mailgun/client.py index e179e2cd..1ceff049 100644 --- a/mailgun/client.py +++ b/mailgun/client.py @@ -187,7 +187,7 @@ def __getattr__(self, name: str) -> Any: AttributeError: If the requested route is unknown or a magic Python method is invoked. """ # Protect Data Model: Ignore magic Python methods - if name.startswith("__") and name.endswith("__"): + if (name.startswith("__") and name.endswith("__")) or name in {"config", "auth"}: msg = f"'{self.__class__.__name__}' object has no attribute '{name}'" raise AttributeError(msg) @@ -360,7 +360,9 @@ def _client(self) -> httpx.AsyncClient: if "transport" not in kwargs: limits = httpx.Limits(max_keepalive_connections=100, max_connections=100) kwargs["transport"] = httpx.AsyncHTTPTransport( - retries=3, limits=limits, verify=ssl_context + retries=3, + limits=limits, + verify=ssl_context, ) self._httpx_client = httpx.AsyncClient(**kwargs) diff --git a/mailgun/config.py b/mailgun/config.py index 399ba258..903820f9 100644 --- a/mailgun/config.py +++ b/mailgun/config.py @@ -147,7 +147,7 @@ class Config: # Use Mapping to denote read-only dictionary-like structures _HEADERS_BASE: Final[Mapping[str, str]] = MappingProxyType({"User-agent": USER_AGENT}) _HEADERS_JSON: Final[Mapping[str, str]] = MappingProxyType( - {"User-agent": USER_AGENT, "Content-Type": "application/json"} + {"User-agent": USER_AGENT, "Content-Type": "application/json"}, ) # --- ENCAPSULATED ROUTING REGISTRIES --- @@ -159,7 +159,7 @@ class Config: _DOMAIN_ALIASES: Final[Mapping[str, str]] = MappingProxyType(routes.DOMAIN_ALIASES) _DOMAIN_ENDPOINTS: Final[Mapping[str, tuple[str, ...]]] = MappingProxyType( - routes.DOMAIN_ENDPOINTS + routes.DOMAIN_ENDPOINTS, ) _V1_ENDPOINTS: Final[frozenset[str]] = frozenset(routes.DOMAIN_ENDPOINTS["v1"]) _V3_ENDPOINTS: Final[frozenset[str]] = frozenset(routes.DOMAIN_ENDPOINTS["v3"]) diff --git a/mailgun/endpoints.py b/mailgun/endpoints.py index 58faa4b0..95300daa 100644 --- a/mailgun/endpoints.py +++ b/mailgun/endpoints.py @@ -194,7 +194,7 @@ def __init__( self._auth = auth self._timeout = timeout self.dry_run = dry_run - self.retry_policy = None + self.retry_policy: RetryPolicy | None = None @staticmethod def _warn_if_deprecated(method: str, target_url: str) -> None: @@ -233,7 +233,9 @@ def _reset_stream_pointers(files: Any) -> None: @staticmethod def _prepare_payload( - data: Any | None, files: Any | None, headers: dict[str, str] + data: Any | None, + files: Any | None, + headers: dict[str, str], ) -> tuple[Any | None, dict[str, str]]: """Prepares headers and minifies JSON payloads or handles multipart files safely. @@ -257,7 +259,7 @@ def _prepare_payload( ) if is_json_request and data is not None and not isinstance(data, (str, bytes)): - data = json.dumps(data, separators=(",", ":")) + data = json.dumps(data, separators=(",", ":"), default=str) return data, working_headers @@ -376,7 +378,10 @@ def _prepare_request( safe_timeout = SecurityGuard.sanitize_timeout(actual_timeout) target_url = self.build_url( - url, domain=target_domain_normalized, method=safe_method, **kwargs + url, + domain=target_domain_normalized, + method=safe_method, + **kwargs, ) self._warn_if_deprecated(safe_method, target_url) @@ -385,6 +390,33 @@ def _prepare_request( return safe_method, target_url, safe_url_for_log, safe_timeout, safe_headers, safe_kwargs + @staticmethod + def _cast_query_param(orig_ref: Any, raw_values: list[str]) -> Any: + """Cast query string values to match the original filter parameter type. + + Args: + orig_ref: Reference value indicating the intended target type. + raw_values: List of string values extracted from the URL query string. + + Returns: + The parsed value cast to bool, int, float, list, tuple, set, or str. + """ + parsed_str = raw_values[0] if len(raw_values) == 1 else raw_values + + if isinstance(orig_ref, bool): + return str(raw_values[0]).lower() in {"true", "1", "yes"} + if isinstance(orig_ref, int): + return int(raw_values[0]) + if isinstance(orig_ref, float): + return float(raw_values[0]) + if isinstance(orig_ref, list): + return raw_values + if isinstance(orig_ref, tuple): + return tuple(raw_values) + if isinstance(orig_ref, set): + return set(raw_values) + return parsed_str + class Endpoint(BaseEndpoint): """Generate synchronous requests and return responses.""" @@ -517,11 +549,17 @@ def api_call( # noqa: PLR0914, PLR0915 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 + "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(), safe_url_for_log + "API Success %s | %s %s", + status_code, + safe_method.upper(), + safe_url_for_log, ) break @@ -645,6 +683,7 @@ def put( Args: data: Payload data to include in the request. filters: Query parameters to include in the request. + domain: Target domain name. **kwargs: Additional arguments to pass to the HTTP client. Returns: @@ -732,7 +771,12 @@ def delete(self, domain: str | None = None, **kwargs: Any) -> APIResponseType: """ merged_headers = self._merge_headers(kwargs) return self.api_call( - self._auth, "delete", self._url, headers=merged_headers, domain=domain, **kwargs + self._auth, + "delete", + self._url, + headers=merged_headers, + domain=domain, + **kwargs, ) def stream( @@ -748,7 +792,8 @@ def stream( Yields: Individual records from the paginated API response. """ - current_filters = dict(filters) if filters else {} + initial_filters = dict(filters) if filters else {} + current_filters = initial_filters.copy() while True: # Pass a copy of the dictionary so the mock (and the underlying request layer) @@ -759,13 +804,18 @@ def stream( response.raise_for_status() data = response.json() - items = data.get("items", []) + if not isinstance(data, dict): + break + items = data.get("items") or [] # Yield items one by one (Lazy Evaluation) yield from items + paging_dict = data.get("paging") or {} # Check for the next page cursor - next_url = data.get("paging", {}).get("next") + next_url = paging_dict.get("next") + if not next_url or not items: + break # Stop if there's no next URL or the current page was empty if not next_url or not items: @@ -778,30 +828,11 @@ def stream( if not v: continue - # Default flatten logic for unknown or string parameters - parsed_str_val = v[0] if len(v) == 1 else v - - # Prevent Query Parameter Type Drift - if k in current_filters: - original_val = current_filters[k] - - # Dynamically cast to the developer's original type - if isinstance(original_val, bool): - current_filters[k] = str(v[0]).lower() in {"true", "1", "yes"} - elif isinstance(original_val, int): - current_filters[k] = int(v[0]) - elif isinstance(original_val, float): - current_filters[k] = float(v[0]) - elif isinstance(original_val, list): - current_filters[k] = v # Always keep as list - elif isinstance(original_val, tuple): - current_filters[k] = tuple(v) # Always keep as tuple - elif isinstance(original_val, set): - current_filters[k] = set(v) # Always keep as set - else: - current_filters[k] = parsed_str_val + orig_ref = initial_filters.get(k, current_filters.get(k)) + if orig_ref is not None: + current_filters[k] = self._cast_query_param(orig_ref, v) else: - current_filters[k] = parsed_str_val + current_filters[k] = v[0] if len(v) == 1 else v # ============================================================================== @@ -1177,7 +1208,12 @@ async def delete(self, domain: str | None = None, **kwargs: Any) -> AsyncAPIResp """ merged_headers = self._merge_headers(kwargs) return await self.api_call( - self._auth, "delete", self._url, headers=merged_headers, domain=domain, **kwargs + self._auth, + "delete", + self._url, + headers=merged_headers, + domain=domain, + **kwargs, ) async def stream( @@ -1186,46 +1222,52 @@ async def stream( domain: str | None = None, **kwargs: Any, ) -> Any: - """Lazy pagination: yield records asynchronously one by one. + """Lazy pagination: yield records asynchronously one by one without loading all into memory. + + Automatically traverses the 'paging' links returned by the Mailgun API. Yields: Individual records from the paginated API response. + + Raises: + ApiError: If the server returns a 4xx or 5xx status code or a network error occurs. """ - current_filters = dict(filters) if filters else {} + initial_filters = dict(filters) if filters else {} + current_filters = initial_filters.copy() while True: response = await self.get(filters=current_filters.copy(), domain=domain, **kwargs) + # Defensive status check: Convert raw HTTPStatusError into SDK's standard ApiError if hasattr(response, "raise_for_status"): - response.raise_for_status() + try: + response.raise_for_status() + except httpx.HTTPStatusError as exc: + raise ApiError(exc.response) from exc data = response.json() - items = data.get("items", []) + # Stop if response payload is not a valid JSON mapping (e.g. error list or gateway shock) + if not isinstance(data, dict): + break + + items = data.get("items") or [] for item in items: yield item - next_url = data.get("paging", {}).get("next") + paging_dict = data.get("paging") or {} + next_url = paging_dict.get("next") if not next_url or not items: break + # 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) for k, v in query_params.items(): if not v: continue - parsed_str_val = v[0] if len(v) == 1 else v - - # Prevent Query Parameter Type Drift - if k in current_filters: - original_val = current_filters[k] - - # Dynamically cast to the developer's original type - if isinstance(original_val, bool): - current_filters[k] = str(v[0]).lower() in {"true", "1", "yes"} - elif isinstance(original_val, int): - current_filters[k] = int(v[0]) - elif isinstance(original_val, float): - current_filters[k] = float(v[0]) - else: - current_filters[k] = parsed_str_val + + orig_ref = initial_filters.get(k, current_filters.get(k)) + if orig_ref is not None: + current_filters[k] = self._cast_query_param(orig_ref, v) else: - current_filters[k] = parsed_str_val + current_filters[k] = v[0] if len(v) == 1 else v diff --git a/mailgun/ext/pydantic/models.py b/mailgun/ext/pydantic/models.py index 137727c7..a125b5a3 100644 --- a/mailgun/ext/pydantic/models.py +++ b/mailgun/ext/pydantic/models.py @@ -34,7 +34,7 @@ def _validate_emails(value: str | list[str]) -> str | list[str]: msg = f"Security Alert (CWE-113): CRLF injection detected in email: '{email}'" raise ValueError(msg) - # Quick format check. Ignore names (e.g., "John Doe ") + # 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}'" @@ -50,11 +50,9 @@ class SendMessageSchema(BaseModel): 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) + strict=True, ) # Required fields @@ -66,29 +64,47 @@ class SendMessageSchema(BaseModel): 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 + subject: str | None = Field(default=None, max_length=998) 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 + # Container for dynamic Mailgun parameters custom_params: dict[str, str] = Field(default_factory=dict) + @field_validator("subject", mode="after") + @classmethod + def validate_subject(cls, v: str | None) -> str | None: + """Poka-yoke: Prevent CRLF Injection in email Subject (CWE-113 / RFC 5322). + + Args: + v: The subject string to validate. + + Returns: + The validated subject string or None. + + Raises: + ValueError: If a CRLF injection sequence is detected in the subject. + """ + if v is not None and _CRLF_REGEX.search(v): + msg = f"Security Alert (CWE-113): CRLF injection detected in subject: '{v}'" + raise ValueError(msg) + return v + @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. + v: Dictionary of custom parameters. Returns: - The validated dictionary of custom parameters. + The validated custom parameters dictionary. Raises: - ValueError: If a key does not start with 'v:', 'h:', 'o:', or contains CRLFs. + ValueError: If an unknown prefix or CRLF injection sequence is detected. """ for key, val in v.items(): if not key.startswith(("v:", "h:", "o:")): @@ -98,7 +114,6 @@ def validate_prefixes(cls, v: dict[str, str]) -> dict[str, str]: ) 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) @@ -110,8 +125,11 @@ def validate_prefixes(cls, v: dict[str, str]) -> dict[str, str]: def check_email_formats(cls, v: Any) -> Any: """Validates the correct format of email addresses. + Args: + v: Raw email string or sequence of email strings. + Returns: - The validated input value. + The validated email string or sequence. """ if v is not None: _validate_emails(v) @@ -122,32 +140,28 @@ def validate_body(self) -> "SendMessageSchema": """Cross-validation of body content. Returns: - The validated schema instance. + The validated instance of SendMessageSchema. Raises: - ValueError: If no body parts are provided or invalid prefixes are used. + ValueError: If no message body components (text, html, template, amp_html) are provided. """ - # 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'." + "'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 + Dictionary payload formatted for direct submission to the Mailgun API. """ - # Get standard fields as a dict data: dict[str, Any] = self.model_dump( - by_alias=True, exclude_none=True, exclude={"custom_params"} + 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 aa406446..cb322c75 100644 --- a/mailgun/filters.py +++ b/mailgun/filters.py @@ -10,7 +10,7 @@ class RedactingFilter(logging.Filter): """ SECRET_PATTERN: Final[re.Pattern[str]] = re.compile(r"(key-|pubkey-)[\w\-]+") - MAX_REDACTION_DEPTH: Final[int] = 4 + MAX_REDACTION_DEPTH: Final[int] = 5 # Standard LogRecord attributes to ignore for maximum performance _STANDARD_ATTRS: Final[frozenset[str]] = frozenset( @@ -38,59 +38,137 @@ class RedactingFilter(logging.Filter): "thread", "threadName", "taskName", - } + }, ) def _redact_str(self, data: str) -> str: + """Sanitize a string using the compiled regex pattern. + + Args: + data: The string to sanitize. + + Returns: + The sanitized string with matching secrets redacted. + """ try: - return self.SECRET_PATTERN.sub(r"\1[REDACTED]", data) + return self.SECRET_PATTERN.sub(r"\1[REDACTED]", str(data)) except Exception: # ruff: ignore[blind-except] - return data + return str(data) if isinstance(data, str) else "" def _redact_dict(self, data: dict[Any, Any], depth: int) -> dict[Any, Any]: - return {k: self._deep_redact(v, depth + 1) for k, v in data.items()} + """Recursively sanitize dictionary values. + + Args: + data: The dictionary to sanitize. + depth: The current recursion depth. + + Returns: + A sanitized dictionary with redacted values. + """ + try: + return {k: self._deep_redact(v, depth + 1) for k, v in list(data.items())} + except Exception: # ruff: ignore[blind-except] + return data def _redact_list(self, data: list[Any], depth: int) -> list[Any]: - return [self._deep_redact(item, depth + 1) for item in data] + """Recursively sanitize list items. + + Args: + data: The list to sanitize. + depth: The current recursion depth. - def _redact_set(self, data: set[Any], depth: int) -> Any: + Returns: + A sanitized list with redacted values. + """ + try: + return [self._deep_redact(item, depth + 1) for item in data] + except Exception: # ruff: ignore[blind-except] + return data + + def _redact_set(self, data: set[Any] | frozenset[Any], depth: int) -> Any: + """Recursively sanitize set items with unhashable fallback. + + Args: + data: The set or frozenset to sanitize. + depth: The current recursion depth. + + Returns: + A sanitized set, frozenset, or list with redacted values. + """ try: - return {self._deep_redact(item, depth + 1) for item in data} + redacted = {self._deep_redact(item, depth + 1) for item in data} + return type(data)(redacted) except TypeError: # Fallback if redacted items become unhashable (e.g. dicts/lists) - return [self._deep_redact(item, depth + 1) for item in data] + try: + return [self._deep_redact(item, depth + 1) for item in data] + except Exception: # ruff: ignore[blind-except] + return data + except Exception: # ruff: ignore[blind-except] + return data def _redact_tuple(self, data: tuple[Any, ...], depth: int) -> tuple[Any, ...]: - if hasattr(data, "_fields"): # Safely unpack NamedTuples - try: - return type(data)(*(self._deep_redact(item, depth + 1) for item in data)) - except Exception: # ruff: ignore[blind-except, try-except-pass] - pass - return tuple(self._deep_redact(item, depth + 1) for item in data) + """Recursively sanitize tuple items preserving namedtuple structure. + + Args: + data: The tuple to sanitize. + depth: The current recursion depth. + + Returns: + A sanitized tuple or NamedTuple instance with redacted values. + """ + try: + if hasattr(data, "_fields"): # Safely unpack NamedTuples + try: + return type(data)(*(self._deep_redact(item, depth + 1) for item in data)) + except Exception: # ruff: ignore[blind-except, try-except-pass] + pass + return tuple(self._deep_redact(item, depth + 1) for item in data) + except Exception: # ruff: ignore[blind-except] + return data def _redact_object(self, data: Any, depth: int) -> Any: - if hasattr(data, "model_dump") and callable(data.model_dump): - try: - return self._deep_redact(data.model_dump(), depth + 1) - except Exception: # ruff: ignore[blind-except, try-except-pass] - pass + """Recursively sanitize custom objects, dataclasses, and Pydantic models. - if hasattr(data, "__dict__"): - try: - return self._deep_redact(vars(data), depth + 1) - except Exception: # ruff: ignore[blind-except, try-except-pass] - pass + Args: + data: The object instance to sanitize. + depth: The current recursion depth. + + Returns: + A sanitized representation of the object. + """ + try: + if hasattr(data, "model_dump") and callable(data.model_dump): + try: + return self._deep_redact(data.model_dump(), depth + 1) + except Exception: # ruff: ignore[blind-except, try-except-pass] + pass + except Exception: # ruff: ignore[blind-except, try-except-pass] + pass + + try: + if hasattr(data, "__dict__"): + try: + return self._deep_redact(vars(data), depth + 1) + except Exception: # ruff: ignore[blind-except, try-except-pass] + pass + except Exception: # ruff: ignore[blind-except, try-except-pass] + pass try: str_val = str(data) except Exception: # ruff: ignore[blind-except] - str_val = "" + return "" return self._redact_str(str_val) def _deep_redact(self, data: Any, depth: int = 0) -> Any: """Recursively sanitize strings, dictionaries, and iterables safely. + Args: + data: The data structure to sanitize. + depth: The current recursion depth. + Returns: A safely sanitized copy of the input data with secrets redacted. """ @@ -107,7 +185,7 @@ def _deep_redact(self, data: Any, depth: int = 0) -> Any: return self._redact_dict(data, depth) if isinstance(data, list): return self._redact_list(data, depth) - if isinstance(data, set): + if isinstance(data, (set, frozenset)): return self._redact_set(data, depth) if isinstance(data, tuple): return self._redact_tuple(data, depth) @@ -121,6 +199,9 @@ def _deep_redact(self, data: Any, depth: int = 0) -> Any: def filter(self, record: logging.LogRecord) -> bool: """Filter out sensitive secrets from log records safely. + Args: + record: The logging record to inspect and redact. + Returns: True to allow the record to be logged. """ @@ -130,13 +211,21 @@ def filter(self, record: logging.LogRecord) -> bool: record.msg = self._redact_str(record.msg) # 2. Redact tuple/dict args WITHOUT changing their types - if isinstance(record.args, (dict, tuple)): - record.args = self._deep_redact(record.args) + raw_args: Any = record.args + if raw_args: + if isinstance(raw_args, tuple): + record.args = tuple(self._deep_redact(arg, 0) for arg in raw_args) + elif isinstance(raw_args, dict): + record.args = {k: self._deep_redact(v, 0) for k, v in list(raw_args.items())} + elif isinstance(raw_args, list): + record.args = [self._deep_redact(item, 0) for item in raw_args] # type: ignore[assignment] + else: + record.args = self._deep_redact(raw_args, 0) # 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) + extra_keys = [k for k in list(record.__dict__.keys()) if k not in self._STANDARD_ATTRS] + for attr_name in extra_keys: + record.__dict__[attr_name] = self._deep_redact(record.__dict__[attr_name], 0) except Exception: # ruff: ignore[blind-except, try-except-pass] # Never let logging filters crash application execution pass diff --git a/mailgun/handlers/domains_handler.py b/mailgun/handlers/domains_handler.py index 2b364ae8..06f47810 100644 --- a/mailgun/handlers/domains_handler.py +++ b/mailgun/handlers/domains_handler.py @@ -9,6 +9,7 @@ from mailgun.endpoints import build_path_from_keys from mailgun.handlers.error_handler import ApiError +from mailgun.routes import DOMAIN_ALIASES from mailgun.security import SecurityGuard @@ -60,9 +61,8 @@ def handle_domains( # noqa: PLR0914 Raises: ApiError: If the domain is missing or options are invalid. """ - keys = list(url.get("keys", [])) - if "domains" in keys: - keys.remove("domains") + # Extract, strip "domains" prefix, and canonicalize aliases in a single pass + keys = [DOMAIN_ALIASES.get(k, k) for k in url.get("keys", []) if k != "domains"] base_url = str(url.get("base", "")).rstrip("/") @@ -108,7 +108,7 @@ def handle_domains( # noqa: PLR0914 safe_webhook = SecurityGuard.sanitize_path_segment(webhook_name) return f"{final_url}/{safe_webhook}" - # B. Credentials Logins (CRITICAL FIX: Correct path segment handling) + # B. Credentials Logins (Preserve literal '@' and handle domain duplicates) login_val = kwargs.pop("login", None) if "credentials" in keys and login_val is not None: login_str = str(login_val) @@ -119,7 +119,10 @@ def handle_domains( # noqa: PLR0914 if domain and domain_part == domain: safe_login = SecurityGuard.sanitize_path_segment(local_part) else: - safe_login = f"{SecurityGuard.sanitize_path_segment(local_part)}@{SecurityGuard.sanitize_path_segment(domain_part)}" + safe_login = ( + f"{SecurityGuard.sanitize_path_segment(local_part)}@" + f"{SecurityGuard.sanitize_path_segment(domain_part)}" + ) else: safe_login = SecurityGuard.sanitize_path_segment(login_str) @@ -191,10 +194,7 @@ def handle_mailboxes_credentials( Raises: ApiError: If the domain is missing. """ - keys = list(url.get("keys", [])) - - if "domains" in keys: - keys.remove("domains") + keys = [k for k in url.get("keys", []) if k != "domains"] base_url = str(url["base"]).rstrip("/") diff --git a/mailgun/handlers/email_validation_handler.py b/mailgun/handlers/email_validation_handler.py index 0af34396..4185b70d 100644 --- a/mailgun/handlers/email_validation_handler.py +++ b/mailgun/handlers/email_validation_handler.py @@ -5,6 +5,7 @@ from __future__ import annotations +from collections.abc import Sequence from typing import Any from mailgun.endpoints import build_path_from_keys @@ -28,8 +29,14 @@ def handle_address_validate( Returns: The final URL for the email validation endpoint. """ - final_keys = build_path_from_keys(url.get("keys", [])[1:]) - base_url = str(url["base"]).rstrip("/") + raw_keys = url.get("keys") if isinstance(url, dict) else None + if isinstance(raw_keys, Sequence) and not isinstance(raw_keys, (str, bytes)): + keys = list(raw_keys) + else: + keys = [] + + final_keys = build_path_from_keys(keys[1:] if len(keys) > 1 else []) + base_url = str(url.get("base", "")).rstrip("/") if "list_name" in kwargs: safe_list = SecurityGuard.sanitize_path_segment(kwargs["list_name"]) diff --git a/mailgun/routes.py b/mailgun/routes.py index b1aaf529..808f6bfb 100644 --- a/mailgun/routes.py +++ b/mailgun/routes.py @@ -37,7 +37,8 @@ # Account level definitions "account_templates": ("v4", ("templates",)), "account_webhooks": ("v1", ("webhooks",)), - # Validation Service + "routes_match": ("v3", ("routes", "match")), + # Validation Service (v4 Validate API) "addressvalidate": ("v4", ("address", "validate")), "addressparse": ("v4", ("address", "parse")), "address_bulk": ("v4", ("address", "validate", "bulk")), @@ -45,7 +46,6 @@ "address_preview": ("v4", ("address", "validate", "preview")), # Standard Domain Endpoints (Merged paths to avoid handle_domains intercept) "spamtraps": ("v2", ("spamtraps",)), - "blocklists": ("v3", ("domains", "{domain}", "blocklists")), # MTLS and DKIM Management "x509": ("v2", ("x509", "{domain}")), "x509_status": ("v2", ("x509", "{domain}", "status")), @@ -74,7 +74,7 @@ "forwards": ("v3", "", None), "ip_pools": ("v3", "", None), "ip_warmups": ("v3", "", None), - "ip_whitelist": ("v2", "ip", "whitelist"), + "ip_whitelist": ("v2", "", "ip_whitelist"), "ips": ("v3", "", None), "lists": ("v3", "", None), "mailboxes": ("v3", "", None), @@ -110,6 +110,7 @@ "preview": ("v1", "", None), "preview_v2": ("v2", "", "preview"), "reputationanalytics": ("v1", "", None), + "reputationanalytics_v2": ("v2", "", "reputationanalytics"), } PREFIX_ROUTES: Final = MappingProxyType(_PREFIX_ROUTES) @@ -129,8 +130,9 @@ # --- DOMAIN_ENDPOINTS --- # Grouping endpoints by versions for smart routing. _DOMAIN_ENDPOINTS: DomainsEndpointsType = { - "v1": ("dkim", "security"), - "v4": ("ips", "connections"), + "v1": ("dkim_management", "monitoring", "security"), + "v2": ("x509",), + "v4": ("connections", "ips", "keys"), "v3": ( "bounces", "click", @@ -175,6 +177,7 @@ # Defined as strings to prevent expensive regex compilation on cold boot. _DEPRECATED_ROUTES_PATTERNS: Final[dict[str, str]] = { r"^/v1/bounce-classification/": "The v1 bounce-classification API is deprecated. Migrate to POST /v2/bounce-classification/metrics.", + r"^/v1/spamtraps": "The v1 spamtraps APIs (/v1/spamtraps, /v1/spamtraps/totals, /v1/spamtraps/filters) are deprecated. Migrate to GET /v2/spamtraps.", r"^/v3/(stats|[^/]+/stats|[^/]+/aggregates)": "The v3 Stats API is deprecated. Migrate to the v1 Metrics API.", r"^/v3/[^/]+/tag(/|$|\?)": "The legacy Tag API is deprecated. Migrate to the new Tags API (/v3/{domain}/tags).", r"^/v3/domains/[^/]+/limits/tag": "The domain tag limits API is deprecated.", @@ -191,5 +194,5 @@ def get_deprecated_regexes() -> MappingProxyType[re.Pattern[str], str]: A read-only mapping of compiled regular expressions to their deprecation messages. """ return MappingProxyType( - {re.compile(pattern): msg for pattern, msg in _DEPRECATED_ROUTES_PATTERNS.items()} + {re.compile(pattern): msg for pattern, msg in _DEPRECATED_ROUTES_PATTERNS.items()}, ) diff --git a/mailgun/security.py b/mailgun/security.py index a69609e9..e4497571 100644 --- a/mailgun/security.py +++ b/mailgun/security.py @@ -24,6 +24,7 @@ # Constants for API error handling and logging (fixes Ruff PLR2004) _AUTH_TUPLE_LEN: Final = 2 +_MAX_IDEMPOTENCY_DEPTH: Final = 50 # Regex to detect any ASCII control character EXCEPT horizontal tab (\x09) # Compliant with RFC 9110 Section 5.5 _CONTROL_CHAR_RE: Final = re.compile(r"[\x00-\x08\x0a-\x1f\x7f]") @@ -32,7 +33,7 @@ _XSS_PATTERN: Final = re.compile(r"<(script|svg)|javascript:|onload=", re.IGNORECASE) ALLOWED_HOSTS: Final = frozenset( - {"mailgun.net", "mailgun.org", "mailgun.com", "localhost", "127.0.0.1"} + {"mailgun.net", "mailgun.org", "mailgun.com", "localhost", "127.0.0.1"}, ) ALLOWED_SUFFIXES: Final = (".mailgun.net", ".mailgun.org", ".mailgun.com") ALLOWED_SCHEMES: Final = frozenset({"https", "http"}) @@ -67,9 +68,7 @@ def proxy_manager_for(self, proxy: str, **proxy_kwargs: Any) -> Any: 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) @@ -90,15 +89,18 @@ class SecurityGuard: easy to extract into a dedicated security module in future releases. """ + ALLOWED_SCHEMES: Final[frozenset[str]] = frozenset({"https", "http"}) ALLOWED_HTTP_METHODS: Final[frozenset[str]] = frozenset( - {"GET", "POST", "PUT", "DELETE", "PATCH", "OPTIONS", "HEAD"} + {"GET", "POST", "PUT", "DELETE", "PATCH", "OPTIONS", "HEAD"}, ) ALLOWED_API_HOSTS: Final[tuple[str, ...]] = ( "mailgun.net", "mailgun.org", + "mailgun.com", "localhost", "127.0.0.1", ) + ALLOWED_SUFFIXES: tuple[str, ...] = (".mailgun.net", ".mailgun.org", ".mailgun.com") ALLOWED_KWARGS: Final[frozenset[str]] = frozenset({"proxies", "cert"}) SAFE_KEY_PATTERN: Final[re.Pattern[str]] = re.compile(r"^[a-z0-9_]+$") CRLF_SLASH_PATTERN: Final[re.Pattern[str]] = re.compile(r"[\r\n/\\]+") @@ -123,6 +125,10 @@ def sanitize_api_url(cls, raw_url: str) -> str: raw_url = f"https://{raw_url}" parsed = urlparse(raw_url) + if parsed.scheme not in cls.ALLOWED_SCHEMES: + msg = f"Security Alert (CWE-918): Forbidden URL scheme '{parsed.scheme}'." + raise ValueError(msg) + if parsed.scheme == "http" and parsed.hostname not in {"localhost", "127.0.0.1"}: msg = ( "CRITICAL SECURITY: Cleartext HTTP transmission is prohibited (CWE-319). Use HTTPS." @@ -188,7 +194,7 @@ def sanitize_key(cls, key: str) -> str: @classmethod def sanitize_domain(cls, domain: str | None) -> str | None: - """Protect against Path Traversal in URL construction. + """Protect against Path Traversal and Encoding Bypasses in URL construction. Args: domain: Target domain name to sanitize. @@ -197,20 +203,31 @@ def sanitize_domain(cls, domain: str | None) -> str | None: The sanitized domain name or None. Raises: - ValueError: If path traversal characters are detected. + ValueError: If path traversal characters or excessive encodings are detected. """ if not domain: return None - decoded_domain = unquote(domain) + decoded = str(domain) + for _ in range(3): + new_decoded = unquote(decoded) + if new_decoded == decoded: + break + decoded = new_decoded + else: + raise ValueError("Security Alert (CWE-116): Excessive URL encoding detected.") + + decoded = unicodedata.normalize("NFKC", decoded) - # Poka-yoke: Actively strip all slashes and newlines (Advanced Traversal & CRLF) - safe_domain = cls.CRLF_SLASH_PATTERN.sub("", decoded_domain).strip() + # Poka-yoke: Cut CRLF and slashes, normalize whitespaces + safe_domain = cls.CRLF_SLASH_PATTERN.sub("", decoded).strip() + # Check path traversal after removing slashes ("mytest.com/....//path" -> "mytest.com....path") if ".." in safe_domain: raise ValueError( - "CRITICAL SECURITY: Path traversal characters detected in domain parameter." + "CRITICAL SECURITY: Path traversal characters detected in domain parameter.", ) + return safe_domain @classmethod @@ -237,8 +254,8 @@ def sanitize_timeout(cls, timeout: TimeoutType) -> TimeoutType: """Prevent Infinite Timeout Thread Exhaustion (DoS). Strict Creation-Time Timeout Constraints & Float Validation. - Prevents thread pool exhaustion from infinite blocking (CWE-400). - Enforces a strict maximum boundary of 300 seconds. + Prevents thread pool exhaustion from infinite blocking. + Enforces a strict maximum boundary of 300 seconds (CWE-400). Args: timeout: The requested timeout value. @@ -247,8 +264,7 @@ def sanitize_timeout(cls, timeout: TimeoutType) -> TimeoutType: The safely verified timeout value. Raises: - ValueError: If the timeout is None, negative, zero, non-finite, - exceeds 300 seconds, or a tuple with an incorrect number of elements. + ValueError: If timeout is None, non-finite, out of bounds, or an invalid tuple. """ if timeout is None: msg = ( @@ -277,7 +293,11 @@ def _validate_float(val: Any) -> float: msg = f"Timeout must be a numeric value, got {type(val).__name__}" raise TypeError(msg) - f_val = float(val) + # In SecurityGuard._validate_float (mailgun/security.py) + try: + f_val = float(val) + except OverflowError as e: + raise ValueError("Timeout value exceeds maximum scalar float capacity.") from e if math.isnan(f_val) or math.isinf(f_val): raise ValueError("Timeout must be a finite number.") @@ -285,7 +305,7 @@ def _validate_float(val: Any) -> float: 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." + "Security Alert: Timeout exceeds maximum allowed boundary of 300 seconds.", ) return f_val @@ -294,7 +314,7 @@ def _validate_float(val: Any) -> float: expected_tuple_length = 2 if len(timeout) != expected_tuple_length: raise ValueError( - "Timeout must be a tuple containing exactly two elements: (connect, read)." + "Timeout must be a tuple containing exactly two elements: (connect, read).", ) return _validate_float(timeout[0]), _validate_float(timeout[1]) @@ -313,26 +333,70 @@ def filter_safe_kwargs(cls, kwargs: dict[str, Any]) -> dict[str, Any]: return {k: v for k, v in kwargs.items() if k in cls.ALLOWED_KWARGS} @staticmethod - def sanitize_headers(headers: dict[str, str] | None) -> dict[str, str] | None: - """Poka-yoke: Prevent HTTP Header Injection (CWE-113). + def sanitize_headers(headers: dict[Any, Any] | None) -> dict[str, str] | None: + """Poka-yoke: Prevent HTTP Header Injection (CWE-113 / RFC 9110). + + Validates header keys and values against CRLF injection, control characters, + and non-ASCII/non-Latin-1 encodings that cause runtime divergences across + HTTP engines (requests vs. httpx). Multi-value headers (lists/tuples/sets) + are safely joined into RFC 9110 comma-delimited strings. + + Args: + headers: Dictionary of headers to sanitize, or None. Returns: - The sanitized headers dictionary, or None if no headers were provided. + The sanitized, strictly string-typed headers dictionary, or None if headers is None. Raises: - ValueError: If a CRLF injection pattern is detected in any header key or value. + ValueError: If CRLF sequences or forbidden control characters are detected. + UnicodeEncodeError: If headers cannot be encoded to ISO-8859-1 (Latin-1). """ - if not headers: - return headers - for key, value in headers.items(): - # Check both key and value - if "\n" in str(key) or "\r" in str(key) or "\n" in str(value) or "\r" in str(value): - # PEP 578: Emit Enterprise security telemetry before crashing - sys.audit("mailgun.security.header_injection", key) - - msg = f"CRLF injection detected in header: {key}" + if headers is None: + return None + + sanitized: dict[str, str] = {} + for raw_key, raw_value in headers.items(): + if raw_key is None: + continue + + k_str = str(raw_key) + + # Flatten multi-value headers (e.g. list, tuple, set) per RFC 9110 + if isinstance(raw_value, (list, tuple, set)): + items = [str(item) for item in raw_value if item is not None] + v_str = ", ".join(items) + elif raw_value is None: + v_str = "" + else: + v_str = str(raw_value) + + # 1. Block CRLF and forbidden control characters (CWE-113 / RFC 9110) + has_crlf = any(c in k_str or c in v_str for c in ("\r", "\n")) + if has_crlf or _CONTROL_CHAR_RE.search(k_str) or _CONTROL_CHAR_RE.search(v_str): + if "sys" in sys.modules: + sys.audit("mailgun.security.header_injection", k_str) + + msg = f"CRLF injection detected in header: {k_str}" raise ValueError(msg) - return headers + + # 2. Strict HTTP Wire Encoding Check (RFC 9110 / ISO-8859-1 Parity) + try: + k_str.encode("latin-1") + v_str.encode("latin-1") + except UnicodeEncodeError as err: + if "sys" in sys.modules: + sys.audit("mailgun.security.header_encoding_divergence", k_str) + raise UnicodeEncodeError( + err.encoding, + err.object, + err.start, + err.end, + f"Header '{k_str}' contains characters outside the ISO-8859-1/Latin-1 standard", + ) from err + + sanitized[k_str] = v_str + + return sanitized @staticmethod def validate_no_control_characters(value: str, context: str = "Input") -> None: @@ -342,7 +406,8 @@ def validate_no_control_characters(value: str, context: str = "Input") -> None: ValueError: If control characters are detected. """ if _CONTROL_CHAR_RE.search(str(value)): - sys.audit("mailgun.security.control_characters", context) + if "sys" in sys.modules: + sys.audit("mailgun.security.control_characters", context) msg = f"Security Alert (CWE-20): Control characters detected in {context}: {value!r}" raise ValueError(msg) @@ -355,13 +420,13 @@ def sanitize_path_segment(cls, segment: Any) -> str: The URL-encoded path segment string. Raises: - TypeError: If the segment is not a string, int, or float. + TypeError: If the segment is a complex container or boolean. ValueError: If path traversal or invalid characters are detected. """ if segment is None: return "" - if isinstance(segment, (dict, list, set, bool)): + if isinstance(segment, (dict, list, set, tuple, bool)): msg = f"Security Alert: Invalid segment type {type(segment).__name__}." raise TypeError(msg) @@ -422,20 +487,22 @@ def validate_mailgun_url(url: str) -> str: if not hostname: raise ValueError("Security Alert: Missing hostname in URL.") - if scheme and scheme not in ALLOWED_SCHEMES: - sys.audit("mailgun.security.ssrf_scheme_violation", scheme) + if not scheme or scheme not in ALLOWED_SCHEMES: + if "sys" in sys.modules: + sys.audit("mailgun.security.ssrf_scheme_violation", scheme) msg = f"Security Alert (CWE-319): Forbidden URL scheme '{scheme}'." raise ValueError(msg) if scheme == "http" and hostname not in {"localhost", "127.0.0.1"}: raise ValueError( - "Security Alert (CWE-319): Plaintext HTTP is forbidden for external URLs." + "Security Alert (CWE-319): Plaintext HTTP is forbidden for external URLs.", ) is_safe = hostname in ALLOWED_HOSTS or hostname.endswith(ALLOWED_SUFFIXES) if not is_safe: - sys.audit("mailgun.security.ssrf_attempt", url) + if "sys" in sys.modules: + sys.audit("mailgun.security.ssrf_attempt", url) msg = f"Security Alert (CWE-918): Untrusted external hostname '{hostname}'." raise ValueError(msg) @@ -443,7 +510,8 @@ def validate_mailgun_url(url: str) -> str: @staticmethod def validate_attachment_path( - file_path: str | Path, safe_base_dir: str | Path | None = None + file_path: str | Path, + safe_base_dir: str | Path | None = None, ) -> Path: """Poka-yoke: Prevent Path Traversal (CWE-22) when reading attachments. @@ -472,7 +540,7 @@ def validate_attachment_path( # 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." + "Security Alert (CWE-22): Path traversal tokens ('..') are explicitly forbidden.", ) # Allow files residing in the OS temporary directory @@ -486,7 +554,7 @@ def validate_attachment_path( path_str = str(target_path).lower() if any(path_str.startswith(root.lower()) for root in forbidden_roots): raise ValueError( - "Security Alert: Access to sensitive OS system directories is explicitly forbidden." + "Security Alert: Access to sensitive OS system directories is explicitly forbidden.", ) forbidden_components = { @@ -501,7 +569,7 @@ def validate_attachment_path( } if any(part.lower() in forbidden_components for part in target_path.parts): raise ValueError( - "Security Alert: Access to sensitive OS system directories is explicitly forbidden." + "Security Alert: Access to sensitive OS system directories is explicitly forbidden.", ) return target_path @@ -551,26 +619,25 @@ def verify_webhook( token: str, timestamp: str | int, signature: str, - max_age_seconds: int = 300, + max_age_seconds: int = 900, ) -> bool: """Cryptographically verify a Mailgun webhook signature. - Protects against CWE-347 (Improper Verification), CWE-208 (Timing Attacks), - and CWE-294 (Capture-Replay Attacks). + Protects against CWE-347, CWE-208 (Timing Attacks), and CWE-294 (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. + max_age_seconds: Maximum allowed age in seconds (default 15m; <=0 disables TTL). Returns: True if the signature mathematically matches and is within TTL, False otherwise. Raises: - TypeError: If the signature components are invalid types. - ValueError: If the cryptographic payload or timestamp is invalid or out of bounds. + TypeError: If signature components are invalid types. + ValueError: If cryptographic payload or timestamp is invalid or out of bounds. """ # 1. Type Guard: Prevent AttributeError and Type Confusion if not isinstance(token, str) or not isinstance(signature, str): @@ -586,15 +653,16 @@ def verify_webhook( raise TypeError("Security Alert: Webhook timestamp must be a valid integer.") from e # 3. TTL/Replay Attack Prevention (CWE-294) - try: - if abs(time.time() - ts_math) > max_age_seconds: - logger.warning("Security Alert (CWE-294): Webhook timestamp expired.") - return False - except (TypeError, ValueError, OverflowError) as e: - # If the timestamp is wildly out of bounds, it's invalid. - raise ValueError( - "Security Alert: Invalid cryptographic payload or timestamp out of bounds." - ) from e + if max_age_seconds > 0: + try: + if abs(time.time() - ts_math) > max_age_seconds: + logger.warning("Security Alert (CWE-294): Webhook timestamp expired.") + return False + except (TypeError, ValueError, OverflowError) as e: + # If the timestamp is wildly out of bounds, it's invalid. + raise ValueError( + "Security Alert: Invalid cryptographic payload or timestamp out of bounds.", + ) from e # 4. Canonicalization: Encode securely if isinstance(signing_key, str): @@ -618,20 +686,26 @@ def normalize_domain(domain: str | None) -> str: attempt to route to or build URLs with non-ASCII domains (e.g., Cyrillic). Args: - domain: The target domain name + domain: The target domain or email address. Returns: - The ASCII-safe Punycode string + The ASCII-safe Punycode string. Raises: ValueError: If invalid domain name encoding. """ - if not domain: + if not domain or not isinstance(domain, str): return "" + if any(c in domain for c in ("\r", "\n", "\x00")): + msg = "Domain contains illegal control characters" + raise ValueError(msg) + 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. + local_part, sep, domain_part = domain.rpartition("@") + if sep: + normalized_host = domain_part.encode("idna").decode("ascii") + return f"{local_part}@{normalized_host}" return domain.encode("idna").decode("ascii") except UnicodeError as e: # Fallback or raise a clear validation error if the domain is completely malformed @@ -650,6 +724,8 @@ class SpamReport(TypedDict): class _SpamGuardParser(HTMLParser): """Internal lightning-fast HTML parser for detecting structural spam triggers.""" + _BLOCKED_TAGS: Final = frozenset({"script", "iframe", "object", "embed", "applet"}) + def __init__(self) -> None: super().__init__() self.issues: list[str] = [] @@ -658,16 +734,26 @@ def __init__(self) -> None: self.has_scripts = False def handle_starttag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None: - attr_dict = dict(attrs) - - if tag == "img": + tag_lower = tag.lower() + if tag_lower == "img": self.image_count += 1 + attr_dict = dict(attrs) if "alt" not in attr_dict or not attr_dict["alt"]: self.has_alt_tags = False - if tag == "script": + if tag_lower in self._BLOCKED_TAGS: self.has_scripts = True - self.issues.append("CRITICAL: " +"" +"" +"', + '