]>
git.siccegge.de Git - tools.git/blob - tls-check
3 from __future__
import print_function
4 from optparse
import OptionParser
5 from ssl
import SSLContext
, PROTOCOL_TLSv1_2
, CERT_REQUIRED
, cert_time_to_seconds
, SSLError
, CertificateError
6 from socket
import socket
, AF_INET6
, create_connection
7 from datetime
import datetime
, timedelta
8 from smtplib
import SMTP
14 def __init__(self
, cafile
, warn
, crit
):
19 def check(self
, proto
, host
, port
, name
):
20 context
= SSLContext(PROTOCOL_TLSv1_2
)
21 context
.verify_mode
= CERT_REQUIRED
22 context
.load_verify_locations(self
.cafile
)
23 if hasattr(self
, 'remote_check_%s' % proto
):
24 getattr(self
, 'remote_check_%s' % proto
)(context
, host
, port
, name
)
26 def remote_check_xmpp(self
, context
, host
, port
, name
):
27 xmpp_open
= ("<stream:stream xmlns='jabber:client' xmlns:stream='"
28 "http://etherx.jabber.org/streams' xmlns:tls='http://www.ietf.org/rfc/"
29 "rfc2595.txt' to='{0}' xml:lang='en' version='1.0'>" )
30 xmpp_starttls
= "<starttls xmlns='urn:ietf:params:xml:ns:xmpp-tls'/>"
32 connection
= create_connection((host
, port
))
33 connection
.sendall(xmpp_open
.format(name
).encode('utf-8'))
34 response
= connection
.recv(4096).decode('utf-8')
36 if not '</stream:features>' in response
:
39 connection
.sendall(xmpp_starttls
.encode('utf-8'))
42 connection
= context
.wrap_socket(connection
, server_hostname
=name
)
43 connection
.do_handshake()
45 cert
= connection
.getpeercert()
46 return self
.check_cert(cert
, host
, port
, name
)
48 def remote_check_smtp(self
, context
, host
, port
, name
):
49 smtp
= SMTP(host
, port
)
51 smtp
.starttls(context
=context
)
53 print("CRIT (invalid certificate) %s:%d" % (host
, port
))
56 cert
= smtp
.sock
.getpeercert()
57 return self
.check_cert(cert
, host
, port
, name
)
59 def remote_check_ssl(self
, context
, host
, port
, name
):
60 connection
= context
.wrap_socket(socket(AF_INET6
),
63 connection
.connect((host
, port
))
65 print("CRIT (invalid certificate) %s:%d" % (host
, port
))
68 cert
= connection
.getpeercert()
69 return self
.check_cert(cert
, host
, port
, name
)
71 def check_cert(self
, data
, host
, port
, name
):
72 expiretimestamp
= cert_time_to_seconds(data
['notAfter'])
73 delta
= datetime
.utcfromtimestamp(expiretimestamp
) - datetime
.utcnow()
74 deltastr
= str(delta
).split(",")
77 print("CRIT (expires in %8s,%16s) %s:%d" % (deltastr
[0], deltastr
[1], name
, port
))
79 elif delta
< self
.warn
:
80 print("WARN (expires in %8s,%16s) %s:%d" % (deltastr
[0], deltastr
[1], name
, port
))
85 parser
= OptionParser()
86 parser
.add_option("--config", action
="store", type="string", dest
="config",
87 help="configuration file to use")
88 parser
.add_option("-n", "--name",
89 action
="append", type="string", dest
="names",
90 help="hostname:port to check for expired certificates")
91 parser
.add_option("-w", "--warning-days",
92 action
="store", type=int, dest
="warn",
93 help="minimum remaining validity in days before a warning is issued")
94 parser
.add_option("-c", "--critical-days",
95 action
="store", type=int, dest
="crit",
96 help="minimum remaining validity in days before a warning is issued")
97 parser
.add_option("-v", action
="store_true", dest
="verbose", default
=False)
98 parser
.add_option("-q", action
="store_false", dest
="verbose")
99 parser
.add_option("--ca", action
="store", type="string", dest
="ca",
100 help="ca certificate bundle")
103 opts
, _args
= parser
.parse_args()
106 configuration
= yaml
.load(open(opts
.config
))
108 configuration
= dict()
111 configuration
['names'] = opts
.names
113 configuration
['warn_days'] = opts
.warn
115 configuration
['crit_days'] = opts
.crit
117 configuration
['cacertificates'] = opts
.ca
119 configuration
['verbose'] = opts
.verbose
121 if 'verbose' in configuration
:
122 VERBOSE
= configuration
['verbose']
124 if not 'names' in configuration
:
125 parser
.error("needs at least one host")
127 verifier
= Verifier(configuration
['cacertificates'] if 'cacertificates' in configuration
else '/etc/ssl/certs/ca-certificates.crt',
128 timedelta(configuration
['warn_days'] if 'warn_days' in configuration
else 15),
129 timedelta(configuration
['crit_days'] if 'crit_days' in configuration
else 5))
132 hosts
= [ (i
[0], i
[1], int(i
[2]), i
[3] if len(i
) == 4 else i
[1]) for i
in [ j
.split(':', 3) for j
in configuration
['names'] ] ]
133 except (ValueError, IndexError):
134 parser
.error("names need to be in PROTO:DNSNAME:PORT format")
136 for proto
, host
, port
, name
in hosts
:
137 verifier
.check(proto
, host
, port
, name
)
139 if __name__
== "__main__":