diff --git a/docs/docs/configuration/cameras.md b/docs/docs/configuration/cameras.md index f21984c8b..b800b1010 100644 --- a/docs/docs/configuration/cameras.md +++ b/docs/docs/configuration/cameras.md @@ -143,6 +143,11 @@ If your ONVIF camera does not require authentication credentials, you may still ::: +If a camera connects but fails to authenticate, two optional fields can help: + +- `tls_insecure`: Skips TLS certificate verification and sends the ONVIF password as plaintext (`PasswordText`) instead of a hashed digest (`PasswordDigest`). Some cameras reject the digest token and only accept plaintext. This weakens connection security, so only enable it on a trusted local network. +- `ignore_time_mismatch`: ONVIF authentication tokens include a timestamp, and a camera will reject the token if its clock differs too much from Frigate's. Enabling this makes Frigate compensate for the time offset so authentication can still succeed. Running NTP on both the camera and the Frigate host is the recommended fix; only use this in a "safe" environment, as it slightly weakens token validation. + If your camera has multiple ONVIF profiles, you can specify which one to use for PTZ control with the `profile` option, matched by token or name. When not set, Frigate selects the first profile with a valid PTZ configuration. Check the Frigate debug logs (`frigate.ptz.onvif: debug`) to see available profile names and tokens for your camera. An ONVIF-capable camera that supports relative movement within the field of view (FOV) can also be configured to automatically track moving objects and keep them in the center of the frame. For autotracking setup, see the [autotracking](autotracking.md) docs. diff --git a/frigate/api/camera.py b/frigate/api/camera.py index 9eb4bec9e..339ee33a1 100644 --- a/frigate/api/camera.py +++ b/frigate/api/camera.py @@ -529,6 +529,68 @@ def _extract_fps(r_frame_rate: str) -> float | None: return None +def _build_digest_transport(username: str, password: str) -> AsyncTransport: + """Build a zeep transport backed by an httpx client using HTTP digest auth.""" + auth = httpx.DigestAuth(username, password) + client = httpx.AsyncClient(auth=auth, timeout=10.0) + return AsyncTransport(client=client) + + +async def _connect_onvif_camera( + host: str, + port: int, + username: str, + password: str, + wsdl_base: str | None, + auth_type: str, +) -> ONVIFCamera: + """Connect to an ONVIF device, trying both WS-Security password encodings. + + Cameras disagree on whether the WS-Security UsernameToken should carry a + hashed PasswordDigest or a plaintext PasswordText. The wizard can't know + which a given camera expects, so we try PasswordDigest first (the common + case) and fall back to PasswordText when the device rejects the token. This + is independent of auth_type, which controls HTTP transport-level auth. + """ + first_error: Fault | None = None + + # encrypt=True -> PasswordDigest, encrypt=False -> PasswordText + for encrypt in (True, False): + onvif_camera = ONVIFCamera( + host, + port, + username or "", + password or "", + wsdl_dir=wsdl_base, + encrypt=encrypt, + ) + + try: + await onvif_camera.update_xaddrs() + except Fault as e: + # A SOAP fault here is how a camera signals the wrong password + # encoding, so retry with the other encoding before giving up. + logger.debug( + "ONVIF connect with %s rejected, trying alternate encoding", + "PasswordDigest" if encrypt else "PasswordText", + ) + if first_error is None: + first_error = e + continue + + if auth_type == "digest" and username and password: + transport = _build_digest_transport(username, password) + for service in ("devicemgmt", "media", "ptz"): + if hasattr(onvif_camera, service): + getattr(onvif_camera, service).zeep_client.transport = transport + logger.debug("Configured digest authentication") + + return onvif_camera + + # Both encodings failed authentication; surface the original fault. + raise first_error + + @router.get( "/onvif/probe", dependencies=[Depends(require_role(["admin"]))], @@ -605,34 +667,10 @@ async def onvif_probe( except Exception: wsdl_base = None - onvif_camera = ONVIFCamera( - host, port, username or "", password or "", wsdl_dir=wsdl_base + onvif_camera = await _connect_onvif_camera( + host, port, username, password, wsdl_base, auth_type ) - # Configure digest authentication if requested - if auth_type == "digest" and username and password: - # Create httpx client with digest auth - auth = httpx.DigestAuth(username, password) - client = httpx.AsyncClient(auth=auth, timeout=10.0) - - # Replace the transport in the zeep client - transport = AsyncTransport(client=client) - - # Update the xaddr before setting transport - await onvif_camera.update_xaddrs() - - # Replace transport in all services - if hasattr(onvif_camera, "devicemgmt"): - onvif_camera.devicemgmt.zeep_client.transport = transport - if hasattr(onvif_camera, "media"): - onvif_camera.media.zeep_client.transport = transport - if hasattr(onvif_camera, "ptz"): - onvif_camera.ptz.zeep_client.transport = transport - - logger.debug("Configured digest authentication") - else: - await onvif_camera.update_xaddrs() - # Get device information device_info = { "manufacturer": "Unknown", @@ -644,10 +682,9 @@ async def onvif_probe( # Update transport for device service if digest auth if auth_type == "digest" and username and password: - auth = httpx.DigestAuth(username, password) - client = httpx.AsyncClient(auth=auth, timeout=10.0) - transport = AsyncTransport(client=client) - device_service.zeep_client.transport = transport + device_service.zeep_client.transport = _build_digest_transport( + username, password + ) device_info_resp = await device_service.GetDeviceInformation() manufacturer = getattr(device_info_resp, "Manufacturer", None) or ( @@ -685,10 +722,9 @@ async def onvif_probe( # Update transport for media service if digest auth if auth_type == "digest" and username and password: - auth = httpx.DigestAuth(username, password) - client = httpx.AsyncClient(auth=auth, timeout=10.0) - transport = AsyncTransport(client=client) - media_service.zeep_client.transport = transport + media_service.zeep_client.transport = _build_digest_transport( + username, password + ) profiles = await media_service.GetProfiles() profiles_count = len(profiles) if profiles else 0 @@ -720,10 +756,9 @@ async def onvif_probe( # Update transport for PTZ service if digest auth if auth_type == "digest" and username and password: - auth = httpx.DigestAuth(username, password) - client = httpx.AsyncClient(auth=auth, timeout=10.0) - transport = AsyncTransport(client=client) - ptz_service.zeep_client.transport = transport + ptz_service.zeep_client.transport = _build_digest_transport( + username, password + ) # Check if PTZ service is available try: @@ -876,10 +911,9 @@ async def onvif_probe( # Update transport for media service if digest auth if auth_type == "digest" and username and password: - auth = httpx.DigestAuth(username, password) - client = httpx.AsyncClient(auth=auth, timeout=10.0) - transport = AsyncTransport(client=client) - media_service.zeep_client.transport = transport + media_service.zeep_client.transport = _build_digest_transport( + username, password + ) if profiles_count and media_service: for p in profiles or []: diff --git a/frigate/test/test_onvif_probe.py b/frigate/test/test_onvif_probe.py new file mode 100644 index 000000000..aee7cc17a --- /dev/null +++ b/frigate/test/test_onvif_probe.py @@ -0,0 +1,124 @@ +import unittest +from unittest.mock import AsyncMock, MagicMock, patch + +from zeep.exceptions import Fault, TransportError +from zeep.transports import AsyncTransport + +from frigate.api.camera import _build_digest_transport, _connect_onvif_camera + + +def _make_camera(update_side_effect=None): + """Build a mock ONVIFCamera whose update_xaddrs can raise or succeed.""" + camera = MagicMock() + camera.update_xaddrs = AsyncMock(side_effect=update_side_effect) + return camera + + +class TestConnectOnvifCamera(unittest.IsolatedAsyncioTestCase): + async def test_password_digest_succeeds_first(self): + # Cameras that accept PasswordDigest authenticate on the first attempt + # and should never trigger the PasswordText fallback. + camera = _make_camera() + + with patch("frigate.api.camera.ONVIFCamera", return_value=camera) as mock_cls: + result = await _connect_onvif_camera( + "cam.local", 80, "user", "pass", None, "basic" + ) + + self.assertIs(result, camera) + mock_cls.assert_called_once() + self.assertTrue(mock_cls.call_args.kwargs["encrypt"]) + + async def test_falls_back_to_password_text(self): + # A PasswordDigest rejection should retry once with PasswordText. + camera_digest = _make_camera(update_side_effect=Fault("token rejected")) + camera_text = _make_camera() + + with patch( + "frigate.api.camera.ONVIFCamera", + side_effect=[camera_digest, camera_text], + ) as mock_cls: + result = await _connect_onvif_camera( + "cam.local", 80, "user", "pass", None, "basic" + ) + + self.assertIs(result, camera_text) + self.assertEqual(mock_cls.call_count, 2) + self.assertTrue(mock_cls.call_args_list[0].kwargs["encrypt"]) + self.assertFalse(mock_cls.call_args_list[1].kwargs["encrypt"]) + + async def test_both_encodings_fail_raises_first_fault(self): + # When both encodings fault, the original (PasswordDigest) fault is + # surfaced so the caller's existing Fault handler reports it. + first_fault = Fault("digest rejected") + camera_digest = _make_camera(update_side_effect=first_fault) + camera_text = _make_camera(update_side_effect=Fault("text rejected")) + + with patch( + "frigate.api.camera.ONVIFCamera", + side_effect=[camera_digest, camera_text], + ) as mock_cls: + with self.assertRaises(Fault) as ctx: + await _connect_onvif_camera( + "cam.local", 80, "user", "pass", None, "basic" + ) + + self.assertIs(ctx.exception, first_fault) + self.assertEqual(mock_cls.call_count, 2) + + async def test_transport_error_is_not_retried(self): + # Connection-level errors (timeout, refused, unreachable) should + # propagate immediately without doubling latency on a second encoding. + camera = _make_camera(update_side_effect=TransportError("unreachable")) + + with patch("frigate.api.camera.ONVIFCamera", side_effect=[camera]) as mock_cls: + with self.assertRaises(TransportError): + await _connect_onvif_camera( + "cam.local", 80, "user", "pass", None, "basic" + ) + + mock_cls.assert_called_once() + + async def test_digest_auth_replaces_service_transports(self): + # auth_type "digest" wires an HTTP digest transport onto each service, + # independently of the WS-Security encoding. + camera = _make_camera() + + with ( + patch("frigate.api.camera.ONVIFCamera", return_value=camera), + patch( + "frigate.api.camera._build_digest_transport", + return_value="TRANSPORT", + ) as mock_transport, + ): + result = await _connect_onvif_camera( + "cam.local", 80, "user", "pass", None, "digest" + ) + + self.assertIs(result, camera) + mock_transport.assert_called_once_with("user", "pass") + self.assertEqual(camera.devicemgmt.zeep_client.transport, "TRANSPORT") + self.assertEqual(camera.media.zeep_client.transport, "TRANSPORT") + self.assertEqual(camera.ptz.zeep_client.transport, "TRANSPORT") + + async def test_basic_auth_does_not_replace_transports(self): + # Without digest auth, no transport override is built. + camera = _make_camera() + + with ( + patch("frigate.api.camera.ONVIFCamera", return_value=camera), + patch("frigate.api.camera._build_digest_transport") as mock_transport, + ): + await _connect_onvif_camera("cam.local", 80, "user", "pass", None, "basic") + + mock_transport.assert_not_called() + + +class TestBuildDigestTransport(unittest.TestCase): + def test_returns_async_transport(self): + transport = _build_digest_transport("user", "pass") + self.assertIsInstance(transport, AsyncTransport) + + +if __name__ == "__main__": + unittest.main()