Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/acme2certifier/acme_srv/helpers/csr.py: 22%

1# -*- coding: utf-8 -*- 

2"""CSR utilities for acme2certifier""" 

3 

4import base64 

5import ipaddress 

6import logging 

7from typing import Dict, List, Optional, Set, Tuple 

8from cryptography import x509 

9from cryptography.hazmat.primitives import serialization 

10from .encoding import ( 

11 convert_string_to_byte, 

12 convert_byte_to_string, 

13 build_pem_file, 

14 b64_url_recode, 

15 b64_encode, 

16) 

17 

18 

19def csr_load(logger: logging.Logger, csr: str) -> x509.CertificateSigningRequest: 

20 """load certificate object from pem _Format""" 

21 logger.debug("Helper.cert_load()") 

22 

23 pem_data = convert_string_to_byte( 

24 build_pem_file(logger, None, b64_url_recode(logger, csr), True, True) 

25 ) 

26 csr_data = x509.load_pem_x509_csr(pem_data) 

27 

28 return csr_data 

29 

30 

31def csr_cn_get(logger: logging.Logger, csr_pem: str) -> str: 

32 """get cn from certificate request""" 

33 logger.debug("Helper.csr_cn_get()") 

34 

35 csr = csr_load(logger, csr_pem) 

36 # Extract the subject's common name 

37 common_name = None 

38 for attribute in csr.subject: 

39 if attribute.oid == x509.NameOID.COMMON_NAME: 

40 common_name = attribute.value 

41 break 

42 

43 logger.debug("Helper.csr_cn_get() ended with: %s", common_name) 

44 return common_name 

45 

46 

47def csr_dn_get(logger: logging.Logger, csr: str) -> str: 

48 """get subject from certificate request in openssl notation""" 

49 logger.debug("Helper.csr_dn_get()") 

50 

51 csr_obj = csr_load(logger, csr) 

52 subject = csr_obj.subject.rfc4514_string() 

53 

54 logger.debug("Helper.csr_dn_get() ended with: %s", subject) 

55 return subject 

56 

57 

58def csr_pubkey_get(logger: logging.Logger, csr, encoding="pem"): 

59 """get public key from certificate request""" 

60 logger.debug("Helper.csr_pubkey_get()") 

61 csr_obj = csr_load(logger, csr) 

62 public_key = csr_obj.public_key() 

63 if encoding == "pem": 

64 pubkey_str = public_key.public_bytes( 

65 encoding=serialization.Encoding.PEM, 

66 format=serialization.PublicFormat.SubjectPublicKeyInfo, 

67 ) 

68 pubkey = convert_byte_to_string(pubkey_str) 

69 elif encoding == "base64der": 

70 pubkey_str = public_key.public_bytes( 

71 encoding=serialization.Encoding.DER, 

72 format=serialization.PublicFormat.SubjectPublicKeyInfo, 

73 ) 

74 pubkey = b64_encode(logger, pubkey_str) 

75 

76 elif encoding == "der": 

77 pubkey = public_key.public_bytes( 

78 encoding=serialization.Encoding.DER, 

79 format=serialization.PublicFormat.SubjectPublicKeyInfo, 

80 ) 

81 else: 

82 pubkey = None 

83 logger.debug("Helper.cert_pubkey_get() ended with: %s", pubkey) 

84 return pubkey 

85 

86 

87def csr_san_get(logger: logging.Logger, csr: str) -> List[str]: 

88 """get subject alternate names from certificate""" 

89 logger.debug("Helper.cert_san_get()") 

90 sans = [] 

91 if csr: 

92 

93 csr_obj = csr_load(logger, csr) 

94 sans = [] 

95 try: 

96 ext = csr_obj.extensions.get_extension_for_oid( 

97 x509.OID_SUBJECT_ALTERNATIVE_NAME 

98 ) 

99 

100 sans_list = ext.value.get_values_for_type(x509.DNSName) 

101 for san in sans_list: 

102 sans.append(f"DNS:{san}") 

103 sans_list = ext.value.get_values_for_type(x509.IPAddress) 

104 for san in sans_list: 

105 sans.append(f"IP:{san}") 

106 sans_list = ext.value.get_values_for_type(x509.RFC822Name) 

107 for san in sans_list: 

108 sans.append(f"EMAIL:{san}") 

109 

110 except Exception as err: 

111 logger.error("Error while getting SANs from CSR: %s", err) 

112 

113 logger.debug("Helper.csr_san_get() ended with: %s", str(sans)) 

114 return sans 

115 

116 

117def _normalize_bound_name( 

118 name_type: str, 

119 name_value: str, 

120 email_identifier_rewrite: bool = False, 

121) -> Optional[Tuple[str, str]]: 

122 """Normalize a (type, value) pair for CSR/order binding comparison. 

123 

124 When email_identifier_rewrite is True (Order.email_identifier_rewrite), 

125 values containing '@' are typed as email so dns:user@host and EMAIL SANs 

126 for the same address compare equal (acme_email / RFC 8823 rewrite path). 

127 """ 

128 if not name_type or name_value is None or name_value == "": 

129 return None 

130 normalized_type = name_type.lower() 

131 normalized_value = name_value.lower() 

132 if email_identifier_rewrite and "@" in normalized_value: 

133 normalized_type = "email" 

134 elif normalized_type == "ip": 

135 try: 

136 normalized_value = str(ipaddress.ip_address(name_value.strip())) 

137 except ValueError: 

138 pass 

139 return (normalized_type, normalized_value) 

140 

141 

142def _cn_bound_type(cn: str) -> str: 

143 """Infer ACME identifier type for a CSR Common Name.""" 

144 if "@" in cn: 

145 return "email" 

146 try: 

147 ipaddress.ip_address(cn) 

148 return "ip" 

149 except ValueError: 

150 return "dns" 

151 

152 

153def san_list_to_bound_names( 

154 logger: logging.Logger, 

155 san_list: List[str], 

156 email_identifier_rewrite: bool = False, 

157) -> Set[Tuple[str, str]]: 

158 """Convert DNS:/IP:/EMAIL: SAN strings to normalized bound-name pairs.""" 

159 names: Set[Tuple[str, str]] = set() 

160 for san in san_list: 

161 try: 

162 san_type, san_value = san.split(":", 1) 

163 except ValueError as err: 

164 logger.error("Error while splitting SAN %s: %s", san, err) 

165 continue 

166 normalized = _normalize_bound_name( 

167 san_type, san_value, email_identifier_rewrite=email_identifier_rewrite 

168 ) 

169 if normalized: 

170 names.add(normalized) 

171 return names 

172 

173 

174def csr_bound_names_get( 

175 logger: logging.Logger, 

176 csr: str, 

177 email_identifier_rewrite: bool = False, 

178) -> Set[Tuple[str, str]]: 

179 """ 

180 Return normalized (type, value) pairs from CSR SANs and subject CN. 

181 

182 SAN types map DNS/IP/EMAIL → dns/ip/email. A subject CN is included when 

183 present (email if it contains '@', else ip if parseable, else dns) and 

184 deduplicated against SANs. With email_identifier_rewrite, DNS SANs that 

185 look like email addresses are typed as email. 

186 """ 

187 logger.debug( 

188 "Helper.csr_bound_names_get(email_identifier_rewrite=%s)", 

189 email_identifier_rewrite, 

190 ) 

191 names: Set[Tuple[str, str]] = set() 

192 if not csr: 

193 logger.debug("Helper.csr_bound_names_get() ended with: %s", names) 

194 return names 

195 

196 for san in csr_san_get(logger, csr): 

197 try: 

198 san_type, san_value = san.split(":", 1) 

199 except ValueError as err: 

200 logger.error("Error while splitting SAN %s: %s", san, err) 

201 continue 

202 normalized = _normalize_bound_name( 

203 san_type, san_value, email_identifier_rewrite=email_identifier_rewrite 

204 ) 

205 if normalized: 

206 names.add(normalized) 

207 

208 cn = csr_cn_get(logger, csr) 

209 if cn: 

210 normalized = _normalize_bound_name( 

211 _cn_bound_type(cn), 

212 cn, 

213 email_identifier_rewrite=email_identifier_rewrite, 

214 ) 

215 if normalized: 

216 names.add(normalized) 

217 

218 logger.debug("Helper.csr_bound_names_get() ended with: %s", names) 

219 return names 

220 

221 

222def csr_san_byte_get(logger: logging.Logger, csr: str) -> bytes: 

223 """get sans from CSR as base64 encoded byte squence""" 

224 # Load the CSR 

225 logger.debug("Helper.csr_san_byte_get()") 

226 

227 csr = csr_load(logger, csr) 

228 

229 # Get the SAN extension 

230 san_extension = csr.extensions.get_extension_for_oid( 

231 x509.oid.ExtensionOID.SUBJECT_ALTERNATIVE_NAME 

232 ) 

233 

234 # Get the SANs 

235 sans = san_extension.value 

236 

237 # Serialize the SANs to a byte sequence 

238 sans_bytes = sans.public_bytes() 

239 

240 # Encode the byte sequence as base64 

241 sans_base64 = b64_encode(logger, sans_bytes) 

242 

243 logger.debug("Helper.csr_san_byte_get() ended") 

244 return sans_base64 

245 

246 

247def csr_extensions_get(logger: logging.Logger, csr: str) -> List[str]: 

248 """get extensions from certificate""" 

249 logger.debug("Helper.csr_extensions_get()") 

250 

251 csr_obj = csr_load(logger, csr) 

252 

253 extension_list = [] 

254 for extension in csr_obj.extensions: 

255 extension_list.append( 

256 convert_byte_to_string(base64.b64encode(extension.value.public_bytes())) 

257 ) 

258 

259 logger.debug("Helper.csr_extensions_get() ended with: %s", extension_list) 

260 return extension_list 

261 

262 

263def csr_subject_get(logger: logging.Logger, csr: str) -> Dict[str, str]: 

264 """get subject from csr as a list of tuples""" 

265 logger.debug("Helper.csr_subject_get()") 

266 # pylint: disable=w0212 

267 

268 # extend OID library from cryptography module 

269 OID_NAME_MAP = { 

270 "2.5.4.97": "organizationIdentifier", 

271 } 

272 

273 csr_obj = csr_load(logger, csr) 

274 subject_dic = {} 

275 # get subject and look for common name 

276 subject = csr_obj.subject 

277 for attr in subject: 

278 # use mapping table as primiary source, otherwise oid names from library 

279 name = OID_NAME_MAP.get(attr.oid.dotted_string, attr.oid._name) 

280 subject_dic[name] = attr.value 

281 

282 logger.debug("Helper.csr_subject_get() ended") 

283 return subject_dic 

284 

285 

286def csr_cn_lookup(logger: logging.Logger, csr: str) -> str: 

287 """lookup CN/ 1st san from CSR""" 

288 logger.debug("Heloer._csr_cn_lookup()") 

289 

290 csr_cn = csr_cn_get(logger, csr) 

291 if not csr_cn: 

292 # lookup first san 

293 san_list = csr_san_get(logger, csr) 

294 if san_list and len(san_list) > 0: 

295 for san in san_list: 

296 try: 

297 csr_cn = san.split(":")[1] 

298 break 

299 except Exception as err: 

300 logger.error("SAN split failed: %s", err) 

301 else: 

302 logger.error("No SANs found in CSR") 

303 

304 logger.debug("Helper._csr_cn_lookup() ended with: %s", csr_cn) 

305 return csr_cn