From 45a229c56a51021afb5fd689c28616c8ccd2514e Mon Sep 17 00:00:00 2001
From: Serhii Kupriienko
<291109589+skupriienko-mailgun@users.noreply.github.com>
Date: Tue, 8 Sep 2026 13:38:20 +0300
Subject: [PATCH 01/27] chore(scripts): make manage.sh executable (mode 100755)
---
manage.sh | 0
1 file changed, 0 insertions(+), 0 deletions(-)
mode change 100644 => 100755 manage.sh
diff --git a/manage.sh b/manage.sh
old mode 100644
new mode 100755
From 3bdba9a4be8933d77b8faaef389c3d281ad2a68d Mon Sep 17 00:00:00 2001
From: Serhii Kupriienko
<291109589+skupriienko-mailgun@users.noreply.github.com>
Date: Tue, 8 Sep 2026 13:46:07 +0300
Subject: [PATCH 02/27] fix(security): harden security, URL sanitization, and
idempotency hashing
- Neutralize multi-pass URL encoding bypasses (CWE-116) and catch timeout float overflows.
- Add _deep_sanitize to IdempotencyGuard against cyclic references and recursion overflows.
- Preserve and restore file stream seek pointers during payload fingerprint hashing.
- Block executable tags and inline event handlers in SpamGuard.
- Update unit tests covering proxy initialization, overflow validation, and tag blocks.
---
mailgun/security.py | 264 +++++++++++++++++++----------
tests/unit/test_client_security.py | 29 ++++
tests/unit/test_security_guards.py | 37 ++++
3 files changed, 238 insertions(+), 92 deletions(-)
diff --git a/mailgun/security.py b/mailgun/security.py
index a69609e9..59b64461 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]")
@@ -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"}
)
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."
)
+
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.")
@@ -314,21 +334,23 @@ def filter_safe_kwargs(cls, kwargs: dict[str, Any]) -> dict[str, Any]:
@staticmethod
def sanitize_headers(headers: dict[str, str] | None) -> dict[str, str] | None:
- """Poka-yoke: Prevent HTTP Header Injection (CWE-113).
+ """Poka-yoke: Prevent HTTP Header Injection (CWE-113 / RFC 9110).
Returns:
The sanitized headers dictionary, or None if no headers were provided.
Raises:
- ValueError: If a CRLF injection pattern is detected in any header key or value.
+ ValueError: If CRLF or control character sequences are detected.
"""
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)
+ k_str, v_str = str(key), str(value)
+ 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:
+ # PEP 578: Emit Enterprise security telemetry before crashing
+ sys.audit("mailgun.security.header_injection", key)
msg = f"CRLF injection detected in header: {key}"
raise ValueError(msg)
@@ -342,7 +364,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 +378,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,8 +445,9 @@ 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)
@@ -435,7 +459,8 @@ def validate_mailgun_url(url: str) -> str:
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)
@@ -551,26 +576,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 +610,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,10 +643,10 @@ 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.
@@ -630,8 +655,10 @@ def normalize_domain(domain: str | None) -> str:
return ""
try:
- # Encode the Unicode string to IDNA bytes, then decode to an ASCII string.
- # If the domain is already ASCII (e.g., 'example.com'), it remains unchanged.
+ 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 +677,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 +687,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: "
+""
+"