| 380 | self.close() |
| 381 | |
| 382 | def _create_ssl_ctx(self, sslp): |
| 383 | if isinstance(sslp, ssl.SSLContext): |
| 384 | return sslp |
| 385 | ca = sslp.get("ca") |
| 386 | capath = sslp.get("capath") |
| 387 | hasnoca = ca is None and capath is None |
| 388 | ctx = ssl.create_default_context(cafile=ca, capath=capath) |
| 389 | |
| 390 | # Python 3.13 enables VERIFY_X509_STRICT by default. |
| 391 | # But self signed certificates that are generated by MySQL automatically |
| 392 | # doesn't pass the verification. |
| 393 | ctx.verify_flags &= ~ssl.VERIFY_X509_STRICT |
| 394 | |
| 395 | ctx.check_hostname = not hasnoca and sslp.get("check_hostname", True) |
| 396 | verify_mode_value = sslp.get("verify_mode") |
| 397 | if verify_mode_value is None: |
| 398 | ctx.verify_mode = ssl.CERT_NONE if hasnoca else ssl.CERT_REQUIRED |
| 399 | elif isinstance(verify_mode_value, bool): |
| 400 | ctx.verify_mode = ssl.CERT_REQUIRED if verify_mode_value else ssl.CERT_NONE |
| 401 | else: |
| 402 | if isinstance(verify_mode_value, str): |
| 403 | verify_mode_value = verify_mode_value.lower() |
| 404 | if verify_mode_value in ("none", "0", "false", "no"): |
| 405 | ctx.verify_mode = ssl.CERT_NONE |
| 406 | elif verify_mode_value == "optional": |
| 407 | ctx.verify_mode = ssl.CERT_OPTIONAL |
| 408 | elif verify_mode_value in ("required", "1", "true", "yes"): |
| 409 | ctx.verify_mode = ssl.CERT_REQUIRED |
| 410 | else: |
| 411 | ctx.verify_mode = ssl.CERT_NONE if hasnoca else ssl.CERT_REQUIRED |
| 412 | if "cert" in sslp: |
| 413 | ctx.load_cert_chain( |
| 414 | sslp["cert"], keyfile=sslp.get("key"), password=sslp.get("password") |
| 415 | ) |
| 416 | if "cipher" in sslp: |
| 417 | ctx.set_ciphers(sslp["cipher"]) |
| 418 | ctx.options |= ssl.OP_NO_SSLv2 |
| 419 | ctx.options |= ssl.OP_NO_SSLv3 |
| 420 | return ctx |
| 421 | |
| 422 | def close(self): |
| 423 | """ |