Coverage for frappe_manager / commands / ssl / external_helpers.py: 8%

346 statements  

« prev     ^ index     » next       coverage.py v7.13.5, created at 2026-07-02 18:13 +0530

1"""Helper functions for external domain SSL certificate operations.""" 

2 

3import re 

4import subprocess 

5from datetime import datetime, timezone 

6 

7import typer 

8from rich.table import Table 

9 

10from frappe_manager import CLI_BENCHES_DIRECTORY, SSL_RENEW_BEFORE_DAYS 

11from frappe_manager.logger import ContextualLogger, log 

12from frappe_manager.logger.context import LoggerContext 

13from frappe_manager.output_manager import spinner, temporary_stop 

14from frappe_manager.output_manager.silent_output import SilentOutputHandler 

15from frappe_manager.site_manager.bench_service import BenchService 

16from frappe_manager.site_manager.site import Bench 

17from frappe_manager.ssl_manager import LETSENCRYPT_PREFERRED_CHALLENGE, SUPPORTED_SSL_TYPES 

18from frappe_manager.ssl_manager.certificate_link_manager import CertificateLinkManager 

19from frappe_manager.ssl_manager.external_domain_manager import ExternalDomainConfig, ExternalDomainConfigManager 

20from frappe_manager.ssl_manager.letsencrypt_certificate import CustomDomainCertificate, LetsencryptSSLCertificate 

21from frappe_manager.ssl_manager.service_factory import create_certificate_service 

22from frappe_manager.ssl_manager.ssl_certificate_manager import SSLCertificateManager 

23from frappe_manager.ssl_manager.standalone_nginx_config_manager import StandaloneNginxConfigManager 

24from frappe_manager.ssl_manager.storage_config import SSLStorageConfig 

25from frappe_manager.utils.helpers import get_certificate_expiry_date 

26 

27from .helpers import get_output_handler 

28 

29 

30def _add_external_certificate( 

31 ctx: typer.Context, 

32 domain: str, 

33 challenge: LETSENCRYPT_PREFERRED_CHALLENGE, 

34 cname: str | None, 

35 dry_run: bool, 

36 skip_dns_check: bool = False, 

37 wait_for_dns: bool = False, 

38): 

39 """Add SSL certificate for external (non-bench) domain.""" 

40 

41 services_manager = ctx.obj["services"] 

42 context = LoggerContext(operation="ssl-add-external") 

43 logger = ContextualLogger(log.get_logger(), context) 

44 output = get_output_handler(ctx, context=context) 

45 

46 external_config_path = services_manager.path / "nginx-proxy" / "external_domains.toml" 

47 external_manager = ExternalDomainConfigManager(external_config_path) 

48 

49 if external_manager.domain_exists(domain): 

50 output.display_error(f"Certificate already exists for external domain '{domain}'") 

51 output.print("To update certificate:", emoji_code="") 

52 output.print(f" 1. Remove existing: fm ssl remove --standalone {domain}", emoji_code="") 

53 output.print(f" 2. Add new: fm ssl add --standalone {domain}", emoji_code="") 

54 raise typer.Exit(1) 

55 

56 if cname and challenge != LETSENCRYPT_PREFERRED_CHALLENGE.dns01: 

57 output.display_error("CNAME delegation (--cname) requires DNS-01 challenge") 

58 with temporary_stop(output): 

59 typer.echo(ctx.get_help()) 

60 raise typer.Exit(1) 

61 

62 output.change_head(f"Adding SSL certificate for {domain} (standalone mode)") 

63 

64 try: 

65 if cname: 

66 cert = CustomDomainCertificate( 

67 domain=domain, 

68 ssl_type=SUPPORTED_SSL_TYPES.le, 

69 api_token=None, 

70 api_key=None, 

71 challenge_type=challenge, 

72 delegation_cname=cname, 

73 ) 

74 output.print(f"Using CNAME delegation: {cname}", emoji_code=":information:") 

75 

76 # Validate CNAME before proceeding (unless skipped) 

77 if not skip_dns_check: 

78 from frappe_manager.ssl_manager.dns_validator import DNSValidator 

79 

80 output.change_head(f"Validating DNS configuration for {domain}") 

81 validator = DNSValidator(output_handler=output) 

82 

83 # If wait_for_dns is True, poll for propagation 

84 if wait_for_dns: 

85 propagation = validator.wait_for_cname_propagation( 

86 domain=domain, 

87 challenge_alias=cname, 

88 timeout=300, # 5 minutes 

89 check_interval=30, # 30 seconds 

90 ) 

91 

92 if not propagation.propagated: 

93 output.display_error("DNS propagation timeout") 

94 output.print("", emoji_code="") 

95 output.print("CNAME record did not propagate within 5 minutes.", emoji_code="") 

96 output.print("Expected CNAME:", emoji_code="") 

97 output.print(f" _acme-challenge.{domain} → _acme-challenge.{cname}", emoji_code="") 

98 output.print("", emoji_code="") 

99 output.print("Please verify your DNS configuration and try again.", emoji_code="") 

100 raise typer.Exit(1) 

101 

102 output.print(f"{propagation.message}", emoji_code=":white_check_mark:") 

103 else: 

104 # Single validation check (no waiting) 

105 validation = validator.validate_cname_for_acme(domain, cname) 

106 

107 if not validation.valid: 

108 output.display_error("DNS validation failed") 

109 output.print("", emoji_code="") 

110 output.print(f"Domain: {domain}", emoji_code="") 

111 output.print("Expected CNAME:", emoji_code="") 

112 output.print(f" _acme-challenge.{domain} → _acme-challenge.{cname}", emoji_code="") 

113 

114 if validation.actual_value: 

115 output.print("", emoji_code="") 

116 output.print("Current CNAME:", emoji_code="") 

117 output.print(f" _acme-challenge.{domain}{validation.actual_value}", emoji_code="") 

118 output.print("", emoji_code="") 

119 output.print("The CNAME record exists but points to the wrong target.", emoji_code="") 

120 else: 

121 output.print("", emoji_code="") 

122 output.print("CNAME record not found.", emoji_code="") 

123 

124 output.print("", emoji_code="") 

125 output.print("Please update your DNS to match the expected value above.", emoji_code="") 

126 output.print("DNS changes may take up to 5 minutes to propagate.", emoji_code="") 

127 output.print("", emoji_code="") 

128 output.print("To skip this check, use: --skip-dns-check", emoji_code="") 

129 output.print("To wait for propagation, use: --wait-for-dns", emoji_code="") 

130 raise typer.Exit(1) 

131 

132 output.print("CNAME record verified", emoji_code=":white_check_mark:") 

133 else: 

134 cert = LetsencryptSSLCertificate( 

135 domain=domain, 

136 ssl_type=SUPPORTED_SSL_TYPES.le, 

137 api_token=None, 

138 api_key=None, 

139 challenge_type=challenge, 

140 ) 

141 

142 if not skip_dns_check and challenge == LETSENCRYPT_PREFERRED_CHALLENGE.http01: 

143 from frappe_manager.ssl_manager.dns_validator import DNSValidator 

144 

145 output.change_head(f"Checking DNS configuration for {domain}") 

146 validator = DNSValidator(output_handler=output) 

147 validation = validator.validate_a_record(domain) 

148 

149 if validation.valid: 

150 output.print(f"Domain resolves to {validation.actual_value}", emoji_code=":white_check_mark:") 

151 else: 

152 output.warning(f"Domain {domain} doesn't have an A record") 

153 output.print("HTTP-01 challenge may fail if DNS is not configured correctly.", emoji_code="") 

154 output.print(f"Make sure {domain} points to this server's IP address.", emoji_code="") 

155 

156 global_proxy_storage = services_manager.proxy_storage 

157 

158 storage_config = SSLStorageConfig( 

159 ssl_dir=global_proxy_storage.dirs.ssl.host, 

160 ssl_dir_container=global_proxy_storage.dirs.ssl.container, 

161 certs_dir=global_proxy_storage.dirs.certs.host, 

162 certs_dir_container=global_proxy_storage.dirs.certs.container, 

163 vhostd_dir=global_proxy_storage.dirs.vhostd.host, 

164 webroot_dir=global_proxy_storage.dirs.html.host, # For HTTP-01 challenge 

165 ) 

166 

167 link_manager = CertificateLinkManager(storage_config) 

168 nginx_controller = services_manager.nginx_controller 

169 

170 standalone_nginx = StandaloneNginxConfigManager( 

171 conf_dir=global_proxy_storage.dirs.confd.host, 

172 webroot_dir_container=global_proxy_storage.dirs.html.container, 

173 certs_dir_container=global_proxy_storage.dirs.certs.container, 

174 ) 

175 

176 # Step 1: Create HTTP-only nginx config (for ACME challenge) 

177 output.change_head(f"Setting up nginx configuration for {domain}") 

178 standalone_nginx.create_http_config(domain) 

179 output.print("Created HTTP configuration for ACME challenge", emoji_code=":white_check_mark:") 

180 

181 # Step 2: Reload nginx to apply the config 

182 output.change_head("Reloading nginx to apply configuration") 

183 nginx_controller.reload() 

184 output.print("Nginx reloaded successfully", emoji_code=":white_check_mark:") 

185 

186 def certificate_service_factory(cert, storage_cfg, output_handler): 

187 return create_certificate_service(logger, cert, storage_cfg, output_handler) 

188 

189 cert_manager = SSLCertificateManager( 

190 logger=logger, 

191 certificates=[], # Start with empty list, we'll add the cert next 

192 service_factory=certificate_service_factory, 

193 link_manager=link_manager, 

194 nginx_controller=nginx_controller, 

195 storage_config=storage_config, 

196 output_handler=output, 

197 config_save_callback=None, # No bench config callback 

198 ) 

199 

200 # Step 3: Generate certificate (HTTP-01 challenge will now work) 

201 try: 

202 with spinner(output, f"Generating SSL certificate for {domain}"): 

203 cert_manager.add_certificate(cert, dry_run=dry_run) 

204 except Exception as cert_error: 

205 # Certificate generation failed - clean up nginx config 

206 output.change_head("Cleaning up after certificate generation failure") 

207 try: 

208 standalone_nginx.remove_config(domain) 

209 nginx_controller.reload() 

210 output.print("Cleaned up nginx configuration", emoji_code=":white_check_mark:") 

211 except Exception as cleanup_error: 

212 output.debug(f"Failed to clean up nginx config: {cleanup_error}") 

213 # Re-raise the original certificate error 

214 raise cert_error 

215 

216 # Step 4: Update nginx config to enable HTTPS 

217 output.change_head("Enabling HTTPS for {domain}") 

218 try: 

219 standalone_nginx.create_https_config(domain) 

220 output.print("Created HTTPS configuration", emoji_code=":white_check_mark:") 

221 

222 # Step 5: Reload nginx again to enable HTTPS 

223 nginx_controller.reload() 

224 output.print("Nginx reloaded with HTTPS enabled", emoji_code=":white_check_mark:") 

225 except Exception as post_cert_error: 

226 # HTTPS config or reload failed — cert was already issued; clean up to avoid orphan 

227 output.change_head("Cleaning up after HTTPS configuration failure") 

228 try: 

229 cert_manager.remove_certificate_by_domain(domain) 

230 standalone_nginx.remove_config(domain) 

231 nginx_controller.reload() 

232 output.print("Cleaned up nginx configuration and certificate", emoji_code=":white_check_mark:") 

233 except Exception as cleanup_error: 

234 output.debug(f"Failed to clean up after HTTPS config failure: {cleanup_error}") 

235 raise post_cert_error 

236 

237 if not dry_run: 

238 # Save to external domains config 

239 external_manager.add_domain( 

240 ExternalDomainConfig( 

241 domain=domain, 

242 ssl_type="letsencrypt", 

243 # Email removed - Let's Encrypt discontinued notifications (June 2025) 

244 added_at=datetime.now().isoformat(), 

245 challenge_type=challenge.lower(), 

246 delegation_cname=cname, 

247 acme_client="acme.sh", 

248 ), 

249 ) 

250 

251 output.print(f"SSL certificate added for {domain}", emoji_code=":white_check_mark:") 

252 output.print("Certificate configured successfully", emoji_code=":zap:") 

253 output.print("", emoji_code="") 

254 output.print("[bold cyan]To use this certificate in your Docker project:[/bold cyan]", emoji_code="") 

255 output.print("", emoji_code="") 

256 output.print("1. Add to your docker-compose.yml:", emoji_code="") 

257 output.print("", emoji_code="") 

258 output.print(" services:", emoji_code="") 

259 output.print(" your-app:", emoji_code="") 

260 output.print(" environment:", emoji_code="") 

261 output.print(f" VIRTUAL_HOST: {domain}", emoji_code="") 

262 output.print(" VIRTUAL_PORT: 80 # Your app's port", emoji_code="") 

263 output.print(" networks:", emoji_code="") 

264 output.print(" - fm-global-frontend-network", emoji_code="") 

265 output.print("", emoji_code="") 

266 output.print(" networks:", emoji_code="") 

267 output.print(" fm-global-frontend-network:", emoji_code="") 

268 output.print(" external: true", emoji_code="") 

269 output.print("", emoji_code="") 

270 output.print("2. Start your project:", emoji_code="") 

271 output.print(" docker compose up -d", emoji_code="") 

272 output.print("", emoji_code="") 

273 output.print(f"3. Access your app at: https://{domain}", emoji_code="") 

274 

275 except ValueError as e: 

276 output.display_error(f"Failed to add certificate: {e}") 

277 raise typer.Exit(1) 

278 except Exception as e: 

279 output.display_error(f"Failed to add certificate: {e}") 

280 raise typer.Exit(1) 

281 

282 

283def _remove_external_certificate(ctx: typer.Context, domain: str, yes: bool): 

284 services_manager = ctx.obj["services"] 

285 context = LoggerContext(operation="ssl-remove-external") 

286 logger = ContextualLogger(log.get_logger(), context) 

287 output = get_output_handler(ctx, context=context) 

288 

289 external_config_path = services_manager.path / "nginx-proxy" / "external_domains.toml" 

290 external_manager = ExternalDomainConfigManager(external_config_path) 

291 

292 if not external_manager.domain_exists(domain): 

293 output.display_error(f"Certificate does not exist for external domain '{domain}'") 

294 raise typer.Exit(1) 

295 

296 domain_config = external_manager.get_domain(domain) 

297 output.change_head(f"Removing SSL certificate for {domain}") 

298 

299 if not yes: 

300 with temporary_stop(output): 

301 choice = output.prompt_ask( 

302 prompt=f"Remove SSL certificate for {domain}?", 

303 choices=["yes", "no"], 

304 default="no", 

305 required_flag="--yes or -y", 

306 ) 

307 if choice != "yes": 

308 output.print("Cancelled.", emoji_code=":x:") 

309 raise typer.Exit(0) 

310 

311 output.change_head(f"Removing SSL certificate for {domain}") 

312 

313 try: 

314 cert_config = external_manager.get_domain(domain) 

315 if not cert_config: 

316 raise ValueError(f"Domain config not found for {domain}") 

317 

318 cert = external_manager.to_ssl_certificate(domain) 

319 if not cert: 

320 raise ValueError(f"Could not create certificate object for {domain}") 

321 

322 global_proxy_storage = services_manager.proxy_storage 

323 

324 storage_config = SSLStorageConfig( 

325 ssl_dir=global_proxy_storage.dirs.ssl.host, 

326 ssl_dir_container=global_proxy_storage.dirs.ssl.container, 

327 certs_dir=global_proxy_storage.dirs.certs.host, 

328 certs_dir_container=global_proxy_storage.dirs.certs.container, 

329 vhostd_dir=global_proxy_storage.dirs.vhostd.host, 

330 webroot_dir=global_proxy_storage.dirs.html.host, 

331 ) 

332 

333 link_manager = CertificateLinkManager(storage_config) 

334 nginx_controller = services_manager.nginx_controller 

335 

336 standalone_nginx = StandaloneNginxConfigManager( 

337 conf_dir=global_proxy_storage.dirs.confd.host, 

338 webroot_dir_container=global_proxy_storage.dirs.html.container, 

339 certs_dir_container=global_proxy_storage.dirs.certs.container, 

340 ) 

341 

342 def certificate_service_factory(cert, storage_cfg, output_handler): 

343 return create_certificate_service(logger, cert, storage_cfg, output_handler) 

344 

345 cert_manager = SSLCertificateManager( 

346 logger=logger, 

347 certificates=[cert], 

348 service_factory=certificate_service_factory, 

349 link_manager=link_manager, 

350 nginx_controller=nginx_controller, 

351 storage_config=storage_config, 

352 output_handler=output, 

353 config_save_callback=None, 

354 ) 

355 

356 # Remove certificate (removes symlinks, vhost.d, and cert files) 

357 with spinner(output, f"Removing SSL certificate for {domain}"): 

358 cert_manager.remove_certificate_by_domain(domain) 

359 

360 # Remove standalone nginx configuration 

361 output.change_head(f"Removing nginx configuration for {domain}") 

362 standalone_nginx.remove_config(domain) 

363 output.print("Removed nginx configuration", emoji_code=":white_check_mark:") 

364 

365 # Reload nginx 

366 nginx_controller.reload() 

367 output.print("Nginx reloaded", emoji_code=":white_check_mark:") 

368 

369 # Remove from external domains config 

370 external_manager.remove_domain(domain) 

371 

372 output.print(f"SSL certificate removed for {domain}", emoji_code=":white_check_mark:") 

373 

374 except Exception as e: 

375 output.display_error(f"Failed to remove certificate: {e}") 

376 raise typer.Exit(1) 

377 

378 

379def _get_non_bench_domains_from_nginx(services_manager) -> list[str]: 

380 """ 

381 Detect domains being proxied by nginx-proxy that are NOT Frappe benches. 

382 

383 Returns list of domain names found in nginx config that: 

384 - Have active backends (VIRTUAL_HOST containers) 

385 - Are not managed by FM benches 

386 """ 

387 try: 

388 nginx_container_name = services_manager.compose_file_manager.get_container_names().get("global-nginx-proxy") 

389 if not nginx_container_name: 

390 return [] 

391 

392 # Read default.conf which docker-gen generates 

393 result = subprocess.run( 

394 ["docker", "exec", nginx_container_name, "cat", "/etc/nginx/conf.d/default.conf"], 

395 capture_output=True, 

396 text=True, 

397 timeout=10, 

398 ) 

399 

400 if result.returncode != 0: 

401 return [] 

402 

403 # Parse upstream blocks to find domains 

404 # Format: "# domain.com/" 

405 # Followed by: "upstream domain.com {" 

406 domain_pattern = r"^# (.+?)/$" 

407 

408 detected_domains = set() 

409 for line in result.stdout.split("\n"): 

410 match = re.match(domain_pattern, line) 

411 if match: 

412 domain = match.group(1) 

413 detected_domains.add(domain) 

414 

415 # Filter out bench domains 

416 bench_service = BenchService(CLI_BENCHES_DIRECTORY, services_manager) 

417 benches = bench_service.get_bench_names() 

418 

419 bench_domains = set() 

420 for bench_name in benches: 

421 try: 

422 output = SilentOutputHandler() 

423 bench = Bench.get_object(bench_name, services_manager, logger=None, output_handler=output) 

424 bench_domains.update(bench.bench_config.get_all_domains()) 

425 except Exception: 

426 continue 

427 

428 # Return only non-bench domains 

429 non_bench_domains = detected_domains - bench_domains 

430 return sorted(list(non_bench_domains)) 

431 

432 except Exception as e: 

433 # Silently fail if we can't detect domains 

434 return [] 

435 

436 

437def _list_external_certificates(ctx: typer.Context): 

438 """List all external domain SSL certificates and detected non-SSL domains.""" 

439 

440 services_manager = ctx.obj["services"] 

441 context = LoggerContext(operation="ssl-list-external") 

442 output = get_output_handler(ctx, context=context) 

443 

444 external_config_path = services_manager.path / "nginx-proxy" / "external_domains.toml" 

445 external_manager = ExternalDomainConfigManager(external_config_path) 

446 

447 external_domains = external_manager.list_domains() 

448 

449 detected_domains = _get_non_bench_domains_from_nginx(services_manager) 

450 

451 # Filter out domains that already have SSL certificates 

452 external_domain_names = {d.domain for d in external_domains} 

453 non_ssl_domains = [d for d in detected_domains if d not in external_domain_names] 

454 

455 if not external_domains and not non_ssl_domains: 

456 output.print("No external domains or SSL certificates configured", emoji_code=":information:") 

457 output.print("", emoji_code="") 

458 output.print("To add an external certificate:", emoji_code="") 

459 output.print(" fm ssl add --standalone <domain>", emoji_code="") 

460 return 

461 

462 global_proxy_storage = services_manager.proxy_storage 

463 storage_config = SSLStorageConfig( 

464 ssl_dir=global_proxy_storage.dirs.ssl.host, 

465 ssl_dir_container=global_proxy_storage.dirs.ssl.container, 

466 certs_dir=global_proxy_storage.dirs.certs.host, 

467 certs_dir_container=global_proxy_storage.dirs.certs.container, 

468 vhostd_dir=global_proxy_storage.dirs.vhostd.host, 

469 webroot_dir=global_proxy_storage.dirs.html.host, 

470 ) 

471 

472 link_manager = CertificateLinkManager(storage_config) 

473 

474 table = Table(title="External Domains & SSL Certificates", show_header=True, header_style="bold magenta") 

475 table.add_column("Domain", style="cyan") 

476 table.add_column("Type", style="yellow") 

477 table.add_column("Status", style="green") 

478 table.add_column("Expiry", style="blue") 

479 table.add_column("Days Left", justify="right") 

480 table.add_column("Renewal", style="red") 

481 

482 # Add SSL-enabled domains 

483 for domain_config in external_domains: 

484 domain = domain_config.domain 

485 ssl_type = domain_config.ssl_type 

486 

487 try: 

488 privkey_path, fullchain_path = link_manager.get_certificate_paths(domain) 

489 

490 expiry_date = get_certificate_expiry_date(fullchain_path) 

491 if expiry_date: 

492 expiry = expiry_date.strftime("%Y-%m-%d %H:%M") 

493 # Make datetime.now() timezone-aware to match expiry_date 

494 now = datetime.now(timezone.utc) 

495 days_left = (expiry_date - now).days 

496 needs_renewal = days_left <= SSL_RENEW_BEFORE_DAYS 

497 renewal = "⚠️ DUE" if needs_renewal else "✓ OK" 

498 status = "✅ Issued" 

499 else: 

500 expiry = "N/A" 

501 days_left = "N/A" 

502 renewal = "N/A" 

503 status = "⚠️ Unknown" 

504 except Exception as e: 

505 output.debug(f"Error getting certificate status for {domain}: {e}") 

506 status = "❌ Missing" 

507 expiry = "N/A" 

508 days_left = "N/A" 

509 renewal = "N/A" 

510 

511 table.add_row(domain, ssl_type, status, expiry, str(days_left), renewal) 

512 

513 # Add detected non-SSL domains 

514 for domain in non_ssl_domains: 

515 table.add_row(domain, "none", "🔓 No SSL", "N/A", "N/A", "N/A") 

516 

517 output.stop() 

518 output.print_data(table) 

519 

520 if non_ssl_domains: 

521 output.print("\n[yellow]💡 Tip: Add SSL certificates for non-SSL domains:[/yellow]", emoji_code="") 

522 output.print("[dim] fm ssl add --standalone <domain>[/dim]", emoji_code="") 

523 

524 

525def _renew_external_certificate(ctx: typer.Context, domain: str, dry_run: bool, force: bool = False): 

526 """Renew SSL certificate for a specific external domain.""" 

527 

528 services_manager = ctx.obj["services"] 

529 context = LoggerContext(operation="ssl-renew-external") 

530 logger = ContextualLogger(log.get_logger(), context) 

531 output = get_output_handler(ctx, context=context) 

532 

533 external_config_path = services_manager.path / "nginx-proxy" / "external_domains.toml" 

534 external_manager = ExternalDomainConfigManager(external_config_path) 

535 

536 if not external_manager.domain_exists(domain): 

537 output.display_error(f"No external certificate found for domain '{domain}'") 

538 output.print("To list external certificates: fm ssl list --standalone", emoji_code="") 

539 raise typer.Exit(1) 

540 

541 output.change_head(f"Renewing certificate for {domain}") 

542 

543 try: 

544 cert = external_manager.to_ssl_certificate(domain) 

545 if not cert: 

546 raise ValueError(f"Could not create certificate object for {domain}") 

547 

548 global_proxy_storage = services_manager.proxy_storage 

549 

550 storage_config = SSLStorageConfig( 

551 ssl_dir=global_proxy_storage.dirs.ssl.host, 

552 ssl_dir_container=global_proxy_storage.dirs.ssl.container, 

553 certs_dir=global_proxy_storage.dirs.certs.host, 

554 certs_dir_container=global_proxy_storage.dirs.certs.container, 

555 vhostd_dir=global_proxy_storage.dirs.vhostd.host, 

556 webroot_dir=global_proxy_storage.dirs.html.host, 

557 ) 

558 

559 link_manager = CertificateLinkManager(storage_config) 

560 nginx_controller = services_manager.nginx_controller 

561 

562 def certificate_service_factory(cert, storage_cfg, output_handler): 

563 return create_certificate_service(logger, cert, storage_cfg, output_handler) 

564 

565 cert_manager = SSLCertificateManager( 

566 logger=logger, 

567 certificates=[cert], 

568 service_factory=certificate_service_factory, 

569 link_manager=link_manager, 

570 nginx_controller=nginx_controller, 

571 storage_config=storage_config, 

572 output_handler=output, 

573 config_save_callback=None, 

574 ) 

575 

576 with spinner(output, f"Renewing certificate for {domain}"): 

577 cert_manager.renew_certificate(domain=domain, dry_run=dry_run, force=force) 

578 output.print(f"Certificate renewal for {domain} completed", emoji_code=":white_check_mark:") 

579 

580 except Exception as e: 

581 output.display_error(f"Failed to renew certificate: {e}") 

582 raise typer.Exit(1) 

583 

584 

585def _renew_all_external_certificates(ctx: typer.Context, dry_run: bool, force: bool = False): 

586 """Renew all external domain SSL certificates.""" 

587 

588 services_manager = ctx.obj["services"] 

589 context = LoggerContext(operation="ssl-renew-external-all") 

590 output = get_output_handler(ctx, context=context) 

591 

592 external_config_path = services_manager.path / "nginx-proxy" / "external_domains.toml" 

593 external_manager = ExternalDomainConfigManager(external_config_path) 

594 

595 external_domains = external_manager.list_domains() 

596 

597 if not external_domains: 

598 output.print("No external SSL certificates to renew", emoji_code=":information:") 

599 return 

600 

601 output.change_head(f"Renewing {len(external_domains)} external certificate(s)") 

602 

603 for domain_config in external_domains: 

604 domain = domain_config.domain 

605 try: 

606 _renew_external_certificate(ctx, domain, dry_run, force) 

607 except Exception as e: 

608 output.warning(f"Failed to renew {domain}: {e}")