SSL Proxy

system/proxy_module/cert_manager.py

@@ -0,0 +1,120 @@
+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