/
/
/
1"""SSL helpers for the webserver controller."""
2
3from __future__ import annotations
4
5import asyncio
6import contextlib
7import logging
8import ssl
9import subprocess
10import tempfile
11from dataclasses import dataclass
12from pathlib import Path
13
14import aiofiles
15
16LOGGER = logging.getLogger(__name__)
17
18
19@dataclass
20class SSLCertificateInfo:
21 """Information about an SSL certificate."""
22
23 is_valid: bool
24 key_type: str # "RSA", "ECDSA", or "Unknown"
25 subject: str
26 expiry: str
27 is_expired: bool
28 is_expiring_soon: bool # Within 30 days
29 error_message: str | None = None
30
31
32async def get_ssl_content(value: str) -> str:
33 """
34 Get SSL content from either a file path or the raw PEM content.
35
36 :param value: Either an absolute file path or the raw PEM content.
37 :return: The PEM content as a string.
38 :raises FileNotFoundError: If the file path doesn't exist.
39 :raises ValueError: If the path is not a file.
40 """
41 value = value.strip()
42 # Check if this looks like a file path (absolute path starting with /)
43 # PEM content always starts with "-----BEGIN"
44 if value.startswith("/") and not value.startswith("-----BEGIN"):
45 # This looks like a file path
46 path = Path(value)
47 if not path.exists():
48 raise FileNotFoundError(f"SSL file not found: {value}")
49 if not path.is_file():
50 raise ValueError(f"SSL path is not a file: {value}")
51 async with aiofiles.open(path) as f:
52 content: str = await f.read()
53 return content
54 # Otherwise, treat as raw PEM content
55 return value
56
57
58def _run_openssl_command(args: list[str]) -> subprocess.CompletedProcess[str]:
59 """
60 Run an openssl command synchronously.
61
62 :param args: List of arguments for the openssl command (excluding 'openssl').
63 :return: CompletedProcess result.
64 """
65 return subprocess.run( # noqa: S603
66 ["openssl", *args], # noqa: S607
67 capture_output=True,
68 text=True,
69 timeout=10,
70 check=False,
71 )
72
73
74async def create_server_ssl_context(
75 certificate: str,
76 private_key: str,
77 logger: logging.Logger | None = None,
78) -> ssl.SSLContext | None:
79 """
80 Create an SSL context for a server from certificate and private key.
81
82 :param certificate: The SSL certificate (file path or PEM content).
83 :param private_key: The SSL private key (file path or PEM content).
84 :param logger: Optional logger for error messages.
85 :return: SSL context if successful, None otherwise.
86 """
87 log = logger or LOGGER
88 if not certificate or not private_key:
89 log.error(
90 "SSL is enabled but certificate or private key is missing. "
91 "Server will start without SSL."
92 )
93 return None
94
95 cert_path = None
96 key_path = None
97 try:
98 # Load certificate and key content (supports both file paths and raw content)
99 cert_content = await get_ssl_content(certificate)
100 key_content = await get_ssl_content(private_key)
101
102 # Create SSL context
103 ssl_context = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)
104
105 # Write certificate and key to temporary files
106 # This is necessary because ssl.SSLContext.load_cert_chain requires file paths
107 with tempfile.NamedTemporaryFile(mode="w", suffix=".pem", delete=False) as cert_file:
108 cert_file.write(cert_content)
109 cert_path = cert_file.name
110
111 with tempfile.NamedTemporaryFile(mode="w", suffix=".pem", delete=False) as key_file:
112 key_file.write(key_content)
113 key_path = key_file.name
114
115 # Load certificate and private key
116 ssl_context.load_cert_chain(cert_path, key_path)
117 log.info("SSL/TLS enabled for server")
118 return ssl_context
119
120 except Exception:
121 log.exception("Failed to create SSL context. Server will start without SSL.")
122 return None
123 finally:
124 # Clean up temporary files
125 if cert_path:
126 with contextlib.suppress(Exception):
127 Path(cert_path).unlink()
128 if key_path:
129 with contextlib.suppress(Exception):
130 Path(key_path).unlink()
131
132
133async def verify_ssl_certificate(certificate: str, private_key: str) -> SSLCertificateInfo:
134 """
135 Verify SSL certificate and private key are valid and match.
136
137 :param certificate: The SSL certificate (file path or PEM content).
138 :param private_key: The SSL private key (file path or PEM content).
139 :return: SSLCertificateInfo with verification results.
140 """
141 if not certificate or not private_key:
142 return SSLCertificateInfo(
143 is_valid=False,
144 key_type="Unknown",
145 subject="",
146 expiry="",
147 is_expired=False,
148 is_expiring_soon=False,
149 error_message="Both certificate and private key are required.",
150 )
151
152 # Load certificate and key content
153 try:
154 cert_content = await get_ssl_content(certificate)
155 except FileNotFoundError as e:
156 return SSLCertificateInfo(
157 is_valid=False,
158 key_type="Unknown",
159 subject="",
160 expiry="",
161 is_expired=False,
162 is_expiring_soon=False,
163 error_message=f"Certificate file not found: {e}",
164 )
165 except Exception as e:
166 return SSLCertificateInfo(
167 is_valid=False,
168 key_type="Unknown",
169 subject="",
170 expiry="",
171 is_expired=False,
172 is_expiring_soon=False,
173 error_message=f"Error loading certificate: {e}",
174 )
175
176 try:
177 key_content = await get_ssl_content(private_key)
178 except FileNotFoundError as e:
179 return SSLCertificateInfo(
180 is_valid=False,
181 key_type="Unknown",
182 subject="",
183 expiry="",
184 is_expired=False,
185 is_expiring_soon=False,
186 error_message=f"Private key file not found: {e}",
187 )
188 except Exception as e:
189 return SSLCertificateInfo(
190 is_valid=False,
191 key_type="Unknown",
192 subject="",
193 expiry="",
194 is_expired=False,
195 is_expiring_soon=False,
196 error_message=f"Error loading private key: {e}",
197 )
198
199 # Verify with temp files
200 try:
201 return await _verify_ssl_with_temp_files(cert_content, key_content)
202 except ssl.SSLError as e:
203 return SSLCertificateInfo(
204 is_valid=False,
205 key_type="Unknown",
206 subject="",
207 expiry="",
208 is_expired=False,
209 is_expiring_soon=False,
210 error_message=_format_ssl_error(e),
211 )
212 except Exception as e:
213 return SSLCertificateInfo(
214 is_valid=False,
215 key_type="Unknown",
216 subject="",
217 expiry="",
218 is_expired=False,
219 is_expiring_soon=False,
220 error_message=f"Verification failed: {e}",
221 )
222
223
224async def _verify_ssl_with_temp_files(cert_content: str, key_content: str) -> SSLCertificateInfo:
225 """
226 Verify SSL using temporary files.
227
228 :param cert_content: Certificate PEM content.
229 :param key_content: Private key PEM content.
230 :return: SSLCertificateInfo with verification results.
231 """
232 cert_path = None
233 key_path = None
234 try:
235 with tempfile.NamedTemporaryFile(mode="w", suffix=".pem", delete=False) as cert_file:
236 cert_file.write(cert_content)
237 cert_path = cert_file.name
238
239 with tempfile.NamedTemporaryFile(mode="w", suffix=".pem", delete=False) as key_file:
240 key_file.write(key_content)
241 key_path = key_file.name
242
243 # Test loading into SSL context
244 test_ctx = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)
245 test_ctx.load_cert_chain(cert_path, key_path)
246
247 # Get certificate details using openssl
248 return await _get_certificate_details(cert_path)
249 finally:
250 # Clean up temp files
251 if cert_path:
252 with contextlib.suppress(Exception):
253 Path(cert_path).unlink()
254 if key_path:
255 with contextlib.suppress(Exception):
256 Path(key_path).unlink()
257
258
259async def _get_certificate_details(cert_path: str) -> SSLCertificateInfo:
260 """
261 Get certificate details using openssl.
262
263 :param cert_path: Path to the certificate file.
264 :return: SSLCertificateInfo with certificate details.
265 """
266 # Get certificate info
267 result = await asyncio.to_thread(
268 _run_openssl_command,
269 ["x509", "-in", cert_path, "-noout", "-subject", "-dates", "-issuer"],
270 )
271
272 if result.returncode != 0:
273 return SSLCertificateInfo(
274 is_valid=True,
275 key_type="Unknown",
276 subject="",
277 expiry="",
278 is_expired=False,
279 is_expiring_soon=False,
280 )
281
282 # Parse certificate info
283 expiry = ""
284 subject = ""
285 for line in result.stdout.strip().split("\n"):
286 if line.startswith("notAfter="):
287 expiry = line.replace("notAfter=", "")
288 elif line.startswith("subject="):
289 subject = line.replace("subject=", "")
290
291 # Check expiry status
292 expiry_check = await asyncio.to_thread(
293 _run_openssl_command,
294 ["x509", "-in", cert_path, "-noout", "-checkend", "0"],
295 )
296 is_expired = expiry_check.returncode != 0
297
298 expiring_soon_check = await asyncio.to_thread(
299 _run_openssl_command,
300 ["x509", "-in", cert_path, "-noout", "-checkend", str(30 * 24 * 60 * 60)],
301 )
302 is_expiring_soon = expiring_soon_check.returncode != 0
303
304 # Detect key type
305 key_type_result = await asyncio.to_thread(
306 _run_openssl_command,
307 ["x509", "-in", cert_path, "-noout", "-text"],
308 )
309 key_type = "Unknown"
310 if "rsaEncryption" in key_type_result.stdout:
311 key_type = "RSA"
312 elif "id-ecPublicKey" in key_type_result.stdout:
313 key_type = "ECDSA"
314
315 return SSLCertificateInfo(
316 is_valid=True,
317 key_type=key_type,
318 subject=subject,
319 expiry=expiry,
320 is_expired=is_expired,
321 is_expiring_soon=is_expiring_soon,
322 )
323
324
325def _format_ssl_error(e: ssl.SSLError) -> str:
326 """
327 Format an SSL error into a user-friendly message.
328
329 :param e: The SSL error.
330 :return: User-friendly error message.
331 """
332 error_msg = str(e)
333 if "PEM lib" in error_msg:
334 return (
335 "Invalid certificate or key format. "
336 "Make sure both are valid PEM format and the key is not encrypted."
337 )
338 if "key values mismatch" in error_msg.lower():
339 return (
340 "Certificate and private key do not match. "
341 "Please verify you're using the correct key for this certificate."
342 )
343 return f"SSL Error: {error_msg}"
344
345
346def format_certificate_info(info: SSLCertificateInfo) -> str:
347 """
348 Format SSLCertificateInfo into a human-readable string.
349
350 :param info: The certificate info to format.
351 :return: Human-readable string.
352 """
353 if not info.is_valid:
354 return f"Error: {info.error_message}"
355
356 status = "VALID"
357 warning = ""
358 if info.is_expired:
359 status = "EXPIRED"
360 warning = " (Certificate has expired!)"
361 elif info.is_expiring_soon:
362 status = "EXPIRING SOON"
363 warning = " (Certificate expires within 30 days)"
364
365 lines = [f"Certificate verification: {status}{warning}", f"Key type: {info.key_type}"]
366 if info.subject:
367 lines.append(f"Subject: {info.subject}")
368 if info.expiry:
369 lines.append(f"Expires: {info.expiry}")
370
371 return "\n".join(lines)
372