Added support for Natted IP addresses

onvif/client.py

@@ -22,6 +22,7 @@
 from onvif.definition import SERVICES
 from onvif.exceptions import ONVIFAuthError, ONVIFError, ONVIFTimeoutError
 from requests import Response
+from urllib.parse import urlparse, urlunparse
 
 from .const import KEEPALIVE_EXPIRY
 from .managers import NotificationManager, PullPointManager
@@ -457,6 +458,47 @@ async def get_capabilities(self) -> dict[str, Any]:
             await self.update_xaddrs()
         return self._capabilities
 
+    def rewrite_xaddr(self, original_xaddr):
+        """Replace host:port in XAddr with the connection host:port"""
+        if not original_xaddr:
+            return None
+
+        parsed = urlparse(original_xaddr)
+
+        # Check if the reported address differs from connection address
+        # If they match, no rewriting needed
+        if parsed.hostname == self.host and (
+            not parsed.port or parsed.port == self.port
+        ):
+            return original_xaddr
+
+        # Build new netloc with our connection host/port
+        if self.port and self.port != 80:
+            new_netloc = f"{self.host}:{self.port}"
+        else:
+            new_netloc = self.host
+
+        # Reconstruct URL with new host:port but same path
+        rewritten = urlunparse(
+            (
+                parsed.scheme,  # Keep original scheme (http/https)
+                new_netloc,  # Use connection host:port
+                parsed.path,  # Keep original path
+                parsed.params,  # Keep params
+                parsed.query,  # Keep query
+                parsed.fragment,  # Keep fragment
+            )
+        )
+
+        logger.debug(
+            "%s: NAT detected - rewriting XAddr from %s to %s",
+            self.host,
+            original_xaddr,
+            rewritten,
+        )
+
+        return rewritten
+
     async def update_xaddrs(self):
         """Update xaddrs for services."""
         self.dt_diff = None
@@ -489,7 +531,11 @@ async def update_xaddrs(self):
             try:
                 if name.lower() in SERVICES and capability is not None:
                     namespace = SERVICES[name.lower()]["ns"]
-                    self.xaddrs[namespace] = normalize_url(capability["XAddr"])
+                    original_xaddr = normalize_url(capability["XAddr"])
+                    # Rewrite the xaddr for NAT before storing
+                    rewritten_xaddr = self.rewrite_xaddr(original_xaddr)
+                    self.xaddrs[namespace] = rewritten_xaddr
+                    capability["XAddr"] = rewritten_xaddr
             except Exception:
                 logger.exception("Unexpected service type")
         try:

tests/test_nat.py

@@ -0,0 +1,65 @@
+from __future__ import annotations
+
+import os
+from unittest.mock import AsyncMock, patch
+
+import pytest
+from onvif.client import ONVIFCamera
+
+_WSDL_PATH = os.path.join(os.path.dirname(os.path.dirname(__file__)), "onvif", "wsdl")
+
+
+@pytest.mark.asyncio
+async def test_rewrite_xaddr_logic():
+    """Test the core rewrite_xaddr logic."""
+    with patch("onvif.client.TCPConnector"), patch("onvif.client.ClientSession"):
+        device = ONVIFCamera("203.0.113.5", 8080, "user", "pass", wsdl_dir=_WSDL_PATH)
+
+        # different IP/Port
+        assert (
+            device.rewrite_xaddr("http://192.168.1.10/service")
+            == "http://203.0.113.5:8080/service"
+        )
+
+        # dame IP/Port
+        assert (
+            device.rewrite_xaddr("http://203.0.113.5:8080/service")
+            == "http://203.0.113.5:8080/service"
+        )
+
+        # default port
+        device_80 = ONVIFCamera("203.0.113.5", 80, "user", "pass", wsdl_dir=_WSDL_PATH)
+        assert (
+            device_80.rewrite_xaddr("http://10.0.0.1/service")
+            == "http://203.0.113.5/service"
+        )
+
+
+@pytest.mark.asyncio
+async def test_update_xaddrs_nat_rewrite():
+    """Test that update_xaddrs properly rewrites XAddrs in capabilities."""
+    with patch("onvif.client.TCPConnector"), patch("onvif.client.ClientSession"):
+        device = ONVIFCamera("203.0.113.5", 8080, "user", "pass", wsdl_dir=_WSDL_PATH)
+
+        mock_capabilities = {
+            "Media": {"XAddr": "http://192.168.1.10/onvif/media_service"},
+            "Events": {"XAddr": "http://192.168.1.10/onvif/event_service"},
+        }
+
+        mock_devicemgmt = AsyncMock()
+        mock_devicemgmt.GetCapabilities = AsyncMock(return_value=mock_capabilities)
+        mock_devicemgmt.binding_key = "device"
+
+        with patch.object(
+            device, "create_devicemgmt_service", return_value=mock_devicemgmt
+        ):
+            await device.update_xaddrs()
+
+            expected_media = "http://203.0.113.5:8080/onvif/media_service"
+            expected_events = "http://203.0.113.5:8080/onvif/event_service"
+
+            assert device._capabilities["Media"]["XAddr"] == expected_media
+            assert device._capabilities["Events"]["XAddr"] == expected_events
+
+            assert any(v == expected_media for v in device.xaddrs.values())
+            assert any(v == expected_events for v in device.xaddrs.values())