Add additional tests for SSL authentication and fix minor bug

ca_cert variable in request_with_key_password might be undefined
before use.

cabby/dispatcher.py

@@ -404,6 +404,7 @@ def request_with_key_password(
     # See also 'get_generic_session' which sets many of these attributes.
 
     # session 'verify' attribute can be a bool or a path to a CA bundle:
+    ca_cert = None
     if not isinstance(session.verify, bool):
         ca_cert = session.verify
     context = ssl.create_default_context(

tests/ssl_test_files/client_no_pass.key

@@ -0,0 +1,27 @@
+-----BEGIN RSA PRIVATE KEY-----
+MIIEogIBAAKCAQEA0fYP7Q2xmfLoVNjtFsOQwTtYzQcvuR/3mxSXWDrDM/FcSgbX
+spedvf26tzr8PX68YljJN9IwaRVx+20lykH8ZPKminYSht8RdcLUXH+vzpFqGFFd
+edKsO+//apNw87KV6rYONp+nbE5HJ7GRiAj3Aop5EAtHEjANdJAizyYsMoJiImtP
+f8doritQ11QSs59mTPsyRuM2ZRvu3CzloBEzx/ClTXSVAMP7jZFanxXcwnrJYSbL
+/BKxrbPVyzPYBeMjx5V9GUg/YZcDC62C+s0x1CQjacK9meSIx7sXRap3OPFbZ/9c
+/OWJiTxx6lzywecoELRX/nfDqsr1gn4/v+GvXwIDAQABAoIBAA7fIO2nhKbk37Py
+0YhFPeGR9I5BLg4sx3sCkF29e8oYFHNh3LFXr+KfMDR+RxOMAIfuxgHgL3GiBTrL
+1ltnJRt5XHZ+On23GyN+M1CB8s/s2Nj0GmzgkTaFn0/LNbrtMVU4o/UWheNUABI9
+r5M1H1ncuQp74gLVyH4zH2QQzhydorho/RQ4C8wHIm/yUu2RlrlmoV/CCtuFql7Q
+EOCwfuf3wom/LUbL2mFNdqYg2LS5ChzFps69zeiO7LF7cppqGAM6hQmtvacmUC7z
+H1l1Qn6c7uWxI4zezvEF+zEYg2bZhV7Bm2G4k0H5NyC3lP9DD9OjAem9SQaVtO46
+b7MLrzECgYEA8ypZBcT5YuKH6v5M/ndzY5tek2PrTuOuzkNY9XC3NgOafjyunbxR
+o7kn6z6ttpaV8nr43PP5k5FiWyek9FxZp893myEWbxpCZ4s8iKGQs13ICqb3NhAb
+kmLqEeaHkgqGt2pISOnwKwq323wWhl9iYOlipgpLa2E5aBTGOAAq/esCgYEA3QsP
+AdwNcVTL9irThaBnVZUw0UViv5oaoYIVJBfkjAZeLSi5Oar1+SGErmD8c/RtrB07
+FRABemjRk1Kz1Y1DM4qP6ldWlFaNqvvm+sH8MXtazNeHaFLWUZ5zeh33hV48FUBH
+5CG9nXGWSOU/KNYbYk2oFS4fnlVUeXfmsKSBE10CgYBfU46qsFmD5oKaIS9V4sYd
+ml8tMNKijqeMvOI29gUc67S5IFjkBVuL754yntPC7K3D7Wl6VTrWGvyP9663DS0o
+mDCvY/1DeOvnY7JLbesoJe+yHVp0m0Pz00sn0VP8cJv3c3b6/prkhMMnDDJYYzRS
+Aaxmo0qFwgubPemnMomoGQKBgF2OAqoM0vu2oiTsd15FR7cnT+Qi8+qYdNEK15vR
+KQBC6bU/WTYZL8Zj815C2lbRi2GfdZQylA2VM66hAFBQW7MC4GqU0KY0A/3sZBSZ
+6f2fcgzk5AC5ntAoukNjT5H5EoEEFeluhmyO2Ma9kH/eLvTqXUupm+RNxVUr9E2r
+Mp5ZAoGAJWKK/jR2Bl/f/mNasbmoxs86wwCAHLd8hJ3YMxiN0sRkQfroV/+9En2H
+LhWqQHIb+lQQzECwnayf4K+SxQUoR5GxOzLe8QIzcsaFfmSjA7pxlIqvzXBVCQKy
+/kNpbVHGK/SsA0/YDdFA5XA9lO23+dL0lxsoptScrHP0cTKoMtI=
+-----END RSA PRIVATE KEY-----

tests/ssl_test_files/client_no_pass.pem

@@ -0,0 +1,23 @@
+-----BEGIN CERTIFICATE-----
+MIIDwTCCAqkCCQDNW0hGu6KfqTANBgkqhkiG9w0BAQsFADCBnTELMAkGA1UEBhMC
+TkwxFjAUBgNVBAgMDU5vcnRoLUhvbGxhbmQxEjAQBgNVBAcMCUFtc3RlcmRhbTET
+MBEGA1UECgwKRWNsZWN0aWNJUTETMBEGA1UECwwKQ2FiYnkgVGVzdDETMBEGA1UE
+AwwKY2FiYnktdGVzdDEjMCEGCSqGSIb3DQEJARYUY2FiYnlAZWNsZWN0aWNpcS5j
+b20wHhcNMTgwMzI5MDkwODEzWhcNMjEwMTE2MDkwODEzWjCBpjELMAkGA1UEBhMC
+TkwxFjAUBgNVBAgMDU5vcnRoLUhvbGxhbmQxEjAQBgNVBAcMCUFtc3RlcmRhbTET
+MBEGA1UECgwKRWNsZWN0aWNJUTEOMAwGA1UECwwFQ2FiYnkxITAfBgNVBAMMGGNh
+YmJ5LXRlc3Qtbm8tcGFzc3BocmFzZTEjMCEGCSqGSIb3DQEJARYUY2FiYnlAZWNs
+ZWN0aWNpcS5jb20wggEiMA0GCSqGSIb3DQEBAQUAA4IBDwAwggEKAoIBAQDR9g/t
+DbGZ8uhU2O0Ww5DBO1jNBy+5H/ebFJdYOsMz8VxKBteyl529/bq3Ovw9frxiWMk3
+0jBpFXH7bSXKQfxk8qaKdhKG3xF1wtRcf6/OkWoYUV150qw77/9qk3DzspXqtg42
+n6dsTkcnsZGICPcCinkQC0cSMA10kCLPJiwygmIia09/x2iuK1DXVBKzn2ZM+zJG
+4zZlG+7cLOWgETPH8KVNdJUAw/uNkVqfFdzCeslhJsv8ErGts9XLM9gF4yPHlX0Z
+SD9hlwMLrYL6zTHUJCNpwr2Z5IjHuxdFqnc48Vtn/1z85YmJPHHqXPLB5ygQtFf+
+d8OqyvWCfj+/4a9fAgMBAAEwDQYJKoZIhvcNAQELBQADggEBACh1tE0dIfOF0Mi2
+52976pChfkUGNlVmCbkI8UdwcPq9W3a298s8xbg2s+BnxBZhdRgF4xRepq4Db3h9
+S7f9iQ0iCHN61yh0UNS/NgsM/ALGhHFCv1sP8ajs3vMDa4P4X5xoBavc+Pf+TKIm
+CFq/yfXhnMXzfTvLh82zzaE8LM3x+IpthxLkq3F80OXNgyRs5gVFI3poln8TMWiB
+I1PnrzBRJcHaod67Mly/iHjH0Zwy6jgkEgrbWLmzkmUXU2kEryt0HlaGO4pj5/2A
+CeZqZxjgzp5NSkdFUEauHYH6bzwhSv+BT0jZOgm5+zt/lOGU8a1Y+BJmc2UhgWQ1
+Rhf+7yc=
+-----END CERTIFICATE-----

tests/test_ssl.py

@@ -3,6 +3,8 @@
 
 import pytest
 import werkzeug.serving
+import werkzeug.wrappers
+from six.moves import urllib
 
 import cabby
 import fixtures11
@@ -34,29 +36,32 @@ def get_free_port():
 
 
 def remote_taxii_app(env, start_response):
-    if env['PATH_INFO'] == '/auth':
+    path = env['PATH_INFO']
+
+    if path == '/auth':
         start_response('200 OK',  [('Content-Type', 'application/json')])
         return [b'{"token": "123"}']
-    elif env['PATH_INFO'] == fixtures11.DISCOVERY_PATH:
+
+    if path == fixtures11.DISCOVERY_PATH:
         taxii_headers = [
             ('Content-Type', 'application/xml'),
             ('X-TAXII-Content-Type', 'urn:taxii.mitre.org:message:xml:1.1'),
         ]
         start_response('200 OK', taxii_headers)
         return [fixtures11.DISCOVERY_RESPONSE.encode()]
-    else:
-        raise Exception('Unknown test path')
 
+    raise Exception('Unknown test path')
 
-def test_jwt_auth_with_ssl(httpsserver):
-    # Serve a JWT token
+
+def test_set_auth_ssl(httpsserver):
     host, port = httpsserver.server_address
     client = cabby.create_client(
         host=host,
         port=port,
         use_https=True,
         discovery_path=fixtures11.DISCOVERY_PATH)
-    client.set_auth(
+
+    auth_params = dict(
         username='cabby',
         password='test',
         ca_cert='tests/ssl_test_files/root_ca.pem',
@@ -65,5 +70,41 @@ def test_jwt_auth_with_ssl(httpsserver):
         key_password='cabby-test',
         jwt_auth_url='/auth',
         verify_ssl=True)
+
+    # Test client certificate with key_password.
+    # This doesn't use the requests library, instead
+    # it uses urllib2 or 3 through the 'request_with_key_password' function:
+    client.set_auth(**auth_params)
+    services = client.discover_services()
+    assert len(services) == 4
+
+    # Test we get the correct error when forgetting a key_password:
+    missing_key_password = dict(auth_params, key_password=None)
+    client.set_auth(**missing_key_password)
+    error_msg = "Key file is encrypted but key password was not provided"
+    with pytest.raises(ValueError, matches=error_msg):
+        client.discover_services()
+
+    # Test that server certificate authentication can fail:
+    invalid_server_ca_cert = dict(auth_params, ca_cert=None)
+    client.set_auth(**invalid_server_ca_cert)
+    error_msg = "certificate verify failed"
+    with pytest.raises(urllib.error.URLError, matches=error_msg):
+        client.discover_services()
+
+    # Test that verify_ssl=False ignores the invalid certificate:
+    invalid_server_ca_cert_no_verify = dict(
+        auth_params, ca_cert=None, verify_ssl=False)
+    client.set_auth(**invalid_server_ca_cert_no_verify)
+    services = client.discover_services()
+    assert len(services) == 4
+
+    # Test client certificate with key_password.
+    # This uses the requests library as usual:
+    client_key_without_passphrase = dict(
+        auth_params,
+        key_file='tests/ssl_test_files/client_no_pass.key',
+        cert_file='tests/ssl_test_files/client_no_pass.pem')
+    client.set_auth(**client_key_without_passphrase)
     services = client.discover_services()
     assert len(services) == 4