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: