Update protocol. Support SIP inbound/outbound trunks, grants.

livekit-api/livekit/api/__init__.py

@@ -25,6 +25,6 @@
 
 from .twirp_client import TwirpError, TwirpErrorCode
 from .livekit_api import LiveKitAPI
-from .access_token import VideoGrants, AccessToken, TokenVerifier
+from .access_token import VideoGrants, SIPGrants, AccessToken, TokenVerifier
 from .webhook import WebhookReceiver
 from .version import __version__

livekit-api/livekit/api/_service.py

@@ -2,7 +2,7 @@
 import aiohttp
 from abc import ABC
 from .twirp_client import TwirpClient
-from .access_token import AccessToken, VideoGrants
+from .access_token import AccessToken, VideoGrants, SIPGrants
 
 AUTHORIZATION = "authorization"
 
@@ -15,8 +15,14 @@ def __init__(
         self.api_key = api_key
         self.api_secret = api_secret
 
-    def _auth_header(self, grants: VideoGrants) -> Dict[str, str]:
-        token = AccessToken(self.api_key, self.api_secret).with_grants(grants).to_jwt()
+    def _auth_header(
+        self, grants: VideoGrants, sip: SIPGrants | None = None
+    ) -> Dict[str, str]:
+        tok = AccessToken(self.api_key, self.api_secret).with_grants(grants)
+        if sip is not None:
+            tok = tok.with_sip_grants(sip)
+
+        token = tok.to_jwt()
 
         headers = {}
         headers[AUTHORIZATION] = "Bearer {}".format(token)

livekit-api/livekit/api/access_token.py

@@ -63,11 +63,20 @@ class VideoGrants:
     agent: bool = False
 
 
+@dataclasses.dataclass
+class SIPGrants:
+    # manage sip resources
+    admin: bool = False
+    # make outbound calls
+    call: bool = False
+
+
 @dataclasses.dataclass
 class Claims:
     identity: str = ""
     name: str = ""
     video: VideoGrants = dataclasses.field(default_factory=VideoGrants)
+    sip: SIPGrants = dataclasses.field(default_factory=SIPGrants)
     metadata: str = ""
     sha256: str = ""
 
@@ -100,6 +109,10 @@ def with_grants(self, grants: VideoGrants) -> "AccessToken":
         self.claims.video = grants
         return self
 
+    def with_sip_grants(self, grants: SIPGrants) -> "AccessToken":
+        self.claims.sip = grants
+        return self
+
     def with_identity(self, identity: str) -> "AccessToken":
         self.identity = identity
         return self
@@ -173,10 +186,18 @@ def verify(self, token: str) -> Claims:
         }
         video = VideoGrants(**video_dict)
 
+        sip_dict = claims.get("sip", dict())
+        sip_dict = {camel_to_snake(k): v for k, v in sip_dict.items()}
+        sip_dict = {
+            k: v for k, v in sip_dict.items() if k in SIPGrants.__dataclass_fields__
+        }
+        sip = SIPGrants(**sip_dict)
+
         return Claims(
             identity=claims.get("sub", ""),
             name=claims.get("name", ""),
             video=video,
+            sip=sip,
             metadata=claims.get("metadata", ""),
             sha256=claims.get("sha256", ""),
         )

livekit-api/livekit/api/sip_service.py

@@ -1,7 +1,7 @@
 import aiohttp
 from livekit.protocol import sip as proto_sip
 from ._service import Service
-from .access_token import VideoGrants
+from .access_token import VideoGrants, SIPGrants
 
 SVC = "SIP"
 
@@ -19,29 +19,73 @@ async def create_sip_trunk(
             SVC,
             "CreateSIPTrunk",
             create,
-            self._auth_header(VideoGrants()),
+            self._auth_header(VideoGrants(), sip=SIPGrants(admin=True)),
             proto_sip.SIPTrunkInfo,
         )
 
+    async def create_sip_inbound_trunk(
+        self, create: proto_sip.CreateSIPInboundTrunkRequest
+    ) -> proto_sip.SIPInboundTrunkInfo:
+        return await self._client.request(
+            SVC,
+            "CreateSIPInboundTrunk",
+            create,
+            self._auth_header(VideoGrants(), sip=SIPGrants(admin=True)),
+            proto_sip.SIPInboundTrunkInfo,
+        )
+
+    async def create_sip_outbound_trunk(
+        self, create: proto_sip.CreateSIPOutboundTrunkRequest
+    ) -> proto_sip.SIPOutboundTrunkInfo:
+        return await self._client.request(
+            SVC,
+            "CreateSIPOutboundTrunk",
+            create,
+            self._auth_header(VideoGrants(), sip=SIPGrants(admin=True)),
+            proto_sip.SIPOutboundTrunkInfo,
+        )
+
     async def list_sip_trunk(
         self, list: proto_sip.ListSIPTrunkRequest
     ) -> proto_sip.ListSIPTrunkResponse:
         return await self._client.request(
             SVC,
             "ListSIPTrunk",
             list,
-            self._auth_header(VideoGrants()),
+            self._auth_header(VideoGrants(), sip=SIPGrants(admin=True)),
             proto_sip.ListSIPTrunkResponse,
         )
 
+    async def list_sip_inbound_trunk(
+        self, list: proto_sip.ListSIPInboundTrunkRequest
+    ) -> proto_sip.ListSIPInboundTrunkResponse:
+        return await self._client.request(
+            SVC,
+            "ListSIPInboundTrunk",
+            list,
+            self._auth_header(VideoGrants(), sip=SIPGrants(admin=True)),
+            proto_sip.ListSIPInboundTrunkResponse,
+        )
+
+    async def list_sip_outbound_trunk(
+        self, list: proto_sip.ListSIPOutboundTrunkRequest
+    ) -> proto_sip.ListSIPOutboundTrunkResponse:
+        return await self._client.request(
+            SVC,
+            "ListSIPOutboundTrunk",
+            list,
+            self._auth_header(VideoGrants(), sip=SIPGrants(admin=True)),
+            proto_sip.ListSIPOutboundTrunkResponse,
+        )
+
     async def delete_sip_trunk(
         self, delete: proto_sip.DeleteSIPTrunkRequest
     ) -> proto_sip.SIPTrunkInfo:
         return await self._client.request(
             SVC,
             "DeleteSIPTrunk",
             delete,
-            self._auth_header(VideoGrants()),
+            self._auth_header(VideoGrants(), sip=SIPGrants(admin=True)),
             proto_sip.SIPTrunkInfo,
         )
 
@@ -52,7 +96,7 @@ async def create_sip_dispatch_rule(
             SVC,
             "CreateSIPDispatchRule",
             create,
-            self._auth_header(VideoGrants()),
+            self._auth_header(VideoGrants(), sip=SIPGrants(admin=True)),
             proto_sip.SIPDispatchRuleInfo,
         )
 
@@ -63,7 +107,7 @@ async def list_sip_dispatch_rule(
             SVC,
             "ListSIPDispatchRule",
             list,
-            self._auth_header(VideoGrants()),
+            self._auth_header(VideoGrants(), sip=SIPGrants(admin=True)),
             proto_sip.ListSIPDispatchRuleResponse,
         )
 
@@ -74,7 +118,7 @@ async def delete_sip_dispatch_rule(
             SVC,
             "DeleteSIPDispatchRule",
             delete,
-            self._auth_header(VideoGrants()),
+            self._auth_header(VideoGrants(), sip=SIPGrants(admin=True)),
             proto_sip.SIPDispatchRuleInfo,
         )
 
@@ -85,6 +129,6 @@ async def create_sip_participant(
             SVC,
             "CreateSIPParticipant",
             create,
-            self._auth_header(VideoGrants()),
+            self._auth_header(VideoGrants(), sip=SIPGrants(call=True)),
             proto_sip.SIPParticipantInfo,
         )

livekit-api/tests/test_access_token.py

@@ -1,20 +1,22 @@
 import datetime
 
 import pytest  # type: ignore
-from livekit.api import AccessToken, TokenVerifier, VideoGrants
+from livekit.api import AccessToken, TokenVerifier, VideoGrants, SIPGrants
 
 TEST_API_KEY = "myapikey"
 TEST_API_SECRET = "thiskeyistotallyunsafe"
 
 
 def test_verify_token():
     grants = VideoGrants(room_join=True, room="test_room")
+    sip = SIPGrants(admin=True)
 
     token = (
         AccessToken(TEST_API_KEY, TEST_API_SECRET)
         .with_identity("test_identity")
         .with_metadata("test_metadata")
         .with_grants(grants)
+        .with_sip_grants(sip)
         .to_jwt()
     )
 
@@ -24,6 +26,7 @@ def test_verify_token():
     assert claims.identity == "test_identity"
     assert claims.metadata == "test_metadata"
     assert claims.video == grants
+    assert claims.sip == sip
 
 
 def test_verify_token_invalid():