Local address support

httpcore/_async/connection.py

@@ -22,11 +22,13 @@ def __init__(
  http2: bool = False,
  ssl_context: SSLContext = None,
  socket: AsyncSocketStream = None,
+ local_address: str = None,
  ):
  self.origin = origin
  self.http2 = http2
  self.ssl_context = SSLContext() if ssl_context is None else ssl_context
  self.socket = socket
+ self.local_address = local_address
 
  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, local_address=self.local_address
  )
  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_address** - `Optional[str]` - 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_address: str = 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_address = local_address
  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_address=self._local_address,
  )
  logger.trace("created connection=%r", connection)
  await self._add_to_pool(connection, timeout=timeout)

httpcore/_backends/asyncio.py

@@ -225,13 +225,20 @@ async def open_tcp_stream(
  port: int,
  ssl_context: Optional[SSLContext],
  timeout: TimeoutDict,
+ *,
+ local_address: Optional[str],
  ) -> SocketStream:
  host = hostname.decode("ascii")
  connect_timeout = timeout.get("connect")
+ local_addr = None if local_address is None else (local_address, 0)
+
  exc_map = {asyncio.TimeoutError: ConnectTimeout, OSError: ConnectError}
  with map_exceptions(exc_map):
  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_addr
+ ),
+ connect_timeout,
  )
  return SocketStream(
  stream_reader=stream_reader, stream_writer=stream_writer

httpcore/_backends/auto.py

@@ -34,8 +34,12 @@ async def open_tcp_stream(
  port: int,
  ssl_context: Optional[SSLContext],
  timeout: TimeoutDict,
+ *,
+ local_address: Optional[str],
  ) -> 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_address=local_address
+ )
 
  def create_lock(self) -> AsyncLock:
  return self.backend.create_lock()

httpcore/_backends/base.py

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

httpcore/_backends/sync.py

@@ -128,13 +128,19 @@ def open_tcp_stream(
  port: int,
  ssl_context: Optional[SSLContext],
  timeout: TimeoutDict,
+ *,
+ local_address: Optional[str],
  ) -> SyncSocketStream:
  address = (hostname.decode("ascii"), port)
  connect_timeout = timeout.get("connect")
+ source_address = None if local_address is None else (local_address, 0)
  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_address:
+ local_addrport = (local_address, 0)
+ sock = socket.create_connection(address, connect_timeout, source_address=source_address) # type: ignore
  if ssl_context is not None:
  sock = ssl_context.wrap_socket(
  sock, server_hostname=hostname.decode("ascii")

httpcore/_backends/trio.py

@@ -143,16 +143,24 @@ async def open_tcp_stream(
  port: int,
  ssl_context: Optional[SSLContext],
  timeout: TimeoutDict,
+ *,
+ local_address: Optional[str],
  ) -> AsyncSocketStream:
  connect_timeout = none_as_inf(timeout.get("connect"))
+ # Trio will support local_address from 0.16.1 onwards.
+ # We only include the keyword argument if a local_address
+ # argument has been passed.
+ kwargs: dict = {} if local_address is None else {"local_address": local_address}
  exc_map = {
  trio.TooSlowError: ConnectTimeout,
  trio.BrokenResourceError: ConnectError,
  }
 
  with map_exceptions(exc_map):
  with trio.fail_after(connect_timeout):
- stream: trio.abc.Stream = await trio.open_tcp_stream(hostname, port)
+ stream: trio.abc.Stream = await trio.open_tcp_stream(
+ hostname, port, **kwargs
+ )
 
  if ssl_context is not None:
  stream = trio.SSLStream(

httpcore/_sync/connection.py

@@ -22,11 +22,13 @@ def __init__(
  http2: bool = False,
  ssl_context: SSLContext = None,
  socket: SyncSocketStream = None,
+ local_address: str = None,
  ):
  self.origin = origin
  self.http2 = http2
  self.ssl_context = SSLContext() if ssl_context is None else ssl_context
  self.socket = socket
+ self.local_address = local_address
 
  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, local_address=self.local_address
  )
  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_address** - `Optional[str]` - 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_address: str = 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_address = local_address
  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_address=self._local_address,
  )
  logger.trace("created connection=%r", connection)
  self._add_to_pool(connection, timeout=timeout)

tests/async_tests/test_interfaces.py

@@ -212,6 +212,24 @@ async def test_http_proxy(proxy_server: URL, proxy_mode: str) -> None:
  assert reason == b"OK"
 
 
+@pytest.mark.asyncio
+# This doesn't run with trio, since trio doesn't support local_address.
+async def test_http_request_local_address() -> None:
+ async with httpcore.AsyncConnectionPool(local_address="0.0.0.0") 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,24 @@ def test_http_proxy(proxy_server: URL, proxy_mode: str) -> None:
  assert reason == b"OK"
 
 
+
+# This doesn't run with trio, since trio doesn't support local_address.
+def test_http_request_local_address() -> None:
+ with httpcore.SyncConnectionPool(local_address="0.0.0.0") 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"])