From f6a7a621f9b0567e570ca3b18c66a56f399db3b2 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Mateusz=20Gruszczy=C5=84ski?= Date: Mon, 27 Oct 2025 07:21:43 +0100 Subject: [PATCH] logging --- certpusher.py | 133 ++++++++++++++++++++++++++++++++++----------- config.ini.example | 2 + 2 files changed, 104 insertions(+), 31 deletions(-) diff --git a/certpusher.py b/certpusher.py index 2667fa2..ea6a848 100644 --- a/certpusher.py +++ b/certpusher.py @@ -371,11 +371,71 @@ class MikroTikManager(SSHManager): class ProxmoxManager(SSHManager): """Specialized manager for Proxmox VE servers""" - def upload_certificate(self, cert_path: str, key_path: str) -> bool: + def check_certificate(self, source_cert: x509.Certificate, check_url: str) -> bool: + """ + Check if certificate on Proxmox needs update + Returns True if upload needed, False if current cert is OK + """ + try: + logger.info("Checking Proxmox certificate") + + # Method 1: Check via SSH - read cert file directly + success, stdout, stderr = self.execute_command( + 'openssl x509 -in /etc/pve/local/pveproxy-ssl.pem -noout -serial -dates', + ignore_error=True + ) + + if success and stdout: + logger.debug(f"Proxmox certificate info:\n{stdout}") + + # Parse serial number + serial_match = re.search(r'serial=([A-F0-9]+)', stdout) + # Parse expiry date + notAfter_match = re.search(r'notAfter=(.+)', stdout) + + if serial_match and notAfter_match: + proxmox_serial = serial_match.group(1) + source_serial = format(source_cert.serial_number, 'X') + + logger.debug(f"Source serial: {source_serial}") + logger.debug(f"Proxmox serial: {proxmox_serial}") + + if source_serial == proxmox_serial: + logger.info("✓ Proxmox certificate is current. Skipping upload.") + return False + else: + logger.info("Proxmox certificate differs. Upload needed.") + return True + + # Method 2: Fallback - try URL check + if check_url: + cert_manager = CertificateManager() + remote_cert = cert_manager.get_cert_from_url(check_url) + + if remote_cert: + if cert_manager.compare_certificates(source_cert, remote_cert): + logger.info("✓ Certificate verified via URL. Skipping upload.") + return False + + # If we can't verify, proceed with upload + logger.warning("Could not verify certificate. Proceeding with upload.") + return True + + except Exception as e: + logger.warning(f"Error checking certificate: {e}. Proceeding with upload.") + return True + + def upload_certificate(self, cert_path: str, key_path: str, check_first: bool = True, + source_cert: x509.Certificate = None, check_url: str = None) -> bool: """Upload certificate to Proxmox VE""" try: logger.info(f"Proxmox certificate deployment") + # Check if upload is needed + if check_first and source_cert: + if not self.check_certificate(source_cert, check_url): + return True # Certificate is current, skip upload + logger.info("Uploading certificate") if not self.upload_file(cert_path, '/etc/pve/local/pveproxy-ssl.pem'): return False @@ -415,7 +475,6 @@ class ProxmoxManager(SSHManager): logger.error(f"Proxmox deployment failed: {e}") return False - class CertPusher: """Main application class""" @@ -505,37 +564,49 @@ class CertPusher: return False def process_proxmox(self, section: str, hostname: str, port: int, - username: str, ssh_key: str, source_cert_path: str) -> bool: - """Process Proxmox VE server specifically""" - try: - logger.info("Using Proxmox deployment method") - - source_key_path = self.get_key_path(section, source_cert_path) - - if not os.path.exists(source_key_path): - logger.error(f"Private key not found: {source_key_path}") - return False - - proxmox = ProxmoxManager(hostname, port, username, ssh_key) - - if not proxmox.connect(): - self.stats['failed'] += 1 - return False - - if not proxmox.upload_certificate(source_cert_path, source_key_path): - proxmox.disconnect() - self.stats['failed'] += 1 - return False - - proxmox.disconnect() - self.stats['uploaded'] += 1 - logger.info(f"✓ Proxmox processed successfully") - return True - - except Exception as e: - logger.error(f"Proxmox processing failed: {e}") + username: str, ssh_key: str, source_cert_path: str) -> bool: + """Process Proxmox VE server specifically""" + try: + logger.info("Using Proxmox deployment method") + + source_key_path = self.get_key_path(section, source_cert_path) + + if not os.path.exists(source_key_path): + logger.error(f"Private key not found: {source_key_path}") + return False + + # Load source certificate for comparison + source_cert = self.cert_manager.get_cert_from_file(source_cert_path) + + # Get check URL if available + check_url = self.config.get(section, 'check_url', fallback=None) + + # Check if we should verify before upload + check_first = self.config.getboolean(section, 'check_before_upload', fallback=True) + + proxmox = ProxmoxManager(hostname, port, username, ssh_key) + + if not proxmox.connect(): self.stats['failed'] += 1 return False + + # Upload with optional checking + if not proxmox.upload_certificate(source_cert_path, source_key_path, + check_first, source_cert, check_url): + proxmox.disconnect() + self.stats['failed'] += 1 + return False + + proxmox.disconnect() + self.stats['uploaded'] += 1 + logger.info(f"✓ Proxmox processed successfully") + return True + + except Exception as e: + logger.error(f"Proxmox processing failed: {e}") + self.stats['failed'] += 1 + return False + def process_host(self, section: str) -> bool: """Process certificate deployment for a single host""" diff --git a/config.ini.example b/config.ini.example index d351059..7b228ae 100644 --- a/config.ini.example +++ b/config.ini.example @@ -42,6 +42,7 @@ hostname = 10.87.2.150 port = 11922 username = root check_url = https://10.87.2.150:8006 +check_before_upload = true [proxmox2] type = proxmox @@ -49,6 +50,7 @@ hostname = 10.87.2.151 port = 11922 username = root check_url = https://10.87.2.151:8006 +check_before_upload = true # ═══════════════════════════════════════════════════════════ # HOME ASSISTANT INSTALLATIONS