Add default user agent header
MODIFY
src/urllib3/connection.py
src/urllib3/connection.py
@@ -47,9 +47,10 @@ from .util.ssl_ import (
)
-from .util import connection
+from .util import connection, SUPPRESS_USER_AGENT
from ._collections import HTTPHeaderDict
+from ._version import __version__
log = logging.getLogger(__name__)
@@ -200,6 +201,14 @@ class HTTPConnection(_HTTPConnection, object):
return _HTTPConnection.putrequest(self, method, url, *args, **kwargs)
+ def request(self, method, url, body=None, headers=None):
+ headers = HTTPHeaderDict(headers if headers is not None else {})
+ if "user-agent" not in headers:
+ headers["User-Agent"] = _get_default_user_agent()
+ elif headers["user-agent"] == SUPPRESS_USER_AGENT:
+ del headers["user-agent"]
+ super(HTTPConnection, self).request(method, url, body=body, headers=headers)
+
def request_chunked(self, method, url, body=None, headers=None):
"""
Alternative to the common request method, which sends the
@@ -211,6 +220,10 @@ class HTTPConnection(_HTTPConnection, object):
self.putrequest(
method, url, skip_accept_encoding=skip_accept_encoding, skip_host=skip_host
)
+ if "user-agent" not in headers:
+ headers["User-Agent"] = _get_default_user_agent()
+ elif headers["user-agent"] == SUPPRESS_USER_AGENT:
+ del headers["user-agent"]
for header, value in headers.items():
self.putheader(header, value)
if "transfer-encoding" not in headers:
@@ -418,6 +431,10 @@ def _match_hostname(cert, asserted_hostname):
raise
+def _get_default_user_agent():
+ return "python-urllib3/%s" % __version__
+
+
if not ssl:
HTTPSConnection = DummyConnection # noqa: F811

MODIFY
src/urllib3/util/__init__.py
src/urllib3/util/__init__.py
@@ -2,7 +2,7 @@ from __future__ import absolute_import
# For backwards compatibility, provide imports that used to be here.
from .connection import is_connection_dropped
-from .request import make_headers
+from .request import make_headers, SUPPRESS_USER_AGENT
from .response import is_fp_closed
from .ssl_ import (
SSLContext,
@@ -43,4 +43,5 @@ __all__ = (
"ssl_wrap_socket",
"wait_for_read",
"wait_for_write",
+ "SUPPRESS_USER_AGENT",
)

MODIFY
src/urllib3/util/request.py
src/urllib3/util/request.py
@@ -4,6 +4,10 @@ from base64 import b64encode
from ..packages.six import b, integer_types
from ..exceptions import UnrewindableBodyError
+# Use an invalid User-Agent to represent suppressing of default user agent.
+# See https://tools.ietf.org/html/rfc7231#section-5.5.3 and
+# https://tools.ietf.org/html/rfc7230#section-3.2.6
+SUPPRESS_USER_AGENT = "@@@INVALID_USER_AGENT@@@"
ACCEPT_ENCODING = "gzip,deflate"
try:
import brotli as _unused_module_brotli # noqa: F401

MODIFY
test/with_dummyserver/test_chunked_transfer.py
test/with_dummyserver/test_chunked_transfer.py
@@ -4,6 +4,7 @@ import pytest
from urllib3 import HTTPConnectionPool
from urllib3.util.retry import Retry
+from urllib3.util import SUPPRESS_USER_AGENT
from dummyserver.testcase import SocketDummyServerTestCase, consume_socket
# Retry failed tests
@@ -78,16 +79,18 @@ class TestChunkedTransfer(SocketDummyServerTestCase):
def test_empty_iterable_body(self):
self._test_body([])
+ def _get_header_lines(self, prefix):
+ header_block = self.buffer.split(b"\r\n\r\n", 1)[0].lower()
+ header_lines = header_block.split(b"\r\n")[1:]
+ return [x for x in header_lines if x.startswith(prefix)]
+
def test_removes_duplicate_host_header(self):
self.start_chunked_handler()
chunks = ["foo", "bar", "", "bazzzzzzzzzzzzzzzzzzzzzz"]
with HTTPConnectionPool(self.host, self.port, retries=False) as pool:
pool.urlopen("GET", "/", chunks, headers={"Host": "test.org"}, chunked=True)
- header_block = self.buffer.split(b"\r\n\r\n", 1)[0].lower()
- header_lines = header_block.split(b"\r\n")[1:]
-
- host_headers = [x for x in header_lines if x.startswith(b"host")]
+ host_headers = self._get_header_lines(b"host")
assert len(host_headers) == 1
def test_provides_default_host_header(self):
@@ -96,12 +99,33 @@ class TestChunkedTransfer(SocketDummyServerTestCase):
with HTTPConnectionPool(self.host, self.port, retries=False) as pool:
pool.urlopen("GET", "/", chunks, chunked=True)
- header_block = self.buffer.split(b"\r\n\r\n", 1)[0].lower()
- header_lines = header_block.split(b"\r\n")[1:]
-
- host_headers = [x for x in header_lines if x.startswith(b"host")]
+ host_headers = self._get_header_lines(b"host")
assert len(host_headers) == 1
+ def test_provides_default_user_agent_header(self):
+ self.start_chunked_handler()
+ chunks = ["foo", "bar", "", "bazzzzzzzzzzzzzzzzzzzzzz"]
+ with HTTPConnectionPool(self.host, self.port, retries=False) as pool:
+ pool.urlopen("GET", "/", chunks, chunked=True)
+
+ ua_headers = self._get_header_lines(b"user-agent")
+ assert len(ua_headers) == 1
+
+ def test_remove_user_agent_header(self):
+ self.start_chunked_handler()
+ chunks = ["foo", "bar", "", "bazzzzzzzzzzzzzzzzzzzzzz"]
+ with HTTPConnectionPool(self.host, self.port, retries=False) as pool:
+ pool.urlopen(
+ "GET",
+ "/",
+ chunks,
+ headers={"User-Agent": SUPPRESS_USER_AGENT},
+ chunked=True,
+ )
+
+ ua_headers = self._get_header_lines(b"user-agent")
+ assert len(ua_headers) == 0
+
def test_preserve_chunked_on_retry_after(self):
self.chunked_requests = 0
self.socks = []

MODIFY
test/with_dummyserver/test_connectionpool.py
test/with_dummyserver/test_connectionpool.py
import io
+import json
import logging
import socket
import sys
@@ -11,6 +12,7 @@ import mock
from .. import TARPIT_HOST, VALID_SOURCE_ADDRESSES, INVALID_SOURCE_ADDRESSES
from ..port_helpers import find_unused_port
from urllib3 import encode_multipart_formdata, HTTPConnectionPool
+from urllib3.connection import _get_default_user_agent
from urllib3.exceptions import (
ConnectTimeoutError,
EmptyPoolError,
@@ -22,6 +24,7 @@ from urllib3.exceptions import (
)
from urllib3.packages.six import b, u
from urllib3.packages.six.moves.urllib.parse import urlencode
+from urllib3.util import SUPPRESS_USER_AGENT
from urllib3.util.retry import Retry, RequestHistory
from urllib3.util.timeout import Timeout
@@ -793,6 +796,63 @@ class TestConnectionPool(HTTPDummyServerTestCase):
response = pool.request("GET", "/echo_uri/seg0/../seg2")
assert response.data == b"/echo_uri/seg0/../seg2"
+ def test_default_user_agent_header(self):
+ """ ConnectionPool has a default user agent """
+ default_ua = _get_default_user_agent()
+ custom_ua = "I'm not a web scraper, what are you talking about?"
+ custom_ua2 = "Yet Another User Agent"
+ with HTTPConnectionPool(self.host, self.port) as pool:
+ # Use default user agent if no user agent was specified.
+ r = pool.request("GET", "/headers")
+ request_headers = json.loads(r.data.decode("utf8"))
+ assert request_headers.get("User-Agent") == _get_default_user_agent()
+
+ # Prefer the request user agent over the default.
+ headers = {"UsEr-AGENt": custom_ua}
+ r = pool.request("GET", "/headers", headers=headers)
+ request_headers = json.loads(r.data.decode("utf8"))
+ assert request_headers.get("User-Agent") == custom_ua
+
+ # Do not modify pool headers when using the default user agent.
+ pool_headers = {"foo": "bar"}
+ pool.headers = pool_headers
+ r = pool.request("GET", "/headers")
+ request_headers = json.loads(r.data.decode("utf8"))
+ assert request_headers.get("User-Agent") == default_ua
+ assert "User-Agent" not in pool_headers
+
+ pool.headers.update({"User-Agent": custom_ua2})
+ r = pool.request("GET", "/headers")
+ request_headers = json.loads(r.data.decode("utf8"))
+ assert request_headers.get("User-Agent") == custom_ua2
+
+ def test_no_user_agent_header(self):
+ """ ConnectionPool can suppress sending a user agent header """
+ custom_ua = "I'm not a web scraper, what are you talking about?"
+ with HTTPConnectionPool(self.host, self.port) as pool:
+ # Suppress user agent in the request headers.
+ no_ua_headers = {"User-Agent": SUPPRESS_USER_AGENT}
+ r = pool.request("GET", "/headers", headers=no_ua_headers)
+ request_headers = json.loads(r.data.decode("utf8"))
+ assert "User-Agent" not in request_headers
+ assert no_ua_headers["User-Agent"] == SUPPRESS_USER_AGENT
+
+ # Suppress user agent in the pool headers.
+ pool.headers = no_ua_headers
+ r = pool.request("GET", "/headers")
+ request_headers = json.loads(r.data.decode("utf8"))
+ assert "User-Agent" not in request_headers
+ assert no_ua_headers["User-Agent"] == SUPPRESS_USER_AGENT
+
+ # Request headers override pool headers.
+ pool_headers = {"User-Agent": custom_ua}
+ pool.headers = pool_headers
+ r = pool.request("GET", "/headers", headers=no_ua_headers)
+ request_headers = json.loads(r.data.decode("utf8"))
+ assert "User-Agent" not in request_headers
+ assert no_ua_headers["User-Agent"] == SUPPRESS_USER_AGENT
+ assert pool_headers.get("User-Agent") == custom_ua
+
class TestRetry(HTTPDummyServerTestCase):
def test_max_retry(self):

MODIFY
test/with_dummyserver/test_socketlevel.py
test/with_dummyserver/test_socketlevel.py
# rather than the socket level-ness of it.
from urllib3 import HTTPConnectionPool, HTTPSConnectionPool
from urllib3.poolmanager import proxy_from_url
+from urllib3.connection import _get_default_user_agent
from urllib3.exceptions import (
MaxRetryError,
ProxyError,
@@ -937,6 +938,7 @@ class TestProxyManager(SocketDummyServerTestCase):
b"Host: google.com",
b"Accept-Encoding: identity",
b"Accept: */*",
+ b"User-Agent: " + _get_default_user_agent().encode("utf-8"),
b"",
b"",
]
@@ -1433,9 +1435,9 @@ class TestHeaders(SocketDummyServerTestCase):
r = pool.request("GET", "/")
assert HEADERS == dict(r.headers.items()) # to preserve case sensitivity
- def test_headers_are_sent_with_the_original_case(self):
- headers = {"foo": "bar", "bAz": "quux"}
- parsed_headers = {}
+ def start_parsing_handler(self):
+ self.parsed_headers = OrderedDict()
+ self.received_headers = []
def socket_handler(listener):
sock = listener.accept()[0]
@@ -1444,11 +1446,13 @@ class TestHeaders(SocketDummyServerTestCase):
while not buf.endswith(b"\r\n\r\n"):
buf += sock.recv(65536)
- headers_list = [header for header in buf.split(b"\r\n")[1:] if header]
+ self.received_headers = [
+ header for header in buf.split(b"\r\n")[1:] if header
+ ]
- for header in headers_list:
+ for header in self.received_headers:
(key, value) = header.split(b": ")
- parsed_headers[key.decode("ascii")] = value.decode("ascii")
+ self.parsed_headers[key.decode("ascii")] = value.decode("ascii")
sock.send(
("HTTP/1.1 204 No Content\r\nContent-Length: 0\r\n\r\n").encode("utf-8")
@@ -1457,6 +1461,26 @@ class TestHeaders(SocketDummyServerTestCase):
sock.close()
self._start_server(socket_handler)
+
+ def test_headers_are_sent_with_the_original_case(self):
+ headers = {"foo": "bar", "bAz": "quux"}
+
+ self.start_parsing_handler()
+ expected_headers = {
+ "Accept-Encoding": "identity",
+ "Host": "{0}:{1}".format(self.host, self.port),
+ "User-Agent": _get_default_user_agent(),
+ }
+ expected_headers.update(headers)
+
+ with HTTPConnectionPool(self.host, self.port, retries=False) as pool:
+ pool.request("GET", "/", headers=HTTPHeaderDict(headers))
+ assert expected_headers == self.parsed_headers
+
+ def test_ua_header_can_be_overridden(self):
+ headers = {"uSeR-AgENt": "Definitely not urllib3!"}
+
+ self.start_parsing_handler()
expected_headers = {
"Accept-Encoding": "identity",
"Host": "{0}:{1}".format(self.host, self.port),
@@ -1465,7 +1489,7 @@ class TestHeaders(SocketDummyServerTestCase):
with HTTPConnectionPool(self.host, self.port, retries=False) as pool:
pool.request("GET", "/", headers=HTTPHeaderDict(headers))
- assert expected_headers == parsed_headers
+ assert expected_headers == self.parsed_headers
def test_request_headers_are_sent_in_the_original_order(self):
# NOTE: Probability this test gives a false negative is 1/(K!)
@@ -1477,69 +1501,26 @@ class TestHeaders(SocketDummyServerTestCase):
(u"X-Header-%d" % i, str(i)) for i in reversed(range(K))
]
- actual_request_headers = []
+ def filter_non_x_headers(d):
+ return [(k, v) for (k, v) in d.items() if k.startswith("X-Header-")]
- def socket_handler(listener):
- sock = listener.accept()[0]
+ request_headers = OrderedDict()
- buf = b""
- while not buf.endswith(b"\r\n\r\n"):
- buf += sock.recv(65536)
-
- headers_list = [header for header in buf.split(b"\r\n")[1:] if header]
-
- for header in headers_list:
- (key, value) = header.split(b": ")
- if not key.decode("ascii").startswith(u"X-Header-"):
- continue
- actual_request_headers.append(
- (key.decode("ascii"), value.decode("ascii"))
- )
-
- sock.send(
- (u"HTTP/1.1 204 No Content\r\nContent-Length: 0\r\n\r\n").encode(
- "ascii"
- )
- )
-
- sock.close()
-
- self._start_server(socket_handler)
+ self.start_parsing_handler()
with HTTPConnectionPool(self.host, self.port, retries=False) as pool:
pool.request("GET", "/", headers=OrderedDict(expected_request_headers))
- assert expected_request_headers == actual_request_headers
+ request_headers = filter_non_x_headers(self.parsed_headers)
+ assert expected_request_headers == request_headers
@resolvesLocalhostFQDN
def test_request_host_header_ignores_fqdn_dot(self):
-
- received_headers = []
-
- def socket_handler(listener):
- sock = listener.accept()[0]
-
- buf = b""
- while not buf.endswith(b"\r\n\r\n"):
- buf += sock.recv(65536)
-
- for header in buf.split(b"\r\n")[1:]:
- if header:
- received_headers.append(header)
-
- sock.send(
- (u"HTTP/1.1 204 No Content\r\nContent-Length: 0\r\n\r\n").encode(
- "ascii"
- )
- )
-
- sock.close()
-
- self._start_server(socket_handler)
+ self.start_parsing_handler()
with HTTPConnectionPool(self.host + ".", self.port, retries=False) as pool:
pool.request("GET", "/")
self.assert_header_received(
- received_headers, "Host", "%s:%s" % (self.host, self.port)
+ self.received_headers, "Host", "%s:%s" % (self.host, self.port)
)
def test_response_headers_are_returned_in_the_original_order(self):

Add default user agent header (#1750)

Co-authored-by: hodbn <hodbn@users.noreply.github.com>