Add explicit RTP protocol selection (IPv4 or IPv6).

Author: sobomax

sippy/Rtp/EPoint.py

@@ -4,7 +4,6 @@
 import errno
 
 from rtpsynth.RtpServer import RtpQueueFullError, RtpServer
-from sippy.misc import local4remote
 from sippy.Time.MonoTime import MonoTime
 
 from .Core.AudioChunk import AudioChunk
@@ -27,11 +26,12 @@ def __init__(self, rc:RTPConf, rtp_params:RTPParams, audio_in:callable,
         self.id = uuid4()
         self.rtp_params = rtp_params
         self.handlers = handlers or RTPHandlers()
+        self._palloc = rc.palloc
         self._rtp_server = None
         self.state_lock = Lock()
         self.writer = None
         self.rsess = self.make_rtp_instream(rtp_params, audio_in)
-        rserv_opts = self.make_udp_server_opts(rc, rtp_params)
+        rserv_opts = self.make_udp_server_opts(rtp_params)
         self.rserv = self.make_udp_server(rserv_opts)
         if self.rtp_params.rtp_target is not None:
             self.writer_setup()
@@ -42,12 +42,9 @@ def make_writer(self, rtp_params:RTPParams):
     def make_rtp_instream(self, rtp_params:RTPParams, audio_in:callable):
         return self.handlers.rtp_instream_cls(rtp_params, audio_in)
 
-    def make_udp_server_opts(self, rc:RTPConf, rtp_params:RTPParams):
-        if rtp_params.rtp_target is None:
-            rtp_laddr = '0.0.0.0'
-        else:
-            rtp_laddr = local4remote(rtp_params.rtp_target[0])
-        return (rtp_laddr, rc.palloc)
+    def make_udp_server_opts(self, rtp_params:RTPParams):
+        palloc = self._palloc if rtp_params.rtp_lport == 0 else rtp_params.rtp_lport
+        return (rtp_params.rtp_laddr, palloc)
 
     def make_udp_server(self, rserv_opts):
         rtp_laddr, palloc = rserv_opts
@@ -56,30 +53,10 @@ def make_udp_server(self, rserv_opts):
         channel = None
         self._rtp_server = rtp_server
         try:
-            if callable(palloc):
-                ntry = -1
-                while True:
-                    ntry += 1
-                    bind_port = int(palloc(ntry))
-                    try:
-                        channel = rtp_server.create_channel(
-                            pkt_in=self.rtp_received,
-                            bind_host=rtp_laddr,
-                            bind_port=bind_port,
-                        )
-                    except OSError as ex:
-                        if ex.errno == errno.EADDRINUSE:
-                            continue
-                        raise
-                    break
-            else:
-                channel = rtp_server.create_channel(
-                    pkt_in=self.rtp_received,
-                    bind_host=rtp_laddr,
-                    bind_port=int(palloc),
-                )
-            if self.rtp_params.rtp_target is not None:
-                channel.set_target(self.rtp_params.rtp_target[0], self.rtp_params.rtp_target[1])
+            channel = self._bind_channel(rtp_server, rtp_laddr, palloc)
+            target = self.rtp_params.rtp_target
+            if target is not None:
+                channel.set_target(target[0], target[1])
             return channel
         except Exception:
             release_rtp_server(rtp_server)
@@ -88,6 +65,51 @@ def make_udp_server(self, rserv_opts):
                 channel.close()
             raise
 
+    def _create_channel(self, rtp_server:RtpServer, bind_host:str, bind_port:int):
+        bind_family = self.rtp_params.rtp_family
+        ch_kwargs = dict(pkt_in=self.rtp_received, bind_host=bind_host, bind_port=bind_port)
+        ch = rtp_server.create_channel(bind_family=bind_family, **ch_kwargs)
+        self.rtp_params.rtp_lport = bind_port
+        return ch
+
+    def _bind_channel(self, rtp_server:RtpServer, rtp_laddr:str, palloc, preferred_port:int=None):
+        if preferred_port is not None:
+            try:
+                return self._create_channel(rtp_server, rtp_laddr, preferred_port)
+            except OSError as ex:
+                if ex.errno != errno.EADDRINUSE:
+                    raise
+        if callable(palloc):
+            ntry = -1
+            while True:
+                ntry += 1
+                bind_port = int(palloc(ntry))
+                try:
+                    return self._create_channel(rtp_server, rtp_laddr, bind_port)
+                except OSError as ex:
+                    if ex.errno == errno.EADDRINUSE:
+                        continue
+                    raise
+        return self._create_channel(rtp_server, rtp_laddr, int(palloc))
+
+    def _swap_channel(self, old_channel, rtp_params:RTPParams):
+        with self.state_lock:
+            rtp_server = self._rtp_server
+            if rtp_server is None or self.rserv is not old_channel:
+                return
+        rtp_laddr, palloc = self.make_udp_server_opts(rtp_params)
+        preferred_port = old_channel.local_addr[1]
+        new_channel = self._bind_channel(rtp_server, rtp_laddr, palloc, preferred_port=preferred_port)
+        target = rtp_params.rtp_target
+        if target is not None:
+            new_channel.set_target(target[0], target[1])
+        with self.state_lock:
+            if self._rtp_server is None or self.rserv is not old_channel:
+                new_channel.close()
+                return
+            self.rserv = new_channel
+        old_channel.close()
+
     def writer_setup(self):
         assert self.writer is None
         writer = self.make_writer(self.rtp_params)
@@ -124,8 +146,11 @@ def update(self, rtp_params:RTPParams):
         old_writer = None
         need_new_writer = False
         target_changed = False
+        proto_changed = False
         with self.state_lock:
             target_changed = self.rtp_params.rtp_target != rtp_params.rtp_target
+            proto_changed = self.rtp_params.rtp_proto != rtp_params.rtp_proto
+            self.rtp_params.rtp_proto = rtp_params.rtp_proto
             self.rtp_params.rtp_target = rtp_params.rtp_target
             ptime_changed = self.rtp_params.out_ptime != rtp_params.out_ptime
             self.rtp_params.out_ptime = rtp_params.out_ptime
@@ -137,15 +162,18 @@ def update(self, rtp_params:RTPParams):
                 self.writer = None
             elif self.writer is None:
                 need_new_writer = True
-            elif ptime_changed:
+            elif ptime_changed or proto_changed:
                 old_writer = self.writer
                 self.writer = None
                 need_new_writer = True
-        if target_changed and channel is not None and self.rtp_params.rtp_target is not None:
-            channel.set_target(rtp_params.rtp_target[0], rtp_params.rtp_target[1])
         if old_writer is not None:
             old_writer.end()
             old_writer.join()
+        if proto_changed and channel is not None:
+            self._swap_channel(channel, rtp_params)
+        elif target_changed and channel is not None and self.rtp_params.rtp_target is not None:
+            target = self.rtp_params.rtp_target
+            channel.set_target(target[0], target[1])
         if need_new_writer:
             new_writer = self.make_writer(rtp_params)
             new_writer.set_pkt_send_f(self.send_pkt)

sippy/Rtp/Params.py

@@ -1,18 +1,104 @@
+from functools import lru_cache
+from ipaddress import ip_address
 from typing import Optional, Tuple, Type, Union
+from socket import AF_INET, AF_INET6
+
+from sippy.misc import local4remote
 
 from .Codecs.G711 import G711Codec
 from .Codecs.G722 import G722Codec
 
+
+@lru_cache(maxsize=256)
+def canonicalize_rtp_host(host: str) -> str:
+    if host.startswith('[') and host.endswith(']'):
+        host = host[1:-1]
+    try:
+        return str(ip_address(host))
+    except ValueError:
+        return host
+
+
+def canonicalize_rtp_target(rtp_target: Optional[Tuple[str, int]]) -> Optional[Tuple[str, int]]:
+    if rtp_target is None:
+        return None
+    host, port = rtp_target
+    assert isinstance(host, str)
+    return (canonicalize_rtp_host(host), int(port))
+
+
+def canonicalize_rtp_address(address) -> Optional[Tuple[str, int]]:
+    if not isinstance(address, tuple) or len(address) < 2:
+        return None
+    host, port = address[:2]
+    if not isinstance(host, str):
+        return None
+    return (canonicalize_rtp_host(host), int(port))
+
+
 class RTPParams():
-    rtp_target: Optional[Tuple[str, int]]
+    _rtp_target: Optional[Tuple[str, int]]
+    _rtp_proto: str
+    rtp_laddr: str
+    rtp_lport: int = 0
     out_ptime: int
     out_sr: int
+    default_rtp_proto: str = 'IPV4'
     default_ptime: int = 20
     default_sr: int = 16000
     codec: Type[Union[G711Codec, G722Codec]]
+
     def __init__(self, rtp_target:Optional[Tuple[str, int]], out_ptime:int=default_ptime,
-                 out_sr:int=default_sr):
-        assert rtp_target is None or (isinstance(rtp_target, tuple) and len(rtp_target) == 2)
+                 out_sr:int=default_sr, rtp_proto:str=default_rtp_proto):
+        self._rtp_target = None
+        self._rtp_proto = self.default_rtp_proto
+        self.rtp_laddr = '0.0.0.0'
+        self.rtp_proto = rtp_proto
         self.rtp_target = rtp_target
         self.out_ptime = out_ptime
         self.out_sr = out_sr
+
+    @property
+    def rtp_target(self) -> Optional[Tuple[str, int]]:
+        return self._rtp_target
+
+    @rtp_target.setter
+    def rtp_target(self, rtp_target: Optional[Tuple[str, int]]):
+        assert rtp_target is None or (isinstance(rtp_target, tuple) and len(rtp_target) == 2)
+        # Fast path for common "set to current value" updates.
+        if rtp_target == self._rtp_target:
+            return
+        new_rtp_target = canonicalize_rtp_target(rtp_target)
+        if new_rtp_target == self._rtp_target:
+            return
+        self._rtp_target = new_rtp_target
+        self.rtp_laddr = self._get_laddr()
+
+    @property
+    def rtp_proto(self) -> str:
+        return self._rtp_proto
+
+    @rtp_proto.setter
+    def rtp_proto(self, rtp_proto: str):
+        assert isinstance(rtp_proto, str)
+        rtp_proto = rtp_proto.upper()
+        assert rtp_proto in ('IPV4', 'IPV6')
+        if rtp_proto == self._rtp_proto:
+            return
+        self._rtp_proto = rtp_proto
+        self.rtp_laddr = self._get_laddr()
+
+    @property
+    def rtp_family(self):
+        return AF_INET if self.rtp_proto == 'IPV4' else AF_INET6
+
+    def _get_laddr(self):
+        af = self.rtp_family
+        target = self.rtp_target
+        if target is None:
+            rtp_laddr = '0.0.0.0' if af == AF_INET else '::'
+        else:
+            rtp_laddr = local4remote(target[0], family=af)
+            if af == AF_INET6:
+                rtp_laddr = rtp_laddr[1:-1]
+        return rtp_laddr

tests/test_RtpParams.py

@@ -0,0 +1,66 @@
+import unittest
+from socket import AF_INET, AF_INET6
+from unittest.mock import patch
+
+from sippy.Rtp.Params import RTPParams, canonicalize_rtp_address, canonicalize_rtp_target
+
+
+class TestRtpParams(unittest.TestCase):
+    def test_canonicalize_ipv6_expanded_target(self):
+        target = canonicalize_rtp_target(('0:0:0:0:0:0:0:1', 14086))
+        self.assertEqual(target, ('::1', 14086))
+
+    def test_canonicalize_bracketed_ipv6_target(self):
+        target = canonicalize_rtp_target(('[::1]', '13998'))
+        self.assertEqual(target, ('::1', 13998))
+
+    def test_canonicalize_incoming_ipv6_address_tuple(self):
+        source = canonicalize_rtp_address(('0:0:0:0:0:0:0:1', 14086, 0, 0))
+        self.assertEqual(source, ('::1', 14086))
+
+    def test_params_store_canonical_target(self):
+        a = RTPParams(('::1', 14086), rtp_proto='IPV6')
+        b = RTPParams(('0:0:0:0:0:0:0:1', 14086), rtp_proto='IPV6')
+        self.assertEqual(a.rtp_target, b.rtp_target)
+
+    def test_rtp_proto_setter_normalizes_and_updates_laddr(self):
+        params = RTPParams(None, rtp_proto='IPV4')
+        self.assertEqual(params.rtp_laddr, '0.0.0.0')
+        params.rtp_proto = 'ipv6'
+        self.assertEqual(params.rtp_proto, 'IPV6')
+        self.assertEqual(params.rtp_laddr, '::')
+
+    def test_rtp_target_setter_canonicalizes_and_updates_laddr(self):
+        params = RTPParams(None, rtp_proto='IPV6')
+        with patch('sippy.Rtp.Params.local4remote', return_value='[2001:db8::1]') as mock_l4r:
+            params.rtp_target = ('0:0:0:0:0:0:0:1', '14086')
+            mock_l4r.assert_called_once_with('::1', family=AF_INET6)
+        self.assertEqual(params.rtp_target, ('::1', 14086))
+        self.assertEqual(params.rtp_laddr, '2001:db8::1')
+
+    def test_rtp_target_setter_skips_canonicalize_and_laddr_on_unchanged_value(self):
+        params = RTPParams(None, rtp_proto='IPV6')
+        with patch('sippy.Rtp.Params.canonicalize_rtp_target',
+                   side_effect=AssertionError('canonicalize_rtp_target should not be called')):
+            with patch.object(params, '_get_laddr', side_effect=AssertionError('_get_laddr should not be called')):
+                params.rtp_target = None
+
+    def test_rtp_proto_setter_skips_laddr_on_unchanged_value(self):
+        params = RTPParams(None, rtp_proto='IPV4')
+        with patch.object(params, '_get_laddr', side_effect=AssertionError('_get_laddr should not be called')):
+            params.rtp_proto = 'ipv4'
+
+    def test_rtp_family_property_maps_proto(self):
+        params = RTPParams(None, rtp_proto='IPV4')
+        self.assertEqual(params.rtp_family, AF_INET)
+        params.rtp_proto = 'IPV6'
+        self.assertEqual(params.rtp_family, AF_INET6)
+
+    def test_rtp_family_property_is_read_only(self):
+        params = RTPParams(None, rtp_proto='IPV4')
+        with self.assertRaises(AttributeError):
+            params.rtp_family = AF_INET6
+
+
+if __name__ == '__main__':
+    unittest.main()