Prepare SSL certificates for Redis cluster connection. Args: cert_chain: PEM-encoded certificate chain containing client cert + intermediate + CA cert. This is the full certificate chain that will be used to validate the server. Returns: dict: SSL co
(cert_chain: bool)
| 227 | |
| 228 | |
| 229 | def _prepare_ssl_certificates(cert_chain: bool) -> dict: |
| 230 | """ |
| 231 | Prepare SSL certificates for Redis cluster connection. |
| 232 | |
| 233 | Args: |
| 234 | cert_chain: PEM-encoded certificate chain containing client cert + intermediate + CA cert. |
| 235 | This is the full certificate chain that will be used to validate the server. |
| 236 | |
| 237 | Returns: |
| 238 | dict: SSL configuration kwargs for RedisCluster |
| 239 | """ |
| 240 | certs_config_path = os.environ.get("MTLS_CONFIG_PATH", None) |
| 241 | |
| 242 | if not cert_chain: |
| 243 | return { |
| 244 | "ssl_cert_reqs": "none", |
| 245 | "ssl_check_hostname": False, |
| 246 | } |
| 247 | |
| 248 | if not certs_config_path: |
| 249 | raise ValueError( |
| 250 | "MTLS enabled test is triggered but MTLS_CONFIG_PATH environment variable not set" |
| 251 | ) |
| 252 | |
| 253 | # The cert_chain contains the full chain (client cert + intermediate + root CA) |
| 254 | # Use it as CA data for validating the server's certificate |
| 255 | return { |
| 256 | "ssl_cert_reqs": "none", |
| 257 | "ssl_keyfile": os.path.join(certs_config_path, "client.key"), |
| 258 | "ssl_certfile": os.path.join(certs_config_path, "client.crt"), |
| 259 | } |
| 260 | |
| 261 | |
| 262 | @pytest.fixture() |
no test coverage detected