Parse ssl options.
(
options: Mapping[str, Any], is_sync: bool
)
| 85 | |
| 86 | |
| 87 | def _parse_ssl_options( |
| 88 | options: Mapping[str, Any], is_sync: bool |
| 89 | ) -> tuple[Optional[SSLContext], bool]: |
| 90 | """Parse ssl options.""" |
| 91 | use_tls = options.get("tls") |
| 92 | if use_tls is not None: |
| 93 | validate_boolean("tls", use_tls) |
| 94 | |
| 95 | certfile = options.get("tlscertificatekeyfile") |
| 96 | passphrase = options.get("tlscertificatekeyfilepassword") |
| 97 | ca_certs = options.get("tlscafile") |
| 98 | crlfile = options.get("tlscrlfile") |
| 99 | allow_invalid_certificates = options.get("tlsallowinvalidcertificates", False) |
| 100 | allow_invalid_hostnames = options.get("tlsallowinvalidhostnames", False) |
| 101 | disable_ocsp_endpoint_check = options.get("tlsdisableocspendpointcheck", False) |
| 102 | |
| 103 | enabled_tls_opts = [] |
| 104 | for opt in ( |
| 105 | "tlscertificatekeyfile", |
| 106 | "tlscertificatekeyfilepassword", |
| 107 | "tlscafile", |
| 108 | "tlscrlfile", |
| 109 | ): |
| 110 | # Any non-null value of these options implies tls=True. |
| 111 | if opt in options and options[opt]: |
| 112 | enabled_tls_opts.append(opt) |
| 113 | for opt in ( |
| 114 | "tlsallowinvalidcertificates", |
| 115 | "tlsallowinvalidhostnames", |
| 116 | "tlsdisableocspendpointcheck", |
| 117 | ): |
| 118 | # A value of False for these options implies tls=True. |
| 119 | if opt in options and not options[opt]: |
| 120 | enabled_tls_opts.append(opt) |
| 121 | |
| 122 | if enabled_tls_opts: |
| 123 | if use_tls is None: |
| 124 | # Implicitly enable TLS when one of the tls* options is set. |
| 125 | use_tls = True |
| 126 | elif not use_tls: |
| 127 | # Error since tls is explicitly disabled but a tls option is set. |
| 128 | raise ConfigurationError( |
| 129 | "TLS has not been enabled but the " |
| 130 | "following tls parameters have been set: " |
| 131 | "%s. Please set `tls=True` or remove." % ", ".join(enabled_tls_opts) |
| 132 | ) |
| 133 | |
| 134 | if use_tls: |
| 135 | ctx = get_ssl_context( |
| 136 | certfile, |
| 137 | passphrase, |
| 138 | ca_certs, |
| 139 | crlfile, |
| 140 | allow_invalid_certificates, |
| 141 | allow_invalid_hostnames, |
| 142 | disable_ocsp_endpoint_check, |
| 143 | is_sync, |
| 144 | ) |
no test coverage detected