Scrameupeutchi/librespot-python
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

move all login5 logic into TokenProvider

Browse Source 
allow for login5 refresh after expiration
This commit is contained in:
Googolplexed0
2025-08-20 22:36:19 -05:00
parent 7481dca654
commit 9a7776667f
1 changed files with 44 additions and 58 deletions
Download Patch File Download Diff File Expand all files Collapse all files

100
librespot/core.py
View File

@@ -910,8 +910,6 @@ class Session(Closeable, MessageListener, SubListener):
__dealer_client: typing.Union[DealerClient, None] = None
__event_service: typing.Union[EventService, None] = None
__keys: DiffieHellman
__login5_access_token: typing.Union[str, None] = None
__login5_token_expiry: typing.Union[int, None] = None
__mercury_client: MercuryClient
__receiver: typing.Union[Receiver, None] = None
__search: typing.Union[SearchManager, None]
@@ -972,7 +970,6 @@ class Session(Closeable, MessageListener, SubListener):
"""
self.__authenticate_partial(credential, False)
self.__authenticate_login5()
with self.__auth_lock:
self.__mercury_client = MercuryClient(self)
self.__token_provider = TokenProvider(self)
@@ -1208,13 +1205,6 @@ class Session(Closeable, MessageListener, SubListener):
self.__wait_auth_lock()
return self.__ap_welcome is not None and self.connection is not None
def login5(self) -> tuple[str, int]:
""" """
self.__wait_auth_lock()
if self.__login5_access_token is None or self.__login5_token_expiry is None:
raise RuntimeError("Session isn't authenticated!")
return self.__login5_access_token, self.__login5_token_expiry
def mercury(self) -> MercuryClient:
""" """
self.__wait_auth_lock()
@@ -1394,43 +1384,6 @@ class Session(Closeable, MessageListener, SubListener):
else:
raise RuntimeError("Unknown CMD 0x" + packet.cmd.hex())
def __authenticate_login5(self) -> None:
"""Authenticate using Login5 to get access token"""
login5_request = Login5.LoginRequest()
login5_request.client_info.client_id = MercuryRequests.keymaster_client_id
login5_request.client_info.device_id = self.__inner.device_id
# Set stored credential from APWelcome
if hasattr(self, '_Session__ap_welcome') and self.__ap_welcome:
stored_cred = Login5Credentials.StoredCredential()
stored_cred.username = self.__ap_welcome.canonical_username
stored_cred.data = self.__ap_welcome.reusable_auth_credentials
login5_request.stored_credential.CopyFrom(stored_cred)
response = requests.post(
"https://login5.spotify.com/v3/login",
data=login5_request.SerializeToString(),
headers={
"Content-Type": "application/x-protobuf",
"Accept": "application/x-protobuf"
}
)
if response.status_code == 200:
login5_response = Login5.LoginResponse()
login5_response.ParseFromString(response.content)
if login5_response.HasField('ok'):
self.__login5_access_token = login5_response.ok.access_token
self.__login5_token_expiry = int(time.time()) + login5_response.ok.access_token_expires_in
self.logger.info("Login5 authentication successful, got access token")
else:
self.logger.warning("Login5 authentication failed: {}".format(login5_response.error))
else:
self.logger.warning("Login5 request failed with status: {}".format(response.status_code))
else:
self.logger.error("Login5 authentication failed: No APWelcome found")
def __send_unchecked(self, cmd: bytes, payload: bytes) -> None:
self.cipher_pair.send_encoded(self.connection, cmd, payload)
@@ -2343,22 +2296,55 @@ class TokenProvider:
if token is not None:
if token.expired():
self.__tokens.remove(token)
self.logger.debug("Login5 token expired, need to re-authenticate")
else:
return token
login5_token = None
login5_access_token, login5_token_expiry = self.__session.login5()
if int(time.time()) < login5_token_expiry - 60: # 60 second buffer
login5_token = TokenProvider.StoredToken({
"expiresIn": login5_token_expiry - int(time.time()),
"accessToken": login5_access_token,
token = self.login5(scopes)
if token is not None:
self.__tokens.append(token)
self.logger.debug("Using Login5 access token for scopes: {}".format(scopes))
return token
def login5(self, scopes: typing.List[str]) -> typing.Union[StoredToken, None]:
"""Submit Login5 request for a fresh access token"""
if self.__session.ap_welcome():
login5_request = Login5.LoginRequest()
login5_request.client_info.client_id = MercuryRequests.keymaster_client_id
login5_request.client_info.device_id = self.__session.device_id()
stored_cred = Login5Credentials.StoredCredential()
stored_cred.username = self.__session.username()
stored_cred.data = self.__session.ap_welcome().reusable_auth_credentials
login5_request.stored_credential.CopyFrom(stored_cred)
response = requests.post(
"https://login5.spotify.com/v3/login",
data=login5_request.SerializeToString(),
headers={
"Content-Type": "application/x-protobuf",
"Accept": "application/x-protobuf"
})
if response.status_code == 200:
login5_response = Login5.LoginResponse()
login5_response.ParseFromString(response.content)
if login5_response.HasField('ok'):
self.logger.info("Login5 authentication successful, got access token".format(login5_response.ok.access_token))
token = TokenProvider.StoredToken({
"expiresIn": login5_response.ok.access_token_expires_in, # approximately one hour
"accessToken": login5_response.ok.access_token,
"scope": scopes
})
self.__tokens.append(login5_token)
self.logger.debug("Using Login5 access token for scopes: {}".format(scopes))
return token
else:
self.logger.debug("Login5 token expired, need to re-authenticate")
return login5_token
self.logger.warning("Login5 authentication failed: {}".format(login5_response.error))
else:
self.logger.warning("Login5 request failed with status: {}".format(response.status_code))
else:
self.logger.error("Login5 authentication failed: No APWelcome found")
class StoredToken:
""" """