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

Shortcuts on this page

r m x   toggle line displays

j k   next/prev highlighted chunk

0   (zero) top of page

1   (one) first highlighted chunk

172 statements  

1from __future__ import annotations 

2 

3import hashlib 

4import hmac 

5import os 

6import socket 

7import sys 

8import typing 

9import warnings 

10from binascii import unhexlify 

11 

12from ..exceptions import ProxySchemeUnsupported, SSLError 

13from .url import _BRACELESS_IPV6_ADDRZ_RE, _IPV4_RE 

14 

15SSLContext = None 

16SSLTransport = None 

17HAS_NEVER_CHECK_COMMON_NAME = False 

18IS_PYOPENSSL = False 

19ALPN_PROTOCOLS = ["http/1.1"] 

20 

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

22 

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

24HASHFUNC_MAP = { 

25 length: getattr(hashlib, algorithm, None) 

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

27} 

28 

29 

30def _is_bpo_43522_fixed( 

31 implementation_name: str, 

32 version_info: _TYPE_VERSION_INFO, 

33 pypy_version_info: _TYPE_VERSION_INFO | None, 

34) -> bool: 

35 """Return True for CPython 3.9.3+ or 3.10+ and PyPy 7.3.8+ where 

36 setting SSLContext.hostname_checks_common_name to False works. 

37 

38 Outside of CPython and PyPy we don't know which implementations work 

39 or not so we conservatively use our hostname matching as we know that works 

40 on all implementations. 

41 

42 https://github.com/urllib3/urllib3/issues/2192#issuecomment-821832963 

43 https://foss.heptapod.net/pypy/pypy/-/issues/3539 

44 """ 

45 if implementation_name == "pypy": 

46 # https://foss.heptapod.net/pypy/pypy/-/issues/3129 

47 return pypy_version_info >= (7, 3, 8) # type: ignore[operator] 

48 elif implementation_name == "cpython": 

49 major_minor = version_info[:2] 

50 micro = version_info[2] 

51 return (major_minor == (3, 9) and micro >= 3) or major_minor >= (3, 10) 

52 else: # Defensive: 

53 return False 

54 

55 

56def _is_has_never_check_common_name_reliable( 

57 openssl_version: str, 

58 openssl_version_number: int, 

59 implementation_name: str, 

60 version_info: _TYPE_VERSION_INFO, 

61 pypy_version_info: _TYPE_VERSION_INFO | None, 

62) -> bool: 

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

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

65 is_openssl = openssl_version.startswith("OpenSSL ") 

66 # Before fixing OpenSSL issue #14579, the SSL_new() API was not copying hostflags 

67 # like X509_CHECK_FLAG_NEVER_CHECK_SUBJECT, which tripped up CPython. 

68 # https://github.com/openssl/openssl/issues/14579 

69 # This was released in OpenSSL 1.1.1l+ (>=0x101010cf) 

70 is_openssl_issue_14579_fixed = openssl_version_number >= 0x101010CF 

71 

72 return is_openssl and ( 

73 is_openssl_issue_14579_fixed 

74 or _is_bpo_43522_fixed(implementation_name, version_info, pypy_version_info) 

75 ) 

76 

77 

78if typing.TYPE_CHECKING: 

79 from ssl import VerifyMode 

80 from typing import TypedDict 

81 

82 from .ssltransport import SSLTransport as SSLTransportType 

83 

84 class _TYPE_PEER_CERT_RET_DICT(TypedDict, total=False): 

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

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

87 serialNumber: str 

88 

89 

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

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

92 

93try: # Do we have ssl at all? 

94 import ssl 

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

96 CERT_REQUIRED, 

97 HAS_NEVER_CHECK_COMMON_NAME, 

98 OP_NO_COMPRESSION, 

99 OP_NO_TICKET, 

100 OPENSSL_VERSION, 

101 OPENSSL_VERSION_NUMBER, 

102 PROTOCOL_TLS, 

103 PROTOCOL_TLS_CLIENT, 

104 OP_NO_SSLv2, 

105 OP_NO_SSLv3, 

106 SSLContext, 

107 TLSVersion, 

108 ) 

109 

110 PROTOCOL_SSLv23 = PROTOCOL_TLS 

111 

112 # Setting SSLContext.hostname_checks_common_name = False didn't work before CPython 

113 # 3.9.3, and 3.10 (but OK on PyPy) or OpenSSL 1.1.1l+ 

114 if HAS_NEVER_CHECK_COMMON_NAME and not _is_has_never_check_common_name_reliable( 

115 OPENSSL_VERSION, 

116 OPENSSL_VERSION_NUMBER, 

117 sys.implementation.name, 

118 sys.version_info, 

119 sys.pypy_version_info if sys.implementation.name == "pypy" else None, # type: ignore[attr-defined] 

120 ): 

121 HAS_NEVER_CHECK_COMMON_NAME = False 

122 

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

124 # removed in future 'ssl' module implementations. 

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

126 try: 

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

128 TLSVersion, attr 

129 ) 

130 except AttributeError: # Defensive: 

131 continue 

132 

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

134except ImportError: 

135 OP_NO_COMPRESSION = 0x20000 # type: ignore[assignment] 

136 OP_NO_TICKET = 0x4000 # type: ignore[assignment] 

137 OP_NO_SSLv2 = 0x1000000 # type: ignore[assignment] 

138 OP_NO_SSLv3 = 0x2000000 # type: ignore[assignment] 

139 PROTOCOL_SSLv23 = PROTOCOL_TLS = 2 # type: ignore[assignment] 

140 PROTOCOL_TLS_CLIENT = 16 # type: ignore[assignment] 

141 

142 

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

144 

145 

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

147 """ 

148 Checks if given fingerprint matches the supplied certificate. 

149 

150 :param cert: 

151 Certificate as bytes object. 

152 :param fingerprint: 

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

154 """ 

155 

156 if cert is None: 

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

158 

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

160 digest_length = len(fingerprint) 

161 if digest_length not in HASHFUNC_MAP: 

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

163 hashfunc = HASHFUNC_MAP.get(digest_length) 

164 if hashfunc is None: 

165 raise SSLError( 

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

167 ) 

168 

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

170 fingerprint_bytes = unhexlify(fingerprint.encode()) 

171 

172 cert_digest = hashfunc(cert).digest() 

173 

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

175 raise SSLError( 

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

177 ) 

178 

179 

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

181 """ 

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

183 the wrap_socket function/method from the ssl module. 

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

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

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

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

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

189 constant which can directly be passed to wrap_socket. 

190 """ 

191 if candidate is None: 

192 return CERT_REQUIRED 

193 

194 if isinstance(candidate, str): 

195 res = getattr(ssl, candidate, None) 

196 if res is None: 

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

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

199 

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

201 

202 

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

204 """ 

205 like resolve_cert_reqs 

206 """ 

207 if candidate is None: 

208 return PROTOCOL_TLS 

209 

210 if isinstance(candidate, str): 

211 res = getattr(ssl, candidate, None) 

212 if res is None: 

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

214 return typing.cast(int, res) 

215 

216 return candidate 

217 

218 

219def create_urllib3_context( 

220 ssl_version: int | None = None, 

221 cert_reqs: int | None = None, 

222 options: int | None = None, 

223 ciphers: str | None = None, 

224 ssl_minimum_version: int | None = None, 

225 ssl_maximum_version: int | None = None, 

226) -> ssl.SSLContext: 

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

228 

229 :param ssl_version: 

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

231 PROTOCOL_SSLv23 which will negotiate the highest protocol that both 

232 the server and your installation of OpenSSL support. 

233 

234 This parameter is deprecated instead use 'ssl_minimum_version'. 

235 :param ssl_minimum_version: 

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

237 :param ssl_maximum_version: 

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

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

240 default value. 

241 :param cert_reqs: 

242 Whether to require the certificate verification. This defaults to 

243 ``ssl.CERT_REQUIRED``. 

244 :param options: 

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

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

247 :param ciphers: 

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

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

250 :returns: 

251 Constructed SSLContext object with specified options 

252 :rtype: SSLContext 

253 """ 

254 if SSLContext is None: 

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

256 

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

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

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

260 # to avoid conflicts. 

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

262 raise ValueError( 

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

264 "'ssl_minimum_version' or 'ssl_maximum_version'" 

265 ) 

266 

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

268 else: 

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

270 ssl_minimum_version = _SSL_VERSION_TO_TLS_VERSION.get( 

271 ssl_version, TLSVersion.MINIMUM_SUPPORTED 

272 ) 

273 ssl_maximum_version = _SSL_VERSION_TO_TLS_VERSION.get( 

274 ssl_version, TLSVersion.MAXIMUM_SUPPORTED 

275 ) 

276 

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

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

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

280 warnings.warn( 

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

282 "removed in urllib3 v2.1.0. Instead use 'ssl_minimum_version'", 

283 category=DeprecationWarning, 

284 stacklevel=2, 

285 ) 

286 

287 # PROTOCOL_TLS is deprecated in Python 3.10 so we always use PROTOCOL_TLS_CLIENT 

288 context = SSLContext(PROTOCOL_TLS_CLIENT) 

289 

290 if ssl_minimum_version is not None: 

291 context.minimum_version = ssl_minimum_version 

292 else: # Python <3.10 defaults to 'MINIMUM_SUPPORTED' so explicitly set TLSv1.2 here 

293 context.minimum_version = TLSVersion.TLSv1_2 

294 

295 if ssl_maximum_version is not None: 

296 context.maximum_version = ssl_maximum_version 

297 

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

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

300 if ciphers: 

301 context.set_ciphers(ciphers) 

302 

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

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

305 

306 if options is None: 

307 options = 0 

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

309 options |= OP_NO_SSLv2 

310 # SSLv3 has several problems and is now dangerous 

311 options |= OP_NO_SSLv3 

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

313 # (issue #309) 

314 options |= OP_NO_COMPRESSION 

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

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

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

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

319 options |= OP_NO_TICKET 

320 

321 context.options |= options 

322 

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

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

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

326 # an SSLContext created by pyOpenSSL. 

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

328 context.post_handshake_auth = True 

329 

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

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

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

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

334 # 'ssl.match_hostname()' implementation. 

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

336 context.verify_mode = cert_reqs 

337 context.check_hostname = True 

338 else: 

339 context.check_hostname = False 

340 context.verify_mode = cert_reqs 

341 

342 try: 

343 context.hostname_checks_common_name = False 

344 except AttributeError: # Defensive: for CPython < 3.9.3; for PyPy < 7.3.8 

345 pass 

346 

347 sslkeylogfile = os.environ.get("SSLKEYLOGFILE") 

348 if sslkeylogfile: 

349 context.keylog_filename = sslkeylogfile 

350 

351 return context 

352 

353 

354@typing.overload 

355def ssl_wrap_socket( 

356 sock: socket.socket, 

357 keyfile: str | None = ..., 

358 certfile: str | None = ..., 

359 cert_reqs: int | None = ..., 

360 ca_certs: str | None = ..., 

361 server_hostname: str | None = ..., 

362 ssl_version: int | None = ..., 

363 ciphers: str | None = ..., 

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

365 ca_cert_dir: str | None = ..., 

366 key_password: str | None = ..., 

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

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

369) -> ssl.SSLSocket: ... 

370 

371 

372@typing.overload 

373def ssl_wrap_socket( 

374 sock: socket.socket, 

375 keyfile: str | None = ..., 

376 certfile: str | None = ..., 

377 cert_reqs: int | None = ..., 

378 ca_certs: str | None = ..., 

379 server_hostname: str | None = ..., 

380 ssl_version: int | None = ..., 

381 ciphers: str | None = ..., 

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

383 ca_cert_dir: str | None = ..., 

384 key_password: str | None = ..., 

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

386 tls_in_tls: bool = ..., 

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

388 

389 

390def ssl_wrap_socket( 

391 sock: socket.socket, 

392 keyfile: str | None = None, 

393 certfile: str | None = None, 

394 cert_reqs: int | None = None, 

395 ca_certs: str | None = None, 

396 server_hostname: str | None = None, 

397 ssl_version: int | None = None, 

398 ciphers: str | None = None, 

399 ssl_context: ssl.SSLContext | None = None, 

400 ca_cert_dir: str | None = None, 

401 key_password: str | None = None, 

402 ca_cert_data: None | str | bytes = None, 

403 tls_in_tls: bool = False, 

404) -> ssl.SSLSocket | SSLTransportType: 

405 """ 

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

407 ca_cert_dir have the same meaning as they do when using 

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

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

410 

411 :param server_hostname: 

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

413 :param ssl_context: 

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

415 be created using :func:`create_urllib3_context`. 

416 :param ciphers: 

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

418 :param ca_cert_dir: 

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

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

421 SSLContext.load_verify_locations(). 

422 :param key_password: 

423 Optional password if the keyfile is encrypted. 

424 :param ca_cert_data: 

425 Optional string containing CA certificates in PEM format suitable for 

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

427 :param tls_in_tls: 

428 Use SSLTransport to wrap the existing socket. 

429 """ 

430 context = ssl_context 

431 if context is None: 

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

433 # We should consider deprecating and removing this code. 

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

435 

436 if ca_certs or ca_cert_dir or ca_cert_data: 

437 try: 

438 context.load_verify_locations(ca_certs, ca_cert_dir, ca_cert_data) 

439 except OSError as e: 

440 raise SSLError(e) from e 

441 

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

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

444 context.load_default_certs() 

445 

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

447 # keyfile being encrypted and OpenSSL asking for the 

448 # passphrase via the terminal and instead error out. 

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

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

451 

452 if certfile: 

453 if key_password is None: 

454 context.load_cert_chain(certfile, keyfile) 

455 else: 

456 context.load_cert_chain(certfile, keyfile, key_password) 

457 

458 context.set_alpn_protocols(ALPN_PROTOCOLS) 

459 

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

461 return ssl_sock 

462 

463 

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

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

466 Also detects IPv6 addresses with Zone IDs. 

467 

468 :param str hostname: Hostname to examine. 

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

470 """ 

471 if isinstance(hostname, bytes): 

472 # IDN A-label bytes are ASCII compatible. 

473 hostname = hostname.decode("ascii") 

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

475 

476 

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

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

479 with open(key_file) as f: 

480 for line in f: 

481 # Look for Proc-Type: 4,ENCRYPTED 

482 if "ENCRYPTED" in line: 

483 return True 

484 

485 return False 

486 

487 

488def _ssl_wrap_socket_impl( 

489 sock: socket.socket, 

490 ssl_context: ssl.SSLContext, 

491 tls_in_tls: bool, 

492 server_hostname: str | None = None, 

493) -> ssl.SSLSocket | SSLTransportType: 

494 if tls_in_tls: 

495 if not SSLTransport: 

496 # Import error, ssl is not available. 

497 raise ProxySchemeUnsupported( 

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

499 ) 

500 

501 SSLTransport._validate_ssl_context_for_tls_in_tls(ssl_context) 

502 return SSLTransport(sock, ssl_context, server_hostname) 

503 

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