""" SSL/Let's Encrypt checks module """ from tests.utils.logger import Logger from tests.utils.docker_utils import DockerUtils import re import requests import urllib3 # Suppress urllib3 warnings for unverified HTTPS requests (expected in development) urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) class SSLChecks: """Check SSL certificate status and Let's Encrypt configuration""" def __init__(self, logger: Logger, docker_utils: DockerUtils, domain: str = None, version: str = "unknown"): self.logger = logger self.version = version self.docker_utils = docker_utils self.domain = domain if domain else "localhost" def run_all_checks(self) -> bool: """Run all SSL checks""" self.logger.subheader(f"SSL Certificate Check (v{self.version})") # Skip SSL check for localhost/development if self.domain == "localhost" or self.domain.startswith("127.0.0.1") or self.domain.startswith("192.168."): self.logger.info("SSL", "Localhost/Local IP detected - skipping SSL certificate check (expected for development)") return True all_ok = True # Check if domain is configured if not self._check_domain_configured(): all_ok = False # Check Caddy container status if not self._check_caddy_running(): all_ok = False # Check SSL certificate if self.domain: if not self._check_ssl_certificate(): all_ok = False else: self.logger.warning("SSL", "No domain configured - skipping SSL certificate check") if all_ok: self.logger.success("SSL", "All SSL checks passed") else: self.logger.error("SSL", "Some SSL checks failed") return all_ok def _check_domain_configured(self) -> bool: """Check if domain is configured""" self.logger.info("SSL", "Checking domain configuration...") if self.domain: self.logger.success("SSL", f"Domain configured: {self.domain} ✓") return True else: self.logger.warning("SSL", "No domain configured (using localhost)") return True # Not critical for development def _check_caddy_running(self) -> bool: """Check if Caddy container is running""" self.logger.info("SSL", "Checking Caddy container status...") status = self.docker_utils.get_container_status("streama2026-caddy") if status == "running": self.logger.success("SSL", "Caddy container: running ✓") return True elif status: self.logger.error("SSL", f"Caddy container: {status}") return False else: self.logger.error("SSL", "Caddy container: not found") return False def _check_ssl_certificate(self) -> bool: """Check SSL certificate status""" self.logger.info("SSL", f"Checking SSL certificate for {self.domain}...") # Skip SSL certificate check for localhost (expected for development) if self.domain == "localhost" or self.domain.startswith("127.0.0.1") or self.domain.startswith("192.168."): self.logger.info("SSL", "Localhost/Local IP detected - skipping SSL certificate check (expected for development)") return True # Check Caddy logs for certificate status try: logs = self.docker_utils.get_container_logs("streama2026-caddy", tail=100) except Exception as e: self.logger.warning("SSL", f"Could not retrieve Caddy logs: {e}") return True # Not critical for development # Look for certificate acquisition messages cert_obtained = False cert_error = None cert_error_type = None # Check for successful certificate (multiple patterns) success_patterns = [ "certificate obtained successfully", "obtaining certificate", "enabling automatic TLS certificate management", "certificate renewed successfully", "tls handshake successful" ] for pattern in success_patterns: if pattern.lower() in logs.lower(): cert_obtained = True self.logger.success("SSL", "Certificate obtained from Let's Encrypt ✓") break # Rate limit errors are warnings, not critical errors (check first) rate_limit_patterns = [ "too many certificates", "rate limit" ] for pattern in rate_limit_patterns: if re.search(pattern, logs, re.IGNORECASE): self.logger.warning("SSL", f"Let's Encrypt rate limit detected (expected during development/testing)") self.logger.info("SSL", "Rate limits reset after 7 days - see https://letsencrypt.org/docs/rate-limits/") cert_error = pattern cert_error_type = "rate_limit" break # Check for other certificate errors (only if not rate limit) error_patterns = [ "acme:.*error", "failed to obtain certificate", "certificate request failed", "no solution found" ] if not cert_error: for pattern in error_patterns: if re.search(pattern, logs, re.IGNORECASE): cert_error = pattern cert_error_type = "error" self.logger.error("SSL", f"Certificate error detected: {pattern}") break # Check if HTTPS is accessible (actual connectivity test) https_accessible = self._check_https_accessible() # Check if HTTPS is accessible if cert_obtained or https_accessible: self.logger.success("SSL", f"SSL certificate for {self.domain} is active ✓") return True elif cert_error: # Rate limit errors are warnings, not critical errors if cert_error_type == "rate_limit": self.logger.warning("SSL", f"SSL certificate rate limited (expected during development/testing)") return True # Not critical for development else: self.logger.error("SSL", f"SSL certificate failed to obtain: {cert_error}") self._explain_certificate_error(logs, cert_error) return False else: self.logger.warning("SSL", "Certificate status unknown - check Caddy logs") return True # Not critical, may still be in process def _check_https_accessible(self) -> bool: """Check if HTTPS is actually accessible""" try: url = f"https://{self.domain}" # Disable SSL verification for this specific request only (dev/testing) response = requests.get(url, timeout=5, verify=False) if response.status_code == 200: self.logger.success("SSL", f"HTTPS accessible at {url} ✓") return True else: self.logger.warning("SSL", f"HTTPS returned status {response.status_code}") return False except Exception as e: self.logger.warning("SSL", f"HTTPS not accessible: {str(e)}") return False def _explain_certificate_error(self, logs: str, error: str) -> None: """Explain certificate errors and provide solutions""" self.logger.info("SSL", "Analyzing certificate error...") if "rate limit" in error.lower(): self.logger.error("SSL", "Let's Encrypt rate limit exceeded") self.logger.info("SSL", "Wait 7 days or use staging environment") elif "no solution found" in error.lower(): self.logger.error("SSL", "DNS record not found for domain") self.logger.info("SSL", "Ensure A/AAAA record points to server IP") elif "acme:.*error" in error.lower(): self.logger.error("SSL", "ACME protocol error") self.logger.info("SSL", "Check ACME_EMAIL and DOMAIN configuration") # Show relevant log lines self.logger.info("SSL", "Recent Caddy logs:") relevant_logs = self._extract_relevant_logs(logs) for line in relevant_logs[:5]: # Show last 5 relevant lines self.logger.info("SSL", f" {line.strip()}") def _extract_relevant_logs(self, logs: str) -> list: """Extract log lines related to certificate acquisition""" lines = logs.split('\n') relevant = [] for line in lines: if any(keyword in line.lower() for keyword in ['certificate', 'acme', 'tls', 'ssl', 'obtain', 'error']): relevant.append(line) return relevant[-10:] # Return last 10 relevant lines