def compute_response(method, challenge, iters, salt, passw):
from pbkdf2 import pbkdf2_bin
from binascii import hexlify, unhexlify
import hashlib
def md5hex(x):
return hexlify(hashlib.md5(x).digest())
def aes(k, d):
import aes
# nb: aes init vector (iv) is hardcoded to 0, so we return the
# encrypted result sans its iv prefix
return \
hexlify(aes.encryptData(unhexlify(k), unhexlify(d), iv = [0] * 16))[32:]
def sha256(x, outlen = 16):
return hexlify(hashlib.sha256(unhexlify(x)).digest())[:outlen * 2]
if '0' == method:
passw = md5hex(passw)
elif '1' == method:
passw = md5hex(passw.upper())
else:
passw = 'D0DB1CA3B300831A301AF9144FC6986A'
hwid, emac = '00000000', '0000CAFEBABE'
pkey = hexlify(pbkdf2_bin(unhexlify(passw), unhexlify(salt), iters, 16))
response = \
aes(pkey, challenge + sha256('%s%s%s' % (hwid, emac, challenge)))
return hwid, emac, pkey, response
def login(username, password, ctype):
from util import get_url, extract_substr_fields
login_url = 'https://www.siriusxm.com/userservices/authentication' \
'/en-us/xml/user/login/v2/initiate'
pdata = \
'<AuthenticationRequest><userName>%s</userName>' % username + \
'<consumerType>%s</consumerType>' % ctype + \
'<subscriberType>SIRIUS_SUBSCRIBER</subscriberType>' \
'<currency>840</currency></AuthenticationRequest>'
challenge, iters, method, salt = \
extract_substr_fields(get_url(login_url, pdata),
[['<authenticationChallenge>',
'</authenticationChallenge>'],
['<iterationsCount>', '</iterationsCount>'],
['<passwordHashType>', '</passwordHashType>'],
['<salt>', '</salt>']])
hwid, emac, pkey, response = \
compute_response(method, challenge, int(iters), salt, password)
resp = \
'<AuthenticationRequest><userName>%s</userName>' % username + \
'<consumerType>%s</consumerType>' % ctype + \
'<subscriberType>SIRIUS_SUBSCRIBER</subscriberType>' \
'<currency>840</currency><playerIdentification>' \
'<hardwareIdentification>%s</hardwareIdentification>' % hwid + \
'<ethernetMac>%s</ethernetMac></playerIdentification>' % emac + \
'<authenticationData>%s</authenticationData> ' % response + \
'</AuthenticationRequest>'
comp_url = 'https://www.siriusxm.com/userservices/authentication' \
'/en-us/xml/user/login/v2/complete'
session, lineup = \
extract_substr_fields(get_url(comp_url, resp),
[['<sessionId>', '</sessionId>'],
['<channelLineupId>', '</channelLineupId>']])
return session, lineup, pkey
def media_url(cid, user, passw, ctype):
from urllib import quote
from util import get_url, substr_between
from binascii import hexlify, unhexlify
import re
try:
import json
except ImportError:
import simplejson as json
def aes(k, d):
import aes
# nb: iv prefix is 0; see "compute_response"
return hexlify(aes.decryptData(unhexlify(k), unhexlify('00' * 16 + d)))
def sha256(x, outlen = 16):
import hashlib
return hexlify(hashlib.sha256(unhexlify(x)).digest())[:outlen * 2]
def fstr_at(s, x): # nb: fstr is short for "****ed string".
size = 0
for c in unhexlify(s[x:x + 4]):
size += ord(c)
end = x + 4 + 2 * size
return unhexlify(s[x + 4:end]), end
def get_m3u8(mrl):
prefix, suffix = re.findall(r'(.*?/)(\?token=.*)', mrl)[0]
return '%s%s_variant_large.m3u8%s' % (prefix, cid, suffix)
session, _, pkey = login(user, passw, ctype)
data = json.loads(get_url('https://www.siriusxm.com/userservices/token/en-us'
'/json/%s/token/v2/%s?sessionId=%s' % \
(ctype, cid, quote(session))))
token = aes(pkey, data['tokenResponse']['tokenData'])
mrl, pos = fstr_at(token, 8)
mrls = [mrl, fstr_at(token, pos)[0]]
keys = token[-72:-8]
keys = [keys[:32], keys[32:]]
return '%s///keys=%s' % \
([get_m3u8(mrl) for mrl in mrls][0],
','.join(['D0DB1CA3B300831A301AF9144FC6986A'] + keys))
def get_channels_list(user, passwd, ctype):
from util import substr_between, get_url, clean
_, lineup, _ = login(user, passwd, ctype)
lang = 'en-us'
format = 'xml'
channels_url = \
'https://www.siriusxm.com/userservices/cl' \
'/%s/%s/lineup/%s/client/%s' \
% (lang, format, lineup, ctype.strip('2'))
data = get_url(channels_url)
channels = []
try:
while True:
cid, data = substr_between(data, '<contentId>', '</contentId>')
desc, data = substr_between(data, '<displayName>', '</displayName>')
name, data = substr_between(data, '<name>', '</name>')
num, data = \
substr_between(data, '<siriusChannelNo>', '</siriusChannelNo>')
xm_num, data = \
substr_between(data, '<xmChannelNo>', '</xmChannelNo')
if not len(num.strip()):
num = xm_num
prefix = 'SiriusXM'
if not name.startswith(prefix):
name = '%s %s' % (prefix, name)
channels += [[clean(x.decode('utf-8')).encode('utf-8') for x in
[cid, num, name, desc, xm_num]]]
except:
pass
channels.sort(lambda x, y: cmp(int(x[1]), int(y[1])))
return channels
def get_channels(user, passwd, ctype):
return '\n'.join('\n'.join([f for f in x]) \
for x in get_channels_list(user, passwd, ctype))
# def whats_on(cid):
# # nb: useless without a timestamp to sync with, and we have no way
# # to pass it.
# try:
# import json
# except ImportError:
# import simplejson as json
# from time import strftime, localtime, time
# from util import get_url
# ts = strftime('%m-%d-%H:%M:%S', localtime(time()))
# data = json.loads(get_url('https://metadata.siriusxm.com/metadata/pdt/'
# 'en-us/json/events/timestamp/%s' % ts))
# for entry in data['channelMetadataResponse']['metaData']:
# if cid != entry['channelId']:
# continue
# evt = entry['currentEvent']
# track = evt['song']['name']
# artist = evt['artists']['name']
# return '%s - %s' % (artist, track)
# return ''
