Package contrail ::
Package security ::
Package onlineca ::
Package client
|
|
1 """Online CA service client package
2
3 Contrail Project
4 """
5 __author__ = "P J Kershaw"
6 __date__ = "28/05/12"
7 __copyright__ = "(C) 2012 Science and Technology Facilities Council"
8 __license__ = "BSD - see LICENSE file in top-level directory"
9 __contact__ = "Philip.Kershaw@stfc.ac.uk"
10 __revision__ = '$Id$'
11 import logging
12 log = logging.getLogger(__name__)
13 import base64
14 import os
15 import errno
16 import urllib2
17 from urlparse import urlparse, urlunparse
18
19 from OpenSSL import SSL, crypto
20
21 from ndg.httpsclient.ssl_context_util import make_ssl_context
22 from ndg.httpsclient.urllib2_build_opener import build_opener
23 from ndg.httpsclient.utils import (_should_use_proxy, fetch_stream_from_url,
24 Configuration)
28 PRIKEY_NBITS = 2048
29 MESSAGE_DIGEST_TYPE = "md5"
30 CERT_REQ_POST_PARAM_KEYNAME = 'certificate_request'
31 TRUSTED_CERTS_FIELDNAME = 'TRUSTED_CERTS'
32 TRUSTED_CERTS_FILEDATA_FIELDNAME_PREFIX = 'FILEDATA_'
33 SSL_METHOD = SSL.TLSv1_METHOD
34
36 self.__ca_cert_dir = None
37
38 @property
40 return self.__ca_cert_dir
41
42 @ca_cert_dir.setter
44 if not isinstance(val, basestring):
45 raise TypeError('Expecting string type for "ca_cert_dir"; got %r' %
46 type(val))
47
48 self.__ca_cert_dir = val
49
50 @staticmethod
52 """Generate key pair and return as PEM encoded string
53 @type n_bits_for_key: int
54 @param n_bits_for_key: number of bits for private key generation -
55 default is 2048
56 @rtype: OpenSSL.crypto.PKey
57 @return: public/private key pair
58 """
59 key_pair = crypto.PKey()
60 key_pair.generate_key(crypto.TYPE_RSA, n_bits_for_key)
61
62 return key_pair
63
64 @staticmethod
66 """Create a certificate request.
67
68 @type keyPair: string/None
69 @param keyPair: public/private key pair
70 @type messageDigest: basestring
71 @param messageDigest: message digest type - default is MD5
72 @rtype: base string
73 @return certificate request PEM text and private key PEM text
74 """
75
76
77
78 cert_req = crypto.X509Req()
79
80
81 cert_req.set_pubkey(key_pair)
82
83
84 cert_req.sign(key_pair, message_digest)
85
86 cert_req = crypto.dump_certificate_request(crypto.FILETYPE_PEM,
87 cert_req)
88
89 return cert_req
90
91 - def logon(self, username, password, server_url, proxies=None, no_proxy=None,
92 cert_life_time=86400, ssl_ctx=None, pem_out_filepath=None):
93 """Obtain a create a new key pair and invoke the SLCS service to obtain
94 a certificate
95 """
96 if ssl_ctx is None:
97 ssl_ctx = make_ssl_context(ca_dir=self.ca_cert_dir,
98 verify_peer=True,
99 url=server_url,
100 method=self.__class__.SSL_METHOD)
101
102
103 password_mgr = urllib2.HTTPPasswordMgrWithDefaultRealm()
104
105
106 parsed_url = urlparse(server_url)
107 base_url = urlunparse(parsed_url[0:2] + ('/', '', '', ''))
108
109
110
111 password_mgr.add_password(None, base_url, username, password)
112
113 handlers = [urllib2.HTTPBasicAuthHandler(password_mgr)]
114
115 key_pair = self.__class__.create_key_pair()
116 cert_req = self.__class__.create_cert_req(key_pair)
117
118
119 encoded_cert_req = cert_req.replace('+', '%2B')
120 req = "%s=%s\n" % (self.__class__.CERT_REQ_POST_PARAM_KEYNAME,
121 encoded_cert_req)
122 config = Configuration(ssl_ctx, True)
123 res = fetch_stream_from_url(server_url, config, data=req,
124 handlers=handlers)
125
126 pem_out = res.read()
127 cert = crypto.load_certificate(crypto.FILETYPE_PEM, pem_out)
128
129
130
131 if pem_out_filepath:
132 pem_pkey = crypto.dump_privatekey(crypto.FILETYPE_PEM, key_pair)
133 pem_cert = crypto.dump_certificate(crypto.FILETYPE_PEM, cert)
134
135 with open(pem_out_filepath, 'w', 0400) as pem_out_file:
136 pem_out_file.write(pem_pkey)
137 pem_out_file.write(pem_cert)
138
139 return key_pair, cert
140
141 - def get_trustroots(self, server_url, write_to_ca_cert_dir=False,
142 bootstrap=False):
143 """Get trustroots"""
144 if bootstrap:
145 ca_cert_dir = None
146 else:
147 ca_cert_dir = self.ca_cert_dir
148
149 ssl_ctx = make_ssl_context(ca_cert_dir,
150 verify_peer=not bootstrap,
151 url=server_url,
152 method=self.__class__.SSL_METHOD)
153
154 config = Configuration(ssl_ctx, True)
155 res = fetch_stream_from_url(server_url, config)
156
157 prefix = self.__class__.TRUSTED_CERTS_FILEDATA_FIELDNAME_PREFIX
158 field_name = self.__class__.TRUSTED_CERTS_FIELDNAME
159
160 files_dict = {}
161 for line in res.readlines():
162 file_name, enc_file_content = line.strip().split('=', 1)
163 files_dict[file_name] = base64.b64decode(enc_file_content)
164
165 if write_to_ca_cert_dir:
166
167 try:
168 os.makedirs(self.ca_cert_dir)
169 except OSError, e:
170
171 if e.errno != errno.EEXIST:
172 raise
173
174 for file_name, file_contents in files_dict.items():
175 file_path = os.path.join(self.ca_cert_dir, file_name)
176 open(file_path, 'wb').write(file_contents)
177
178 return files_dict
179