proxmox class
This commit is contained in:
Mateusz Gruszczyński
156
certpusher.py
156
certpusher.py
@@ -2,7 +2,7 @@
|
||||
"""
|
||||
CertPusher - Automated SSL Certificate Distribution Tool
|
||||
Distributes SSL certificates to remote servers via SSH/SCP
|
||||
Supports standard Linux servers and MikroTik RouterOS devices
|
||||
Supports standard Linux servers, MikroTik RouterOS, and Proxmox VE
|
||||
"""
|
||||
|
||||
import configparser
|
||||
@@ -11,6 +11,7 @@ import sys
|
||||
import os
|
||||
import ssl
|
||||
import socket
|
||||
import tempfile
|
||||
from datetime import datetime, timezone
|
||||
from pathlib import Path
|
||||
from typing import Dict, Optional, Tuple
|
||||
@@ -119,6 +120,31 @@ Certificate Info:
|
||||
"""
|
||||
except Exception as e:
|
||||
return f"Unable to extract certificate info: {e}"
|
||||
|
||||
@staticmethod
|
||||
def create_combined_cert(cert_path: str, key_path: str, output_path: str) -> bool:
|
||||
"""Create combined certificate file (cert + key) - used by Proxmox"""
|
||||
try:
|
||||
logger.debug(f"Creating combined certificate: {cert_path} + {key_path} -> {output_path}")
|
||||
|
||||
with open(cert_path, 'r') as cert_file:
|
||||
cert_content = cert_file.read()
|
||||
|
||||
with open(key_path, 'r') as key_file:
|
||||
key_content = key_file.read()
|
||||
|
||||
# Combined format: certificate + private key
|
||||
combined_content = cert_content.strip() + "\n" + key_content.strip() + "\n"
|
||||
|
||||
with open(output_path, 'w') as combined_file:
|
||||
combined_file.write(combined_content)
|
||||
|
||||
logger.info(f"✓ Combined certificate created at {output_path}")
|
||||
return True
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to create combined certificate: {e}")
|
||||
return False
|
||||
|
||||
|
||||
class SSHManager:
|
||||
@@ -304,10 +330,8 @@ class MikroTikManager(SSHManager):
|
||||
logger.debug(f"Certificates after import:\n{stdout}")
|
||||
|
||||
# Step 7: Configure services to use new certificate
|
||||
# RouterOS typically names imported certs as certname_0, certname_1, etc.
|
||||
logger.info("Configuring www-ssl service to use new certificate")
|
||||
|
||||
# Find the certificate name (usually ssl-cert_0 for cert, ssl-cert_1 for key)
|
||||
config_commands = [
|
||||
f'/ip service set www-ssl certificate={self.cert_name}_0',
|
||||
'/ip service enable www-ssl',
|
||||
@@ -343,6 +367,71 @@ class MikroTikManager(SSHManager):
|
||||
return False
|
||||
|
||||
|
||||
class ProxmoxManager(SSHManager):
|
||||
"""Specialized manager for Proxmox VE servers"""
|
||||
|
||||
def upload_certificate(self, cert_path: str, key_path: str) -> bool:
|
||||
"""
|
||||
Upload certificate to Proxmox VE
|
||||
|
||||
Proxmox uses two separate files:
|
||||
- /etc/pve/local/pveproxy-ssl.pem (certificate)
|
||||
- /etc/pve/local/pveproxy-ssl.key (private key)
|
||||
|
||||
Args:
|
||||
cert_path: Path to certificate file (fullchain)
|
||||
key_path: Path to private key file
|
||||
"""
|
||||
try:
|
||||
logger.info(f"Starting Proxmox certificate deployment to {self.hostname}")
|
||||
|
||||
# Step 1: Upload certificate
|
||||
logger.debug("Uploading certificate to Proxmox")
|
||||
if not self.upload_file(cert_path, '/etc/pve/local/pveproxy-ssl.pem'):
|
||||
return False
|
||||
|
||||
# Step 2: Upload private key
|
||||
logger.debug("Uploading private key to Proxmox")
|
||||
if not self.upload_file(key_path, '/etc/pve/local/pveproxy-ssl.key'):
|
||||
return False
|
||||
|
||||
# Step 3: Set correct permissions
|
||||
logger.debug("Setting file 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)
|
||||
|
||||
# Step 4: Restart pveproxy
|
||||
logger.info("Restarting pveproxy service")
|
||||
success, stdout, stderr = self.execute_command('systemctl restart pveproxy', timeout=30)
|
||||
|
||||
if not success:
|
||||
logger.error(f"Failed to restart pveproxy: {stderr}")
|
||||
return False
|
||||
|
||||
# Step 5: Verify service is running
|
||||
import time
|
||||
time.sleep(3)
|
||||
|
||||
success, stdout, stderr = self.execute_command('systemctl is-active pveproxy')
|
||||
if success and 'active' in stdout:
|
||||
logger.info(f"✓ Successfully deployed certificate to Proxmox {self.hostname}")
|
||||
return True
|
||||
else:
|
||||
logger.error("pveproxy service is not active after restart")
|
||||
# Show journal logs for debugging
|
||||
self.execute_command('journalctl -u pveproxy -n 20 --no-pager')
|
||||
return False
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Proxmox certificate deployment failed: {e}", exc_info=True)
|
||||
return False
|
||||
|
||||
|
||||
class CertPusher:
|
||||
"""Main application class"""
|
||||
|
||||
@@ -422,6 +511,49 @@ class CertPusher:
|
||||
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"""
|
||||
try:
|
||||
logger.info("Using Proxmox-specific deployment method")
|
||||
|
||||
# Get private key path
|
||||
if self.config.has_option(section, 'source_key_path'):
|
||||
source_key_path = self.config.get(section, 'source_key_path')
|
||||
else:
|
||||
# Try to derive from cert path
|
||||
source_key_path = source_cert_path.replace('fullchain.pem', 'privkey.pem')
|
||||
|
||||
logger.info(f"Certificate: {source_cert_path}")
|
||||
logger.info(f"Private key: {source_key_path}")
|
||||
|
||||
if not os.path.exists(source_key_path):
|
||||
logger.error(f"Private key file not found: {source_key_path}")
|
||||
return False
|
||||
|
||||
# Connect
|
||||
proxmox = ProxmoxManager(hostname, port, username, ssh_key)
|
||||
|
||||
if not proxmox.connect():
|
||||
self.stats['failed'] += 1
|
||||
return False
|
||||
|
||||
# Upload certificate
|
||||
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"✓ Successfully processed Proxmox {section}")
|
||||
return True
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Proxmox processing failed: {e}", exc_info=True)
|
||||
self.stats['failed'] += 1
|
||||
return False
|
||||
|
||||
def process_host(self, section: str) -> bool:
|
||||
"""Process certificate deployment for a single host"""
|
||||
try:
|
||||
@@ -443,16 +575,30 @@ class CertPusher:
|
||||
else:
|
||||
ssh_key = self.config.get('global', 'default_ssh_key')
|
||||
|
||||
source_cert_path = self.config.get('global', 'source_cert_path')
|
||||
# Allow per-host certificate override
|
||||
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: {source_cert_path}")
|
||||
else:
|
||||
source_cert_path = self.config.get('global', 'source_cert_path')
|
||||
logger.debug(f"Using global certificate: {source_cert_path}")
|
||||
|
||||
# Verify certificate exists
|
||||
if not os.path.exists(source_cert_path):
|
||||
logger.error(f"Certificate file 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"Username: {username}")
|
||||
logger.info(f"SSH Key: {ssh_key}")
|
||||
|
||||
# Handle MikroTik devices specially
|
||||
# Handle device-specific deployments
|
||||
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 processing for other devices
|
||||
remote_cert_path = self.config.get(section, 'remote_cert_path')
|
||||
|
||||
Reference in New Issue
Block a user