Rename third_party/webpagereplay to third_party/web-page-replay

telemetry/third_party/webpagereplay/certutils.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.
-
-"""Routines to generate root and server certificates.
-
-Certificate Naming Conventions:
-  ca_cert:  crypto.X509 for the certificate authority (w/ both the pub &
-                priv keys)
-  cert:  a crypto.X509 certificate (w/ just the pub key)
-  cert_str:  a certificate string (w/ just the pub cert)
-  key:  a private crypto.PKey  (from ca or pem)
-  ca_cert_str:  a certificae authority string (w/ both the pub & priv certs)
-"""
-
-import logging
-import os
-import platform
-import socket
-import subprocess
-import time
-
-openssl_import_error = None
-
-Error = None
-SSL_METHOD = None
-SysCallError = None
-VERIFY_PEER = None
-ZeroReturnError = None
-FILETYPE_PEM = None
-
-try:
-  from OpenSSL import crypto, SSL
-
-  Error = SSL.Error
-  SSL_METHOD = SSL.SSLv23_METHOD
-  SysCallError = SSL.SysCallError
-  VERIFY_PEER = SSL.VERIFY_PEER
-  ZeroReturnError = SSL.ZeroReturnError
-  FILETYPE_PEM = crypto.FILETYPE_PEM
-except ImportError, e:
-  openssl_import_error = e
-
-
-def get_ssl_context(method=SSL_METHOD):
-  # One of: One of SSLv2_METHOD, SSLv3_METHOD, SSLv23_METHOD, or TLSv1_METHOD
-  if openssl_import_error:
-    raise openssl_import_error  # pylint: disable=raising-bad-type
-  return SSL.Context(method)
-
-
-class WrappedConnection(object):
-
-  def __init__(self, obj):
-    self._wrapped_obj = obj
-
-  def __getattr__(self, attr):
-    if attr in self.__dict__:
-      return getattr(self, attr)
-    return getattr(self._wrapped_obj, attr)
-
-  def recv(self, buflen=1024, flags=0):
-    try:
-      return self._wrapped_obj.recv(buflen, flags)
-    except SSL.SysCallError, e:
-      if e.args[1] == 'Unexpected EOF':
-        return ''
-      raise
-    except SSL.ZeroReturnError:
-      return ''
-
-
-def get_ssl_connection(context, connection):
-  return WrappedConnection(SSL.Connection(context, connection))
-
-
-def load_privatekey(key, filetype=FILETYPE_PEM):
-  """Loads obj private key object from string."""
-  return crypto.load_privatekey(filetype, key)
-
-
-def load_cert(cert_str, filetype=FILETYPE_PEM):
-  """Loads obj cert object from string."""
-  return crypto.load_certificate(filetype, cert_str)
-
-
-def _dump_privatekey(key, filetype=FILETYPE_PEM):
-  """Dumps obj private key object to string."""
-  return crypto.dump_privatekey(filetype, key)
-
-
-def _dump_cert(cert, filetype=FILETYPE_PEM):
-  """Dumps obj cert object to string."""
-  return crypto.dump_certificate(filetype, cert)
-
-
-def generate_dummy_ca_cert(subject='_WebPageReplayCert'):
-  """Generates dummy certificate authority.
-
-  Args:
-    subject: a string representing the desired root cert issuer
-  Returns:
-    A tuple of the public key and the private key strings for the root
-    certificate
-  """
-  if openssl_import_error:
-    raise openssl_import_error  # pylint: disable=raising-bad-type
-
-  key = crypto.PKey()
-  key.generate_key(crypto.TYPE_RSA, 1024)
-
-  ca_cert = crypto.X509()
-  ca_cert.set_serial_number(int(time.time()*10000))
-  ca_cert.set_version(2)
-  ca_cert.get_subject().CN = subject
-  ca_cert.get_subject().O = subject
-  ca_cert.gmtime_adj_notBefore(-60 * 60 * 24 * 365 * 2)
-  ca_cert.gmtime_adj_notAfter(60 * 60 * 24 * 365 * 2)
-  ca_cert.set_issuer(ca_cert.get_subject())
-  ca_cert.set_pubkey(key)
-  ca_cert.add_extensions([
-      crypto.X509Extension('basicConstraints', True, 'CA:TRUE'),
-      crypto.X509Extension('nsCertType', True, 'sslCA'),
-      crypto.X509Extension('extendedKeyUsage', True,
-                           ('serverAuth,clientAuth,emailProtection,'
-                            'timeStamping,msCodeInd,msCodeCom,msCTLSign,'
-                            'msSGC,msEFS,nsSGC')),
-      crypto.X509Extension('keyUsage', False, 'keyCertSign, cRLSign'),
-      crypto.X509Extension('subjectKeyIdentifier', False, 'hash',
-                           subject=ca_cert),
-      ])
-  ca_cert.sign(key, 'sha256')
-  key_str = _dump_privatekey(key)
-  ca_cert_str = _dump_cert(ca_cert)
-  return ca_cert_str, key_str
-
-
-def get_host_cert(host, port=443):
-  """Contacts the host and returns its certificate."""
-  host_certs = []
-  def verify_cb(conn, cert, errnum, depth, ok):
-    host_certs.append(cert)
-    # Return True to indicates that the certificate was ok.
-    return True
-
-  context = SSL.Context(SSL.SSLv23_METHOD)
-  context.set_verify(SSL.VERIFY_PEER, verify_cb)  # Demand a certificate
-  s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
-  connection = SSL.Connection(context, s)
-  try:
-    connection.connect((host, port))
-    connection.send('')
-  except SSL.SysCallError:
-    pass
-  except socket.gaierror:
-    logging.debug('Host name is not valid')
-  finally:
-    connection.shutdown()
-    connection.close()
-  if not host_certs:
-    logging.warning('Unable to get host certificate from %s:%s', host, port)
-    return ''
-  return _dump_cert(host_certs[-1])
-
-
-def write_dummy_ca_cert(ca_cert_str, key_str, cert_path):
-  """Writes four certificate files.
-
-  For example, if cert_path is "mycert.pem":
-      mycert.pem - CA plus private key
-      mycert-cert.pem - CA in PEM format
-      mycert-cert.cer - CA for Android
-      mycert-cert.p12 - CA in PKCS12 format for Windows devices
-  Args:
-    cert_path: path string such as "mycert.pem"
-    ca_cert_str: certificate string
-    key_str: private key string
-  """
-  dirname = os.path.dirname(cert_path)
-  if dirname and not os.path.exists(dirname):
-    os.makedirs(dirname)
-
-  root_path = os.path.splitext(cert_path)[0]
-  ca_cert_path = root_path + '-cert.pem'
-  android_cer_path = root_path + '-cert.cer'
-  windows_p12_path = root_path + '-cert.p12'
-
-  # Dump the CA plus private key
-  with open(cert_path, 'w') as f:
-    f.write(key_str)
-    f.write(ca_cert_str)
-
-  # Dump the certificate in PEM format
-  with open(ca_cert_path, 'w') as f:
-    f.write(ca_cert_str)
-
-  # Create a .cer file with the same contents for Android
-  with open(android_cer_path, 'w') as f:
-    f.write(ca_cert_str)
-
-  ca_cert = load_cert(ca_cert_str)
-  key = load_privatekey(key_str)
-  # Dump the certificate in PKCS12 format for Windows devices
-  with open(windows_p12_path, 'w') as f:
-    p12 = crypto.PKCS12()
-    p12.set_certificate(ca_cert)
-    p12.set_privatekey(key)
-    f.write(p12.export())
-
-
-def generate_cert(root_ca_cert_str, server_cert_str, server_host):
-  """Generates a cert_str with the sni field in server_cert_str signed by the
-  root_ca_cert_str.
-
-  Args:
-    root_ca_cert_str: PEM formatted string representing the root cert
-    server_cert_str: PEM formatted string representing cert
-    server_host: host name to use if there is no server_cert_str
-  Returns:
-    a PEM formatted certificate string
-  """
-  EXTENSION_WHITELIST = set(['subjectAltName'])
-
-  if openssl_import_error:
-    raise openssl_import_error  # pylint: disable=raising-bad-type
-
-  common_name = server_host
-  reused_extensions = []
-  if server_cert_str:
-    original_cert = load_cert(server_cert_str)
-    common_name = original_cert.get_subject().commonName
-    for i in xrange(original_cert.get_extension_count()):
-      original_cert_extension = original_cert.get_extension(i)
-      if original_cert_extension.get_short_name() in EXTENSION_WHITELIST:
-        reused_extensions.append(original_cert_extension)
-
-  ca_cert = load_cert(root_ca_cert_str)
-  ca_key = load_privatekey(root_ca_cert_str)
-
-  cert = crypto.X509()
-  cert.get_subject().CN = common_name
-  cert.gmtime_adj_notBefore(-60 * 60)
-  cert.gmtime_adj_notAfter(60 * 60 * 24 * 30)
-  cert.set_issuer(ca_cert.get_subject())
-  cert.set_serial_number(int(time.time()*10000))
-  cert.set_pubkey(ca_key)
-  cert.add_extensions(reused_extensions)
-  cert.sign(ca_key, 'sha256')
-
-  return _dump_cert(cert)
-
-
-def install_cert_in_nssdb(home_directory_path, certificate_path):
-  """Installs a certificate into the ~/.pki/nssdb database.
-
-  Args:
-    home_directory_path: Path of the home directory where to install
-    certificate_path: Path of a CA in PEM format
-  """
-  assert os.path.isdir(home_directory_path)
-  assert platform.system() == 'Linux', \
-      'SSL certification authority has only been tested for linux.'
-  if (os.path.abspath(home_directory_path) ==
-          os.path.abspath(os.environ['HOME'])):
-    raise Exception('Modifying $HOME/.pki/nssdb compromises your machine.')
-
-  cert_database_path = os.path.join(home_directory_path, '.pki', 'nssdb')
-  def certutil(args):
-    cmd = ['certutil', '--empty-password', '-d', 'sql:' + cert_database_path]
-    cmd.extend(args)
-    logging.info(subprocess.list2cmdline(cmd))
-    subprocess.check_call(cmd)
-
-  if not os.path.isdir(cert_database_path):
-    os.makedirs(cert_database_path)
-    certutil(['-N'])
-
-  certutil(['-A', '-t', 'PC,,', '-n', certificate_path, '-i', certificate_path])