Module livekit.api.access_token

def camel_to_snake(t: str) -> str:
    return re.sub(r"(?<!^)(?=[A-Z])", "_", t).lower()

def snake_to_lower_camel(t: str) -> str:
    return "".join(word.capitalize() if i else word for i, word in enumerate(t.split("_")))

class AccessToken:
    ParticipantKind = Literal["standard", "egress", "ingress", "sip", "agent"]

    def __init__(
        self,
        api_key: Optional[str] = None,
        api_secret: Optional[str] = None,
    ) -> None:
        api_key = api_key or os.getenv("LIVEKIT_API_KEY")
        api_secret = api_secret or os.getenv("LIVEKIT_API_SECRET")

        if not api_key or not api_secret:
            raise ValueError("api_key and api_secret must be set")

        self.api_key = api_key  # iss
        self.api_secret = api_secret
        self.claims = Claims()

        # default jwt claims
        self.identity = ""  # sub
        self.ttl = DEFAULT_TTL  # exp

    def with_ttl(self, ttl: datetime.timedelta) -> "AccessToken":
        self.ttl = ttl
        return self

    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_observability_grants(self, grants: ObservabilityGrants) -> "AccessToken":
        self.claims.observability = grants
        return self

    def with_inference_grants(self, grants: InferenceGrants) -> "AccessToken":
        self.claims.inference = grants
        return self

    def with_agent_grants(self, grants: AgentGrants) -> "AccessToken":
        self.claims.agent = grants
        return self

    def with_agent_endpoint_grants(self, grants: AgentEndpointGrants) -> "AccessToken":
        self.claims.agent_endpoint = grants
        return self

    def with_agent_thread_grants(self, grants: AgentThreadGrants) -> "AccessToken":
        self.claims.agent_thread = grants
        return self

    def with_identity(self, identity: str) -> "AccessToken":
        self.identity = identity
        return self

    def with_kind(self, kind: ParticipantKind) -> "AccessToken":
        self.claims.kind = kind
        return self

    def with_name(self, name: str) -> "AccessToken":
        self.claims.name = name
        return self

    def with_metadata(self, metadata: str) -> "AccessToken":
        self.claims.metadata = metadata
        return self

    def with_attributes(self, attributes: dict[str, str]) -> "AccessToken":
        self.claims.attributes = attributes
        return self

    def with_sha256(self, sha256: str) -> "AccessToken":
        self.claims.sha256 = sha256
        return self

    def with_room_preset(self, preset: str) -> "AccessToken":
        self.claims.room_preset = preset
        return self

    def with_room_config(self, config: RoomConfiguration) -> "AccessToken":
        self.claims.room_config = config
        return self

    def to_jwt(self) -> str:
        video = self.claims.video
        if video and video.room_join and (not self.identity or not video.room):
            raise ValueError("identity and room must be set when joining a room")

        # we want to exclude None values from the token
        jwt_claims = self.claims.asdict()
        jwt_claims.update(
            {
                "sub": self.identity,
                "iss": self.api_key,
                "nbf": calendar.timegm(datetime.datetime.now(datetime.timezone.utc).utctimetuple()),
                "exp": calendar.timegm(
                    (datetime.datetime.now(datetime.timezone.utc) + self.ttl).utctimetuple()
                ),
            }
        )
        return jwt.encode(jwt_claims, self.api_secret, algorithm="HS256")

@dataclass
class AgentEndpointGrants:
    # invoke an agent's non-public HTTP endpoints
    call: bool = False
    # restrict to one agent; empty grants every agent in the project
    agent_name: str = ""
    # restrict to one deployment; empty grants every deployment
    deployment: str = ""

@dataclass
class AgentGrants:
    # create/update/delete Cloud Agents
    admin: bool = False
    # manage simulations and scenarios for evaluating agents
    simulation_admin: bool = False
    # access the project's agent databases
    database_admin: bool = False
    # manage the project's agent dispatch queue
    dispatch_admin: bool = False

@dataclass
class AgentThreadGrants:
    # every thread and scope of the project; the action flags still apply
    all: bool = False
    # scope in which threads may be listed and created
    scope: str = ""
    # threads that may be read, written and deleted
    thread_ids: Optional[List[str]] = None
    list: bool = False
    create: bool = False
    write: bool = False
    read: bool = False
    delete: bool = False

@dataclass
class Claims:
    identity: str = ""
    name: str = ""
    kind: str = ""
    metadata: str = ""
    video: Optional[VideoGrants] = None
    sip: Optional[SIPGrants] = None
    inference: Optional[InferenceGrants] = None
    observability: Optional[ObservabilityGrants] = None
    agent: Optional[AgentGrants] = None
    agent_endpoint: Optional[AgentEndpointGrants] = None
    agent_thread: Optional[AgentThreadGrants] = None
    attributes: Optional[dict[str, str]] = None
    sha256: Optional[str] = None
    room_preset: Optional[str] = None
    room_config: Optional[RoomConfiguration] = None

    def asdict(self) -> dict:
        # in order to produce minimal jwt size, exclude None or empty values
        claims = dataclasses.asdict(
            self,
            dict_factory=lambda items: {
                snake_to_lower_camel(k): v for k, v in items if v is not None and v != ""
            },
        )
        if self.room_config:
            claims["roomConfig"] = MessageToDict(self.room_config)
        return claims

@dataclass
class InferenceGrants:
    # perform inference
    perform: bool = False

@dataclass
class ObservabilityGrants:
    # write grants to publish observability data
    write: bool = False

@dataclass
class SIPGrants:
    # manage sip resources
    admin: bool = False
    # make outbound calls
    call: bool = False

class TokenVerifier:
    def __init__(
        self,
        api_key: Optional[str] = None,
        api_secret: Optional[str] = None,
        *,
        leeway: datetime.timedelta = DEFAULT_LEEWAY,
    ) -> None:
        api_key = api_key or os.getenv("LIVEKIT_API_KEY")
        api_secret = api_secret or os.getenv("LIVEKIT_API_SECRET")

        self.api_key = api_key
        self.api_secret = api_secret
        self._leeway = leeway

    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")

        # 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__}
        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,
            attributes=claims.get("attributes", {}),
            metadata=claims.get("metadata", ""),
            sha256=claims.get("sha256", ""),
            agent=_parse_grants(claims, "agent", AgentGrants),
            agent_endpoint=_parse_grants(claims, "agentEndpoint", AgentEndpointGrants),
            agent_thread=_parse_grants(claims, "agentThread", AgentThreadGrants),
        )

        if claims.get("roomPreset"):
            grant_claims.room_preset = claims.get("roomPreset")
        room_config = claims.get("roomConfig")
        if room_config:
            grant_claims.room_config = ParseDict(
                room_config,
                RoomConfiguration(),
                ignore_unknown_fields=True,
            )

        return grant_claims

@dataclass
class VideoGrants:
    # actions on rooms
    room_create: Optional[bool] = None
    room_list: Optional[bool] = None
    room_record: Optional[bool] = None

    # actions on a particular room
    room_admin: Optional[bool] = None
    room_join: Optional[bool] = None
    room: str = ""

    # allows forwarding participant to room
    destination_room: Optional[str] = None

    # permissions within a room
    can_publish: bool = True
    can_subscribe: bool = True
    can_publish_data: bool = True

    # TrackSource types that a participant may publish.
    # When set, it supersedes CanPublish. Only sources explicitly set here can be
    # published
    can_publish_sources: Optional[List[str]] = None

    # by default, a participant is not allowed to update its own metadata
    can_update_own_metadata: Optional[bool] = None

    # actions on ingresses
    ingress_admin: Optional[bool] = None  # applies to all ingress

    # participant is not visible to other participants (useful when making bots)
    hidden: Optional[bool] = None

    # [deprecated] indicates to the room that current participant is a recorder
    recorder: Optional[bool] = None

    # indicates that the holder can register as an Agent framework worker
    agent: Optional[bool] = None

    # allow participant to manage an agent session via RemoteSession
    can_manage_agent_session: Optional[bool] = None

    # allow participant to subscribe to metrics
    can_subscribe_metrics: Optional[bool] = None