diff --git a/livekit-api/livekit/api/access_token.py b/livekit-api/livekit/api/access_token.py index 51e6e7e7..19f77261 100644 --- a/livekit-api/livekit/api/access_token.py +++ b/livekit-api/livekit/api/access_token.py @@ -308,40 +308,14 @@ def verify(self, token: str, *, verify_signature: bool = True) -> Claims: leeway=self._leeway.total_seconds(), options={"verify_signature": False}, ) - video_dict = claims.get("video", dict()) - video_dict = {camel_to_snake(k): v for k, v in video_dict.items()} - video_dict = {k: v for k, v in video_dict.items() if k in VideoGrants.__dataclass_fields__} - 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) - - inference_dict = claims.get("inference", dict()) - inference_dict = {camel_to_snake(k): v for k, v in inference_dict.items()} - inference_dict = { - k: v for k, v in inference_dict.items() if k in InferenceGrants.__dataclass_fields__ - } - inference = InferenceGrants(**inference_dict) - - observability_dict = claims.get("observability", dict()) - observability_dict = {camel_to_snake(k): v for k, v in observability_dict.items()} - observability_dict = { - k: v - for k, v in observability_dict.items() - if k in ObservabilityGrants.__dataclass_fields__ - } - observability = ObservabilityGrants(**observability_dict) - grant_claims = Claims( identity=claims.get("sub", ""), name=claims.get("name", ""), kind=claims.get("kind", ""), - video=video, - sip=sip, - inference=inference, - observability=observability, + video=_parse_grants(claims, "video", VideoGrants), + sip=_parse_grants(claims, "sip", SIPGrants), + inference=_parse_grants(claims, "inference", InferenceGrants), + observability=_parse_grants(claims, "observability", ObservabilityGrants), attributes=claims.get("attributes", {}), metadata=claims.get("metadata", ""), sha256=claims.get("sha256", ""), diff --git a/tests/api/test_access_token.py b/tests/api/test_access_token.py index b726ada9..d3e46eec 100644 --- a/tests/api/test_access_token.py +++ b/tests/api/test_access_token.py @@ -8,11 +8,12 @@ AgentEndpointGrants, AgentGrants, AgentThreadGrants, + InferenceGrants, + ObservabilityGrants, SIPGrants, TokenVerifier, VideoGrants, ) -from livekit.api.access_token import ObservabilityGrants from livekit.protocol.room import RoomConfiguration from livekit.protocol.agent_dispatch import RoomAgentDispatch @@ -207,3 +208,34 @@ def test_verify_token_preserves_kind_and_observability() -> None: assert claims.kind == "agent" assert claims.observability is not None assert claims.observability.write is True + + +def test_verify_token_omitted_grants_are_none() -> None: + token = AccessToken(TEST_API_KEY, TEST_API_SECRET).with_identity("test_identity").to_jwt() + + claims = TokenVerifier(TEST_API_KEY, TEST_API_SECRET).verify(token) + assert claims.identity == "test_identity" + assert claims.video is None + assert claims.sip is None + assert claims.inference is None + assert claims.observability is None + assert claims.agent is None + assert claims.agent_endpoint is None + assert claims.agent_thread is None + + +def test_verify_token_with_only_inference_grants() -> None: + inference = InferenceGrants(perform=True) + token = ( + AccessToken(TEST_API_KEY, TEST_API_SECRET) + .with_identity("test_identity") + .with_inference_grants(inference) + .to_jwt() + ) + + claims = TokenVerifier(TEST_API_KEY, TEST_API_SECRET).verify(token) + assert claims.identity == "test_identity" + assert claims.inference == inference + assert claims.video is None + assert claims.sip is None + assert claims.observability is None