Percent-encode targets and URLs in ConnectionPool
MODIFY
dummyserver/handlers.py
dummyserver/handlers.py
@@ -17,7 +17,7 @@ from datetime import timedelta
from urllib3.packages.six.moves.http_client import responses
from urllib3.packages.six.moves.urllib.parse import urlsplit
-from urllib3.packages.six import binary_type
+from urllib3.packages.six import binary_type, ensure_str
log = logging.getLogger(__name__)
@@ -208,6 +208,12 @@ class TestingApp(RequestHandler):
headers = [("Connection", "keep-alive")]
return Response("Keeping alive", headers=headers)
+ def echo_params(self, request):
+ params = sorted(
+ [(ensure_str(k), ensure_str(v)) for k, v in request.params.items()]
+ )
+ return Response(repr(params))
+
def sleep(self, request):
"Sleep for a specified amount of ``seconds``"
# DO NOT USE THIS, IT'S DEPRECATED.

MODIFY
noxfile.py
noxfile.py
@@ -30,8 +30,7 @@ def tests_impl(session, extras="socks,secure,brotli"):
"pytest",
"-r",
"sx",
- "test",
- *session.posargs,
+ *(session.posargs or ("test/",)),
env={"PYTHONWARNINGS": "always::DeprecationWarning"}
)
session.run("coverage", "combine")

MODIFY
src/urllib3/connectionpool.py
src/urllib3/connectionpool.py
@@ -43,7 +43,13 @@ from .util.request import set_file_position
from .util.response import assert_header_parsing
from .util.retry import Retry
from .util.timeout import Timeout
-from .util.url import get_host, Url, _normalize_host as normalize_host
+from .util.url import (
+ get_host,
+ parse_url,
+ Url,
+ _normalize_host as normalize_host,
+ _encode_target,
+)
from .util.queue import LifoQueue
@@ -604,6 +610,12 @@ class HTTPConnectionPool(ConnectionPool, RequestMethods):
if assert_same_host and not self.is_same_host(url):
raise HostChangedError(self, url, retries)
+ # Ensure that the URL we're connecting to is properly encoded
+ if url.startswith("/"):
+ url = six.ensure_str(_encode_target(url))
+ else:
+ url = six.ensure_str(parse_url(url).url)
+
conn = None
# Track whether `conn` needs to be released before

MODIFY
src/urllib3/util/url.py
src/urllib3/util/url.py
@@ -55,6 +55,7 @@ IPV6_PAT = "(?:" + "|".join([x % _subs for x in _variations]) + ")"
ZONE_ID_PAT = "(?:%25|%)(?:[" + UNRESERVED_PAT + "]|%[a-fA-F0-9]{2})+"
IPV6_ADDRZ_PAT = r"\[" + IPV6_PAT + r"(?:" + ZONE_ID_PAT + r")?\]"
REG_NAME_PAT = r"(?:[^\[\]%:/?#]|%[a-fA-F0-9]{2})*"
+TARGET_RE = re.compile(r"^(/[^?]*)(?:\?([^#]+))?(?:#(.*))?$")
IPV4_RE = re.compile("^" + IPV4_PAT + "$")
IPV6_RE = re.compile("^" + IPV6_PAT + "$")
@@ -318,6 +319,22 @@ def _idna_encode(name):
return name.lower().encode("ascii")
+def _encode_target(target):
+ """Percent-encodes a request target so that there are no invalid characters"""
+ if not target.startswith("/"):
+ return target
+
+ path, query, fragment = TARGET_RE.match(target).groups()
+ target = _encode_invalid_chars(path, PATH_CHARS)
+ query = _encode_invalid_chars(query, QUERY_CHARS)
+ fragment = _encode_invalid_chars(fragment, FRAGMENT_CHARS)
+ if query is not None:
+ target += "?" + query
+ if fragment is not None:
+ target += "#" + target
+ return target
+
+
def parse_url(url):
"""
Given a url, return a parsed :class:`.Url` namedtuple. Best-effort is

MODIFY
test/with_dummyserver/test_connectionpool.py
test/with_dummyserver/test_connectionpool.py
@@ -717,6 +717,11 @@ class TestConnectionPool(HTTPDummyServerTestCase):
with pytest.raises(MaxRetryError):
pool.request("GET", "/test", retries=2)
+ def test_percent_encode_invalid_target_chars(self):
+ with HTTPConnectionPool(self.host, self.port) as pool:
+ r = pool.request("GET", "/echo_params?q=\r&k=\n \n")
+ assert r.data == b"[('k', '\\n \\n'), ('q', '\\r')]"
+
def test_source_address(self):
for addr, is_ipv6 in VALID_SOURCE_ADDRESSES:
if is_ipv6 and not HAS_IPV6_AND_DNS:

Percent-encode targets and URLs in ConnectionPool (#1673)