diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index fea0dab..1b692d8 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -105,7 +105,7 @@ repos: name: "🔒 security · Detect private keys" - repo: https://github.com/commitizen-tools/commitizen - rev: v4.18.1 + rev: v4.19.2 hooks: - id: commitizen name: "🌳 git · Validate commit message" @@ -134,7 +134,7 @@ repos: additional_dependencies: [".[toml]"] - repo: https://github.com/semgrep/pre-commit - rev: 'v1.177.0' + rev: 'v1.180.0' hooks: - id: semgrep name: "🔒 security · Static analysis (semgrep)" @@ -143,14 +143,14 @@ repos: # Spelling and typos - repo: https://github.com/crate-ci/typos - rev: v1.50.2 + rev: v1.51.1 hooks: - id: typos name: "📝 spelling · Check typos" # CI/CD validation - repo: https://github.com/python-jsonschema/check-jsonschema - rev: 0.38.0 + rev: 0.38.2 hooks: - id: check-dependabot name: "🔧 ci/cd · Validate Dependabot config" @@ -172,7 +172,7 @@ repos: - responses - repo: https://github.com/astral-sh/ruff-pre-commit - rev: v0.16.8 + rev: v0.16.10 hooks: - id: ruff-check name: "🐍 lint · Check with Ruff" @@ -191,7 +191,7 @@ repos: # Python type checking - repo: https://github.com/pre-commit/mirrors-mypy - rev: v2.3.1 + rev: v2.4.0 hooks: - id: mypy name: "🐍 types · Check with mypy" diff --git a/CHANGELOG.md b/CHANGELOG.md index 12e6804..76c311a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,33 @@ We [keep a changelog.](http://keepachangelog.com/) ## [Unreleased] +## [1.10.0] - 2026-10-08 + +### Changed + +- **Async Engine Migration (`httpx2`):** Migrated the asynchronous engine exclusively to `httpx2 >=2.7.0`, removing `httpx >=0.24.0` from runtime dependencies across `pyproject.toml`, `environment.yaml`, and `environment-dev.yaml`. +- **Core HTTP Import Refactoring:** Updated `mailgun.client`, `mailgun.endpoints`, `mailgun.security`, and `mailgun.types` to interface directly with `httpx2` for client instances (`httpx2.AsyncClient`), requests (`httpx2.Request`), responses (`httpx2.Response`), timeouts (`httpx2.Timeout`), and network exceptions (`httpx2.RequestError`, `httpx2.TimeoutException`, `httpx2.ConnectError`, `httpx2.NetworkError`, `httpx2.HTTPStatusError`). +- **Tooling & Static Analysis Alignment:** Synchronized pre-commit hooks (`commitizen v4.19.2`, `semgrep v1.180.0`, `typos v1.51.1`, `check-jsonschema 0.38.2`, `ruff v0.16.10`, and `mypy v2.4.0`). Added `COM812` (`missing-trailing-comma`) to `[tool.ruff.lint.ignore]` to eliminate Ruff formatter conflicts, and configured test-suite rule relaxations under `[tool.ruff.lint.per-file-ignores]` for `tests/**`. +- **Security Policy Update:** Updated `SECURITY.md` supported versions table to mark `1.10.x` supported and deprecate `< 1.10.0`. +- **Fuzzing Harness Latency & Retry Optimization:** Mocked `time.sleep` in `tests/fuzz/fuzz_audit_events.py` to eliminate artificial delays during fuzz test runs, passed a zero-retry policy configuration (`_NO_RETRY_CONFIG`) to the client, and safely handled expected `ValueError` exceptions when testing hostile URLs via `SecurityGuard.validate_mailgun_url`. + +### Removed + +- **`_httpx_compat` Bridge Layer:** Removed `mailgun/_httpx_compat.py` and its runtime detection flag (`HAS_HTTPX2`), eliminating dual-dependency branching and legacy import redirection. +- **Obsolete Compat Tests:** Removed `tests/unit/test_httpx_compat.py`. + +### Fixed + +- **Stream Pagination Hardening:** Added duplicate and stagnant cursor detection (`prev_url`) in both `Endpoint.stream()` and `AsyncEndpoint.stream()` to prevent infinite loops. Wrapped synchronous `requests.HTTPError` in `ApiError` for contract parity with the async client, and eliminated redundant loop termination checks. +- **Retry Short-Circuiting on Unrecoverable Client Errors:** Suppressed retry attempts in `Endpoint` and `AsyncEndpoint` for permanent client-side exceptions. Configuration errors (`InvalidURL`, `InvalidHeader`, `InvalidSchema`, `MissingSchema`) and protocol/decoding errors (`UnsupportedProtocol`, `DecodingError`) are now dispatched directly to `_handle_api_error` without backoff delays. +- **Telemetry Log Item Index Guard:** Added an `items` existence check in `test_post_query_get_account_logs` across `test_integration_async.py` and `test_integration_sync.py` to prevent `IndexError` when querying accounts with empty log retention windows. +- **Test Suite Quality & Formatting:** Corrected verbose exception re-raising in `test_integration_coverage.py`, added missing placeholder formatting to `sdk_logger.info` calls, and formatted all unit, integration, and fuzz test targets using `ruff format`. + +### Pull Requests Merged + +- PR #68: build(deps): Bump the minor-and-patch group with 2 updates. +- PR #70: build(deps): Bump the minor-and-patch group with 2 updates. + ## [1.9.1] - 2026-09-21 ### Security @@ -505,6 +532,7 @@ We [keep a changelog.](http://keepachangelog.com/) [1.0.1]: https://github.com/mailgun/mailgun-python/releases/tag/v1.0.1 [1.0.2]: https://github.com/mailgun/mailgun-python/releases/tag/v1.0.2 [1.1.0]: https://github.com/mailgun/mailgun-python/releases/tag/v1.1.0 +[1.10.0]: https://github.com/mailgun/mailgun-python/releases/tag/v1.10.0 [1.2.0]: https://github.com/mailgun/mailgun-python/releases/tag/v1.2.0 [1.3.0]: https://github.com/mailgun/mailgun-python/releases/tag/v1.3.0 [1.4.0]: https://github.com/mailgun/mailgun-python/releases/tag/v1.4.0 @@ -515,4 +543,4 @@ We [keep a changelog.](http://keepachangelog.com/) [1.8.0]: https://github.com/mailgun/mailgun-python/releases/tag/v1.8.0 [1.9.0]: https://github.com/mailgun/mailgun-python/releases/tag/v1.9.0 [1.9.1]: https://github.com/mailgun/mailgun-python/releases/tag/v1.9.1 -[unreleased]: https://github.com/mailgun/mailgun-python/compare/v1.9.1...HEAD +[unreleased]: https://github.com/mailgun/mailgun-python/compare/v1.10.0...HEAD diff --git a/README.md b/README.md index 0d85539..c9dd7bf 100644 --- a/README.md +++ b/README.md @@ -145,8 +145,7 @@ 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 `httpx2 >=2.7.0`. -Async client automatically detects and uses `httpx2` if available, falling back seamlessly to legacy `httpx`. +At runtime the package requires `requests >=2.33.0`. For async support, it requires `httpx2 >=2.7.0`. ### Test dependencies @@ -675,7 +674,7 @@ data = { "to": ["recipient1@example.com", "recipient2@example.com"], "subject": "Advanced Mailgun Features", "text": "Testing out tags, custom variables, and testmode!", - "o:tag": ["newsletter", "python-sdk"], # Multiple tags + "o:tag": ["async-integration-test", "httpx2-sdk"], # Multiple tags "o:testmode": "yes", # Validates payload without actually sending "o:deliverytime-optimize-period": "24h", # Send Time Optimization "v:my-custom-id": "USER-12345", # Custom user-defined variable diff --git a/SECURITY.md b/SECURITY.md index 979fb7e..5dc453d 100644 --- a/SECURITY.md +++ b/SECURITY.md @@ -2,10 +2,10 @@ ## Supported Versions -| Version | Supported | -| ------- | ------------------ | -| 1.9.x | :white_check_mark: | -| < 1.9.0 | :x: | +| Version | Supported | +| -------- | ------------------ | +| 1.10.x | :white_check_mark: | +| < 1.10.0 | :x: | # Vulnerability Disclosure diff --git a/environment-dev.yaml b/environment-dev.yaml index ce5eb38..07541f1 100644 --- a/environment-dev.yaml +++ b/environment-dev.yaml @@ -12,7 +12,6 @@ dependencies: # runtime deps - requests >=2.33.0 - httpx2 >=2.7.0 - - httpx >=0.24.0 # extras - fastapi - pydantic >=2.0.0 diff --git a/environment.yaml b/environment.yaml index 5e49b0c..b222416 100644 --- a/environment.yaml +++ b/environment.yaml @@ -9,7 +9,6 @@ dependencies: # runtime deps - requests >=2.33.0 - httpx2 >=2.7.0 - - httpx >=0.24.0 # tests - pytest >=9.0.3 # other diff --git a/mailgun/_httpx_compat.py b/mailgun/_httpx_compat.py deleted file mode 100644 index eb9f532..0000000 --- a/mailgun/_httpx_compat.py +++ /dev/null @@ -1,28 +0,0 @@ -"""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/client.py b/mailgun/client.py index 1ceff04..8bccd50 100644 --- a/mailgun/client.py +++ b/mailgun/client.py @@ -22,9 +22,9 @@ from http import HTTPStatus from typing import TYPE_CHECKING, Any, Final, Self +import httpx2 import requests # pyright: ignore[reportMissingModuleSource] -from mailgun._httpx_compat import httpx from mailgun.config import Config from mailgun.endpoints import AsyncEndpoint, BaseEndpoint, Endpoint from mailgun.filters import RedactingFilter @@ -282,7 +282,7 @@ def ping(self) -> bool: class AsyncClient(BaseClient): - """Async client class using httpx.""" + """Async client class using httpx2.""" __slots__ = ("_client_kwargs", "_httpx_client") @@ -301,7 +301,7 @@ def __init__( """ super().__init__(auth=auth, **kwargs) self._client_kwargs = kwargs.get("client_kwargs", {}) - self._httpx_client: httpx.AsyncClient | None = None + self._httpx_client: httpx2.AsyncClient | None = None def __getattr__(self, name: str) -> Any: """Resolve and return the requested API endpoint instance. @@ -341,14 +341,14 @@ def __getattr__(self, name: str) -> Any: return endpoint @property - def _client(self) -> httpx.AsyncClient: - """Provide lazy initialization for the underlying httpx.AsyncClient. + def _client(self) -> httpx2.AsyncClient: + """Provide lazy initialization for the underlying httpx2.AsyncClient. Returns: - The active httpx.AsyncClient instance. + The active httpx2.AsyncClient instance. """ # Assign to a local variable so Pyright can properly narrow the type - current_client: httpx.AsyncClient | None = getattr(self, "_httpx_client", None) + current_client: httpx2.AsyncClient | None = getattr(self, "_httpx_client", None) if current_client is None or current_client.is_closed: # Enforce TLS 1.2+ for httpx (CWE-319) @@ -358,20 +358,20 @@ def _client(self) -> httpx.AsyncClient: # Check if the user already provided a custom transport (e.g. for mocking) kwargs = self._client_kwargs.copy() if "transport" not in kwargs: - limits = httpx.Limits(max_keepalive_connections=100, max_connections=100) - kwargs["transport"] = httpx.AsyncHTTPTransport( + limits = httpx2.Limits(max_keepalive_connections=100, max_connections=100) + kwargs["transport"] = httpx2.AsyncHTTPTransport( retries=3, limits=limits, verify=ssl_context, ) - self._httpx_client = httpx.AsyncClient(**kwargs) + self._httpx_client = httpx2.AsyncClient(**kwargs) return self._httpx_client return current_client async def aclose(self) -> None: - """Close the underlying httpx.AsyncClient and purge memory.""" + """Close the underlying httpx2.AsyncClient and purge memory.""" if self._httpx_client: try: # CWE-316: Clear async session diff --git a/mailgun/endpoints.py b/mailgun/endpoints.py index 95300da..24e2aa9 100644 --- a/mailgun/endpoints.py +++ b/mailgun/endpoints.py @@ -10,11 +10,11 @@ from typing import TYPE_CHECKING, Any, Final from urllib.parse import parse_qs, urlparse +import httpx2 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 @@ -276,12 +276,12 @@ def _handle_api_error(e: Exception, method: str, target_url: str) -> None: MailgunTimeoutError: If the request times out. ApiError: If network routing fails or the API request fails. """ - if isinstance(e, (requests.Timeout, httpx.TimeoutException)): + if isinstance(e, (requests.Timeout, httpx2.TimeoutException)): msg = f"Request timed out for {method.upper()} {target_url}" raise MailgunTimeoutError(msg) from e if isinstance(e, ApiError): raise e - if isinstance(e, (requests.ConnectionError, httpx.ConnectError, httpx.NetworkError)): + if isinstance(e, (requests.ConnectionError, httpx2.ConnectError, httpx2.NetworkError)): msg = f"Network routing failed for {method.upper()} {target_url}: {e}" raise ApiError(msg) from e msg = f"API request failed for {method.upper()} {target_url}: {e}" @@ -343,7 +343,7 @@ def _merge_headers(self, kwargs: dict[str, Any]) -> dict[str, str]: req_headers.update(custom_headers) # CWE-400 / Crash Prevention: Enforce string keys and values to - # prevent HTTP protocol serialization crashes in requests/httpx. + # prevent HTTP protocol serialization crashes in requests/httpx2. return {str(k): str(v) for k, v in req_headers.items()} def _prepare_request( @@ -564,6 +564,18 @@ def api_call( # noqa: PLR0914, PLR0915 break except requests.RequestException as e: + # Client-side configuration and validation errors are unrecoverable; do not retry + if isinstance( + e, + ( + requests.exceptions.InvalidURL, + requests.exceptions.InvalidHeader, + requests.exceptions.InvalidSchema, + requests.exceptions.MissingSchema, + ), + ): + self._handle_api_error(e, safe_method, target_url) + if attempt < max_attempts - 1: delay = policy.calculate_delay(attempt) @@ -791,9 +803,13 @@ def stream( 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. """ initial_filters = dict(filters) if filters else {} current_filters = initial_filters.copy() + prev_url: str | None = None while True: # Pass a copy of the dictionary so the mock (and the underlying request layer) @@ -801,7 +817,10 @@ def stream( response = self.get(filters=current_filters.copy(), domain=domain, **kwargs) if hasattr(response, "raise_for_status"): - response.raise_for_status() + try: + response.raise_for_status() + except requests.HTTPError as exc: + raise ApiError(exc.response) from exc data = response.json() if not isinstance(data, dict): @@ -812,14 +831,10 @@ def stream( yield from items paging_dict = data.get("paging") or {} - # Check for the next page cursor 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: + if not next_url or not items or next_url == prev_url: break + prev_url = next_url # 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. @@ -841,7 +856,7 @@ def stream( class AsyncEndpoint(BaseEndpoint): - """Generate async requests and return responses using httpx.""" + """Generate async requests and return responses using httpx2.""" __slots__ = ("_client",) @@ -850,7 +865,7 @@ def __init__( url: dict[str, Any], headers: dict[str, str], auth: tuple[str, str] | None, - client: httpx.AsyncClient | None = None, + client: httpx2.AsyncClient | None = None, timeout: TimeoutType = 60, *, dry_run: bool = False, @@ -861,12 +876,12 @@ def __init__( url: URL dictionary with pairs {"base": "keys"}. headers: Headers dictionary. auth: httpx auth tuple or None. - client: Optional httpx.AsyncClient instance to reuse. + client: Optional httpx2.AsyncClient instance to reuse. timeout: Base request timeout. dry_run: Execution sandbox flag. """ super().__init__(url, headers, auth, timeout=timeout, dry_run=dry_run) - self._client = client or httpx.AsyncClient() + self._client = client or httpx2.AsyncClient() async def api_call( # noqa: PLR0912, PLR0914, PLR0915 self, @@ -913,15 +928,15 @@ async def api_call( # noqa: PLR0912, PLR0914, PLR0915 safe_method.upper(), safe_url_for_log, ) - mock_request = httpx.Request(safe_method.upper(), target_url) - return httpx.Response( + mock_request = httpx2.Request(safe_method.upper(), target_url) + return httpx2.Response( HTTPStatus.OK, request=mock_request, content=b'{"message": "Dry run successful - request intercepted", "id": ""}', ) if isinstance(safe_timeout, tuple) and len(safe_timeout) == 2: # noqa: PLR2004 - safe_timeout = httpx.Timeout(safe_timeout[1], connect=safe_timeout[0]) + safe_timeout = httpx2.Timeout(safe_timeout[1], connect=safe_timeout[0]) request_kwargs: dict[str, Any] = { "method": safe_method.upper(), @@ -998,7 +1013,11 @@ async def api_call( # noqa: PLR0912, PLR0914, PLR0915 break - except httpx.RequestError as e: + except httpx2.RequestError as e: + # Protocol and decoding violations are non-transient; do not retry + if isinstance(e, (httpx2.UnsupportedProtocol, httpx2.DecodingError)): + self._handle_api_error(e, safe_method, target_url) + if attempt < max_attempts - 1: delay = policy.calculate_delay(attempt) @@ -1017,14 +1036,14 @@ async def api_call( # noqa: PLR0912, PLR0914, PLR0915 continue - if isinstance(e, httpx.TimeoutException): + if isinstance(e, httpx2.TimeoutException): logger.exception( "Request timed out for %s %s", safe_method.upper(), safe_url_for_log, ) - elif isinstance(e, (httpx.ConnectError, httpx.NetworkError)): + elif isinstance(e, (httpx2.ConnectError, httpx2.NetworkError)): logger.critical( "Network routing failed for %s %s: %s", safe_method.upper(), @@ -1234,6 +1253,7 @@ async def stream( """ initial_filters = dict(filters) if filters else {} current_filters = initial_filters.copy() + prev_url: str | None = None while True: response = await self.get(filters=current_filters.copy(), domain=domain, **kwargs) @@ -1242,7 +1262,7 @@ async def stream( if hasattr(response, "raise_for_status"): try: response.raise_for_status() - except httpx.HTTPStatusError as exc: + except httpx2.HTTPStatusError as exc: raise ApiError(exc.response) from exc data = response.json() @@ -1256,8 +1276,9 @@ async def stream( paging_dict = data.get("paging") or {} next_url = paging_dict.get("next") - if not next_url or not items: + if not next_url or not items or next_url == prev_url: break + prev_url = next_url # 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. diff --git a/mailgun/security.py b/mailgun/security.py index e449757..d1e28de 100644 --- a/mailgun/security.py +++ b/mailgun/security.py @@ -272,7 +272,7 @@ def sanitize_timeout(cls, timeout: TimeoutType) -> TimeoutType: ) raise ValueError(msg) - # Extract values from httpx.Timeout object cleanly + # Extract values from httpx2.Timeout object cleanly if hasattr(timeout, "read") and hasattr(timeout, "connect"): timeout = (getattr(timeout, "connect", 60.0), getattr(timeout, "read", 60.0)) diff --git a/mailgun/types.py b/mailgun/types.py index f1e8dda..2191ec2 100644 --- a/mailgun/types.py +++ b/mailgun/types.py @@ -9,8 +9,8 @@ if TYPE_CHECKING: - from mailgun._httpx_compat import Response as HttpxResponse - from mailgun._httpx_compat import Timeout as HttpxTimeout + from httpx2 import Response as HttpxResponse + from httpx2 import Timeout as HttpxTimeout __all__ = [ "APIResponseType", diff --git a/pyproject.toml b/pyproject.toml index dabd1d3..115bfe5 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -44,8 +44,7 @@ classifiers = [ dynamic = [ "version" ] dependencies = [ - "httpx2 >=2.7.0", # The new primary async engine - "httpx >=0.24.0", # Kept as a fallback/safety net + "httpx2 >=2.7.0", "requests>=2.33.0", ] @@ -178,6 +177,7 @@ docstring-code-format = true select = [ "ALL" ] ignore = [ # --- Formatter Conflicts --- + "missing-trailing-comma", # COM812: Recommended by Ruff when using ruff format "single-line-implicit-string-concatenation", # Implicit string concatenation (Conflicts with ruff-format) "incorrect-blank-line-before-class", # one-blank-line-before-class (conflicts with D211) "multi-line-summary-second-line", # multi-line-summary-second-line (conflicts with D212) @@ -189,7 +189,9 @@ ignore = [ "line-too-long", # Line too long (Let the formatter handle wrapping) # --- Docstring --- - "undocumented-param", "undocumented-public-module", "undocumented-public-package", + "undocumented-param", + "undocumented-public-module", + "undocumented-public-package", # --- Keep existing TODO ignores --- "complex-structure", @@ -204,10 +206,108 @@ dummy-variable-rgx = "^(_+|(_+[a-zA-Z0-9_]*[a-zA-Z0-9]+?))$" external = [ "DOC", "PLR" ] preview = true +# Test suite rules relaxation [tool.ruff.lint.per-file-ignores] -"__init__.py" = [ "module-import-not-at-top-of-file" ] +"__init__.py" = ["module-import-not-at-top-of-file"] "mailgun/endpoints.py" = ["import-outside-top-level"] +# Test suite relaxations +"tests/**" = [ + # Core test assertions & naming + "assert", + "magic-value-comparison", + "invalid-function-name", + "non-lowercase-variable-in-function", + "shebang-not-executable", + "commented-out-code", + "implicit-namespace-package", + "class-as-data-structure", + "collections-named-tuple", + + # Docstring relaxations in tests + "missing-blank-line-after-summary", + "undocumented-public-module", + "undocumented-public-package", + "undocumented-public-class", + "undocumented-public-method", + "undocumented-public-function", + "undocumented-public-init", + "undocumented-public-nested-class", + "undocumented-magic-method", + "docstring-missing-returns", + "docstring-missing-yields", + "docstring-missing-exception", + "missing-terminal-punctuation", + + # Test execution & white-box state inspections + "private-member-access", + "import-private-name", + "import-outside-top-level", + "set-attr-with-constant", + "builtin-argument-shadowing", + "unused-function-argument", + "unused-method-argument", + "unused-lambda-argument", + "unused-loop-control-variable", + "unused-unpacked-variable", + "unused-async", + "manual-list-comprehension", + "if-exp-instead-of-or-operator", + "no-self-use", + + # String & float equality test assertions + "compare-to-empty-string", + "float-equality-comparison", + "boolean-positional-value-in-call", + "boolean-type-hint-positional-argument", + "boolean-default-value-positional-argument", + + # Pytest & Unittest specifics + "pytest-unittest-assertion", + "pytest-unittest-raises-assertion", + "pytest-fixture-param-without-value", + "pytest-raises-too-broad", + "pytest-warns-too-broad", + "pytest-warns-with-multiple-statements", + "pytest-raises-with-multiple-statements", + + # Security & Logging test vectors + "hardcoded-password-string", + "hardcoded-password-func-arg", + "ambiguous-unicode-character-string", + "static-join-to-f-string", + "suspicious-subprocess-import", + "subprocess-without-shell-equals-true", + "start-process-with-partial-path", + "root-logger-call", + "logging-extra-attr-clash", + + # Complexity & Flow in fuzzing harnesses + "too-many-branches", + "too-many-statements", + "too-many-locals", + "too-many-nested-blocks", + "too-many-public-methods", + "f-string-in-exception", + "raise-within-try", + "type-check-without-type-error", + "raise-without-from-inside-except", + "error-suffix-on-exception-name", + "print", + "blind-except", + "try-except-pass", + "suppressible-exception", + "blocking-open-call-in-async-function", + "blocking-sleep-in-async-function", + "unspecified-encoding", + "call-datetime-now-without-tzinfo", + "typing-only-first-party-import", + "typing-only-standard-library-import", + "missing-todo-author", + "missing-todo-link", + "line-contains-todo", +] + [tool.ruff.lint.pylint] max-args = 9 # Default is 5. SDK endpoints need more (auth, data, headers, files, etc.) max-returns = 9 # Default is 6. diff --git a/tests/conftest.py b/tests/conftest.py index e0b4761..905af35 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -35,14 +35,16 @@ def pytest_addoption(parser: pytest.Parser) -> None: def pytest_collection_modifyitems( - config: pytest.Config, items: list[pytest.Item] + config: pytest.Config, + items: list[pytest.Item], ) -> None: if config.getoption("--run-skipped"): for item in items: for marker in list(item.iter_markers()): - if marker.name in ("skip", "skipif"): + if marker.name in {"skip", "skipif"}: item.own_markers.remove(marker) + @pytest.fixture(autouse=True) def bypass_retry_delays(monkeypatch: pytest.MonkeyPatch) -> None: """Automatically speeds up test execution by blocking the actual delays diff --git a/tests/fuzz/fuzz.dict b/tests/fuzz/fuzz.dict index f458c58..ba3afa0 100644 --- a/tests/fuzz/fuzz.dict +++ b/tests/fuzz/fuzz.dict @@ -1,5 +1,6 @@ " \xea\x98\x80\xe2\x88\x80\xed\x94\x80\x00\xef\x97\xbd\xe2\x92\x96\xc7\x9a\xe5\xb4\xb1" "!%D0Dify=9%3MExpecevent_typest: " +"!%D5G%D\x7f\x7f\x7f\x7f\x01\x00" "!>" "#/" "#{7*7}" @@ -178,6 +179,7 @@ "5%seMr{\xef\xbf\xbd%for0" "6[" "6\x00\x00\x00\x00\x00\x00\x00" +"7.\x02\x02{\"%80" "77" "8:" "9.999999999999999e95" @@ -198,6 +200,7 @@ "; wget http://attacker.com" ";/" ";id;" +"---" "q!-q" "" +"ccccc" +"" "\x15q2&\x15\x15<%do" "<" "<\x1b" "" "<" "=\x18\x00\x00\x00\x00" +"E" "}&l\x7f\x7f\x7f" "}><" "}<" "]>" "}!t" "E" -"im1" -"ccccc" -"---" -"\xff\xff\xff\xff\xff\xff\xff\xf0" -"" -"<" -" compat_httpx.Response: + async def handle_async_request(self, request: httpx2.Request) -> httpx2.Response: return self._static_resp -original_init = compat_httpx.AsyncClient.__init__ +original_init = httpx2.AsyncClient.__init__ -def secure_init(self: compat_httpx.AsyncClient, *args: Any, **kwargs: Any) -> None: +def secure_init(self: httpx2.AsyncClient, *args: Any, **kwargs: Any) -> None: kwargs["transport"] = MockAsyncTransport() original_init(self, *args, **kwargs) -compat_httpx.AsyncClient.__init__ = secure_init # type: ignore[method-assign] +httpx2.AsyncClient.__init__ = secure_init # type: ignore[method-assign] async def _worker_task(client: AsyncClient, fdp: atheris.FuzzedDataProvider) -> None: @@ -64,7 +64,7 @@ async def _worker_task(client: AsyncClient, fdp: atheris.FuzzedDataProvider) -> RuntimeError, TypeError, ValueError, - compat_httpx.RequestError, + httpx2.RequestError, ): # Fuzzing intentionally suppresses expected malformed-input/runtime errors # so one failing path does not stop concurrent exploration. @@ -96,7 +96,7 @@ async def _async_fuzz_target(data: bytes) -> None: RuntimeError, TypeError, ValueError, - compat_httpx.RequestError, + httpx2.RequestError, ): pass @@ -110,5 +110,5 @@ def TestOneInput(data: bytes) -> None: if __name__ == "__main__": atheris.instrument_all() atheris.Setup(sys.argv, TestOneInput) - atexit.register(lambda: _FUZZ_LOOP.close()) + atexit.register(_FUZZ_LOOP.close) atheris.Fuzz() diff --git a/tests/fuzz/fuzz_async_evil_server.py b/tests/fuzz/fuzz_async_evil_server.py index fd53783..9b6f1a2 100755 --- a/tests/fuzz/fuzz_async_evil_server.py +++ b/tests/fuzz/fuzz_async_evil_server.py @@ -14,8 +14,10 @@ import atheris + with atheris.instrument_imports(): - from mailgun._httpx_compat import httpx as compat_httpx + import httpx2 + from mailgun.client import AsyncClient, Config from mailgun.config import RetryPolicy from mailgun.handlers.error_handler import ApiError, MailgunTimeoutError @@ -55,18 +57,20 @@ def TestOneInput(data: bytes) -> None: fdp = atheris.FuzzedDataProvider(data) - original_send = compat_httpx.AsyncClient.send + original_send = httpx2.AsyncClient.send async def evil_send( - self: compat_httpx.AsyncClient, request: compat_httpx.Request, **kwargs: Any - ) -> compat_httpx.Response: + self: httpx2.AsyncClient, + request: httpx2.Request, + **kwargs: Any, + ) -> httpx2.Response: if fdp.ConsumeBool(): exceptions = [ - compat_httpx.ConnectError("Fuzzed Connection Drop"), - compat_httpx.NetworkError("Fuzzed Network Error"), - compat_httpx.ProtocolError("Fuzzed Protocol Error"), - compat_httpx.ReadTimeout("Fuzzed Timeout"), - compat_httpx.TooManyRedirects("Infinite Redirect Loop"), + httpx2.ConnectError("Fuzzed Connection Drop"), + httpx2.NetworkError("Fuzzed Network Error"), + httpx2.ProtocolError("Fuzzed Protocol Error"), + httpx2.ReadTimeout("Fuzzed Timeout"), + httpx2.TooManyRedirects("Infinite Redirect Loop"), ] raise fdp.PickValueInList(exceptions) @@ -80,26 +84,30 @@ async def evil_send( headers = { b"content-type": fdp.PickValueInList( - [b"application/json", b"image/png", b"text/html", b"application/octet-stream"] + [b"application/json", b"image/png", b"text/html", b"application/octet-stream"], ), b"content-length": str(fdp.ConsumeIntInRange(-100, 10000)).encode(), b"Retry-After": retry_val.encode(errors="ignore"), } garbage_bytes = fdp.ConsumeBytes(512) - return compat_httpx.Response( + return httpx2.Response( status_code=status, headers=headers, content=garbage_bytes, request=request, ) - compat_httpx.AsyncClient.send = evil_send # type: ignore[method-assign] + httpx2.AsyncClient.send = evil_send # type: ignore[method-assign] async def run_fuzz() -> None: - with Path(os.devnull).open("w") as devnull, contextlib.redirect_stdout( - devnull - ), contextlib.redirect_stderr(devnull): + with ( + Path(os.devnull).open("w") as devnull, + contextlib.redirect_stdout( + devnull, + ), + contextlib.redirect_stderr(devnull), + ): try: action_choice = fdp.ConsumeIntInRange(0, 2) if action_choice == 0: @@ -117,20 +125,19 @@ async def run_fuzz() -> None: MailgunTimeoutError, TypeError, ValueError, - compat_httpx.RequestError, + httpx2.RequestError, json.JSONDecodeError, TimeoutError, - asyncio.TimeoutError, ): pass except OverflowError as oe: raise RuntimeError(f"CRASH: Retry-After exponential overflow: {oe}") from oe except Exception as e: raise RuntimeError( - f"SDK crashed handling Async Evil Server response: {type(e).__name__} - {e}" + f"SDK crashed handling Async Evil Server response: {type(e).__name__} - {e}", ) from e finally: - compat_httpx.AsyncClient.send = original_send # type: ignore[method-assign] + httpx2.AsyncClient.send = original_send # type: ignore[method-assign] # Global patch prevents any residual retry or backoff sleeps from blocking the runner with patch("time.sleep", return_value=None), patch("asyncio.sleep", return_value=None): diff --git a/tests/fuzz/fuzz_audit_events.py b/tests/fuzz/fuzz_audit_events.py index fa3733a..cf8da97 100644 --- a/tests/fuzz/fuzz_audit_events.py +++ b/tests/fuzz/fuzz_audit_events.py @@ -3,17 +3,26 @@ import logging import sys +import time from typing import Any import atheris import requests + with atheris.instrument_imports(): from mailgun.client import Client + from mailgun.config import RetryPolicy from mailgun.handlers.error_handler import ApiError + from mailgun.security import SecurityGuard logging.disable(logging.CRITICAL) +# Eliminate all sleep latency during fuzzing +time.sleep = lambda *args, **kwargs: None + +_NO_RETRY_CONFIG = RetryPolicy(max_retries=0, base_delay=0.0, max_delay=0.0) + _AUDIT_LOG: list[tuple[str, tuple[Any, ...]]] = [] @@ -57,7 +66,7 @@ def TestOneInput(data: bytes) -> None: fuzzed_domain = fdp.ConsumeUnicodeNoSurrogates(64) fuzzed_method = fdp.PickValueInList(["get", "post", "delete", "put"]) - client = Client(auth=("api", "test-key")) + client = Client(auth=("api", "test-key"), retry_policy=_NO_RETRY_CONFIG) try: # Route through actual SDK endpoint execution @@ -65,13 +74,12 @@ def TestOneInput(data: bytes) -> None: action = getattr(client.domains, fuzzed_method) action(domain=fuzzed_domain) - # Probe header injection & control character audit hooks if fdp.ConsumeBool(): hostile_headers = { fdp.ConsumeUnicodeNoSurrogates(16): fdp.PickValueInList( - ["valid", "bad\r\nHeader: 1", "control\x01char", "null\x00byte"] - ) + ["valid", "bad\r\nHeader: 1", "control\x01char", "null\x00byte"], + ), } client.messages.create( domain=fuzzed_domain, @@ -82,16 +90,20 @@ def TestOneInput(data: bytes) -> None: # Probe SSRF URL audit hooks if fdp.ConsumeBool(): hostile_url = fdp.PickValueInList( - ["ftp://api.mailgun.net", "gopher://127.0.0.1", "http://attacker.com/v3"] + ["ftp://api.mailgun.net", "gopher://127.0.0.1", "http://attacker.com/v3"], ) - client.messages.api_call(method="get", url=hostile_url) + try: + SecurityGuard.validate_mailgun_url(hostile_url) + except ValueError: + # Expected for hostile fuzzed URLs; rejection is a valid outcome. + pass # Verify invariant: if audit hook fired, arguments must be safe for event, args in _AUDIT_LOG: for arg in args: if isinstance(arg, str) and "\x00" in arg: raise RuntimeError( - f"CRITICAL: Embedded null byte leaked into sys.audit hook '{event}': {arg!r}" + f"CRITICAL: Embedded null byte leaked into sys.audit hook '{event}': {arg!r}", ) except (ApiError, TypeError, ValueError): @@ -100,7 +112,7 @@ def TestOneInput(data: bytes) -> None: except Exception as e: if "embedded null" in str(e).lower(): raise RuntimeError( - f"CRASH: Unsanitized null byte reached runtime boundary: {e}" + f"CRASH: Unsanitized null byte reached runtime boundary: {e}", ) from e raise RuntimeError(f"UNHANDLED CRASH in Audit Events execution: {e}") from e diff --git a/tests/fuzz/fuzz_builders.py b/tests/fuzz/fuzz_builders.py index bc8aed9..cae7882 100644 --- a/tests/fuzz/fuzz_builders.py +++ b/tests/fuzz/fuzz_builders.py @@ -7,6 +7,7 @@ import atheris + with atheris.instrument_imports(): from mailgun.builders import MailgunMessageBuilder, MailgunTemplateBuilder @@ -57,7 +58,9 @@ def TestOneInput(data: bytes) -> None: elif op_code == 2: builder.add_option( fdp.ConsumeUnicodeNoSurrogates(20), - value=fdp.PickValueInList([True, False, fdp.ConsumeUnicodeNoSurrogates(10)]), + value=fdp.PickValueInList( + [True, False, fdp.ConsumeUnicodeNoSurrogates(10)], + ), ) elif op_code == 3: builder.add_recipient( @@ -82,13 +85,9 @@ def TestOneInput(data: bytes) -> None: else: # Target 2: MailgunTemplateBuilder - template_name = ( - fdp.ConsumeUnicodeNoSurrogates(30) if fdp.ConsumeBool() else None - ) + template_name = fdp.ConsumeUnicodeNoSurrogates(30) if fdp.ConsumeBool() else None t_builder = ( - MailgunTemplateBuilder(template_name) - if template_name - else MailgunTemplateBuilder() + MailgunTemplateBuilder(template_name) if template_name else MailgunTemplateBuilder() ) for _ in range(fdp.ConsumeIntInRange(1, 8)): @@ -99,7 +98,7 @@ def TestOneInput(data: bytes) -> None: t_builder.set_description(fdp.ConsumeUnicodeNoSurrogates(50)) elif op == 2: t_builder.set_engine( - fdp.PickValueInList(["handlebars", "jinja2", "{{7*7}}", "none"]) + fdp.PickValueInList(["handlebars", "jinja2", "{{7*7}}", "none"]), ) elif op == 3: t_builder.set_template_content(fdp.ConsumeUnicodeNoSurrogates(200)) @@ -108,7 +107,6 @@ def TestOneInput(data: bytes) -> None: _ = t_builder.build() - except (ValueError, AttributeError) as e: error_msg = str(e) if isinstance(e, ValueError) and not any(msg in error_msg for msg in _ALLOWED_ERRORS): diff --git a/tests/fuzz/fuzz_builders_advanced.py b/tests/fuzz/fuzz_builders_advanced.py index f77bbbb..b9815ad 100644 --- a/tests/fuzz/fuzz_builders_advanced.py +++ b/tests/fuzz/fuzz_builders_advanced.py @@ -21,7 +21,8 @@ def _test_cyclic_and_stream_pointer_invariants( - fdp: atheris.FuzzedDataProvider, builder: MailgunMessageBuilder + fdp: atheris.FuzzedDataProvider, + builder: MailgunMessageBuilder, ) -> None: # Explicitly annotate cyclic_dict as dict[str, Any] to allow self-referential structures cyclic_dict: dict[str, Any] = {"domain": "test.com", "to": ["a@b.com"]} @@ -105,13 +106,12 @@ def TestOneInput(data: bytes) -> None: nested_ast = _generate_nested_ast(fdp) builder.add_custom_variable(var_key, nested_ast) - elif op_code == 5: # Attach in-memory BytesIO directly raw_bytes = io.BytesIO(fdp.ConsumeBytes(128)) try: getattr(builder, "add_attachment", lambda *a, **k: None)( - ("payload.bin", raw_bytes, "application/octet-stream") + ("payload.bin", raw_bytes, "application/octet-stream"), ) except (AttributeError, TypeError, ValueError): # Expected during fuzzing: optional API may be missing or reject malformed input. diff --git a/tests/fuzz/fuzz_client.py b/tests/fuzz/fuzz_client.py index f0bde30..bde149b 100755 --- a/tests/fuzz/fuzz_client.py +++ b/tests/fuzz/fuzz_client.py @@ -9,6 +9,7 @@ import atheris import requests + with atheris.instrument_imports(): from mailgun import routes from mailgun.client import Client @@ -33,12 +34,12 @@ def _generate_chaotic_file_payload( ".env", "payload.exe\x00.jpg", "＼．．／＼．．／.txt", - ] + ], ) ) content = fdp.ConsumeBytes(64) mime_type = fdp.PickValueInList( - ["application/json", "text/plain", "image/png", fdp.ConsumeUnicodeNoSurrogates(16)] + ["application/json", "text/plain", "image/png", fdp.ConsumeUnicodeNoSurrogates(16)], ) files.append(("attachment", (filename, content, mime_type))) return files diff --git a/tests/fuzz/fuzz_config_router.py b/tests/fuzz/fuzz_config_router.py index e81dfda..74003a3 100755 --- a/tests/fuzz/fuzz_config_router.py +++ b/tests/fuzz/fuzz_config_router.py @@ -6,6 +6,7 @@ import atheris + with atheris.instrument_imports(): from mailgun import routes from mailgun.config import Config @@ -33,7 +34,7 @@ def TestOneInput(data: bytes) -> None: "https://user:pass@api.mailgun.net", # pragma: allowlist secret "ftp://api.mailgun.net", "/v3/relative/path", - ] + ], ) ) diff --git a/tests/fuzz/fuzz_crypto_idna.py b/tests/fuzz/fuzz_crypto_idna.py index 7c8ef39..7d03726 100644 --- a/tests/fuzz/fuzz_crypto_idna.py +++ b/tests/fuzz/fuzz_crypto_idna.py @@ -7,6 +7,7 @@ import atheris + with atheris.instrument_imports(): from mailgun.security import SecretAuth, SecurityGuard @@ -75,7 +76,9 @@ def TestOneInput(data: bytes) -> None: except ValueError: pass except Exception as e: - raise RuntimeError(f"UNHANDLED CRASH in Crypto/IDNA boundaries: {type(e).__name__} - {e}") from e + raise RuntimeError( + f"UNHANDLED CRASH in Crypto/IDNA boundaries: {type(e).__name__} - {e}", + ) from e if __name__ == "__main__": diff --git a/tests/fuzz/fuzz_diff_serialization.py b/tests/fuzz/fuzz_diff_serialization.py index 33cc8d3..38dfe52 100644 --- a/tests/fuzz/fuzz_diff_serialization.py +++ b/tests/fuzz/fuzz_diff_serialization.py @@ -7,11 +7,11 @@ import atheris + with atheris.instrument_imports(): + import httpx2 import requests - from mailgun._httpx_compat import httpx as compat_httpx - logging.disable(logging.CRITICAL) @@ -48,7 +48,7 @@ def TestOneInput(data: bytes) -> None: payload[key] = fdp.ConsumeBool() sync_req = requests.Request("POST", "https://api.mailgun.net/v3/fuzz", data=payload) - async_req = compat_httpx.Request("POST", "https://api.mailgun.net/v3/fuzz", data=payload) + async_req = httpx2.Request("POST", "https://api.mailgun.net/v3/fuzz", data=payload) sync_exc: Exception | None = None async_exc: Exception | None = None @@ -74,7 +74,7 @@ def TestOneInput(data: bytes) -> None: raise RuntimeError( f"DIFFERENTIAL DIVERGENCE: Sync resolved to '{sync_category}' ({type(sync_exc).__name__}), " f"but Async resolved to '{async_category}' ({type(async_exc).__name__}). " - f"Payload: {payload!r}" + f"Payload: {payload!r}", ) diff --git a/tests/fuzz/fuzz_differential.py b/tests/fuzz/fuzz_differential.py index 152378e..83e3077 100644 --- a/tests/fuzz/fuzz_differential.py +++ b/tests/fuzz/fuzz_differential.py @@ -8,10 +8,9 @@ from unittest.mock import patch import atheris +import httpx2 import requests -from mailgun._httpx_compat import httpx as compat_httpx - with atheris.instrument_imports(): from mailgun import routes @@ -37,7 +36,9 @@ _ASYNC_CLIENT = AsyncClient(auth=("api", "key-test"), config=_NO_RETRY_CONFIG) -def _build_fuzzed_response_data(fdp: atheris.FuzzedDataProvider) -> tuple[int, dict[str, str], bytes]: +def _build_fuzzed_response_data( + fdp: atheris.FuzzedDataProvider, +) -> tuple[int, dict[str, str], bytes]: """Generate dynamic status codes, headers, and payloads for mock responses.""" status_code = fdp.PickValueInList(_STATUS_CODES) content_types = [ @@ -62,7 +63,7 @@ def _build_fuzzed_response_data(fdp: atheris.FuzzedDataProvider) -> tuple[int, d if choice == 0: msg = fdp.ConsumeUnicodeNoSurrogates(32) total = fdp.ConsumeInt(100) - body = f'{{"message": "{msg}", "total": {total}}}'.encode("utf-8") + body = f'{{"message": "{msg}", "total": {total}}}'.encode() elif choice == 1: body = b"{}" if fdp.ConsumeBool() else b"" else: @@ -99,21 +100,21 @@ def mock_requests_send( # Dynamic mock for Async (httpx) byte_headers = { - k.encode("latin-1"): v.encode("latin-1", "replace") - for k, v in resp_headers.items() + k.encode("latin-1"): v.encode("latin-1", "replace") for k, v in resp_headers.items() } async def mock_httpx_handle( - self: compat_httpx.AsyncBaseTransport, request: compat_httpx.Request - ) -> compat_httpx.Response: - return compat_httpx.Response( + self: httpx2.AsyncBaseTransport, + request: httpx2.Request, + ) -> httpx2.Response: + return httpx2.Response( status_code=status_code, headers=byte_headers, content=resp_body, request=request, ) - compat_httpx.AsyncHTTPTransport.handle_async_request = mock_httpx_handle # type: ignore[method-assign] + httpx2.AsyncHTTPTransport.handle_async_request = mock_httpx_handle # type: ignore[method-assign] target_attr = fdp.PickValueInList(_VALID_ENDPOINTS) method_name = fdp.PickValueInList(_HTTP_METHODS) @@ -133,7 +134,7 @@ async def mock_httpx_handle( call_kwargs["domain"] = fdp.ConsumeUnicodeNoSurrogates(24) if fdp.ConsumeBool(): call_kwargs["params"] = { - fdp.ConsumeUnicodeNoSurrogates(8): fdp.ConsumeUnicodeNoSurrogates(16) + fdp.ConsumeUnicodeNoSurrogates(8): fdp.ConsumeUnicodeNoSurrogates(16), } if fdp.ConsumeBool(): call_kwargs["data"] = { @@ -159,7 +160,7 @@ async def mock_httpx_handle( ares = _FUZZ_LOOP.run_until_complete(coro) if hasattr(ares, "status_code"): async_result = f"status_{ares.status_code}" - except (TimeoutError, asyncio.TimeoutError): + except TimeoutError: async_result = "TimeoutError" except Exception as exc: async_result = type(exc).__name__ @@ -170,7 +171,7 @@ async def mock_httpx_handle( f"Call kwargs: {call_kwargs!r}\n" f"Mock Status: {status_code}\n" f"Sync Result: {sync_result}\n" - f"Async Result: {async_result}" + f"Async Result: {async_result}", ) diff --git a/tests/fuzz/fuzz_endpoint_headers.py b/tests/fuzz/fuzz_endpoint_headers.py index 20fc360..26adf63 100755 --- a/tests/fuzz/fuzz_endpoint_headers.py +++ b/tests/fuzz/fuzz_endpoint_headers.py @@ -54,7 +54,7 @@ def TestOneInput(data: bytes) -> None: "X-Mailgun-Tag", "X-Mailgun-Variables", fdp.ConsumeUnicodeNoSurrogates(12), - ] + ], ) base_headers[header_key] = _generate_header_value(fdp) @@ -81,7 +81,7 @@ def TestOneInput(data: bytes) -> None: "Content-Type", "x-mailgun-tag", fdp.ConsumeUnicodeNoSurrogates(16), - ] + ], ) merged_headers[header_key] = _generate_header_value(fdp) kwargs["headers"] = merged_headers @@ -89,7 +89,14 @@ def TestOneInput(data: bytes) -> None: # HTTP transport kwargs injection for _ in range(fdp.ConsumeIntInRange(0, 5)): prop_key = fdp.PickValueInList( - ["timeout", "verify", "proxies", "params", "allow_redirects", fdp.ConsumeUnicodeNoSurrogates(10)] + [ + "timeout", + "verify", + "proxies", + "params", + "allow_redirects", + fdp.ConsumeUnicodeNoSurrogates(10), + ], ) prop_val: Any val_mode = fdp.ConsumeIntInRange(0, 3) @@ -115,7 +122,7 @@ def TestOneInput(data: bytes) -> None: f"Unhandled exception during _merge_headers:\n" f"Base headers: {base_headers!r}\n" f"Kwargs: {kwargs!r}\n" - f"Exception: {type(exc).__name__}: {exc}" + f"Exception: {type(exc).__name__}: {exc}", ) from exc diff --git a/tests/fuzz/fuzz_endpoint_lifecycle.py b/tests/fuzz/fuzz_endpoint_lifecycle.py index 0d239f1..95c8362 100755 --- a/tests/fuzz/fuzz_endpoint_lifecycle.py +++ b/tests/fuzz/fuzz_endpoint_lifecycle.py @@ -44,7 +44,7 @@ def _generate_fuzzed_id(fdp: atheris.FuzzedDataProvider) -> str: if choice == 2: # URL encoding and special characters return fdp.PickValueInList( - ["test@example.com", "tag:special", "id#123", "id?filter=1", "id%20space", ""] + ["test@example.com", "tag:special", "id#123", "id?filter=1", "id%20space", ""], ) if choice == 3: # Sub-resource path representation @@ -82,12 +82,18 @@ def TestOneInput(data: bytes) -> None: # CREATE new_id = _generate_fuzzed_id(fdp) if hasattr(endpoint, "create"): - endpoint.create(data={"id": new_id, "name": fdp.ConsumeUnicodeNoSurrogates(10)}) + endpoint.create( + data={"id": new_id, "name": fdp.ConsumeUnicodeNoSurrogates(10)}, + ) active_ids.append(new_id) elif op == 1: # GET - target_id = fdp.PickValueInList(active_ids) if (active_ids and fdp.ConsumeBool()) else _generate_fuzzed_id(fdp) + target_id = ( + fdp.PickValueInList(active_ids) + if (active_ids and fdp.ConsumeBool()) + else _generate_fuzzed_id(fdp) + ) if hasattr(endpoint, "get"): endpoint.get(domain=target_id) @@ -104,7 +110,10 @@ def TestOneInput(data: bytes) -> None: if hasattr(endpoint, "update"): endpoint.update( domain=target_id, - data={"fuzz_key": fdp.ConsumeUnicodeNoSurrogates(12), "val": fdp.ConsumeInt(1000)}, + data={ + "fuzz_key": fdp.ConsumeUnicodeNoSurrogates(12), + "val": fdp.ConsumeInt(1000), + }, ) elif op == 4 and active_ids: @@ -125,7 +134,9 @@ def TestOneInput(data: bytes) -> None: ): pass except Exception as exc: - raise RuntimeError(f"Unexpected crash in endpoint lifecycle harness: {type(exc).__name__}: {exc}") from exc + raise RuntimeError( + f"Unexpected crash in endpoint lifecycle harness: {type(exc).__name__}: {exc}", + ) from exc if __name__ == "__main__": diff --git a/tests/fuzz/fuzz_error_parser.py b/tests/fuzz/fuzz_error_parser.py index 7674de5..c76731a 100644 --- a/tests/fuzz/fuzz_error_parser.py +++ b/tests/fuzz/fuzz_error_parser.py @@ -6,10 +6,9 @@ import sys import atheris +import httpx2 import requests -from mailgun._httpx_compat import httpx as compat_httpx - with atheris.instrument_imports(): from mailgun.handlers.error_handler import ApiError, DeliverabilityError, MailgunTimeoutError @@ -25,24 +24,32 @@ def _build_fuzzed_error_content(fdp: atheris.FuzzedDataProvider) -> bytes: if payload_type == 0: # Standard Mailgun error structure - return json.dumps({ - "message": fdp.ConsumeUnicodeNoSurrogates(64), - "details": [fdp.ConsumeUnicodeNoSurrogates(32) for _ in range(fdp.ConsumeIntInRange(0, 3))], - }).encode("utf-8") + return json.dumps( + { + "message": fdp.ConsumeUnicodeNoSurrogates(64), + "details": [ + fdp.ConsumeUnicodeNoSurrogates(32) for _ in range(fdp.ConsumeIntInRange(0, 3)) + ], + }, + ).encode("utf-8") if payload_type == 1: # Single error key or legacy dictionary format - return json.dumps({ - "error": fdp.ConsumeUnicodeNoSurrogates(64), - "code": fdp.ConsumeInt(1000), - }).encode("utf-8") + return json.dumps( + { + "error": fdp.ConsumeUnicodeNoSurrogates(64), + "code": fdp.ConsumeInt(1000), + }, + ).encode("utf-8") if payload_type == 2: # Array-wrapped errors - return json.dumps([ - {"message": fdp.ConsumeUnicodeNoSurrogates(32)}, - {"reason": fdp.ConsumeUnicodeNoSurrogates(32)}, - ]).encode("utf-8") + return json.dumps( + [ + {"message": fdp.ConsumeUnicodeNoSurrogates(32)}, + {"reason": fdp.ConsumeUnicodeNoSurrogates(32)}, + ], + ).encode("utf-8") if payload_type == 3: # Non-JSON HTML / XML error page (Cloudflare/Nginx gateway crashes) @@ -67,8 +74,7 @@ def TestOneInput(data: bytes) -> None: try: score = fdp.ConsumeFloat() issues = [ - fdp.ConsumeUnicodeNoSurrogates(24) - for _ in range(fdp.ConsumeIntInRange(0, 20)) + fdp.ConsumeUnicodeNoSurrogates(24) for _ in range(fdp.ConsumeIntInRange(0, 20)) ] err = DeliverabilityError(score=score, issues=issues) _ = str(err) @@ -96,7 +102,7 @@ def TestOneInput(data: bytes) -> None: "text/html", "text/plain", fdp.ConsumeUnicodeNoSurrogates(16), - ] + ], ) # Ensure header values conform to valid ASCII per HTTP header specifications content_type = raw_content_type.encode("ascii", "replace").decode("ascii") @@ -106,12 +112,12 @@ def TestOneInput(data: bytes) -> None: if status_code == 429 and fdp.ConsumeBool(): httpx_headers["retry-after"] = str(fdp.ConsumeIntInRange(0, 300)) - httpx_resp = compat_httpx.Response( + httpx_resp = httpx2.Response( status_code=status_code, headers=httpx_headers, content=content, - request=compat_httpx.Request("POST", "https://api.mailgun.net/v3/messages"), - ) + request=httpx2.Request("POST", "https://api.mailgun.net/v3/messages"), + ) try: api_error_httpx = ApiError(httpx_resp) diff --git a/tests/fuzz/fuzz_evil_server.py b/tests/fuzz/fuzz_evil_server.py index c2078bd..3d6f07f 100755 --- a/tests/fuzz/fuzz_evil_server.py +++ b/tests/fuzz/fuzz_evil_server.py @@ -31,7 +31,11 @@ def TestOneInput(data: bytes) -> None: client = Client(auth=("api", "test-key")) original_send = requests.Session.send - def evil_send(self: requests.Session, request: requests.PreparedRequest, **kwargs: Any) -> requests.Response: + def evil_send( + self: requests.Session, + request: requests.PreparedRequest, + **kwargs: Any, + ) -> requests.Response: # Inject network-level transport faults if fdp.ConsumeBool(): exceptions = [ @@ -47,7 +51,7 @@ def evil_send(self: requests.Session, request: requests.PreparedRequest, **kwarg status = fdp.PickValueInList(_STATUS_CODES) headers: dict[str, str] = { "content-type": fdp.PickValueInList( - ["application/json", "text/html", "application/octet-stream", "image/png"] + ["application/json", "text/html", "application/octet-stream", "image/png"], ), "content-length": str(fdp.ConsumeIntInRange(-1000, 20000)), } @@ -55,15 +59,13 @@ def evil_send(self: requests.Session, request: requests.PreparedRequest, **kwarg # Hostile Retry-After injection if fdp.ConsumeBool(): headers["Retry-After"] = ( - fdp.ConsumeUnicodeNoSurrogates(16) - if fdp.ConsumeBool() - else str(fdp.ConsumeFloat()) + fdp.ConsumeUnicodeNoSurrogates(16) if fdp.ConsumeBool() else str(fdp.ConsumeFloat()) ) # Redirect loop simulation - if status in (301, 302): + if status in {301, 302}: headers["Location"] = fdp.PickValueInList( - ["https://api.mailgun.net/v3/messages", "http://127.0.0.1:80", "/relative/loop"] + ["https://api.mailgun.net/v3/messages", "http://127.0.0.1:80", "/relative/loop"], ) mock_response = requests.Response() @@ -77,9 +79,13 @@ def evil_send(self: requests.Session, request: requests.PreparedRequest, **kwarg requests.Session.send = evil_send # type: ignore[method-assign] - with Path(os.devnull).open("w") as devnull, contextlib.redirect_stdout( - devnull - ), contextlib.redirect_stderr(devnull): + with ( + Path(os.devnull).open("w") as devnull, + contextlib.redirect_stdout( + devnull, + ), + contextlib.redirect_stderr(devnull), + ): try: target_method = fdp.PickValueInList(_HTTP_METHODS) client.messages.api_call( @@ -98,7 +104,9 @@ def evil_send(self: requests.Session, request: requests.PreparedRequest, **kwarg ): pass except Exception as exc: - raise RuntimeError(f"Unhandled crash during evil server simulation: {type(exc).__name__}: {exc}") from exc + raise RuntimeError( + f"Unhandled crash during evil server simulation: {type(exc).__name__}: {exc}", + ) from exc finally: requests.Session.send = original_send # type: ignore[method-assign] if hasattr(client, "close"): diff --git a/tests/fuzz/fuzz_handlers.py b/tests/fuzz/fuzz_handlers.py index 7533844..8573deb 100755 --- a/tests/fuzz/fuzz_handlers.py +++ b/tests/fuzz/fuzz_handlers.py @@ -13,6 +13,7 @@ import atheris + with atheris.instrument_imports(): from mailgun.handlers.bounce_classification_handler import ( handle_bounce_classification, @@ -148,10 +149,7 @@ def _generate_chaotic_value(fdp: atheris.FuzzedDataProvider, depth: int = 0) -> if choice == 4: return None if choice == 5: - return [ - _generate_chaotic_value(fdp, depth + 1) - for _ in range(fdp.ConsumeIntInRange(0, 4)) - ] + return [_generate_chaotic_value(fdp, depth + 1) for _ in range(fdp.ConsumeIntInRange(0, 4))] if choice == 6: return { fdp.ConsumeUnicodeNoSurrogates(12): _generate_chaotic_value(fdp, depth + 1) @@ -245,8 +243,7 @@ def TestOneInput(data: bytes) -> None: if key == "event_types": kwargs[key] = [ - fdp.ConsumeUnicodeNoSurrogates(8) - for _ in range(fdp.ConsumeIntInRange(1, 4)) + fdp.ConsumeUnicodeNoSurrogates(8) for _ in range(fdp.ConsumeIntInRange(1, 4)) ] elif key == "filters" and fdp.ConsumeBool(): kwargs[key] = { @@ -266,7 +263,7 @@ def TestOneInput(data: bytes) -> None: if not isinstance(result, str): handler_name = getattr(handler, "__name__", type(handler).__name__) raise RuntimeError( - f"CONTRACT VIOLATION: Handler {handler_name} returned non-string: {type(result)}" + f"CONTRACT VIOLATION: Handler {handler_name} returned non-string: {type(result)}", ) except (ApiError, TypeError, ValueError): # Expected rejections for malformed URLs, missing keys, or unsupported methods @@ -274,7 +271,7 @@ def TestOneInput(data: bytes) -> None: except Exception as e: handler_name = getattr(handler, "__name__", type(handler).__name__) raise RuntimeError( - f"UNHANDLED CRASH in {handler_name} with kwargs {list(kwargs.keys())}: {e}" + f"UNHANDLED CRASH in {handler_name} with kwargs {list(kwargs.keys())}: {e}", ) from e diff --git a/tests/fuzz/fuzz_headers.py b/tests/fuzz/fuzz_headers.py index 57fc0aa..bdae621 100755 --- a/tests/fuzz/fuzz_headers.py +++ b/tests/fuzz/fuzz_headers.py @@ -12,6 +12,7 @@ import atheris + with atheris.instrument_imports(): from mailgun.security import SecurityGuard @@ -89,10 +90,7 @@ def TestOneInput(data: bytes) -> None: val = fdp.ConsumeIntInRange(-100000, 100000) elif val_choice == 3: # Multi-value headers - val = [ - fdp.ConsumeUnicodeNoSurrogates(24) - for _ in range(fdp.ConsumeIntInRange(1, 3)) - ] + val = [fdp.ConsumeUnicodeNoSurrogates(24) for _ in range(fdp.ConsumeIntInRange(1, 3))] elif val_choice == 4: val = None elif val_choice == 5: @@ -107,26 +105,26 @@ def TestOneInput(data: bytes) -> None: if not isinstance(sanitized, dict): raise RuntimeError( - f"CONTRACT VIOLATION: sanitize_headers returned {type(sanitized)}" + f"CONTRACT VIOLATION: sanitize_headers returned {type(sanitized)}", ) # Invariant checks: sanitized headers must never leak raw CRLF or null bytes for k, v in sanitized.items(): if not isinstance(k, str) or not isinstance(v, str): raise RuntimeError( - f"TYPE DRIFT: Sanitized header contains non-str: {type(k)}={type(v)}" + f"TYPE DRIFT: Sanitized header contains non-str: {type(k)}={type(v)}", ) if "\r" in k or "\n" in k or "\x00" in k: - raise RuntimeError(f"INJECTION LEAK in header key: {repr(k)}") + raise RuntimeError(f"INJECTION LEAK in header key: {k!r}") if "\r" in v or "\n" in v or "\x00" in v: - raise RuntimeError(f"INJECTION LEAK in header value: {repr(v)}") + raise RuntimeError(f"INJECTION LEAK in header value: {v!r}") except (TypeError, ValueError, UnicodeEncodeError): # Expected security rejection for malformed headers or control characters pass except Exception as e: raise RuntimeError( - f"UNHANDLED CRASH in SecurityGuard.sanitize_headers: {e}" + f"UNHANDLED CRASH in SecurityGuard.sanitize_headers: {e}", ) from e diff --git a/tests/fuzz/fuzz_log_redaction.py b/tests/fuzz/fuzz_log_redaction.py index 77fd85d..b33e519 100755 --- a/tests/fuzz/fuzz_log_redaction.py +++ b/tests/fuzz/fuzz_log_redaction.py @@ -12,6 +12,7 @@ import atheris + with atheris.instrument_imports(): from mailgun.filters import RedactingFilter @@ -56,7 +57,9 @@ def _generate_nested_payload( if choice == 3: return { fdp.ConsumeUnicodeNoSurrogates(10): _generate_nested_payload( - fdp, depth + 1, state + fdp, + depth + 1, + state, ) for _ in range(fdp.ConsumeIntInRange(1, 3)) } @@ -97,7 +100,7 @@ def TestOneInput(data: bytes) -> None: "pubkey-", "secret-", "https://api.mailgun.net/v3/domains/", - ] + ], ) multiplier = fdp.ConsumeIntInRange(10, 80) suffix = fdp.ConsumeUnicodeNoSurrogates(120) @@ -143,14 +146,15 @@ def TestOneInput(data: bytes) -> None: # Invariant 2: String formatting safety # Guard against LibFuzzer OOM on massive format specifiers like %999999999s if isinstance(record.msg, str) and re.search( - r"%[^a-zA-Z%]*[0-9]{4,}", record.msg + r"%[^a-zA-Z%]*[0-9]{4,}", + record.msg, ): return formatted = record.getMessage() if not isinstance(formatted, str): raise RuntimeError( - f"CONTRACT VIOLATION: getMessage returned {type(formatted)}" + f"CONTRACT VIOLATION: getMessage returned {type(formatted)}", ) except (KeyError, OverflowError, TypeError, ValueError): @@ -158,7 +162,7 @@ def TestOneInput(data: bytes) -> None: pass except RecursionError as e: raise RuntimeError( - f"SECURITY BREACH: RedactingFilter failed to prevent infinite recursion: {e}" + f"SECURITY BREACH: RedactingFilter failed to prevent infinite recursion: {e}", ) from e except Exception as e: raise RuntimeError(f"UNHANDLED CRASH in RedactingFilter: {e}") from e diff --git a/tests/fuzz/fuzz_logger.py b/tests/fuzz/fuzz_logger.py index 95f2e33..eec41a1 100644 --- a/tests/fuzz/fuzz_logger.py +++ b/tests/fuzz/fuzz_logger.py @@ -13,6 +13,7 @@ import atheris + with atheris.instrument_imports(): from mailgun.logger import get_logger @@ -72,7 +73,7 @@ def TestOneInput(data: bytes) -> None: logger.warning, logger.error, logger.critical, - ] + ], ) # Message with log forging payloads diff --git a/tests/fuzz/fuzz_pagination.py b/tests/fuzz/fuzz_pagination.py index b0c86fc..6f6d257 100644 --- a/tests/fuzz/fuzz_pagination.py +++ b/tests/fuzz/fuzz_pagination.py @@ -13,6 +13,7 @@ import atheris + with atheris.instrument_imports(): from mailgun.security import SecurityGuard @@ -83,10 +84,10 @@ def TestOneInput(data: bytes) -> None: if "ascending" in params: for a_val in params["ascending"]: - _ = a_val.lower() in ("true", "1", "yes") + _ = a_val.lower() in {"true", "1", "yes"} # SSRF guard on pagination host - if parsed.scheme in ("http", "https") and parsed.netloc: + if parsed.scheme in {"http", "https"} and parsed.netloc: try: SecurityGuard.validate_mailgun_url(target_url) except (ValueError, TypeError): @@ -102,7 +103,9 @@ def TestOneInput(data: bytes) -> None: query_str = fdp.ConsumeUnicodeNoSurrogates(200) try: parsed_qs = urllib.parse.parse_qs( - query_str, keep_blank_values=True, max_num_fields=30 + query_str, + keep_blank_values=True, + max_num_fields=30, ) for k, vals in parsed_qs.items(): if "\x00" in k or any("\x00" in v for v in vals): @@ -111,7 +114,6 @@ def TestOneInput(data: bytes) -> None: # Expected for malformed fuzz inputs; continue fuzzing other paths. pass - else: # URL construction with mutated cursor tokens cursor_token = fdp.ConsumeUnicodeNoSurrogates(128) @@ -119,7 +121,7 @@ def TestOneInput(data: bytes) -> None: constructed_url = f"{base}?page={urllib.parse.quote(cursor_token)}" try: split_url = urllib.parse.urlsplit(constructed_url) - if split_url.scheme not in ("http", "https"): + if split_url.scheme not in {"http", "https"}: raise RuntimeError("Scheme corruption occurred during cursor interpolation") except (ValueError, UnicodeDecodeError): # Expected for malformed fuzz inputs; continue fuzzing other paths. diff --git a/tests/fuzz/fuzz_pagination_stream.py b/tests/fuzz/fuzz_pagination_stream.py index 5782b43..4764810 100644 --- a/tests/fuzz/fuzz_pagination_stream.py +++ b/tests/fuzz/fuzz_pagination_stream.py @@ -11,6 +11,7 @@ import atheris + with atheris.instrument_imports(): from mailgun.client import Endpoint @@ -47,7 +48,7 @@ def TestOneInput(data: bytes) -> None: initial_filters["page"] = fdp.ConsumeUnicodeNoSurrogates(16) if fdp.ConsumeBool(): initial_filters["event"] = fdp.PickValueInList( - ["delivered", "failed", "opened", "clicked", None] + ["delivered", "failed", "opened", "clicked", None], ) if fdp.ConsumeBool(): @@ -72,9 +73,7 @@ def TestOneInput(data: bytes) -> None: { "items": [{"id": "event_2", "recipient": "user2@example.com"}], "paging": { - "next": ( - next_url if fdp.ConsumeBool() else None - ), # 50% test cyclic cursor + "next": (next_url if fdp.ConsumeBool() else None), # 50% test cyclic cursor "previous": None, }, }, @@ -105,9 +104,7 @@ def mock_get(*args: Any, **kwargs: Any) -> Any: stream_gen = ep.stream(filters=initial_filters) # Pull up to 3 pages to test iteration, type casting, and termination - iterations = 0 - for _ in stream_gen: - iterations += 1 + for iterations, _ in enumerate(stream_gen, start=1): if iterations >= 3: # Invariant: prevent endless loops on cyclic next URLs break @@ -117,7 +114,7 @@ def mock_get(*args: Any, **kwargs: Any) -> Any: pass except Exception as e: raise RuntimeError( - f"UNHANDLED CRASH in Endpoint.stream with filters {initial_filters} and url {repr(next_url)}: {e}" + f"UNHANDLED CRASH in Endpoint.stream with filters {initial_filters} and url {next_url!r}: {e}", ) from e diff --git a/tests/fuzz/fuzz_pydantic_models.py b/tests/fuzz/fuzz_pydantic_models.py index 077c954..5bd26cd 100644 --- a/tests/fuzz/fuzz_pydantic_models.py +++ b/tests/fuzz/fuzz_pydantic_models.py @@ -12,6 +12,7 @@ import atheris + with atheris.instrument_imports(): from pydantic import ValidationError @@ -25,7 +26,7 @@ "evil\r\nBcc: victim@target.com", "user@localhost", "test@[127.0.0.1]", - "\"Recipient with quotes\" ", + '"Recipient with quotes" ', "user+tag@domain.co.uk", "a" * 255 + "@long.domain.com", "user@" + "sub." * 50 + "com", @@ -44,8 +45,7 @@ def _generate_fuzzed_recipient(fdp: atheris.FuzzedDataProvider) -> Any: return fdp.ConsumeUnicodeNoSurrogates(64) if choice == 2: return [ - fdp.PickValueInList(_MALICIOUS_RECIPIENTS) - for _ in range(fdp.ConsumeIntInRange(1, 4)) + fdp.PickValueInList(_MALICIOUS_RECIPIENTS) for _ in range(fdp.ConsumeIntInRange(1, 4)) ] if choice == 3: return fdp.ConsumeInt(10000) @@ -64,15 +64,9 @@ def TestOneInput(data: bytes) -> None: if fdp.ConsumeBool() else fdp.ConsumeUnicodeNoSurrogates(48) ) - subject_val = ( - fdp.ConsumeUnicodeNoSurrogates(64) if fdp.ConsumeBool() else None - ) - text_val = ( - fdp.ConsumeUnicodeNoSurrogates(128) if fdp.ConsumeBool() else None - ) - html_val = ( - fdp.ConsumeUnicodeNoSurrogates(128) if fdp.ConsumeBool() else None - ) + subject_val = fdp.ConsumeUnicodeNoSurrogates(64) if fdp.ConsumeBool() else None + text_val = fdp.ConsumeUnicodeNoSurrogates(128) if fdp.ConsumeBool() else None + html_val = fdp.ConsumeUnicodeNoSurrogates(128) if fdp.ConsumeBool() else None # Custom parameter mutations (testing h:, v:, o: prefix handling & CRLF rejection) custom_params: dict[str, Any] = {} @@ -117,7 +111,7 @@ def TestOneInput(data: bytes) -> None: for field in ("from_", "subject"): field_val = dumped.get(field) if isinstance(field_val, str) and ("\r" in field_val or "\n" in field_val): - raise RuntimeError(f"CRLF injection leak in model field {field}: {repr(field_val)}") + raise RuntimeError(f"CRLF injection leak in model field {field}: {field_val!r}") except (ValidationError, ValueError, TypeError): # Expected defensive rejections from Pydantic validators diff --git a/tests/fuzz/fuzz_security_primitives.py b/tests/fuzz/fuzz_security_primitives.py index 951a35c..1abfa34 100755 --- a/tests/fuzz/fuzz_security_primitives.py +++ b/tests/fuzz/fuzz_security_primitives.py @@ -17,6 +17,7 @@ import atheris + with atheris.instrument_imports(): from mailgun.security import SecurityGuard @@ -73,7 +74,7 @@ def TestOneInput(data: bytes) -> None: # Invariant: Output must never contain directory traversal sequences or null bytes if ".." in sanitized or "\x00" in sanitized or "/" in sanitized or "\\" in sanitized: - raise RuntimeError(f"PATH ESCAPE LEAK in sanitized segment: {repr(sanitized)}") + raise RuntimeError(f"PATH ESCAPE LEAK in sanitized segment: {sanitized!r}") elif target == 1: # Target 2: SSRF and Scheme/Host Validation (CWE-918) @@ -102,9 +103,9 @@ def TestOneInput(data: bytes) -> None: if sanitized_headers is not None: for k, v in sanitized_headers.items(): if "\r" in k or "\n" in k or "\x00" in k: - raise RuntimeError(f"CRLF LEAK in header key: {repr(k)}") + raise RuntimeError(f"CRLF LEAK in header key: {k!r}") if "\r" in v or "\n" in v or "\x00" in v: - raise RuntimeError(f"CRLF LEAK in header value: {repr(v)}") + raise RuntimeError(f"CRLF LEAK in header value: {v!r}") elif target == 3: # Target 4: Timeout Bounds and Overflow Defense (CWE-400) @@ -115,14 +116,20 @@ def TestOneInput(data: bytes) -> None: elif timeout_choice == 1: timeout = (fdp.ConsumeFloat(), fdp.ConsumeFloat()) elif timeout_choice == 2: - timeout = fdp.PickValueInList([float("inf"), float("-inf"), float("nan"), -1.0, 0.0]) + timeout = fdp.PickValueInList( + [float("inf"), float("-inf"), float("nan"), -1.0, 0.0], + ) else: timeout = fdp.ConsumeUnicodeNoSurrogates(16) sanitized_timeout = SecurityGuard.sanitize_timeout(timeout) # Invariant: Returned timeout must be positive and non-infinite if isinstance(sanitized_timeout, (int, float)): - if sanitized_timeout <= 0 or math.isnan(sanitized_timeout) or math.isinf(sanitized_timeout): + if ( + sanitized_timeout <= 0 + or math.isnan(sanitized_timeout) + or math.isinf(sanitized_timeout) + ): raise RuntimeError(f"INVALID TIMEOUT escaped: {sanitized_timeout}") elif isinstance(sanitized_timeout, tuple): for t in sanitized_timeout: diff --git a/tests/fuzz/fuzz_semantic_payloads.py b/tests/fuzz/fuzz_semantic_payloads.py index 3ff60e5..9ad890f 100644 --- a/tests/fuzz/fuzz_semantic_payloads.py +++ b/tests/fuzz/fuzz_semantic_payloads.py @@ -14,6 +14,7 @@ import atheris + with atheris.instrument_imports(): from mailgun.client import Client from mailgun.handlers.error_handler import ApiError @@ -37,8 +38,7 @@ def _generate_nested_json_tree(fdp: atheris.FuzzedDataProvider, depth: int = 0) return None if choice == 4: return [ - _generate_nested_json_tree(fdp, depth + 1) - for _ in range(fdp.ConsumeIntInRange(1, 3)) + _generate_nested_json_tree(fdp, depth + 1) for _ in range(fdp.ConsumeIntInRange(1, 3)) ] return { fdp.ConsumeUnicodeNoSurrogates(8): _generate_nested_json_tree(fdp, depth + 1) @@ -61,8 +61,7 @@ def _generate_semantic_message(fdp: atheris.FuzzedDataProvider) -> dict[str, Any # Option parameters (o:tag, o:tracking, o:deliverytime) if fdp.ConsumeBool(): payload["o:tag"] = [ - fdp.ConsumeUnicodeNoSurrogates(12) - for _ in range(fdp.ConsumeIntInRange(1, 3)) + fdp.ConsumeUnicodeNoSurrogates(12) for _ in range(fdp.ConsumeIntInRange(1, 3)) ] if fdp.ConsumeBool(): payload["o:tracking"] = fdp.PickValueInList([True, False, "yes", "no", None]) @@ -74,7 +73,7 @@ def _generate_semantic_message(fdp: atheris.FuzzedDataProvider) -> dict[str, Any "2024-10-25T23:10:10Z", fdp.ConsumeInt(10**10), "invalid-time-format", - ] + ], ) # Deep recipient-variables serialization @@ -122,6 +121,7 @@ def TestOneInput(data: bytes) -> None: except Exception as e: raise RuntimeError(f"SEMANTIC CRASH: {type(e).__name__} - {e}") from e + if __name__ == "__main__": atheris.instrument_all() atheris.Setup(sys.argv, TestOneInput) diff --git a/tests/fuzz/fuzz_spamguard.py b/tests/fuzz/fuzz_spamguard.py index 797b685..d96d0ee 100644 --- a/tests/fuzz/fuzz_spamguard.py +++ b/tests/fuzz/fuzz_spamguard.py @@ -13,6 +13,7 @@ import atheris + with atheris.instrument_imports(): from mailgun.security import SpamGuard @@ -22,11 +23,11 @@ 'Click here', 'No image', '
Hidden Spam
', - '