Projekt

Obecné

Profil

Stáhnout (17.2 KB) Statistiky
| Větev: | Tag: | Revize:
1
import re
2
import subprocess
3
import time
4
import random
5

    
6
from src.model.subject import Subject
7
from src.utils.temporary_file import TemporaryFile
8

    
9
# encryption method to be used when generating private keys
10
PRIVATE_KEY_ENCRYPTION_METHOD = "-aes256"
11

    
12
# openssl executable name
13
OPENSSL_EXECUTABLE = "openssl"
14

    
15
# format of NOT_BEFORE NOT_AFTER date fields
16
NOT_AFTER_BEFORE_DATE_FORMAT = "%b %d %H:%M:%S %Y %Z"
17

    
18
# minimal configuration file to be used for openssl req command
19
# specifies distinguished_name that references empty section only
20
# openssl requires this option to be present
21
MINIMAL_CONFIG_FILE = "[req]\ndistinguished_name = req_distinguished_name\n[req_distinguished_name]\n\n"
22

    
23
# section to be used to specify extensions when creating a SSCRT
24
SSCRT_SECTION = "sscrt_ext"
25

    
26
CA_EXTENSIONS = "basicConstraints=critical,CA:TRUE"
27

    
28
# upper bound of the range of random serial numbers to be generated
29
MAX_SN = 4294967296
30

    
31

    
32
class CryptographyService:
33

    
34
    @staticmethod
35
    def __subject_to_param_format(subject):
36
        """
37
        Converts the given subject to a dictionary containing openssl field names mapped to subject's fields
38
        :param subject: subject to be converted
39
        :return: a dictionary containing openssl field names mapped to subject's fields
40
        """
41
        subj_dict = {}
42
        if subject.common_name is not None:
43
            subj_dict["CN"] = subject.common_name
44
        if subject.country is not None:
45
            subj_dict["C"] = subject.country
46
        if subject.locality is not None:
47
            subj_dict["L"] = subject.locality
48
        if subject.state is not None:
49
            subj_dict["ST"] = subject.state
50
        if subject.organization is not None:
51
            subj_dict["O"] = subject.organization
52
        if subject.organization_unit is not None:
53
            subj_dict["OU"] = subject.organization_unit
54
        if subject.email_address is not None:
55
            subj_dict["emailAddress"] = subject.email_address
56

    
57
        # merge the subject into a "subj" parameter format
58
        return "".join([f"/{key}={value}" for key, value in subj_dict.items()])
59

    
60
    @staticmethod
61
    def __run_for_output(args=None, proc_input=None, executable=OPENSSL_EXECUTABLE):
62
        """
63
        Launches a new process in which the given executable is run. STDIN and process arguments can be set.
64
        If the process ends with a non-zero then <CryptographyException> is raised.
65

    
66
        :param args: Arguments to be passed to the program.
67
        :param proc_input: String input to be passed to the stdin of the created process.
68
        :param executable: Executable to be run (defaults to openssl)
69
        :return: If the process ends with a zero return code then the STDOUT of the process is returned as a byte array.
70
        """
71
        if args is None:
72
            args = []
73
        try:
74
            # prepend the name of the executable
75
            args.insert(0, executable)
76

    
77
            # create a new process
78
            proc = subprocess.Popen(args, stdin=subprocess.PIPE if proc_input is not None else None,
79
                                    stdout=subprocess.PIPE,
80
                                    stderr=subprocess.PIPE)
81

    
82
            out, err = proc.communicate(proc_input)
83

    
84
            if proc.returncode != 0:
85
                # if the process did not result in zero result code, then raise an exception
86
                if err is not None and len(err) > 0:
87
                    raise CryptographyException(executable, args, err.decode())
88
                else:
89
                    raise CryptographyException(executable, args,
90
                                                f""""Execution resulted in non-zero argument""")
91

    
92
            return out
93
        except FileNotFoundError:
94
            raise CryptographyException(executable, args, f""""{executable}" not found in the current PATH.""")
95

    
96
    def create_private_key(self, passphrase=None):
97
        """
98
        Creates a private key with the option to encrypt it using a passphrase.
99
        :param passphrase: A passphrase to be used when encrypting the key (if none is passed then the key is not
100
        encrypted at all). Empty passphrase ("") also results in a key that is not encrypted.
101
        :return: string containing the generated private key in PEM format
102
        """
103
        if passphrase is None or len(passphrase) == 0:
104
            return self.__run_for_output(["genrsa", "2048"]).decode()
105
        else:
106
            return self.__run_for_output(
107
                ["genrsa", PRIVATE_KEY_ENCRYPTION_METHOD, "-passout", f"pass:{passphrase}", "2048"]).decode()
108

    
109
    def create_sscrt(self, subject, key, config="", extensions="", key_pass=None, days=30, sn: int =None):
110
        """
111
        Creates a root CA
112

    
113
        :param subject: an instance of <Subject> representing the subject to be added to the certificate
114
        :param key: private key of the CA to be used
115
        :param config: string containing the configuration to be used
116
        :param extensions: name of the section in the configuration representing extensions
117
        :param key_pass: passphrase of the private key
118
        :param days: number of days for which the certificate will be valid
119
        :param sn: serial number to be set, when "None" is set a random serial number is generated
120

    
121
        :return: string containing the generated certificate in PEM format
122
        """
123
        assert key is not None
124
        assert subject is not None
125

    
126
        subj = self.__subject_to_param_format(subject)
127

    
128
        # To specify extension for creating a SSCRT, one has to use a configuration
129
        # file instead of an extension file. Therefore the following code creates
130
        # the most basic configuration file with sscrt_ext section, that is later
131
        # reference in openssl req command using -extensions option.
132
        extensions += "\n"+CA_EXTENSIONS
133
        if len(config) == 0:
134
            config += MINIMAL_CONFIG_FILE
135
        config += "\n[ " + SSCRT_SECTION + " ]" + "\n" + extensions
136

    
137
        with TemporaryFile("openssl.conf", config) as conf_path:
138
            args = ["req", "-x509", "-new", "-subj", subj, "-days", f"{days}",
139
                    "-key", "-"]
140

    
141
            # serial number passed, use it when generating the certificate,
142
            # without passing it openssl generates a random one
143
            if sn is not None:
144
                args.extend(["-set_serial", str(sn)])
145

    
146
            if len(config) > 0:
147
                args.extend(["-config", conf_path])
148
            if len(extensions) > 0:
149
                args.extend(["-extensions", SSCRT_SECTION]) # when creating SSCRT, section references section in config
150

    
151
            # it would be best to not send the pass phrase at all, but for some reason pytest then prompts for
152
            # the pass phrase (this does not happen when run from pycharm)
153

    
154
            #  add the passphrase even when None is passed. Otherwise when running tests with pytest some tests freeze
155
            # waiting for the passphrase to be typed in
156
            args.extend(["-passin", f"pass:{key_pass}"])
157

    
158
            return self.__run_for_output(args, proc_input=bytes(key, encoding="utf-8")).decode()
159

    
160
    def __create_csr(self, subject, key, key_pass=""):
161
        """
162
        Creates a CSR (Certificate Signing Request)
163

    
164
        :param subject: an instance of <Subject> representing the subject to be added to the CSR
165
        :param key: the private key of the subject to be used to generate the CSR
166
        :param key_pass: passphrase of the subject's private key
167
        :return: string containing the generated certificate signing request in PEM format
168
        """
169

    
170
        subj_param = self.__subject_to_param_format(subject)
171

    
172
        args = ["req", "-new", "-subj", subj_param, "-key", "-"]
173

    
174
        # add the passphrase even when None is passed. Otherwise when running tests with pytest some tests freeze
175
        # waiting for the passphrase to be typed in
176
        args.extend(["-passin", f"pass:{key_pass}"])
177

    
178
        return self.__run_for_output(args, proc_input=bytes(key, encoding="utf-8")).decode()
179

    
180
    def __sign_csr(self, csr, issuer_pem, issuer_key, issuer_key_pass=None, extensions="", days=30, sn: int = None):
181
        """
182
        Signs the given CSR by the given issuer CA
183

    
184
        :param csr: a string containing the CSR to be signed
185
        :param issuer_pem: string containing the certificate of the issuer that will sign this CSR in PEM format
186
        :param issuer_key: string containing the private key of the issuer's certificate in PEM format
187
        :param issuer_key_pass: string containing the passphrase of the private key of the issuer's certificate in PEM
188
        format
189
        :param extensions: extensions to be applied when signing the CSR
190
        :param days: number of days for which the certificate will be valid
191
        :param sn: serial number to be set, when "None" is set a random serial number is generated
192
        :return: string containing the generated and signed certificate in PEM format
193
        """
194

    
195
        # concatenate CSR, issuer certificate and issuer's key (will be used in the openssl call)
196
        proc_input = csr + issuer_pem + issuer_key
197

    
198
        # TODO find a better way to generate a random serial number or let openssl generate a .srl file
199
        # when serial number is not passed generate a random one
200
        if sn is None:
201
            sn = random.randint(0, MAX_SN)
202

    
203
        # prepare openssl parameters...
204
        # CSR, CA and CA's private key will be passed via stdin (that's the meaning of the '-' symbol)
205
        params = ["x509", "-req", "-in", "-", "-CA", "-", "-CAkey", "-", "-CAcreateserial", "-days", str(days),
206
                  "-set_serial", str(sn)]
207

    
208
        # TODO delete created -.srl file
209

    
210
        with TemporaryFile("extensions.conf", extensions) as ext_path:
211
            # add the passphrase even when None is passed. Otherwise when running tests with pytest some tests freeze
212
            # waiting for the passphrase to be typed in
213
            params.extend(["-passin", f"pass:{issuer_key_pass}"])
214

    
215
            if len(extensions) > 0:
216
                params.extend(["-extfile", ext_path])
217

    
218
            return self.__run_for_output(params, proc_input=(bytes(proc_input, encoding="utf-8"))).decode()
219

    
220
    def create_crt(self, subject, subject_key, issuer_pem, issuer_key, subject_key_pass=None, issuer_key_pass=None,
221
                   extensions="",
222
                   days=30,
223
                   sn: int = None):
224
        """
225
        Creates a certificate by using the given subject, subject's key, issuer and its key.
226

    
227
        :param subject: subject to be added to the created certificate
228
        :param subject_key: string containing the private key to be used when creating the certificate in PEM format
229
        :param issuer_key: string containing the private key of the issuer's certificate in PEM format
230
        :param issuer_pem: string containing the certificate of the issuer that will sign this CSR in PEM format
231
        :param subject_key_pass: string containing the passphrase of the private key used when creating the certificate
232
        in PEM format
233
        :param issuer_key_pass: string containing the passphrase of the private key of the issuer's certificate in PEM
234
        format
235
        :param extensions: extensions to be applied when creating the certificate
236
        :param days: number of days for which the certificate will be valid
237
        :param sn: serial number to be set, when "None" is set a random serial number is generated
238
        :return: string containing the generated certificate in PEM format
239
        """
240
        csr = self.__create_csr(subject, subject_key, key_pass=subject_key_pass)
241
        return self.__sign_csr(csr, issuer_pem, issuer_key, issuer_key_pass=issuer_key_pass, extensions=extensions,
242
                               days=days, sn=sn)
243

    
244
    @staticmethod
245
    def verify_cert(certificate):
246
        """
247
        Verifies whether the given certificate is not expired.
248

    
249
        :param certificate: certificate to be verified in PEM format
250
        :return: Returns `true` if the certificate is not expired, `false` when expired.
251
        """
252
        # call openssl to check whether the certificate is valid to this date
253
        args = [OPENSSL_EXECUTABLE, "x509", "-checkend", "0", "-noout", "-text", "-in", "-"]
254

    
255
        # create a new process
256
        proc = subprocess.Popen(args, stdin=subprocess.PIPE,
257
                                stdout=subprocess.PIPE,
258
                                stderr=subprocess.PIPE)
259

    
260
        out, err = proc.communicate(bytes(certificate, encoding="utf-8"))
261

    
262
        # zero return code means that the certificate is valid
263
        if proc.returncode == 0:
264
            return True
265
        elif proc.returncode == 1 and "Certificate will expire" in out.decode():
266
            # 1 return code means that the certificate is invalid but such message has to be present in the proc output
267
            return False
268
        else:
269
            # the process failed because of some other reason (incorrect cert format)
270
            raise CryptographyException(OPENSSL_EXECUTABLE, args, err.decode())
271

    
272
    def extract_public_key_from_private_key(self, private_key_pem: str, passphrase=None) -> str:
273
        """
274
        Extracts a public key from the given private key passed in PEM format
275
        :param private_key_pem: PEM data representing the private key from which a public key should be extracted
276
        :param passphrase: passphrase to be provided when the supplied private key is encrypted
277
        :return: a string containing the extracted public key in PEM format
278
        """
279
        args = ["rsa", "-in", "-", "-pubout"]
280
        if passphrase is not None:
281
            args.extend(["-passin", f"pass:{passphrase}"])
282
        return self.__run_for_output(args, proc_input=bytes(private_key_pem, encoding="utf-8")).decode()
283

    
284
    def extract_public_key_from_certificate(self, cert_pem: str) -> str:
285
        """
286
        Extracts a public key from the given certificate passed in PEM format
287
        :param cert_pem: PEM data representing a certificate from which a public key should be extracted
288
        :return: a string containing the extracted public key in PEM format
289
        """
290
        # extracting public key from a certificate does not seem to require a passphrase even when
291
        # signed using an encrypted PK
292
        args = ["x509", "-in", "-", "-noout", "-pubkey"]
293
        return self.__run_for_output(args, proc_input=bytes(cert_pem, encoding="utf-8")).decode()
294

    
295
    def parse_cert_pem(self, cert_pem):
296
        """
297
        Parses the given certificate in PEM format and returns the subject of the certificate and it's NOT_BEFORE
298
        and NOT_AFTER field
299
        :param cert_pem: a certificated in a PEM format to be parsed
300
        :return: a tuple containing a subject, NOT_BEFORE and NOT_AFTER dates
301
        """
302
        # run openssl x509 to view certificate content
303
        args = ["x509", "-noout", "-subject", "-startdate", "-enddate", "-in", "-"]
304

    
305
        cert_info_raw = self.__run_for_output(args, proc_input=bytes(cert_pem, encoding="utf-8")).decode()
306

    
307
        # split lines
308
        results = re.split("\n", cert_info_raw)
309
        subj_line = results[0].strip()
310
        not_before_line = results[1].strip()
311
        not_after_line = results[2].strip()
312

    
313
        # attempt to extract subject via regex
314
        match = re.search(r"subject=(.*)", subj_line)
315
        if match is None:
316
            # TODO use logger
317
            print(f"Could not find subject to parse: {subj_line}")
318
            return None
319
        else:
320
            # find all attributes (key = value)
321
            found = re.findall(r"\s?([^c=\s]+)\s?=\s?([^,\n]+)", match.group(1))
322
            subj = Subject()
323
            for key, value in found:
324
                if key == "C":
325
                    subj.country = value.strip()
326
                elif key == "ST":
327
                    subj.state = value.strip()
328
                elif key == "L":
329
                    subj.locality = value.strip()
330
                elif key == "O":
331
                    subj.organization = value.strip()
332
                elif key == "OU":
333
                    subj.organization_unit = value.strip()
334
                elif key == "CN":
335
                    subj.common_name = value.strip()
336
                elif key == "emailAddress":
337
                    subj.email_address = value.strip()
338

    
339
        # extract notBefore and notAfter date fields
340
        not_before = re.search(r"notBefore=(.*)", not_before_line)
341
        not_after = re.search(r"notAfter=(.*)", not_after_line)
342

    
343
        # if date fields are found parse them into date objects
344
        if not_before is not None:
345
            not_before = time.strptime(not_before.group(1).strip(), NOT_AFTER_BEFORE_DATE_FORMAT)
346
        if not_after is not None:
347
            not_after = time.strptime(not_after.group(1).strip(), NOT_AFTER_BEFORE_DATE_FORMAT)
348

    
349
        # TODO wrapper class?
350
        # return it as a tuple
351
        return subj, not_before, not_after
352

    
353
    def get_openssl_version(self) -> str:
354
        """
355
        Get version of the OpenSSL installed on the system
356
        :return: version of the OpenSSL as returned from the process
357
        """
358
        return self.__run_for_output(["version"]).decode("utf-8")
359

    
360

    
361
class CryptographyException(Exception):
362

    
363
    def __init__(self, executable, args, message):
364
        self.executable = executable
365
        self.args = args
366
        self.message = message
367

    
368
    def __str__(self):
369
        return f"""
370
        EXECUTABLE: {self.executable}
371
        ARGS: {self.args}
372
        MESSAGE: {self.message}
373
        """
(3-3/4)