From 343b83f02b93349e4f9cdf888be8afdaf5481bb2 Mon Sep 17 00:00:00 2001 From: Cursor Date: Sun, 1 Nov 2026 19:04:40 +0000 Subject: [PATCH] fix: restrict pycurl high-level API to HTTP(S) and reject CRLF Block file:// and other non-HTTP schemes in pycurl_proxy.request(), pin libcurl protocols to HTTP(S), and validate origin headers and methods so CR/LF cannot smuggle extra request lines. Co-authored-by: ProxyMesh AI --- python_proxy_headers/header_utils.py | 42 +++++++++++++++++++++-- python_proxy_headers/pycurl_proxy.py | 26 +++++++++++++-- test_header_security.py | 50 ++++++++++++++++++++++++++++ 3 files changed, 113 insertions(+), 5 deletions(-) diff --git a/python_proxy_headers/header_utils.py b/python_proxy_headers/header_utils.py index eb4d69d..cea0681 100644 --- a/python_proxy_headers/header_utils.py +++ b/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 [] diff --git a/python_proxy_headers/pycurl_proxy.py b/python_proxy_headers/pycurl_proxy.py index d703a54..fe141c6 100644 --- a/python_proxy_headers/pycurl_proxy.py +++ b/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': diff --git a/test_header_security.py b/test_header_security.py index 6be189f..a9a93ba 100644 --- a/test_header_security.py +++ b/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: