120 lines
4.8 KiB
Python
120 lines
4.8 KiB
Python
import logging
|
|
import os
|
|
from pathlib import Path
|
|
from typing import Dict, Any, Optional
|
|
from datetime import datetime
|
|
import ssl
|
|
import OpenSSL
|
|
from dataclasses import dataclass
|
|
|
|
@dataclass
|
|
class CertificateInfo:
|
|
subject: str
|
|
issuer: str
|
|
valid_from: datetime
|
|
valid_until: datetime
|
|
serial_number: str
|
|
is_valid: bool
|
|
|
|
class CertificateManager:
|
|
def __init__(self, proxy_manager):
|
|
self.proxy_manager = proxy_manager
|
|
self.logger = logging.getLogger("ProxyCertManager")
|
|
self.certs_path = proxy_manager.certs_path
|
|
|
|
def validate_certificate(self, cert_path: Path) -> Optional[CertificateInfo]:
|
|
"""Validate and get information about a certificate"""
|
|
try:
|
|
with open(cert_path, 'rb') as cert_file:
|
|
cert_data = cert_file.read()
|
|
x509 = OpenSSL.crypto.load_certificate(OpenSSL.crypto.FILETYPE_PEM, cert_data)
|
|
|
|
valid_from = datetime.strptime(x509.get_notBefore().decode(), '%Y%m%d%H%M%SZ')
|
|
valid_until = datetime.strptime(x509.get_notAfter().decode(), '%Y%m%d%H%M%SZ')
|
|
now = datetime.now()
|
|
|
|
return CertificateInfo(
|
|
subject=x509.get_subject().CN,
|
|
issuer=x509.get_issuer().CN,
|
|
valid_from=valid_from,
|
|
valid_until=valid_until,
|
|
serial_number=str(x509.get_serial_number()),
|
|
is_valid=valid_from <= now <= valid_until
|
|
)
|
|
except Exception as e:
|
|
self.logger.error(f"Certificate validation error: {e}")
|
|
return None
|
|
|
|
def install_certificate(self, cert_path: Path, key_path: Path, name: str) -> bool:
|
|
try:
|
|
import shutil
|
|
# Validate certificate first
|
|
cert_info = self.validate_certificate(cert_path)
|
|
if not cert_info or not cert_info.is_valid:
|
|
self.logger.error("Invalid or corrupt certificate")
|
|
return False
|
|
|
|
cert_dest = self.certs_path / f"{name}_cert.pem"
|
|
key_dest = self.certs_path / f"{name}_key.pem"
|
|
|
|
shutil.copy2(cert_path, cert_dest)
|
|
shutil.copy2(key_path, key_dest)
|
|
|
|
# Update config
|
|
config = self.proxy_manager.config_manager.get_instance()
|
|
config.proxy_settings.gradio_endpoint.cert_name = name
|
|
config.proxy_settings.api_endpoint.cert_name = name
|
|
config.save()
|
|
|
|
self.logger.info(f"Certificate '{name}' installed successfully")
|
|
return True
|
|
|
|
except Exception as e:
|
|
self.logger.error(f"Certificate installation error: {e}")
|
|
return False
|
|
|
|
def remove_certificate(self, name: str) -> bool:
|
|
"""Remove a certificate"""
|
|
try:
|
|
cert_path = self.certs_path / f"{name}_cert.pem"
|
|
key_path = self.certs_path / f"{name}_key.pem"
|
|
|
|
if cert_path.exists():
|
|
cert_path.unlink()
|
|
if key_path.exists():
|
|
key_path.unlink()
|
|
|
|
self.logger.info(f"Certificate '{name}' removed successfully")
|
|
return True
|
|
except Exception as e:
|
|
self.logger.error(f"Certificate removal error: {e}")
|
|
return False
|
|
|
|
def get_certificates_status(self) -> Dict[str, Any]:
|
|
"""Get status of all installed certificates"""
|
|
certificates = {}
|
|
for cert_file in self.certs_path.glob("*_cert.pem"):
|
|
name = cert_file.name.replace("_cert.pem", "")
|
|
cert_info = self.validate_certificate(cert_file)
|
|
if cert_info:
|
|
certificates[name] = {
|
|
"subject": cert_info.subject,
|
|
"issuer": cert_info.issuer,
|
|
"valid_from": cert_info.valid_from.strftime('%Y-%m-%d %H:%M:%S'),
|
|
"valid_until": cert_info.valid_until.strftime('%Y-%m-%d %H:%M:%S'),
|
|
"is_valid": cert_info.is_valid,
|
|
"days_until_expiry": (cert_info.valid_until - datetime.now()).days
|
|
}
|
|
return certificates
|
|
|
|
def check_expiring_certificates(self, days_warning: int = 30) -> Dict[str, int]:
|
|
"""Check for certificates nearing expiration"""
|
|
expiring = {}
|
|
for cert_file in self.certs_path.glob("*_cert.pem"):
|
|
name = cert_file.name.replace("_cert.pem", "")
|
|
cert_info = self.validate_certificate(cert_file)
|
|
if cert_info:
|
|
days_left = (cert_info.valid_until - datetime.now()).days
|
|
if days_left < days_warning:
|
|
expiring[name] = days_left
|
|
return expiring |