Adds basic xfail tests for secure proxy support

requirements/test.txt

@@ -5,6 +5,7 @@ cryptography==3.3.1; platform_machine!="i686" and python_version<"3.9" # no 32-b
 freezegun==1.1.0
 mypy==0.910; implementation_name=="cpython"
 mypy-extensions==0.4.3; implementation_name=="cpython"
+proxy.py==2.0.0
 pytest==6.1.2
 pytest-cov==2.12.1
 pytest-mock==3.6.1

tests/conftest.py

@@ -69,6 +69,12 @@ def tls_ca_certificate_pem_path(tls_certificate_authority):
         yield ca_cert_pem
 
 
+@pytest.fixture
+def tls_ca_private_key_pem_path(tls_certificate_authority):
+    with tls_certificate_authority.private_key_pem.tempfile() as ca_key_pem:
+        yield ca_key_pem
+
+
 @pytest.fixture
 def tls_certificate_pem_path(tls_certificate):
     with tls_certificate.private_key_and_cert_chain_pem.tempfile() as cert_pem:

tests/test_proxy_functional.py

@@ -1,15 +1,137 @@
 import asyncio
 import os
 import pathlib
+import ssl
 from unittest import mock
 
+import proxy
 import pytest
 from yarl import URL
 
 import aiohttp
 from aiohttp import web
 
 
+# @pytest.fixture
+# def secure_proxy_url(
+#     aiohttp_unused_port, tls_ca_certificate_pem_path, tls_ca_private_key_pem_path
+# ):
+#     """Return the an instance of a running secure proxy."""
+#     unused_port = str(aiohttp_unused_port())
+#     address = "127.0.0.1"
+#     input_args = [
+#         "--num-workers",
+#         "1",
+#         "--hostname",
+#         "127.0.0.1",
+#         "--port",
+#         unused_port,
+#         "--cert-file",
+#         tls_ca_certificate_pem_path,
+#         "--key-file",
+#         tls_ca_private_key_pem_path,
+#     ]
+#     proxy_url = URL.build(scheme="https", host=address, port=unused_port)
+#     with proxy.Proxy(input_args=input_args):
+#         yield proxy_url
+
+
+@pytest.fixture
+def secure_proxy_url(
+    aiohttp_unused_port, tls_certificate,
+):
+    """Return the an instance of a running secure proxy."""
+    unused_port = str(aiohttp_unused_port())
+    address = "127.0.0.1"
+    with tls_certificate.cert_chain_pems[0].tempfile() as server_cert_path:
+        with tls_certificate.private_key_pem.tempfile() as server_key:
+            input_args = [
+                "--num-workers",
+                "1",
+                "--hostname",
+                "127.0.0.1",
+                "--port",
+                unused_port,
+                "--cert-file",
+                server_cert_path,
+                "--key-file",
+                server_key,
+            ]
+            proxy_url = URL.build(scheme="https", host=address, port=unused_port)
+            with proxy.Proxy(input_args=input_args):
+                yield proxy_url
+
+
+# @pytest.mark.xfail
+async def test_secure_proxy_http_absolute_path(
+    loop,
+    aiohttp_server,
+    client_ssl_ctx,
+    secure_proxy_url,
+) -> None:
+    """Test urls can be requested through a secure proxy."""
+    response_body = "Test message"
+
+    async def handler(*args, **kwargs):
+        return web.Response(text=response_body)
+
+    app = web.Application()
+    app.router.add_route("GET", "/", handler)
+    server = await aiohttp_server(app)
+
+    conn = aiohttp.TCPConnector(loop=loop)
+    sess = aiohttp.ClientSession(connector=conn, loop=loop)
+
+    url = URL.build(scheme="http", host=server.host, port=server.port)
+    response = await sess.get(url, proxy=secure_proxy_url, ssl=client_ssl_ctx)
+    assert response.status == 200
+    assert await response.text() == response_body
+
+
+# @pytest.mark.xfail
+async def test_secure_proxy_https_absolute_path(
+    loop,
+    aiohttp_server,
+    tls_certificate_pem_path,
+    ssl_ctx,
+    client_ssl_ctx,
+    tls_ca_certificate_pem_path,
+    tls_certificate_authority,
+    secure_proxy_url,
+) -> None:
+    """Test urls can be requested through a secure proxy."""
+    response_body = "Test message"
+
+    async def handler(*args, **kwargs):
+        return web.Response(text=response_body)
+
+    app = web.Application()
+    app.router.add_route("GET", "/", handler)
+    server = await aiohttp_server(app, ssl=ssl_ctx)
+
+    conn = aiohttp.TCPConnector(loop=loop)
+    sess = aiohttp.ClientSession(connector=conn, loop=loop)
+
+    url = URL.build(scheme="https", host=server.host, port=server.port)
+    # breakpoint()
+    qq = ssl.create_default_context()
+    url = "https://httpbin.org/get"
+    # secure_proxy_url = "https://127.0.0.1:8899"
+    import logging
+    logging.getLogger().info(f"secure_proxy_url:  {secure_proxy_url}")
+    with tls_certificate_authority.cert_pem.tempfile() as tt:
+        logging.getLogger().info(f"path to client CA file: {tt}")
+        qq.load_verify_locations(tt)
+    #     qq = ssl.create_default_context(cafile="/home/vagrant/client.pem.ORIG")
+    #     client_ssl_ctx.load_verify_locations("/home/vagrant/client.pem")
+    #     response = await sess.get(url, proxy=secure_proxy_url, ssl=client_ssl_ctx)
+    #     breakpoint()
+    #     qq.load_verify_locations("/home/vagrant/client.pem.ORIG")
+    response = await sess.get(url, proxy=secure_proxy_url, ssl=qq)
+    assert response.status == 200
+    # assert await response.text() == response_body
+
+
 @pytest.fixture
 def proxy_test_server(aiohttp_raw_server, loop, monkeypatch):
     # Handle all proxy requests and imitate remote server response.