diff --git a/patroni/api.py b/patroni/api.py index f461811b..7be2a049 100644 --- a/patroni/api.py +++ b/patroni/api.py @@ -1541,11 +1541,11 @@ class RestApiServer(ThreadingMixIn, HTTPServer, Thread): if self.__ssl_options.get('certfile'): import ssl try: - ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT) - crts = ctx.load_verify_locations(self.__ssl_options['certfile']) - if crts: - return crts[0].get('serialNumber') - except Exception as e: + crt: Dict[str, Any] = ssl._ssl._test_decode_cert(self.__ssl_options['certfile']) # pyright: ignore + if TYPE_CHECKING: # pragma: no cover + assert isinstance(crt, dict) + return crt.get('serialNumber') + except ssl.SSLError as e: logger.error('Failed to get serial number from certificate %s: %r', self.__ssl_options['certfile'], e) def reload_local_certificate(self) -> Optional[bool]: diff --git a/tests/test_api.py b/tests/test_api.py index 166d3eb1..25342dca 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -180,7 +180,6 @@ class MockRestApiServer(RestApiServer): @patch('ssl.SSLContext.load_cert_chain', Mock()) @patch('ssl.SSLContext.wrap_socket', Mock(return_value=0)) -@patch('ssl.SSLContext.load_verify_locations', Mock(return_value=[Mock()])) @patch.object(HTTPServer, '__init__', Mock()) class TestRestApiHandler(unittest.TestCase): @@ -589,7 +588,6 @@ class TestRestApiServer(unittest.TestCase): @patch('ssl.SSLContext.load_cert_chain', Mock()) @patch('ssl.SSLContext.set_ciphers', Mock()) @patch('ssl.SSLContext.wrap_socket', Mock(return_value=0)) - @patch('ssl.SSLContext.load_verify_locations', Mock(return_value=[Mock()])) @patch.object(HTTPServer, '__init__', Mock()) def setUp(self): self.srv = MockRestApiServer(Mock(), '', {'listen': '*:8008', 'certfile': 'a', 'verify_client': 'required', @@ -652,10 +650,9 @@ class TestRestApiServer(unittest.TestCase): mock_get_request.return_value = (self.__create_socket(), ('127.0.0.1', 55555)) self.srv._handle_request_noblock() - @patch('ssl.SSLContext.load_verify_locations', Mock(return_value=[Mock()])) + @patch('ssl._ssl._test_decode_cert', Mock()) def test_reload_local_certificate(self): self.assertTrue(self.srv.reload_local_certificate()) - @patch('ssl.SSLContext.load_verify_locations', Mock(side_effect=Exception)) def test_get_certificate_serial_number(self): self.assertIsNone(self.srv.get_certificate_serial_number())