Fix connecting to npipe://, tcp://, and unix:// urls

CHANGES/8632.bugfix.rst

@@ -0,0 +1 @@
+Fixed connecting to ``npipe://`` and ``unix://`` urls -- by :user:`bdraco`.

aiohttp/client.py

@@ -80,7 +80,13 @@
     ClientWebSocketResponse,
     ClientWSTimeout,
 )
-from .connector import BaseConnector, NamedPipeConnector, TCPConnector, UnixConnector
+from .connector import (
+    HTTP_AND_EMPTY_SCHEMA_SET,
+    BaseConnector,
+    NamedPipeConnector,
+    TCPConnector,
+    UnixConnector,
+)
 from .cookiejar import CookieJar
 from .helpers import (
     _SENTINEL,
@@ -210,9 +216,7 @@ class ClientTimeout:
 
 # https://www.rfc-editor.org/rfc/rfc9110#section-9.2.2
 IDEMPOTENT_METHODS = frozenset({"GET", "HEAD", "OPTIONS", "TRACE", "PUT", "DELETE"})
-HTTP_SCHEMA_SET = frozenset({"http", "https", ""})
-WS_SCHEMA_SET = frozenset({"ws", "wss"})
-ALLOWED_PROTOCOL_SCHEMA_SET = HTTP_SCHEMA_SET | WS_SCHEMA_SET
+
 
 _RetType = TypeVar("_RetType")
 _CharsetResolver = Callable[[ClientResponse, bytes], str]
@@ -455,7 +459,8 @@ async def _request(
         except ValueError as e:
             raise InvalidUrlClientError(str_or_url) from e
 
-        if url.scheme not in ALLOWED_PROTOCOL_SCHEMA_SET:
+        assert self._connector is not None
+        if url.scheme not in self._connector.allowed_protocol_schema_set:
             raise NonHttpUrlClientError(url)
 
         skip_headers = set(self._skip_auto_headers)
@@ -586,7 +591,6 @@ async def _request(
                             real_timeout.connect,
                             ceil_threshold=real_timeout.ceil_threshold,
                         ):
-                            assert self._connector is not None
                             conn = await self._connector.connect(
                                 req, traces=traces, timeout=real_timeout
                             )
@@ -682,7 +686,7 @@ async def _request(
                             ) from e
 
                         scheme = parsed_redirect_url.scheme
-                        if scheme not in HTTP_SCHEMA_SET:
+                        if scheme not in HTTP_AND_EMPTY_SCHEMA_SET:
                             resp.close()
                             raise NonHttpUrlRedirectClientError(r_url)
                         elif not scheme:

aiohttp/connector.py

@@ -9,6 +9,7 @@
 import warnings
 from collections import defaultdict, deque
 from contextlib import suppress
+from functools import cached_property
 from http import HTTPStatus
 from http.cookies import SimpleCookie
 from itertools import cycle, islice
@@ -21,6 +22,7 @@
     Callable,
     DefaultDict,
     Dict,
+    FrozenSet,
     Iterator,
     List,
     Literal,
@@ -63,6 +65,23 @@
     SSLContext = object  # type: ignore[misc,assignment]
 
 
+EMPTY_SCHEMA_SET = frozenset({""})
+HTTP_SCHEMA_SET = frozenset({"http", "https"})
+WS_SCHEMA_SET = frozenset({"ws", "wss"})
+TCP_PROTOCOL_SCHEMA_SET = frozenset({"tcp"})
+UNIX_PROTOCOL_SCHEMA_SET = frozenset({"unix"})
+NAMED_PIPE_PROTOCOL_SCHEMA_SET = frozenset({"npipe"})
+
+HTTP_AND_EMPTY_SCHEMA_SET = HTTP_SCHEMA_SET | EMPTY_SCHEMA_SET
+HIGH_LEVEL_SCHEMA_SET = HTTP_AND_EMPTY_SCHEMA_SET | WS_SCHEMA_SET
+ALLOWED_PROTOCOL_SCHEMA_SET = (
+    HIGH_LEVEL_SCHEMA_SET
+    | TCP_PROTOCOL_SCHEMA_SET
+    | UNIX_PROTOCOL_SCHEMA_SET
+    | NAMED_PIPE_PROTOCOL_SCHEMA_SET
+)
+
+
 __all__ = ("BaseConnector", "TCPConnector", "UnixConnector", "NamedPipeConnector")
 
 
@@ -243,6 +262,14 @@ def __init__(
         self._cleanup_closed_transports: List[Optional[asyncio.Transport]] = []
         self._cleanup_closed()
 
+    @cached_property
+    def allowed_protocol_schema_set(self) -> FrozenSet[str]:
+        """Return allowed protocol schema set.
+
+        By default we allow all protocols for backwards compatibility.
+        """
+        return ALLOWED_PROTOCOL_SCHEMA_SET
+
     def __del__(self, _warnings: Any = warnings) -> None:
         if self._closed:
             return
@@ -790,6 +817,11 @@ def _close_immediately(self) -> List["asyncio.Future[None]"]:
             ev.cancel()
         return super()._close_immediately()
 
+    @cached_property
+    def allowed_protocol_schema_set(self) -> FrozenSet[str]:
+        """Return allowed protocol schema set."""
+        return HIGH_LEVEL_SCHEMA_SET | TCP_PROTOCOL_SCHEMA_SET
+
     @property
     def family(self) -> int:
         """Socket family like AF_INET."""
@@ -1357,6 +1389,11 @@ def __init__(
         )
         self._path = path
 
+    @cached_property
+    def allowed_protocol_schema_set(self) -> FrozenSet[str]:
+        """Return allowed protocol schema set."""
+        return HIGH_LEVEL_SCHEMA_SET | UNIX_PROTOCOL_SCHEMA_SET
+
     @property
     def path(self) -> str:
         """Path to unix socket."""
@@ -1417,6 +1454,11 @@ def __init__(
             )
         self._path = path
 
+    @cached_property
+    def allowed_protocol_schema_set(self) -> FrozenSet[str]:
+        """Return allowed protocol schema set."""
+        return HIGH_LEVEL_SCHEMA_SET | NAMED_PIPE_PROTOCOL_SCHEMA_SET
+
     @property
     def path(self) -> str:
         """Path to the named pipe."""

tests/test_client_session.py

@@ -467,7 +467,7 @@ async def create_connection(req, traces, timeout):
         c.__del__()
 
 
-@pytest.mark.parametrize("protocol", ["http", "https", "ws", "wss"])
+@pytest.mark.parametrize("protocol", ["http", "https", "npipe", "ws", "wss", "unix"])
 async def test_ws_connect_allowed_protocols(
     create_session: Any,
     create_mocked_conn: Any,
@@ -482,15 +482,17 @@ async def test_ws_connect_allowed_protocols(
         hdrs.CONNECTION: "upgrade",
         hdrs.SEC_WEBSOCKET_ACCEPT: ws_key,
     }
-    resp.url = URL(f"{protocol}://example.com")
+    resp.url = URL(f"{protocol}://example")
     resp.cookies = SimpleCookie()
     resp.start = mock.AsyncMock()
 
     req = mock.create_autospec(aiohttp.ClientRequest, spec_set=True)
     req_factory = mock.Mock(return_value=req)
     req.send = mock.AsyncMock(return_value=resp)
+    # BaseConnector allows all protocols by default
+    connector = BaseConnector()
 
-    session = await create_session(request_class=req_factory)
+    session = await create_session(connector=connector, request_class=req_factory)
 
     connections = []
     original_connect = session._connector.connect
@@ -510,7 +512,7 @@ async def create_connection(req, traces, timeout):
         "aiohttp.client.os"
     ) as m_os:
         m_os.urandom.return_value = key_data
-        await session.ws_connect(f"{protocol}://example.com")
+        await session.ws_connect(f"{protocol}://example")
 
     # normally called during garbage collection.  triggers an exception
     # if the connection wasn't already closed

tests/test_connector.py

@@ -1490,6 +1490,11 @@ async def test_tcp_connector_ctor(loop: Any) -> None:
     assert conn.family == 0
 
 
+async def test_tcp_connector_allowed_protocols(loop: Any) -> None:
+    conn = aiohttp.TCPConnector()
+    assert conn.allowed_protocol_schema_set == {"", "tcp", "http", "https", "ws", "wss"}
+
+
 async def test_invalid_ssl_param() -> None:
     with pytest.raises(TypeError):
         aiohttp.TCPConnector(ssl=object())
@@ -1649,6 +1654,19 @@ async def test_ctor_with_default_loop(loop: Any) -> None:
     assert loop is conn._loop
 
 
+async def test_base_connector_allows_all_protocols(loop: Any) -> None:
+    conn = aiohttp.BaseConnector()
+    assert conn.allowed_protocol_schema_set == {
+        "",
+        "http",
+        "https",
+        "ws",
+        "wss",
+        "unix",
+        "npipe",
+    }
+
+
 async def test_connect_with_limit(loop: Any, key: Any) -> None:
     proto = create_mocked_conn(loop)
     proto.is_connected.return_value = True
@@ -2420,6 +2438,14 @@ async def handler(request):
 
     connector = aiohttp.UnixConnector(unix_sockname)
     assert unix_sockname == connector.path
+    assert connector.allowed_protocol_schema_set == {
+        "",
+        "http",
+        "https",
+        "ws",
+        "wss",
+        "unix",
+    }
 
     session = client.ClientSession(connector=connector)
     r = await session.get(url)
@@ -2445,6 +2471,14 @@ async def handler(request):
 
     connector = aiohttp.NamedPipeConnector(pipe_name)
     assert pipe_name == connector.path
+    assert connector.allowed_protocol_schema_set == {
+        "",
+        "http",
+        "https",
+        "ws",
+        "wss",
+        "npipe",
+    }
 
     session = client.ClientSession(connector=connector)
     r = await session.get(url)