feat(security): implement advanced input guardrails and fuzzing tests

- client: add CRLF header validation before request dispatch
- config: enforce strict float coercion and math invariants for timeouts
- endpoint: explicitly encode dots in CSV routes to prevent path traversal
- guardrails: patch ReDoS in secret redaction; add recursive log scrubbing and 1000-char truncation limits
- fuzzing: implement network fault injection (Evil Server) and file I/O mocking
- manage.sh: support dynamic LibFuzzer args and fail CI on non-zero exit codes

Author: skupriienko

.github/workflows/codeql.yml

@@ -13,7 +13,7 @@ jobs:
       contents: read
       actions: read
     steps:
-      - uses: actions/checkout@v4
+      - uses: actions/checkout@v6
       - uses: github/codeql-action/init@v3
         with:
           languages: python

.github/workflows/security.yml

@@ -12,8 +12,8 @@ jobs:
   static-analysis:
     runs-on: ubuntu-latest
     steps:
-      - uses: actions/checkout@v4
-      - uses: actions/setup-python@v5
+      - uses: actions/checkout@v6
+      - uses: actions/setup-python@v6
         with:
           python-version: "3.13"
           cache: 'pip'
@@ -26,7 +26,7 @@ jobs:
   semgrep:
     runs-on: ubuntu-latest
     steps:
-      - uses: actions/checkout@v4
+      - uses: actions/checkout@v6
       - uses: returntocorp/semgrep-action@v1
         with:
           config: >-
@@ -40,8 +40,8 @@ jobs:
   pip-audit:
     runs-on: ubuntu-latest
     steps:
-      - uses: actions/checkout@v4
-      - uses: actions/setup-python@v5
+      - uses: actions/checkout@v6
+      - uses: actions/setup-python@v6
         with: { python-version: "3.13" }
       - run: pip install pip-audit
       - run: pip-audit --strict

mailjet_rest/client.py

@@ -406,6 +406,10 @@ def _execute_request(
         Returns:
             requests.Response: The HTTP response from the Mailjet API.
         """
+        # Prevent CRLF header injection (CWE-113)
+        if req_kwargs.get("headers"):
+            SecurityGuard.validate_crlf_headers(req_kwargs["headers"])
+
         logger.debug(
             "Sending Request: %s %s%s",
             method,

mailjet_rest/config.py

@@ -1,7 +1,8 @@
 """Configuration settings for the Mailjet SDK."""
 
+import math
 from dataclasses import dataclass
-from typing import ClassVar, cast
+from typing import ClassVar
 
 from mailjet_rest._version import __version__
 from mailjet_rest.types import _DEFAULT_TIMEOUT, _JSON_HEADERS, _TEXT_HEADERS, TimeoutType
@@ -40,20 +41,41 @@ def __post_init__(self) -> None:
         if not self.api_url.endswith("/"):
             self.api_url += "/"
 
-        def _validate_timeout(t: float) -> None:
-            if t <= 0 or t > 300:
-                err_msg = f"Timeout values must be strictly between 1 and 300 seconds, got {t}."
-                raise ValueError(err_msg)
-
         if self.timeout is not None:
             if isinstance(self.timeout, tuple):
                 if len(self.timeout) != 2:
-                    msg = f"Timeout tuple must contain exactly two elements, got {self.timeout}."
+                    msg = "Timeout tuple must contain exactly two elements (connect, read)."
                     raise ValueError(msg)
-                for t_val in self.timeout:
-                    _validate_timeout(t_val)
+                clean_tuple = []
+                for t in self.timeout:
+                    # 1. Scope type coercion strictly
+                    try:
+                        val = float(t)
+                    except (ValueError, TypeError) as e:
+                        msg = f"Invalid timeout tuple element: {t}"
+                        raise ValueError(msg) from e
+
+                    # 2. Scope invariant validation separately
+                    if math.isnan(val) or math.isinf(val) or val <= 0:
+                        msg = f"Timeout tuple values must be positive finite numbers, got {val}"
+                        raise ValueError(msg)
+
+                    clean_tuple.append(val)
+                self.timeout = tuple(clean_tuple)  # type: ignore[assignment]
             else:
-                _validate_timeout(cast("float", self.timeout))
+                # 1. Scope type coercion strictly
+                try:
+                    val = float(self.timeout)  # type: ignore[arg-type]
+                except (ValueError, TypeError) as e:
+                    msg = f"Security Violation: Timeout must be numeric, got {type(self.timeout).__name__}."
+                    raise ValueError(msg) from e
+
+                # 2. Scope invariant validation separately
+                if math.isnan(val) or math.isinf(val) or val <= 0:
+                    msg = f"Timeout must be a strictly positive finite number, got {val}"
+                    raise ValueError(msg)
+
+                self.timeout = val
 
     def __getitem__(self, key: str) -> tuple[str, dict[str, str]]:
         """Retrieve the base API endpoint URL and default headers for a given key.

mailjet_rest/endpoint.py

@@ -32,7 +32,11 @@ def _route_send(base: str, ver: str, _parts: list[str], _id_val: str, _action: s
 
 
 def _route_csv(base: str, ver: str, parts: list[str], id_val: str, _action: str, name: str) -> str:
-    url = f"{base}/{ver}/DATA/{parts[0]}"
+    # 1. quote() encodes slashes and spaces (CWE-20)
+    # 2. replace() explicitly encodes dots to prevent strict ".." evaluation (CWE-22)
+    safe_part = quote(parts[0], safe="").replace(".", "%2E")
+
+    url = f"{base}/{ver}/DATA/{safe_part}"
     if id_val:  # Only append suffix if an ID was passed
         suffix = "CSVData/text:plain" if name.endswith("_csvdata") else "CSVError/text:csv"
         url += f"{id_val}/{suffix}"

mailjet_rest/utils/guardrails.py

@@ -27,11 +27,17 @@
 def _get_secret_pattern() -> re.Pattern[str]:
     """Lazy-compile strict patterns to minimize cold-boot overhead.
 
+    ReDoS prevented by eliminating overlapping quantifiers (+)
+    and enforcing strict finite bounds ({1,5} and {1,200}).
+
     Returns:
         re.Pattern[str]: Compiled regular expression for secret pattern matching.
     """
+    # Group 1: The key/header name (e.g., Authorization)
+    # Group 2: The separator and scheme (e.g., ': Bearer ')
+    # Group 3: The actual secret (matched but NOT included in the substitution)
     return re.compile(
-        r"(?i)(Authorization|api[_-]key|api[_-]secret|token)([:\s=]+(?:Bearer\s+|Basic\s+|Token\\s+)?)([^\s'\"]+)"
+        r"(?i)(Authorization|api[_-]key|api[_-]secret|token)([:\s=]{1,5}(?:(?:Bearer|Basic|Token)\s{1,5})?)([^\s'\"]{1,200})"
     )
 
 
@@ -48,7 +54,7 @@ def init_poolmanager(self, connections: int, maxsize: int, block: bool = False,
 
 
 class RedactingFilter(logging.Filter):
-    """Filters out sensitive patterns from log messages and arguments."""
+    """Filter that intercepts and masks sensitive credentials in log records."""
 
     @override
     def filter(self, record: logging.LogRecord) -> bool:
@@ -62,20 +68,28 @@ def filter(self, record: logging.LogRecord) -> bool:
         if isinstance(record.msg, str):
             record.msg = _get_secret_pattern().sub(r"\1\2********", record.msg)
 
-        # Redact arguments
-        if record.args:
-            new_args: list[Any] = []
+        if isinstance(record.args, tuple) and record.args:
+            new_args = []
             for arg in record.args:
                 if isinstance(arg, str):
                     new_args.append(_get_secret_pattern().sub(r"\1\2********", arg))
                 else:
-                    new_args.append(arg)
+                    new_args.append(arg)  # type: ignore[arg-type]
             record.args = tuple(new_args)
+        elif isinstance(record.args, dict) and record.args:
+            new_dict_args = {}
+            for k, v in record.args.items():
+                if isinstance(v, str):
+                    new_dict_args[k] = _get_secret_pattern().sub(r"\1\2********", v)
+                else:
+                    new_dict_args[k] = v
+            record.args = new_dict_args
+
         return True
 
 
 class SecurityGuard:
-    """Centralized OWASP API security guardrails."""
+    """Centralized security validation and sanitization (Defense in Depth)."""
 
     _audit_hook_installed: ClassVar[bool] = False
 
@@ -132,11 +146,14 @@ def sanitize_log_trace(val: Any) -> str:
         Returns:
             str: The sanitized string value.
         """
-        s = str(val)
-        # If the input contains control characters, reject/scrub to prevent injection.
-        if _CRLF_RE.search(s):
-            return "[INVALID_DATA_REDACTED]"
-        return s
+        # CAST TO STRING FIRST to prevent TypeError on ints/floats
+        val_str = str(val)
+
+        # Fail-fast on insanely large strings (CWE-400 Resource Exhaustion)
+        if len(val_str) > 1000:
+            val_str = val_str[:1000] + "... [TRUNCATED]"
+
+        return _CRLF_RE.sub("", val_str)
 
     @staticmethod
     def check_request_security(kwargs: dict[str, Any]) -> None:
@@ -220,8 +237,8 @@ def validate_crlf_headers(custom_headers: dict[str, str]) -> None:
         """
         for key, value in custom_headers.items():
             if _CRLF_RE.search(str(value)):
-                err_msg = f"CRLF Injection detected in header '{key}'"
-                raise ValueError(err_msg)
+                msg = f"Security Violation: CRLF Injection detected in header '{key}'."
+                raise ValueError(msg)
 
     @staticmethod
     def validate_attachment_path(file_path: str | Path, safe_base_dir: str | Path | None = None) -> Path:
@@ -276,5 +293,6 @@ def sanitize_segment(segment: Any) -> str:
             return ""
 
         clean_str = str(segment).replace("\r", "").replace("\n", "")
-        # Safe is empty string to strictly encode EVERYTHING, including slashes
-        return quote(clean_str, safe="")
+        # quote() ignores dots. We explicitly encode them to prevent
+        # strict ".." traversal when injected into middle of URI templates.
+        return quote(clean_str, safe="").replace(".", "%2E")

manage.sh

@@ -103,37 +103,64 @@ test_strict_warnings() {
     pytest -W "error::DeprecationWarning" "$@"
 }
 
+
 # ==============================================================================
 # SECURITY & FUZZING
 # ==============================================================================
 fuzz_all() {
-    # Usage: ./manage.sh fuzz_all [duration]
+    # Usage: ./manage.sh fuzz_all [duration_in_seconds]
     local duration=${1:-30}
     local fuzzer_dir="tests/fuzz"
     local dictionary="tests/fuzz/fuzzer.dict"
+    local corpus_dir="tests/fuzz/corpus"
 
     if [ ! -d "$fuzzer_dir" ]; then
         error "Fuzzer directory '$fuzzer_dir' not found."
         return 1
     fi
 
-    info "🚀 Starting security fuzzing suite (duration: ${duration}s)..."
+    # Ensure the dictionary exists before passing the argument
+    local dict_arg=""
+    if [ -f "$dictionary" ]; then
+        dict_arg="-dict=$dictionary"
+    else
+        echo "⚠️ Warning: Dictionary '$dictionary' not found. Running without it."
+    fi
+
+    info "🚀 Starting security fuzzing suite (duration: ${duration} seconds per fuzzer)..."
+
+    # Safely gather fuzzer files to prevent errors if none exist
+    shopt -s nullglob
+    local fuzzers=("$fuzzer_dir"/fuzz_*.py)
+    shopt -u nullglob
 
-    for fuzzer in "$fuzzer_dir"/fuzz_*.py; do
-        if [[ "$fuzzer" == *".dict" ]]; then continue; fi
+    if [ ${#fuzzers[@]} -eq 0 ]; then
+        error "No fuzzer scripts found in '$fuzzer_dir'."
+        return 1
+    fi
+
+    for fuzzer in "${fuzzers[@]}"; do
+        local fuzzer_name=$(basename "$fuzzer" .py)
+        local fuzzer_corpus="$corpus_dir/$fuzzer_name"
+        mkdir -p "$fuzzer_corpus"
 
-        info "🔍 Running fuzzer: $fuzzer"
+        info "🔍 Running fuzzer: $fuzzer (Corpus: $fuzzer_corpus)"
 
         conda run --name "${CONDA_ENV_NAME}" python "$fuzzer" \
-            -dict="$dictionary" \
+            $dict_arg \
             -max_len=512 \
-            -max_total_time="$duration"
-
-        if [ $? -eq 77 ]; then
-            error "❌ Fuzzing failed: Crash detected in $fuzzer."
-            return 77
+            -max_total_time="$duration" \
+            "$fuzzer_corpus"
+
+        local exit_code=$?
+        # libFuzzer returns 1 for crash, 70 for OOM, 77 for timeout.
+        # Catching any non-zero exit ensures we don't miss Python tracebacks.
+        if [ $exit_code -ne 0 ]; then
+            error "❌ Fuzzing failed: Crash or error detected in $fuzzer (Exit Code: $exit_code)."
+            return $exit_code
         fi
     done
+
     success "✅ All fuzz tests passed successfully."
 }
 
@@ -272,9 +299,40 @@ COMMAND=$1
 shift # Remove the command from the arguments list, leaving only extra flags
 
 case "$COMMAND" in
-    env_setup|format|lint|test_all|test_unit|test_integration|test_cov|test_no_warnings|test_strict_warnings|fuzz_all|perf_bench|perf_profile|audit_deps|run_hooks|build_pkg|release|clean|help)
+    env_setup|format|lint|test_all|test_unit|test_integration|test_cov|test_no_warnings|test_strict_warnings|perf_bench|perf_profile|audit_deps|run_hooks|build_pkg|release|clean|help)
         "$COMMAND" "$@" # Execute the function with any remaining arguments
         ;;
+    fuzz_all)
+        # 1. Grab the duration, defaulting to 20 if not provided
+        DURATION=${1:-20}
+
+        # 2. Shift the duration out of the arguments list (if it was provided)
+        # This leaves ONLY the extra flags (like -max_len=16384) in "$@"
+        if [ $# -gt 0 ]; then shift; fi
+
+        info "🚀 Starting security fuzzing suite (duration: ${DURATION} seconds per fuzzer)..."
+        if [ $# -gt 0 ]; then
+            info "Applying extra LibFuzzer arguments: $@"
+        fi
+
+        # Find all fuzzers
+        FUZZERS=$(find tests/fuzz -maxdepth 1 -name "fuzz_*.py" -type f)
+
+        for fuzzer in $FUZZERS; do
+            fuzzer_name=$(basename "$fuzzer" .py)
+            corpus_dir="tests/fuzz/corpus/$fuzzer_name"
+
+            mkdir -p "$corpus_dir"
+
+            info "🔍 Running fuzzer: $fuzzer (Corpus: $corpus_dir)"
+
+            # 3. Append "$@" to the end to forward any extra arguments
+            python "$fuzzer" "$corpus_dir" -dict=tests/fuzz/fuzzer.dict -max_total_time="$DURATION" "$@"
+
+            echo ""
+        done
+        success "✅ All fuzz tests passed successfully."
+        ;;
     *)
         error "Unknown command: $COMMAND"
         help