AIOHTTPTransport default ssl cert validation add warning

gql/transport/aiohttp.py

@@ -46,7 +46,7 @@ def __init__(
         headers: Optional[LooseHeaders] = None,
         cookies: Optional[LooseCookies] = None,
         auth: Optional[Union[BasicAuth, "AppSyncAuthentication"]] = None,
-        ssl: Union[SSLContext, bool, Fingerprint] = False,
+        ssl: Union[SSLContext, bool, Fingerprint] = True,
         timeout: Optional[int] = None,
         ssl_close_timeout: Optional[Union[int, float]] = 10,
         json_serialize: Callable = json.dumps,

tests/test_aiohttp.py

@@ -1301,7 +1301,10 @@ async def handler(request):
     assert str(url).startswith("https://")
 
     transport = AIOHTTPTransport(
-        url=url, timeout=10, ssl_close_timeout=ssl_close_timeout
+        url=url,
+        timeout=10,
+        ssl_close_timeout=ssl_close_timeout,
+        ssl=False,  # Disable cert verification as we are using a self certificate
     )
 
     async with Client(transport=transport) as session:
@@ -1318,6 +1321,39 @@ async def handler(request):
         assert africa["code"] == "AF"
 
 
+@pytest.mark.asyncio
+async def test_aiohttp_query_https_self_cert_fail(event_loop, ssl_aiohttp_server):
+    """By default, we should verify the ssl certificate"""
+    from aiohttp.client_exceptions import ClientConnectorCertificateError
+    from aiohttp import web
+    from gql.transport.aiohttp import AIOHTTPTransport
+
+    async def handler(request):
+        return web.Response(text=query1_server_answer, content_type="application/json")
+
+    app = web.Application()
+    app.router.add_route("POST", "/", handler)
+    server = await ssl_aiohttp_server(app)
+
+    url = server.make_url("/")
+
+    assert str(url).startswith("https://")
+
+    transport = AIOHTTPTransport(url=url, timeout=10)
+
+    with pytest.raises(ClientConnectorCertificateError) as exc_info:
+        async with Client(transport=transport) as session:
+            query = gql(query1_str)
+
+            # Execute query asynchronously
+            await session.execute(query)
+
+    expected_error = "certificate verify failed: self-signed certificate"
+
+    assert expected_error in str(exc_info.value)
+    assert transport.session is None
+
+
 @pytest.mark.asyncio
 async def test_aiohttp_error_fetching_schema(event_loop, aiohttp_server):
     from aiohttp import web

tests/test_httpx.py

@@ -77,6 +77,103 @@ def test_code():
     await run_sync_test(event_loop, server, test_code)
 
 
+@pytest.mark.aiohttp
+@pytest.mark.asyncio
+async def test_httpx_query_https(event_loop, ssl_aiohttp_server, run_sync_test):
+    from aiohttp import web
+    from gql.transport.httpx import HTTPXTransport
+
+    async def handler(request):
+        return web.Response(
+            text=query1_server_answer,
+            content_type="application/json",
+            headers={"dummy": "test1234"},
+        )
+
+    app = web.Application()
+    app.router.add_route("POST", "/", handler)
+    server = await ssl_aiohttp_server(app)
+
+    url = str(server.make_url("/"))
+
+    print(url)
+
+    assert str(url).startswith("https://")
+
+    def test_code():
+        transport = HTTPXTransport(
+            url=url,
+            verify=False,
+        )
+
+        with Client(transport=transport) as session:
+
+            query = gql(query1_str)
+
+            # Execute query synchronously
+            result = session.execute(query)
+
+            continents = result["continents"]
+
+            africa = continents[0]
+
+            assert africa["code"] == "AF"
+
+            # Checking response headers are saved in the transport
+            assert hasattr(transport, "response_headers")
+            assert isinstance(transport.response_headers, Mapping)
+            assert transport.response_headers["dummy"] == "test1234"
+
+    await run_sync_test(event_loop, server, test_code)
+
+
+@pytest.mark.aiohttp
+@pytest.mark.asyncio
+async def test_httpx_query_https_self_cert_fail(
+    event_loop, ssl_aiohttp_server, run_sync_test
+):
+    """By default, we should verify the ssl certificate"""
+    from aiohttp import web
+    from httpx import ConnectError
+    from gql.transport.httpx import HTTPXTransport
+
+    async def handler(request):
+        return web.Response(
+            text=query1_server_answer,
+            content_type="application/json",
+            headers={"dummy": "test1234"},
+        )
+
+    app = web.Application()
+    app.router.add_route("POST", "/", handler)
+    server = await ssl_aiohttp_server(app)
+
+    url = str(server.make_url("/"))
+
+    print(url)
+
+    assert str(url).startswith("https://")
+
+    def test_code():
+        transport = HTTPXTransport(
+            url=url,
+        )
+
+        with pytest.raises(ConnectError) as exc_info:
+            with Client(transport=transport) as session:
+
+                query = gql(query1_str)
+
+                # Execute query synchronously
+                session.execute(query)
+
+        expected_error = "certificate verify failed: self-signed certificate"
+
+        assert expected_error in str(exc_info.value)
+
+    await run_sync_test(event_loop, server, test_code)
+
+
 @pytest.mark.aiohttp
 @pytest.mark.asyncio
 async def test_httpx_cookies(event_loop, aiohttp_server, run_sync_test):

tests/test_requests.py

@@ -77,6 +77,100 @@ def test_code():
     await run_sync_test(event_loop, server, test_code)
 
 
+@pytest.mark.aiohttp
+@pytest.mark.asyncio
+async def test_requests_query_https(event_loop, ssl_aiohttp_server, run_sync_test):
+    from aiohttp import web
+    from gql.transport.requests import RequestsHTTPTransport
+    import warnings
+
+    async def handler(request):
+        return web.Response(
+            text=query1_server_answer,
+            content_type="application/json",
+            headers={"dummy": "test1234"},
+        )
+
+    app = web.Application()
+    app.router.add_route("POST", "/", handler)
+    server = await ssl_aiohttp_server(app)
+
+    url = server.make_url("/")
+
+    def test_code():
+        with warnings.catch_warnings():
+            # Ignoring Insecure Request warning
+            warnings.filterwarnings("ignore")
+
+            transport = RequestsHTTPTransport(
+                url=url,
+                verify=False,
+            )
+
+            with Client(transport=transport) as session:
+
+                query = gql(query1_str)
+
+                # Execute query synchronously
+                result = session.execute(query)
+
+                continents = result["continents"]
+
+                africa = continents[0]
+
+                assert africa["code"] == "AF"
+
+                # Checking response headers are saved in the transport
+                assert hasattr(transport, "response_headers")
+                assert isinstance(transport.response_headers, Mapping)
+                assert transport.response_headers["dummy"] == "test1234"
+
+    await run_sync_test(event_loop, server, test_code)
+
+
+@pytest.mark.aiohttp
+@pytest.mark.asyncio
+async def test_requests_query_https_self_cert_fail(
+    event_loop, ssl_aiohttp_server, run_sync_test
+):
+    """By default, we should verify the ssl certificate"""
+    from aiohttp import web
+    from gql.transport.requests import RequestsHTTPTransport
+    from requests.exceptions import SSLError
+
+    async def handler(request):
+        return web.Response(
+            text=query1_server_answer,
+            content_type="application/json",
+            headers={"dummy": "test1234"},
+        )
+
+    app = web.Application()
+    app.router.add_route("POST", "/", handler)
+    server = await ssl_aiohttp_server(app)
+
+    url = server.make_url("/")
+
+    def test_code():
+        transport = RequestsHTTPTransport(
+            url=url,
+        )
+
+        with pytest.raises(SSLError) as exc_info:
+            with Client(transport=transport) as session:
+
+                query = gql(query1_str)
+
+                # Execute query synchronously
+                session.execute(query)
+
+        expected_error = "certificate verify failed: self-signed certificate"
+
+        assert expected_error in str(exc_info.value)
+
+    await run_sync_test(event_loop, server, test_code)
+
+
 @pytest.mark.aiohttp
 @pytest.mark.asyncio
 async def test_requests_cookies(event_loop, aiohttp_server, run_sync_test):