rdelaage
/
librespot-python
mirror of https://github.com/kokarare1212/librespot-python.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Restyled by yapf

This commit is contained in:
Restyled.io 2023-09-05 07:21:12 +00:00 committed by d.rathmer
parent 12621ef31c
commit ecd0006874
1 changed files with 223 additions and 253 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

476
librespot/core.py
View File

@ -94,9 +94,8 @@ class ApiClient(Closeable):
if self.__client_token_str is None: if self.__client_token_str is None:
resp = self.__client_token() resp = self.__client_token()
self.__client_token_str = resp.granted_token.token self.__client_token_str = resp.granted_token.token
self.logger.debug( self.logger.debug("Updated client token: {}".format(
"Updated client token: {}".format(self.__client_token_str) self.__client_token_str))
)
request = requests.PreparedRequest() request = requests.PreparedRequest()
request.method = method request.method = method
@ -105,8 +104,7 @@ class ApiClient(Closeable):
if headers is not None: if headers is not None:
request.headers = headers request.headers = headers
request.headers["Authorization"] = "Bearer {}".format( request.headers["Authorization"] = "Bearer {}".format(
self.__session.tokens().get("playlist-read") self.__session.tokens().get("playlist-read"))
)
request.headers["client-token"] = self.__client_token_str request.headers["client-token"] = self.__client_token_str
request.url = self.__base_url + suffix request.url = self.__base_url + suffix
return request return request
@ -130,13 +128,11 @@ class ApiClient(Closeable):
""" """
response = self.__session.client().send( response = self.__session.client().send(
self.build_request(method, suffix, headers, body) self.build_request(method, suffix, headers, body))
)
return response return response
def put_connect_state( def put_connect_state(self, connection_id: str,
self, connection_id: str, proto: Connect.PutStateRequest proto: Connect.PutStateRequest) -> None:
) -> None:
""" """
:param connection_id: str: :param connection_id: str:
@ -154,16 +150,11 @@ class ApiClient(Closeable):
) )
if response.status_code == 413: if response.status_code == 413:
self.logger.warning( self.logger.warning(
"PUT state payload is too large: {} bytes uncompressed.".format( "PUT state payload is too large: {} bytes uncompressed.".
len(proto.SerializeToString()) format(len(proto.SerializeToString())))
)
)
elif response.status_code != 200: elif response.status_code != 200:
self.logger.warning( self.logger.warning("PUT state returned {}. headers: {}".format(
"PUT state returned {}. headers: {}".format( response.status_code, response.headers))
response.status_code, response.headers
)
)
def get_metadata_4_track(self, track: TrackId) -> Metadata.Track: def get_metadata_4_track(self, track: TrackId) -> Metadata.Track:
""" """
@ -171,9 +162,9 @@ class ApiClient(Closeable):
:param track: TrackId: :param track: TrackId:
""" """
response = self.send( response = self.send("GET",
"GET", "/metadata/4/track/{}".format(track.hex_id()), None, None "/metadata/4/track/{}".format(track.hex_id()),
) None, None)
ApiClient.StatusCodeException.check_status(response) ApiClient.StatusCodeException.check_status(response)
body = response.content body = response.content
if body is None: if body is None:
@ -188,9 +179,9 @@ class ApiClient(Closeable):
:param episode: EpisodeId: :param episode: EpisodeId:
""" """
response = self.send( response = self.send("GET",
"GET", "/metadata/4/episode/{}".format(episode.hex_id()), None, None "/metadata/4/episode/{}".format(episode.hex_id()),
) None, None)
ApiClient.StatusCodeException.check_status(response) ApiClient.StatusCodeException.check_status(response)
body = response.content body = response.content
if body is None: if body is None:
@ -205,9 +196,9 @@ class ApiClient(Closeable):
:param album: AlbumId: :param album: AlbumId:
""" """
response = self.send( response = self.send("GET",
"GET", "/metadata/4/album/{}".format(album.hex_id()), None, None "/metadata/4/album/{}".format(album.hex_id()),
) None, None)
ApiClient.StatusCodeException.check_status(response) ApiClient.StatusCodeException.check_status(response)
body = response.content body = response.content
@ -223,9 +214,9 @@ class ApiClient(Closeable):
:param artist: ArtistId: :param artist: ArtistId:
""" """
response = self.send( response = self.send("GET",
"GET", "/metadata/4/artist/{}".format(artist.hex_id()), None, None "/metadata/4/artist/{}".format(artist.hex_id()),
) None, None)
ApiClient.StatusCodeException.check_status(response) ApiClient.StatusCodeException.check_status(response)
body = response.content body = response.content
if body is None: if body is None:
@ -240,9 +231,9 @@ class ApiClient(Closeable):
:param show: ShowId: :param show: ShowId:
""" """
response = self.send( response = self.send("GET",
"GET", "/metadata/4/show/{}".format(show.hex_id()), None, None "/metadata/4/show/{}".format(show.hex_id()), None,
) None)
ApiClient.StatusCodeException.check_status(response) ApiClient.StatusCodeException.check_status(response)
body = response.content body = response.content
if body is None: if body is None:
@ -251,15 +242,16 @@ class ApiClient(Closeable):
proto.ParseFromString(body) proto.ParseFromString(body)
return proto return proto
def get_playlist(self, _id: PlaylistId) -> Playlist4External.SelectedListContent: def get_playlist(self,
_id: PlaylistId) -> Playlist4External.SelectedListContent:
""" """
:param _id: PlaylistId: :param _id: PlaylistId:
""" """
response = self.send( response = self.send("GET",
"GET", "/playlist/v2/playlist/{}".format(_id.id()), None, None "/playlist/v2/playlist/{}".format(_id.id()), None,
) None)
ApiClient.StatusCodeException.check_status(response) ApiClient.StatusCodeException.check_status(response)
body = response.content body = response.content
if body is None: if body is None:
@ -278,7 +270,8 @@ class ApiClient(Closeable):
def __client_token(self): def __client_token(self):
proto_req = ClientToken.ClientTokenRequest( proto_req = ClientToken.ClientTokenRequest(
request_type=ClientToken.ClientTokenRequestType.REQUEST_CLIENT_DATA_REQUEST, request_type=ClientToken.ClientTokenRequestType.
REQUEST_CLIENT_DATA_REQUEST,
client_data=ClientToken.ClientDataRequest( client_data=ClientToken.ClientDataRequest(
client_id=MercuryRequests.keymaster_client_id, client_id=MercuryRequests.keymaster_client_id,
client_version=Version.version_name, client_version=Version.version_name,
@ -293,8 +286,7 @@ class ApiClient(Closeable):
something7=332, something7=332,
something8=33404, something8=33404,
something10=True, something10=True,
), ), ),
),
), ),
), ),
) )
@ -345,7 +337,8 @@ class ApResolver:
:returns: The resulting object will be returned :returns: The resulting object will be returned
""" """
response = requests.get("{}?type={}".format(ApResolver.base_url, service_type)) response = requests.get("{}?type={}".format(ApResolver.base_url,
service_type))
if response.status_code != 200: if response.status_code != 200:
if response.status_code == 502: if response.status_code == 502:
raise RuntimeError( raise RuntimeError(
@ -414,7 +407,8 @@ class DealerClient(Closeable):
def __init__(self, session: Session): def __init__(self, session: Session):
self.__session = session self.__session = session
def add_message_listener(self, listener: MessageListener, uris: list[str]) -> None: def add_message_listener(self, listener: MessageListener,
uris: list[str]) -> None:
""" """
:param listener: MessageListener: :param listener: MessageListener:
@ -424,8 +418,7 @@ class DealerClient(Closeable):
with self.__message_listeners_lock: with self.__message_listeners_lock:
if listener in self.__message_listeners: if listener in self.__message_listeners:
raise TypeError( raise TypeError(
"A listener for {} has already been added.".format(uris) "A listener for {} has already been added.".format(uris))
)
self.__message_listeners[listener] = uris self.__message_listeners[listener] = uris
self.__message_listeners_lock.notify_all() self.__message_listeners_lock.notify_all()
@ -439,8 +432,7 @@ class DealerClient(Closeable):
with self.__request_listeners_lock: with self.__request_listeners_lock:
if uri in self.__request_listeners: if uri in self.__request_listeners:
raise TypeError( raise TypeError(
"A listener for '{}' has already been added.".format(uri) "A listener for '{}' has already been added.".format(uri))
)
self.__request_listeners[uri] = listener self.__request_listeners[uri] = listener
self.__request_listeners_lock.notify_all() self.__request_listeners_lock.notify_all()
@ -469,7 +461,8 @@ class DealerClient(Closeable):
self.__last_scheduled_reconnection = None self.__last_scheduled_reconnection = None
self.connect() self.connect()
self.__last_scheduled_reconnection = self.__scheduler.enter(10, 1, anonymous) self.__last_scheduled_reconnection = self.__scheduler.enter(
10, 1, anonymous)
def handle_message(self, obj: typing.Any) -> None: def handle_message(self, obj: typing.Any) -> None:
""" """
@ -527,10 +520,8 @@ class DealerClient(Closeable):
sender = payload.get("sent_by_device_id") sender = payload.get("sent_by_device_id")
command = payload.get("command") command = payload.get("command")
self.logger.debug( self.logger.debug(
"Received request. [mid: {}, key: {}, pid: {}, sender: {}, command: {}]".format( "Received request. [mid: {}, key: {}, pid: {}, sender: {}, command: {}]"
mid, key, pid, sender, command .format(mid, key, pid, sender, command))
)
)
interesting = False interesting = False
with self.__request_listeners_lock: with self.__request_listeners_lock:
for mid_prefix in self.__request_listeners: for mid_prefix in self.__request_listeners:
@ -544,8 +535,8 @@ class DealerClient(Closeable):
if self.__connection is not None: if self.__connection is not None:
self.__connection.send_reply(key, result) self.__connection.send_reply(key, result)
self.logger.warning( self.logger.warning(
"Handled request. [key: {}, result: {}]".format(key, result) "Handled request. [key: {}, result: {}]".format(
) key, result))
self.__worker.submit(anonymous) self.__worker.submit(anonymous)
if not interesting: if not interesting:
@ -597,7 +588,8 @@ class DealerClient(Closeable):
__url: str __url: str
__ws: websocket.WebSocketApp __ws: websocket.WebSocketApp
def __init__(self, session: Session, dealer_client: DealerClient, url: str): def __init__(self, session: Session, dealer_client: DealerClient,
url: str):
self.__session = session self.__session = session
self.__dealer_client = dealer_client self.__dealer_client = dealer_client
self.__url = url self.__url = url
@ -621,8 +613,7 @@ class DealerClient(Closeable):
if self.__closed: if self.__closed:
return return
self.__dealer_client.logger.warning( self.__dealer_client.logger.warning(
"An exception occurred. Reconnecting..." "An exception occurred. Reconnecting...")
)
self.close() self.close()
def on_message(self, ws: websocket.WebSocketApp, text: str): def on_message(self, ws: websocket.WebSocketApp, text: str):
@ -644,7 +635,8 @@ class DealerClient(Closeable):
elif typ == MessageType.PING: elif typ == MessageType.PING:
pass pass
else: else:
raise RuntimeError("Unknown message type for {}".format(typ.value)) raise RuntimeError("Unknown message type for {}".format(
typ.value))
def on_open(self, ws: websocket.WebSocketApp): def on_open(self, ws: websocket.WebSocketApp):
""" """
@ -654,13 +646,10 @@ class DealerClient(Closeable):
""" """
if self.__closed: if self.__closed:
self.__dealer_client.logger.fatal( self.__dealer_client.logger.fatal(
"I wonder what happened here... Terminating. [closed: {}]".format( "I wonder what happened here... Terminating. [closed: {}]".
self.__closed format(self.__closed))
)
)
self.__dealer_client.logger.debug( self.__dealer_client.logger.debug(
"Dealer connected! [url: {}]".format(self.__url) "Dealer connected! [url: {}]".format(self.__url))
)
def anonymous(): def anonymous():
""" """ """ """
@ -680,9 +669,11 @@ class DealerClient(Closeable):
self.__received_pong = False self.__received_pong = False
self.__scheduler.enter(3, 1, anonymous2) self.__scheduler.enter(3, 1, anonymous2)
self.__last_scheduled_ping = self.__scheduler.enter(30, 1, anonymous) self.__last_scheduled_ping = self.__scheduler.enter(
30, 1, anonymous)
self.__last_scheduled_ping = self.__scheduler.enter(30, 1, anonymous) self.__last_scheduled_ping = self.__scheduler.enter(
30, 1, anonymous)
def send_ping(self): def send_ping(self):
""" """ """ """
@ -695,12 +686,11 @@ class DealerClient(Closeable):
:param result: DealerClient.RequestResult: :param result: DealerClient.RequestResult:
""" """
success = ( success = ("true" if result == DealerClient.RequestResult.SUCCESS
"true" if result == DealerClient.RequestResult.SUCCESS else "false" else "false")
)
self.__ws.send( self.__ws.send(
'{"type":"reply","key":"%s","payload":{"success":%s}' % (key, success) '{"type":"reply","key":"%s","payload":{"success":%s}' %
) (key, success))
class RequestResult(enum.Enum): class RequestResult(enum.Enum):
""" """ """ """
@ -727,21 +717,19 @@ class EventService(Closeable):
try: try:
body = event_builder.to_array() body = event_builder.to_array()
resp = self.__session.mercury().send_sync( resp = self.__session.mercury().send_sync(
RawMercuryRequest.Builder() RawMercuryRequest.Builder().set_uri(
.set_uri("hm://event-service/v1/events") "hm://event-service/v1/events").set_method("POST").
.set_method("POST") add_user_field("Accept-Language", "en").add_user_field(
.add_user_field("Accept-Language", "en") "X-ClientTimeStamp",
.add_user_field("X-ClientTimeStamp", int(time.time() * 1000)) int(time.time() * 1000)).add_payload_part(body).build())
.add_payload_part(body) self.logger.debug("Event sent. body: {}, result: {}".format(
.build() body, resp.status_code))
)
self.logger.debug(
"Event sent. body: {}, result: {}".format(body, resp.status_code)
)
except IOError as ex: except IOError as ex:
self.logger.error("Failed sending event: {} {}".format(event_builder, ex)) self.logger.error("Failed sending event: {} {}".format(
event_builder, ex))
def send_event(self, event_or_builder: typing.Union[GenericEvent, EventBuilder]): def send_event(self, event_or_builder: typing.Union[GenericEvent,
EventBuilder]):
""" """
:param event_or_builder: typing.Union[GenericEvent: :param event_or_builder: typing.Union[GenericEvent:
@ -788,6 +776,7 @@ class EventService(Closeable):
class GenericEvent: class GenericEvent:
""" """ """ """
def build(self) -> EventService.EventBuilder: def build(self) -> EventService.EventBuilder:
""" """ """ """
raise NotImplementedError raise NotImplementedError
@ -811,7 +800,9 @@ class EventService(Closeable):
s = "" s = ""
self.body.write(s.encode()) self.body.write(s.encode())
def append(self, c: int = None, s: str = None) -> EventService.EventBuilder: def append(self,
c: int = None,
s: str = None) -> EventService.EventBuilder:
""" """
:param c: int: (Default value = None) :param c: int: (Default value = None)
@ -889,23 +880,21 @@ class Session(Closeable, MessageListener, SubListener):
__mercury_client: MercuryClient __mercury_client: MercuryClient
__receiver: typing.Union[Receiver, None] = None __receiver: typing.Union[Receiver, None] = None
__search: typing.Union[SearchManager, None] __search: typing.Union[SearchManager, None]
__server_key = ( __server_key = (b"\xac\xe0F\x0b\xff\xc20\xaf\xf4k\xfe\xc3\xbf\xbf\x86="
b"\xac\xe0F\x0b\xff\xc20\xaf\xf4k\xfe\xc3\xbf\xbf\x86=" b"\xa1\x91\xc6\xcc3l\x93\xa1O\xb3\xb0\x16\x12\xac\xacj"
b"\xa1\x91\xc6\xcc3l\x93\xa1O\xb3\xb0\x16\x12\xac\xacj" b"\xf1\x80\xe7\xf6\x14\xd9B\x9d\xbe.4fC\xe3b\xd22z\x1a"
b"\xf1\x80\xe7\xf6\x14\xd9B\x9d\xbe.4fC\xe3b\xd22z\x1a" b"\r\x92;\xae\xdd\x14\x02\xb1\x81U\x05a\x04\xd5,\x96\xa4"
b"\r\x92;\xae\xdd\x14\x02\xb1\x81U\x05a\x04\xd5,\x96\xa4" b"L\x1e\xcc\x02J\xd4\xb2\x0c\x00\x1f\x17\xed\xc2/\xc45"
b"L\x1e\xcc\x02J\xd4\xb2\x0c\x00\x1f\x17\xed\xc2/\xc45" b"!\xc8\xf0\xcb\xae\xd2\xad\xd7+\x0f\x9d\xb3\xc52\x1a*"
b"!\xc8\xf0\xcb\xae\xd2\xad\xd7+\x0f\x9d\xb3\xc52\x1a*" b"\xfeY\xf3Z\r\xach\xf1\xfab\x1e\xfb,\x8d\x0c\xb79-\x92"
b"\xfeY\xf3Z\r\xach\xf1\xfab\x1e\xfb,\x8d\x0c\xb79-\x92" b"G\xe3\xd75\x1am\xbd$\xc2\xae%[\x88\xff\xabs)\x8a\x0b"
b"G\xe3\xd75\x1am\xbd$\xc2\xae%[\x88\xff\xabs)\x8a\x0b" b"\xcc\xcd\x0cXg1\x89\xe8\xbd4\x80xJ_\xc9k\x89\x9d\x95k"
b"\xcc\xcd\x0cXg1\x89\xe8\xbd4\x80xJ_\xc9k\x89\x9d\x95k" b"\xfc\x86\xd7O3\xa6x\x17\x96\xc9\xc3-\r2\xa5\xab\xcd\x05'"
b"\xfc\x86\xd7O3\xa6x\x17\x96\xc9\xc3-\r2\xa5\xab\xcd\x05'" b"\xe2\xf7\x10\xa3\x96\x13\xc4/\x99\xc0'\xbf\xed\x04\x9c"
b"\xe2\xf7\x10\xa3\x96\x13\xc4/\x99\xc0'\xbf\xed\x04\x9c" b"<'X\x04\xb6\xb2\x19\xf9\xc1/\x02\xe9Hc\xec\xa1\xb6B\xa0"
b"<'X\x04\xb6\xb2\x19\xf9\xc1/\x02\xe9Hc\xec\xa1\xb6B\xa0" b"\x9dH%\xf8\xb3\x9d\xd0\xe8j\xf9HM\xa1\xc2\xba\x860B\xea"
b"\x9dH%\xf8\xb3\x9d\xd0\xe8j\xf9HM\xa1\xc2\xba\x860B\xea" b"\x9d\xb3\x08l\x19\x0eH\xb3\x9df\xeb\x00\x06\xa2Z\xee\xa1"
b"\x9d\xb3\x08l\x19\x0eH\xb3\x9df\xeb\x00\x06\xa2Z\xee\xa1" b"\x1b\x13\x87<\xd7\x19\xe6U\xbd")
b"\x1b\x13\x87<\xd7\x19\xe6U\xbd"
)
__stored_str: str = "" __stored_str: str = ""
__token_provider: typing.Union[TokenProvider, None] __token_provider: typing.Union[TokenProvider, None]
__user_attributes = {} __user_attributes = {}
@ -915,11 +904,8 @@ class Session(Closeable, MessageListener, SubListener):
self.connection = Session.ConnectionHolder.create(address, None) self.connection = Session.ConnectionHolder.create(address, None)
self.__inner = inner self.__inner = inner
self.__keys = DiffieHellman() self.__keys = DiffieHellman()
self.logger.info( self.logger.info("Created new session! device_id: {}, ap: {}".format(
"Created new session! device_id: {}, ap: {}".format( inner.device_id, address))
inner.device_id, address
)
)
def api(self) -> ApiClient: def api(self) -> ApiClient:
""" """ """ """
@ -942,7 +928,8 @@ class Session(Closeable, MessageListener, SubListener):
raise RuntimeError("Session isn't authenticated!") raise RuntimeError("Session isn't authenticated!")
return self.__audio_key_manager return self.__audio_key_manager
def authenticate(self, credential: Authentication.LoginCredentials) -> None: def authenticate(self,
credential: Authentication.LoginCredentials) -> None:
"""Log in to Spotify """Log in to Spotify
:param credential: Spotify account login information :param credential: Spotify account login information
@ -965,13 +952,11 @@ class Session(Closeable, MessageListener, SubListener):
self.__auth_lock_bool = False self.__auth_lock_bool = False
self.__auth_lock.notify_all() self.__auth_lock.notify_all()
self.dealer().connect() self.dealer().connect()
self.logger.info( self.logger.info("Authenticated as {}!".format(
"Authenticated as {}!".format(self.__ap_welcome.canonical_username) self.__ap_welcome.canonical_username))
)
self.mercury().interested_in("spotify:user:attributes:update", self) self.mercury().interested_in("spotify:user:attributes:update", self)
self.dealer().add_message_listener( self.dealer().add_message_listener(
self, ["hm://connect-state/v1/connect/logout"] self, ["hm://connect-state/v1/connect/logout"])
)
def cache(self) -> CacheManager: def cache(self) -> CacheManager:
""" """ """ """
@ -1000,9 +985,8 @@ class Session(Closeable, MessageListener, SubListener):
def close(self) -> None: def close(self) -> None:
"""Close instance""" """Close instance"""
self.logger.info( self.logger.info("Closing session. device_id: {}".format(
"Closing session. device_id: {}".format(self.__inner.device_id) self.__inner.device_id))
)
self.__closing = True self.__closing = True
if self.__dealer_client is not None: if self.__dealer_client is not None:
self.__dealer_client.close() self.__dealer_client.close()
@ -1028,7 +1012,8 @@ class Session(Closeable, MessageListener, SubListener):
self.__ap_welcome = None self.__ap_welcome = None
self.cipher_pair = None self.cipher_pair = None
self.__closed = True self.__closed = True
self.logger.info("Closed session. device_id: {}".format(self.__inner.device_id)) self.logger.info("Closed session. device_id: {}".format(
self.__inner.device_id))
def connect(self) -> None: def connect(self) -> None:
"""Connect to the Spotify Server""" """Connect to the Spotify Server"""
@ -1038,12 +1023,12 @@ class Session(Closeable, MessageListener, SubListener):
client_hello_proto = Keyexchange.ClientHello( client_hello_proto = Keyexchange.ClientHello(
build_info=Version.standard_build_info(), build_info=Version.standard_build_info(),
client_nonce=nonce, client_nonce=nonce,
cryptosuites_supported=[Keyexchange.Cryptosuite.CRYPTO_SUITE_SHANNON], cryptosuites_supported=[
Keyexchange.Cryptosuite.CRYPTO_SUITE_SHANNON
],
login_crypto_hello=Keyexchange.LoginCryptoHelloUnion( login_crypto_hello=Keyexchange.LoginCryptoHelloUnion(
diffie_hellman=Keyexchange.LoginCryptoDiffieHellmanHello( diffie_hellman=Keyexchange.LoginCryptoDiffieHellmanHello(
gc=self.__keys.public_key_bytes(), server_keys_known=1 gc=self.__keys.public_key_bytes(), server_keys_known=1), ),
),
),
padding=b"\x1e", padding=b"\x1e",
) )
client_hello_bytes = client_hello_proto.SerializeToString() client_hello_bytes = client_hello_proto.SerializeToString()
@ -1057,25 +1042,25 @@ class Session(Closeable, MessageListener, SubListener):
# Read APResponseMessage # Read APResponseMessage
ap_response_message_length = self.connection.read_int() ap_response_message_length = self.connection.read_int()
acc.write_int(ap_response_message_length) acc.write_int(ap_response_message_length)
ap_response_message_bytes = self.connection.read(ap_response_message_length - 4) ap_response_message_bytes = self.connection.read(
ap_response_message_length - 4)
acc.write(ap_response_message_bytes) acc.write(ap_response_message_bytes)
ap_response_message_proto = Keyexchange.APResponseMessage() ap_response_message_proto = Keyexchange.APResponseMessage()
ap_response_message_proto.ParseFromString(ap_response_message_bytes) ap_response_message_proto.ParseFromString(ap_response_message_bytes)
shared_key = util.int_to_bytes( shared_key = util.int_to_bytes(
self.__keys.compute_shared_key( self.__keys.compute_shared_key(
ap_response_message_proto.challenge.login_crypto_challenge.diffie_hellman.gs ap_response_message_proto.challenge.login_crypto_challenge.
) diffie_hellman.gs))
)
# Check gs_signature # Check gs_signature
rsa = RSA.construct((int.from_bytes(self.__server_key, "big"), 65537)) rsa = RSA.construct((int.from_bytes(self.__server_key, "big"), 65537))
pkcs1_v1_5 = PKCS1_v1_5.new(rsa) pkcs1_v1_5 = PKCS1_v1_5.new(rsa)
sha1 = SHA1.new() sha1 = SHA1.new()
sha1.update( sha1.update(ap_response_message_proto.challenge.login_crypto_challenge.
ap_response_message_proto.challenge.login_crypto_challenge.diffie_hellman.gs diffie_hellman.gs)
)
if not pkcs1_v1_5.verify( if not pkcs1_v1_5.verify(
sha1, sha1,
ap_response_message_proto.challenge.login_crypto_challenge.diffie_hellman.gs_signature, ap_response_message_proto.challenge.login_crypto_challenge.
diffie_hellman.gs_signature,
): ):
raise RuntimeError("Failed signature check!") raise RuntimeError("Failed signature check!")
# Solve challenge # Solve challenge
@ -1093,14 +1078,11 @@ class Session(Closeable, MessageListener, SubListener):
crypto_response=Keyexchange.CryptoResponseUnion(), crypto_response=Keyexchange.CryptoResponseUnion(),
login_crypto_response=Keyexchange.LoginCryptoResponseUnion( login_crypto_response=Keyexchange.LoginCryptoResponseUnion(
diffie_hellman=Keyexchange.LoginCryptoDiffieHellmanResponse( diffie_hellman=Keyexchange.LoginCryptoDiffieHellmanResponse(
hmac=challenge hmac=challenge)),
)
),
pow_response=Keyexchange.PoWResponseUnion(), pow_response=Keyexchange.PoWResponseUnion(),
) )
client_response_plaintext_bytes = ( client_response_plaintext_bytes = (
client_response_plaintext_proto.SerializeToString() client_response_plaintext_proto.SerializeToString())
)
self.connection.write_int(4 + len(client_response_plaintext_bytes)) self.connection.write_int(4 + len(client_response_plaintext_bytes))
self.connection.write(client_response_plaintext_bytes) self.connection.write(client_response_plaintext_bytes)
self.connection.flush() self.connection.flush()
@ -1108,7 +1090,8 @@ class Session(Closeable, MessageListener, SubListener):
self.connection.set_timeout(1) self.connection.set_timeout(1)
scrap = self.connection.read(4) scrap = self.connection.read(4)
if len(scrap) == 4: if len(scrap) == 4:
payload = self.connection.read(struct.unpack(">i", scrap)[0] - 4) payload = self.connection.read(
struct.unpack(">i", scrap)[0] - 4)
failed = Keyexchange.APResponseMessage() failed = Keyexchange.APResponseMessage()
failed.ParseFromString(payload) failed.ParseFromString(payload)
raise RuntimeError(failed) raise RuntimeError(failed)
@ -1169,9 +1152,8 @@ class Session(Closeable, MessageListener, SubListener):
attributes_update.ParseFromString(resp.payload) attributes_update.ParseFromString(resp.payload)
for pair in attributes_update.pairs_list: for pair in attributes_update.pairs_list:
self.__user_attributes[pair.key] = pair.value self.__user_attributes[pair.key] = pair.value
self.logger.info( self.logger.info("Updated user attribute: {} -> {}".format(
"Updated user attribute: {} -> {}".format(pair.key, pair.value) pair.key, pair.value))
)
def get_user_attribute(self, key: str, fallback: str = None) -> str: def get_user_attribute(self, key: str, fallback: str = None) -> str:
""" """
@ -1180,11 +1162,8 @@ class Session(Closeable, MessageListener, SubListener):
:param fallback: str: (Default value = None) :param fallback: str: (Default value = None)
""" """
return ( return (self.__user_attributes.get(key)
self.__user_attributes.get(key) if self.__user_attributes.get(key) is not None else fallback)
if self.__user_attributes.get(key) is not None
else fallback
)
def is_valid(self) -> bool: def is_valid(self) -> bool:
""" """ """ """
@ -1200,7 +1179,8 @@ class Session(Closeable, MessageListener, SubListener):
raise RuntimeError("Session isn't authenticated!") raise RuntimeError("Session isn't authenticated!")
return self.__mercury_client return self.__mercury_client
def on_message(self, uri: str, headers: typing.Dict[str, str], payload: bytes): def on_message(self, uri: str, headers: typing.Dict[str, str],
payload: bytes):
""" """
:param uri: str: :param uri: str:
@ -1226,7 +1206,8 @@ class Session(Closeable, MessageListener, SubListener):
return return
for i in range(len(product)): for i in range(len(product)):
self.__user_attributes[product[i].tag] = product[i].text self.__user_attributes[product[i].tag] = product[i].text
self.logger.debug("Parsed product info: {}".format(self.__user_attributes)) self.logger.debug("Parsed product info: {}".format(
self.__user_attributes))
def preferred_locale(self) -> str: def preferred_locale(self) -> str:
""" """ """ """
@ -1238,8 +1219,7 @@ class Session(Closeable, MessageListener, SubListener):
self.connection.close() self.connection.close()
self.__receiver.stop() self.__receiver.stop()
self.connection = Session.ConnectionHolder.create( self.connection = Session.ConnectionHolder.create(
ApResolver.get_random_accesspoint(), self.__inner.conf ApResolver.get_random_accesspoint(), self.__inner.conf)
)
self.connect() self.connect()
self.__authenticate_partial( self.__authenticate_partial(
Authentication.LoginCredentials( Authentication.LoginCredentials(
@ -1249,9 +1229,8 @@ class Session(Closeable, MessageListener, SubListener):
), ),
True, True,
) )
self.logger.info( self.logger.info("Re-authenticated as {}!".format(
"Re-authenticated as {}!".format(self.__ap_welcome.canonical_username) self.__ap_welcome.canonical_username))
)
def reconnecting(self) -> bool: def reconnecting(self) -> bool:
""" """ """ """
@ -1298,9 +1277,9 @@ class Session(Closeable, MessageListener, SubListener):
""" """ """ """
return self.__stored_str return self.__stored_str
def __authenticate_partial( def __authenticate_partial(self,
self, credential: Authentication.LoginCredentials, remove_lock: bool credential: Authentication.LoginCredentials,
) -> None: remove_lock: bool) -> None:
""" """
Login to Spotify Login to Spotify
Args: Args:
@ -1319,8 +1298,8 @@ class Session(Closeable, MessageListener, SubListener):
version_string=Version.version_string(), version_string=Version.version_string(),
) )
self.__send_unchecked( self.__send_unchecked(
Packet.Type.login, client_response_encrypted_proto.SerializeToString() Packet.Type.login,
) client_response_encrypted_proto.SerializeToString())
packet = self.cipher_pair.receive_encoded(self.connection) packet = self.cipher_pair.receive_encoded(self.connection)
if packet.is_cmd(Packet.Type.ap_welcome): if packet.is_cmd(Packet.Type.ap_welcome):
self.__ap_welcome = Authentication.APWelcome() self.__ap_welcome = Authentication.APWelcome()
@ -1329,12 +1308,11 @@ class Session(Closeable, MessageListener, SubListener):
bytes0x0f = Random.get_random_bytes(0x14) bytes0x0f = Random.get_random_bytes(0x14)
self.__send_unchecked(Packet.Type.unknown_0x0f, bytes0x0f) self.__send_unchecked(Packet.Type.unknown_0x0f, bytes0x0f)
preferred_locale = io.BytesIO() preferred_locale = io.BytesIO()
preferred_locale.write( preferred_locale.write(b"\x00\x00\x10\x00\x02preferred-locale" +
b"\x00\x00\x10\x00\x02preferred-locale" self.__inner.preferred_locale.encode())
+ self.__inner.preferred_locale.encode()
)
preferred_locale.seek(0) preferred_locale.seek(0)
self.__send_unchecked(Packet.Type.preferred_locale, preferred_locale.read()) self.__send_unchecked(Packet.Type.preferred_locale,
preferred_locale.read())
if remove_lock: if remove_lock:
with self.__auth_lock: with self.__auth_lock:
self.__auth_lock_bool = False self.__auth_lock_bool = False
@ -1342,19 +1320,19 @@ class Session(Closeable, MessageListener, SubListener):
if self.__inner.conf.store_credentials: if self.__inner.conf.store_credentials:
reusable = self.__ap_welcome.reusable_auth_credentials reusable = self.__ap_welcome.reusable_auth_credentials
reusable_type = Authentication.AuthenticationType.Name( reusable_type = Authentication.AuthenticationType.Name(
self.__ap_welcome.reusable_auth_credentials_type self.__ap_welcome.reusable_auth_credentials_type)
)
if self.__inner.conf.stored_credentials_file is None: if self.__inner.conf.stored_credentials_file is None:
raise TypeError("The file path to be saved is not specified") raise TypeError(
"The file path to be saved is not specified")
self.__stored_str = base64.b64encode( self.__stored_str = base64.b64encode(
json.dumps( json.dumps({
{ "username":
"username": self.__ap_welcome.canonical_username, self.__ap_welcome.canonical_username,
"credentials": base64.b64encode(reusable).decode(), "credentials":
"type": reusable_type, base64.b64encode(reusable).decode(),
} "type":
).encode() reusable_type,
).decode() }).encode()).decode()
with open(self.__inner.conf.stored_credentials_file, "w") as f: with open(self.__inner.conf.stored_credentials_file, "w") as f:
json.dump( json.dump(
{ {
@ -1432,8 +1410,7 @@ class Session(Closeable, MessageListener, SubListener):
return self return self
def set_device_type( def set_device_type(
self, device_type: Connect.DeviceType self, device_type: Connect.DeviceType) -> Session.AbsBuilder:
) -> Session.AbsBuilder:
""" """
:param device_type: Connect.DeviceType: :param device_type: Connect.DeviceType:
@ -1502,12 +1479,13 @@ class Session(Closeable, MessageListener, SubListener):
""" """
if self.device_id is None: if self.device_id is None:
raise TypeError("You must specify the device ID first.") raise TypeError("You must specify the device ID first.")
self.login_credentials = self.decrypt_blob(self.device_id, username, blob) self.login_credentials = self.decrypt_blob(self.device_id,
username, blob)
return self return self
def decrypt_blob( def decrypt_blob(
self, device_id: str, username: str, encrypted_blob: bytes self, device_id: str, username: str,
) -> Authentication.LoginCredentials: encrypted_blob: bytes) -> Authentication.LoginCredentials:
""" """
:param device_id: str: :param device_id: str:
@ -1519,9 +1497,11 @@ class Session(Closeable, MessageListener, SubListener):
sha1 = SHA1.new() sha1 = SHA1.new()
sha1.update(device_id.encode()) sha1.update(device_id.encode())
secret = sha1.digest() secret = sha1.digest()
base_key = PBKDF2( base_key = PBKDF2(secret,
secret, username.encode(), 20, 0x100, hmac_hash_module=SHA1 username.encode(),
) 20,
0x100,
hmac_hash_module=SHA1)
sha1 = SHA1.new() sha1 = SHA1.new()
sha1.update(base_key) sha1.update(base_key)
key = sha1.digest() + b"\x00\x00\x00\x14" key = sha1.digest() + b"\x00\x00\x00\x14"
@ -1539,8 +1519,8 @@ class Session(Closeable, MessageListener, SubListener):
type_ = Authentication.AuthenticationType.Name(type_int) type_ = Authentication.AuthenticationType.Name(type_int)
if type_ is None: if type_ is None:
raise IOError( raise IOError(
TypeError("Unknown AuthenticationType: {}".format(type_int)) TypeError(
) "Unknown AuthenticationType: {}".format(type_int)))
blob.read(1) blob.read(1)
l = self.read_blob_int(blob) l = self.read_blob_int(blob)
auth_data = blob.read(l) auth_data = blob.read(l)
@ -1578,7 +1558,8 @@ class Session(Closeable, MessageListener, SubListener):
else: else:
try: try:
self.login_credentials = Authentication.LoginCredentials( self.login_credentials = Authentication.LoginCredentials(
typ=Authentication.AuthenticationType.Value(obj["type"]), typ=Authentication.AuthenticationType.Value(
obj["type"]),
username=obj["username"], username=obj["username"],
auth_data=base64.b64decode(obj["credentials"]), auth_data=base64.b64decode(obj["credentials"]),
) )
@ -1586,7 +1567,8 @@ class Session(Closeable, MessageListener, SubListener):
pass pass
return self return self
def stored_file(self, stored_credentials: str = None) -> Session.Builder: def stored_file(self,
stored_credentials: str = None) -> Session.Builder:
"""Create credential from stored file """Create credential from stored file
:param stored_credentials: str: (Default value = None) :param stored_credentials: str: (Default value = None)
@ -1604,7 +1586,8 @@ class Session(Closeable, MessageListener, SubListener):
else: else:
try: try:
self.login_credentials = Authentication.LoginCredentials( self.login_credentials = Authentication.LoginCredentials(
typ=Authentication.AuthenticationType.Value(obj["type"]), typ=Authentication.AuthenticationType.Value(
obj["type"]),
username=obj["username"], username=obj["username"],
auth_data=base64.b64decode(obj["credentials"]), auth_data=base64.b64decode(obj["credentials"]),
) )
@ -1722,7 +1705,8 @@ class Session(Closeable, MessageListener, SubListener):
# Stored credentials # Stored credentials
store_credentials: bool = True store_credentials: bool = True
stored_credentials_file: str = os.path.join(os.getcwd(), "credentials.json") stored_credentials_file: str = os.path.join(
os.getcwd(), "credentials.json")
# Fetching # Fetching
retry_on_chunk_error: bool = True retry_on_chunk_error: bool = True
@ -1762,8 +1746,8 @@ class Session(Closeable, MessageListener, SubListener):
# return self # return self
def set_cache_enabled( def set_cache_enabled(
self, cache_enabled: bool self,
) -> Session.Configuration.Builder: cache_enabled: bool) -> Session.Configuration.Builder:
"""Set cache_enabled """Set cache_enabled
:param cache_enabled: bool: :param cache_enabled: bool:
@ -1773,7 +1757,8 @@ class Session(Closeable, MessageListener, SubListener):
self.cache_enabled = cache_enabled self.cache_enabled = cache_enabled
return self return self
def set_cache_dir(self, cache_dir: str) -> Session.Configuration.Builder: def set_cache_dir(self,
cache_dir: str) -> Session.Configuration.Builder:
"""Set cache_dir """Set cache_dir
:param cache_dir: str: :param cache_dir: str:
@ -1784,8 +1769,8 @@ class Session(Closeable, MessageListener, SubListener):
return self return self
def set_do_cache_clean_up( def set_do_cache_clean_up(
self, do_cache_clean_up: bool self,
) -> Session.Configuration.Builder: do_cache_clean_up: bool) -> Session.Configuration.Builder:
"""Set do_cache_clean_up """Set do_cache_clean_up
:param do_cache_clean_up: bool: :param do_cache_clean_up: bool:
@ -1796,8 +1781,8 @@ class Session(Closeable, MessageListener, SubListener):
return self return self
def set_store_credentials( def set_store_credentials(
self, store_credentials: bool self,
) -> Session.Configuration.Builder: store_credentials: bool) -> Session.Configuration.Builder:
"""Set store_credentials """Set store_credentials
:param store_credentials: bool: :param store_credentials: bool:
@ -1808,7 +1793,7 @@ class Session(Closeable, MessageListener, SubListener):
return self return self
def set_stored_credential_file( def set_stored_credential_file(
self, stored_credential_file: str self, stored_credential_file: str
) -> Session.Configuration.Builder: ) -> Session.Configuration.Builder:
"""Set stored_credential_file """Set stored_credential_file
@ -1820,7 +1805,7 @@ class Session(Closeable, MessageListener, SubListener):
return self return self
def set_retry_on_chunk_error( def set_retry_on_chunk_error(
self, retry_on_chunk_error: bool self, retry_on_chunk_error: bool
) -> Session.Configuration.Builder: ) -> Session.Configuration.Builder:
"""Set retry_on_chunk_error """Set retry_on_chunk_error
@ -1975,9 +1960,8 @@ class Session(Closeable, MessageListener, SubListener):
self.conf = conf self.conf = conf
self.device_type = device_type self.device_type = device_type
self.device_name = device_name self.device_name = device_name
self.device_id = ( self.device_id = (device_id if device_id is not None else
device_id if device_id is not None else util.random_hex_string(40) util.random_hex_string(40))
)
class Receiver: class Receiver:
""" """ """ """
@ -2004,21 +1988,18 @@ class Session(Closeable, MessageListener, SubListener):
cmd: bytes cmd: bytes
try: try:
packet = self.__session.cipher_pair.receive_encoded( packet = self.__session.cipher_pair.receive_encoded(
self.__session.connection self.__session.connection)
)
cmd = Packet.Type.parse(packet.cmd) cmd = Packet.Type.parse(packet.cmd)
if cmd is None: if cmd is None:
self.__session.logger.info( self.__session.logger.info(
"Skipping unknown command cmd: 0x{}, payload: {}".format( "Skipping unknown command cmd: 0x{}, payload: {}".
util.bytes_to_hex(packet.cmd), packet.payload format(util.bytes_to_hex(packet.cmd),
) packet.payload))
)
continue continue
except (RuntimeError, ConnectionResetError) as ex: except (RuntimeError, ConnectionResetError) as ex:
if self.__running: if self.__running:
self.__session.logger.fatal( self.__session.logger.fatal(
"Failed reading packet! {}".format(ex) "Failed reading packet! {}".format(ex))
)
self.__session.reconnect() self.__session.reconnect()
break break
if not self.__running: if not self.__running:
@ -2026,19 +2007,16 @@ class Session(Closeable, MessageListener, SubListener):
if cmd == Packet.Type.ping: if cmd == Packet.Type.ping:
if self.__session.scheduled_reconnect is not None: if self.__session.scheduled_reconnect is not None:
self.__session.scheduler.cancel( self.__session.scheduler.cancel(
self.__session.scheduled_reconnect self.__session.scheduled_reconnect)
)
def anonymous(): def anonymous():
""" """ """ """
self.__session.logger.warning( self.__session.logger.warning(
"Socket timed out. Reconnecting..." "Socket timed out. Reconnecting...")
)
self.__session.reconnect() self.__session.reconnect()
self.__session.scheduled_reconnect = self.__session.scheduler.enter( self.__session.scheduled_reconnect = self.__session.scheduler.enter(
2 * 60 + 5, 1, anonymous 2 * 60 + 5, 1, anonymous)
)
self.__session.send(Packet.Type.pong, packet.payload) self.__session.send(Packet.Type.pong, packet.payload)
elif cmd == Packet.Type.pong_ack: elif cmd == Packet.Type.pong_ack:
continue continue
@ -2046,49 +2024,47 @@ class Session(Closeable, MessageListener, SubListener):
self.__session.__country_code = packet.payload.decode() self.__session.__country_code = packet.payload.decode()
self.__session.logger.info( self.__session.logger.info(
"Received country_code: {}".format( "Received country_code: {}".format(
self.__session.__country_code self.__session.__country_code))
)
)
elif cmd == Packet.Type.license_version: elif cmd == Packet.Type.license_version:
license_version = io.BytesIO(packet.payload) license_version = io.BytesIO(packet.payload)
license_id = struct.unpack(">h", license_version.read(2))[0] license_id = struct.unpack(">h",
license_version.read(2))[0]
if license_id != 0: if license_id != 0:
buffer = license_version.read() buffer = license_version.read()
self.__session.logger.info( self.__session.logger.info(
"Received license_version: {}, {}".format( "Received license_version: {}, {}".format(
license_id, buffer.decode() license_id, buffer.decode()))
)
)
else: else:
self.__session.logger.info( self.__session.logger.info(
"Received license_version: {}".format(license_id) "Received license_version: {}".format(license_id))
)
elif cmd == Packet.Type.unknown_0x10: elif cmd == Packet.Type.unknown_0x10:
self.__session.logger.debug( self.__session.logger.debug("Received 0x10: {}".format(
"Received 0x10: {}".format(util.bytes_to_hex(packet.payload)) util.bytes_to_hex(packet.payload)))
)
elif cmd in [ elif cmd in [
Packet.Type.mercury_sub, Packet.Type.mercury_sub,
Packet.Type.mercury_unsub, Packet.Type.mercury_unsub,
Packet.Type.mercury_event, Packet.Type.mercury_event,
Packet.Type.mercury_req, Packet.Type.mercury_req,
]: ]:
self.__session.mercury().dispatch(packet) self.__session.mercury().dispatch(packet)
elif cmd in [Packet.Type.aes_key, Packet.Type.aes_key_error]: elif cmd in [Packet.Type.aes_key, Packet.Type.aes_key_error]:
self.__session.audio_key().dispatch(packet) self.__session.audio_key().dispatch(packet)
elif cmd in [Packet.Type.channel_error, Packet.Type.stream_chunk_res]: elif cmd in [
Packet.Type.channel_error, Packet.Type.stream_chunk_res
]:
self.__session.channel().dispatch(packet) self.__session.channel().dispatch(packet)
elif cmd == Packet.Type.product_info: elif cmd == Packet.Type.product_info:
self.__session.parse_product_info(packet.payload) self.__session.parse_product_info(packet.payload)
else: else:
self.__session.logger.info( self.__session.logger.info("Skipping {}".format(
"Skipping {}".format(util.bytes_to_hex(cmd)) util.bytes_to_hex(cmd)))
)
class SpotifyAuthenticationException(Exception): class SpotifyAuthenticationException(Exception):
""" """ """ """
def __init__(self, login_failed: Keyexchange.APLoginFailed): def __init__(self, login_failed: Keyexchange.APLoginFailed):
super().__init__(Keyexchange.ErrorCode.Name(login_failed.error_code)) super().__init__(
Keyexchange.ErrorCode.Name(login_failed.error_code))
class SearchManager: class SearchManager:
@ -2112,17 +2088,15 @@ class SearchManager:
if request.get_locale() == "": if request.get_locale() == "":
request.set_locale(self.__session.preferred_locale()) request.set_locale(self.__session.preferred_locale())
response = self.__session.mercury().send_sync( response = self.__session.mercury().send_sync(
RawMercuryRequest.new_builder() RawMercuryRequest.new_builder().set_method("GET").set_uri(
.set_method("GET") request.build_url()).build())
.set_uri(request.build_url())
.build()
)
if response.status_code != 200: if response.status_code != 200:
raise SearchManager.SearchException(response.status_code) raise SearchManager.SearchException(response.status_code)
return json.loads(response.payload) return json.loads(response.payload)
class SearchException(Exception): class SearchException(Exception):
""" """ """ """
def __init__(self, status_code: int): def __init__(self, status_code: int):
super().__init__("Search failed with code {}.".format(status_code)) super().__init__("Search failed with code {}.".format(status_code))
@ -2195,7 +2169,8 @@ class SearchManager:
self.__country = country self.__country = country
return self return self
def set_image_size(self, image_size: str) -> SearchManager.SearchRequest: def set_image_size(self,
image_size: str) -> SearchManager.SearchRequest:
""" """
:param image_size: str: :param image_size: str:
@ -2243,8 +2218,7 @@ class TokenProvider:
self._session = session self._session = session
def find_token_with_all_scopes( def find_token_with_all_scopes(
self, scopes: typing.List[str] self, scopes: typing.List[str]) -> typing.Union[StoredToken, None]:
) -> typing.Union[StoredToken, None]:
""" """
:param scopes: typing.List[str]: :param scopes: typing.List[str]:
@ -2279,19 +2253,15 @@ class TokenProvider:
else: else:
return token return token
self.logger.debug( self.logger.debug(
"Token expired or not suitable, requesting again. scopes: {}, old_token: {}".format( "Token expired or not suitable, requesting again. scopes: {}, old_token: {}"
scopes, token .format(scopes, token))
)
)
response = self._session.mercury().send_sync_json( response = self._session.mercury().send_sync_json(
MercuryRequests.request_token(self._session.device_id(), ",".join(scopes)) MercuryRequests.request_token(self._session.device_id(),
) ",".join(scopes)))
token = TokenProvider.StoredToken(response) token = TokenProvider.StoredToken(response)
self.logger.debug( self.logger.debug(
"Updated token successfully! scopes: {}, new_token: {}".format( "Updated token successfully! scopes: {}, new_token: {}".format(
scopes, token scopes, token))
)
)
self.__tokens.append(token) self.__tokens.append(token)
return token return token
@ -2310,9 +2280,9 @@ class TokenProvider:
def expired(self) -> bool: def expired(self) -> bool:
""" """ """ """
return self.timestamp + ( return self.timestamp + (self.expires_in - TokenProvider.
self.expires_in - TokenProvider.token_expire_threshold token_expire_threshold) * 1000 < int(
) * 1000 < int(time.time_ns() / 1000) time.time_ns() / 1000)
def has_scope(self, scope: str) -> bool: def has_scope(self, scope: str) -> bool:
""" """