mikrotik check cert
This commit is contained in:
Mateusz Gruszczyński
271
certpusher.py
271
certpusher.py
@@ -332,158 +332,157 @@ class MikroTikManager(SSHManager):
|
|||||||
return True
|
return True
|
||||||
|
|
||||||
def upload_certificate(self, cert_path: str, key_path: str, check_first: bool, source_cert: x509.Certificate, services: list = None) -> Tuple[bool, bool]:
|
def upload_certificate(self, cert_path: str, key_path: str, check_first: bool, source_cert: x509.Certificate, services: list = None) -> Tuple[bool, bool]:
|
||||||
"""
|
"""
|
||||||
Upload certificate to MikroTik
|
Upload certificate to MikroTik
|
||||||
|
|
||||||
Args:
|
Args:
|
||||||
cert_path: Path to certificate file
|
cert_path: Path to certificate file
|
||||||
key_path: Path to private key file
|
key_path: Path to private key file
|
||||||
check_first: Whether to check existing certificate first
|
check_first: Whether to check existing certificate first
|
||||||
source_cert: Source certificate object for comparison
|
source_cert: Source certificate object for comparison
|
||||||
services: List of services to configure (e.g. ['www-ssl', 'api-ssl'])
|
services: List of services to configure (e.g. ['www-ssl', 'api-ssl'])
|
||||||
"""
|
"""
|
||||||
try:
|
|
||||||
if check_first and source_cert:
|
|
||||||
if not self.check_certificate_expiry(source_cert):
|
|
||||||
return True, False
|
|
||||||
|
|
||||||
# Default to www-ssl if not specified
|
|
||||||
if not services:
|
|
||||||
services = ['www-ssl']
|
|
||||||
|
|
||||||
logger.info(f"Deploying MikroTik certificate for services: {', '.join(services)}")
|
|
||||||
|
|
||||||
# Step 1: Disable all services that will use the certificate
|
|
||||||
for service in services:
|
|
||||||
logger.info(f"Disabling {service} service")
|
|
||||||
self.execute_command(f'/ip service disable {service}', ignore_error=True)
|
|
||||||
|
|
||||||
import time
|
|
||||||
time.sleep(1)
|
|
||||||
|
|
||||||
# Step 2: Remove old files and certificates
|
|
||||||
logger.info("Cleaning up old certificates")
|
|
||||||
cleanup_commands = [
|
|
||||||
'/certificate remove [find name~"letsencrypt"]',
|
|
||||||
'/file remove "letsencrypt.pem"',
|
|
||||||
'/file remove "letsencrypt-key.pem"',
|
|
||||||
]
|
|
||||||
|
|
||||||
for cmd in cleanup_commands:
|
|
||||||
self.execute_command(cmd, ignore_error=True)
|
|
||||||
|
|
||||||
time.sleep(1)
|
|
||||||
|
|
||||||
# Step 3: Upload certificate via SCP
|
|
||||||
logger.info(f"Uploading certificate: {cert_path} -> letsencrypt.pem")
|
|
||||||
try:
|
try:
|
||||||
with SCPClient(self.ssh_client.get_transport(), progress=None) as scp:
|
if check_first and source_cert:
|
||||||
scp.put(cert_path, 'letsencrypt.pem')
|
if not self.check_certificate_expiry(source_cert):
|
||||||
logger.info("✓ Certificate file uploaded")
|
return True, False
|
||||||
except Exception as e:
|
|
||||||
logger.error(f"SCP upload failed: {e}")
|
|
||||||
return False, False
|
|
||||||
|
|
||||||
# Step 4: Upload private key
|
# Default to www-ssl if not specified
|
||||||
if key_path:
|
if not services:
|
||||||
logger.info(f"Uploading key: {key_path} -> letsencrypt-key.pem")
|
services = ['www-ssl']
|
||||||
|
|
||||||
|
logger.info(f"Deploying MikroTik certificate for services: {', '.join(services)}")
|
||||||
|
|
||||||
|
# Step 1: Disable all services that will use the certificate
|
||||||
|
for service in services:
|
||||||
|
logger.info(f"Disabling {service} service")
|
||||||
|
self.execute_command(f'/ip service disable {service}', ignore_error=True)
|
||||||
|
|
||||||
|
import time
|
||||||
|
time.sleep(1)
|
||||||
|
|
||||||
|
# Step 2: Remove old files and certificates
|
||||||
|
logger.info("Cleaning up old certificates")
|
||||||
|
cleanup_commands = [
|
||||||
|
'/certificate remove [find name~"letsencrypt"]',
|
||||||
|
'/file remove "letsencrypt.pem"',
|
||||||
|
'/file remove "letsencrypt-key.pem"',
|
||||||
|
]
|
||||||
|
|
||||||
|
for cmd in cleanup_commands:
|
||||||
|
self.execute_command(cmd, ignore_error=True)
|
||||||
|
|
||||||
|
time.sleep(1)
|
||||||
|
|
||||||
|
# Step 3: Upload certificate via SCP
|
||||||
|
logger.info(f"Uploading certificate: {cert_path} -> letsencrypt.pem")
|
||||||
try:
|
try:
|
||||||
with SCPClient(self.ssh_client.get_transport(), progress=None) as scp:
|
with SCPClient(self.ssh_client.get_transport(), progress=None) as scp:
|
||||||
scp.put(key_path, 'letsencrypt-key.pem')
|
scp.put(cert_path, 'letsencrypt.pem')
|
||||||
logger.info("✓ Key file uploaded")
|
logger.info("✓ Certificate file uploaded")
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
logger.error(f"Key upload failed: {e}")
|
logger.error(f"SCP upload failed: {e}")
|
||||||
return False, False
|
return False, False
|
||||||
|
|
||||||
time.sleep(2)
|
# Step 4: Upload private key
|
||||||
|
if key_path:
|
||||||
|
logger.info(f"Uploading key: {key_path} -> letsencrypt-key.pem")
|
||||||
|
try:
|
||||||
|
with SCPClient(self.ssh_client.get_transport(), progress=None) as scp:
|
||||||
|
scp.put(key_path, 'letsencrypt-key.pem')
|
||||||
|
logger.info("✓ Key file uploaded")
|
||||||
|
except Exception as e:
|
||||||
|
logger.error(f"Key upload failed: {e}")
|
||||||
|
return False, False
|
||||||
|
|
||||||
# Step 5: Verify files were uploaded
|
time.sleep(2)
|
||||||
logger.info("Verifying uploaded files...")
|
|
||||||
success, stdout, stderr = self.execute_command(
|
|
||||||
'/file print where name~"letsencrypt"',
|
|
||||||
ignore_error=True
|
|
||||||
)
|
|
||||||
|
|
||||||
if not success or 'letsencrypt.pem' not in stdout:
|
# Step 5: Verify files were uploaded
|
||||||
logger.error("Certificate file not found on MikroTik!")
|
logger.info("Verifying uploaded files...")
|
||||||
self.execute_command('/file print')
|
|
||||||
return False, False
|
|
||||||
|
|
||||||
logger.info("✓ Files verified on MikroTik")
|
|
||||||
logger.debug(f"Files:\n{stdout}")
|
|
||||||
|
|
||||||
# Step 6: Import certificate
|
|
||||||
logger.info("Importing certificate into MikroTik")
|
|
||||||
import_cmd = '/certificate import file-name=letsencrypt.pem passphrase=""'
|
|
||||||
success, stdout, stderr = self.execute_command(import_cmd, timeout=30)
|
|
||||||
|
|
||||||
if not success:
|
|
||||||
logger.error(f"Import failed: {stderr}")
|
|
||||||
return False, False
|
|
||||||
|
|
||||||
logger.info("✓ Certificate imported")
|
|
||||||
logger.debug(f"Import output: {stdout}")
|
|
||||||
|
|
||||||
time.sleep(2)
|
|
||||||
|
|
||||||
# Step 7: Use predictable certificate name
|
|
||||||
# MikroTik always creates filename_0 for the leaf certificate
|
|
||||||
imported_cert_name = "letsencrypt.pem_0"
|
|
||||||
logger.info(f"Using certificate: {imported_cert_name}")
|
|
||||||
|
|
||||||
# Verify it exists
|
|
||||||
success, stdout, stderr = self.execute_command(
|
|
||||||
f'/certificate print where name="{imported_cert_name}"'
|
|
||||||
)
|
|
||||||
|
|
||||||
if not success or not stdout:
|
|
||||||
logger.warning(f"Certificate {imported_cert_name} not found, trying to find it...")
|
|
||||||
success, stdout, stderr = self.execute_command('/certificate print where name~"letsencrypt"')
|
|
||||||
if stdout:
|
|
||||||
logger.debug(f"Available certificates:\n{stdout}")
|
|
||||||
# Try to parse actual name
|
|
||||||
match = re.search(r'name="?([a-zA-Z0-9._-]*letsencrypt[^"\s]*)"?', stdout)
|
|
||||||
if match:
|
|
||||||
imported_cert_name = match.group(1)
|
|
||||||
logger.info(f"Found certificate: {imported_cert_name}")
|
|
||||||
|
|
||||||
# Step 8: Configure all specified services
|
|
||||||
for service in services:
|
|
||||||
logger.info(f"Configuring {service} to use certificate")
|
|
||||||
config_cmd = f'/ip service set {service} certificate="{imported_cert_name}"'
|
|
||||||
success, stdout, stderr = self.execute_command(config_cmd)
|
|
||||||
|
|
||||||
if not success:
|
|
||||||
logger.error(f"Failed to configure {service}: {stderr}")
|
|
||||||
logger.warning(f"Continuing with other services...")
|
|
||||||
else:
|
|
||||||
logger.info(f"✓ {service} configured")
|
|
||||||
|
|
||||||
# Step 9: Enable all services
|
|
||||||
for service in services:
|
|
||||||
logger.info(f"Enabling {service} service")
|
|
||||||
self.execute_command(f'/ip service enable {service}')
|
|
||||||
|
|
||||||
time.sleep(1)
|
|
||||||
|
|
||||||
# Step 10: Verify services status
|
|
||||||
for service in services:
|
|
||||||
success, stdout, stderr = self.execute_command(
|
success, stdout, stderr = self.execute_command(
|
||||||
f'/ip service print where name="{service}"'
|
'/file print where name~"letsencrypt"',
|
||||||
|
ignore_error=True
|
||||||
)
|
)
|
||||||
|
|
||||||
if success and stdout:
|
if not success or 'letsencrypt.pem' not in stdout:
|
||||||
logger.debug(f"{service} status:\n{stdout}")
|
logger.error("Certificate file not found on MikroTik!")
|
||||||
|
self.execute_command('/file print')
|
||||||
|
return False, False
|
||||||
|
|
||||||
logger.info(f"✓ MikroTik deployment completed successfully")
|
logger.info("✓ Files verified on MikroTik")
|
||||||
return True, True
|
logger.debug(f"Files:\n{stdout}")
|
||||||
|
|
||||||
except Exception as e:
|
# Step 6: Import certificate
|
||||||
logger.error(f"MikroTik deployment failed: {e}")
|
logger.info("Importing certificate into MikroTik")
|
||||||
import traceback
|
import_cmd = '/certificate import file-name=letsencrypt.pem passphrase=""'
|
||||||
logger.error(traceback.format_exc())
|
success, stdout, stderr = self.execute_command(import_cmd, timeout=30)
|
||||||
return False, False
|
|
||||||
|
|
||||||
|
if not success:
|
||||||
|
logger.error(f"Import failed: {stderr}")
|
||||||
|
return False, False
|
||||||
|
|
||||||
|
logger.info("✓ Certificate imported")
|
||||||
|
logger.debug(f"Import output: {stdout}")
|
||||||
|
|
||||||
|
time.sleep(2)
|
||||||
|
|
||||||
|
# Step 7: Use predictable certificate name
|
||||||
|
# MikroTik always creates filename_0 for the leaf certificate
|
||||||
|
imported_cert_name = "letsencrypt.pem_0"
|
||||||
|
logger.info(f"Using certificate: {imported_cert_name}")
|
||||||
|
|
||||||
|
# Verify it exists
|
||||||
|
success, stdout, stderr = self.execute_command(
|
||||||
|
f'/certificate print where name="{imported_cert_name}"'
|
||||||
|
)
|
||||||
|
|
||||||
|
if not success or not stdout:
|
||||||
|
logger.warning(f"Certificate {imported_cert_name} not found, trying to find it...")
|
||||||
|
success, stdout, stderr = self.execute_command('/certificate print where name~"letsencrypt"')
|
||||||
|
if stdout:
|
||||||
|
logger.debug(f"Available certificates:\n{stdout}")
|
||||||
|
# Try to parse actual name
|
||||||
|
match = re.search(r'name="?([a-zA-Z0-9._-]*letsencrypt[^"\s]*)"?', stdout)
|
||||||
|
if match:
|
||||||
|
imported_cert_name = match.group(1)
|
||||||
|
logger.info(f"Found certificate: {imported_cert_name}")
|
||||||
|
|
||||||
|
# Step 8: Configure all specified services
|
||||||
|
for service in services:
|
||||||
|
logger.info(f"Configuring {service} to use certificate")
|
||||||
|
config_cmd = f'/ip service set {service} certificate="{imported_cert_name}"'
|
||||||
|
success, stdout, stderr = self.execute_command(config_cmd)
|
||||||
|
|
||||||
|
if not success:
|
||||||
|
logger.error(f"Failed to configure {service}: {stderr}")
|
||||||
|
logger.warning(f"Continuing with other services...")
|
||||||
|
else:
|
||||||
|
logger.info(f"✓ {service} configured")
|
||||||
|
|
||||||
|
# Step 9: Enable all services
|
||||||
|
for service in services:
|
||||||
|
logger.info(f"Enabling {service} service")
|
||||||
|
self.execute_command(f'/ip service enable {service}')
|
||||||
|
|
||||||
|
time.sleep(1)
|
||||||
|
|
||||||
|
# Step 10: Verify services status
|
||||||
|
for service in services:
|
||||||
|
success, stdout, stderr = self.execute_command(
|
||||||
|
f'/ip service print where name="{service}"'
|
||||||
|
)
|
||||||
|
|
||||||
|
if success and stdout:
|
||||||
|
logger.debug(f"{service} status:\n{stdout}")
|
||||||
|
|
||||||
|
logger.info(f"✓ MikroTik deployment completed successfully")
|
||||||
|
return True, True
|
||||||
|
|
||||||
|
except Exception as e:
|
||||||
|
logger.error(f"MikroTik deployment failed: {e}")
|
||||||
|
import traceback
|
||||||
|
logger.error(traceback.format_exc())
|
||||||
|
return False, False
|
||||||
|
|
||||||
class ProxmoxManager(SSHManager):
|
class ProxmoxManager(SSHManager):
|
||||||
"""Specialized manager for Proxmox VE servers"""
|
"""Specialized manager for Proxmox VE servers"""
|
||||||
|
|||||||
Reference in New Issue
Block a user