JDM170/librespot-spotizerr-dev
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: license + login5

Browse Source
This commit is contained in:
xoconoch
2025-08-30 11:25:07 -06:00
parent 28bfcbf826
commit 8bca880d33
10 changed files with 494 additions and 51 deletions
Download Patch File Download Diff File Expand all files Collapse all files

120
librespot/core.py
View File

@@ -40,7 +40,6 @@ from librespot.crypto import CipherPair
from librespot.crypto import DiffieHellman
from librespot.crypto import Packet
from librespot.mercury import MercuryClient
from librespot.mercury import MercuryRequests
from librespot.mercury import RawMercuryRequest
from librespot.metadata import AlbumId
from librespot.metadata import ArtistId
@@ -56,6 +55,8 @@ from librespot.proto import Keyexchange_pb2 as Keyexchange
from librespot.proto import Metadata_pb2 as Metadata
from librespot.proto import Playlist4External_pb2 as Playlist4External
from librespot.proto.ExplicitContentPubsub_pb2 import UserAttributesUpdate
from librespot.proto.spotify.login5.v3 import Login5_pb2 as Login5
from librespot.proto.spotify.login5.v3.credentials import Credentials_pb2 as Login5Credentials
from librespot.structure import Closeable
from librespot.structure import MessageListener
from librespot.structure import RequestListener
@@ -203,7 +204,7 @@ class ApiClient(Closeable):
proto_req = ClientToken.ClientTokenRequest(
request_type=ClientToken.ClientTokenRequestType.REQUEST_CLIENT_DATA_REQUEST,
client_data=ClientToken.ClientDataRequest(
client_id=MercuryRequests.keymaster_client_id,
client_id=Session.CLIENT_ID,
client_version=Version.version_name,
connectivity_sdk_data=Connectivity.ConnectivitySdkData(
device_id=self.__session.device_id(),
@@ -781,6 +782,7 @@ class MessageType(enum.Enum):
class Session(Closeable, MessageListener, SubListener):
""" """
CLIENT_ID = "65b708073fc0480ea92a077233ca87bd" # Spotify client ID for librespot
cipher_pair: typing.Union[CipherPair, None]
country_code: str = "EN"
connection: typing.Union[ConnectionHolder, None]
@@ -823,6 +825,8 @@ class Session(Closeable, MessageListener, SubListener):
__stored_str: str = ""
__token_provider: typing.Union[TokenProvider, None]
__user_attributes = {}
__login5_access_token: typing.Union[str, None] = None
__login5_token_expiry: typing.Union[int, None] = None
def __init__(self, inner: Inner, address: str) -> None:
self.__client = Session.create_client(inner.conf)
@@ -862,6 +866,10 @@ class Session(Closeable, MessageListener, SubListener):
"""
self.__authenticate_partial(credential, False)
# Try Login5 authentication for access token
self.__authenticate_login5(credential)
with self.__auth_lock:
self.__mercury_client = MercuryClient(self)
self.__token_provider = TokenProvider(self)
@@ -1279,6 +1287,66 @@ class Session(Closeable, MessageListener, SubListener):
def __send_unchecked(self, cmd: bytes, payload: bytes) -> None:
self.cipher_pair.send_encoded(self.connection, cmd, payload)
def __authenticate_login5(self, credential: Authentication.LoginCredentials) -> None:
"""Authenticate using Login5 to get access token"""
try:
# Build Login5 request
login5_request = Login5.LoginRequest()
# Set client info
login5_request.client_info.client_id = Session.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)
# Send Login5 request
login5_url = "https://login5.spotify.com/v3/login"
headers = {
"Content-Type": "application/x-protobuf",
"Accept": "application/x-protobuf"
}
response = requests.post(
login5_url,
data=login5_request.SerializeToString(),
headers=headers
)
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:
error_msg = "Login5 authentication failed"
if login5_response.HasField('error'):
error_msg += f": {login5_response.error}"
self.logger.error(error_msg)
# Don't raise exception here, as the session can still work with some limitations
else:
self.logger.error("Login5 request failed with status: {}".format(response.status_code))
# Don't raise exception here, as the session can still work with some limitations
except Exception as e:
self.logger.error("Failed to authenticate with Login5: {}".format(e))
# Don't raise exception here, as the session can still work with some limitations
def get_login5_token(self) -> typing.Union[str, None]:
"""Get the Login5 access token if available and not expired"""
if self.__login5_access_token and self.__login5_token_expiry:
if int(time.time()) < self.__login5_token_expiry - 60: # 60 second buffer
return self.__login5_access_token
else:
self.logger.debug("Login5 token expired, need to re-authenticate")
return None
def __wait_auth_lock(self) -> None:
if self.__closing and self.connection is None:
self.logger.debug("Connection was broken while closing.")
@@ -1921,7 +1989,7 @@ class Session(Closeable, MessageListener, SubListener):
format(util.bytes_to_hex(packet.cmd),
packet.payload))
continue
except (RuntimeError, ConnectionResetError) as ex:
except (RuntimeError, ConnectionResetError, OSError) as ex:
if self.__running:
self.__session.logger.fatal(
"Failed reading packet! {}".format(ex))
@@ -2171,24 +2239,46 @@ class TokenProvider:
scopes = list(scopes)
if len(scopes) == 0:
raise RuntimeError("The token doesn't have any scope")
# Use Login5 token
login5_token = self._session.get_login5_token()
if login5_token:
# Create a StoredToken-compatible object using Login5 token
login5_stored_token = TokenProvider.Login5StoredToken(login5_token, scopes)
self.logger.debug("Using Login5 access token for scopes: {}".format(scopes))
return login5_stored_token
# Check existing tokens
token = self.find_token_with_all_scopes(scopes)
if token is not None:
if token.expired():
self.__tokens.remove(token)
else:
return token
self.logger.debug(
"Token expired or not suitable, requesting again. scopes: {}, old_token: {}"
.format(scopes, token))
response = self._session.mercury().send_sync_json(
MercuryRequests.request_token(self._session.device_id(),
",".join(scopes)))
token = TokenProvider.StoredToken(response)
self.logger.debug(
"Updated token successfully! scopes: {}, new_token: {}".format(
scopes, token))
self.__tokens.append(token)
return token
# No valid token available
raise RuntimeError("Unable to obtain access token. Login5 authentication may have failed.")
class Login5StoredToken:
"""StoredToken-compatible wrapper for Login5 access tokens"""
access_token: str
scopes: typing.List[str]
def __init__(self, access_token: str, scopes: typing.List[str]):
self.access_token = access_token
self.scopes = scopes
def expired(self) -> bool:
"""Login5 tokens are managed by Session, so delegate expiry check"""
return False # Session handles expiry
def has_scope(self, scope: str) -> bool:
"""Login5 tokens are general-purpose, assume they have all scopes"""
return True
def has_scopes(self, sc: typing.List[str]) -> bool:
"""Login5 tokens are general-purpose, assume they have all scopes"""
return True
class StoredToken:
""" """

7
librespot/mercury.py
View File

@@ -299,13 +299,6 @@ class MercuryRequests:
@TODO implement function
"""
@staticmethod
def request_token(device_id, scope):
return JsonMercuryRequest(
RawMercuryRequest.get(
"hm://keymaster/token/authenticated?scope={}&client_id={}&device_id={}"
.format(scope, MercuryRequests.keymaster_client_id,
device_id)))
class RawMercuryRequest:

24
librespot/proto/spotify/login5/v3/Login5_pb2.py
View File

@@ -7,19 +7,19 @@ from google.protobuf import message as _message
from google.protobuf import reflection as _reflection
from google.protobuf import symbol_database as _symbol_database
from google.protobuf.internal import enum_type_wrapper
from spotify.login5.v3 import \
client_info_pb2 as spotify_dot_login5_dot_v3_dot_client__info__pb2
from spotify.login5.v3 import \
user_info_pb2 as spotify_dot_login5_dot_v3_dot_user__info__pb2
from spotify.login5.v3.challenges import \
code_pb2 as spotify_dot_login5_dot_v3_dot_challenges_dot_code__pb2
from spotify.login5.v3.challenges import \
hashcash_pb2 as spotify_dot_login5_dot_v3_dot_challenges_dot_hashcash__pb2
from spotify.login5.v3.credentials import \
credentials_pb2 as \
from librespot.proto.spotify.login5.v3 import \
ClientInfo_pb2 as spotify_dot_login5_dot_v3_dot_client__info__pb2
from librespot.proto.spotify.login5.v3 import \
UserInfo_pb2 as spotify_dot_login5_dot_v3_dot_user__info__pb2
from librespot.proto.spotify.login5.v3.challenges import \
Code_pb2 as spotify_dot_login5_dot_v3_dot_challenges_dot_code__pb2
from librespot.proto.spotify.login5.v3.challenges import \
Hashcash_pb2 as spotify_dot_login5_dot_v3_dot_challenges_dot_hashcash__pb2
from librespot.proto.spotify.login5.v3.credentials import \
Credentials_pb2 as \
spotify_dot_login5_dot_v3_dot_credentials_dot_credentials__pb2
from spotify.login5.v3.identifiers import \
identifiers_pb2 as \
from librespot.proto.spotify.login5.v3.identifiers import \
Identifiers as \
spotify_dot_login5_dot_v3_dot_identifiers_dot_identifiers__pb2
# @@protoc_insertion_point(imports)