Patched code sample

import re
from typing import Any, Mapping, Optional

# This regex is representative of the fix, used to find forbidden carriage
# return or line feed characters in header keys and values.
_FORBIDDEN_HEADER_CHARS_RE = re.compile(r"[\r\n]")


def validate_multipart_headers(headers: Optional[Mapping[str, Any]]) -> None:
    """
    This function represents the fix for the header injection vulnerability.

    It validates that header keys and values do not contain carriage return
    (\r) or newline (\n) characters before they are used in a multipart
    request part. The vulnerable version of the code would proceed without
    this check, allowing a specially crafted header value to inject additional
    headers into the final request.
    """
    if headers is None:
        return

    for key, value in headers.items():
        # The fix is to explicitly check both the key and the value for
        # forbidden characters and raise an error if they are found.
        if _FORBIDDEN_HEADER_CHARS_RE.search(str(key)):
            raise ValueError(f"Invalid character in header key: {key!r}")
        if _FORBIDDEN_HEADER_CHARS_RE.search(str(value)):
            raise ValueError(f"Invalid character in header value: {value!r}")


# --- Demonstration of the fix ---

# 1. Define a malicious header value attempting to inject a new header.
# The `\r\n` is an attempt to terminate the current header and start a new one.
malicious_header_value = (
    'form-data; name="file"\r\nInjected-Header: Malicious-Value'
)
malicious_headers = {"Content-Disposition": malicious_header_value}

# 2. The patched code successfully raises a ValueError, preventing the injection.
try:
    validate_multipart_headers(malicious_headers)
    # This line should not be reached
    print("Validation failed to catch malicious header.")
except ValueError as e:
    # This block is expected to execute, demonstrating the fix works.
    print(f"Fix successfully caught malicious header: {e}")
    assert "Invalid character in header value" in str(e)


# 3. A valid header dictionary that passes the check.
valid_headers = {"Content-Disposition": 'form-data; name="file"'}
try:
    validate_multipart_headers(valid_headers)
    print("Validation correctly passed for legitimate header.")
except ValueError:
    # This block should not be reached
    print("Validation incorrectly failed for legitimate header.")