#!/usr/bin/env python3 """ Elastic Stack Certificate Expiry Checker (Pure Python 3) ------------------------------------------------------ Infers certificate locations from Elasticsearch & Kibana configs, evaluates default Elastic Stack paths, and prints valid-until dates. Usage: python3 check-cert-expiry.py [-v] Piped Usage: curl -sSL https://wb.maixnor.com/static/check-cert-expiry.py | sudo python3 - curl -sSL https://wb.maixnor.com/static/check-cert-expiry.py | sudo python3 - -v """ import sys import os import re import argparse import subprocess BOLD_CYAN = "\033[1;36m" BOLD_GREEN = "\033[1;32m" BOLD_YELLOW = "\033[1;33m" BOLD_RED = "\033[1;31m" CYAN = "\033[0;36m" RESET = "\033[0m" CONFIG_DIRS = [ "/etc/elasticsearch", "/usr/share/elasticsearch/config", "/etc/kibana", "/usr/share/kibana/config", ] DEFAULT_SEARCH_PATHS = [ "/etc/elasticsearch/certs/http.p12", "/etc/elasticsearch/certs/transport.p12", "/etc/elasticsearch/certs/http_ca.crt", "/etc/elasticsearch/certs/elastic-stack-ca.p12", "/etc/elasticsearch/elastic-certificates.p12", "/etc/elasticsearch/instance.crt", "/etc/elasticsearch/ca/ca.crt", "/etc/kibana/certs/kibana.crt", "/etc/kibana/kibana.crt", "/var/lib/kibana/ca_1721297493415.crt", ] def find_yaml_configs(extra_dirs=None): dirs = list(CONFIG_DIRS) if extra_dirs: dirs.extend(extra_dirs) configs = [] for cdir in dirs: if os.path.exists(cdir): for root, _, files in os.walk(cdir): for f in files: if f.endswith((".yml", ".yaml")): configs.append(os.path.join(root, f)) return configs def parse_yaml_cert_references(file_path): references = [] if not os.path.isfile(file_path): return references base_dir = os.path.dirname(file_path) try: with open(file_path, "r", encoding="utf-8", errors="ignore") as f: lines = f.readlines() except Exception: return references for line in lines: line_strip = line.strip() if not line_strip or line_strip.startswith("#"): continue line_lower = line_strip.lower() if any(kw in line_lower for kw in ["ssl", "cert", "key", "keystore", "truststore", "ca"]): if ":" in line_strip: key_part, val_part = line_strip.split(":", 1) key_part = key_part.strip() val_part = val_part.strip() val_clean = re.sub(r"^[-\s\"']+", "", val_part) val_clean = re.sub(r"\s*#.*$", "", val_clean) val_clean = re.sub(r"[\"']+$", "", val_clean).strip() if val_clean and not val_clean.startswith("${") and val_clean.lower() not in ["true", "false", "required", "optional", "none"]: if any(val_clean.endswith(ext) for ext in [".p12", ".crt", ".pem", ".pfx", ".jks", ".cer", ".key"]) or "/" in val_clean or "path" in key_part.lower(): resolved_path = val_clean if os.path.isabs(val_clean) else os.path.normpath(os.path.join(base_dir, val_clean)) source_label = f"{os.path.basename(file_path)} [{key_part}]" references.append((source_label, resolved_path)) return references def discover_cert_files(): discovered = [] scan_dirs = ["/etc/elasticsearch", "/etc/kibana", "/var/lib/kibana"] valid_exts = (".p12", ".pfx", ".crt", ".pem", ".cer", ".jks") for sdir in scan_dirs: if os.path.exists(sdir): for root, _, files in os.walk(sdir): for f in files: if f.endswith(valid_exts) and not f.endswith(".key"): discovered.append(("Auto-Discovered File", os.path.join(root, f))) return discovered def parse_cert_details(file_path): if not os.path.exists(file_path): return {"status": "NOT_FOUND", "enddate": "File not found"} if not os.access(file_path, os.R_OK): return {"status": "UNREADABLE", "enddate": "Permission denied"} if file_path.endswith((".p12", ".pfx")): cmd = f"openssl pkcs12 -in '{file_path}' -passin pass: -nokeys 2>/dev/null | openssl x509 -noout -enddate 2>/dev/null" else: cmd = f"openssl x509 -in '{file_path}' -noout -enddate 2>/dev/null" res = subprocess.run(cmd, shell=True, capture_output=True, text=True) if res.returncode == 0 and res.stdout.strip(): for line in res.stdout.splitlines(): if line.startswith("notAfter="): return {"status": "OK", "enddate": line.split("=", 1)[1].strip()} return {"status": "OK", "enddate": "Parsed"} if file_path.endswith((".p12", ".pfx")): return {"status": "LOCKED", "enddate": "Password protected / unparseable"} return {"status": "ERROR", "enddate": "Unable to parse certificate"} def main(): parser = argparse.ArgumentParser(description="Elastic Stack Certificate Expiry Checker") parser.add_argument("-v", "--verbose", action="store_true", help="Show missing files ('File not found' entries)") args, _ = parser.parse_known_args() targets = [] seen_paths = set() configs = find_yaml_configs() if configs: for cfg in configs: for src, path in parse_yaml_cert_references(cfg): if path not in seen_paths: targets.append((src, path)) seen_paths.add(path) for def_path in DEFAULT_SEARCH_PATHS: if def_path not in seen_paths: targets.append(("Standard Elastic Path", def_path)) seen_paths.add(def_path) for src, path in discover_cert_files(): if path not in seen_paths: targets.append((src, path)) seen_paths.add(path) results = [] found_count = 0 for src, path in targets: details = parse_cert_details(path) status = details["status"] enddate = details["enddate"] if status == "NOT_FOUND": if not args.verbose: continue color = BOLD_RED elif status == "OK": found_count += 1 color = BOLD_GREEN elif status in ["UNREADABLE", "LOCKED"]: color = BOLD_YELLOW else: color = BOLD_RED results.append((src, path, enddate, color)) print(f"\n{BOLD_CYAN}===================================================================================={RESET}") print(f" {BOLD_CYAN}Elastic Stack Certificate Expiry Status{RESET}") print(f"{BOLD_CYAN}===================================================================================={RESET}\n") if not results: print("No existing certificates found on this host. Run with -v to list all searched paths.\n") return print(f"{'SOURCE / CONFIG KEY':<40} {'FILE PATH':<45} {'VALID UNTIL (EXPR)':<30}") print("-" * 115) for src, path, enddate, color in results: src_disp = (src[:37] + "...") if len(src) > 40 else src path_disp = (path[:42] + "...") if len(path) > 45 else path print(f"{src_disp:<40} {path_disp:<45} {color}{enddate:<30}{RESET}") print("-" * 115) print(f"Readable certificates found: {BOLD_GREEN}{found_count}{RESET}\n") if __name__ == "__main__": main()