[DPE-8900] fix: harden pg_hba replication rules and ensure dynamic IP updates (#1506)

* feat: harden pg_hba replication rules and ensure dynamic IP updates

- Use /32 CIDR masks instead of /0 for peer replication connections
- Add explicit self_ip/32 replication entry for local connections
- Convert Patroni.peers_ips to dynamic property for always-fresh IPs
- Update async replication data on IP changes and peer relation events

This reduces the attack surface by restricting replication access to
specific IPs and ensures configuration stays synchronized when cluster
member IPs change dynamically.

Signed-off-by: Marcelo Henrique Neppel <marcelo.neppel@canonical.com>

* test(ha): add retry logic for sync-standby verification

Add retry mechanism when checking sync-standby changes after failover,
as Patroni may need time to update cluster state after unit removal.

Signed-off-by: Marcelo Henrique Neppel <marcelo.neppel@canonical.com>

* fix: handle IP changes during async replication with Raft cleanup

Clear stale Raft data and re-render config when unit IP changes to
prevent Patroni startup failures. Add update-status safety net to
detect missed IP changes and prevent race condition in ip-to-remove
cleanup. Includes integration test for IP change during async
replication.

Signed-off-by: Marcelo Henrique Neppel <marcelo.neppel@canonical.com>

* refactor: use Raft member removal instead of destructive rmtree for IP changes

Replace shutil.rmtree(raft_path) with remove_raft_member() to cleanly
remove the old IP from the Raft cluster before stopping Patroni. This
avoids destructively deleting the entire Raft data directory and instead
lets PySyncObj handle the membership change gracefully.

Signed-off-by: Marcelo Henrique Neppel <marcelo.neppel@canonical.com>

* refactor(test): extract async replication fixtures to conftest

Move fast_forward helper and model fixtures (first_model, second_model)
from test_async_replication.py to conftest.py to enable reuse across
multiple async replication test files

Signed-off-by: Marcelo Henrique Neppel <marcelo.neppel@canonical.com>

* refactor: remove IP change detection from update-status hook

Remove safety net IP change detection from _on_update_status handler and associated unit test. IP changes are now handled exclusively through config-changed hook.

Signed-off-by: Marcelo Henrique Neppel <marcelo.neppel@canonical.com>

* test(async-replication): fix IP change test to target standby leader

Change the async replication IP change test to cut network from the standby leader instead of a random primary unit. The standby leader is the unit that connects to the primary cluster, making it the correct target for testing pg_hba rule updates. Remove obsolete helper functions and simplify test logic.

Signed-off-by: Marcelo Henrique Neppel <marcelo.neppel@canonical.com>

* test: remove amd64_only marker from IP change test

Signed-off-by: Marcelo Henrique Neppel <marcelo.neppel@canonical.com>

* test: refactor IP change test to use helper functions consistently

Replace subprocess calls with existing helper functions (get_ip_from_inside_the_unit,
get_unit_ip) that support multi-model operations. Remove manual DHCP release step and
use hostname -I consistently for IP retrieval to avoid Juju cache staleness issues.

Signed-off-by: Marcelo Henrique Neppel <marcelo.neppel@canonical.com>

* test: add spread task for async replication IP change test

Add spread test configuration for ha_tests/test_async_replication_ip_change.py
with Juju 2.9 variant and allure results artifact collection

Signed-off-by: Marcelo Henrique Neppel <marcelo.neppel@canonical.com>

* revert(cluster): restore peers_ips as constructor parameter

Revert dynamic property pattern from 036e2c7 and pass
peers_ips explicitly to Patroni. Remove update_config() call.

Signed-off-by: Marcelo Henrique Neppel <marcelo.neppel@canonical.com>

---------

Signed-off-by: Marcelo Henrique Neppel <marcelo.neppel@canonical.com>

Author: marceloneppel

src/charm.py

@@ -737,6 +737,9 @@ def _on_peer_relation_changed(self, event: HookEvent):
             logger.error("Invalid configuration: %s", str(e))
             return
 
+        # Update the async replication data with the current unit IP.
+        self.async_replication.update_async_replication_data()
+
         # Should not override a blocked status
         if isinstance(self.unit.status, BlockedStatus):
             logger.debug("on_peer_relation_changed early exit: Unit in blocked status")
@@ -857,11 +860,28 @@ def _update_member_ip(self) -> bool:
             logger.info(f"ip changed from {stored_ip} to {current_ip}")
             self.unit_peer_data.update({"ip-to-remove": stored_ip})
             self.unit_peer_data.update({"ip": current_ip})
+            # Update the async replication data with the new unit IP.
+            self.async_replication.update_async_replication_data()
+            # Remove the old IP from the Raft cluster while Patroni is still
+            # running. When Patroni restarts with the new self_addr, PySyncObj
+            # will create fresh journal files for the new IP — the orphaned
+            # old-IP files in the raft data_dir are harmless.
+            try:
+                self._patroni.remove_raft_member(stored_ip)
+            except RemoveRaftMemberFailedError:
+                logger.warning(
+                    "Failed to remove old IP %s from Raft cluster during IP change", stored_ip
+                )
             self._patroni.stop_patroni()
             self._update_certificate()
             return True
         else:
-            self.unit_peer_data.update({"ip-to-remove": ""})
+            # Only clear ip-to-remove after the leader has processed it
+            # (removed the old IP from members_ips). This prevents a race
+            # where the leader sees an empty ip-to-remove and skips cleanup.
+            ip_to_remove = self.unit_peer_data.get("ip-to-remove")
+            if not ip_to_remove or ip_to_remove not in self.members_ips:
+                self.unit_peer_data.update({"ip-to-remove": ""})
             return False
 
     def _add_members(self, event):
@@ -1331,6 +1351,8 @@ def _on_start(self, event: StartEvent) -> None:
             return
 
         self.unit_peer_data.update({"ip": self.get_hostname_by_unit(None)})
+        # Update the async replication data with the current unit IP.
+        self.async_replication.update_async_replication_data()
 
         self.unit.set_workload_version(self._patroni.get_postgresql_version())
 

templates/patroni.yml.j2

@@ -176,12 +176,13 @@ postgresql:
     {%- endfor %}
     {%- endif %}
     - {{ 'hostssl' if enable_tls else 'host' }} replication replication 127.0.0.1/32 md5
+    - {{ 'hostssl' if enable_tls else 'host' }} replication replication {{ self_ip }}/32 md5
     # Allow replications connections from other cluster members.
     {%- for endpoint in extra_replication_endpoints %}
     - {{ 'hostssl' if enable_tls else 'host' }} replication replication {{ endpoint }}/32 md5
     {%- endfor %}
     {%- for peer_ip in peers_ips %}
-    - {{ 'hostssl' if enable_tls else 'host' }}     replication    replication    {{ peer_ip }}/0    md5
+    - {{ 'hostssl' if enable_tls else 'host' }}     replication    replication    {{ peer_ip }}/32    md5
     {% endfor %}
   pg_ident:
   - operator snap_daemon backup

tests/integration/ha_tests/conftest.py

@@ -1,12 +1,17 @@
 #!/usr/bin/env python3
 # Copyright 2022 Canonical Ltd.
 # See LICENSE file for licensing details.
+import contextlib
+import logging
+import subprocess
 from asyncio import gather
 
 import pytest as pytest
+from juju.model import Model
 from pytest_operator.plugin import OpsTest
 from tenacity import Retrying, stop_after_delay, wait_fixed
 
+from .. import architecture
 from ..helpers import get_password, run_command_on_unit
 from .helpers import (
     APPLICATION_NAME,
@@ -20,6 +25,48 @@
     update_restart_condition,
 )
 
+logger = logging.getLogger(__name__)
+
+
+@contextlib.asynccontextmanager
+async def fast_forward(model: Model, fast_interval: str = "10s", slow_interval: str | None = None):
+    """Adaptation of OpsTest.fast_forward to work with different models."""
+    update_interval_key = "update-status-hook-interval"
+    interval_after = (
+        slow_interval if slow_interval else (await model.get_config())[update_interval_key]
+    )
+
+    await model.set_config({update_interval_key: fast_interval})
+    yield
+    await model.set_config({update_interval_key: interval_after})
+
+
+@pytest.fixture(scope="module")
+def first_model(ops_test: OpsTest) -> Model:
+    """Return the first model."""
+    first_model = ops_test.model
+    return first_model
+
+
+@pytest.fixture(scope="module")
+async def second_model(ops_test: OpsTest, first_model, request) -> Model:
+    """Create and return the second model."""
+    second_model_name = f"{first_model.info.name}-other"
+    if second_model_name not in await ops_test._controller.list_models():
+        await ops_test._controller.add_model(second_model_name)
+        subprocess.run(["juju", "switch", second_model_name], check=True)
+        subprocess.run(
+            ["juju", "set-model-constraints", f"arch={architecture.architecture}"], check=True
+        )
+        subprocess.run(["juju", "switch", first_model.info.name], check=True)
+    second_model = Model()
+    await second_model.connect(model_name=second_model_name)
+    yield second_model
+    if request.config.getoption("--keep-models"):
+        return
+    logger.info("Destroying second model")
+    await ops_test._controller.destroy_model(second_model_name, destroy_storage=True)
+
 
 @pytest.fixture()
 async def continuous_writes(ops_test: OpsTest) -> None:

tests/integration/ha_tests/helpers.py

@@ -415,8 +415,11 @@ async def get_controller_machine(ops_test: OpsTest) -> str:
     )
 
 
-async def get_ip_from_inside_the_unit(ops_test: OpsTest, unit_name: str) -> str:
-    command = f"exec --unit {unit_name} -- hostname -I"
+async def get_ip_from_inside_the_unit(
+    ops_test: OpsTest, unit_name: str, model: Model | None = None
+) -> str:
+    model = model or ops_test.model
+    command = f"exec -m {model.info.name} --unit {unit_name} -- hostname -I"
     return_code, stdout, stderr = await ops_test.juju(*command.split())
     if return_code != 0:
         raise ProcessError(
@@ -547,7 +550,7 @@ async def get_unit_ip(ops_test: OpsTest, unit_name: str, model: Model = None) ->
                 break
         return await instance_ip(ops_test, unit.machine.hostname)
     else:
-        return get_unit_address(ops_test, unit_name)
+        return get_unit_address(ops_test, unit_name, model)
 
 
 @retry(stop=stop_after_attempt(8), wait=wait_fixed(15), reraise=True)

tests/integration/ha_tests/test_async_replication.py

@@ -1,9 +1,7 @@
 #!/usr/bin/env python3
 # Copyright 2024 Canonical Ltd.
 # See LICENSE file for licensing details.
-import contextlib
 import logging
-import subprocess
 from asyncio import gather
 
 import psycopg2
@@ -12,7 +10,7 @@
 from pytest_operator.plugin import OpsTest
 from tenacity import Retrying, stop_after_delay, wait_fixed
 
-from .. import architecture, markers
+from .. import markers
 from ..helpers import (
     APPLICATION_NAME,
     DATABASE_APP_NAME,
@@ -23,6 +21,7 @@
     scale_application,
     wait_for_relation_removed_between,
 )
+from .conftest import fast_forward
 from .helpers import (
     app_name,
     are_writes_increasing,
@@ -43,46 +42,6 @@
 DATA_INTEGRATOR_APP_NAME = "data-integrator"
 
 
-@contextlib.asynccontextmanager
-async def fast_forward(model: Model, fast_interval: str = "10s", slow_interval: str | None = None):
-    """Adaptation of OpsTest.fast_forward to work with different models."""
-    update_interval_key = "update-status-hook-interval"
-    interval_after = (
-        slow_interval if slow_interval else (await model.get_config())[update_interval_key]
-    )
-
-    await model.set_config({update_interval_key: fast_interval})
-    yield
-    await model.set_config({update_interval_key: interval_after})
-
-
-@pytest.fixture(scope="module")
-def first_model(ops_test: OpsTest) -> Model:
-    """Return the first model."""
-    first_model = ops_test.model
-    return first_model
-
-
-@pytest.fixture(scope="module")
-async def second_model(ops_test: OpsTest, first_model, request) -> Model:
-    """Create and return the second model."""
-    second_model_name = f"{first_model.info.name}-other"
-    if second_model_name not in await ops_test._controller.list_models():
-        await ops_test._controller.add_model(second_model_name)
-        subprocess.run(["juju", "switch", second_model_name], check=True)
-        subprocess.run(
-            ["juju", "set-model-constraints", f"arch={architecture.architecture}"], check=True
-        )
-        subprocess.run(["juju", "switch", first_model.info.name], check=True)
-    second_model = Model()
-    await second_model.connect(model_name=second_model_name)
-    yield second_model
-    if request.config.getoption("--keep-models"):
-        return
-    logger.info("Destroying second model")
-    await ops_test._controller.destroy_model(second_model_name, destroy_storage=True)
-
-
 @pytest.fixture
 async def second_model_continuous_writes(second_model) -> None:
     """Cleans up continuous writes on the second model after a test run."""
@@ -490,10 +449,13 @@ async def test_async_replication_failover_in_main_cluster(
             ),
         )
 
-    # Check that the sync-standby unit is not the same as before.
-    new_sync_standby = await get_sync_standby(ops_test, first_model, DATABASE_APP_NAME)
-    logger.info(f"New sync-standby: {new_sync_standby}")
-    assert new_sync_standby != sync_standby, "Sync-standby is the same as before"
+    # Check that the sync-standby unit is not the same as before (retry because
+    # Patroni may take some time to update its cluster state after unit removal).
+    for attempt in Retrying(stop=stop_after_delay(300), wait=wait_fixed(10), reraise=True):
+        with attempt:
+            new_sync_standby = await get_sync_standby(ops_test, first_model, DATABASE_APP_NAME)
+            logger.info(f"New sync-standby: {new_sync_standby}")
+            assert new_sync_standby != sync_standby, "Sync-standby is the same as before"
 
     logger.info("Ensure continuous_writes after the crashed unit")
     await are_writes_increasing(ops_test)