···
1
-
"""Automated SSL certificate management using ACME protocol."""
9
-
from datetime import datetime, timezone
10
-
from pathlib import Path
11
-
from typing import Optional, Tuple
14
-
from acme import challenges, client, errors, messages
15
-
from cryptography import x509
16
-
from cryptography.hazmat.backends import default_backend
17
-
from cryptography.hazmat.primitives import hashes, serialization
18
-
from cryptography.hazmat.primitives.asymmetric import rsa
19
-
from cryptography.x509.oid import NameOID
20
-
from fastapi import FastAPI
21
-
from fastapi.responses import PlainTextResponse
23
-
import josepy as jose
25
-
logger = structlog.get_logger()
27
-
LETSENCRYPT_DIRECTORY_URL = "https://acme-v02.api.letsencrypt.org/directory"
28
-
LETSENCRYPT_STAGING_URL = "https://acme-staging-v02.api.letsencrypt.org/directory"
31
-
class CertificateManager:
32
-
"""Manages SSL certificates using ACME protocol."""
39
-
staging: bool = False,
42
-
"""Initialize certificate manager.
45
-
domain: Domain name for the certificate
46
-
email: Email for Let's Encrypt account
47
-
cert_dir: Directory to store certificates
48
-
staging: Use Let's Encrypt staging server
49
-
port: Port for HTTP-01 challenge server
51
-
self.domain = domain
53
-
self.cert_dir = Path(cert_dir)
54
-
self.cert_dir.mkdir(parents=True, exist_ok=True)
55
-
self.staging = staging
56
-
self.challenge_port = port
58
-
self.directory_url = LETSENCRYPT_STAGING_URL if staging else LETSENCRYPT_DIRECTORY_URL
59
-
self.account_key_path = self.cert_dir / "account_key.pem"
60
-
self.cert_path = self.cert_dir / f"{domain}_cert.pem"
61
-
self.key_path = self.cert_dir / f"{domain}_key.pem"
62
-
self.fullchain_path = self.cert_dir / f"{domain}_fullchain.pem"
64
-
# For HTTP-01 challenge
65
-
self.challenge_tokens = {}
66
-
self.challenge_server = None
67
-
self.challenge_thread = None
69
-
def _generate_private_key(self) -> rsa.RSAPrivateKey:
70
-
"""Generate a new RSA private key."""
71
-
return rsa.generate_private_key(
72
-
public_exponent=65537,
74
-
backend=default_backend()
77
-
def _get_or_create_account_key(self) -> jose.JWK:
78
-
"""Get existing account key or create a new one."""
79
-
if self.account_key_path.exists():
80
-
with open(self.account_key_path, 'rb') as f:
82
-
private_key = serialization.load_pem_private_key(
83
-
key_data, password=None, backend=default_backend()
86
-
private_key = self._generate_private_key()
87
-
key_pem = private_key.private_bytes(
88
-
encoding=serialization.Encoding.PEM,
89
-
format=serialization.PrivateFormat.TraditionalOpenSSL,
90
-
encryption_algorithm=serialization.NoEncryption()
92
-
with open(self.account_key_path, 'wb') as f:
94
-
logger.info("Created new account key", path=str(self.account_key_path))
96
-
return jose.JWK.load(private_key.private_bytes(
97
-
encoding=serialization.Encoding.PEM,
98
-
format=serialization.PrivateFormat.TraditionalOpenSSL,
99
-
encryption_algorithm=serialization.NoEncryption()
102
-
def _create_csr(self, private_key: rsa.RSAPrivateKey) -> bytes:
103
-
"""Create a Certificate Signing Request."""
104
-
csr = x509.CertificateSigningRequestBuilder().subject_name(
106
-
x509.NameAttribute(NameOID.COMMON_NAME, self.domain),
108
-
).sign(private_key, hashes.SHA256(), backend=default_backend())
110
-
return csr.public_bytes(serialization.Encoding.DER)
112
-
def _start_challenge_server(self):
113
-
"""Start HTTP server for ACME challenges."""
116
-
@app.get("/.well-known/acme-challenge/{token}")
117
-
async def acme_challenge(token: str):
118
-
"""Serve ACME challenge responses."""
119
-
if token in self.challenge_tokens:
120
-
logger.info("Serving ACME challenge", token=token)
121
-
return PlainTextResponse(self.challenge_tokens[token])
122
-
logger.warning("Unknown ACME challenge token", token=token)
123
-
return PlainTextResponse("Not found", status_code=404)
126
-
"""Run the challenge server in a thread."""
131
-
port=self.challenge_port,
134
-
except Exception as e:
135
-
logger.error("Challenge server error", error=str(e))
137
-
self.challenge_thread = threading.Thread(target=run_server, daemon=True)
138
-
self.challenge_thread.start()
140
-
# Give the server time to start
142
-
logger.info("Started ACME challenge server", port=self.challenge_port)
144
-
def _stop_challenge_server(self):
145
-
"""Stop the challenge server."""
146
-
if self.challenge_thread and self.challenge_thread.is_alive():
147
-
# The thread is daemon, so it will stop when the main process exits
148
-
logger.info("Challenge server will stop with main process")
150
-
def _perform_http01_challenge(
152
-
acme_client: client.ClientV2,
153
-
authz: messages.Authorization
155
-
"""Perform HTTP-01 challenge."""
156
-
# Find HTTP-01 challenge
157
-
http_challenge = None
158
-
for challenge in authz.body.challenges:
159
-
if isinstance(challenge.chall, challenges.HTTP01):
160
-
http_challenge = challenge
163
-
if not http_challenge:
164
-
logger.error("No HTTP-01 challenge found")
167
-
# Prepare challenge response
168
-
response, validation = http_challenge.chall.response_and_validation(
169
-
acme_client.net.key
172
-
# Store challenge token and response
173
-
self.challenge_tokens[http_challenge.chall.token.decode('utf-8')] = validation
176
-
"Prepared HTTP-01 challenge",
177
-
token=http_challenge.chall.token.decode('utf-8'),
181
-
# Notify ACME server that we're ready
182
-
acme_client.answer_challenge(http_challenge, response)
184
-
# Wait for challenge validation
186
-
for attempt in range(max_attempts):
189
-
authz, _ = acme_client.poll(authz)
190
-
if authz.body.status == messages.STATUS_VALID:
191
-
logger.info("Challenge validated successfully")
193
-
elif authz.body.status == messages.STATUS_INVALID:
194
-
logger.error("Challenge validation failed")
196
-
except errors.TimeoutError:
197
-
if attempt == max_attempts - 1:
198
-
logger.error("Challenge validation timeout")
204
-
def needs_renewal(self) -> bool:
205
-
"""Check if certificate needs renewal."""
206
-
if not self.cert_path.exists():
210
-
with open(self.cert_path, 'rb') as f:
211
-
cert_data = f.read()
212
-
cert = x509.load_pem_x509_certificate(cert_data, default_backend())
214
-
# Renew if less than 30 days remaining
215
-
days_remaining = (cert.not_valid_after_utc -
216
-
datetime.now(timezone.utc)).days
218
-
if days_remaining < 30:
219
-
logger.info("Certificate needs renewal", days_remaining=days_remaining)
222
-
logger.info("Certificate still valid", days_remaining=days_remaining)
225
-
except Exception as e:
226
-
logger.error("Error checking certificate", error=str(e))
229
-
def obtain_certificate(self) -> Tuple[Path, Path, Path]:
230
-
"""Obtain or renew SSL certificate.
233
-
Tuple of (cert_path, key_path, fullchain_path)
235
-
if not self.needs_renewal():
236
-
logger.info("Certificate is still valid, skipping renewal")
237
-
return self.cert_path, self.key_path, self.fullchain_path
240
-
"Obtaining SSL certificate",
241
-
domain=self.domain,
242
-
staging=self.staging
246
-
# Start challenge server
247
-
self._start_challenge_server()
249
-
# Get or create account key
250
-
account_key = self._get_or_create_account_key()
252
-
# Create ACME client
253
-
net = client.ClientNetwork(account_key)
254
-
directory = messages.Directory.from_json(
255
-
net.get(self.directory_url).json()
257
-
acme_client = client.ClientV2(directory, net=net)
259
-
# Register or get existing account
261
-
account = acme_client.new_account(
262
-
messages.NewRegistration.from_data(
264
-
terms_of_service_agreed=True
267
-
logger.info("Created new ACME account")
268
-
except errors.ConflictError:
269
-
# Account already exists
270
-
account = acme_client.query_registration(
271
-
messages.Registration(key=account_key.public_key())
273
-
logger.info("Using existing ACME account")
275
-
# Generate certificate private key
276
-
cert_key = self._generate_private_key()
279
-
csr = self._create_csr(cert_key)
281
-
# Request certificate
282
-
order = acme_client.new_order(csr)
284
-
# Complete challenges
285
-
for authz in order.authorizations:
286
-
if not self._perform_http01_challenge(acme_client, authz):
287
-
raise Exception(f"Failed to complete challenge for {authz.body.identifier.value}")
290
-
order = acme_client.poll_and_finalize(order)
292
-
if order.fullchain_pem:
293
-
# Save certificate and key
294
-
with open(self.cert_path, 'w') as f:
295
-
f.write(order.fullchain_pem.split('\n\n')[0] + '\n')
297
-
with open(self.fullchain_path, 'w') as f:
298
-
f.write(order.fullchain_pem)
300
-
key_pem = cert_key.private_bytes(
301
-
encoding=serialization.Encoding.PEM,
302
-
format=serialization.PrivateFormat.TraditionalOpenSSL,
303
-
encryption_algorithm=serialization.NoEncryption()
306
-
with open(self.key_path, 'w') as f:
309
-
# Set proper permissions
310
-
os.chmod(self.key_path, 0o600)
311
-
os.chmod(self.cert_path, 0o644)
312
-
os.chmod(self.fullchain_path, 0o644)
315
-
"Certificate obtained successfully",
316
-
cert_path=str(self.cert_path),
317
-
key_path=str(self.key_path),
318
-
fullchain_path=str(self.fullchain_path)
321
-
return self.cert_path, self.key_path, self.fullchain_path
323
-
raise Exception("Failed to obtain certificate")
325
-
except Exception as e:
326
-
logger.error("Failed to obtain certificate", error=str(e))
329
-
# Clean up challenge tokens and stop server
330
-
self.challenge_tokens.clear()
331
-
self._stop_challenge_server()
333
-
def setup_auto_renewal(self, check_interval: int = 86400):
334
-
"""Setup automatic certificate renewal.
337
-
check_interval: Interval in seconds to check for renewal (default: 24 hours)
339
-
def renewal_loop():
340
-
"""Background renewal loop."""
343
-
if self.needs_renewal():
344
-
logger.info("Certificate renewal needed")
345
-
self.obtain_certificate()
347
-
logger.debug("Certificate renewal not needed")
348
-
except Exception as e:
349
-
logger.error("Certificate renewal check failed", error=str(e))
351
-
time.sleep(check_interval)
353
-
renewal_thread = threading.Thread(target=renewal_loop, daemon=True)
354
-
renewal_thread.start()
355
-
logger.info("Started automatic certificate renewal", interval_hours=check_interval/3600)
anil.recoil.org