REST API would not reload SSL certificate upon receiving an SIGHUP (#2722)

Revert to using `ssl._ssl._test_decode_cert`

A change has been included as part of Patroni 3.0.3 release: use
public functions instead of `ssl._ssl._test_decode_cert` to get
serial number of certificates.

There was a slight bug in that implementation: it was only loading
the certificates through `load_verify_locations`, but was missing
to get the certificates through `get_ca_certs`. As a consequence
Patroni was not able anymore to reload REST API cert on SIGHUP.

An attempt to fix that issue was made through commit
`20f578f09f3aa604e5288710d4fd4e611152ed5f`. However, even with the
correct call of `get_ca_certs`, it was detected a corner case where
`load_verify_locations` would skip loading a certificate: if it was
issued with `CA:FALSE`. That essentially means the implementation is
still buggy in that situation. See [CPython](https://github.com/python/cpython/blob/c283a0cff5603540f06d9017e484b3602cc62e7c/Modules/_ssl.c#L4618C14-L4619)
for the underlying problem.

In order to get back a functional implementation again we are reverting
the code to use the private function `ssl._ssl._test_decode_cert`.

We can later study a possible more elegant alternative for solving this,
if any.

---------

Signed-off-by: Israel Barth Rubio <[email protected]>

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]:

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())