diff --git a/livekit-api/livekit/api/access_token.py b/livekit-api/livekit/api/access_token.py index e6516819..4f4d1658 100644 --- a/livekit-api/livekit/api/access_token.py +++ b/livekit-api/livekit/api/access_token.py @@ -233,14 +233,26 @@ def verify(self, token: str, *, verify_signature: bool = True) -> Claims: if verify_signature and (not self.api_key or not self.api_secret): raise ValueError("api_key and api_secret must be set") - claims = jwt.decode( - token, - key=self.api_secret or "", - issuer=self.api_key or "", - algorithms=["HS256"], - leeway=self._leeway.total_seconds(), - options={"verify_signature": verify_signature}, - ) + # First-party minters always set exp. Without this, a hand-rolled token + # with a valid signature and no exp verifies forever (livekit/protocol#1706). + if verify_signature: + claims = jwt.decode( + token, + key=self.api_secret or "", + issuer=self.api_key or "", + algorithms=["HS256"], + leeway=self._leeway.total_seconds(), + options={"verify_signature": True, "require": ["exp"]}, + ) + else: + claims = jwt.decode( + token, + key=self.api_secret or "", + issuer=self.api_key or "", + algorithms=["HS256"], + 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__} diff --git a/tests/api/test_access_token.py b/tests/api/test_access_token.py index 85aeb6ce..df5f4678 100644 --- a/tests/api/test_access_token.py +++ b/tests/api/test_access_token.py @@ -1,5 +1,7 @@ +import calendar import datetime +import jwt import pytest from livekit.api import AccessToken, TokenVerifier, VideoGrants, SIPGrants from livekit.protocol.room import RoomConfiguration @@ -101,3 +103,16 @@ def test_verify_token_expired() -> None: ) with pytest.raises(Exception): token_verifier.verify(token) + + +def test_verify_token_missing_exp() -> None: + now = calendar.timegm(datetime.datetime.now(datetime.timezone.utc).utctimetuple()) + token = jwt.encode( + {"sub": "test_identity", "iss": TEST_API_KEY, "nbf": now}, + TEST_API_SECRET, + algorithm="HS256", + ) + + token_verifier = TokenVerifier(TEST_API_KEY, TEST_API_SECRET) + with pytest.raises(Exception): + token_verifier.verify(token)