TLS on Patroni REST API

src/charm.py

@@ -410,6 +410,7 @@ def _patroni(self) -> Patroni:
             self._peer_members_ips,
             self._get_password(),
             self._replication_password,
+            bool(self.unit_peer_data.get("tls")),
         )
 
     @property
@@ -820,12 +821,10 @@ def push_tls_files_to_workload(self) -> None:
         self.update_config()
 
     def _restart(self, _) -> None:
-        """Restart PostgreSQL."""
-        try:
-            self._patroni.restart_postgresql()
-        except RetryError as e:
+        """Restart Patroni and PostgreSQL."""
+        if not self._patroni.restart_patroni():
             logger.exception("failed to restart PostgreSQL")
-            self.unit.status = BlockedStatus(f"failed to restart PostgreSQL with error {e}")
+            self.unit.status = BlockedStatus("failed to restart Patroni and PostgreSQL")
 
     def update_config(self) -> None:
         """Updates Patroni config file based on the existence of the TLS files."""
@@ -838,6 +837,7 @@ def update_config(self) -> None:
 
         restart_postgresql = enable_tls != self.postgresql.is_tls_enabled()
         self._patroni.reload_patroni_configuration()
+        self.unit_peer_data.update({"tls": "enabled" if enable_tls else ""})
 
         # Restart PostgreSQL if TLS configuration has changed
         # (so the both old and new connections use the configuration).

src/cluster.py

@@ -13,6 +13,7 @@
 from charms.operator_libs_linux.v0.apt import DebianPackage
 from charms.operator_libs_linux.v1.systemd import (
     daemon_reload,
+    service_restart,
     service_running,
     service_start,
 )
@@ -28,7 +29,7 @@
     wait_fixed,
 )
 
-from constants import USER
+from constants import TLS_CA_FILE, USER
 
 logger = logging.getLogger(__name__)
 
@@ -62,6 +63,7 @@ def __init__(
         peers_ips: Set[str],
         superuser_password: str,
         replication_password: str,
+        tls_enabled: bool,
     ):
         """Initialize the Patroni class.
 
@@ -74,6 +76,7 @@ def __init__(
             planned_units: number of units planned for the cluster
             superuser_password: password for the operator user
             replication_password: password for the user used in the replication
+            tls_enabled: whether TLS is enabled
         """
         self.unit_ip = unit_ip
         self.storage_path = storage_path
@@ -83,6 +86,16 @@ def __init__(
         self.peers_ips = peers_ips
         self.superuser_password = superuser_password
         self.replication_password = replication_password
+        self.tls_enabled = tls_enabled
+        # Variable mapping to requests library verify parameter.
+        # The CA bundle file is used to validate the server certificate when
+        # TLS is enabled, otherwise True is set because it's the default value.
+        self.verify = f"{self.storage_path}/{TLS_CA_FILE}" if tls_enabled else True
+
+    @property
+    def _patroni_url(self) -> str:
+        """Patroni REST API URL."""
+        return f"{'https' if self.tls_enabled else 'http'}://{self.unit_ip}:8008"
 
     def bootstrap_cluster(self) -> bool:
         """Bootstrap a PostgreSQL cluster using Patroni."""
@@ -117,7 +130,7 @@ def _change_owner(self, path: str) -> None:
     def cluster_members(self) -> set:
         """Get the current cluster members."""
         # Request info from cluster endpoint (which returns all members of the cluster).
-        cluster_status = requests.get(f"http://{self.unit_ip}:8008/cluster")
+        cluster_status = requests.get(f"{self._patroni_url}/cluster", verify=self.verify)
         return set([member["name"] for member in cluster_status.json()["members"]])
 
     def _create_directory(self, path: str, mode: int) -> None:
@@ -150,7 +163,7 @@ def get_member_ip(self, member_name: str) -> str:
         """
         ip = None
         # Request info from cluster endpoint (which returns all members of the cluster).
-        cluster_status = requests.get(f"http://{self.unit_ip}:8008/cluster")
+        cluster_status = requests.get(f"{self._patroni_url}/cluster", verify=self.verify)
         for member in cluster_status.json()["members"]:
             if member["name"] == member_name:
                 ip = member["host"]
@@ -168,7 +181,7 @@ def get_primary(self, unit_name_pattern=False) -> str:
         """
         primary = None
         # Request info from cluster endpoint (which returns all members of the cluster).
-        cluster_status = requests.get(f"http://{self.unit_ip}:8008/cluster")
+        cluster_status = requests.get(f"{self._patroni_url}/cluster", verify=self.verify)
         for member in cluster_status.json()["members"]:
             if member["role"] == "leader":
                 primary = member["name"]
@@ -190,7 +203,9 @@ def are_all_members_ready(self) -> bool:
         try:
             for attempt in Retrying(stop=stop_after_delay(10), wait=wait_fixed(3)):
                 with attempt:
-                    cluster_status = requests.get(f"http://{self.unit_ip}:8008/cluster")
+                    cluster_status = requests.get(
+                        f"{self._patroni_url}/cluster", verify=self.verify
+                    )
         except RetryError:
             return False
 
@@ -212,7 +227,7 @@ def member_started(self) -> bool:
         try:
             for attempt in Retrying(stop=stop_after_delay(60), wait=wait_fixed(3)):
                 with attempt:
-                    r = requests.get(f"http://{self.unit_ip}:8008/health")
+                    r = requests.get(f"{self._patroni_url}/health", verify=self.verify)
         except RetryError:
             return False
 
@@ -300,8 +315,9 @@ def switchover(self) -> None:
             with attempt:
                 current_primary = self.get_primary()
                 r = requests.post(
-                    f"http://{self.unit_ip}:8008/switchover",
+                    f"{self._patroni_url}/switchover",
                     json={"leader": current_primary},
+                    verify=self.verify,
                 )
 
         # Check whether the switchover was unsuccessful.
@@ -348,9 +364,18 @@ def remove_raft_member(self, member_ip: str) -> None:
     @retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10))
     def reload_patroni_configuration(self):
         """Reload Patroni configuration after it was changed."""
-        requests.post(f"http://{self.unit_ip}:8008/reload")
+        requests.post(f"{self._patroni_url}/reload", verify=self.verify)
+
+    def restart_patroni(self) -> bool:
+        """Restart Patroni.
+
+        Returns:
+            Whether the service restarted successfully.
+        """
+        service_restart(PATRONI_SERVICE)
+        return service_running(PATRONI_SERVICE)
 
     @retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10))
     def restart_postgresql(self) -> None:
         """Restart PostgreSQL."""
-        requests.post(f"http://{self.unit_ip}:8008/restart")
+        requests.post(f"{self._patroni_url}/restart", verify=self.verify)

templates/patroni.yml.j2

@@ -17,6 +17,18 @@ log:
 restapi:
   listen: '{{ self_ip }}:8008'
   connect_address: '{{ self_ip }}:8008'
+  {%- if enable_tls %}
+  cafile: {{ conf_path }}/ca.pem
+  certfile: {{ conf_path }}/cert.pem
+  keyfile: {{ conf_path }}/key.pem
+  {%- endif %}
+
+{%- if enable_tls %}
+ctl:
+  cacert: {{ conf_path }}/ca.pem
+  certfile: {{ conf_path }}/cert.pem
+  keyfile: {{ conf_path }}/key.pem
+{%- endif %}
 
 raft:
   data_dir: {{ conf_path }}/raft

tests/integration/helpers.py

@@ -2,6 +2,7 @@
 # Copyright 2022 Canonical Ltd.
 # See LICENSE file for licensing details.
 import itertools
+import json
 import tempfile
 import zipfile
 from datetime import datetime
@@ -412,6 +413,32 @@ async def get_primary(ops_test: OpsTest, unit_name: str) -> str:
     return action.results["primary"]
 
 
+async def get_tls_ca(
+    ops_test: OpsTest,
+    unit_name: str,
+) -> str:
+    """Returns the TLS CA used by the unit.
+
+    Args:
+        ops_test: The ops test framework instance
+        unit_name: The name of the unit
+
+    Returns:
+        TLS CA or an empty string if there is no CA.
+    """
+    raw_data = (await ops_test.juju("show-unit", unit_name))[1]
+    if not raw_data:
+        raise ValueError(f"no unit info could be grabbed for {unit_name}")
+    data = yaml.safe_load(raw_data)
+    # Filter the data based on the relation name.
+    relation_data = [
+        v for v in data[unit_name]["relation-info"] if v["endpoint"] == "certificates"
+    ]
+    if len(relation_data) == 0:
+        return ""
+    return json.loads(relation_data[0]["application-data"]["certificates"])[0].get("ca")
+
+
 def get_unit_address(ops_test: OpsTest, unit_name: str) -> str:
     """Get unit IP address.
 
@@ -490,6 +517,47 @@ async def check_tls(ops_test: OpsTest, unit_name: str, enabled: bool) -> bool:
         return False
 
 
+async def check_tls_patroni_api(ops_test: OpsTest, unit_name: str, enabled: bool) -> bool:
+    """Returns whether TLS is enabled on Patroni REST API.
+
+    Args:
+        ops_test: The ops test framework instance.
+        unit_name: The name of the unit where Patroni is running.
+        enabled: check if TLS is enabled/disabled
+
+    Returns:
+        Whether TLS is enabled/disabled on Patroni REST API.
+    """
+    unit_address = get_unit_address(ops_test, unit_name)
+    tls_ca = await get_tls_ca(ops_test, unit_name)
+
+    # If there is no TLS CA in the relation, something is wrong in
+    # the relation between the TLS Certificates Operator and PostgreSQL.
+    if enabled and not tls_ca:
+        return False
+
+    try:
+        for attempt in Retrying(
+            stop=stop_after_attempt(10), wait=wait_exponential(multiplier=1, min=2, max=30)
+        ):
+            with attempt, tempfile.NamedTemporaryFile() as temp_ca_file:
+                # Write the TLS CA to a temporary file to use it in a request.
+                temp_ca_file.write(tls_ca.encode("utf-8"))
+                temp_ca_file.seek(0)
+
+                # The CA bundle file is used to validate the server certificate when
+                # TLS is enabled, otherwise True is set because it's the default value
+                # for the verify parameter.
+                health_info = requests.get(
+                    f"{'https' if enabled else 'http'}://{unit_address}:8008/health",
+                    verify=temp_ca_file.name if enabled else True,
+                )
+                return health_info.status_code == 200
+    except RetryError:
+        return False
+    return False
+
+
 async def scale_application(ops_test: OpsTest, application_name: str, count: int) -> None:
     """Scale a given application to a specific unit count.
 

tests/integration/test_tls.py

@@ -5,7 +5,11 @@
 from pytest_operator.plugin import OpsTest
 
 from tests.helpers import METADATA
-from tests.integration.helpers import DATABASE_APP_NAME, check_tls
+from tests.integration.helpers import (
+    DATABASE_APP_NAME,
+    check_tls,
+    check_tls_patroni_api,
+)
 
 APP_NAME = METADATA["name"]
 TLS_CERTIFICATES_APP_NAME = "tls-certificates-operator"
@@ -43,6 +47,7 @@ async def test_tls_enabled(ops_test: OpsTest) -> None:
         # Wait for all units enabling TLS.
         for unit in ops_test.model.applications[DATABASE_APP_NAME].units:
             assert await check_tls(ops_test, unit.name, enabled=True)
+            assert await check_tls_patroni_api(ops_test, unit.name, enabled=True)
 
         # Remove the relation.
         await ops_test.model.applications[DATABASE_APP_NAME].remove_relation(
@@ -53,3 +58,4 @@ async def test_tls_enabled(ops_test: OpsTest) -> None:
         # Wait for all units disabling TLS.
         for unit in ops_test.model.applications[DATABASE_APP_NAME].units:
             assert await check_tls(ops_test, unit.name, enabled=False)
+            assert await check_tls_patroni_api(ops_test, unit.name, enabled=False)

tests/unit/test_cluster.py

@@ -27,6 +27,7 @@ def setUp(self):
             peers_ips,
             "fake-superuser-password",
             "fake-replication-password",
+            False,
         )
 
     @patch("charms.operator_libs_linux.v0.apt.DebianPackage.from_system")