diff --git a/napalm_linux/linux.py b/napalm_linux/linux.py
index e753752..b870940 100644
--- a/napalm_linux/linux.py
+++ b/napalm_linux/linux.py
@@ -1999,6 +1999,14 @@ class LinuxDriver(
             output = f"{output}\n{_SUDO_PASSWORD_HINT}"
         return {"ok": status == 0, "output": (output.strip()[-300:] if status != 0 else done)}
 
+    def get_management_peer(self) -> str | None:
+        """The address this host sees the SSH connection come from (``SSH_CONNECTION``)."""
+        out = self.run_command('printf "%s\\n" "$SSH_CONNECTION"', timeout=15).stdout.split()
+        try:
+            return str(ipaddress.ip_address(out[0])) if out else None
+        except ValueError:
+            return None
+
     def _install_snmp_agent(self) -> FixStepOutcomeDict:
         pm = self._pkg_manager or self._detect_pkg_manager()
         command = _SNMP_INSTALL.get(pm or "")
diff --git a/tests/test_fix_templates.py b/tests/test_fix_templates.py
index 3b1e6d4..7adfe74 100644
--- a/tests/test_fix_templates.py
+++ b/tests/test_fix_templates.py
@@ -191,3 +191,23 @@ def test_every_annotation_resolves():
     for name, member in inspect.getmembers(LinuxDriver, inspect.isfunction):
         if member.__module__ == "napalm_linux.linux":
             typing.get_type_hints(member)
+
+
+class TestManagementPeer:
+    """The address the host sees this connection come from (``SSH_CONNECTION``)."""
+
+    def _driver(self, answer):
+        d = LinuxDriver("h", "admin", "p")
+        d._device = MagicMock()
+        d.run_command = MagicMock(return_value=MagicMock(stdout=answer, exit_code=0))
+        return d
+
+    def test_the_client_side_of_ssh_connection(self):
+        assert self._driver("10.7.224.20 51234 10.0.0.5 22\n").get_management_peer() == "10.7.224.20"
+
+    def test_an_ipv6_client(self):
+        assert self._driver("fd00::20 51234 fd00::5 22\n").get_management_peer() == "fd00::20"
+
+    @pytest.mark.parametrize("answer", ["", "\n", "garbage", "not-an-ip 1 2 3"])
+    def test_nothing_usable_is_none(self, answer):
+        assert self._driver(answer).get_management_peer() is None