certutils_test.py

# Copyright 2014 Google Inc. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#      http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

"""Test routines to generate dummy certificates."""

import BaseHTTPServer
import os
import shutil
import ssl
import tempfile
import threading
import unittest

import certutils


class Server(BaseHTTPServer.HTTPServer):

  def __init__(self, https_root_ca_cert_path):
    BaseHTTPServer.HTTPServer.__init__(
        self, ('localhost', 0), BaseHTTPServer.BaseHTTPRequestHandler)
    self.socket = ssl.wrap_socket(
        self.socket, certfile=https_root_ca_cert_path, server_side=True,
        do_handshake_on_connect=False)

  def __enter__(self):
    thread = threading.Thread(target=self.serve_forever)
    thread.daemon = True
    thread.start()
    return self

  def cleanup(self):
    try:
      self.shutdown()
    except KeyboardInterrupt:
      pass

  def __exit__(self, type_, value_, traceback_):
    self.cleanup()


class CertutilsTest(unittest.TestCase):

  def _check_cert_file(self, cert_file_path, cert_str, key_str=None):
    cert_load = open(cert_file_path, 'r').read()
    if key_str:
      expected_cert = key_str + cert_str
    else:
      expected_cert = cert_str
    self.assertEqual(expected_cert, cert_load)

  def setUp(self):
    self._temp_dir = tempfile.mkdtemp(prefix='certutils_', dir='/tmp')

  def tearDown(self):
    if self._temp_dir:
      shutil.rmtree(self._temp_dir)

  def test_generate_dummy_ca_cert(self):
    subject = 'testSubject'
    c, _ = certutils.generate_dummy_ca_cert(subject)
    c = certutils.load_cert(c)
    self.assertEqual(c.get_subject().commonName, subject)

  def test_get_host_cert(self):
    ca_cert_path = os.path.join(self._temp_dir, 'rootCA.pem')
    issuer = 'testCA'
    certutils.write_dummy_ca_cert(*certutils.generate_dummy_ca_cert(issuer),
                                  cert_path=ca_cert_path)

    with Server(ca_cert_path) as server:
      cert_str = certutils.get_host_cert('localhost', server.server_port)
      cert = certutils.load_cert(cert_str)
      self.assertEqual(issuer, cert.get_subject().commonName)

  def test_get_host_cert_gives_empty_for_bad_host(self):
    cert_str = certutils.get_host_cert('not_a_valid_host_name_2472341234234234')
    self.assertEqual('', cert_str)

  def test_write_dummy_ca_cert(self):
    base_path = os.path.join(self._temp_dir, 'testCA')
    ca_cert_path = base_path + '.pem'
    cert_path = base_path + '-cert.pem'
    ca_cert_android = base_path + '-cert.cer'
    ca_cert_windows = base_path + '-cert.p12'

    self.assertFalse(os.path.exists(ca_cert_path))
    self.assertFalse(os.path.exists(cert_path))
    self.assertFalse(os.path.exists(ca_cert_android))
    self.assertFalse(os.path.exists(ca_cert_windows))
    c, k = certutils.generate_dummy_ca_cert()
    certutils.write_dummy_ca_cert(c, k, ca_cert_path)

    self._check_cert_file(ca_cert_path, c, k)
    self._check_cert_file(cert_path, c)
    self._check_cert_file(ca_cert_android, c)
    self.assertTrue(os.path.exists(ca_cert_windows))

  def test_generate_cert(self):
    ca_cert_path = os.path.join(self._temp_dir, 'testCA.pem')
    issuer = 'testIssuer'
    certutils.write_dummy_ca_cert(
        *certutils.generate_dummy_ca_cert(issuer), cert_path=ca_cert_path)

    with open(ca_cert_path, 'r') as root_file:
      root_string = root_file.read()
    subject = 'testSubject'
    cert_string = certutils.generate_cert(
        root_string, '', subject)
    cert = certutils.load_cert(cert_string)
    self.assertEqual(issuer, cert.get_issuer().commonName)
    self.assertEqual(subject, cert.get_subject().commonName)

    with open(ca_cert_path, 'r') as ca_cert_file:
      ca_cert_str = ca_cert_file.read()
    cert_string = certutils.generate_cert(ca_cert_str, cert_string,
                                          'host.com')
    cert = certutils.load_cert(cert_string)
    self.assertEqual(issuer, cert.get_issuer().commonName)
    self.assertEqual(subject, cert.get_subject().commonName)
    self.assertEqual(2, cert.get_version())
    self.assertEqual(2, cert.get_extension_count())
    self.assertEqual(b"subjectAltName", cert.get_extension(0).get_short_name())
    self.assertEqual(b"extendedKeyUsage",
                     cert.get_extension(1).get_short_name())



if __name__ == '__main__':
  unittest.main()