/
/
/
1"""JWT token helper for Music Assistant authentication.
2
3Future OIDC Support:
4- Consuming external OIDC providers (Google, Keycloak, etc.): Can be added without
5 changes to token structure. MA would validate external OIDC tokens and issue its
6 own JWT tokens (similar to current Home Assistant OAuth flow).
7
8- Acting as OIDC provider for third parties: Would require implementing OAuth2
9 refresh token flow with a dedicated /auth/token endpoint for token refresh.
10 Short-lived access tokens (15 min) + long-lived refresh tokens would be needed
11 for proper OIDC compliance.
12"""
13
14from __future__ import annotations
15
16import secrets
17from datetime import datetime
18from typing import TYPE_CHECKING, Any
19
20import jwt
21
22from music_assistant.helpers.datetime import utc
23
24if TYPE_CHECKING:
25 from music_assistant_models.auth import User
26
27
28class JWTHelper:
29 """Helper class for JWT token operations."""
30
31 def __init__(self, secret_key: str) -> None:
32 """Initialize JWT helper.
33
34 :param secret_key: Secret key for signing JWTs.
35 """
36 self.secret_key = secret_key
37 self.algorithm = "HS256"
38
39 def encode_token(
40 self,
41 user: User,
42 token_id: str,
43 token_name: str,
44 expires_at: datetime,
45 is_long_lived: bool = False,
46 ) -> str:
47 """Encode a JWT token for a user.
48
49 :param user: User object to create token for.
50 :param token_id: Unique token identifier.
51 :param token_name: Human-readable token name.
52 :param expires_at: Token expiration datetime.
53 :param is_long_lived: Whether this is a long-lived token.
54 :return: Encoded JWT token string.
55 """
56 now = utc()
57 payload = {
58 "sub": user.user_id,
59 "jti": token_id,
60 "iat": int(now.timestamp()),
61 "exp": int(expires_at.timestamp()),
62 "username": user.username,
63 "role": user.role.value,
64 "token_name": token_name,
65 "is_long_lived": is_long_lived,
66 }
67
68 return jwt.encode(payload, self.secret_key, algorithm=self.algorithm)
69
70 def decode_token(self, token: str, verify_exp: bool = True) -> dict[str, Any]:
71 """Decode and verify a JWT token.
72
73 :param token: JWT token string to decode.
74 :param verify_exp: Whether to verify token expiration.
75 :return: Decoded token payload.
76 :raises jwt.InvalidTokenError: If token is invalid or expired.
77 """
78 payload: dict[str, Any] = jwt.decode(
79 token,
80 self.secret_key,
81 algorithms=[self.algorithm],
82 options={"verify_exp": verify_exp},
83 )
84 return payload
85
86 @staticmethod
87 def generate_secret_key() -> str:
88 """Generate a secure random secret key for JWT signing.
89
90 :return: Base64-encoded 256-bit random key.
91 """
92 return secrets.token_urlsafe(32) # 32 bytes = 256 bits
93
94 def get_token_id(self, token: str) -> str | None:
95 """Extract token ID (jti) from JWT without full validation.
96
97 :param token: JWT token string.
98 :return: Token ID or None if invalid.
99 """
100 try:
101 payload: dict[str, Any] = jwt.decode(
102 token,
103 options={"verify_signature": False, "verify_exp": False},
104 )
105 jti = payload.get("jti")
106 return str(jti) if jti else None
107 except Exception:
108 return None
109