From 529adb3e0d763a8ee9ba9b4c5b13f933d723e8de Mon Sep 17 00:00:00 2001 From: Dan Lavu Date: Fri, 5 Feb 2016 08:51:07 -0500 Subject: [PATCH] Adding SSL encryption to integration tests. --- src/tests/intg/ca.py | 166 ++++++++++++++++++++++++++++++++++++++++++ src/tests/intg/ds_openldap.py | 14 ++++ 2 files changed, 180 insertions(+) create mode 100644 src/tests/intg/ca.py diff --git a/src/tests/intg/ca.py b/src/tests/intg/ca.py new file mode 100644 index 0000000000000000000000000000000000000000..a44a92e5d5053338dabd7d8d82d2b1d50ec7594e --- /dev/null +++ b/src/tests/intg/ca.py @@ -0,0 +1,166 @@ +# +# SSSD LOCAL domain tests +# +# Copyright (c) 2016 Red Hat, Inc. +# Author: Dan Lavu +# +# This is free software; you can redistribute it and/or modify it +# under the terms of the GNU General Public License as published by +# the Free Software Foundation; version 2 only +# +# This program is distributed in the hope that it will be useful, but +# WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU +# General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with this program. If not, see . +# + +from OpenSSL import crypto +from os.path import exists, join + +import socket +import os +import fnmatch + + +class CA: + """CA Class""" + + def __init__(self, subject=None, country=None, state=None, + city=None, organization=None, unit=None, config_dir=None): + if subject is None: + self.subject = socket.gethostname() + if country is None: + self.country = 'US' + if state is None: + self.state = 'NC' + if city is None: + self.city = 'Raleigh' + if organization is None: + self.organization = 'Red Hat' + if unit is None: + self.unit = 'SSSD' + if config_dir is None: + self.config_dir = '/etc/pki' + + self.hostname = socket.gethostname() + self.csr_dir = self.config_dir + '/CA/newcerts' + self.key_dir = self.config_dir + '/tls/private' + self.cert_dir = self.config_dir + '/tls/certs' + + self.index = int(1000) + + + def setup(self): + """Setup CA using OpenSSL""" + cacert = socket.gethostname() + '-ca.crt' + cakey = socket.gethostname() + '-ca.key' + + if not exists(join(self.cert_dir, cacert)) or not exists(join(self.key_dir, cakey)): + key = crypto.PKey() + key.generate_key(crypto.TYPE_RSA, 2048) + + ca = crypto.X509() + ca.get_subject().C = self.country + ca.get_subject().ST = self.state + ca.get_subject().L = self.city + ca.get_subject().O = self.organization + ca.get_subject().OU = self.unit + ca.get_subject().CN = self.subject + ca.set_serial_number(self.index) + ca.gmtime_adj_notBefore(0) + ca.gmtime_adj_notAfter(10 * 365 * 24 * 60 * 60) + ca.set_issuer(ca.get_subject()) + ca.set_pubkey(key) + ca.sign(key, 'sha1') + + open(os.path.join(self.cert_dir, cacert), 'wt').write \ + (crypto.dump_certificate(crypto.FILETYPE_PEM, ca)) + open(os.path.join(self.key_dir, cakey), 'wt').write \ + (crypto.dump_privatekey(crypto.FILETYPE_PEM, key)) + + + def teardown(self): + """Teardown CA certificate files""" + for i in os.listdir(self.csr_dir): + if fnmatch.fnmatch(i, socket.gethostname() + '.*'): + os.unlink(os.path.join(self.csr_dir, i)) + for i in os.listdir(self.key_dir): + if fnmatch.fnmatch(i, socket.gethostname() + '.*'): + os.unlink(os.path.join(self.key_dir, i)) + for i in os.listdir(self.cert_dir): + if fnmatch.fnmatch(i, socket.gethostname() + '.*'): + os.unlink(os.path.join(self.cert_dir, i)) + + + def get_ca(self): + """Returns CA certificate in ASCII PEM encoding""" + ca = crypto.load_certificate(crypto.FILETYPE_PEM, open\ + (os.path.join(self.cert_dir, socket.gethostname() + '-ca.crt'), 'rt').read()) + ca_crt = crypto.dump_certificate(crypto.FILETYPE_PEM, ca) + + return ca_crt + + + def get_cert(self, csr, text=False): + """Retrieves certificate + required: csr -csr_type + optional: text + False - will return a pem object type (Default) + True - will return the certificate as ASCII + """ + cacert = socket.gethostname() + '-ca.crt' + cakey = socket.gethostname() + '-ca.key' + ca_crt = crypto.load_certificate(crypto.FILETYPE_PEM, open(os.path.join(self.cert_dir, cacert), 'rt').read()) + ca_key = crypto.load_privatekey(crypto.FILETYPE_PEM, open(os.path.join(self.key_dir, cakey), 'rt').read()) + + cert = crypto.X509() + cert.set_subject(csr.get_subject()) + cert.set_serial_number(self.index+1) + cert.gmtime_adj_notBefore(0) + cert.gmtime_adj_notAfter(10 * 365 * 24 * 60 * 60) + cert.set_issuer(ca_crt.get_subject()) + cert.set_pubkey(csr.get_pubkey()) + cert.sign(ca_key, 'sha1') + + open(os.path.join(self.cert_dir, socket.gethostname() + '.crt'), 'wt')\ + .write(crypto.dump_certificate(crypto.FILETYPE_PEM, cert)) + server_crt = crypto.dump_certificate(crypto.FILETYPE_PEM, cert) + + if text is True: + return server_crt + else: + return cert + + + def request_cert(self, fqdn=None, text=False): + """Generates CSR + optional: fqdn, socket.gethostname() (Default) + optional: text + False - will return a pem object type (Default) + True - will return the certificate as ASCII + """ + if fqdn is None: + fqdn = socket.getfqdn() + + hostname = socket.gethostname() + key = crypto.PKey() + key.generate_key(crypto.TYPE_RSA, 2048) + csr = crypto.X509Req() + csr.get_subject().CN = fqdn + csr.set_pubkey(key) + csr.sign(key, 'sha1') + + open(os.path.join(self.key_dir, hostname + '.key'), 'wt').\ + write(crypto.dump_privatekey(crypto.FILETYPE_PEM, key)) + open(os.path.join(self.csr_dir, hostname + '.csr'), 'wt').\ + write(crypto.dump_certificate_request(crypto.FILETYPE_PEM, csr)) + + server_csr = crypto.dump_certificate_request(crypto.FILETYPE_PEM, csr) + + if text is True: + return server_csr + else: + return csr \ No newline at end of file diff --git a/src/tests/intg/ds_openldap.py b/src/tests/intg/ds_openldap.py index fb230a081b58bd4e135585daa5a6ddf8f494861c..bc3d9e3b78c41b9dabd5fbbf438694d048b673fd 100644 --- a/src/tests/intg/ds_openldap.py +++ b/src/tests/intg/ds_openldap.py @@ -27,8 +27,10 @@ import errno import signal import shutil import sys +import socket from util import * from ds import DS +from ca import CA def hash_password(password): @@ -60,6 +62,9 @@ class DSOpenLDAP(DS): self.conf_dir = self.dir + "/etc/ldap" self.conf_slapd_d_dir = self.conf_dir + "/slapd.d" self.data_dir = self.dir + "/var/lib/ldap" + self.ca_inst = CA() + + def _setup_config(self): """Setup the instance initial configuration.""" @@ -73,6 +78,10 @@ class DSOpenLDAP(DS): uid = os.geteuid() gid = os.getegid() + self.ca_inst.setup() + self.ca_inst.get_cert(self.ca_inst.request_cert()) + fqdn = socket.gethostname() + # # Add configuration # @@ -82,6 +91,9 @@ class DSOpenLDAP(DS): cn: config olcPidFile: {self.pid_path} olcArgsFile: {args_file} + olcTLSCertificateKeyFile: /etc/pki/tls/private/{fqdn}.key + olcTLSCertificateFile: /etc/pki/tls/certs/{fqdn}.crt + olcTLSCACertificatePath: /etc/pki/tls/certs/{fqdn}-ca.crt # Read slapd.conf(5) for possible values olcLogLevel: none @@ -282,3 +294,5 @@ class DSOpenLDAP(DS): for path in (self.conf_slapd_d_dir, self.run_dir, self.data_dir): shutil.rmtree(path, True) + + self.ca_inst.teardown() -- 1.8.3.1