Integrate trustme fake CA lib for testing TLS

Also:
* Use real TLS context where it's disabled in tests
* Add a change note about trustme integration

Closes #3487

CHANGES/3487.misc

@@ -0,0 +1 @@
+Integrate [`trustme`](https://trustme.readthedocs.io/en/latest/) to better test TLS support.

requirements/ci-wheel.txt

@@ -13,6 +13,7 @@ pytest==4.0.2
 pytest-cov==2.6.0
 pytest-mock==1.10.0
 tox==3.6.1
+trustme==0.4.0
 twine==1.12.1
 yarl==1.3.0
 

tests/conftest.py

@@ -1,8 +1,11 @@
+import hashlib
 import pathlib
 import shutil
+import ssl
 import tempfile
 
 import pytest
+import trustme
 
 pytest_plugins = ['aiohttp.pytest_plugin', 'pytester']
 
@@ -15,3 +18,54 @@ def shorttmpdir():
     tmpdir = pathlib.Path(tempfile.mkdtemp())
     yield tmpdir
     shutil.rmtree(tmpdir, ignore_errors=True)
+
+
+@pytest.fixture
+def tls_certificate_authority():
+    return trustme.CA()
+
+
+@pytest.fixture
+def tls_certificate(tls_certificate_authority):
+    return tls_certificate_authority.issue_server_cert(
+        'localhost',
+        '127.0.0.1',
+        '::1',
+    )
+
+
+@pytest.fixture
+def ssl_ctx(tls_certificate):
+    ssl_ctx = ssl.SSLContext(ssl.PROTOCOL_SSLv23)
+    tls_certificate.configure_cert(ssl_ctx)
+    return ssl_ctx
+
+
+@pytest.fixture
+def client_ssl_ctx(tls_certificate_authority):
+    ssl_ctx = ssl.create_default_context(purpose=ssl.Purpose.SERVER_AUTH)
+    tls_certificate_authority.configure_trust(ssl_ctx)
+    return ssl_ctx
+
+
+@pytest.fixture
+def tls_ca_certificate_pem_path(tls_certificate_authority):
+    with tls_certificate_authority.cert_pem.tempfile() as ca_cert_pem:
+        yield ca_cert_pem
+
+
+@pytest.fixture
+def tls_certificate_pem_path(tls_certificate):
+    with tls_certificate.private_key_and_cert_chain_pem.tempfile() as cert_pem:
+        yield cert_pem
+
+
+@pytest.fixture
+def tls_certificate_pem_bytes(tls_certificate):
+    return tls_certificate.cert_chain_pems[0].bytes()
+
+
+@pytest.fixture
+def tls_certificate_fingerprint_sha256(tls_certificate_pem_bytes):
+    tls_cert_der = ssl.PEM_cert_to_DER_cert(tls_certificate_pem_bytes.decode())
+    return hashlib.sha256(tls_cert_der).digest()

tests/test_client_functional.py

@@ -6,7 +6,6 @@
 import json
 import pathlib
 import socket
-import ssl
 from unittest import mock
 
 import pytest
@@ -25,18 +24,9 @@ def here():
     return pathlib.Path(__file__).parent
 
 
-@pytest.fixture
-def ssl_ctx(here):
-    ssl_ctx = ssl.SSLContext(ssl.PROTOCOL_SSLv23)
-    ssl_ctx.load_cert_chain(
-        str(here / 'sample.crt'),
-        str(here / 'sample.key'))
-    return ssl_ctx
-
-
 @pytest.fixture
 def fname(here):
-    return here / 'sample.key'
+    return here / 'conftest.py'
 
 
 def ceil(val):
@@ -274,8 +264,11 @@ async def handler(request):
     assert 200 == resp.status
 
 
-async def test_ssl_client(ssl_ctx, aiohttp_server, aiohttp_client) -> None:
-    connector = aiohttp.TCPConnector(ssl=False)
+async def test_ssl_client(
+        aiohttp_server, ssl_ctx,
+        aiohttp_client, client_ssl_ctx,
+) -> None:
+    connector = aiohttp.TCPConnector(ssl=client_ssl_ctx)
 
     async def handler(request):
         return web.Response(text='Test message')
@@ -291,17 +284,16 @@ async def handler(request):
     assert txt == 'Test message'
 
 
-async def test_tcp_connector_fingerprint_ok(aiohttp_server, aiohttp_client,
-                                            ssl_ctx):
-
-    fingerprint = (b'0\x9a\xc9D\x83\xdc\x91\'\x88\x91\x11\xa1d\x97\xfd'
-                   b'\xcb~7U\x14D@L'
-                   b'\x11\xab\x99\xa8\xae\xb7\x14\xee\x8b')
+async def test_tcp_connector_fingerprint_ok(
+        aiohttp_server, aiohttp_client,
+        ssl_ctx, tls_certificate_fingerprint_sha256,
+):
+    tls_fingerprint = Fingerprint(tls_certificate_fingerprint_sha256)
 
     async def handler(request):
         return web.Response(text='Test message')
 
-    connector = aiohttp.TCPConnector(ssl=Fingerprint(fingerprint))
+    connector = aiohttp.TCPConnector(ssl=tls_fingerprint)
     app = web.Application()
     app.router.add_route('GET', '/', handler)
     server = await aiohttp_server(app, ssl=ssl_ctx)
@@ -312,17 +304,14 @@ async def handler(request):
     resp.close()
 
 
-async def test_tcp_connector_fingerprint_fail(aiohttp_server, aiohttp_client,
-                                              ssl_ctx):
-
-    fingerprint = (b'0\x9a\xc9D\x83\xdc\x91\'\x88\x91\x11\xa1d\x97\xfd'
-                   b'\xcb~7U\x14D@L'
-                   b'\x11\xab\x99\xa8\xae\xb7\x14\xee\x8b')
-
+async def test_tcp_connector_fingerprint_fail(
+        aiohttp_server, aiohttp_client,
+        ssl_ctx, tls_certificate_fingerprint_sha256,
+):
     async def handler(request):
         return web.Response(text='Test message')
 
-    bad_fingerprint = b'\x00' * len(fingerprint)
+    bad_fingerprint = b'\x00' * len(tls_certificate_fingerprint_sha256)
 
     connector = aiohttp.TCPConnector(ssl=Fingerprint(bad_fingerprint))
 
@@ -335,7 +324,7 @@ async def handler(request):
         await client.get('/')
     exc = cm.value
     assert exc.expected == bad_fingerprint
-    assert exc.got == fingerprint
+    assert exc.got == tls_certificate_fingerprint_sha256
 
 
 async def test_format_task_get(aiohttp_server) -> None:
@@ -1402,7 +1391,7 @@ async def handler(request):
                                         'text/plain',
                                         'application/octet-stream']
         assert request.headers['content-disposition'] == (
-            "inline; filename=\"sample.key\"; filename*=utf-8''sample.key")
+            "inline; filename=\"conftest.py\"; filename*=utf-8''conftest.py")
 
         return web.Response()
 
@@ -1428,6 +1417,7 @@ async def handler(request):
         # then use 'application/octet-stream' default
         assert request.content_type in ['application/pgp-keys',
                                         'text/plain',
+                                        'text/x-python',
                                         'application/octet-stream']
         return web.Response()
 

tests/test_client_request.py

@@ -801,7 +801,7 @@ async def test_chunked_transfer_encoding(loop, conn) -> None:
 
 async def test_file_upload_not_chunked(loop) -> None:
     here = os.path.dirname(__file__)
-    fname = os.path.join(here, 'sample.key')
+    fname = os.path.join(here, 'aiohttp.png')
     with open(fname, 'rb') as f:
         req = ClientRequest(
             'post', URL('http://python.org/'),
@@ -828,7 +828,7 @@ async def test_precompressed_data_stays_intact(loop) -> None:
 
 async def test_file_upload_not_chunked_seek(loop) -> None:
     here = os.path.dirname(__file__)
-    fname = os.path.join(here, 'sample.key')
+    fname = os.path.join(here, 'aiohttp.png')
     with open(fname, 'rb') as f:
         f.seek(100)
         req = ClientRequest(
@@ -842,7 +842,7 @@ async def test_file_upload_not_chunked_seek(loop) -> None:
 
 async def test_file_upload_force_chunked(loop) -> None:
     here = os.path.dirname(__file__)
-    fname = os.path.join(here, 'sample.key')
+    fname = os.path.join(here, 'aiohttp.png')
     with open(fname, 'rb') as f:
         req = ClientRequest(
             'post', URL('http://python.org/'),
@@ -1270,11 +1270,11 @@ async def create_connection(req, traces, timeout):
     conn.close()
 
 
-def test_verify_ssl_false_with_ssl_context(loop) -> None:
+def test_verify_ssl_false_with_ssl_context(loop, ssl_ctx) -> None:
     with pytest.warns(DeprecationWarning):
         with pytest.raises(ValueError):
             _merge_ssl_params(None, verify_ssl=False,
-                              ssl_context=mock.Mock(), fingerprint=None)
+                              ssl_context=ssl_ctx, fingerprint=None)
 
 
 def test_bad_fingerprint(loop) -> None:

tests/test_connector.py

@@ -3,7 +3,6 @@
 import asyncio
 import gc
 import hashlib
-import os.path
 import platform
 import socket
 import ssl
@@ -1999,20 +1998,16 @@ async def test_resolver_not_called_with_address_is_ip(loop) -> None:
     resolver.resolve.assert_not_called()
 
 
-async def test_tcp_connector_raise_connector_ssl_error(aiohttp_server) -> None:
+async def test_tcp_connector_raise_connector_ssl_error(
+        aiohttp_server, ssl_ctx,
+) -> None:
     async def handler(request):
         return web.Response()
 
     app = web.Application()
     app.router.add_get('/', handler)
 
-    here = os.path.join(os.path.dirname(__file__), '..', 'tests')
-    keyfile = os.path.join(here, 'sample.key')
-    certfile = os.path.join(here, 'sample.crt')
-    sslcontext = ssl.SSLContext(ssl.PROTOCOL_SSLv23)
-    sslcontext.load_cert_chain(certfile, keyfile)
-
-    srv = await aiohttp_server(app, ssl=sslcontext)
+    srv = await aiohttp_server(app, ssl=ssl_ctx)
 
     port = unused_port()
     conn = aiohttp.TCPConnector(local_addr=('127.0.0.1', port))
@@ -2038,27 +2033,22 @@ async def handler(request):
 
 
 async def test_tcp_connector_do_not_raise_connector_ssl_error(
-        aiohttp_server) -> None:
+        aiohttp_server, ssl_ctx, client_ssl_ctx,
+) -> None:
     async def handler(request):
         return web.Response()
 
     app = web.Application()
     app.router.add_get('/', handler)
 
-    here = os.path.join(os.path.dirname(__file__), '..', 'tests')
-    keyfile = os.path.join(here, 'sample.key')
-    certfile = os.path.join(here, 'sample.crt')
-    sslcontext = ssl.SSLContext(ssl.PROTOCOL_SSLv23)
-    sslcontext.load_cert_chain(certfile, keyfile)
-
-    srv = await aiohttp_server(app, ssl=sslcontext)
+    srv = await aiohttp_server(app, ssl=ssl_ctx)
     port = unused_port()
     conn = aiohttp.TCPConnector(local_addr=('127.0.0.1', port))
 
     session = aiohttp.ClientSession(connector=conn)
     url = srv.make_url('/')
 
-    r = await session.get(url, ssl=sslcontext)
+    r = await session.get(url, ssl=client_ssl_ctx)
 
     r.release()
     first_conn = next(iter(conn._conns.values()))[0][0]
@@ -2068,7 +2058,7 @@ async def handler(request):
     except AttributeError:
         _sslcontext = first_conn.transport._sslcontext
 
-    assert _sslcontext is sslcontext
+    assert _sslcontext is client_ssl_ctx
     r.close()
 
     await session.close()

tests/test_route_def.py

@@ -128,8 +128,8 @@ def test_static(router) -> None:
     info = resource.get_info()
     assert info['prefix'] == '/prefix'
     assert info['directory'] == folder
-    url = resource.url_for(filename='sample.key')
-    assert url == URL('/prefix/sample.key')
+    url = resource.url_for(filename='aiohttp.png')
+    assert url == URL('/prefix/aiohttp.png')
 
 
 def test_head_deco(router) -> None: