Coverage for frappe_manager / ssl_manager / certificate_link_manager.py: 21%

58 statements  

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

1""" 

2Manages symlinks between certificate files and nginx-proxy directories. 

3 

4This module handles the creation and removal of symbolic links that connect 

5Let's Encrypt certificates (stored in ssl/ directory) to the nginx-proxy certs/ 

6directory where nginx-proxy expects to find them. 

7""" 

8 

9from pathlib import Path 

10 

11from frappe_manager.ssl_manager.storage_config import SSLStorageConfig 

12from frappe_manager.utils.helpers import create_symlink 

13 

14 

15class CertificateLinkManager: 

16 """ 

17 Manages certificate symlinks for nginx-proxy. 

18 

19 This class is responsible for creating and removing symbolic links between: 

20 - Certificate files generated by Let's Encrypt (in ssl/ directory) 

21 - Certificate locations expected by nginx-proxy (in certs/ directory) 

22 

23 The symlinks allow nginx-proxy to access certificates for both primary 

24 and alias domains without duplicating certificate files. 

25 

26 Attributes: 

27 storage_config: Configuration for SSL storage paths 

28 """ 

29 

30 def __init__(self, storage_config: SSLStorageConfig): 

31 """ 

32 Initialize the certificate link manager. 

33 

34 Args: 

35 storage_config: Configuration object with paths to ssl and certs directories 

36 

37 Raises: 

38 ValueError: If storage_config paths are invalid 

39 """ 

40 self.storage_config = storage_config 

41 # Validate that required directories exist 

42 self._validate_prerequisites() 

43 

44 def _validate_prerequisites(self): 

45 """ 

46 Validate that prerequisite directories exist. 

47 

48 Raises: 

49 ValueError: If required directories don't exist 

50 """ 

51 if not self.storage_config.certs_dir.exists(): 

52 raise ValueError( 

53 f"Certificate directory does not exist: {self.storage_config.certs_dir}. " 

54 "Ensure nginx-proxy is running and volumes are mounted correctly.", 

55 ) 

56 

57 if not self.storage_config.ssl_dir.exists(): 

58 raise ValueError( 

59 f"SSL directory does not exist: {self.storage_config.ssl_dir}. " 

60 "Ensure nginx-proxy is running and volumes are mounted correctly.", 

61 ) 

62 

63 def link_certificate( 

64 self, 

65 cert_type: str, 

66 domain: str, 

67 privkey_path: Path, 

68 fullchain_path: Path, 

69 alias_domains: list[str] | None = None, 

70 ): 

71 """ 

72 Create symlinks for a certificate and its alias domains. 

73 

74 This creates symlinks from the certs/ directory to the actual certificate 

75 files in the ssl/ directory. Both the primary domain and any alias domains 

76 get their own symlinks pointing to the same certificate files. 

77 

78 Args: 

79 cert_type: Type of certificate (e.g., 'letsencrypt', 'self-signed') 

80 domain: Primary domain name for the certificate 

81 privkey_path: Path to the private key file (host path) 

82 fullchain_path: Path to the full chain certificate file (host path) 

83 alias_domains: Optional list of alias domains that should also point to this cert 

84 """ 

85 # Convert host paths to container paths for symlink targets 

86 container_privkey_path = self._host_to_container_path(privkey_path, cert_type) 

87 container_fullchain_path = self._host_to_container_path(fullchain_path, cert_type) 

88 

89 primary_privkey_link = self.storage_config.certs_dir / f"{domain}.key" 

90 primary_fullchain_link = self.storage_config.certs_dir / f"{domain}.crt" 

91 

92 create_symlink(container_privkey_path, primary_privkey_link) 

93 create_symlink(container_fullchain_path, primary_fullchain_link) 

94 

95 if alias_domains: 

96 for alias_domain in alias_domains: 

97 alias_privkey_link = self.storage_config.certs_dir / f"{alias_domain}.key" 

98 alias_fullchain_link = self.storage_config.certs_dir / f"{alias_domain}.crt" 

99 

100 create_symlink(container_privkey_path, alias_privkey_link) 

101 create_symlink(container_fullchain_path, alias_fullchain_link) 

102 

103 def unlink_certificate(self, domain: str, alias_domains: list[str] | None = None): 

104 """ 

105 Remove symlinks for a certificate and its alias domains. 

106 

107 Args: 

108 domain: Primary domain name for the certificate 

109 alias_domains: Optional list of alias domains to also unlink 

110 """ 

111 primary_privkey_link = self.storage_config.certs_dir / f"{domain}.key" 

112 primary_fullchain_link = self.storage_config.certs_dir / f"{domain}.crt" 

113 

114 self._safe_unlink(primary_privkey_link) 

115 self._safe_unlink(primary_fullchain_link) 

116 

117 # Remove alias domain symlinks 

118 if alias_domains: 

119 for alias_domain in alias_domains: 

120 alias_privkey_link = self.storage_config.certs_dir / f"{alias_domain}.key" 

121 alias_fullchain_link = self.storage_config.certs_dir / f"{alias_domain}.crt" 

122 

123 self._safe_unlink(alias_privkey_link) 

124 self._safe_unlink(alias_fullchain_link) 

125 

126 def get_certificate_paths(self, domain: str) -> tuple[Path, Path]: 

127 """ 

128 Get the actual certificate file paths by resolving symlinks. 

129 

130 This follows the symlinks in the certs/ directory to find the actual 

131 certificate files in the ssl/ directory. 

132 

133 Args: 

134 domain: Domain name to get certificate paths for 

135 

136 Returns: 

137 Tuple of (privkey_path, fullchain_path) as resolved host paths 

138 

139 Raises: 

140 FileNotFoundError: If the certificate symlinks don't exist 

141 """ 

142 privkey_link = self.storage_config.certs_dir / f"{domain}.key" 

143 fullchain_link = self.storage_config.certs_dir / f"{domain}.crt" 

144 

145 # Resolve symlinks to get container paths 

146 container_privkey_path = privkey_link.readlink() 

147 container_fullchain_path = fullchain_link.readlink() 

148 

149 # Convert container paths back to host paths 

150 privkey_path = self._container_to_host_path(container_privkey_path) 

151 fullchain_path = self._container_to_host_path(container_fullchain_path) 

152 

153 return (privkey_path, fullchain_path) 

154 

155 def _host_to_container_path(self, host_path: Path, cert_type: str) -> Path: 

156 """ 

157 Convert a host filesystem path to a container filesystem path. 

158 

159 Args: 

160 host_path: Path on the host filesystem 

161 cert_type: Certificate type (e.g., 'letsencrypt') 

162 

163 Returns: 

164 Equivalent path as seen from inside the container 

165 """ 

166 # Calculate relative path from the ssl type directory 

167 ssl_type_host_dir = self.storage_config.ssl_dir / cert_type 

168 relative_path = host_path.relative_to(ssl_type_host_dir) 

169 

170 # Build container path 

171 container_path = self.storage_config.ssl_dir_container / cert_type / relative_path 

172 return container_path 

173 

174 def _container_to_host_path(self, container_path: Path) -> Path: 

175 """ 

176 Convert a container filesystem path to a host filesystem path. 

177 

178 Args: 

179 container_path: Path as seen from inside the container 

180 

181 Returns: 

182 Equivalent path on the host filesystem 

183 """ 

184 # Calculate relative path from the container ssl directory 

185 relative_path = container_path.relative_to(self.storage_config.ssl_dir_container) 

186 

187 # Build host path 

188 host_path = self.storage_config.ssl_dir / relative_path 

189 return host_path 

190 

191 def _safe_unlink(self, path: Path): 

192 """ 

193 Safely remove a symlink, ignoring errors if it doesn't exist. 

194 

195 Args: 

196 path: Path to the symlink to remove 

197 """ 

198 try: 

199 path.unlink() 

200 except (FileNotFoundError, OSError): 

201 # Ignore errors - symlink might already be removed or never existed 

202 pass