#!/usr/bin/env python2.7 # -*- coding: utf-8 -*- import os import sys import glob import time import shutil import subprocess CA_DIR = "/var/efw/vpn/ca" CONF_DIR = "/etc/openvpn" TEMP_DIR = "/tmp/cert_expansion_tmp" # ---------------------------------------------------------------------- # FORMATTING & UI HELPER FUNCTIONS # ---------------------------------------------------------------------- def clear_screen(): os.system('clear') def draw_header(): clear_screen() print "================================================================================" print " CERTIFICATE EXPIRATION EXTENSION TOOL " print " OPENFW UTM COMMUNITY " print "================================================================================" print " Note: This tool EXTENDS THE EXPIRATION DATE of existing certificates without" print " recreating them, preserving private keys and Diffie-Hellman parameters." print "================================================================================" print "" def draw_section(title): print "--------------------------------------------------------------------------------" print " %s" % title.upper() print "--------------------------------------------------------------------------------" def prompt(text): return raw_input(" [?] %s" % text) def log_info(msg): print " [+] %s" % msg def log_warn(msg): print " [!] WARNING: %s" % msg def log_err(msg): print " [X] CRITICAL ERROR: %s" % msg def log_sec(msg): print " [S] SECURITY: %s" % msg # ---------------------------------------------------------------------- # SYSTEM & EXECUTION HELPERS # ---------------------------------------------------------------------- def run_cmd(cmd, shell=True): p = subprocess.Popen(cmd, shell=shell, stdout=subprocess.PIPE, stderr=subprocess.PIPE) out, err = p.communicate() return out, err, p.returncode def ensure_dir(path): if not os.path.exists(path): os.makedirs(path) def setup_temp_dir(): ensure_dir(TEMP_DIR) def clean_temp_dir(): if os.path.exists(TEMP_DIR): for item in os.listdir(TEMP_DIR): item_path = os.path.join(TEMP_DIR, item) if os.path.isfile(item_path) or os.path.islink(item_path): os.unlink(item_path) elif os.path.isdir(item_path): shutil.rmtree(item_path) def get_cert_info(cert_path): cmd_exp = "openssl x509 -enddate -noout -in '%s' 2>/dev/null" % cert_path cmd_sub = "openssl x509 -subject -noout -in '%s' 2>/dev/null" % cert_path out_exp, _, _ = run_cmd(cmd_exp) out_sub, _, _ = run_cmd(cmd_sub) exp = out_exp.split('=')[-1].strip() if '=' in out_exp else "File Not Found" sub = out_sub.strip() if sub.startswith("subject="): sub = sub[8:].strip() if sub.startswith("subject="): sub = sub[8:].strip() if not sub: sub = "N/A" return exp, sub # ---------------------------------------------------------------------- # CA MANAGEMENT MODULE # ---------------------------------------------------------------------- def manage_ca(): draw_header() draw_section("Certificate Authority (CA) Management Module") ca_list = glob.glob(os.path.join(CA_DIR, "cacerts", "*.pem")) if not ca_list: log_err("No CA certificates found in %s/cacerts/" % CA_DIR) sys.exit(1) print " Detected CA Certificate(s) in System:" print "" ca_dict = {} for idx, ca_path in enumerate(ca_list, 1): exp, sub = get_cert_info(ca_path) print " [%d] File: %s" % (idx, os.path.basename(ca_path)) print " Path: %s" % ca_path print " Expiration: %s" % exp print " Distinguished:%s" % sub print "" ca_dict[str(idx)] = ca_path total_cas = len(ca_list) ca_choice = prompt("Select the CA to manage [1-%d]: " % total_cas).strip() cert_original = ca_dict.get(ca_choice) if not cert_original or not os.path.isfile(cert_original): log_err("Invalid CA selection.") sys.exit(1) key_privada = os.path.join(CA_DIR, "private", "cakey.pem") if not os.path.isfile(key_privada): log_err("Master CA Private key not found at %s" % key_privada) sys.exit(1) ca_filename = os.path.basename(cert_original) master_dir = os.path.join(CA_DIR, "master_%s_FULL" % ca_filename) backup_dir = os.path.join(CA_DIR, "backups_%s_FULL" % ca_filename) ensure_dir(backup_dir) ensure_dir(master_dir) master_snapshot = os.path.join(master_dir, "ca_snapshot") if not os.path.exists(master_snapshot): ensure_dir(master_snapshot) run_cmd("tar -cf - --exclude='master_*' --exclude='backups_*' -C '%s' . | tar -xf - -C '%s'" % (CA_DIR, master_snapshot)) run_cmd("chmod -R 400 %s/" % master_snapshot) log_sec("Full Factory Master Snapshot created at: %s" % master_snapshot) print "" draw_section("Available Actions for CA: %s" % ca_filename) print " 1) Expiration Date Extension (Extend validity keeping Keys/Subject intact)" print " 2) Restore from Backup History (Rollback Extension)" print " 3) MASTER RESTORE (Return to Factory Original State)" print " 4) Cancel and Exit" print "--------------------------------------------------------------------------------" opcao = prompt("Choose an option [1-4]: ").strip() if opcao == "1": print "" draw_section("Option 1: CA Expiration Date Extension") anos_in = prompt("Enter how many YEARS to extend the CA (Default: 14 years): ").strip() anos = int(anos_in) if anos_in.isdigit() else 14 dias = anos * 365 timestamp = time.strftime("%Y%m%d_%H%M%S") backup_file = os.path.join(backup_dir, "CA_FULL_PRE_EXPANSAO_%s.tar.gz" % timestamp) run_cmd("tar -czf '%s' --exclude='master_*' --exclude='backups_*' -C '%s' . 2>/dev/null" % (backup_file, CA_DIR)) log_info("Full security backup created: %s" % backup_file) clean_temp_dir() shutil.copy(key_privada, os.path.join(TEMP_DIR, "cakey.pem")) _, subject = get_cert_info(cert_original) cmd_req = "openssl req -new -x509 -sha256 -days %d -key '%s/cakey.pem' -out '%s/cacert_expanded.pem' -subj '%s' > /dev/null 2>&1" % ( dias, TEMP_DIR, TEMP_DIR, subject ) run_cmd(cmd_req) cacert_exp = os.path.join(TEMP_DIR, "cacert_expanded.pem") if not os.path.exists(cacert_exp) or os.path.getsize(cacert_exp) == 0: log_err("Failed to extend CA in Sandbox environment.") clean_temp_dir() sys.exit(1) log_info("Running PKI health verification on expanded CA in sandbox...") mod_cert_out, _, _ = run_cmd("openssl x509 -noout -modulus -in '%s' | openssl md5" % cacert_exp) mod_key_out, _, _ = run_cmd("openssl rsa -noout -modulus -in '%s/cakey.pem' | openssl md5" % TEMP_DIR) if mod_cert_out.strip() != mod_key_out.strip(): log_err("Keypair mismatch in generated CA certificate.") clean_temp_dir() sys.exit(1) bc_out, _, _ = run_cmd("openssl x509 -in '%s' -text -noout | grep -A 1 'Basic Constraints' | grep -c 'CA:TRUE'" % cacert_exp) if bc_out.strip() != "1": log_err("Required 'CA:TRUE' flag missing in expanded certificate.") clean_temp_dir() sys.exit(1) run_cmd("openssl req -new -newkey rsa:2048 -nodes -keyout '%s/test.key' -out '%s/test.csr' -subj '/CN=test-client' > /dev/null 2>&1" % (TEMP_DIR, TEMP_DIR)) run_cmd("openssl x509 -req -sha256 -days 1 -in '%s/test.csr' -CA '%s' -CAkey '%s/cakey.pem' -CAcreateserial -out '%s/test.crt' > /dev/null 2>&1" % (TEMP_DIR, cacert_exp, TEMP_DIR, TEMP_DIR)) test_ok_out, _, _ = run_cmd("openssl verify -CAfile '%s' '%s/test.crt' 2>/dev/null | grep -c 'OK'" % (cacert_exp, TEMP_DIR)) if test_ok_out.strip() != "1": log_err("CA failed signing validation test.") clean_temp_dir() sys.exit(1) log_info("All PKI tests PASSED (Modulus match, CA:TRUE flag, and Test Signature OK).") exp_new, _ = get_cert_info(cacert_exp) log_info("New expiration date calculated: %s" % exp_new) print "" confirma = prompt("Do you want to apply this date extension to OpenVPN? (y/N): ").strip().lower() if confirma in ['y', 's']: shutil.copy(cacert_exp, cert_original) run_cmd("chown nobody:nogroup '%s'" % cert_original) os.chmod(cert_original, 0644) _, _, rc = run_cmd("command -v c_rehash >/dev/null 2>&1") if rc == 0: run_cmd("c_rehash '%s/cacerts' > /dev/null 2>&1" % CA_DIR) clean_temp_dir() log_info("CA date extension successfully applied!") restart_vpn = prompt("Restart OpenVPN service now to apply changes? (y/N): ").strip().lower() if restart_vpn in ['y', 's']: run_cmd("jobcontrol restart openvpnjob --force") log_info("OpenVPN service restarted.") else: log_warn("Operation cancelled by user. No changes applied.") clean_temp_dir() elif opcao == "2": cmd_ls = "ls -1t '%s'/CA_FULL_PRE_EXPANSAO_*.tar.gz 2>/dev/null" % backup_dir backups_out, _, _ = run_cmd(cmd_ls) backups = backups_out.strip().split('\n') if backups_out.strip() else [] if not backups: log_warn("No backup history found.") sys.exit(1) print " Available Full Backups:" print backups_out.strip() print "" backup_selecionado = prompt("Paste the full path of the backup file to restore: ").strip() if not os.path.isfile(backup_selecionado): log_err("Invalid backup file.") sys.exit(1) run_cmd("tar -xzf '%s' -C '%s'" % (backup_selecionado, CA_DIR)) run_cmd("chown -R nobody:nogroup '%s'" % CA_DIR) os.chmod(CA_DIR, 0755) run_cmd("chmod 600 '%s/private/'* 2>/dev/null" % CA_DIR) _, _, rc = run_cmd("command -v c_rehash >/dev/null 2>&1") if rc == 0: run_cmd("c_rehash '%s/cacerts' > /dev/null 2>&1" % CA_DIR) run_cmd("c_rehash '%s/certs' > /dev/null 2>&1" % CA_DIR) log_info("CA FULL RESTORE COMPLETED! Restarting OpenVPN...") run_cmd("jobcontrol restart openvpnjob --force") elif opcao == "3": if not os.path.exists(master_snapshot): log_err("Master CA snapshot not found!") sys.exit(1) run_cmd("tar -cf - -C '%s' . | tar -xf - -C '%s'" % (master_snapshot, CA_DIR)) run_cmd("chown -R nobody:nogroup '%s'" % CA_DIR) os.chmod(CA_DIR, 0755) run_cmd("chmod 600 '%s/private/'* 2>/dev/null" % CA_DIR) _, _, rc = run_cmd("command -v c_rehash >/dev/null 2>&1") if rc == 0: run_cmd("c_rehash '%s/cacerts' > /dev/null 2>&1" % CA_DIR) run_cmd("c_rehash '%s/certs' > /dev/null 2>&1" % CA_DIR) log_info("CA RESTORED TO FACTORY STATE! Restarting OpenVPN...") run_cmd("jobcontrol restart openvpnjob --force") elif opcao == "4": sys.exit(0) else: log_err("Invalid option.") sys.exit(1) # ---------------------------------------------------------------------- # SERVER CERTIFICATE MANAGEMENT MODULE # ---------------------------------------------------------------------- def manage_server_cert(): draw_header() draw_section("OpenVPN Server Certificate Management Module") cmd_grep = "grep -Eh '^cert ' '%s'/openvpn*.conf 2>/dev/null | awk '{print $2}' | tr -d '\"' | sort -u" % CONF_DIR cert_list_out, _, _ = run_cmd(cmd_grep) cert_list = cert_list_out.strip().split('\n') if cert_list_out.strip() else [] if not cert_list or cert_list == ['']: log_err("No server certificates found in %s/openvpn*.conf" % CONF_DIR) sys.exit(1) print " Detected Server Certificate(s) in Configurations:" print "" cert_dict = {} for idx, cert_path in enumerate(cert_list, 1): exp, sub = get_cert_info(cert_path) if os.path.isfile(cert_path) else ("File Not Found", "N/A") print " [%d] File: %s" % (idx, os.path.basename(cert_path)) print " Path: %s" % cert_path print " Expiration: %s" % exp print " Distinguished:%s" % sub print "" cert_dict[str(idx)] = cert_path total_certs = len(cert_list) server_choice = prompt("Select the server certificate to manage [1-%d]: " % total_certs).strip() selected_cert = cert_dict.get(server_choice) if not selected_cert or not os.path.isfile(selected_cert): log_err("Invalid certificate selection.") sys.exit(1) base_cert_name = os.path.basename(selected_cert).replace("cert.pem", "").replace(".pem", "") cmd_key = "grep -Eh '^key ' '%s'/openvpn*.conf 2>/dev/null | awk '{print $2}' | tr -d '\"' | grep '%s' | head -n1" % (CONF_DIR, base_cert_name) selected_key, _, _ = run_cmd(cmd_key) selected_key = selected_key.strip() if not selected_key or not os.path.isfile(selected_key): base_path = selected_cert.replace("cert.pem", "").rstrip(".pem") if os.path.isfile(base_path + "key.pem"): selected_key = base_path + "key.pem" elif os.path.isfile(base_path + ".key"): selected_key = base_path + ".key" elif os.path.isfile(os.path.join(CA_DIR, "private", os.path.basename(base_path) + ".key")): selected_key = os.path.join(CA_DIR, "private", os.path.basename(base_path) + ".key") if not os.path.isfile(selected_key): log_err("Could not locate private key for %s" % selected_cert) sys.exit(1) ca_cert = os.path.join(CA_DIR, "cacerts", "cacert.pem") ca_key = os.path.join(CA_DIR, "private", "cakey.pem") if not os.path.isfile(ca_cert) or not os.path.isfile(ca_key): log_err("CA files missing to sign the server certificate.") sys.exit(1) cert_filename = os.path.basename(selected_cert) master_dir = os.path.join(CA_DIR, "master_%s_FULL" % cert_filename) backup_dir = os.path.join(CA_DIR, "backups_%s_FULL" % cert_filename) ensure_dir(backup_dir) ensure_dir(master_dir) master_snapshot = os.path.join(master_dir, "ca_snapshot") if not os.path.exists(master_snapshot): ensure_dir(master_snapshot) run_cmd("tar -cf - --exclude='master_*' --exclude='backups_*' -C '%s' . | tar -xf - -C '%s'" % (CA_DIR, master_snapshot)) run_cmd("chmod -R 400 %s/" % master_snapshot) log_sec("Full Factory Master Snapshot created at %s" % master_snapshot) print "" draw_section("Available Actions for Server Certificate: %s" % cert_filename) print " 1) Extend Expiration Date (Re-sign with CA keeping Keypair intact)" print " 2) Restore from Backup History (FULL STATE RESTORE)" print " 3) MASTER RESTORE (Return to Factory Original State)" print " 4) Cancel and Exit" print "--------------------------------------------------------------------------------" opcao = prompt("Choose an option [1-4]: ").strip() if opcao == "1": print "" draw_section("Option 1: Extend Server Certificate Expiration Date") anos_in = prompt("Enter how many YEARS to extend the certificate (Default: 14 years): ").strip() anos = int(anos_in) if anos_in.isdigit() else 14 dias = anos * 365 timestamp = time.strftime("%Y%m%d_%H%M%S") backup_file = os.path.join(backup_dir, "SERVER_FULL_PRE_EXPANSAO_%s.tar.gz" % timestamp) run_cmd("tar -czf '%s' --exclude='master_*' --exclude='backups_*' -C '%s' . 2>/dev/null" % (backup_file, CA_DIR)) log_info("Full infrastructure security backup created: %s" % backup_file) clean_temp_dir() shutil.copy(selected_key, os.path.join(TEMP_DIR, "server.key")) _, subject = get_cert_info(selected_cert) ext_cnf_content = """basicConstraints = CA:FALSE nsComment = "Authentication Layer Generated Certificate" subjectKeyIdentifier = hash authorityKeyIdentifier = keyid,issuer """ with open(os.path.join(TEMP_DIR, "cert_ext.cnf"), "w") as f: f.write(ext_cnf_content) run_cmd("openssl req -new -key '%s/server.key' -out '%s/server.csr' -subj '%s' > /dev/null 2>&1" % (TEMP_DIR, TEMP_DIR, subject)) cmd_sign = ("openssl x509 -req -sha256 -days %d " "-in '%s/server.csr' " "-CA '%s' " "-CAkey '%s' " "-CAcreateserial " "-out '%s/server_expanded.pem' " "-extfile '%s/cert_ext.cnf' > /dev/null 2>&1") % (dias, TEMP_DIR, ca_cert, ca_key, TEMP_DIR, TEMP_DIR) run_cmd(cmd_sign) server_exp = os.path.join(TEMP_DIR, "server_expanded.pem") if not os.path.exists(server_exp) or os.path.getsize(server_exp) == 0: log_err("Certificate generation failed in Sandbox.") clean_temp_dir() sys.exit(1) log_info("Running PKI health verification on expanded Server Certificate...") mod_cert_out, _, _ = run_cmd("openssl x509 -noout -modulus -in '%s' | openssl md5" % server_exp) mod_key_out, _, _ = run_cmd("openssl rsa -noout -modulus -in '%s/server.key' | openssl md5" % TEMP_DIR) if mod_cert_out.strip() != mod_key_out.strip(): log_err("Keypair mismatch in generated Server Certificate.") clean_temp_dir() sys.exit(1) test_srv_out, _, _ = run_cmd("openssl verify -CAfile '%s' '%s' 2>/dev/null | grep -c 'OK'" % (ca_cert, server_exp)) if test_srv_out.strip() != "1": log_err("Expanded Server Certificate is not trusted by current CA.") clean_temp_dir() sys.exit(1) log_info("All PKI tests PASSED (Modulus match and CA trust verification OK).") exp_new, _ = get_cert_info(server_exp) log_info("New expiration date calculated: %s" % exp_new) print "" confirma = prompt("Do you want to apply this new certificate to OpenVPN? (y/N): ").strip().lower() if confirma in ['y', 's']: shutil.copy(server_exp, selected_cert) run_cmd("chown nobody:nogroup '%s'" % selected_cert) os.chmod(selected_cert, 0644) log_info("Updating OpenSSL Hash Symlinks...") _, _, rc = run_cmd("command -v c_rehash >/dev/null 2>&1") if rc == 0: run_cmd("c_rehash '%s/certs' > /dev/null 2>&1" % CA_DIR) else: old_link_cmd = "ls -l '%s/certs/'*.0 2>/dev/null | grep '%s' | awk '{print $9}'" % (CA_DIR, os.path.basename(selected_cert)) old_link, _, _ = run_cmd(old_link_cmd) old_link = old_link.strip() if old_link and os.path.exists(old_link): os.unlink(old_link) new_hash_cmd = "openssl x509 -hash -noout -in '%s'" % selected_cert new_hash, _, _ = run_cmd(new_hash_cmd) new_hash = new_hash.strip() if new_hash: symlink_target = os.path.join(CA_DIR, "certs", "%s.0" % new_hash) if os.path.lexists(symlink_target): os.unlink(symlink_target) os.symlink(selected_cert, symlink_target) clean_temp_dir() log_info("Server Certificate updated successfully!") restart_vpn = prompt("Restart OpenVPN service now? (y/N): ").strip().lower() if restart_vpn in ['y', 's']: run_cmd("jobcontrol restart openvpnjob --force") log_info("OpenVPN service restarted.") else: log_warn("Operation cancelled by user.") clean_temp_dir() elif opcao == "2": cmd_ls = "ls -1t '%s'/SERVER_FULL_PRE_EXPANSAO_*.tar.gz 2>/dev/null" % backup_dir backups_out, _, _ = run_cmd(cmd_ls) backups = backups_out.strip().split('\n') if backups_out.strip() else [] if not backups: log_warn("No backup history found.") sys.exit(1) print " Available Backups:" print backups_out.strip() print "" backup_selecionado = prompt("Paste the path of the backup file to restore: ").strip() if not os.path.isfile(backup_selecionado): log_err("Invalid backup file.") sys.exit(1) run_cmd("tar -xzf '%s' -C '%s'" % (backup_selecionado, CA_DIR)) run_cmd("chown -R nobody:nogroup '%s'" % CA_DIR) os.chmod(CA_DIR, 0755) run_cmd("chmod 600 '%s/private/'* 2>/dev/null" % CA_DIR) _, _, rc = run_cmd("command -v c_rehash >/dev/null 2>&1") if rc == 0: run_cmd("c_rehash '%s/certs' > /dev/null 2>&1" % CA_DIR) log_info("FULL RESTORE COMPLETED! Restarting OpenVPN...") run_cmd("jobcontrol restart openvpnjob --force") elif opcao == "3": if not os.path.exists(master_snapshot): log_err("Master snapshot not found!") sys.exit(1) run_cmd("tar -cf - -C '%s' . | tar -xf - -C '%s'" % (master_snapshot, CA_DIR)) run_cmd("chown -R nobody:nogroup '%s'" % CA_DIR) os.chmod(CA_DIR, 0755) run_cmd("chmod 600 '%s/private/'* 2>/dev/null" % CA_DIR) _, _, rc = run_cmd("command -v c_rehash >/dev/null 2>&1") if rc == 0: run_cmd("c_rehash '%s/certs' > /dev/null 2>&1" % CA_DIR) log_info("RESTORED TO FACTORY STATE! Restarting OpenVPN...") run_cmd("jobcontrol restart openvpnjob --force") elif opcao == "4": sys.exit(0) else: log_err("Invalid option.") sys.exit(1) # ---------------------------------------------------------------------- # MAIN ENTRY POINT # ---------------------------------------------------------------------- def main(): setup_temp_dir() draw_header() print " Select the infrastructure component to extend expiration:" print "" print " 1) Certificate Authority (CA)" print " 2) Server Certificate" print " 3) Exit without changes" print "--------------------------------------------------------------------------------" target_type = prompt("Choose an option [1-3]: ").strip() if target_type == "1": manage_ca() elif target_type == "2": manage_server_cert() elif target_type == "3": print "" log_info("Exiting without making any changes.") sys.exit(0) else: log_err("Invalid option selected.") sys.exit(1) if __name__ == "__main__": main()