[PR #10726/feff48d backport][3.11] Disable TLS in TLS warning for uvloop (#10767)

Co-authored-by: Matthew Go <lezgomatt@gmail.com>

Author: bdraco

CHANGES/7686.bugfix.rst

@@ -0,0 +1 @@
+Disabled TLS in TLS warning (when using HTTPS proxies) for uvloop and newer Python versions -- by :user:`lezgomatt`.

CONTRIBUTORS.txt

@@ -241,6 +241,7 @@ Martin Sucha
 Mathias Fröjdman
 Mathieu Dugré
 Matt VanEseltine
+Matthew Go
 Matthias Marquardt
 Matthieu Hauglustaine
 Matthieu Rigal

aiohttp/connector.py

@@ -1203,7 +1203,13 @@ def _warn_about_tls_in_tls(
         if req.request_info.url.scheme != "https":
             return
 
-        asyncio_supports_tls_in_tls = getattr(
+        # Check if uvloop is being used, which supports TLS in TLS,
+        # otherwise assume that asyncio's native transport is being used.
+        if type(underlying_transport).__module__.startswith("uvloop"):
+            return
+
+        # Support in asyncio was added in Python 3.11 (bpo-44011)
+        asyncio_supports_tls_in_tls = sys.version_info >= (3, 11) or getattr(
             underlying_transport,
             "_start_tls_compatible",
             False,

tests/conftest.py

@@ -7,7 +7,7 @@
 from hashlib import md5, sha1, sha256
 from pathlib import Path
 from tempfile import TemporaryDirectory
-from typing import Any, Generator
+from typing import Any, Generator, Iterator
 from unittest import mock
 from uuid import uuid4
 
@@ -27,6 +27,12 @@
 except ImportError:
     TRUSTME = False
 
+
+try:
+    import uvloop
+except ImportError:
+    uvloop = None  # type: ignore[assignment]
+
 pytest_plugins = ["aiohttp.pytest_plugin", "pytester"]
 
 IS_HPUX = sys.platform.startswith("hp-ux")
@@ -193,6 +199,16 @@ def selector_loop():
         yield _loop
 
 
+@pytest.fixture
+def uvloop_loop() -> Iterator[asyncio.AbstractEventLoop]:
+    policy = uvloop.EventLoopPolicy()
+    asyncio.set_event_loop_policy(policy)
+
+    with loop_context(policy.new_event_loop) as _loop:
+        asyncio.set_event_loop(_loop)
+        yield _loop
+
+
 @pytest.fixture
 def netrc_contents(
     tmp_path: Path,

tests/test_proxy_functional.py

@@ -1,6 +1,7 @@
 import asyncio
 import os
 import pathlib
+import platform
 import ssl
 import sys
 from re import match as match_regex
@@ -202,6 +203,32 @@ async def test_https_proxy_unsupported_tls_in_tls(
     await asyncio.sleep(0.1)
 
 
+@pytest.mark.usefixtures("uvloop_loop")
+@pytest.mark.skipif(
+    platform.system() == "Windows" or sys.implementation.name != "cpython",
+    reason="uvloop is not supported on Windows and non-CPython implementations",
+)
+@pytest.mark.filterwarnings(r"ignore:.*ssl.OP_NO_SSL*")
+# Filter out the warning from
+# https://github.com/abhinavsingh/proxy.py/blob/30574fd0414005dfa8792a6e797023e862bdcf43/proxy/common/utils.py#L226
+# otherwise this test will fail because the proxy will die with an error.
+async def test_uvloop_secure_https_proxy(
+    client_ssl_ctx: ssl.SSLContext,
+    secure_proxy_url: URL,
+) -> None:
+    """Ensure HTTPS sites are accessible through a secure proxy without warning when using uvloop."""
+    conn = aiohttp.TCPConnector()
+    sess = aiohttp.ClientSession(connector=conn)
+    url = URL("https://example.com")
+
+    async with sess.get(url, proxy=secure_proxy_url, ssl=client_ssl_ctx) as response:
+        assert response.status == 200
+
+    await sess.close()
+    await conn.close()
+    await asyncio.sleep(0.1)
+
+
 @pytest.fixture
 def proxy_test_server(aiohttp_raw_server, loop, monkeypatch):
     # Handle all proxy requests and imitate remote server response.