Coverage for frappe_manager / commands / ssl / acme_sh.py: 24%

1"""acme.sh passthrough command.""" 

2 

3import os 

4 

5import typer 

6from typer_examples import example 

7 

8from frappe_manager.utils.subprocess import stream_command_output 

9 

10from .helpers import get_output_handler 

11 

12 

13@example( 

14 "Show acme.sh help and available commands", 

15 "", 

16 detail="Displays acme.sh help. Use for learning available subcommands and flags.", 

17) 

18@example( 

19 "List all certificates managed by acme.sh", 

20 "--list", 

21 detail="Lists certificates that acme.sh currently manages in its home directory.", 

22) 

23@example( 

24 "Show certificate information for a domain", 

25 "--info -d example.com", 

26 detail="Shows detailed information for a managed certificate for the domain.", 

27) 

28@example( 

29 "Check acme.sh version", 

30 "--version", 

31 detail="Prints the installed acme.sh version used by FM.", 

32) 

33@example( 

34 "Upgrade acme.sh to latest version", 

35 "--upgrade", 

36 detail="Upgrades the bundled acme.sh installation to the latest release.", 

37) 

38@example( 

39 "Force renew certificate for a domain", 

40 "--renew -d example.com --force", 

41 detail="Forces a renewal for a certificate using acme.sh; advanced option for recovery and testing.", 

42) 

43def acmesh_passthrough( 

44 ctx: typer.Context, 

45): 

46 """ 

47 Run acme.sh commands directly with FM's environment (advanced users). 

48 

49 Advanced users only: this bypasses FM's certificate management. Prefer 'fm ssl add/remove/renew' for normal workflows. 

50 """ 

51 args = ctx.args 

52 

53 services_manager = ctx.obj["services"] 

54 output = get_output_handler(ctx) 

55 

56 global_proxy_storage = services_manager.proxy_storage 

57 ssl_dir = global_proxy_storage.dirs.ssl.host 

58 acmesh_home = ssl_dir / "acmesh" / ".acme.sh" 

59 acmesh_bin = acmesh_home / "acme.sh" 

60 

61 if not acmesh_bin.exists(): 

62 output.display_error("acme.sh is not installed yet") 

63 output.info("Run 'fm ssl add <benchname> <domain>' to install acme.sh first") 

64 raise typer.Exit(1) 

65 

66 cmd = [str(acmesh_bin), "--home", str(acmesh_home)] 

67 

68 # If no args provided, show help 

69 if not args: 

70 cmd.append("--help") 

71 else: 

72 cmd.extend(args) 

73 

74 env = os.environ.copy() 

75 env["LE_WORKING_DIR"] = str(acmesh_home) 

76 

77 output.change_head("Running acme.sh") 

78 output.info(f"Command: acme.sh {' '.join(args or ['--help'])}") 

79 output.info(f"Home: {acmesh_home}") 

80 output.print("") 

81 

82 # Stream output directly to user 

83 exit_code_holder = [0] 

84 

85 def stream_with_exit_tracking(): 

86 """Generator that tracks exit code while yielding output.""" 

87 for source, line in stream_command_output(cmd, env=env, cwd=None): 

88 if source == "exit_code": 

89 exit_code_holder[0] = int(line.decode()) 

90 yield source, line 

91 

92 # Display all output (print directly for raw acme.sh output) 

93 for source, line in stream_with_exit_tracking(): 

94 if source in ("stdout", "stderr"): 

95 # Print directly without prefix for raw acme.sh output 

96 decoded = line.decode() 

97 print(decoded, flush=True) 

98 

99 # Exit with acme.sh's exit code 

100 if exit_code_holder[0] != 0: 

101 output.print("") 

102 output.display_error(f"acme.sh exited with code {exit_code_holder[0]}") 

103 raise typer.Exit(exit_code_holder[0]) 

104 output.print("") 

105 output.print("Command completed successfully", emoji_code=":white_check_mark:")