Fix pycurl file:// fetches and origin-header CRLF injection

python_proxy_headers/header_utils.py

@@ -9,6 +9,7 @@
 from __future__ import annotations
 
 from typing import Any, Dict, List, Mapping, MutableMapping, Optional, Sequence, Tuple, Union
+from urllib.parse import urlparse
 
 RawHeaders = Union[Mapping[Any, Any], Sequence[Tuple[Any, Any]], None]
 
@@ -88,11 +89,11 @@ def validate_header_value(value: Any) -> str:
 
 
 def validate_headers(headers: Optional[Mapping[Any, Any]]) -> Dict[str, str]:
-	"""Validate a mapping of CONNECT request headers.
+	"""Validate a mapping of HTTP header names and values.
 
 	Raises:
 		ValueError: If any name or value contains CR, LF, NUL, or other
-			characters illegal in a single CONNECT header line.
+			characters illegal in a single header line.
 	"""
 	if not headers:
 		return {}
@@ -102,6 +103,43 @@ def validate_headers(headers: Optional[Mapping[Any, Any]]) -> Dict[str, str]:
 	return validated
 
 
+def validate_http_method(method: Any) -> str:
+	"""Return an HTTP method token, or raise ValueError if it is unsafe.
+
+	Methods must be a single token (no spaces, slashes, or control characters)
+	so they cannot smuggle extra request lines through CURLOPT_CUSTOMREQUEST.
+	"""
+	method_s = _as_str(method).upper()
+	if not method_s:
+		raise ValueError("HTTP method must not be empty")
+	for char in method_s:
+		code = ord(char)
+		if char in "()<>@,;:\\\"/[]?={} \t" or code <= 32 or code == 127:
+			raise ValueError(f"Invalid HTTP method {method_s!r}")
+	return method_s
+
+
+def validate_http_url(url: Any) -> str:
+	"""Return an http(s) URL, or raise ValueError for other schemes.
+
+	High-level HTTP helpers must not pass caller URLs to protocol-agile
+	backends such as libcurl, which otherwise honor ``file://`` and similar
+	schemes (and ignore an HTTP proxy for those schemes).
+	"""
+	url_s = _as_str(url)
+	if not url_s or any(char in url_s for char in "\r\n\x00"):
+		raise ValueError(f"Invalid URL {url_s!r}")
+	parsed = urlparse(url_s)
+	scheme = parsed.scheme.lower()
+	if scheme not in ("http", "https"):
+		raise ValueError(
+			f"Unsupported URL scheme {parsed.scheme!r}; only http and https are allowed"
+		)
+	if not parsed.netloc:
+		raise ValueError(f"URL must include a host: {url_s!r}")
+	return url_s
+
+
 def _header_items(headers: RawHeaders) -> List[Tuple[Any, Any]]:
 	if not headers:
 		return []

python_proxy_headers/pycurl_proxy.py

@@ -38,7 +38,7 @@
 from dataclasses import dataclass, field
 from typing import Dict, List, Optional, Tuple
 
-from .header_utils import validate_headers
+from .header_utils import validate_headers, validate_http_method, validate_http_url
 
 try:
     import pycurl
@@ -263,6 +263,18 @@ def raise_for_status(self) -> None:
             raise Exception(f"HTTP Error {self.status_code}")
 
 
+def _restrict_http_protocols(curl) -> None:
+    """Limit libcurl to HTTP(S) so file:// and similar schemes cannot run."""
+    http_https = pycurl.PROTO_HTTP | pycurl.PROTO_HTTPS
+    curl.setopt(pycurl.PROTOCOLS, http_https)
+    redir_protocols = getattr(pycurl, "REDIR_PROTOCOLS", None)
+    if redir_protocols is not None:
+        try:
+            curl.setopt(redir_protocols, http_https)
+        except pycurl.error:
+            pass
+
+
 def request(
     method: str,
     url: str,
@@ -278,7 +290,7 @@ def request(
 
     Args:
         method: HTTP method (GET, POST, etc.)
-        url: Target URL
+        url: Target URL (http and https only)
         proxy: Proxy URL (e.g., 'http://user:pass@proxy:8080')
         proxy_headers: Headers to send to the proxy
         headers: Headers to send to the origin server
@@ -288,17 +300,25 @@ def request(
 
     Returns:
         Response object with body, headers, and proxy_headers
+
+    Raises:
+        ValueError: If the URL scheme is not http/https, the method is not a
+            single token, or a header name/value contains CR, LF, or NUL.
     """
+    url = validate_http_url(url)
+    method = validate_http_method(method)
+    headers = validate_headers(headers)
+
     c = pycurl.Curl()
     body = BytesIO()
     capture = HeaderCapture(c)
 
     try:
+        _restrict_http_protocols(c)
         c.setopt(pycurl.URL, url)
         c.setopt(pycurl.WRITEFUNCTION, body.write)
 
         # HTTP method
-        method = method.upper()
         if method == 'GET':
             c.setopt(pycurl.HTTPGET, 1)
         elif method == 'POST':

test_header_security.py

@@ -19,6 +19,8 @@
     validate_header_name,
     validate_header_value,
     validate_headers,
+    validate_http_method,
+    validate_http_url,
 )
 
 
@@ -73,6 +75,54 @@ def test_pycurl_rejects_crlf(self) -> None:
         finally:
             curl.close()
 
+    def test_accepts_http_methods_and_urls(self) -> None:
+        self.assertEqual(validate_http_method("get"), "GET")
+        self.assertEqual(validate_http_method("PROPFIND"), "PROPFIND")
+        self.assertEqual(validate_http_url("https://example.com/path"), "https://example.com/path")
+        self.assertEqual(validate_http_url("http://127.0.0.1:8080/"), "http://127.0.0.1:8080/")
+
+    def test_rejects_unsafe_http_method(self) -> None:
+        with self.assertRaises(ValueError):
+            validate_http_method("GET / HTTP/1.1\r\nX-Injected: pwned\r\n")
+        with self.assertRaises(ValueError):
+            validate_http_method("GET\nPUT")
+
+    def test_rejects_non_http_url_schemes(self) -> None:
+        with self.assertRaises(ValueError):
+            validate_http_url("file:///etc/passwd")
+        with self.assertRaises(ValueError):
+            validate_http_url("FILE://localhost/etc/passwd")
+        with self.assertRaises(ValueError):
+            validate_http_url("dict://127.0.0.1:1/default:help")
+        with self.assertRaises(ValueError):
+            validate_http_url("http://example.com\r\nX-Injected: pwned/")
+
+    def test_pycurl_request_rejects_file_scheme(self) -> None:
+        try:
+            from python_proxy_headers.pycurl_proxy import get, request
+        except ImportError:
+            self.skipTest("pycurl is not installed")
+        with self.assertRaises(ValueError):
+            get("file:///etc/passwd")
+        with self.assertRaises(ValueError):
+            get("file:///etc/passwd", proxy="http://127.0.0.1:9")
+        with self.assertRaises(ValueError):
+            request("GET", "dict://127.0.0.1:1/default:help")
+
+    def test_pycurl_request_rejects_origin_header_and_method_crlf(self) -> None:
+        try:
+            from python_proxy_headers.pycurl_proxy import request
+        except ImportError:
+            self.skipTest("pycurl is not installed")
+        with self.assertRaises(ValueError):
+            request(
+                "GET",
+                "http://127.0.0.1:9/",
+                headers={"X-Foo": "bar\r\nX-Injected: pwned"},
+            )
+        with self.assertRaises(ValueError):
+            request("GET / HTTP/1.1\r\nX-Injected: pwned\r\n", "http://127.0.0.1:9/")
+
 
 class MergeProxyHeadersTests(unittest.TestCase):
     def test_merges_custom_proxy_header(self) -> None: