| 449 | raise |
| 450 | |
| 451 | def _get_ssl_options( |
| 452 | self, scheme: str |
| 453 | ) -> Union[None, Dict[str, Any], ssl.SSLContext]: |
| 454 | if scheme == "https": |
| 455 | if self.request.ssl_options is not None: |
| 456 | return self.request.ssl_options |
| 457 | # If we are using the defaults, don't construct a |
| 458 | # new SSLContext. |
| 459 | if ( |
| 460 | self.request.validate_cert |
| 461 | and self.request.ca_certs is None |
| 462 | and self.request.client_cert is None |
| 463 | and self.request.client_key is None |
| 464 | ): |
| 465 | return _client_ssl_defaults |
| 466 | ssl_ctx = ssl.create_default_context( |
| 467 | ssl.Purpose.SERVER_AUTH, cafile=self.request.ca_certs |
| 468 | ) |
| 469 | if not self.request.validate_cert: |
| 470 | ssl_ctx.check_hostname = False |
| 471 | ssl_ctx.verify_mode = ssl.CERT_NONE |
| 472 | if self.request.client_cert is not None: |
| 473 | ssl_ctx.load_cert_chain( |
| 474 | self.request.client_cert, self.request.client_key |
| 475 | ) |
| 476 | if hasattr(ssl, "OP_NO_COMPRESSION"): |
| 477 | # See netutil.ssl_options_to_context |
| 478 | ssl_ctx.options |= ssl.OP_NO_COMPRESSION |
| 479 | return ssl_ctx |
| 480 | return None |
| 481 | |
| 482 | def _on_timeout(self, info: Optional[str] = None) -> None: |
| 483 | """Timeout callback of _HTTPConnection instance. |