fix: decouple relay-only diagnostics from LAN TURN setup

api/routes/webrtc_signaling.py

@@ -58,18 +58,53 @@ class NonRelayFilterPolicy(Enum):
     ALL = "all"          # filter all non-relay candidates (relay-only mode)
 
 
-if FORCE_TURN_RELAY:
-    # Outbound: relay-only so browser is forced through TURN.
-    # Inbound: pass all — browser host candidate needed for relay→host pairs.
-    ICE_OUTBOUND_POLICY = NonRelayFilterPolicy.ALL
-    ICE_INBOUND_POLICY = NonRelayFilterPolicy.NONE
-elif ENVIRONMENT == Environment.LOCAL.value:
-    ICE_OUTBOUND_POLICY = NonRelayFilterPolicy.NONE
-    ICE_INBOUND_POLICY = NonRelayFilterPolicy.NONE
-else:
-    # Non-local: drop private-IP host candidates to avoid coturn denied-peer-ip errors.
-    ICE_OUTBOUND_POLICY = NonRelayFilterPolicy.PRIVATE
-    ICE_INBOUND_POLICY = NonRelayFilterPolicy.PRIVATE
+def is_local_or_cgnat_ip(ip_str: str) -> bool:
+    """Return True for RFC1918, loopback, link-local, and CGNAT addresses."""
+
+    try:
+        ip = ipaddress.ip_address(ip_str)
+    except ValueError:
+        return False
+
+    is_cgnat = ip.version == 4 and ip in ipaddress.ip_network("100.64.0.0/10")
+    return ip.is_private or ip.is_loopback or ip.is_link_local or is_cgnat
+
+
+def resolve_ice_filter_policies(
+    environment: str,
+    force_turn_relay: bool,
+    server_ip: str,
+) -> tuple[NonRelayFilterPolicy, NonRelayFilterPolicy]:
+    """Resolve outbound and inbound non-relay filtering for this deployment."""
+
+    private_lan_deployment = (
+        environment != Environment.LOCAL.value and is_local_or_cgnat_ip(server_ip)
+    )
+
+    if force_turn_relay:
+        # Relay-only diagnostics stay explicit. On private LAN deployments we
+        # must still accept inbound private candidates for relay<->host pairs.
+        outbound_policy = NonRelayFilterPolicy.ALL
+        inbound_policy = (
+            NonRelayFilterPolicy.NONE
+            if private_lan_deployment
+            else NonRelayFilterPolicy.PRIVATE
+        )
+        return outbound_policy, inbound_policy
+
+    if environment == Environment.LOCAL.value or private_lan_deployment:
+        return NonRelayFilterPolicy.NONE, NonRelayFilterPolicy.NONE
+
+    # Public remote deployment: drop private-IP host candidates to avoid
+    # coturn denied-peer-ip errors against Docker bridge and LAN interfaces.
+    return NonRelayFilterPolicy.PRIVATE, NonRelayFilterPolicy.PRIVATE
+
+
+ICE_OUTBOUND_POLICY, ICE_INBOUND_POLICY = resolve_ice_filter_policies(
+    ENVIRONMENT,
+    FORCE_TURN_RELAY,
+    os.getenv("SERVER_IP", ""),
+)
 
 
 def is_private_ip_candidate(candidate_str: str) -> bool:
@@ -92,9 +127,7 @@ def is_private_ip_candidate(candidate_str: str) -> bool:
         if "typ" in parts:
             typ_index = parts.index("typ")
             ip_str = parts[typ_index - 2]
-            ip = ipaddress.ip_address(ip_str)
-            is_cgnat = ip in ipaddress.ip_network("100.64.0.0/10")
-            return ip.is_private or is_cgnat
+            return is_local_or_cgnat_ip(ip_str)
     except (ValueError, IndexError):
         pass
     return False

api/tests/test_is_private_ip_candidate.py

@@ -1,4 +1,11 @@
-from api.routes.webrtc_signaling import is_private_ip_candidate
+from api.enums import Environment
+from api.routes.webrtc_signaling import (
+    NonRelayFilterPolicy,
+    _keep_candidate,
+    is_local_or_cgnat_ip,
+    is_private_ip_candidate,
+    resolve_ice_filter_policies,
+)
 
 
 class TestIsPrivateIpCandidate:
@@ -142,3 +149,78 @@ class TestIsPrivateIpCandidate:
             "candidate:999 1 tcp 1518280447 192.168.1.100 9 typ host tcptype active"
         )
         assert is_private_ip_candidate(candidate) is True
+
+
+class TestIsLocalOrCgnatIp:
+    def test_loopback_is_local(self):
+        assert is_local_or_cgnat_ip("127.0.0.1") is True
+
+    def test_link_local_is_local(self):
+        assert is_local_or_cgnat_ip("169.254.1.1") is True
+
+    def test_cgnat_is_local(self):
+        assert is_local_or_cgnat_ip("100.64.0.1") is True
+
+    def test_public_ipv4_is_not_local(self):
+        assert is_local_or_cgnat_ip("8.8.8.8") is False
+
+
+class TestKeepCandidate:
+    def test_private_relay_candidate_survives_private_policy(self):
+        candidate = (
+            "candidate:111 1 udp 41885439 192.168.1.50 50000 typ relay raddr 0.0.0.0 rport 0"
+        )
+        assert _keep_candidate(candidate, NonRelayFilterPolicy.PRIVATE) is True
+
+    def test_private_host_candidate_drops_under_private_policy(self):
+        candidate = (
+            "candidate:123 1 udp 2122260223 192.168.50.24 63603 typ host generation 0"
+        )
+        assert _keep_candidate(candidate, NonRelayFilterPolicy.PRIVATE) is False
+
+
+class TestResolveIceFilterPolicies:
+    def test_local_deployment_keeps_all_candidates(self):
+        outbound, inbound = resolve_ice_filter_policies(
+            Environment.LOCAL.value,
+            False,
+            "",
+        )
+        assert outbound == NonRelayFilterPolicy.NONE
+        assert inbound == NonRelayFilterPolicy.NONE
+
+    def test_private_lan_remote_keeps_all_candidates(self):
+        outbound, inbound = resolve_ice_filter_policies(
+            Environment.PRODUCTION.value,
+            False,
+            "192.168.50.24",
+        )
+        assert outbound == NonRelayFilterPolicy.NONE
+        assert inbound == NonRelayFilterPolicy.NONE
+
+    def test_public_remote_filters_private_candidates(self):
+        outbound, inbound = resolve_ice_filter_policies(
+            Environment.PRODUCTION.value,
+            False,
+            "8.8.8.8",
+        )
+        assert outbound == NonRelayFilterPolicy.PRIVATE
+        assert inbound == NonRelayFilterPolicy.PRIVATE
+
+    def test_force_turn_relay_stays_relay_only_on_private_lan(self):
+        outbound, inbound = resolve_ice_filter_policies(
+            Environment.PRODUCTION.value,
+            True,
+            "192.168.50.24",
+        )
+        assert outbound == NonRelayFilterPolicy.ALL
+        assert inbound == NonRelayFilterPolicy.NONE
+
+    def test_force_turn_relay_keeps_public_remote_private_filter(self):
+        outbound, inbound = resolve_ice_filter_policies(
+            Environment.PRODUCTION.value,
+            True,
+            "8.8.8.8",
+        )
+        assert outbound == NonRelayFilterPolicy.ALL
+        assert inbound == NonRelayFilterPolicy.PRIVATE