Add proxy_auth argument to HTTPProxy

httpcore/_async/http_proxy.py

@@ -1,8 +1,17 @@
 import ssl
+from base64 import b64encode
 from typing import List, Mapping, Optional, Sequence, Tuple, Union
 
 from .._exceptions import ProxyError
-from .._models import URL, Origin, Request, Response, enforce_headers, enforce_url
+from .._models import (
+    URL,
+    Origin,
+    Request,
+    Response,
+    enforce_bytes,
+    enforce_headers,
+    enforce_url,
+)
 from .._ssl import default_ssl_context
 from .._synchronization import AsyncLock
 from .._trace import Trace
@@ -35,6 +44,11 @@ def merge_headers(
     return default_headers + override_headers
 
 
+def build_auth_header(username: bytes, password: bytes) -> bytes:
+    userpass = username + b":" + password
+    return b"Basic " + b64encode(userpass)
+
+
 class AsyncHTTPProxy(AsyncConnectionPool):
     """
     A connection pool that sends requests via an HTTP proxy.
@@ -43,6 +57,7 @@ class AsyncHTTPProxy(AsyncConnectionPool):
     def __init__(
         self,
         proxy_url: Union[URL, bytes, str],
+        proxy_auth: Tuple[Union[bytes, str], Union[bytes, str]] = None,
         proxy_headers: Union[HeadersAsMapping, HeadersAsSequence] = None,
         ssl_context: ssl.SSLContext = None,
         max_connections: Optional[int] = 10,
@@ -61,6 +76,8 @@ def __init__(
         Parameters:
             proxy_url: The URL to use when connecting to the proxy server.
                 For example `"http://127.0.0.1:8080/"`.
+            proxy_auth: Any proxy authentication as a two-tuple of
+                (username, password). May be either bytes or ascii-only str.
             proxy_headers: Any HTTP headers to use for the proxy requests.
                 For example `{"Proxy-Authorization": "Basic <username>:<password>"}`.
             ssl_context: An SSL context to use for verifying connections.
@@ -102,6 +119,13 @@ def __init__(
         self._ssl_context = ssl_context
         self._proxy_url = enforce_url(proxy_url, name="proxy_url")
         self._proxy_headers = enforce_headers(proxy_headers, name="proxy_headers")
+        if proxy_auth is not None:
+            username = enforce_bytes(proxy_auth[0], name="proxy_auth")
+            password = enforce_bytes(proxy_auth[1], name="proxy_auth")
+            authorization = build_auth_header(username, password)
+            self._proxy_headers = [
+                (b"Proxy-Authorization", authorization)
+            ] + self._proxy_headers
 
     def create_connection(self, origin: Origin) -> AsyncConnectionInterface:
         if origin.scheme == b"http":

httpcore/_sync/http_proxy.py

@@ -1,8 +1,17 @@
 import ssl
+from base64 import b64encode
 from typing import List, Mapping, Optional, Sequence, Tuple, Union
 
 from .._exceptions import ProxyError
-from .._models import URL, Origin, Request, Response, enforce_headers, enforce_url
+from .._models import (
+    URL,
+    Origin,
+    Request,
+    Response,
+    enforce_bytes,
+    enforce_headers,
+    enforce_url,
+)
 from .._ssl import default_ssl_context
 from .._synchronization import Lock
 from .._trace import Trace
@@ -35,6 +44,11 @@ def merge_headers(
     return default_headers + override_headers
 
 
+def build_auth_header(username: bytes, password: bytes) -> bytes:
+    userpass = username + b":" + password
+    return b"Basic " + b64encode(userpass)
+
+
 class HTTPProxy(ConnectionPool):
     """
     A connection pool that sends requests via an HTTP proxy.
@@ -43,6 +57,7 @@ class HTTPProxy(ConnectionPool):
     def __init__(
         self,
         proxy_url: Union[URL, bytes, str],
+        proxy_auth: Tuple[Union[bytes, str], Union[bytes, str]] = None,
         proxy_headers: Union[HeadersAsMapping, HeadersAsSequence] = None,
         ssl_context: ssl.SSLContext = None,
         max_connections: Optional[int] = 10,
@@ -61,6 +76,8 @@ def __init__(
         Parameters:
             proxy_url: The URL to use when connecting to the proxy server.
                 For example `"http://127.0.0.1:8080/"`.
+            proxy_auth: Any proxy authentication as a two-tuple of
+                (username, password). May be either bytes or ascii-only str.
             proxy_headers: Any HTTP headers to use for the proxy requests.
                 For example `{"Proxy-Authorization": "Basic <username>:<password>"}`.
             ssl_context: An SSL context to use for verifying connections.
@@ -102,6 +119,13 @@ def __init__(
         self._ssl_context = ssl_context
         self._proxy_url = enforce_url(proxy_url, name="proxy_url")
         self._proxy_headers = enforce_headers(proxy_headers, name="proxy_headers")
+        if proxy_auth is not None:
+            username = enforce_bytes(proxy_auth[0], name="proxy_auth")
+            password = enforce_bytes(proxy_auth[1], name="proxy_auth")
+            authorization = build_auth_header(username, password)
+            self._proxy_headers = [
+                (b"Proxy-Authorization", authorization)
+            ] + self._proxy_headers
 
     def create_connection(self, origin: Origin) -> ConnectionInterface:
         if origin.scheme == b"http":

tests/_async/test_http_proxy.py

@@ -82,7 +82,6 @@ async def test_proxy_tunneling():
 
     async with AsyncHTTPProxy(
         proxy_url="http://localhost:8080/",
-        max_connections=10,
         network_backend=network_backend,
     ) as proxy:
         # Sending an intial request, which once complete will return to the pool, IDLE.
@@ -169,7 +168,6 @@ async def test_proxy_tunneling_http2():
 
     async with AsyncHTTPProxy(
         proxy_url="http://localhost:8080/",
-        max_connections=10,
         network_backend=network_backend,
         http2=True,
     ) as proxy:
@@ -219,10 +217,43 @@ async def test_proxy_tunneling_with_403():
 
     async with AsyncHTTPProxy(
         proxy_url="http://localhost:8080/",
-        max_connections=10,
         network_backend=network_backend,
     ) as proxy:
         with pytest.raises(ProxyError) as exc_info:
             await proxy.request("GET", "https://example.com/")
         assert str(exc_info.value) == "403 Permission Denied"
         assert not proxy.connections
+
+
+@pytest.mark.anyio
+async def test_proxy_tunneling_with_auth():
+    """
+    Send an authenticated HTTPS request via a proxy.
+    """
+    network_backend = AsyncMockBackend(
+        [
+            # The initial response to the proxy CONNECT
+            b"HTTP/1.1 200 OK\r\n\r\n",
+            # The actual response from the remote server
+            b"HTTP/1.1 200 OK\r\n",
+            b"Content-Type: plain/text\r\n",
+            b"Content-Length: 13\r\n",
+            b"\r\n",
+            b"Hello, world!",
+        ]
+    )
+
+    async with AsyncHTTPProxy(
+        proxy_url="http://localhost:8080/",
+        proxy_auth=("username", "password"),
+        network_backend=network_backend,
+    ) as proxy:
+        response = await proxy.request("GET", "https://example.com/")
+        assert response.status == 200
+        assert response.content == b"Hello, world!"
+
+        # Dig into this private property as a cheap lazy way of
+        # checking that the proxy header is set correctly.
+        assert proxy._proxy_headers == [  # type: ignore
+            (b"Proxy-Authorization", b"Basic dXNlcm5hbWU6cGFzc3dvcmQ=")
+        ]

tests/_sync/test_http_proxy.py

@@ -82,7 +82,6 @@ def test_proxy_tunneling():
 
     with HTTPProxy(
         proxy_url="http://localhost:8080/",
-        max_connections=10,
         network_backend=network_backend,
     ) as proxy:
         # Sending an intial request, which once complete will return to the pool, IDLE.
@@ -169,7 +168,6 @@ def test_proxy_tunneling_http2():
 
     with HTTPProxy(
         proxy_url="http://localhost:8080/",
-        max_connections=10,
         network_backend=network_backend,
         http2=True,
     ) as proxy:
@@ -219,10 +217,43 @@ def test_proxy_tunneling_with_403():
 
     with HTTPProxy(
         proxy_url="http://localhost:8080/",
-        max_connections=10,
         network_backend=network_backend,
     ) as proxy:
         with pytest.raises(ProxyError) as exc_info:
             proxy.request("GET", "https://example.com/")
         assert str(exc_info.value) == "403 Permission Denied"
         assert not proxy.connections
+
+
+
+def test_proxy_tunneling_with_auth():
+    """
+    Send an authenticated HTTPS request via a proxy.
+    """
+    network_backend = MockBackend(
+        [
+            # The initial response to the proxy CONNECT
+            b"HTTP/1.1 200 OK\r\n\r\n",
+            # The actual response from the remote server
+            b"HTTP/1.1 200 OK\r\n",
+            b"Content-Type: plain/text\r\n",
+            b"Content-Length: 13\r\n",
+            b"\r\n",
+            b"Hello, world!",
+        ]
+    )
+
+    with HTTPProxy(
+        proxy_url="http://localhost:8080/",
+        proxy_auth=("username", "password"),
+        network_backend=network_backend,
+    ) as proxy:
+        response = proxy.request("GET", "https://example.com/")
+        assert response.status == 200
+        assert response.content == b"Hello, world!"
+
+        # Dig into this private property as a cheap lazy way of
+        # checking that the proxy header is set correctly.
+        assert proxy._proxy_headers == [  # type: ignore
+            (b"Proxy-Authorization", b"Basic dXNlcm5hbWU6cGFzc3dvcmQ=")
+        ]