Revize 6c098d6e
Přidáno uživatelem Stanislav Král před asi 4 roky(ů)
proj/services/cryptography.py | ||
---|---|---|
12 | 12 |
class CryptographyService: |
13 | 13 |
|
14 | 14 |
@staticmethod |
15 |
def _run_for_output(args=None, stdin=None, executable=OPENSSL_EXECUTABLE): |
|
15 |
def subject_to_param_format(subject): |
|
16 |
subj_dict = {} |
|
17 |
if subject.common_name is not None: |
|
18 |
subj_dict["CN"] = subject.common_name |
|
19 |
if subject.country is not None: |
|
20 |
subj_dict["C"] = subject.country |
|
21 |
if subject.locality is not None: |
|
22 |
subj_dict["L"] = subject.locality |
|
23 |
if subject.state is not None: |
|
24 |
subj_dict["ST"] = subject.state |
|
25 |
if subject.organization is not None: |
|
26 |
subj_dict["O"] = subject.organization |
|
27 |
if subject.organization_unit is not None: |
|
28 |
subj_dict["OU"] = subject.organization_unit |
|
29 |
if subject.email_address is not None: |
|
30 |
subj_dict["emailAddress"] = subject.email_address |
|
31 |
|
|
32 |
# merge the subject into a "subj" parameter format |
|
33 |
return "".join([f"/{key}={value}" for key, value in subj_dict.items()]) |
|
34 |
|
|
35 |
@staticmethod |
|
36 |
def _run_for_output(args=None, proc_input=None, executable=OPENSSL_EXECUTABLE): |
|
16 | 37 |
""" |
17 | 38 |
Launches a new process in which the given executable is run. STDIN and process arguments can be set. |
18 | 39 |
If the process ends with a non-zero then <CryptographyException> is raised. |
19 | 40 |
:param args: Arguments to be passed to the program. |
20 |
:param stdin: String input to be passed to the stdin of the created process.
|
|
41 |
:param proc_input: String input to be passed to the stdin of the created process.
|
|
21 | 42 |
:param executable: Executable to be run (defaults to openssl) |
22 | 43 |
:return: If the process ends with a zero return code then the STDOUT of the process is returned as a byte array. |
23 | 44 |
""" |
... | ... | |
28 | 49 |
args.insert(0, executable) |
29 | 50 |
|
30 | 51 |
# create a new process |
31 |
proc = subprocess.Popen(args, stdin=subprocess.PIPE if stdin is not None else None, stdout=subprocess.PIPE,
|
|
52 |
proc = subprocess.Popen(args, stdin=subprocess.PIPE if proc_input is not None else None, stdout=subprocess.PIPE,
|
|
32 | 53 |
stderr=subprocess.PIPE) |
33 | 54 |
|
34 |
out, err = proc.communicate(stdin)
|
|
55 |
out, err = proc.communicate(proc_input)
|
|
35 | 56 |
|
36 | 57 |
if proc.returncode != 0: |
37 | 58 |
# if the process did not result in zero result code, then raise an exception |
... | ... | |
73 | 94 |
assert key is not None |
74 | 95 |
assert subject is not None |
75 | 96 |
|
76 |
subj_dict = {} |
|
77 |
if subject.common_name is not None: |
|
78 |
subj_dict["CN"] = subject.common_name |
|
79 |
if subject.country is not None: |
|
80 |
subj_dict["C"] = subject.country |
|
81 |
if subject.locality is not None: |
|
82 |
subj_dict["L"] = subject.locality |
|
83 |
if subject.state is not None: |
|
84 |
subj_dict["ST"] = subject.state |
|
85 |
if subject.organization is not None: |
|
86 |
subj_dict["O"] = subject.organization |
|
87 |
if subject.organization_unit is not None: |
|
88 |
subj_dict["OU"] = subject.organization_unit |
|
89 |
if subject.email_address is not None: |
|
90 |
subj_dict["emailAddress"] = subject.email_address |
|
91 |
|
|
92 |
# merge the subject into a "subj" parameter |
|
93 |
subj = "".join([f"/{key}={value}" for key, value in subj_dict.items()]) |
|
97 |
subj = self.subject_to_param_format(subject) |
|
94 | 98 |
|
95 | 99 |
with TemporaryFile("openssl.conf", config) as conf_path: |
96 | 100 |
args = ["req", "-x509", "-new", "-subj", subj, |
... | ... | |
107 | 111 |
# if key_passphrase is not None: |
108 | 112 |
args.extend(["-passin", f"pass:{key_passphrase}"]) |
109 | 113 |
|
110 |
return self._run_for_output(args, stdin=bytes(key, encoding="utf-8")).decode() |
|
114 |
return self._run_for_output(args, proc_input=bytes(key, encoding="utf-8")).decode() |
|
115 |
|
|
116 |
def make_csr(self, subject, subject_key, subject_key_pass=None): |
|
117 |
""" |
|
118 |
Makes a CSR (Certificate Signing Request) |
|
119 |
|
|
120 |
:param subject: an instance of <Subject> representing the subject to be added to the CSR |
|
121 |
:param subject_key: the private key of the subject to be used to generate the CSR |
|
122 |
:param subject_key_pass: passphrase of the subject's private key |
|
123 |
:return: byte array containing the generated certificate signing request |
|
124 |
""" |
|
125 |
|
|
126 |
subj_param = self.subject_to_param_format(subject) |
|
127 |
|
|
128 |
args = ["req", "-new", "-subj", subj_param, "-key", "-"] |
|
129 |
|
|
130 |
if subject_key_pass is not None: |
|
131 |
args.extend(["-passin", f"pass:{subject_key_pass}"]) |
|
132 |
|
|
133 |
return self._run_for_output(args, proc_input=bytes(subject_key, encoding="utf-8")).decode() |
|
111 | 134 |
|
112 | 135 |
|
113 | 136 |
class CryptographyException(Exception): |
Také k dispozici: Unified diff
Re #8472 - Added make_csr method that makes a CSR
Added 2 unit tests testing the added method.
Moved CryptographyService tests into a separate files