napalm_linux/linux.py

@@ -46,6 +46,7 @@ from napalm_device_types import (
     OpenSshServerMixin,
     OSDriver,
     PackageProxyMixin,
+    ServiceClustersMixin,
     SnmpAgentMixin,
     SystemIdentityMixin,
     SystemdServicesMixin,
@@ -294,6 +295,7 @@ class LinuxDriver(
     KernelFactsMixin,
     ListeningSocketsMixin,
     VirtualIpsMixin,
+    ServiceClustersMixin,
     SystemIdentityMixin,
     OpenSshServerMixin,
     SystemdServicesMixin,
@@ -1179,6 +1181,13 @@ class LinuxDriver(
         """
         return self.run_command(command, privileged=privileged, timeout=60).stdout
 
+    def _run_service_clusters_command(self, command: str, *, privileged: bool) -> str:
+        """The transport for ``ServiceClustersMixin.get_service_clusters``: an exec
+        channel, like the virtual IPs' command, which it resembles in size.
+        Read-only either way; root only because Patroni's configuration is the
+        postgres user's alone."""
+        return self.run_command(command, privileged=privileged, timeout=60).stdout
+
     def _run_system_identity_command(self, command: str, *, privileged: bool) -> str:
         """The transport for ``SystemIdentityMixin.get_system_identity``: an exec channel,
         as root, since ``product_uuid`` and ``product_serial`` are root's."""

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "napalm-linux"
-version = "0.9.0"
+version = "0.10.0"
 description = "NAPALM driver for generic Linux systems via SSH"
 readme = "README.md"
 requires-python = ">=3.9"
@@ -37,7 +37,7 @@ classifiers = [
 ]
 dependencies = [
     "napalm>=4.0",
-    "napalm-device-types>=4.14.0",
+    "napalm-device-types>=4.17.0",
     "netmiko>=4.0.0",
     "paramiko>=5.0.0",  # CVE-2026-44405
 ]

tests/test_linux.py

@@ -1721,6 +1721,44 @@ class TestGetVirtualIps:
         assert plain.startswith("sh -c '")
 
 
+# ---------------------------------------------------------------------------
+# get_service_clusters (napalm-device-types ServiceClustersMixin)
+# ---------------------------------------------------------------------------
+
+_CLUSTERS = (
+    "SVCC_BEGIN\n[systemd]\n[patroni patroni.service]\nactive active\nscope netork-db\n"
+    "name db-01\nrestapi 10.7.234.12:8008\nrest-unreachable\n[no-docker]\nSVCC_END\n"
+)
+
+
+class TestGetServiceClusters:
+    """The same template as get_virtual_ips: an exec channel, root first."""
+
+    _exec = staticmethod(TestGetVirtualIps._exec)
+
+    def test_it_reads_with_the_shared_command_on_an_exec_channel(self, driver):
+        from napalm_device_types import ServiceClustersMixin
+
+        assert isinstance(driver, ServiceClustersMixin)
+        calls = self._exec(driver, (_CLUSTERS, "", 0))
+
+        reading = driver.get_service_clusters()
+
+        assert reading["patroni"][0]["cluster"] == "netork-db"
+        [(command, privileged)] = calls
+        assert privileged is True
+        assert command.startswith("sh -c '")
+        driver._device.send_command.assert_not_called()
+
+    def test_when_sudo_refuses_it_reads_what_the_user_may_see(self, driver):
+        calls = self._exec(driver, ("", "sudo: a password is required\n", 1), (_CLUSTERS, "", 0))
+
+        reading = driver.get_service_clusters()
+
+        assert reading["bunkerweb"] == []
+        assert [privileged for _, privileged in calls] == [True, False]
+
+
 class TestHardensItsSshServer:
     """A stock Linux host runs a stock OpenSSH: the hardening netOrk offers for
     ssh-audit findings comes from napalm-device-types, carried over run_command