From afafb8aeef0a46800f306daade412a4ec63574f1 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Mateusz=20Gruszczy=C5=84ski?= Date: Mon, 27 Oct 2025 07:27:30 +0100 Subject: [PATCH] logging --- certpusher.py | 511 +++++++++++++++++++++----------------------------- 1 file changed, 214 insertions(+), 297 deletions(-) diff --git a/certpusher.py b/certpusher.py index 66006f9..fb750da 100644 --- a/certpusher.py +++ b/certpusher.py @@ -1,8 +1,7 @@ #!/usr/bin/env python3 """ CertPusher - Automated SSL Certificate Distribution Tool -Distributes SSL certificates to remote servers via SSH/SCP -Supports standard Linux servers, MikroTik RouterOS, and Proxmox VE +Version 1.1 - With unified certificate checking for all host types """ import configparser @@ -21,12 +20,12 @@ from scp import SCPClient import requests from cryptography import x509 from cryptography.hazmat.backends import default_backend +from cryptography.hazmat.primitives import hashes -# Create logs directory if it doesn't exist +# Create logs directory LOG_DIR = Path(__file__).parent / 'logs' LOG_DIR.mkdir(exist_ok=True) -# Logging configuration LOG_FORMAT = '%(asctime)s - %(name)s - %(levelname)s - %(message)s' LOG_FILE = LOG_DIR / f'certpusher_{datetime.now().strftime("%Y%m%d_%H%M%S")}.log' @@ -43,7 +42,6 @@ def setup_logging(debug: bool = False): ] ) - # Reduce paramiko logging noise logging.getLogger('paramiko').setLevel(logging.WARNING) logging.getLogger('paramiko.transport').setLevel(logging.WARNING) @@ -93,16 +91,13 @@ class CertificateManager: @staticmethod def compare_certificates(cert1: x509.Certificate, cert2: x509.Certificate) -> bool: - """Compare two certificates by serial number and fingerprint""" + """Compare two certificates by serial number""" try: - same_serial = cert1.serial_number == cert2.serial_number + serial1 = format(cert1.serial_number, 'X').upper() + serial2 = format(cert2.serial_number, 'X').upper() - from cryptography.hazmat.primitives import hashes - fingerprint1 = cert1.fingerprint(hashes.SHA256()) - fingerprint2 = cert2.fingerprint(hashes.SHA256()) - same_fingerprint = fingerprint1 == fingerprint2 - - return same_serial and same_fingerprint + logger.debug(f"Comparing serials: {serial1} vs {serial2}") + return serial1 == serial2 except Exception as e: logger.error(f"Failed to compare certificates: {e}") return False @@ -112,21 +107,14 @@ class CertificateManager: """Get human-readable certificate information""" try: subject = cert.subject.rfc4514_string() - issuer = cert.issuer.rfc4514_string() - valid_from = cert.not_valid_before_utc valid_to = cert.not_valid_after_utc - now = datetime.now(timezone.utc) days_left = (valid_to - now).days - return f""" -Certificate Info: - Subject: {subject} - Issuer: {issuer} - Valid From: {valid_from} - Valid To: {valid_to} - Days Until Expiry: {days_left} -""" + return f"""Certificate: {subject} + Serial: {format(cert.serial_number, 'X').upper()} + Expires: {valid_to} + Days left: {days_left}""" except Exception as e: return f"Unable to extract certificate info: {e}" @@ -182,13 +170,13 @@ class SSHManager: return True except Exception as e: - logger.error(f"SSH connection failed to {self.hostname}:{self.port}: {e}") + logger.error(f"SSH connection failed: {e}") return False def upload_file(self, local_path: str, remote_path: str) -> bool: """Upload file via SCP""" try: - logger.info(f"Uploading to {self.hostname}:{remote_path}") + logger.info(f"Uploading to {remote_path}") remote_dir = os.path.dirname(remote_path) if remote_dir: @@ -215,13 +203,10 @@ class SSHManager: stdout_text = stdout.read().decode('utf-8', errors='ignore') stderr_text = stderr.read().decode('utf-8', errors='ignore') - if exit_status == 0: - logger.debug(f"Command completed successfully") - else: - if not ignore_error: - logger.error(f"Command failed with exit code {exit_status}") - if stderr_text: - logger.error(f"Error: {stderr_text}") + if exit_status != 0 and not ignore_error: + logger.error(f"Command failed with exit code {exit_status}") + if stderr_text: + logger.debug(f"Error: {stderr_text}") return exit_status == 0, stdout_text, stderr_text @@ -229,11 +214,49 @@ class SSHManager: logger.error(f"Command execution failed: {e}") return False, "", str(e) + def check_remote_certificate(self, remote_cert_path: str, source_cert: x509.Certificate) -> bool: + """ + Check if remote certificate matches source certificate + Returns True if upload needed, False if certificates match + """ + try: + logger.info("Checking remote certificate via SSH") + + # Try to read and compare certificate via SSH + success, stdout, stderr = self.execute_command( + f'openssl x509 -in {remote_cert_path} -noout -serial 2>/dev/null || echo "NOTFOUND"', + ignore_error=True + ) + + if success and stdout and "NOTFOUND" not in stdout: + serial_match = re.search(r'serial=([A-F0-9]+)', stdout, re.IGNORECASE) + + if serial_match: + remote_serial = serial_match.group(1).upper() + source_serial = format(source_cert.serial_number, 'X').upper() + + logger.info(f"Source serial: {source_serial}") + logger.info(f"Remote serial: {remote_serial}") + + if source_serial == remote_serial: + logger.info("✓ Certificates match. Skipping upload.") + return False + else: + logger.info("✗ Certificates differ. Upload needed.") + return True + + logger.warning("Could not read remote certificate via SSH") + return True # Upload if we can't verify + + except Exception as e: + logger.warning(f"Error checking remote certificate: {e}") + return True + def disconnect(self): """Close SSH connection""" if self.ssh_client: self.ssh_client.close() - logger.debug(f"Disconnected from {self.hostname}") + logger.debug(f"Disconnected") class MikroTikManager(SSHManager): @@ -242,13 +265,9 @@ class MikroTikManager(SSHManager): def __init__(self, hostname: str, port: int, username: str, key_path: str): super().__init__(hostname, port, username, key_path) self.cert_name = "ssl-cert" - self.key_name = "ssl-key" def check_certificate_expiry(self, source_cert: x509.Certificate) -> bool: - """ - Check if certificate on MikroTik needs update - Returns True if upload needed, False if current cert is OK - """ + """Check if certificate on MikroTik needs update""" try: logger.info("Checking MikroTik certificate") @@ -258,13 +277,13 @@ class MikroTikManager(SSHManager): ) if not success or not stdout: - logger.info("No certificate found on MikroTik. Upload needed.") + logger.info("No certificate found. Upload needed.") return True invalid_after_match = re.search(r'invalid-after:\s+([a-zA-Z]{3}/\d{2}/\d{4}\s+\d{2}:\d{2}:\d{2})', stdout) if not invalid_after_match: - logger.warning("Could not parse certificate expiry. Proceeding with upload.") + logger.warning("Could not parse expiry. Proceeding with upload.") return True mikrotik_expiry_str = invalid_after_match.group(1) @@ -272,44 +291,44 @@ class MikroTikManager(SSHManager): try: mikrotik_expiry = datetime.strptime(mikrotik_expiry_str, '%b/%d/%Y %H:%M:%S') mikrotik_expiry = mikrotik_expiry.replace(tzinfo=timezone.utc) - except Exception as e: - logger.warning(f"Could not parse date: {e}. Proceeding with upload.") + except Exception: return True source_expiry = source_cert.not_valid_after_utc time_diff = abs((source_expiry - mikrotik_expiry).total_seconds()) if time_diff < 86400: - logger.info("✓ MikroTik certificate is current. Skipping upload.") + logger.info("✓ MikroTik certificate is current. Skipping.") return False else: logger.info(f"MikroTik certificate differs. Upload needed.") return True except Exception as e: - logger.warning(f"Error checking certificate: {e}. Proceeding with upload.") + logger.warning(f"Error checking: {e}. Proceeding with upload.") return True - def upload_certificate(self, cert_path: str, key_path: str = None, check_first: bool = True, source_cert: x509.Certificate = None) -> bool: - """Upload and import certificate to MikroTik RouterOS""" + def upload_certificate(self, cert_path: str, key_path: str, check_first: bool, source_cert: x509.Certificate) -> Tuple[bool, bool]: + """ + Upload certificate to MikroTik + Returns (success, was_uploaded) + """ try: - logger.info(f"MikroTik certificate deployment") - if check_first and source_cert: if not self.check_certificate_expiry(source_cert): - return True + return True, False # Success but skipped + + logger.info("Deploying MikroTik certificate") - logger.debug("Disabling www-ssl service") self.execute_command('/ip service disable www-ssl', ignore_error=True) - logger.debug("Removing old certificates") cleanup_commands = [ f'/certificate remove [find name~"{self.cert_name}"]', f'/file remove "{self.cert_name}.pem"', ] if key_path: - cleanup_commands.append(f'/file remove "{self.key_name}.pem"') + cleanup_commands.append(f'/file remove "ssl-key.pem"') for cmd in cleanup_commands: self.execute_command(cmd, ignore_error=True) @@ -321,20 +340,14 @@ class MikroTikManager(SSHManager): if key_path: logger.info("Uploading private key") with SCPClient(self.ssh_client.get_transport()) as scp: - scp.put(key_path, f'{self.key_name}.pem') + scp.put(key_path, 'ssl-key.pem') logger.info("Importing certificate") - import_cmd = f'/certificate import file-name={self.cert_name}.pem passphrase=""' - success, stdout, stderr = self.execute_command(import_cmd, timeout=30) - - if not success and "failure" in stderr.lower(): - logger.error(f"Certificate import failed: {stderr}") - return False + self.execute_command(f'/certificate import file-name={self.cert_name}.pem passphrase=""', timeout=30) import time time.sleep(2) - logger.info("Configuring www-ssl service") config_commands = [ f'/ip service set www-ssl certificate={self.cert_name}_0', '/ip service enable www-ssl', @@ -344,136 +357,106 @@ class MikroTikManager(SSHManager): self.execute_command(cmd, ignore_error=True) logger.info(f"✓ MikroTik deployment successful") - return True + return True, True # Success and uploaded except Exception as e: logger.error(f"MikroTik deployment failed: {e}") - return False - - def verify_certificate(self) -> bool: - """Verify certificate is properly installed""" - try: - success, stdout, stderr = self.execute_command( - '/certificate print detail where name~"ssl-cert"' - ) - - if success and stdout: - logger.debug(f"Certificate verified") - return True - - return False - - except Exception as e: - logger.error(f"Certificate verification failed: {e}") - return False + return False, False class ProxmoxManager(SSHManager): """Specialized manager for Proxmox VE servers""" 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 - """ + """Check if certificate on Proxmox needs update""" try: logger.info("Checking Proxmox certificate") - # Method 1: Check via SSH - read cert file directly + # Method 1: Check via SSH success, stdout, stderr = self.execute_command( - 'openssl x509 -in /etc/pve/local/pveproxy-ssl.pem -noout -serial -dates', + 'openssl x509 -in /etc/pve/local/pveproxy-ssl.pem -noout -serial -dates 2>/dev/null', ignore_error=True ) if success and stdout: - logger.debug(f"Proxmox certificate info:\n{stdout}") + serial_match = re.search(r'serial=([A-F0-9]+)', stdout, re.IGNORECASE) - # 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') + if serial_match: + proxmox_serial = serial_match.group(1).upper() + source_serial = format(source_cert.serial_number, 'X').upper() - logger.debug(f"Source serial: {source_serial}") - logger.debug(f"Proxmox serial: {proxmox_serial}") + logger.info(f"Source serial: {source_serial}") + logger.info(f"Proxmox serial: {proxmox_serial}") if source_serial == proxmox_serial: - logger.info("✓ Proxmox certificate is current. Skipping upload.") + logger.info("✓ Certificates match. Skipping upload.") return False else: - logger.info("Proxmox certificate differs. Upload needed.") + logger.info("✗ Certificates differ. Upload needed.") return True - # Method 2: Fallback - try URL check + # Method 2: Fallback to URL check if check_url: + logger.info("Trying URL-based check") 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 remote_cert and cert_manager.compare_certificates(source_cert, remote_cert): + logger.info("✓ Certificates match via URL. Skipping.") + return False - # If we can't verify, proceed with upload - logger.warning("Could not verify certificate. Proceeding with upload.") + logger.warning("Could not verify. Proceeding with upload.") return True except Exception as e: - logger.warning(f"Error checking certificate: {e}. Proceeding with upload.") + logger.warning(f"Error checking: {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""" + def upload_certificate(self, cert_path: str, key_path: str, check_first: bool, + source_cert: x509.Certificate, check_url: str) -> Tuple[bool, bool]: + """ + Upload certificate to Proxmox + Returns (success, was_uploaded) + """ 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 + return True, False # Success but skipped + + logger.info("Deploying Proxmox certificate") - logger.info("Uploading certificate") if not self.upload_file(cert_path, '/etc/pve/local/pveproxy-ssl.pem'): - return False + return False, False - logger.info("Uploading private key") if not self.upload_file(key_path, '/etc/pve/local/pveproxy-ssl.key'): - return False + return False, False - logger.debug("Setting permissions") commands = [ 'chmod 640 /etc/pve/local/pveproxy-ssl.key', 'chown root:www-data /etc/pve/local/pveproxy-ssl.key', ] for cmd in commands: - self.execute_command(cmd, ignore_error=False) + self.execute_command(cmd) logger.info("Restarting pveproxy") - success, stdout, stderr = self.execute_command('systemctl restart pveproxy', timeout=30) - - if not success: - logger.error(f"Failed to restart pveproxy") - return False + self.execute_command('systemctl restart pveproxy', timeout=30) import time time.sleep(3) - success, stdout, stderr = self.execute_command('systemctl is-active pveproxy') + success, stdout, _ = self.execute_command('systemctl is-active pveproxy') if success and 'active' in stdout: logger.info(f"✓ Proxmox deployment successful") - return True + return True, True else: - logger.error("pveproxy service is not active") - return False + logger.error("pveproxy not active") + return False, False except Exception as e: logger.error(f"Proxmox deployment failed: {e}") - return False + return False, False + class CertPusher: """Main application class""" @@ -482,61 +465,52 @@ class CertPusher: self.config_file = config_file self.config = configparser.ConfigParser() self.cert_manager = CertificateManager() - self.stats = { - 'total': 0, - 'uploaded': 0, - 'skipped': 0, - 'failed': 0 - } + self.stats = {'total': 0, 'uploaded': 0, 'skipped': 0, 'failed': 0} def load_config(self) -> bool: - """Load configuration from INI file""" + """Load configuration""" try: - logger.info(f"Loading configuration from {self.config_file}") + logger.info(f"Loading config: {self.config_file}") self.config.read(self.config_file) if 'global' not in self.config: - logger.error("Missing [global] section in config file") + logger.error("Missing [global] section") return False - required_global = ['source_cert_path', 'default_ssh_key'] - for key in required_global: + required = ['source_cert_path', 'default_ssh_key'] + for key in required: if not self.config.has_option('global', key): - logger.error(f"Missing required global option: {key}") + logger.error(f"Missing: {key}") return False - logger.info(f"✓ Configuration loaded") - logger.info(f"Found {len(self.config.sections()) - 1} host(s)") + logger.info(f"✓ Config loaded ({len(self.config.sections()) - 1} hosts)") return True except Exception as e: - logger.error(f"Failed to load configuration: {e}") + logger.error(f"Config load failed: {e}") return False def get_key_path(self, section: str, cert_path: str) -> str: - """Get private key path for certificate""" + """Get private key path""" if self.config.has_option(section, 'source_key_path'): return self.config.get(section, 'source_key_path') - if self.config.has_option('global', 'source_key_path'): return self.config.get('global', 'source_key_path') - - key_path = cert_path.replace('fullchain.pem', 'privkey.pem').replace('cert.pem', 'privkey.pem') - return key_path + return cert_path.replace('fullchain.pem', 'privkey.pem').replace('cert.pem', 'privkey.pem') - def process_mikrotik(self, section: str, hostname: str, port: int, - username: str, ssh_key: str, source_cert_path: str) -> bool: - """Process MikroTik device specifically""" + def process_mikrotik(self, section: str, hostname: str, port: int, username: str, ssh_key: str, source_cert_path: str) -> bool: + """Process MikroTik device""" try: - logger.info("Using MikroTik 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}") + logger.error(f"Key not found: {source_key_path}") return False source_cert = self.cert_manager.get_cert_from_file(source_cert_path) + if not source_cert: + return False + check_first = self.config.getboolean(section, 'check_before_upload', fallback=True) mikrotik = MikroTikManager(hostname, port, username, ssh_key) @@ -545,98 +519,73 @@ class CertPusher: self.stats['failed'] += 1 return False - result = mikrotik.upload_certificate(source_cert_path, source_key_path, check_first, source_cert) + success, was_uploaded = mikrotik.upload_certificate(source_cert_path, source_key_path, check_first, source_cert) + mikrotik.disconnect() - if not result: - mikrotik.disconnect() + if success: + if was_uploaded: + self.stats['uploaded'] += 1 + else: + self.stats['skipped'] += 1 + logger.info("✓ MikroTik processed") + return True + else: self.stats['failed'] += 1 return False - mikrotik.verify_certificate() - mikrotik.disconnect() - self.stats['uploaded'] += 1 - logger.info(f"✓ MikroTik processed successfully") - return True - except Exception as e: - logger.error(f"MikroTik processing failed: {e}") + logger.error(f"MikroTik failed: {e}") self.stats['failed'] += 1 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""" + def process_proxmox(self, section: str, hostname: str, port: int, username: str, ssh_key: str, source_cert_path: str) -> bool: + """Process Proxmox server""" try: - logger.info("Using Proxmox deployment method") - source_key_path = self.get_key_path(section, source_cert_path) - # Show which certificate we're using - logger.info(f"Source certificate: {source_cert_path}") - logger.info(f"Source key: {source_key_path}") - if not os.path.exists(source_key_path): - logger.error(f"Private key not found: {source_key_path}") + logger.error(f"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) - - if source_cert: - # Show source cert details - source_serial = format(source_cert.serial_number, 'X').upper() - logger.info(f"Source cert serial: {source_serial}") - logger.info(f"Source cert expires: {source_cert.not_valid_after_utc}") - else: - logger.error("Failed to load source certificate") + if not source_cert: return False - # Get check URL if available + logger.info(f"Using cert: {source_cert_path}") + logger.info(self.cert_manager.get_certificate_info(source_cert)) + 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) - logger.info(f"Check before upload: {check_first}") - if check_url: - logger.info(f"Check URL: {check_url}") - proxmox = ProxmoxManager(hostname, port, username, ssh_key) if not proxmox.connect(): self.stats['failed'] += 1 return False - # Upload with optional checking - result = proxmox.upload_certificate(source_cert_path, source_key_path, - check_first, source_cert, check_url) - + success, was_uploaded = proxmox.upload_certificate(source_cert_path, source_key_path, check_first, source_cert, check_url) proxmox.disconnect() - if result: - # Check if it was actually uploaded or skipped - # This is a bit tricky - we need to track this in upload_certificate - self.stats['uploaded'] += 1 - logger.info(f"✓ Proxmox processed successfully") + if success: + if was_uploaded: + self.stats['uploaded'] += 1 + else: + self.stats['skipped'] += 1 + logger.info("✓ Proxmox processed") return True else: self.stats['failed'] += 1 return False except Exception as e: - logger.error(f"Proxmox processing failed: {e}") - import traceback - logger.debug(traceback.format_exc()) + logger.error(f"Proxmox failed: {e}") self.stats['failed'] += 1 return False - def process_host(self, section: str) -> bool: - """Process certificate deployment for a single host""" + """Process standard host""" try: - logger.info(f"\n{'='*60}") - logger.info(f"Processing: {section}") - logger.info(f"{'='*60}") + logger.info(f"\n{'='*60}\nProcessing: {section}\n{'='*60}") self.stats['total'] += 1 @@ -645,49 +594,33 @@ class CertPusher: username = self.config.get(section, 'username', fallback='root') device_type = self.config.get(section, 'type', fallback='standard') - if self.config.has_option(section, 'ssh_key_path'): - ssh_key = self.config.get(section, 'ssh_key_path') - else: - ssh_key = self.config.get('global', 'default_ssh_key') - - if self.config.has_option(section, 'source_cert_path'): - source_cert_path = self.config.get(section, 'source_cert_path') - logger.info(f"Using host-specific certificate") - else: - source_cert_path = self.config.get('global', 'source_cert_path') + ssh_key = self.config.get(section, 'ssh_key_path', fallback=None) or self.config.get('global', 'default_ssh_key') + source_cert_path = self.config.get(section, 'source_cert_path', fallback=None) or self.config.get('global', 'source_cert_path') if not os.path.exists(source_cert_path): logger.error(f"Certificate not found: {source_cert_path}") self.stats['failed'] += 1 return False - logger.info(f"Host: {hostname}:{port}") - logger.info(f"Type: {device_type}") + logger.info(f"Host: {hostname}:{port} ({device_type})") logger.info(f"User: {username}") + # Route to specialized handlers if device_type.lower() == 'mikrotik': return self.process_mikrotik(section, hostname, port, username, ssh_key, source_cert_path) elif device_type.lower() == 'proxmox': return self.process_proxmox(section, hostname, port, username, ssh_key, source_cert_path) + # Standard host processing remote_cert_path = self.config.get(section, 'remote_cert_path') post_upload_command = self.config.get(section, 'post_upload_command', fallback='') check_url = self.config.get(section, 'check_url', fallback='') + check_first = self.config.getboolean(section, 'check_before_upload', fallback=True) - if check_url: - logger.info(f"Checking certificate at {check_url}") - source_cert = self.cert_manager.get_cert_from_file(source_cert_path) - remote_cert = self.cert_manager.get_cert_from_url(check_url) - - if source_cert and remote_cert: - if self.cert_manager.compare_certificates(source_cert, remote_cert): - logger.info(f"✓ Certificate is up to date. Skipping.") - self.stats['skipped'] += 1 - return True - else: - logger.info(f"Certificate is outdated. Uploading.") - else: - logger.warning(f"Could not compare certificates. Proceeding.") + source_cert = self.cert_manager.get_cert_from_file(source_cert_path) + if not source_cert: + self.stats['failed'] += 1 + return False ssh = SSHManager(hostname, port, username, ssh_key) @@ -695,65 +628,71 @@ class CertPusher: self.stats['failed'] += 1 return False + # Check if upload needed + upload_needed = True + + if check_first: + # Try SSH check first + if not ssh.check_remote_certificate(remote_cert_path, source_cert): + upload_needed = False + # Try URL check if SSH check failed + elif check_url: + logger.info(f"Checking via URL: {check_url}") + remote_cert = self.cert_manager.get_cert_from_url(check_url) + if remote_cert and self.cert_manager.compare_certificates(source_cert, remote_cert): + logger.info("✓ Certificate up to date via URL. Skipping.") + upload_needed = False + + if not upload_needed: + ssh.disconnect() + self.stats['skipped'] += 1 + return True + + # Upload certificate if not ssh.upload_file(source_cert_path, remote_cert_path): ssh.disconnect() self.stats['failed'] += 1 return False + # Upload key if specified if self.config.has_option(section, 'remote_key_path'): remote_key_path = self.config.get(section, 'remote_key_path') source_key_path = self.get_key_path(section, source_cert_path) - logger.info(f"Uploading private key") - - if not os.path.exists(source_key_path): - logger.error(f"Private key not found: {source_key_path}") - ssh.disconnect() - self.stats['failed'] += 1 - return False - - if not ssh.upload_file(source_key_path, remote_key_path): - logger.warning(f"Failed to upload private key") + if os.path.exists(source_key_path): + ssh.upload_file(source_key_path, remote_key_path) + # Additional files if self.config.has_option(section, 'additional_files'): - additional_files = self.config.get(section, 'additional_files') - for file_pair in additional_files.split(','): + for file_pair in self.config.get(section, 'additional_files').split(','): if ':' in file_pair: local, remote = file_pair.strip().split(':', 1) - logger.info(f"Uploading additional: {os.path.basename(local)}") - if not ssh.upload_file(local, remote): - logger.warning(f"Failed to upload additional file") + ssh.upload_file(local, remote) + # Post-upload command if post_upload_command: - logger.info(f"Executing post-upload command") - success, stdout, stderr = ssh.execute_command(post_upload_command) - - if not success: - logger.warning(f"Post-upload command failed") - else: - logger.info(f"✓ Post-upload command completed") + logger.info("Executing post-upload command") + ssh.execute_command(post_upload_command) ssh.disconnect() self.stats['uploaded'] += 1 - logger.info(f"✓ Host processed successfully") + logger.info("✓ Host processed") return True except Exception as e: - logger.error(f"Failed to process host: {e}") + logger.error(f"Failed: {e}") self.stats['failed'] += 1 return False def run(self): - """Main execution method""" + """Main execution""" logger.info("="*60) logger.info(" CertPusher - SSL Certificate Distribution") logger.info("="*60) logger.info(f"Started: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}") - logger.info(f"Log file: {LOG_FILE}") - logger.info("") + logger.info(f"Log: {LOG_FILE}\n") if not self.load_config(): - logger.error("Configuration loading failed") sys.exit(1) source_cert = self.config.get('global', 'source_cert_path') @@ -761,31 +700,21 @@ class CertPusher: logger.error(f"Source certificate not found: {source_cert}") sys.exit(1) - logger.info(f"Source certificate: {source_cert}") - - cert = self.cert_manager.get_cert_from_file(source_cert) - if cert: - logger.info(self.cert_manager.get_certificate_info(cert)) + logger.info(f"Global certificate: {source_cert}\n") for section in self.config.sections(): if section == 'global': continue - - try: - self.process_host(section) - except Exception as e: - logger.error(f"Unexpected error: {e}") - self.stats['failed'] += 1 + self.process_host(section) logger.info("\n" + "="*60) logger.info(" SUMMARY") logger.info("="*60) - logger.info(f"Total hosts: {self.stats['total']}") - logger.info(f"✓ Uploaded: {self.stats['uploaded']}") - logger.info(f"○ Skipped: {self.stats['skipped']}") - logger.info(f"✗ Failed: {self.stats['failed']}") + logger.info(f"Total: {self.stats['total']}") + logger.info(f"✓ Uploaded: {self.stats['uploaded']}") + logger.info(f"○ Skipped: {self.stats['skipped']}") + logger.info(f"✗ Failed: {self.stats['failed']}") logger.info("="*60) - logger.info(f"Finished: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}") if self.stats['failed'] > 0: sys.exit(1) @@ -793,42 +722,30 @@ class CertPusher: def main(): """Entry point""" - parser = argparse.ArgumentParser( - description='CertPusher - SSL Certificate Distribution Tool', - formatter_class=argparse.RawDescriptionHelpFormatter, - epilog=""" -Examples: - %(prog)s config.ini # Normal operation - %(prog)s config.ini --debug # Debug mode with verbose logging - %(prog)s config.ini -d # Same as above - """ - ) - - parser.add_argument('config', help='Path to configuration file') + parser = argparse.ArgumentParser(description='CertPusher - SSL Certificate Distribution') + parser.add_argument('config', help='Configuration file') parser.add_argument('-d', '--debug', action='store_true', help='Enable debug logging') - parser.add_argument('-v', '--version', action='version', version='CertPusher 1.0') - + parser.add_argument('-v', '--version', action='version', version='CertPusher 1.1') args = parser.parse_args() - # Setup logging based on debug flag setup_logging(debug=args.debug) print(""" ╔═══════════════════════════════════════════════════════════╗ -║ CertPusher v1.0 ║ +║ CertPusher v1.1 ║ ║ Automated SSL Certificate Distribution Tool ║ ╚═══════════════════════════════════════════════════════════╝ """) if not os.path.exists(args.config): - print(f"Error: Configuration file '{args.config}' not found") + print(f"Error: Config file '{args.config}' not found") sys.exit(1) try: pusher = CertPusher(args.config) pusher.run() except KeyboardInterrupt: - print("\n\nInterrupted by user. Exiting...") + print("\n\nInterrupted. Exiting...") sys.exit(130) except Exception as e: logger.error(f"Fatal error: {e}", exc_info=True)