#!/usr/bin/env python3 """ ============================================================================== Elastic Stack Certificate Exchange Command Proposer & Wrapper (Python) ============================================================================== Parses configuration files for Elasticsearch and Kibana, extracts SSL certificate target paths, SANs, and service endpoints, and proposes the exact command to run the certificate renewal script (update-certs.py or update-certs.sh). ============================================================================== """ import os import sys import re import argparse import socket import subprocess import json # Terminal ANSI Color Codes CYAN = "\033[0;36m" BOLD_CYAN = "\033[1;36m" BLUE = "\033[0;34m" BOLD_BLUE = "\033[1;34m" GREEN = "\033[0;32m" BOLD_GREEN = "\033[1;32m" YELLOW = "\033[0;33m" BOLD_YELLOW = "\033[1;33m" RED = "\033[0;31m" BOLD_RED = "\033[1;31m" DIM = "\033[2m" BOLD = "\033[1m" RESET = "\033[0m" def parse_simple_yaml(file_path): """ Parses key-value pairs and nested paths from YAML file. Uses PyYAML if available, otherwise falls back to a regex line parser. """ if not os.path.exists(file_path): return {} try: import yaml with open(file_path, "r", encoding="utf-8") as f: data = yaml.safe_load(f) return data if isinstance(data, dict) else {} except Exception: pass config = {} try: with open(file_path, "r", encoding="utf-8") as f: for line in f: line = line.strip() if not line or line.startswith("#"): continue if ":" in line: parts = line.split(":", 1) k = parts[0].strip() v = parts[1].strip().strip("\"'") if v: config[k] = v except Exception: pass return config def get_nested_val(data, key_path, default=None): """ Retrieves nested dictionary values given a dot-separated key path or flat key. """ if not data: return default if key_path in data: return data[key_path] parts = key_path.split(".") curr = data for p in parts: if isinstance(curr, dict) and p in curr: curr = curr[p] else: return default return curr if curr is not None else default def discover_network_sans(): """ Discovers local DNS names and IP addresses to include in certificate Subject Alternative Names (SANs). """ dns_names = set(["localhost"]) ip_addresses = set(["127.0.0.1"]) try: hostname = socket.gethostname() if hostname: dns_names.add(hostname) fqdn = socket.getfqdn() if fqdn: dns_names.add(fqdn) except Exception: pass try: res = subprocess.run(["hostname", "-I"], capture_output=True, text=True, timeout=2) if res.returncode == 0: for ip in res.stdout.strip().split(): if ":" not in ip: ip_addresses.add(ip) except Exception: pass if len(ip_addresses) == 1: try: res = subprocess.run(["ip", "-4", "addr", "show"], capture_output=True, text=True, timeout=2) if res.returncode == 0: matches = re.findall(r"inet\s+([0-9]+\.[0-9]+\.[0-9]+\.[0-9]+)", res.stdout) for ip in matches: ip_addresses.add(ip) except Exception: pass for default_dns in ["cmsrv2024", "siem2024"]: dns_names.add(default_dns) for default_ip in ["10.1.10.3", "192.168.196.95"]: ip_addresses.add(default_ip) return sorted(list(dns_names)), sorted(list(ip_addresses)) def extract_es_settings(es_config_path): """ Extracts SSL target locations and network configuration from elasticsearch.yml. """ data = parse_simple_yaml(es_config_path) found = os.path.exists(es_config_path) http_p12 = get_nested_val(data, "xpack.security.http.ssl.keystore.path", "/etc/elasticsearch/certs/http.p12") transport_p12 = get_nested_val(data, "xpack.security.transport.ssl.keystore.path", "/etc/elasticsearch/certs/transport.p12") http_ca = get_nested_val(data, "xpack.security.http.ssl.certificate_authorities", "/etc/elasticsearch/certs/http_ca.crt") if isinstance(http_ca, list) and len(http_ca) > 0: http_ca = http_ca[0] net_host = get_nested_val(data, "network.host", "10.1.10.3") http_port = get_nested_val(data, "http.port", "9200") return { "config_exists": found, "config_path": es_config_path, "http_p12": http_p12, "transport_p12": transport_p12, "http_ca": http_ca, "network_host": net_host, "http_port": http_port, } def extract_kibana_settings(kibana_config_path): """ Extracts SSL targets and Fleet Server configurations from kibana.yml. """ data = parse_simple_yaml(kibana_config_path) found = os.path.exists(kibana_config_path) kibana_crt = get_nested_val(data, "server.ssl.certificate", "/etc/kibana/certs/kibana.crt") kibana_key = get_nested_val(data, "server.ssl.key", "/etc/kibana/certs/kibana.key") kibana_ca = get_nested_val(data, "elasticsearch.ssl.certificateAuthorities", "/var/lib/kibana/ca_1721297493415.crt") if isinstance(kibana_ca, list) and len(kibana_ca) > 0: kibana_ca = kibana_ca[0] es_hosts = get_nested_val(data, "elasticsearch.hosts", ["https://10.1.10.3:9200"]) if isinstance(es_hosts, str): es_hosts = [es_hosts] fleet_url = get_nested_val(data, "fleet.agent.fleet_server.hosts", ["https://10.1.10.3:8220"]) if isinstance(fleet_url, list) and len(fleet_url) > 0: fleet_url = fleet_url[0] fleet_policy = get_nested_val(data, "fleet.agent.policy", "fleet-server-policy") return { "config_exists": found, "config_path": kibana_config_path, "kibana_crt": kibana_crt, "kibana_key": kibana_key, "kibana_ca": kibana_ca, "es_url": es_hosts[0] if es_hosts else "https://10.1.10.3:9200", "fleet_url": fleet_url if isinstance(fleet_url, str) else "https://10.1.10.3:8220", "fleet_policy": fleet_policy, } def propose_command(es_info, kibana_info, dns_sans, ip_sans, update_script_path): """ Synthesizes environment variables and constructs the proposed execution command. """ env_vars = {} if es_info["http_p12"] != "/etc/elasticsearch/certs/http.p12": env_vars["ES_HTTP_P12_TARGET"] = es_info["http_p12"] if es_info["transport_p12"] != "/etc/elasticsearch/certs/transport.p12": env_vars["ES_TRANSPORT_P12_TARGET"] = es_info["transport_p12"] if es_info["http_ca"] != "/etc/elasticsearch/certs/http_ca.crt": env_vars["ES_HTTP_CA_TARGET"] = es_info["http_ca"] if kibana_info["kibana_crt"] != "/etc/kibana/certs/kibana.crt": env_vars["KIBANA_CRT_TARGET"] = kibana_info["kibana_crt"] if kibana_info["kibana_key"] != "/etc/kibana/certs/kibana.key": env_vars["KIBANA_KEY_TARGET"] = kibana_info["kibana_key"] if kibana_info["kibana_ca"] != "/var/lib/kibana/ca_1721297493415.crt": env_vars["KIBANA_CA_TARGET"] = kibana_info["kibana_ca"] env_vars["FLEET_SERVER_ES_URL"] = kibana_info["es_url"] env_vars["FLEET_SERVER_URL"] = kibana_info["fleet_url"] dns_str = " ".join(dns_sans) ip_str = " ".join(ip_sans) env_vars["SAN_DNS"] = f'"{dns_str}"' env_vars["SAN_IPS"] = f'"{ip_str}"' env_part = " ".join([f"{k}={v}" for k, v in env_vars.items()]) proposed_cmd = f"sudo {env_part} {update_script_path}" return proposed_cmd, env_vars def print_banner(): print(f"\n{BOLD_CYAN}===================================================={RESET}") print(f" {BOLD_BLUE}Elastic Stack Certificate Exchange Proposer (Python){RESET}") print(f"{BOLD_CYAN}===================================================={RESET}\n") def print_report(es_info, kibana_info, dns_sans, ip_sans, proposed_cmd, script_exists): print(f"{BOLD_CYAN}[1] Configuration Discovery Status:{RESET}") es_status = f"{BOLD_GREEN}Found{RESET} ({es_info['config_path']})" if es_info["config_exists"] else f"{YELLOW}Not found (using default paths){RESET}" kib_status = f"{BOLD_GREEN}Found{RESET} ({kibana_info['config_path']})" if kibana_info["config_exists"] else f"{YELLOW}Not found (using default paths){RESET}" print(f" {CYAN}↳{RESET} Elasticsearch Config: {es_status}") print(f" {CYAN}↳{RESET} Kibana Config: {kib_status}") print(f"\n{BOLD_CYAN}[2] Extracted Certificate Target Paths:{RESET}") print(f" {CYAN}↳{RESET} Elasticsearch HTTP P12: {BOLD}{es_info['http_p12']}{RESET}") print(f" {CYAN}↳{RESET} Elasticsearch Transport P12: {BOLD}{es_info['transport_p12']}{RESET}") print(f" {CYAN}↳{RESET} Elasticsearch HTTP CA: {BOLD}{es_info['http_ca']}{RESET}") print(f" {CYAN}↳{RESET} Kibana Certificate: {BOLD}{kibana_info['kibana_crt']}{RESET}") print(f" {CYAN}↳{RESET} Kibana Private Key: {BOLD}{kibana_info['kibana_key']}{RESET}") print(f" {CYAN}↳{RESET} Kibana CA Target: {BOLD}{kibana_info['kibana_ca']}{RESET}") print(f"\n{BOLD_CYAN}[3] Extracted Cluster & Fleet Services:{RESET}") print(f" {CYAN}↳{RESET} Fleet Elasticsearch Endpoint: {BOLD}{kibana_info['es_url']}{RESET}") print(f" {CYAN}↳{RESET} Fleet Server Endpoint: {BOLD}{kibana_info['fleet_url']}{RESET}") print(f" {CYAN}↳{RESET} Fleet Policy: {BOLD}{kibana_info['fleet_policy']}{RESET}") print(f"\n{BOLD_CYAN}[4] Discovered Subject Alternative Names (SANs):{RESET}") print(f" {CYAN}↳{RESET} DNS Names: {GREEN}{' '.join(dns_sans)}{RESET}") print(f" {CYAN}↳{RESET} IP Addrs: {GREEN}{' '.join(ip_sans)}{RESET}") print(f"\n{BOLD_CYAN}===================================================={RESET}") print(f" {BOLD_GREEN}PROPOSED CERTIFICATE EXCHANGE COMMAND:{RESET}") print(f"{BOLD_CYAN}===================================================={RESET}") print(f"\n {BOLD_YELLOW}{proposed_cmd}{RESET}\n") if not script_exists: print(f" {BOLD_RED}⚠ Warning: update script target path not found!{RESET}\n") def main(): parser = argparse.ArgumentParser(description="Wrapper to parse Elastic Stack configs and propose cert update command.") parser.add_argument("--es-config", default="/etc/elasticsearch/elasticsearch.yml", help="Path to elasticsearch.yml") parser.add_argument("--kibana-config", default="/etc/kibana/kibana.yml", help="Path to kibana.yml") parser.add_argument("--script-path", default="", help="Path to update-certs script") parser.add_argument("--execute", "-e", action="store_true", help="Execute the proposed command immediately with sudo") parser.add_argument("--json", action="store_true", help="Output raw configuration analysis in JSON format") args = parser.parse_args() script_dir = os.path.dirname(os.path.abspath(__file__)) if args.script_path: script_path = os.path.abspath(args.script_path) else: # Prefer update-certs.py if exists, fallback to update-certs.sh py_script = os.path.join(script_dir, "update-certs.py") sh_script = os.path.join(script_dir, "update-certs.sh") script_path = py_script if os.path.exists(py_script) else sh_script script_exists = os.path.exists(script_path) es_info = extract_es_settings(args.es_config) kibana_info = extract_kibana_settings(args.kibana_config) dns_sans, ip_sans = discover_network_sans() proposed_cmd, env_vars = propose_command(es_info, kibana_info, dns_sans, ip_sans, script_path) if args.json: out = { "elasticsearch": es_info, "kibana": kibana_info, "dns_sans": dns_sans, "ip_sans": ip_sans, "proposed_command": proposed_cmd, "env_vars": env_vars, "script_path": script_path, "script_exists": script_exists, } print(json.dumps(out, indent=2)) return print_banner() print_report(es_info, kibana_info, dns_sans, ip_sans, proposed_cmd, script_exists) if args.execute: if not script_exists: print(f"{BOLD_RED}Cannot execute: {script_path} does not exist.{RESET}") sys.exit(1) print(f"{BOLD_GREEN}Executing proposed command...{RESET}\n") env = os.environ.copy() for k, v in env_vars.items(): env[k] = v.strip('"') cmd_list = ["sudo", "-E", sys.executable, script_path] if script_path.endswith(".py") else ["sudo", "-E", script_path] subprocess.run(cmd_list, env=env) if __name__ == "__main__": main()