initial import for Open Source 🎉

util/security/test/test_jwtutil.py

@@ -0,0 +1,119 @@
+import time
+
+import pytest
+import jwt
+
+from Crypto.PublicKey import RSA
+from jwkest.jwk import RSAKey
+
+from util.security.jwtutil import (decode, exp_max_s_option, jwk_dict_to_public_key,
+                                   InvalidTokenError, InvalidAlgorithmError)
+
+
+@pytest.fixture(scope="session")
+def private_key():
+  return RSA.generate(2048)
+
+
+@pytest.fixture(scope="session")
+def private_key_pem(private_key):
+  return private_key.exportKey('PEM')
+
+
+@pytest.fixture(scope="session")
+def public_key(private_key):
+  return private_key.publickey().exportKey('PEM')
+
+
+def _token_data(audience, subject, iss, iat=None, exp=None, nbf=None):
+  return {
+    'iss': iss,
+    'aud': audience,
+    'nbf': nbf() if nbf is not None else int(time.time()),
+    'iat': iat() if iat is not None else int(time.time()),
+    'exp': exp() if exp is not None else int(time.time() + 3600),
+    'sub': subject,
+  }
+
+
+@pytest.mark.parametrize('aud, iss, nbf, iat, exp, expected_exception', [
+  pytest.param('invalidaudience', 'someissuer', None, None, None, 'Invalid audience',
+               id='invalid audience'),
+  pytest.param('someaudience', 'invalidissuer', None, None, None, 'Invalid issuer',
+               id='invalid issuer'),
+  pytest.param('someaudience', 'someissuer', lambda: time.time() + 120, None, None,
+               'The token is not yet valid',
+               id='invalid not before'),
+  pytest.param('someaudience', 'someissuer', None, lambda: time.time() + 120, None,
+               'Issued At claim',
+               id='issued at in future'),
+  pytest.param('someaudience', 'someissuer', None, None, lambda: time.time() - 100,
+               'Signature has expired',
+               id='already expired'),
+  pytest.param('someaudience', 'someissuer', None, None, lambda: time.time() + 10000,
+               'Token was signed for more than',
+               id='expiration too far in future'),
+
+  pytest.param('someaudience', 'someissuer', lambda: time.time() + 10, None, None,
+               None,
+               id='not before in future by within leeway'),
+  pytest.param('someaudience', 'someissuer', None, lambda: time.time() + 10, None,
+               None,
+               id='issued at in future but within leeway'),
+  pytest.param('someaudience', 'someissuer', None, None, lambda: time.time() - 10,
+               None,
+               id='expiration in past but within leeway'),
+])
+def test_decode_jwt_validation(aud, iss, nbf, iat, exp, expected_exception, private_key_pem,
+                               public_key):
+  token = jwt.encode(_token_data(aud, 'subject', iss, iat, exp, nbf), private_key_pem, 'RS256')
+
+  if expected_exception is not None:
+    with pytest.raises(InvalidTokenError) as ite:
+      max_exp = exp_max_s_option(3600)
+      decode(token, public_key, algorithms=['RS256'], audience='someaudience',
+             issuer='someissuer', options=max_exp, leeway=60)
+    assert ite.match(expected_exception)
+  else:
+    max_exp = exp_max_s_option(3600)
+    decode(token, public_key, algorithms=['RS256'], audience='someaudience',
+           issuer='someissuer', options=max_exp, leeway=60)
+
+
+def test_decode_jwt_invalid_key(private_key_pem):
+  # Encode with the test private key.
+  token = jwt.encode(_token_data('aud', 'subject', 'someissuer'), private_key_pem, 'RS256')
+
+  # Try to decode with a different public key.
+  another_public_key = RSA.generate(2048).publickey().exportKey('PEM')
+  with pytest.raises(InvalidTokenError) as ite:
+    max_exp = exp_max_s_option(3600)
+    decode(token, another_public_key, algorithms=['RS256'], audience='aud',
+           issuer='someissuer', options=max_exp, leeway=60)
+  assert ite.match('Signature verification failed')
+
+
+def test_decode_jwt_invalid_algorithm(private_key_pem, public_key):
+  # Encode with the test private key.
+  token = jwt.encode(_token_data('aud', 'subject', 'someissuer'), private_key_pem, 'RS256')
+
+  # Attempt to decode but only with a different algorithm than that used.
+  with pytest.raises(InvalidAlgorithmError) as ite:
+    max_exp = exp_max_s_option(3600)
+    decode(token, public_key, algorithms=['ES256'], audience='aud',
+           issuer='someissuer', options=max_exp, leeway=60)
+  assert ite.match('are not whitelisted')
+
+
+def test_jwk_dict_to_public_key(private_key, private_key_pem):
+  public_key = private_key.publickey()
+  jwk = RSAKey(key=private_key.publickey()).serialize()
+  converted = jwk_dict_to_public_key(jwk)
+
+  # Encode with the test private key.
+  token = jwt.encode(_token_data('aud', 'subject', 'someissuer'), private_key_pem, 'RS256')
+
+  # Decode with the converted key.
+  max_exp = exp_max_s_option(3600)
+  decode(token, converted, algorithms=['RS256'], audience='aud',
+         issuer='someissuer', options=max_exp, leeway=60)

util/security/test/test_ssl_util.py

@@ -0,0 +1,116 @@
+from tempfile import NamedTemporaryFile
+
+import pytest
+
+from OpenSSL import crypto
+
+from util.security.ssl import load_certificate, CertInvalidException, KeyInvalidException
+
+def generate_test_cert(hostname='somehostname', san_list=None, expires=1000000):
+  """ Generates a test SSL certificate and returns the certificate data and private key data. """
+
+  # Based on: http://blog.richardknop.com/2012/08/create-a-self-signed-x509-certificate-in-python/
+  # Create a key pair.
+  k = crypto.PKey()
+  k.generate_key(crypto.TYPE_RSA, 1024)
+
+  # Create a self-signed cert.
+  cert = crypto.X509()
+  cert.get_subject().CN = hostname
+
+  # Add the subjectAltNames (if necessary).
+  if san_list is not None:
+    cert.add_extensions([crypto.X509Extension("subjectAltName", False, ", ".join(san_list))])
+
+  cert.set_serial_number(1000)
+  cert.gmtime_adj_notBefore(0)
+  cert.gmtime_adj_notAfter(expires)
+  cert.set_issuer(cert.get_subject())
+
+  cert.set_pubkey(k)
+  cert.sign(k, 'sha1')
+
+  # Dump the certificate and private key in PEM format.
+  cert_data = crypto.dump_certificate(crypto.FILETYPE_PEM, cert)
+  key_data = crypto.dump_privatekey(crypto.FILETYPE_PEM, k)
+
+  return (cert_data, key_data)
+
+
+def test_load_certificate():
+  # Try loading an invalid certificate.
+  with pytest.raises(CertInvalidException):
+    load_certificate('someinvalidcontents')
+
+  # Load a valid certificate.
+  (public_key_data, _) = generate_test_cert()
+
+  cert = load_certificate(public_key_data)
+  assert not cert.expired
+  assert cert.names == set(['somehostname'])
+  assert cert.matches_name('somehostname')
+
+def test_expired_certificate():
+  (public_key_data, _) = generate_test_cert(expires=-100)
+
+  cert = load_certificate(public_key_data)
+  assert cert.expired
+
+def test_hostnames():
+  (public_key_data, _) = generate_test_cert(hostname='foo', san_list=['DNS:bar', 'DNS:baz'])
+  cert = load_certificate(public_key_data)
+  assert cert.names == set(['foo', 'bar', 'baz'])
+
+  for name in cert.names:
+    assert cert.matches_name(name)
+
+def test_wildcard_hostnames():
+  (public_key_data, _) = generate_test_cert(hostname='foo', san_list=['DNS:*.bar'])
+  cert = load_certificate(public_key_data)
+  assert cert.names == set(['foo', '*.bar'])
+
+  for name in cert.names:
+    assert cert.matches_name(name)
+
+  assert cert.matches_name('something.bar')
+  assert cert.matches_name('somethingelse.bar')
+  assert cert.matches_name('cool.bar')
+  assert not cert.matches_name('*')
+
+def test_nondns_hostnames():
+  (public_key_data, _) = generate_test_cert(hostname='foo', san_list=['URI:yarg'])
+  cert = load_certificate(public_key_data)
+  assert cert.names == set(['foo'])
+
+def test_validate_private_key():
+  (public_key_data, private_key_data) = generate_test_cert()
+
+  private_key = NamedTemporaryFile(delete=True)
+  private_key.write(private_key_data)
+  private_key.seek(0)
+
+  cert = load_certificate(public_key_data)
+  cert.validate_private_key(private_key.name)
+
+def test_invalid_private_key():
+  (public_key_data, _) = generate_test_cert()
+
+  private_key = NamedTemporaryFile(delete=True)
+  private_key.write('somerandomdata')
+  private_key.seek(0)
+
+  cert = load_certificate(public_key_data)
+  with pytest.raises(KeyInvalidException):
+    cert.validate_private_key(private_key.name)
+
+def test_mismatch_private_key():
+  (public_key_data, _) = generate_test_cert()
+  (_, private_key_data) = generate_test_cert()
+
+  private_key = NamedTemporaryFile(delete=True)
+  private_key.write(private_key_data)
+  private_key.seek(0)
+
+  cert = load_certificate(public_key_data)
+  with pytest.raises(KeyInvalidException):
+    cert.validate_private_key(private_key.name)