Commit d8e5afec authored by Jean-Francois Dockes's avatar Jean-Francois Dockes
Browse files

Qobuz plugin working again, for now...

parent 84eea83d
......@@ -191,15 +191,14 @@ dist_cdpluginstidallib_DATA = \
cdpluginsqobuzdir = $(pkgdatadir)/cdplugins/qobuz
dist_cdpluginsqobuz_DATA = \
src/mediaserver/cdplugins/qobuz/debug.py \
src/mediaserver/cdplugins/qobuz/exception.py \
src/mediaserver/cdplugins/qobuz/__init__.py \
src/mediaserver/cdplugins/qobuz/qobuz-app.py \
src/mediaserver/cdplugins/qobuz/session.py
cdpluginsqobuzlibdir = $(pkgdatadir)/cdplugins/qobuz/api
dist_cdpluginsqobuzlib_DATA = \
src/mediaserver/cdplugins/qobuz/api/__init__.py \
src/mediaserver/cdplugins/qobuz/api/raw.py
src/mediaserver/cdplugins/qobuz/api/raw.py \
src/mediaserver/cdplugins/qobuz/api/spoofbuz.py
cdpluginshradir = $(pkgdatadir)/cdplugins/hra
dist_cdpluginshra_DATA = \
......
AC_INIT([upmpdcli], [1.5.1], [jfd@lesbonscomptes.com],
AC_INIT([upmpdcli], [1.5.2], [jfd@lesbonscomptes.com],
[upmpdcli], [http://www.lesbonscomptes.com/upmpdcli])
AC_PREREQ([2.53])
AC_CONFIG_MACRO_DIRS([m4])
......
= Upmpdcli release notes
== 2020-11-24 upmpdcli 1.5.2
- Modify a few things which should not have mattered but which prevented
Lumin to work with upmpdcli
- Qobuz plugin working again (for now), by identifying itself as the
web client (thanks to qobuz-dl on github).
- Fixed crash.
- Will now generate events for bitrate changes when playing.
== 2020-11-22 upmpdcli 1.5.1
- Fix small issue resulting in missing metadata in ohinfo
== 2020-11-15 upplay 1.4.6
- Create the GUI even if the library initialisation fails so that the
......
......@@ -10,39 +10,45 @@
:license: GPLv3, see LICENSE for more details.
'''
import sys
from time import time
import time
import math
import hashlib
import socket
import binascii
from itertools import cycle
import requests
from qobuz.exception import QobuzXbmcError
from qobuz.debug import warn, info
from . import spoofbuz
socket.timeout = 5
_loglevel = 3
def debug(s):
if _loglevel >= 2:
print("%s" %s, file=sys.stderr)
def warn(s):
if _loglevel >= 3:
print("%s" %s, file=sys.stderr)
class RawApi(object):
def __init__(self, appid, configvalue):
self.appid = appid
#print("app_id: %s"% self.appid, file=sys.stderr)
self.configvalue = configvalue.encode('ASCII')
self.bappid = self.appid.encode('ASCII')
if appid and configvalue:
self.configvalue = configvalue
self.appid = appid
self.__set_s4()
else:
self.spoofer = spoofbuz.Spoofer()
self.appid = self.spoofer.getAppId()
self.version = '0.2'
self.baseUrl = 'http://www.qobuz.com/api.json/'
self.user_auth_token = None
self.user_id = None
self.error = None
self.status_code = None
self._baseUrl = self.baseUrl + self.version
self.session = requests.Session()
self.statContentSizeTotal = 0
self.statTotalRequest = 0
self.error = None
self.__set_s4()
def _api_error_string(self, request, url='', params={}, json=''):
return '{reason} (code={status_code})\n' \
......@@ -56,19 +62,14 @@ class RawApi(object):
'''Checking parameters before sending our request
- if mandatory parameter is missing raise error
- if a given parameter is neither in mandatory or allowed
raise error (Creating exception class like MissingParameter
may be a good idea)
raise error
'''
for label in mandatory:
if not label in ka:
raise QobuzXbmcError(who=self,
what='missing_parameter',
additional=label)
raise Exception("Qobuz: missing parameter [%s]" % label)
for label in ka:
if label not in mandatory and label not in allowed:
raise QobuzXbmcError(who=self,
what='invalid_parameter',
additional=label)
raise Exception("Qobuz: invalid parameter [%s]" % label)
def __set_s4(self):
'''appid and associated secret is for this app usage only
......@@ -76,9 +77,10 @@ class RawApi(object):
General Terms and Conditions
(http://www.qobuz.com/apps/api/QobuzAPI-TermsofUse.pdf)
'''
s3b = self.configvalue
s3b = self.configvalue.encode('ASCII')
s3s = binascii.a2b_base64(s3b)
a = cycle(self.bappid)
bappid = self.appid.encode('ASCII')
a = cycle(bappid)
b = zip(s3s, a)
self.s4 = b''.join((x ^ y).to_bytes(1, byteorder='big') for (x, y) in b)
#print("S4: %s"% self.s4.decode('ASCII'), file=sys.stderr)
......@@ -114,50 +116,45 @@ class RawApi(object):
Error: [200]
Error: Bad Request [400]
'''
self.statTotalRequest += 1
self.error = ''
self.status_code = None
url = self._baseUrl + uri
useToken = False if (opt and 'noToken' in opt) else True
headers = {}
headers = {
"User-Agent":
"Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:67.0) Gecko/20100101 Firefox/67.0"
}
if useToken and self.user_auth_token:
headers['x-user-auth-token'] = self.user_auth_token
headers['x-app-id'] = self.appid
'''DEBUG'''
import copy
_copy_params = copy.deepcopy(params)
if 'password' in _copy_params:
_copy_params['password'] = '***'
info(self, 'URI {} POST PARAMS: {} HEADERS: {}', uri,
str(_copy_params), str(headers))
'''END / DEBUG'''
headers['X-User-Auth-Token'] = self.user_auth_token
headers['X-App-Id'] = self.appid
r = None
#debug("POST %s params %s headers %s" % (url, params, headers))
try:
r = self.session.post(url, data=params, headers=headers)
except:
self.error = 'Post request fail'
warn(self, self.error)
warn(self.error)
return None
self.status_code = int(r.status_code)
if self.status_code != 200:
self.error = self._api_error_string(r, url, _copy_params)
warn(self, self.error)
self.error = self._api_error_string(r, url, params)
warn(self.error)
return None
if not r.content:
self.error = 'Request return no content'
warn(self, self.error)
warn(self.error)
return None
self.statContentSizeTotal += sys.getsizeof(r.content)
'''Retry get if connexion fail'''
try:
response_json = r.json()
except Exception as e:
warn(self, 'Json loads failed to load... retrying!\n{}', repr(e))
warn('Json loads failed to load... retrying!\n{}', repr(e))
try:
response_json = r.json()
except:
self.error = "Failed to load json two times...abort"
warn(self, self.error)
warn(self.error)
return None
status = None
try:
......@@ -165,19 +162,16 @@ class RawApi(object):
except:
pass
if status == 'error':
self.error = self._api_error_string(r, url, _copy_params,
self.error = self._api_error_string(r, url, params,
response_json)
warn(self, self.error)
warn(self.error)
return None
return response_json
def set_user_data(self, user_id, user_auth_token):
if not (user_id and user_auth_token):
raise QobuzXbmcError(who=self, what='missing_argument',
additional='uid|token')
self.user_auth_token = user_auth_token
self.user_id = user_id
self.logged_on = time()
raise Exception("Qobuz: missing argument uid|token")
self.logged_on = time.time()
def logout(self):
self.user_auth_token = None
......@@ -185,29 +179,38 @@ class RawApi(object):
self.logged_on = None
def user_login(self, **ka):
data = self._user_login(**ka)
if not data:
self.logout()
return None
self.set_user_data(data['user']['id'], data['user_auth_token'])
return data
def _user_login(self, **ka):
self._check_ka(ka, ['username', 'password'], ['email'])
data = self._api_request(ka, '/user/login', noToken=True)
if not data:
return None
if not 'user' in data:
return None
if not 'id' in data['user']:
if not data or not 'user' in data or not 'credential' in data['user'] \
or not 'id' in data['user'] \
or not 'parameters' in data['user']['credential']:
warn("/user/login returns %s" % data)
self.logout()
return None
if not data['user']['id']:
if not data["user"]["credential"]["parameters"]:
warn("Free accounts are not eligible to download tracks.")
return None
self.user_id = data['user']['id']
self.user_auth_token = data["user_auth_token"]
self.label = data["user"]["credential"]["parameters"]["short_label"]
debug("Membership: {}".format(self.label))
data['user']['email'] = ''
data['user']['firstname'] = ''
data['user']['lastname'] = ''
self.setSec()
return data
def setSec(self):
global _loglevel
savedlevel = _loglevel
_loglevel = 1
for value in self.spoofer.getSecrets().values():
self.s4 = value.encode('utf-8')
if self.userlib_getAlbums(sec=self.s4) is not None:
#debug("SECRET [%s]"%self.s4)
_loglevel = savedloglevel
return
def user_update(self, **ka):
self._check_ka(ka, [], ['player_settings'])
return self._api_request(ka, '/user/update')
......@@ -218,23 +221,35 @@ class RawApi(object):
def track_getFileUrl(self, intent="stream", **ka):
self._check_ka(ka, ['format_id', 'track_id'])
ka['request_ts'] = time()
stringvalue = 'trackgetFileUrlformat_id' \
+ str(ka['format_id']) \
+ 'intent'+intent \
+ 'track_id' \
+ str(ka['track_id']) \
+ str(ka['request_ts'])
ts = str(time.time())
track_id = str(ka['track_id'])
fmt_id = str(ka['format_id'])
stringvalue = 'trackgetFileUrlformat_id' + fmt_id \
+ 'intent' + intent \
+ 'track_id' + track_id + ts
stringvalue = stringvalue.encode('ASCII')
stringvalue += self.s4
params = {'format_id': str(ka['format_id']),
rq_sig = str(hashlib.md5(stringvalue).hexdigest())
params = {'format_id': fmt_id,
'intent': intent,
'request_ts': ka['request_ts'],
'request_sig': str(hashlib.md5(stringvalue).hexdigest()),
'track_id': str(ka['track_id'])
'request_ts': ts,
'request_sig': rq_sig,
'track_id': track_id
}
return self._api_request(params, '/track/getFileUrl')
def userlib_getAlbums(self, **ka):
ts = str(time.time())
r_sig = "userLibrarygetAlbumsList" + str(ts) + str(ka["sec"])
r_sig_hashed = hashlib.md5(r_sig.encode("utf-8")).hexdigest()
params = {
"app_id": self.appid,
"user_auth_token": self.user_auth_token,
"request_ts": ts,
"request_sig": r_sig_hashed,
}
return self._api_request(params, '/userLibrary/getAlbumsList')
def track_search(self, **ka):
self._check_ka(ka, ['query'], ['limit'])
return self._api_request(ka, '/track/search')
......@@ -260,14 +275,14 @@ class RawApi(object):
def track_resportStreamingEnd(self, track_id, duration):
duration = math.floor(int(duration))
if duration < 5:
info(self, 'Duration lesser than 5s, abort reporting')
warn(self, 'Duration lesser than 5s, abort reporting')
return None
# @todo ???
user_auth_token = '' # @UnusedVariable
try:
user_auth_token = self.user_auth_token # @UnusedVariable
except:
warn(self, 'No authentification token')
warn('No authentification token')
return None
params = {'user_id': self.user_id,
'track_id': track_id,
......@@ -289,13 +304,7 @@ class RawApi(object):
return self._api_request(ka, '/purchase/getUserPurchases')
def search_getResults(self, **ka):
self._check_ka(ka, ['query'], ['type', 'limit', 'offset'])
mandatory = ['query', 'type']
for label in mandatory:
if not label in ka:
raise QobuzXbmcError(who=self,
what='missing_parameter',
additional=label)
self._check_ka(ka, ['query', 'type'], ['limit', 'offset'])
return self._api_request(ka, '/search/getResults')
def favorite_getUserFavorites(self, **ka):
......@@ -309,8 +318,7 @@ class RawApi(object):
if label in ka:
found = label
if not found:
raise QobuzXbmcError(who=self, what='missing_parameter',
additional='artist_ids|albums_ids|track_ids')
raise Exception("Qobuz: missing parameter: artist_ids|albums_ids|track_ids")
return self._api_request(ka, '/favorite/create')
def favorite_delete(self, **ka):
......@@ -320,8 +328,7 @@ class RawApi(object):
if label in ka:
found = label
if not found:
raise QobuzXbmcError(who=self, what='missing_parameter',
additional='artist_ids|albums_ids|track_ids')
raise Exception("Qobuz: missing parameter: artist_ids|albums_ids|track_ids")
return self._api_request(ka, '/favorite/delete')
def playlist_get(self, **ka):
......@@ -348,14 +355,7 @@ class RawApi(object):
return self._api_request(ka, '/playlist/deleteTracks')
def playlist_subscribe(self, **ka):
mandatory = ['playlist_id']
found = None
for label in mandatory:
if label in ka:
found = label
if not found:
raise QobuzXbmcError(
who=self, what='missing_parameter', additional='playlist_id')
self._check_ka(ka, ['playlist_id'], ['playlist_track_ids'])
return self._api_request(ka, '/playlist/subscribe')
def playlist_unsubscribe(self, **ka):
......@@ -373,9 +373,6 @@ class RawApi(object):
def playlist_delete(self, **ka):
self._check_ka(ka, ['playlist_id'])
if not 'playlist_id' in ka:
raise QobuzXbmcError(
who=self, what='missing_parameter', additional='playlist_id')
return self._api_request(ka, '/playlist/delete')
def playlist_update(self, **ka):
......
import base64
import re
from collections import OrderedDict
import requests
class Spoofer:
def __init__(self):
self.seed_timezone_regex = r'[a-z]\.initialSeed\("(?P<seed>[\w=]+)",window\.utimezone\.(?P<timezone>[a-z]+)\)'
# note: {timezones} should be replaced with every capitalized timezone joined by a |
self.info_extras_regex = r'name:"\w+/(?P<timezone>{timezones})",info:"(?P<info>[\w=]+)",extras:"(?P<extras>[\w=]+)"'
self.appId_regex = r'{app_id:"(?P<app_id>\d{9})",app_secret:"\w{32}",base_port:"80",base_url:"https://www\.qobuz\.com",base_method:"/api\.json/0\.2/"},n\.base_url="https://play\.qobuz\.com"'
login_page_request = requests.get("https://play.qobuz.com/login")
login_page = login_page_request.text
bundle_url_match = re.search(
r'<script src="(/resources/\d+\.\d+\.\d+-[a-z]\d{3}/bundle\.js)"></script>',
login_page,
)
bundle_url = bundle_url_match.group(1)
bundle_req = requests.get("https://play.qobuz.com" + bundle_url)
self.bundle = bundle_req.text
def getAppId(self):
return re.search(self.appId_regex, self.bundle).group("app_id")
def getSecrets(self):
seed_matches = re.finditer(self.seed_timezone_regex, self.bundle)
secrets = OrderedDict()
for match in seed_matches:
seed, timezone = match.group("seed", "timezone")
secrets[timezone] = [seed]
"""The code that follows switches around the first and second timezone. Why? Read on:
Qobuz uses two ternary (a shortened if statement) conditions that should always return false.
The way Javascript's ternary syntax works, the second option listed is what runs if the condition returns false.
Because of this, we must prioritize the *second* seed/timezone pair captured, not the first.
"""
keypairs = list(secrets.items())
secrets.move_to_end(keypairs[1][0], last=False)
info_extras_regex = self.info_extras_regex.format(
timezones="|".join([timezone.capitalize() for timezone in secrets])
)
info_extras_matches = re.finditer(info_extras_regex, self.bundle)
for match in info_extras_matches:
timezone, info, extras = match.group("timezone", "info", "extras")
secrets[timezone.lower()] += [info, extras]
for secret_pair in secrets:
secrets[secret_pair] = base64.standard_b64decode(
"".join(secrets[secret_pair])[:-44]
).decode("utf-8")
return secrets
if __name__ == '__main__':
spoofer = Spoofer()
print("%s"%spoofer.getSecrets())
print("%s"%spoofer.getAppId())
from __future__ import print_function
'''
qobuz.debug
~~~~~~~~~~~
:part_of: xbmc-qobuz
:copyright: (c) 2012 by Joachim Basmaison, Cyril Leclerc
:license: GPLv3, see LICENSE for more details.
'''
import sys
PY3 = sys.version > '3'
if PY3:
basestring = str
__debugging__ = True
ourlog = None
LOGDEBUG = None
LOGNOTICE = None
LOGERROR = None
LOGSEVERE = None
try:
import xbmc # @UnresolvedImport
import xbmcaddon # @UnresolvedImport
ourlog = xbmc.log
LOGDEBUG = xbmc.LOGDEBUG
LOGNOTICE = xbmc.LOGNOTICE
LOGERROR = xbmc.LOGERROR
LOGSEVERE = xbmc.LOGSEVERE
__debugging__ = False
if xbmcaddon.Addon(id='plugin.audio.qobuz').getSetting('debug') == 'true':
__debugging__ = True
except Exception as e:
LOGDEBUG = '[DEBUG]'
LOGNOTICE = '[NOTICE]'
LOGERROR = '[ERROR]'
LOGSEVERE = '[SEVERE]'
def logfunc(msg):
print('%s' % msg, file=sys.stderr)
ourlog = logfunc
def log(obj, *a, **ka):
'''Base for all logging function, run in/out Xbmc
Inside Xbmc loggin functions use xbmc.log else they just print
message to STDOUT
'''
lvl = LOGNOTICE
if 'lvl' in ka:
lvl = ka['lvl']
del ka['lvl']
if not __debugging__:
return
name = None
if isinstance(obj, basestring):
name = obj
else:
try:
name = obj.__class__.__name__
except:
name = type(obj)
msg = None
num_argument = len(a)
if num_argument == 1:
msg = a[0].format(**ka)
elif num_argument > 1:
msg = a[0].format(*a[1:], **ka)
ourlog('[Qobuz/{name}][{level}] {msg}'
.format(name=str(name), msg=msg, level=lvl))
def warn(obj, *a, **ka):
'''facility: LOGERROR'''
ka['lvl']= LOGERROR
log(obj, *a, **ka)
def info(obj, *a, **ka):
'''facility: LOGNOTICE'''
ka['lvl'] = LOGNOTICE
log(obj, *a, **ka)
def debug(obj, *a, **ka):
'''facility: LOGDEBUG'''
ka['lvl'] = LOGDEBUG
log(obj, *a, **ka)
def error(obj, *a, **ka):
'''facility: LOGSEVERE'''
log(obj, *a, **ka)
from __future__ import print_function
'''
qobuz.exception
~~~~~~~~~~~~~~~
:part_of: xbmc-qobuz
:copyright: (c) 2012 by Joachim Basmaison, Cyril Leclerc
:license: GPLv3, see LICENSE for more details.
'''
import sys
import pprint
import traceback
class QobuzXbmcError(Exception):
def __init__(self, **ka):
if not 'additional' in ka or ka['additional'] is None:
ka['additional'] = ''
if (not 'who' in ka) or (not 'what' in ka):
raise Exception(
'QobuzXbmcError', 'Missing constructor arguments (who|what)')
nl = "\n"
msg = "[QobuzXbmcError]" + nl
msg += " - who : " + pprint.pformat(ka['who']) + nl
msg += " - what : " + ka['what'] + nl
msg += " - additional : " + repr(ka['additional']) + nl
# msg += " - type : " + self.exc_type + nl
# msg += " - value : " + self.exc_value + nl
msg += " - Stack : " + nl
print("%s" % msg, file=sys.stderr)
print("%s" % traceback.print_exc(10), file=sys.stderr)
......@@ -71,11 +71,7 @@ def maybelogin(a={}):
upconfig = conftree.ConfSimple(os.environ["UPMPD_CONFIG"])
formatid = upconfig.get('qobuzformatid')
appid = upconfig.get('qobuzappid')
if not appid:
appid = '285473059'
cfvalue = upconfig.get('qobuzcfvalue')
if not cfvalue:
cfvalue = 'Bg8HAA5XAFBYV15UAlVVBAZYCw0MVwcKUVRaVlpWUQ8='