gru/certpusher
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

mikrotik checks

Browse Source
This commit is contained in:
Mateusz Gruszczyński
2025-10-27 07:12:15 +01:00
parent 7fc8b3894c
commit 5a37e451a5
2 changed files with 348 additions and 205 deletions
Download Patch File Download Diff File Expand all files Collapse all files

384
certpusher.py
View File

@@ -11,7 +11,7 @@ import sys
import os
import ssl
import socket
import tempfile
import re
from datetime import datetime, timezone
from pathlib import Path
from typing import Dict, Optional, Tuple
@@ -120,31 +120,6 @@ 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:
@@ -267,17 +242,86 @@ class MikroTikManager(SSHManager):
self.cert_name = "ssl-cert"
self.key_name = "ssl-key"
def upload_certificate(self, cert_path: str, key_path: str = None) -> bool:
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
"""
try:
logger.info("Checking certificate on MikroTik")
# Get certificate details from MikroTik
success, stdout, stderr = self.execute_command(
f'/certificate print detail where name~"{self.cert_name}"',
ignore_error=True
)
if not success or not stdout:
logger.info("No certificate found on MikroTik or cannot read it. Upload needed.")
return True
logger.debug(f"MikroTik certificate info:\n{stdout}")
# Parse expiry date from output
# Looking for "invalid-after" field
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 date from MikroTik. Proceeding with upload.")
return True
mikrotik_expiry_str = invalid_after_match.group(1)
logger.debug(f"MikroTik certificate expires: {mikrotik_expiry_str}")
# Parse MikroTik date format (e.g., "jan/24/2026 08:34:12")
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 MikroTik date: {e}. Proceeding with upload.")
return True
# Compare with source certificate
source_expiry = source_cert.not_valid_after_utc
# Also check fingerprint/serial if available
fingerprint_match = re.search(r'fingerprint:\s+([a-f0-9]+)', stdout)
if fingerprint_match:
mikrotik_fingerprint = fingerprint_match.group(1)
logger.debug(f"MikroTik certificate fingerprint: {mikrotik_fingerprint}")
# Compare expiry dates (allowing 1 day tolerance for timezone differences)
time_diff = abs((source_expiry - mikrotik_expiry).total_seconds())
if time_diff < 86400: # Less than 24 hours difference
logger.info("✓ Certificate on MikroTik is current. Skipping upload.")
return False
else:
logger.info(f"Certificate on MikroTik differs. Source expires: {source_expiry}, MikroTik expires: {mikrotik_expiry}")
return True
except Exception as e:
logger.warning(f"Error checking MikroTik certificate: {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
Args:
cert_path: Path to certificate file (PEM format, can be fullchain)
key_path: Optional path to private key file
check_first: If True, check existing certificate before uploading
source_cert: Source certificate object for comparison
"""
try:
logger.info(f"Starting MikroTik certificate deployment to {self.hostname}")
# Check if upload is needed
if check_first and source_cert:
if not self.check_certificate_expiry(source_cert):
return True # Certificate is current, no upload needed
# Step 1: Disable www-ssl service
logger.debug("Disabling www-ssl service")
self.execute_command('/ip service disable www-ssl', ignore_error=True)
@@ -470,20 +514,45 @@ class CertPusher:
logger.error(f"Failed to load configuration: {e}")
return False
def get_key_path(self, section: str, cert_path: str) -> str:
"""
Get private key path for certificate
Priority: section-specific > global > derived from cert path
"""
# Check section-specific key
if self.config.has_option(section, 'source_key_path'):
return self.config.get(section, 'source_key_path')
# Check global key
if self.config.has_option('global', 'source_key_path'):
return self.config.get('global', 'source_key_path')
# Derive from certificate path
key_path = cert_path.replace('fullchain.pem', 'privkey.pem').replace('cert.pem', 'privkey.pem')
logger.debug(f"Derived key path: {key_path}")
return key_path
def process_mikrotik(self, section: str, hostname: str, port: int,
username: str, ssh_key: str, source_cert_path: str) -> bool:
"""Process MikroTik device specifically"""
try:
logger.info("Using MikroTik-specific deployment method")
# Get optional private key path
source_key_path = self.config.get(section, 'source_key_path', fallback=None)
# Get private key path
source_key_path = self.get_key_path(section, source_cert_path)
if source_key_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
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
# Load source certificate for comparison
source_cert = self.cert_manager.get_cert_from_file(source_cert_path)
# Check if certificate check is enabled
check_first = self.config.getboolean(section, 'check_before_upload', fallback=True)
# Connect
mikrotik = MikroTikManager(hostname, port, username, ssh_key)
@@ -492,17 +561,29 @@ class CertPusher:
self.stats['failed'] += 1
return False
# Upload and import certificate
if not mikrotik.upload_certificate(source_cert_path, source_key_path):
# Upload and import certificate (with optional check)
if not mikrotik.upload_certificate(source_cert_path, source_key_path, check_first, source_cert):
mikrotik.disconnect()
self.stats['failed'] += 1
return False
# If we got here and check was enabled, certificate was actually uploaded
if check_first and source_cert:
# Check if it was actually uploaded or skipped
success, stdout, stderr = mikrotik.execute_command(
f'/certificate print where name~"{mikrotik.cert_name}"'
)
if success and stdout and mikrotik.cert_name in stdout:
self.stats['uploaded'] += 1
else:
self.stats['skipped'] += 1
else:
self.stats['uploaded'] += 1
# Verify installation
mikrotik.verify_certificate()
mikrotik.disconnect()
self.stats['uploaded'] += 1
logger.info(f"✓ Successfully processed MikroTik {section}")
return True
@@ -518,11 +599,7 @@ class CertPusher:
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')
source_key_path = self.get_key_path(section, source_cert_path)
logger.info(f"Certificate: {source_cert_path}")
logger.info(f"Private key: {source_key_path}")
@@ -555,117 +632,136 @@ class CertPusher:
return False
def process_host(self, section: str) -> bool:
"""Process certificate deployment for a single host"""
try:
logger.info(f"\n{'='*60}")
logger.info(f"Processing host: {section}")
logger.info(f"{'='*60}")
"""Process certificate deployment for a single host"""
try:
logger.info(f"\n{'='*60}")
logger.info(f"Processing host: {section}")
logger.info(f"{'='*60}")
self.stats['total'] += 1
# Get configuration
hostname = self.config.get(section, 'hostname')
port = self.config.getint(section, 'port', fallback=22)
username = self.config.get(section, 'username', fallback='root')
device_type = self.config.get(section, 'type', fallback='standard')
# Determine SSH key to use
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')
# 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 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')
post_upload_command = self.config.get(section, 'post_upload_command', fallback='')
check_url = self.config.get(section, 'check_url', fallback='')
logger.info(f"Remote certificate path: {remote_cert_path}")
# Check if upload is needed
if check_url:
logger.info(f"Checking current 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)
self.stats['total'] += 1
# Get configuration
hostname = self.config.get(section, 'hostname')
port = self.config.getint(section, 'port', fallback=22)
username = self.config.get(section, 'username', fallback='root')
device_type = self.config.get(section, 'type', fallback='standard')
# Determine SSH key to use
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')
# 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 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')
post_upload_command = self.config.get(section, 'post_upload_command', fallback='')
check_url = self.config.get(section, 'check_url', fallback='')
logger.info(f"Remote path: {remote_cert_path}")
# Check if upload is needed
if check_url:
logger.info(f"Checking current 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 on {hostname} is already up to date. Skipping upload.")
self.stats['skipped'] += 1
return True
else:
logger.info(f"Certificate on {hostname} is outdated. Upload needed.")
logger.debug(self.cert_manager.get_certificate_info(source_cert))
if source_cert and remote_cert:
if self.cert_manager.compare_certificates(source_cert, remote_cert):
logger.info(f"✓ Certificate on {hostname} is already up to date. Skipping upload.")
self.stats['skipped'] += 1
return True
else:
logger.warning(f"Could not compare certificates. Proceeding with upload.")
logger.info(f"Certificate on {hostname} is outdated. Upload needed.")
logger.debug(self.cert_manager.get_certificate_info(source_cert))
else:
logger.warning(f"Could not compare certificates. Proceeding with upload.")
# Connect and upload
ssh = SSHManager(hostname, port, username, ssh_key)
if not ssh.connect():
self.stats['failed'] += 1
return False
# Upload certificate
if not ssh.upload_file(source_cert_path, remote_cert_path):
ssh.disconnect()
self.stats['failed'] += 1
return False
# Upload private key if remote_key_path is 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)
# Connect and upload
ssh = SSHManager(hostname, port, username, ssh_key)
logger.info(f"Remote key path: {remote_key_path}")
logger.info(f"Uploading private key: {source_key_path} -> {remote_key_path}")
if not ssh.connect():
self.stats['failed'] += 1
return False
if not ssh.upload_file(source_cert_path, remote_cert_path):
if not os.path.exists(source_key_path):
logger.error(f"Private key file not found: {source_key_path}")
ssh.disconnect()
self.stats['failed'] += 1
return False
# Upload additional files if specified
if self.config.has_option(section, 'additional_files'):
additional_files = self.config.get(section, 'additional_files')
# Format: local_path:remote_path,local_path:remote_path
for file_pair in additional_files.split(','):
if ':' in file_pair:
local, remote = file_pair.strip().split(':', 1)
logger.info(f"Uploading additional file: {local} -> {remote}")
if not ssh.upload_file(local, remote):
logger.warning(f"Failed to upload additional file: {local}")
if not ssh.upload_file(source_key_path, remote_key_path):
logger.warning(f"Failed to upload private key to {remote_key_path}")
# Continue anyway - key might be uploaded via additional_files
# Upload additional files if specified
if self.config.has_option(section, 'additional_files'):
additional_files = self.config.get(section, 'additional_files')
# Format: local_path:remote_path,local_path:remote_path
for file_pair in additional_files.split(','):
if ':' in file_pair:
local, remote = file_pair.strip().split(':', 1)
logger.info(f"Uploading additional file: {local} -> {remote}")
if not ssh.upload_file(local, remote):
logger.warning(f"Failed to upload additional file: {local}")
# Execute post-upload command
if post_upload_command:
logger.info(f"Executing post-upload command: {post_upload_command}")
success, stdout, stderr = ssh.execute_command(post_upload_command)
# Execute post-upload command
if post_upload_command:
logger.info(f"Executing post-upload command: {post_upload_command}")
success, stdout, stderr = ssh.execute_command(post_upload_command)
if not success:
logger.warning(f"Post-upload command failed, but file was uploaded successfully")
else:
logger.info(f"✓ Post-upload command completed successfully")
ssh.disconnect()
self.stats['uploaded'] += 1
logger.info(f"✓ Successfully processed {section}")
return True
except Exception as e:
logger.error(f"Failed to process host {section}: {e}", exc_info=True)
self.stats['failed'] += 1
return False
if not success:
logger.warning(f"Post-upload command failed, but file was uploaded successfully")
else:
logger.info(f"✓ Post-upload command completed successfully")
ssh.disconnect()
self.stats['uploaded'] += 1
logger.info(f"✓ Successfully processed {section}")
return True
except Exception as e:
logger.error(f"Failed to process host {section}: {e}", exc_info=True)
self.stats['failed'] += 1
return False
def run(self):
"""Main execution method"""