Local address support.

httpcore/_async/connection.py

@@ -22,11 +22,13 @@ def __init__(
  http2: bool = False,
  ssl_context: SSLContext = None,
  socket: AsyncSocketStream = None,
+ local_addr: bytes = None,
  ):
  self.origin = origin
  self.http2 = http2
  self.ssl_context = SSLContext() if ssl_context is None else ssl_context
  self.socket = socket
+ self.local_addr = local_addr
 
  if self.http2:
  self.ssl_context.set_alpn_protocols(["http/1.1", "h2"])
@@ -97,7 +99,7 @@ async def _open_socket(self, timeout: TimeoutDict = None) -> AsyncSocketStream:
  ssl_context = self.ssl_context if scheme == b"https" else None
  try:
  return await self.backend.open_tcp_stream(
- hostname, port, ssl_context, timeout
+ hostname, port, ssl_context, timeout, self.local_addr
  )
  except Exception:
  self.connect_failed = True

httpcore/_async/connection_pool.py

@@ -76,6 +76,7 @@ class AsyncConnectionPool(AsyncHTTPTransport):
  * **keepalive_expiry** - `Optional[float]` - The maximum time to allow
  before closing a keep-alive connection.
  * **http2** - `bool` - Enable HTTP/2 support.
+ * **local_addr** - `Optional[bytes]` - Local address to connect from.
  """
 
  def __init__(
@@ -85,12 +86,14 @@ def __init__(
  max_keepalive: int = None,
  keepalive_expiry: float = None,
  http2: bool = False,
+ local_addr: bytes = None,
  ):
  self._ssl_context = SSLContext() if ssl_context is None else ssl_context
  self._max_connections = max_connections
  self._max_keepalive = max_keepalive
  self._keepalive_expiry = keepalive_expiry
  self._http2 = http2
+ self._local_addr = local_addr
  self._connections: Dict[Origin, Set[AsyncHTTPConnection]] = {}
  self._thread_lock = ThreadLock()
  self._backend = AutoBackend()
@@ -146,7 +149,10 @@ async def request(
 
  if connection is None:
  connection = AsyncHTTPConnection(
- origin=origin, http2=self._http2, ssl_context=self._ssl_context,
+ origin=origin,
+ http2=self._http2,
+ ssl_context=self._ssl_context,
+ local_addr=self._local_addr,
  )
  logger.trace("created connection=%r", connection)
  await self._add_to_pool(connection, timeout=timeout)

httpcore/_backends/asyncio.py

@@ -225,13 +225,18 @@ async def open_tcp_stream(
  port: int,
  ssl_context: Optional[SSLContext],
  timeout: TimeoutDict,
+ local_addr: Optional[bytes],
  ) -> SocketStream:
  host = hostname.decode("ascii")
  connect_timeout = timeout.get("connect")
  exc_map = {asyncio.TimeoutError: ConnectTimeout, OSError: ConnectError}
  with map_exceptions(exc_map):
+ local_addrport = None if local_addr is None else (local_addr, 0)
  stream_reader, stream_writer = await asyncio.wait_for(
- asyncio.open_connection(host, port, ssl=ssl_context), connect_timeout,
+ asyncio.open_connection(
+ host, port, ssl=ssl_context, local_addr=local_addrport
+ ),
+ connect_timeout,
  )
  return SocketStream(
  stream_reader=stream_reader, stream_writer=stream_writer

httpcore/_backends/auto.py

@@ -34,8 +34,11 @@ async def open_tcp_stream(
  port: int,
  ssl_context: Optional[SSLContext],
  timeout: TimeoutDict,
+ local_addr: Optional[bytes],
  ) -> AsyncSocketStream:
- return await self.backend.open_tcp_stream(hostname, port, ssl_context, timeout)
+ return await self.backend.open_tcp_stream(
+ hostname, port, ssl_context, timeout, local_addr=local_addr
+ )
 
  def create_lock(self) -> AsyncLock:
  return self.backend.create_lock()

httpcore/_backends/base.py

@@ -76,6 +76,7 @@ async def open_tcp_stream(
  port: int,
  ssl_context: Optional[SSLContext],
  timeout: TimeoutDict,
+ local_addr: Optional[bytes],
  ) -> AsyncSocketStream:
  raise NotImplementedError() # pragma: no cover
 

httpcore/_backends/sync.py

@@ -128,13 +128,17 @@ def open_tcp_stream(
  port: int,
  ssl_context: Optional[SSLContext],
  timeout: TimeoutDict,
+ local_addr: Optional[bytes],
  ) -> SyncSocketStream:
  address = (hostname.decode("ascii"), port)
  connect_timeout = timeout.get("connect")
  exc_map = {socket.timeout: ConnectTimeout, socket.error: ConnectError}
 
  with map_exceptions(exc_map):
- sock = socket.create_connection(address, connect_timeout)
+ local_addrport = None
+ if local_addr:
+ local_addrport = (local_addr, 0)
+ sock = socket.create_connection(address, connect_timeout, local_addrport) # type: ignore
  if ssl_context is not None:
  sock = ssl_context.wrap_socket(
  sock, server_hostname=hostname.decode("ascii")

httpcore/_backends/trio.py

@@ -143,7 +143,12 @@ async def open_tcp_stream(
  port: int,
  ssl_context: Optional[SSLContext],
  timeout: TimeoutDict,
+ local_addr: Optional[bytes],
  ) -> AsyncSocketStream:
+ # trio doesn't currently support specifying the local address; it will
+ # as of 0.16.1.
+ if local_addr:
+ raise NotImplementedError()
  connect_timeout = none_as_inf(timeout.get("connect"))
  exc_map = {
  trio.TooSlowError: ConnectTimeout,

httpcore/_sync/connection.py

@@ -22,11 +22,13 @@ def __init__(
  http2: bool = False,
  ssl_context: SSLContext = None,
  socket: SyncSocketStream = None,
+ local_addr: bytes = None,
  ):
  self.origin = origin
  self.http2 = http2
  self.ssl_context = SSLContext() if ssl_context is None else ssl_context
  self.socket = socket
+ self.local_addr = local_addr
 
  if self.http2:
  self.ssl_context.set_alpn_protocols(["http/1.1", "h2"])
@@ -97,7 +99,7 @@ def _open_socket(self, timeout: TimeoutDict = None) -> SyncSocketStream:
  ssl_context = self.ssl_context if scheme == b"https" else None
  try:
  return self.backend.open_tcp_stream(
- hostname, port, ssl_context, timeout
+ hostname, port, ssl_context, timeout, self.local_addr
  )
  except Exception:
  self.connect_failed = True

httpcore/_sync/connection_pool.py

@@ -76,6 +76,7 @@ class SyncConnectionPool(SyncHTTPTransport):
  * **keepalive_expiry** - `Optional[float]` - The maximum time to allow
  before closing a keep-alive connection.
  * **http2** - `bool` - Enable HTTP/2 support.
+ * **local_addr** - `Optional[bytes]` - Local address to connect from.
  """
 
  def __init__(
@@ -85,12 +86,14 @@ def __init__(
  max_keepalive: int = None,
  keepalive_expiry: float = None,
  http2: bool = False,
+ local_addr: bytes = None,
  ):
  self._ssl_context = SSLContext() if ssl_context is None else ssl_context
  self._max_connections = max_connections
  self._max_keepalive = max_keepalive
  self._keepalive_expiry = keepalive_expiry
  self._http2 = http2
+ self._local_addr = local_addr
  self._connections: Dict[Origin, Set[SyncHTTPConnection]] = {}
  self._thread_lock = ThreadLock()
  self._backend = SyncBackend()
@@ -146,7 +149,10 @@ def request(
 
  if connection is None:
  connection = SyncHTTPConnection(
- origin=origin, http2=self._http2, ssl_context=self._ssl_context,
+ origin=origin,
+ http2=self._http2,
+ ssl_context=self._ssl_context,
+ local_addr=self._local_addr,
  )
  logger.trace("created connection=%r", connection)
  self._add_to_pool(connection, timeout=timeout)

tests/async_tests/test_interfaces.py

@@ -212,6 +212,25 @@ async def test_http_proxy(proxy_server: URL, proxy_mode: str) -> None:
  assert reason == b"OK"
 
 
+@pytest.mark.parametrize("local_addr", [b"0.0.0.0"])
+@pytest.mark.asyncio
+# This doesn't run with trio, since trio doesn't support local_addr.
+async def test_http_request_local_addr(local_addr: str) -> None:
+ async with httpcore.AsyncConnectionPool(local_addr=local_addr) as http:
+ method = b"GET"
+ url = (b"http", b"example.org", 80, b"/")
+ headers = [(b"host", b"example.org")]
+ http_version, status_code, reason, headers, stream = await http.request(
+ method, url, headers
+ )
+ body = await read_body(stream)
+
+ assert http_version == b"HTTP/1.1"
+ assert status_code == 200
+ assert reason == b"OK"
+ assert len(http._connections[url[:3]]) == 1 # type: ignore
+
+
 # mitmproxy does not support forwarding HTTPS requests
 @pytest.mark.parametrize("proxy_mode", ["DEFAULT", "TUNNEL_ONLY"])
 @pytest.mark.usefixtures("async_environment")

tests/sync_tests/test_interfaces.py

@@ -212,6 +212,25 @@ def test_http_proxy(proxy_server: URL, proxy_mode: str) -> None:
  assert reason == b"OK"
 
 
+@pytest.mark.parametrize("local_addr", [b"0.0.0.0"])
+
+# This doesn't run with trio, since trio doesn't support local_addr.
+def test_http_request_local_addr(local_addr: str) -> None:
+ with httpcore.SyncConnectionPool(local_addr=local_addr) as http:
+ method = b"GET"
+ url = (b"http", b"example.org", 80, b"/")
+ headers = [(b"host", b"example.org")]
+ http_version, status_code, reason, headers, stream = http.request(
+ method, url, headers
+ )
+ body = read_body(stream)
+
+ assert http_version == b"HTTP/1.1"
+ assert status_code == 200
+ assert reason == b"OK"
+ assert len(http._connections[url[:3]]) == 1 # type: ignore
+
+
 # mitmproxy does not support forwarding HTTPS requests
 @pytest.mark.parametrize("proxy_mode", ["DEFAULT", "TUNNEL_ONLY"])