Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/urllib3/util/ssl_.py: 26%

1from __future__ import annotations 

2 

3import hashlib 

4import hmac 

5import os 

6import socket 

7import sys 

8import typing 

9import warnings 

10from binascii import Error as BinasciiError 

11from binascii import unhexlify 

12 

13from ..exceptions import ProxySchemeUnsupported, SSLError 

14from .url import _BRACELESS_IPV6_ADDRZ_RE, _IPV4_RE 

15 

16SSLContext = None 

17SSLTransport = None 

18HAS_NEVER_CHECK_COMMON_NAME = False 

19IS_PYOPENSSL = False 

20ALPN_PROTOCOLS = ["http/1.1"] 

21 

22_TYPE_VERSION_INFO = tuple[int, int, int, str, int] 

23 

24# Maps the length of a digest to a possible hash function producing this digest 

25HASHFUNC_MAP = { 

26 length: getattr(hashlib, algorithm, None) 

27 for length, algorithm in ((32, "md5"), (40, "sha1"), (64, "sha256")) 

28} 

29 

30 

31def _is_has_never_check_common_name_reliable( 

32 openssl_version: str, 

33) -> bool: 

34 # As of May 2023, all released versions of LibreSSL fail to reject certificates with 

35 # only common names, see https://github.com/urllib3/urllib3/pull/3024 

36 is_openssl = openssl_version.startswith("OpenSSL ") 

37 

38 return is_openssl 

39 

40 

41if typing.TYPE_CHECKING: 

42 from ssl import VerifyMode 

43 from typing import TypedDict 

44 

45 from .ssltransport import SSLTransport as SSLTransportType 

46 

47 class _TYPE_PEER_CERT_RET_DICT(TypedDict, total=False): 

48 subjectAltName: tuple[tuple[str, str], ...] 

49 subject: tuple[tuple[tuple[str, str], ...], ...] 

50 serialNumber: str 

51 

52 

53# Mapping from 'ssl.PROTOCOL_TLSX' to 'TLSVersion.X' 

54_SSL_VERSION_TO_TLS_VERSION: dict[int, int] = {} 

55 

56try: # Do we have ssl at all? 

57 import ssl 

58 from ssl import ( # type: ignore[assignment] 

59 CERT_REQUIRED, 

60 HAS_NEVER_CHECK_COMMON_NAME, 

61 OP_NO_COMPRESSION, 

62 OP_NO_TICKET, 

63 OPENSSL_VERSION, 

64 PROTOCOL_TLS, 

65 PROTOCOL_TLS_CLIENT, 

66 VERIFY_X509_PARTIAL_CHAIN, 

67 VERIFY_X509_STRICT, 

68 OP_NO_SSLv2, 

69 OP_NO_SSLv3, 

70 SSLContext, 

71 TLSVersion, 

72 ) 

73 

74 PROTOCOL_SSLv23 = PROTOCOL_TLS 

75 

76 # Setting SSLContext.hostname_checks_common_name = False didn't work with 

77 # LibreSSL, check details in the used function. 

78 if HAS_NEVER_CHECK_COMMON_NAME and not _is_has_never_check_common_name_reliable( 

79 OPENSSL_VERSION, 

80 ): # Defensive: 

81 HAS_NEVER_CHECK_COMMON_NAME = False 

82 

83 # Need to be careful here in case old TLS versions get 

84 # removed in future 'ssl' module implementations. 

85 for attr in ("TLSv1", "TLSv1_1", "TLSv1_2"): 

86 try: 

87 _SSL_VERSION_TO_TLS_VERSION[getattr(ssl, f"PROTOCOL_{attr}")] = getattr( 

88 TLSVersion, attr 

89 ) 

90 except AttributeError: # Defensive: 

91 continue 

92 

93 from .ssltransport import SSLTransport # type: ignore[assignment] 

94except ImportError: 

95 OP_NO_COMPRESSION = 0x20000 # type: ignore[assignment,misc] 

96 OP_NO_TICKET = 0x4000 # type: ignore[assignment,misc] 

97 OP_NO_SSLv2 = 0x1000000 # type: ignore[assignment,misc] 

98 OP_NO_SSLv3 = 0x2000000 # type: ignore[assignment,misc] 

99 PROTOCOL_SSLv23 = PROTOCOL_TLS = 2 # type: ignore[assignment,misc] 

100 PROTOCOL_TLS_CLIENT = 16 # type: ignore[assignment,misc] 

101 VERIFY_X509_PARTIAL_CHAIN = 0x80000 # type: ignore[assignment,misc] 

102 VERIFY_X509_STRICT = 0x20 # type: ignore[assignment,misc] 

103 

104 

105_TYPE_PEER_CERT_RET = typing.Union["_TYPE_PEER_CERT_RET_DICT", bytes, None] 

106 

107 

108def assert_fingerprint(cert: bytes | None, fingerprint: str) -> None: 

109 """ 

110 Checks if given fingerprint matches the supplied certificate. 

111 

112 :param cert: 

113 Certificate as bytes object. 

114 :param fingerprint: 

115 Fingerprint as string of hexdigits, can be interspersed by colons. 

116 """ 

117 

118 if cert is None: 

119 raise SSLError("No certificate for the peer.") 

120 

121 fingerprint = fingerprint.replace(":", "").lower() 

122 digest_length = len(fingerprint) 

123 if digest_length not in HASHFUNC_MAP: 

124 raise SSLError(f"Fingerprint of invalid length: {fingerprint}") 

125 hashfunc = HASHFUNC_MAP.get(digest_length) 

126 if hashfunc is None: 

127 raise SSLError( 

128 f"Hash function implementation unavailable for fingerprint length: {digest_length}" 

129 ) 

130 

131 # We need encode() here for py32; works on py2 and p33. 

132 try: 

133 fingerprint_bytes = unhexlify(fingerprint.encode()) 

134 except BinasciiError as e: 

135 raise SSLError(e) from e 

136 

137 cert_digest = hashfunc(cert).digest() 

138 

139 if not hmac.compare_digest(cert_digest, fingerprint_bytes): 

140 raise SSLError( 

141 f'Fingerprints did not match. Expected "{fingerprint}", got "{cert_digest.hex()}"' 

142 ) 

143 

144 

145def resolve_cert_reqs(candidate: None | int | str) -> VerifyMode: 

146 """ 

147 Resolves the argument to a numeric constant, which can be passed to 

148 the wrap_socket function/method from the ssl module. 

149 Defaults to :data:`ssl.CERT_REQUIRED`. 

150 If given a string it is assumed to be the name of the constant in the 

151 :mod:`ssl` module or its abbreviation. 

152 (So you can specify `REQUIRED` instead of `CERT_REQUIRED`. 

153 If it's neither `None` nor a string we assume it is already the numeric 

154 constant which can directly be passed to wrap_socket. 

155 """ 

156 if candidate is None: 

157 return CERT_REQUIRED 

158 

159 if isinstance(candidate, str): 

160 res = getattr(ssl, candidate, None) 

161 if res is None: 

162 res = getattr(ssl, "CERT_" + candidate) 

163 return res # type: ignore[no-any-return] 

164 

165 return candidate # type: ignore[return-value] 

166 

167 

168def resolve_ssl_version(candidate: None | int | str) -> int: 

169 """ 

170 like resolve_cert_reqs 

171 """ 

172 if candidate is None: 

173 return PROTOCOL_TLS 

174 

175 if isinstance(candidate, str): 

176 res = getattr(ssl, candidate, None) 

177 if res is None: 

178 res = getattr(ssl, "PROTOCOL_" + candidate) 

179 return typing.cast(int, res) 

180 

181 return candidate 

182 

183 

184def create_urllib3_context( 

185 ssl_version: int | None = None, 

186 cert_reqs: int | None = None, 

187 options: int | None = None, 

188 ciphers: str | None = None, 

189 ssl_minimum_version: int | None = None, 

190 ssl_maximum_version: int | None = None, 

191 verify_flags: int | None = None, 

192) -> ssl.SSLContext: 

193 """Creates and configures an :class:`ssl.SSLContext` instance for use with urllib3. 

194 

195 :param ssl_version: 

196 The desired protocol version to use. This will default to 

197 PROTOCOL_SSLv23 which will negotiate the highest protocol that both 

198 the server and your installation of OpenSSL support. 

199 

200 This parameter is deprecated instead use 'ssl_minimum_version'. 

201 :param ssl_minimum_version: 

202 The minimum version of TLS to be used. Use the 'ssl.TLSVersion' enum for specifying the value. 

203 :param ssl_maximum_version: 

204 The maximum version of TLS to be used. Use the 'ssl.TLSVersion' enum for specifying the value. 

205 Not recommended to set to anything other than 'ssl.TLSVersion.MAXIMUM_SUPPORTED' which is the 

206 default value. 

207 :param cert_reqs: 

208 Whether to require the certificate verification. This defaults to 

209 ``ssl.CERT_REQUIRED``. 

210 :param options: 

211 Specific OpenSSL options. These default to ``ssl.OP_NO_SSLv2``, 

212 ``ssl.OP_NO_SSLv3``, ``ssl.OP_NO_COMPRESSION``, and ``ssl.OP_NO_TICKET``. 

213 :param ciphers: 

214 Which cipher suites to allow the server to select. Defaults to either system configured 

215 ciphers if OpenSSL 1.1.1+, otherwise uses a secure default set of ciphers. 

216 :param verify_flags: 

217 The flags for certificate verification operations. These default to 

218 ``ssl.VERIFY_X509_PARTIAL_CHAIN`` and ``ssl.VERIFY_X509_STRICT`` for Python 3.13+. 

219 :returns: 

220 Constructed SSLContext object with specified options 

221 :rtype: SSLContext 

222 """ 

223 if SSLContext is None: 

224 raise TypeError("Can't create an SSLContext object without an ssl module") 

225 

226 # This means 'ssl_version' was specified as an exact value. 

227 if ssl_version not in (None, PROTOCOL_TLS, PROTOCOL_TLS_CLIENT): 

228 # Disallow setting 'ssl_version' and 'ssl_minimum|maximum_version' 

229 # to avoid conflicts. 

230 if ssl_minimum_version is not None or ssl_maximum_version is not None: 

231 raise ValueError( 

232 "Can't specify both 'ssl_version' and either " 

233 "'ssl_minimum_version' or 'ssl_maximum_version'" 

234 ) 

235 

236 # 'ssl_version' is deprecated and will be removed in the future. 

237 else: 

238 # Use 'ssl_minimum_version' and 'ssl_maximum_version' instead. 

239 ssl_minimum_version = _SSL_VERSION_TO_TLS_VERSION.get( 

240 ssl_version, TLSVersion.MINIMUM_SUPPORTED 

241 ) 

242 ssl_maximum_version = _SSL_VERSION_TO_TLS_VERSION.get( 

243 ssl_version, TLSVersion.MAXIMUM_SUPPORTED 

244 ) 

245 

246 # This warning message is pushing users to use 'ssl_minimum_version' 

247 # instead of both min/max. Best practice is to only set the minimum version and 

248 # keep the maximum version to be it's default value: 'TLSVersion.MAXIMUM_SUPPORTED' 

249 warnings.warn( 

250 "'ssl_version' option is deprecated and will be " 

251 "removed in urllib3 v3.0. Instead use 'ssl_minimum_version'", 

252 category=FutureWarning, 

253 stacklevel=2, 

254 ) 

255 

256 context = SSLContext(PROTOCOL_TLS_CLIENT) 

257 if ssl_minimum_version is not None: 

258 context.minimum_version = ssl_minimum_version 

259 else: # pyOpenSSL defaults to 'MINIMUM_SUPPORTED' so explicitly set TLSv1.2 here 

260 context.minimum_version = TLSVersion.TLSv1_2 

261 

262 if ssl_maximum_version is not None: 

263 context.maximum_version = ssl_maximum_version 

264 

265 # Unless we're given ciphers defer to either system ciphers in 

266 # the case of OpenSSL 1.1.1+ or use our own secure default ciphers. 

267 if ciphers: 

268 context.set_ciphers(ciphers) 

269 

270 # Setting the default here, as we may have no ssl module on import 

271 cert_reqs = ssl.CERT_REQUIRED if cert_reqs is None else cert_reqs 

272 

273 if options is None: 

274 options = 0 

275 # SSLv2 is easily broken and is considered harmful and dangerous 

276 options |= OP_NO_SSLv2 

277 # SSLv3 has several problems and is now dangerous 

278 options |= OP_NO_SSLv3 

279 # Disable compression to prevent CRIME attacks for OpenSSL 1.0+ 

280 # (issue #309) 

281 options |= OP_NO_COMPRESSION 

282 # TLSv1.2 only. Unless set explicitly, do not request tickets. 

283 # This may save some bandwidth on wire, and although the ticket is encrypted, 

284 # there is a risk associated with it being on wire, 

285 # if the server is not rotating its ticketing keys properly. 

286 options |= OP_NO_TICKET 

287 

288 context.options |= options 

289 

290 if verify_flags is None: 

291 verify_flags = 0 

292 # In Python 3.13+ ssl.create_default_context() sets VERIFY_X509_PARTIAL_CHAIN 

293 # and VERIFY_X509_STRICT so we do the same 

294 if sys.version_info >= (3, 13): 

295 verify_flags |= VERIFY_X509_PARTIAL_CHAIN 

296 verify_flags |= VERIFY_X509_STRICT 

297 

298 context.verify_flags |= verify_flags 

299 

300 # Enable post-handshake authentication for TLS 1.3, see GH #1634. PHA is 

301 # necessary for conditional client cert authentication with TLS 1.3. 

302 # The attribute is None for OpenSSL <= 1.1.0 or does not exist when using 

303 # an SSLContext created by pyOpenSSL. 

304 if getattr(context, "post_handshake_auth", None) is not None: 

305 context.post_handshake_auth = True 

306 

307 # The order of the below lines setting verify_mode and check_hostname 

308 # matter due to safe-guards SSLContext has to prevent an SSLContext with 

309 # check_hostname=True, verify_mode=NONE/OPTIONAL. 

310 # We always set 'check_hostname=False' for pyOpenSSL so we rely on our own 

311 # 'ssl.match_hostname()' implementation. 

312 if cert_reqs == ssl.CERT_REQUIRED and not IS_PYOPENSSL: 

313 context.verify_mode = cert_reqs 

314 context.check_hostname = True 

315 else: 

316 context.check_hostname = False 

317 context.verify_mode = cert_reqs 

318 

319 context.hostname_checks_common_name = False 

320 

321 if "SSLKEYLOGFILE" in os.environ: 

322 sslkeylogfile = os.path.expandvars(os.environ.get("SSLKEYLOGFILE")) 

323 else: 

324 sslkeylogfile = None 

325 if sslkeylogfile: 

326 context.keylog_filename = sslkeylogfile 

327 

328 return context 

329 

330 

331@typing.overload 

332def ssl_wrap_socket( 

333 sock: socket.socket, 

334 keyfile: str | None = ..., 

335 certfile: str | None = ..., 

336 cert_reqs: int | None = ..., 

337 ca_certs: str | None = ..., 

338 server_hostname: str | None = ..., 

339 ssl_version: int | None = ..., 

340 ciphers: str | None = ..., 

341 ssl_context: ssl.SSLContext | None = ..., 

342 ca_cert_dir: str | None = ..., 

343 key_password: str | None = ..., 

344 ca_cert_data: None | str | bytes = ..., 

345 tls_in_tls: typing.Literal[False] = ..., 

346) -> ssl.SSLSocket: ... 

347 

348 

349@typing.overload 

350def ssl_wrap_socket( 

351 sock: socket.socket, 

352 keyfile: str | None = ..., 

353 certfile: str | None = ..., 

354 cert_reqs: int | None = ..., 

355 ca_certs: str | None = ..., 

356 server_hostname: str | None = ..., 

357 ssl_version: int | None = ..., 

358 ciphers: str | None = ..., 

359 ssl_context: ssl.SSLContext | None = ..., 

360 ca_cert_dir: str | None = ..., 

361 key_password: str | None = ..., 

362 ca_cert_data: None | str | bytes = ..., 

363 tls_in_tls: bool = ..., 

364) -> ssl.SSLSocket | SSLTransportType: ... 

365 

366 

367def ssl_wrap_socket( 

368 sock: socket.socket, 

369 keyfile: str | None = None, 

370 certfile: str | None = None, 

371 cert_reqs: int | None = None, 

372 ca_certs: str | None = None, 

373 server_hostname: str | None = None, 

374 ssl_version: int | None = None, 

375 ciphers: str | None = None, 

376 ssl_context: ssl.SSLContext | None = None, 

377 ca_cert_dir: str | None = None, 

378 key_password: str | None = None, 

379 ca_cert_data: None | str | bytes = None, 

380 tls_in_tls: bool = False, 

381) -> ssl.SSLSocket | SSLTransportType: 

382 """ 

383 All arguments except for server_hostname, ssl_context, tls_in_tls, ca_cert_data and 

384 ca_cert_dir have the same meaning as they do when using 

385 :func:`ssl.create_default_context`, :meth:`ssl.SSLContext.load_cert_chain`, 

386 :meth:`ssl.SSLContext.set_ciphers` and :meth:`ssl.SSLContext.wrap_socket`. 

387 

388 :param server_hostname: 

389 When SNI is supported, the expected hostname of the certificate 

390 :param ssl_context: 

391 A pre-made :class:`SSLContext` object. If none is provided, one will 

392 be created using :func:`create_urllib3_context`. 

393 :param ciphers: 

394 A string of ciphers we wish the client to support. 

395 :param ca_cert_dir: 

396 A directory containing CA certificates in multiple separate files, as 

397 supported by OpenSSL's -CApath flag or the capath argument to 

398 SSLContext.load_verify_locations(). 

399 :param key_password: 

400 Optional password if the keyfile is encrypted. 

401 :param ca_cert_data: 

402 Optional string containing CA certificates in PEM format suitable for 

403 passing as the cadata parameter to SSLContext.load_verify_locations() 

404 :param tls_in_tls: 

405 Use SSLTransport to wrap the existing socket. 

406 """ 

407 context = ssl_context 

408 if context is None: 

409 # Note: This branch of code and all the variables in it are only used in tests. 

410 # We should consider deprecating and removing this code. 

411 context = create_urllib3_context(ssl_version, cert_reqs, ciphers=ciphers) 

412 

413 if ca_certs or ca_cert_dir or ca_cert_data: 

414 try: 

415 context.load_verify_locations(ca_certs, ca_cert_dir, ca_cert_data) 

416 except OSError as e: 

417 raise SSLError(e) from e 

418 

419 elif ssl_context is None and hasattr(context, "load_default_certs"): 

420 # try to load OS default certs; works well on Windows. 

421 context.load_default_certs() 

422 

423 # Attempt to detect if we get the goofy behavior of the 

424 # keyfile being encrypted and OpenSSL asking for the 

425 # passphrase via the terminal and instead error out. 

426 if keyfile and key_password is None and _is_key_file_encrypted(keyfile): 

427 raise SSLError("Client private key is encrypted, password is required") 

428 

429 if certfile: 

430 if key_password is None: 

431 context.load_cert_chain(certfile, keyfile) 

432 else: 

433 context.load_cert_chain(certfile, keyfile, key_password) 

434 

435 context.set_alpn_protocols(ALPN_PROTOCOLS) 

436 

437 ssl_sock = _ssl_wrap_socket_impl(sock, context, tls_in_tls, server_hostname) 

438 return ssl_sock 

439 

440 

441def is_ipaddress(hostname: str | bytes) -> bool: 

442 """Detects whether the hostname given is an IPv4 or IPv6 address. 

443 Also detects IPv6 addresses with Zone IDs. 

444 

445 :param str hostname: Hostname to examine. 

446 :return: True if the hostname is an IP address, False otherwise. 

447 """ 

448 if isinstance(hostname, bytes): 

449 # IDN A-label bytes are ASCII compatible. 

450 hostname = hostname.decode("ascii") 

451 return bool(_IPV4_RE.match(hostname) or _BRACELESS_IPV6_ADDRZ_RE.match(hostname)) 

452 

453 

454def _is_key_file_encrypted(key_file: str) -> bool: 

455 """Detects if a key file is encrypted or not.""" 

456 with open(key_file) as f: 

457 for line in f: 

458 # Look for Proc-Type: 4,ENCRYPTED 

459 if "ENCRYPTED" in line: 

460 return True 

461 

462 return False 

463 

464 

465def _ssl_wrap_socket_impl( 

466 sock: socket.socket, 

467 ssl_context: ssl.SSLContext, 

468 tls_in_tls: bool, 

469 server_hostname: str | None = None, 

470) -> ssl.SSLSocket | SSLTransportType: 

471 if tls_in_tls: 

472 if not SSLTransport: 

473 # Import error, ssl is not available. 

474 raise ProxySchemeUnsupported( 

475 "TLS in TLS requires support for the 'ssl' module" 

476 ) 

477 

478 SSLTransport._validate_ssl_context_for_tls_in_tls(ssl_context) 

479 return SSLTransport(sock, ssl_context, server_hostname) 

480 

481 return ssl_context.wrap_socket(sock, server_hostname=server_hostname)