From b7beb70e6130bb11e500d37bc0adf8b986ff83e4 Mon Sep 17 00:00:00 2001 From: Reda DRISSI Date: Thu, 8 Jan 2026 17:19:09 +0100 Subject: [PATCH 1/4] Added support for Natted IP addresses This adds support for natted ip addresses where previously onvif only worked when in local address. Instead of returning the local IP address, it returns the same IP/port used to create the ONVIFCamera object --- onvif/client.py | 43 ++++++++++++++++++++++++++++++++++++++++++- 1 file changed, 42 insertions(+), 1 deletion(-) diff --git a/onvif/client.py b/onvif/client.py index 55df04c..d48774b 100644 --- a/onvif/client.py +++ b/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,42 @@ 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 +526,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: From 40925baaf792d80e1621e08f634ec6ff50a4c877 Mon Sep 17 00:00:00 2001 From: "pre-commit-ci[bot]" <66853113+pre-commit-ci[bot]@users.noreply.github.com> Date: Thu, 8 Jan 2026 16:21:17 +0000 Subject: [PATCH 2/4] chore(pre-commit.ci): auto fixes --- onvif/client.py | 37 +++++++++++++++++++++---------------- 1 file changed, 21 insertions(+), 16 deletions(-) diff --git a/onvif/client.py b/onvif/client.py index d48774b..e8bb690 100644 --- a/onvif/client.py +++ b/onvif/client.py @@ -462,38 +462,43 @@ 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): + 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 - )) - + 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 + rewritten, ) - + return rewritten + async def update_xaddrs(self): """Update xaddrs for services.""" self.dt_diff = None From 689f95efe99ac7e0d2d2551a66e64aaccee5a7df Mon Sep 17 00:00:00 2001 From: DrissiReda Date: Sun, 1 Mar 2026 20:24:46 +0100 Subject: [PATCH 3/4] Added tests for nat support --- tests/test_nat.py | 55 +++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 55 insertions(+) create mode 100644 tests/test_nat.py diff --git a/tests/test_nat.py b/tests/test_nat.py new file mode 100644 index 0000000..321d892 --- /dev/null +++ b/tests/test_nat.py @@ -0,0 +1,55 @@ +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()) From 3dc74c6f1ee39e3d451826b9a5117e26200dedac Mon Sep 17 00:00:00 2001 From: "pre-commit-ci[bot]" <66853113+pre-commit-ci[bot]@users.noreply.github.com> Date: Sun, 1 Mar 2026 19:25:38 +0000 Subject: [PATCH 4/4] chore(pre-commit.ci): auto fixes --- tests/test_nat.py | 48 ++++++++++++++++++++++++++++------------------- 1 file changed, 29 insertions(+), 19 deletions(-) diff --git a/tests/test_nat.py b/tests/test_nat.py index 321d892..4d192aa 100644 --- a/tests/test_nat.py +++ b/tests/test_nat.py @@ -8,48 +8,58 @@ _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"): + 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" - + 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" - + 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" + 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"): - + 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"} + "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): + + 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())