diff --git a/.gitignore b/.gitignore index 6b0159c..76e574c 100644 --- a/.gitignore +++ b/.gitignore @@ -1,3 +1,5 @@ .idea *.iml *.pyc +*.csv +*.txt \ No newline at end of file diff --git a/README.md b/README.md index bed77b9..88daf01 100644 --- a/README.md +++ b/README.md @@ -15,7 +15,7 @@ To use the script simply run it from the command line, along with a list of the The script will list the status of each domain's certificate, displaying '`OK`' if the certificate was retrieved and is not expiring soon, '`WARN`' if the certificate's expiry date is getting close, or '`ERROR`' if the certificate has already expired, or if there is some other problem such as the host could not be found, or no certificate could be retrieved. -By default '`WARN`' will be displayed if there are less than 7 days until a certificate expires, but this interval can be changed by altering the value of the [WARN_IF_DAYS_LESS_THAN](https://github.com/codebox/https-certificate-expiry-checker/blob/main/check-certificates.py#L13) variable. +By default '`WARN`' will be displayed if there are less than 7 days until a certificate expires, but this interval can be changed by adding a `--warn-days` execution parameter If any of the domains are using a non-standard port for HTTPS then this should be specified using the usual notation of `host:port`, for example: @@ -31,5 +31,21 @@ The script returns an exit code indicating whether the checks passed or not, mak | Both of the previous conditions occurred | 3 | | No domain list was provided when running the script | 9 | -Certificate checks are performed in parallel, making the process of checking multiple domains much quicker. The number of concurrent checks that will be performed is determined by the value of the [WORKER_THREAD_COUNT](https://github.com/codebox/https-certificate-expiry-checker/blob/main/check-certificates.py#L11) variable. - \ No newline at end of file +Certificate checks are performed in parallel, making the process of checking multiple domains much quicker. The number of concurrent checks that will be performed is determined by the value of the [WORKER_THREAD_COUNT](https://github.com/codebox/https-certificate-expiry-checker/blob/main/check-certificates.py#L12) variable. + +## Optional Execution Parameters + +- `--domains-file` + Provide a file path containing one domain per line. This removes the need to pass endpoints on the command line. + +- `--csv` + Output results in CSV format, useful for parsing by other processes. No extra console messages will be printed. Always includes issuer info. + +- `--show-issuer` + Display the certificate's organization name (issuer) alongside its expiry info. + +- `--warn-days` + Override the default (7 days) warning threshold for expiring certificates. + +### Example: + > python check-certificates.py --domains-file domains.txt --show-issuer --csv --warn-days 10 \ No newline at end of file diff --git a/check-certificates.py b/check-certificates.py index de2e30c..43b90e5 100644 --- a/check-certificates.py +++ b/check-certificates.py @@ -5,6 +5,7 @@ import datetime import concurrent.futures import math +import argparse DEFAULT_HTTPS_PORT = 443 @@ -25,14 +26,20 @@ def make_host_port_pair(endpoint): def pluralise(singular, count): return '{} {}{}'.format(count, singular, '' if count == 1 else 's') -def get_certificate_expiry_date_time(context, host, port): +def get_certificate_expiry_and_issuer(context, host, port): with socket.create_connection((host, port), SOCKET_CONNECTION_TIMEOUT_SECONDS) as tcp_socket: with context.wrap_socket(tcp_socket, server_hostname=host) as ssl_socket: - # certificate_info is a dict with lots of information about the certificate - certificate_info = ssl_socket.getpeercert() - exp_date_text = certificate_info['notAfter'] - return datetime.datetime.fromtimestamp(ssl.cert_time_to_seconds(exp_date_text), datetime.timezone.utc) - + cert = ssl_socket.getpeercert() + exp_date_text = cert['notAfter'] + expiry = datetime.datetime.fromtimestamp(ssl.cert_time_to_seconds(exp_date_text), datetime.timezone.utc) + issuer_blocks = cert.get('issuer', []) + org_name = None + for block in issuer_blocks: + for k, v in block: + if k == 'organizationName': + org_name = v + break + return expiry, org_name def format_time_remaining(time_remaining): day_count = time_remaining.days @@ -56,13 +63,12 @@ def format_time_remaining(time_remaining): pluralise('min', minutes) ) -def get_exit_code(err_count, min_days): +def get_exit_code(err_count, min_days, warn_days): code = EXIT_SUCCESS - if err_count: code += EXIT_ERROR - if min_days < WARN_IF_DAYS_LESS_THAN: + if min_days < warn_days: code += EXIT_EXPIRING_SOON return code @@ -70,48 +76,80 @@ def get_exit_code(err_count, min_days): def format_host_port(host, port): return host + ('' if port == DEFAULT_HTTPS_PORT else ':{}'.format(port)) -def check_certificates(endpoints): +def check_certificates(endpoints, show_issuer, csv_output, warn_days): context = ssl.create_default_context() host_port_pairs = [make_host_port_pair(endpoint) for endpoint in endpoints] with concurrent.futures.ThreadPoolExecutor(max_workers=WORKER_THREAD_COUNT) as executor: futures = { - executor.submit(get_certificate_expiry_date_time, context, host, port): - (host, port) for host, port in host_port_pairs + executor.submit(get_certificate_expiry_and_issuer, context, host, port): (host, port) + for host, port in host_port_pairs } - endpoint_count = len(endpoints) err_count = 0 min_days = math.inf - max_host_port_len = max([len(format_host_port(host, port)) for host, port in host_port_pairs]) - print('Checking {}...'.format(pluralise('endpoint', endpoint_count))) + max_host_port_len = max(len(format_host_port(h, p)) for h, p in host_port_pairs) + + if not csv_output: + print('Checking {}...'.format(pluralise('endpoint', len(endpoints)))) + for future in concurrent.futures.as_completed(futures): host, port = futures[future] try: - expiry_time = future.result() + expiry, issuer = future.result() except Exception as ex: err_count += 1 - print('{} ERROR {}'.format(format_host_port(host, port).ljust(max_host_port_len), ex)) + if csv_output: + print(f"{host},{port},ERROR,,\"{ex}\",\"\"") + else: + print('{} ERROR {}'.format(format_host_port(host, port).ljust(max_host_port_len), ex)) else: - time_remaining = expiry_time - datetime.datetime.now(datetime.timezone.utc) + time_remaining = expiry - datetime.datetime.now(datetime.timezone.utc) time_remaining_txt = format_time_remaining(time_remaining) days_remaining = time_remaining.days min_days = min(min_days, days_remaining) - - print('{} {:<5} expires in {}'.format( - format_host_port(host, port).ljust(max_host_port_len), - 'WARN' if days_remaining < WARN_IF_DAYS_LESS_THAN else 'OK', - time_remaining_txt)) - - exit_code = get_exit_code(err_count, min_days) + status = 'WARN' if days_remaining < warn_days else 'OK' + issuer_text = issuer if issuer else 'not found' + if csv_output: + print(f"{host},{port},{status},{days_remaining},\"{time_remaining_txt}\",\"{issuer_text}\"") + else: + if show_issuer: + print('{} {:<5} expires in {} [Issuer: {}]'.format( + format_host_port(host, port).ljust(max_host_port_len), + status, + time_remaining_txt, + issuer_text + )) + else: + print('{} {:<5} expires in {}'.format( + format_host_port(host, port).ljust(max_host_port_len), + status, + time_remaining_txt + )) + + exit_code = get_exit_code(err_count, min_days, warn_days) sys.exit(exit_code) -if __name__ == '__main__': - endpoints = sys.argv[1:] - - if len(endpoints): - check_certificates(endpoints) +def main(): + parser = argparse.ArgumentParser() + parser.add_argument("--domains-file", help="File with one domain per line") + parser.add_argument("--show-issuer", action="store_true", help="Display certificate supplier info") + parser.add_argument("--csv", action="store_true", help="Output in CSV format") + parser.add_argument("--warn-days", type=int, default=WARN_IF_DAYS_LESS_THAN, + help="Override the default days threshold for warning") + parser.add_argument("endpoints", nargs="*", help="Endpoints (host:port or host)") + args = parser.parse_args() + + if args.domains_file: + with open(args.domains_file, "r") as f: + endpoints = [line.strip() for line in f if line.strip()] else: - print('Usage: {} '.format(sys.argv[0])) + endpoints = args.endpoints + + if not endpoints: sys.exit(EXIT_NO_HOST_LIST) + check_certificates(endpoints, args.show_issuer, args.csv, args.warn_days) + +if __name__ == "__main__": + main()