(
ssl_context: ssl.SSLContext,
log_obj: Any | None = None,
)
| 10 | |
| 11 | |
| 12 | def _try_patch_aiohttp_ssl_context( |
| 13 | ssl_context: ssl.SSLContext, |
| 14 | log_obj: Any | None = None, |
| 15 | ) -> bool: |
| 16 | log = log_obj or logger |
| 17 | attr_name = "_SSL_CONTEXT_VERIFIED" |
| 18 | |
| 19 | if not hasattr(aiohttp_connector, attr_name): |
| 20 | log.warning( |
| 21 | "aiohttp connector does not expose _SSL_CONTEXT_VERIFIED; skipped patch.", |
| 22 | ) |
| 23 | return False |
| 24 | |
| 25 | current_value = getattr(aiohttp_connector, attr_name, None) |
| 26 | if current_value is not None and not isinstance(current_value, ssl.SSLContext): |
| 27 | log.warning( |
| 28 | "aiohttp connector exposes _SSL_CONTEXT_VERIFIED with unexpected type; skipped patch.", |
| 29 | ) |
| 30 | return False |
| 31 | |
| 32 | setattr(aiohttp_connector, attr_name, ssl_context) |
| 33 | return True |
| 34 | |
| 35 | |
| 36 | def configure_runtime_ca_bundle(log_obj: Any | None = None) -> bool: |
no outgoing calls
no test coverage detected