1
0
mirror of https://github.com/webrecorder/pywb.git synced 2025-03-15 08:04:49 +01:00

certauth: clean up CertificatAuthority, add cli interface for creating root cert and host certs

CertificateAuthority instance creates per-host certs, assume root cert exists
static method generate_ca_root() used to create root cert once
add proxy_options to enable https support
This commit is contained in:
Ilya Kreymer 2014-07-27 19:35:16 -07:00
parent 2a9197137e
commit b6fb0e510e
3 changed files with 235 additions and 115 deletions

View File

@ -1,105 +0,0 @@
import logging
import os
import OpenSSL
import random
#=================================================================
class CertificateAuthority(object):
logger = logging.getLogger('pywb.CertificateAuthority')
def __init__(self, ca_file='pywb-ca.pem',
certs_dir='./pywb-ca',
certname='pywb CA'):
self.ca_file = ca_file
self.certs_dir = certs_dir
self.certname = certname
if not os.path.exists(ca_file):
self._generate_ca()
else:
self._read_ca(ca_file)
if not os.path.exists(certs_dir):
os.mkdir(certs_dir)
def _generate_ca(self):
# Generate key
self.key = OpenSSL.crypto.PKey()
self.key.generate_key(OpenSSL.crypto.TYPE_RSA, 2048)
# Generate certificate
self.cert = OpenSSL.crypto.X509()
self.cert.set_version(3)
# avoid sec_error_reused_issuer_and_serial
self.cert.set_serial_number(random.randint(0,2**64-1))
self.cert.get_subject().CN = self.certname
self.cert.gmtime_adj_notBefore(0) # now
self.cert.gmtime_adj_notAfter(100*365*24*60*60) # 100 yrs in future
self.cert.set_issuer(self.cert.get_subject())
self.cert.set_pubkey(self.key)
self.cert.add_extensions([
OpenSSL.crypto.X509Extension(b"basicConstraints",
True,
b"CA:TRUE, pathlen:0"),
OpenSSL.crypto.X509Extension(b"keyUsage",
True,
b"keyCertSign, cRLSign"),
OpenSSL.crypto.X509Extension(b"subjectKeyIdentifier",
False,
b"hash",
subject=self.cert),
])
self.cert.sign(self.key, "sha1")
with open(self.ca_file, 'wb+') as f:
f.write(OpenSSL.crypto.dump_privatekey(OpenSSL.SSL.FILETYPE_PEM,
self.key))
f.write(OpenSSL.crypto.dump_certificate(OpenSSL.SSL.FILETYPE_PEM,
self.cert))
def _read_ca(self, filename):
with open(filename) as cert_fh:
self.cert = OpenSSL.crypto.load_certificate(
OpenSSL.SSL.FILETYPE_PEM, cert_fh.read())
cert_fh.seek(0)
self.key = OpenSSL.crypto.load_privatekey(
OpenSSL.SSL.FILETYPE_PEM, cert_fh.read())
def __getitem__(self, cn):
cnp = os.path.sep.join([self.certs_dir, '%s.pem' % cn])
if not os.path.exists(cnp):
# create certificate
key = OpenSSL.crypto.PKey()
key.generate_key(OpenSSL.crypto.TYPE_RSA, 2048)
# Generate CSR
req = OpenSSL.crypto.X509Req()
req.get_subject().CN = cn
req.set_pubkey(key)
req.sign(key, 'sha1')
# Sign CSR
cert = OpenSSL.crypto.X509()
cert.set_subject(req.get_subject())
cert.set_serial_number(random.randint(0,2**64-1))
cert.gmtime_adj_notBefore(0)
cert.gmtime_adj_notAfter(10*365*24*60*60)
cert.set_issuer(self.cert.get_subject())
cert.set_pubkey(req.get_pubkey())
cert.sign(self.key, 'sha1')
with open(cnp, 'wb+') as f:
f.write(OpenSSL.crypto.dump_privatekey(
OpenSSL.SSL.FILETYPE_PEM, key))
f.write(OpenSSL.crypto.dump_certificate(
OpenSSL.SSL.FILETYPE_PEM, cert))
return cnp

201
pywb/framework/certauth.py Normal file
View File

@ -0,0 +1,201 @@
import logging
import os
from OpenSSL import crypto
from OpenSSL.SSL import FILETYPE_PEM
import random
from argparse import ArgumentParser
#=================================================================
# Duration of 100 years
CERT_DURATION = 100 * 365 * 24 * 60 * 60
CERTS_DIR = './pywb-certs/'
CERT_NAME = 'pywb https proxy replay CA'
CERT_CA_FILE = './pywb-ca.pem'
#=================================================================
class CertificateAuthority(object):
"""
Utility class for signing individual certificate
with a root cert.
Static generate_ca_root() method for creating the root cert
All certs saved on filesystem. Individual certs are stored
in specified certs_dir and reused if previously created.
"""
def __init__(self, ca_file, certs_dir):
if not ca_file:
ca_file = CERT_CA_FILE
if not certs_dir:
certs_dir = CERTS_DIR
self.ca_file = ca_file
self.certs_dir = certs_dir
# read previously created root cert
self.cert, self.key = self.read_pem(ca_file)
if not os.path.exists(certs_dir):
os.mkdir(certs_dir)
def get_cert_for_host(self, host, overwrite=False):
host_filename = os.path.sep.join([self.certs_dir, '%s.pem' % host])
if not overwrite and os.path.exists(host_filename):
return False, host_filename
self.generate_host_cert(host, self.cert, self.key, host_filename)
return True, host_filename
@staticmethod
def _make_cert(certname):
cert = crypto.X509()
cert.set_version(3)
cert.set_serial_number(random.randint(0, 2 ** 64 - 1))
cert.get_subject().CN = certname
cert.gmtime_adj_notBefore(0)
cert.gmtime_adj_notAfter(CERT_DURATION)
return cert
@staticmethod
def generate_ca_root(ca_file, certname=None, overwrite=False):
if not certname:
certname = CERT_NAME
if not ca_file:
ca_file = CERT_CA_FILE
if not overwrite and os.path.exists(ca_file):
cert, key = CertificateAuthority.read_pem(ca_file)
return False, cert, key
# Generate key
key = crypto.PKey()
key.generate_key(crypto.TYPE_RSA, 2048)
# Generate cert
cert = CertificateAuthority._make_cert(certname)
cert.set_issuer(cert.get_subject())
cert.set_pubkey(key)
cert.add_extensions([
crypto.X509Extension(b"basicConstraints",
True,
b"CA:TRUE, pathlen:0"),
crypto.X509Extension(b"keyUsage",
True,
b"keyCertSign, cRLSign"),
crypto.X509Extension(b"subjectKeyIdentifier",
False,
b"hash",
subject=cert),
])
cert.sign(key, "sha1")
# Write cert + key
CertificateAuthority.write_pem(ca_file, cert, key)
return True, cert, key
@staticmethod
def generate_host_cert(host, root_cert, root_key, host_filename):
# Generate key
key = crypto.PKey()
key.generate_key(crypto.TYPE_RSA, 2048)
# Generate CSR
req = crypto.X509Req()
req.get_subject().CN = host
req.set_pubkey(key)
req.sign(key, 'sha1')
# Generate Cert
cert = CertificateAuthority._make_cert(host)
cert.set_issuer(root_cert.get_subject())
cert.set_pubkey(req.get_pubkey())
cert.sign(root_key, 'sha1')
# Write cert + key
CertificateAuthority.write_pem(host_filename, cert, key)
return cert, key
@staticmethod
def write_pem(filename, cert, key):
with open(filename, 'wb+') as f:
f.write(crypto.dump_privatekey(FILETYPE_PEM, key))
f.write(crypto.dump_certificate(FILETYPE_PEM, cert))
@staticmethod
def read_pem(filename):
with open(filename, 'r') as f:
cert = crypto.load_certificate(FILETYPE_PEM, f.read())
f.seek(0)
key = crypto.load_privatekey(FILETYPE_PEM, f.read())
return cert, key
#=================================================================
def main():
parser = ArgumentParser(description='Cert Auth Cert Maker')
parser.add_argument('output_file', help='path to certificate file')
parser.add_argument('-r', '--use-root',
help='use specified root cert to create signed cert')
parser.add_argument('-n', '--name', action='store', default=CERT_NAME,
help='name for root certificate')
parser.add_argument('-d', '--certs-dir', default=CERTS_DIR)
parser.add_argument('-f', '--force', action='store_true')
result = parser.parse_args()
overwrite = result.force
# Create a new signed certificate using specified root
if result.use_root:
certs_dir = result.certs_dir
ca = CertificateAuthority(ca_file=result.use_root,
certs_dir=result.certs_dir,
certname=result.name)
created, host_filename = ca.get_cert_for_host(result.output_file,
overwrite)
if created:
print ('Created new cert "' + host_filename +
'" signed by root cert ' +
result.use_root)
else:
print ('Cert "' + host_filename + '" already exists,' +
' use -f to overwrite')
# Create new root certificate
else:
created, c, k = (CertificateAuthority.
generate_ca_root(result.output_file,
result.name,
overwrite))
if created:
print 'Created new root cert: "' + result.output_file + '"'
else:
print ('Root cert "' + result.output_file + '" already exists,' +
' use -f to overwrite')
if __name__ == "__main__":
main()

View File

@ -12,7 +12,7 @@ from pywb.rewrite.url_rewriter import HttpsUrlRewriter
from pywb.utils.statusandheaders import StatusAndHeaders
from pywb.utils.wbexception import BadRequestException
from certa import CertificateAuthority
from certauth import CertificateAuthority
#=================================================================
@ -68,8 +68,19 @@ class ProxyRouter(object):
self.unaltered = proxy_options.get('unaltered_replay', False)
self.ca = CertificateAuthority()
if proxy_options.get('enable_https_proxy'):
ca_file = proxy_options.get('root_ca_file')
# attempt to create the root_ca_file if doesn't exist
# (generally recommended to create this seperately)
certname = proxy_options.get('root_ca_name')
CertificateAuthority.generate_ca_root(certname, ca_file)
certs_dir = proxy_options.get('certs_dir')
self.ca = CertificateAuthority(ca_file=ca_file,
certs_dir=certs_dir)
else:
self.ca = None
def __call__(self, env):
is_https = (env['REQUEST_METHOD'] == 'CONNECT')
@ -119,7 +130,9 @@ class ProxyRouter(object):
# do connect, then get updated url
if is_https:
self.handle_connect(env)
response = self.handle_connect(env)
if response:
return response
url = env['REL_REQUEST_URI']
@ -142,14 +155,23 @@ class ProxyRouter(object):
return route.handler(wbrequest)
def get_request_socket(self, env):
if not self.ca:
return None
sock = None
if env.get('uwsgi.version'):
import uwsgi
fd = uwsgi.connection_fd()
conn = socket.fromfd(fd, socket.AF_INET, socket.SOCK_STREAM)
sock = socket.socket(_sock=conn)
try:
import uwsgi
fd = uwsgi.connection_fd()
conn = socket.fromfd(fd, socket.AF_INET, socket.SOCK_STREAM)
sock = socket.socket(_sock=conn)
except Exception:
pass
elif env.get('gunicorn.socket'):
sock = env['gunicorn.socket']
else:
if not sock:
# attempt to find socket from wsgi.input
input_ = env.get('wsgi.input')
if input_ and hasattr(input_, '_sock'):
@ -168,9 +190,11 @@ class ProxyRouter(object):
sock.send('\r\n')
hostname = env['REL_REQUEST_URI'].split(':')[0]
created, certfile = self.ca.get_cert_for_host(hostname)
ssl_sock = ssl.wrap_socket(sock, server_side=True,
certfile=self.ca[hostname])
ssl_sock = ssl.wrap_socket(sock,
server_side=True,
certfile=certfile)
#ssl_version=ssl.PROTOCOL_SSLv23)
env['pywb.proxy_ssl_sock'] = ssl_sock