Additional execution parameters

.gitignore

@@ -1,3 +1,5 @@
 .idea
 *.iml
 *.pyc
+*.csv
+*.txt

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.
-
+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

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,62 +63,93 @@ 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
 
 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: {} <list of endpoints>'.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()