|
0 |
- |
1 |
#!/usr/bin/python2.7 |
|
|
2 |
# coding: utf-8 |
3 |
# |
4 |
# Univention Management Console module: |
5 |
# System Diagnosis UMC module |
6 |
# |
7 |
# Copyright 2017 Univention GmbH |
8 |
# |
9 |
# http://www.univention.de/ |
10 |
# |
11 |
# All rights reserved. |
12 |
# |
13 |
# The source code of this program is made available |
14 |
# under the terms of the GNU Affero General Public License version 3 |
15 |
# (GNU AGPL V3) as published by the Free Software Foundation. |
16 |
# |
17 |
# Binary versions of this program provided by Univention to you as |
18 |
# well as other copyrighted, protected or trademarked materials like |
19 |
# Logos, graphics, fonts, specific documentations and configurations, |
20 |
# cryptographic keys etc. are subject to a license agreement between |
21 |
# you and Univention and not subject to the GNU AGPL V3. |
22 |
# |
23 |
# In the case you use this program under the terms of the GNU AGPL V3, |
24 |
# the program is provided in the hope that it will be useful, |
25 |
# but WITHOUT ANY WARRANTY; without even the implied warranty of |
26 |
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the |
27 |
# GNU Affero General Public License for more details. |
28 |
# |
29 |
# You should have received a copy of the GNU Affero General Public |
30 |
# License with the Debian GNU/Linux or Univention distribution in file |
31 |
# /usr/share/common-licenses/AGPL-3; if not, see |
32 |
# <http://www.gnu.org/licenses/>. |
33 |
|
34 |
import re |
35 |
import shutil |
36 |
import os.path |
37 |
import datetime |
38 |
import tempfile |
39 |
import subprocess |
40 |
import contextlib |
41 |
|
42 |
import requests |
43 |
import dateutil.tz |
44 |
from OpenSSL import crypto |
45 |
|
46 |
import univention.config_registry |
47 |
from univention.management.console.modules.diagnostic import Critical, Warning |
48 |
|
49 |
from univention.lib.i18n import Translation |
50 |
_ = Translation('univention-management-console-module-diagnostic').translate |
51 |
|
52 |
title = _('Check validity of SSL certificates') |
53 |
description = _('All SSL certificates valid.') |
54 |
links = [{ |
55 |
'name': 'sdb', |
56 |
'href': _('http://sdb.univention.de/1183'), |
57 |
'label': _('Univention Support Database - Renewing the TLS/SSL certificates') |
58 |
}] |
59 |
|
60 |
|
61 |
WARNING_PERIOD = datetime.timedelta(days=50) |
62 |
|
63 |
|
64 |
class CertificateWarning(Exception): |
65 |
def __init__(self, path): |
66 |
super(CertificateWarning, self).__init__(path) |
67 |
self.path = path |
68 |
|
69 |
|
70 |
class CertificateWillExpire(CertificateWarning): |
71 |
def __init__(self, path, remaining): |
72 |
super(CertificateWillExpire, self).__init__(path) |
73 |
self.remaining = remaining |
74 |
|
75 |
def __str__(self): |
76 |
msg = _('Certificate {path!r} will expire in {days} days.') |
77 |
days = int(self.remaining.total_seconds() / 60 / 60 / 24) |
78 |
return msg.format(path=self.path, days=days) |
79 |
|
80 |
|
81 |
class CertificateError(CertificateWarning): |
82 |
pass |
83 |
|
84 |
|
85 |
class CertificateNotYetValid(CertificateError): |
86 |
def __str__(self): |
87 |
msg = _('Found not yet valid certificate {path!r}.') |
88 |
return msg.format(path=self.path) |
89 |
|
90 |
|
91 |
class CertificateExpired(CertificateError): |
92 |
def __str__(self): |
93 |
msg = _('Found expired certificate {path!r}.') |
94 |
return msg.format(path=self.path) |
95 |
|
96 |
|
97 |
class CertificateInvalid(CertificateError): |
98 |
def __init__(self, path, message): |
99 |
super(CertificateInvalid, self).__init__(path) |
100 |
self.message = message |
101 |
|
102 |
def __str__(self): |
103 |
msg = _('Found invalid certificate {path!r}:\n{message}') |
104 |
return msg.format(path=self.path, message=self.message) |
105 |
|
106 |
|
107 |
class CertificateVerifier(object): |
108 |
def __init__(self, root_cert_path, crl_path): |
109 |
self.root_cert_path = root_cert_path |
110 |
self.crl_path = crl_path |
111 |
|
112 |
@staticmethod |
113 |
def parse_generalized_time(generalized_time): |
114 |
# ASN.1 GeneralizedTime |
115 |
# Local time only. ``YYYYMMDDHH[MM[SS[.fff]]]'' |
116 |
# Universal time (UTC time) only. ``YYYYMMDDHH[MM[SS[.fff]]]Z''. |
117 |
# Difference between local and UTC times. ``YYYYMMDDHH[MM[SS[.fff]]]+-HHMM''. |
118 |
|
119 |
sans_mircoseconds = re.sub('\.\d{3}', '', generalized_time) |
120 |
sans_difference = re.sub('[+-]\d{4}', '', sans_mircoseconds) |
121 |
date_format = { |
122 |
10: '%Y%m%d%H', 12: '%Y%m%d%H%M', 14: '%Y%m%d%H%M%S', |
123 |
11: '%Y%m%d%HZ', 13: '%Y%m%d%H%MZ', 15: '%Y%m%d%H%M%SZ', |
124 |
}.get(len(sans_difference)) |
125 |
|
126 |
if date_format is None: |
127 |
raise ValueError('Unparsable generalized_time {!r}'.format(generalized_time)) |
128 |
|
129 |
date = datetime.datetime.strptime(sans_mircoseconds, date_format) |
130 |
utc_difference = re.search('([+-])(\d{2})(\d{2})', sans_mircoseconds) |
131 |
|
132 |
if sans_mircoseconds.endswith('Z'): |
133 |
return date.replace(tzinfo=dateutil.tz.tzutc()) |
134 |
elif utc_difference: |
135 |
(op, hours_str, minutes_str) = utc_difference.groups() |
136 |
try: |
137 |
(hours, minutes) = (int(hours_str), int(minutes_str)) |
138 |
except ValueError: |
139 |
raise ValueError('Unparsable generalized_time {!r}'.format(generalized_time)) |
140 |
|
141 |
if op == '+': |
142 |
offset = datetime.timedelta(hours=hours, minutes=minutes) |
143 |
else: |
144 |
offset = datetime.timedelta(hours=-hours, minutes=-minutes) |
145 |
with_offset = date.replace(tzinfo=dateutil.tz.tzoffset('unknown', offset)) |
146 |
return with_offset.astimezone(dateutil.tz.tzutc()) |
147 |
as_local = date.replace(tzinfo=dateutil.tz.tzlocal()) |
148 |
return as_local.astimezone(dateutil.tz.tzutc()) |
149 |
|
150 |
def _verify_timestamps(self, cert_path): |
151 |
now = datetime.datetime.now(dateutil.tz.tzutc()) |
152 |
|
153 |
with open(cert_path) as fob: |
154 |
cert = crypto.load_certificate(crypto.FILETYPE_PEM, fob.read()) |
155 |
valid_from = self.parse_generalized_time(cert.get_notBefore()) |
156 |
|
157 |
if now < valid_from: |
158 |
yield CertificateNotYetValid(cert_path) |
159 |
|
160 |
valid_until = self.parse_generalized_time(cert.get_notAfter()) |
161 |
expires_in = valid_until - now |
162 |
|
163 |
if expires_in < datetime.timedelta(): |
164 |
yield CertificateExpired(cert_path) |
165 |
elif expires_in < WARNING_PERIOD: |
166 |
yield CertificateWillExpire(cert_path, expires_in) |
167 |
|
168 |
def _openssl_verify(self, path): |
169 |
# XXX It would be nice to do this in python. `python-openssl` has the |
170 |
# capability to check against CRL since version 16.1.0, but |
171 |
# unfortunately only version 0.14 is available in debian. |
172 |
cmd = ('openssl', 'verify', '-CAfile', self.root_cert_path, |
173 |
'-CRLfile', self.crl_path, '-crl_check', path) |
174 |
verify = subprocess.Popen(cmd, stdout=subprocess.PIPE) |
175 |
(stdout, stderr) = verify.communicate() |
176 |
if verify.poll() != 0: |
177 |
yield CertificateInvalid(path, stdout) |
178 |
|
179 |
def verify_root(self): |
180 |
for error in self.verify(self.root_cert_path): |
181 |
yield error |
182 |
|
183 |
def verify(self, cert_path): |
184 |
for error in self._verify_timestamps(cert_path): |
185 |
yield error |
186 |
for error in self._openssl_verify(cert_path): |
187 |
yield error |
188 |
|
189 |
|
190 |
def certificates(configRegistry): |
191 |
fqdn = '{}.{}'.format(configRegistry.get('hostname'), configRegistry.get('domainname')) |
192 |
default_certificate = '/etc/univention/ssl/{}/cert.pem'.format(fqdn) |
193 |
yield configRegistry.get('apache2/ssl/certificate', default_certificate) |
194 |
|
195 |
saml_certificate = configRegistry.get('saml/idp/certificate/certificate') |
196 |
if saml_certificate: |
197 |
yield saml_certificate |
198 |
|
199 |
postfix_certificate = configRegistry.get('mail/postfix/ssl/certificate') |
200 |
if postfix_certificate: |
201 |
yield postfix_certificate |
202 |
|
203 |
if os.path.exists('/etc/univention/ssl/ucsCA/index.txt'): |
204 |
with open('/etc/univention/ssl/ucsCA/index.txt') as fob: |
205 |
for line in fob.readlines(): |
206 |
try: |
207 |
(status, _expiry, _revoked, serial, _path, _subject) = line.split('\t', 6) |
208 |
except ValueError: |
209 |
pass |
210 |
else: |
211 |
if status.strip() == 'V': |
212 |
yield '/etc/univention/ssl/ucsCA/certs/{}.pem'.format(serial) |
213 |
|
214 |
|
215 |
@contextlib.contextmanager |
216 |
def download_tempfile(url): |
217 |
with tempfile.NamedTemporaryFile() as fob: |
218 |
response = requests.get(url, stream=True) |
219 |
shutil.copyfileobj(response.raw, fob) |
220 |
fob.flush() |
221 |
yield fob.name |
222 |
|
223 |
|
224 |
@contextlib.contextmanager |
225 |
def convert_crl_to_pem(path): |
226 |
with tempfile.NamedTemporaryFile() as fob: |
227 |
convert = ('openssl', 'crl', '-inform', 'DER', '-in', path, '-outform', |
228 |
'PEM', '-out', fob.name) |
229 |
subprocess.check_call(convert) |
230 |
yield fob.name |
231 |
|
232 |
|
233 |
def verify_local(all_certificates): |
234 |
with convert_crl_to_pem('/etc/univention/ssl/ucsCA/crl/ucsCA.crl') as crl: |
235 |
verifier = CertificateVerifier('/etc/univention/ssl/ucsCA/CAcert.pem', crl) |
236 |
for error in verifier.verify_root(): |
237 |
yield error |
238 |
for cert in all_certificates: |
239 |
for error in verifier.verify(cert): |
240 |
yield error |
241 |
|
242 |
|
243 |
def verify_from_master(master, all_certificates): |
244 |
root_ca_uri = 'http://{}/ucs-root-ca.crt'.format(master) |
245 |
crl_uri = 'http://{}/ucsCA.crl'.format(master) |
246 |
with download_tempfile(root_ca_uri) as root_ca, download_tempfile(crl_uri) as crl: |
247 |
with convert_crl_to_pem(crl) as crl_pem: |
248 |
verifier = CertificateVerifier(root_ca, crl_pem) |
249 |
for error in verifier.verify_root(): |
250 |
yield error |
251 |
for cert in all_certificates: |
252 |
for error in verifier.verify(cert): |
253 |
yield error |
254 |
|
255 |
|
256 |
def run(): |
257 |
configRegistry = univention.config_registry.ConfigRegistry() |
258 |
configRegistry.load() |
259 |
|
260 |
all_certificates = certificates(configRegistry) |
261 |
is_local_check = configRegistry.get('server/role') in \ |
262 |
('domaincontroller_master', 'domaincontroller_backup') |
263 |
|
264 |
if is_local_check: |
265 |
cert_verify = list(verify_local(all_certificates)) |
266 |
else: |
267 |
cert_verify = list(verify_from_master(configRegistry.get('ldap/master'), |
268 |
all_certificates)) |
269 |
|
270 |
error_descriptions = [str(error) for error in cert_verify if |
271 |
isinstance(error, CertificateWarning)] |
272 |
|
273 |
if error_descriptions: |
274 |
error_descriptions.append(_('Please see {sdb} on how to renew certificates.')) |
275 |
if any(isinstance(error, CertificateError) for error in cert_verify): |
276 |
raise Critical(description='\n'.join(error_descriptions)) |
277 |
raise Warning(description='\n'.join(error_descriptions)) |
278 |
|
279 |
|
280 |
if __name__ == '__main__': |
281 |
from univention.management.console.modules.diagnostic import main |
282 |
main() |
1 |
certificate_check.py (po) |
283 |
certificate_check.py (po) |
2 |
-- |
|
|
3 |
.../umc/python/diagnostic/de.po | 45 +++++++++++++++++++++- |
284 |
.../umc/python/diagnostic/de.po | 45 +++++++++++++++++++++- |
4 |
1 file changed, 43 insertions(+), 2 deletions(-) |
285 |
1 file changed, 43 insertions(+), 2 deletions(-) |