devnoname120
commented
4 months ago @mkb79 OK so here is a dump of code I have. It should be helpful for your implementation.
I don't remember exactly what works/what doesn't and I don't have time to test but I'm pretty sure that something does run up to the point of initializing Widevine, sending back the challenge to Audible, and Audible complaining that the provisioned certificate inside Widevine is wrong (it's not signed with the Audible cert, which should be extracted as discussed above with @szescxz).
diff --git a/src/audible/client.py b/src/audible/client.py
index 369435e..a4991de 100644
--- a/src/audible/client.py
+++ b/src/audible/client.py
@@ -7,6 +7,22 @@ import httpx
from httpx import URL
from httpx._models import Headers, HeaderTypes # noqa: F401
+LEVEL = logging.DEBUG
+
+logger_httpx = logging.getLogger("httpx")
+logger_httpx.setLevel(LEVEL)
+#logger_httpx.setConsoleLogger(LEVEL)
+# log_helper.set_file_logger(FILENAME, LEVEL)
+#logger_httpx.captureWarnings()
+
+
+
+logging.basicConfig(
+ format="%(levelname)s [%(asctime)s] %(name)s - %(message)s",
+ datefmt="%Y-%m-%d %H:%M:%S",
+ level=logging.DEBUG
+)
+
from .auth import Authenticator
from .exceptions import (
BadRequest,
@@ -88,7 +104,8 @@ class Client:
default_headers.update(headers)
self.session = self._SESSION(
- headers=default_headers, timeout=timeout, auth=auth, **session_kwargs
+ headers=default_headers, timeout=timeout, auth=auth, verify=False, **session_kwargs
+ #headers=default_headers, timeout=timeout, auth=auth, proxies="http://localhost:6617", verify=False, **session_kwargs
)
if response_callback is None:
diff --git a/src/audible/login.py b/src/audible/login.py
index 90692da..f39b607 100644
--- a/src/audible/login.py
+++ b/src/audible/login.py
@@ -19,11 +19,10 @@ from .metadata import encrypt_metadata, meta_audible_app
logger = logging.getLogger("audible.login")
USER_AGENT = (
- "Mozilla/5.0 (iPhone; CPU iPhone OS 15_0 like Mac OS X) "
- "AppleWebKit/605.1.15 (KHTML, like Gecko) Mobile/15E148"
+ "[REDACTED user agent of Android device that supports Audible Atmos]"
)
-
def default_captcha_callback(captcha_url: str) -> str:
"""Helper function for handling captcha."""
captcha = httpx.get(captcha_url).content
@@ -202,8 +201,8 @@ def build_oauth_url(
if with_username:
base_url = f"https://www.audible.{domain}/ap/signin"
return_to = f"https://www.audible.{domain}/ap/maplanding"
- assoc_handle = f"amzn_audible_ios_lap_{country_code}"
- page_id = "amzn_audible_ios_privatepool"
+ assoc_handle = f"amzn_audible_android_aui_{country_code}"
+ page_id = f"amzn_audible_android_aui_{country_code}"
oauth_params = {
"openid.oa2.response_type": "code",
diff --git a/src/audible/register.py b/src/audible/register.py
index 27a1abf..f29b9a9 100644
--- a/src/audible/register.py
+++ b/src/audible/register.py
@@ -1,10 +1,26 @@
+import base64
+from cryptography.hazmat.primitives import serialization
+
from datetime import datetime, timedelta
from typing import Any, Dict, Optional
+import json
import httpx
from .login import build_client_id
+def base64_der_to_pkcs1(base64_key):
+ der_private_key = base64.b64decode(base64_key)
+ private_key = serialization.load_der_private_key(der_private_key, password=None)
+
+ pkcs1_private_key = private_key.private_bytes(
+ encoding=serialization.Encoding.PEM,
+ format=serialization.PrivateFormat.TraditionalOpenSSL,
+ encryption_algorithm=serialization.NoEncryption()
+ )
+
+ return pkcs1_private_key.decode('utf-8')#.replace('\n', '\\n')
+
def register(
authorization_code: str,
@@ -39,19 +55,30 @@ def register(
"store_authentication_cookie",
],
"cookies": {"website_cookies": [], "domain": f".amazon.{domain}"},
+ "device_metadata": {
+ "device_os_family": "android",
+ "device_serial": "[REDACTED]",
+ "device_type": "A10KISP2GWF0E4",
+ "manufacturer": "[REDACTED]",
+ "model": "[REDACTED]",
+ "os_version": "25",
+ "product": "[REDACTED]",
+ },
"registration_data": {
- "domain": "Device",
- "app_version": "3.56.2",
- "device_serial": serial,
- "device_type": "A2CZJZGLK2JJVM",
+ "domain": "DeviceLegacy",
+ # "app_version": "3.59.0",
+ "app_version": "139018",
+ "device_serial": "[REDACTED]",
+ "device_type": "A10KISP2GWF0E4",
"device_name": (
- "%FIRST_NAME%%FIRST_NAME_POSSESSIVE_STRING%%DUPE_"
- "STRATEGY_1ST%Audible for iPhone"
+ "%FIRST_NAME%%FIRST_NAME_POSSESSIVE_STRING%%DUPE_STRATEGY_1ST%Audible for Android"
),
- "os_version": "15.0.0",
- "software_version": "35602678",
- "device_model": "iPhone",
- "app_name": "Audible",
+ "os_version": "[REDACTED an Android release-keys string]",
+ # "software_version": "35900789",
+ "software_version": "130050002",
+ "device_model": "[REDACTED]",
+ "app_name": "com.audible.application",
},
"auth_data": {
"client_id": build_client_id(serial),
@@ -71,11 +98,15 @@ def register(
if resp.status_code != 200:
raise Exception(resp_json)
+ print(json.dumps(resp_json))
+
success_response = resp_json["response"]["success"]
tokens = success_response["tokens"]
adp_token = tokens["mac_dms"]["adp_token"]
- device_private_key = tokens["mac_dms"]["device_private_key"]
+ device_private_key = base64_der_to_pkcs1(tokens["mac_dms"]["device_private_key"])
+ print(device_private_key)
+ # device_private_key = "-----BEGIN RSA PRIVATE KEY-----\n" + device_private_key + "-----END RSA PRIVATE KEY-----\n"
store_authentication_cookie = tokens["store_authentication_cookie"]
access_token = tokens["bearer"]["access_token"]
refresh_token = tokens["bearer"]["refresh_token"]
device_register.py:
from audible import Authenticator, Client, log_helper
def prompt_captcha_callback(captcha_url: str) -> str:
"""Helper function for handling captcha."""
print("Captcha found")
print(captcha_url)
guess = input("Answer for CAPTCHA")
return str(guess).strip().lower()
def prompt_otp_callback() -> str:
"""Helper function for handling 2-factor authentication."""
print("2FA is activated for this account.")
guess = input("Please enter OTP Code")
return str(guess).strip().lower()
auth = Authenticator.from_login(
username="redacted@example.com",
password="[REDACTED]",
locale="fr",
captcha_callback=prompt_captcha_callback,
otp_callback=prompt_otp_callback)
device_name = auth.device_info["device_name"]
print(f"Successfully registered {device_name}.")
ab = auth.get_activation_bytes()
print(ab)
auth.to_file(filename="/Users/paul/.audible/android.json")ec3-download.py:
#!/usr/bin/env python3
# widevine imports
import base64, requests, sys, xmltodict
import logging
import json
# from urllib.parse import urlparse
from pywidevine.cdm import Cdm
from pywidevine.device import Device
from pywidevine.pssh import PSSH
# from cdm import cdm, deviceconfig
from getPSSH import get_pssh
# from wvdecryptcustom import WvDecrypt
# from cdm.formats import wv_proto2_pb2 as wv_proto2
from base64 import b64encode
# import json
from audible import Authenticator, Client, log_helper
# import logging
from http.client import HTTPConnection
LEVEL = "debug"
HTTPConnection.debuglevel = 1
log_helper.set_level(LEVEL)
log_helper.set_console_logger(LEVEL)
# log_helper.set_file_logger(FILENAME, LEVEL)
log_helper.capture_warnings()
requests_log = logging.getLogger("requests.packages.urllib3")
requests_log.setLevel(logging.DEBUG)
requests_log.propagate = True
# pywidevine stuff
device = Device.load("/Users/paul/dev/Audible/my_audible_l3.wvd")
cdm = Cdm.from_device(device)
session_id = cdm.open()
# Audible stuff
auth_file_path = "/Users/paul/.audible/atmos-test-3" # FILL OUT
asin = "[REDACTED]" # FILL OUT
auth = Authenticator.from_file(auth_file_path)
with Client(auth) as client:
body = {
"consumption_type": "Download",
"quality": "High",
"response_groups": "content_reference,chapter_info,pdf_url,ad_insertion",
"spatial": "true",
"supported_media_features": {
"chapter_titles_type": "Tree",
"codecs": [
"mp4a.40.2",
"ec+3"
],
"drm_types": [
"Widevine",
"Adrm",
"Mpeg"
]
}
}
# body = {
# "quality": "High",
# "response_groups": "chapter_info,content_reference,last_position_heard,pdf_url",
# "consumption_type": "Download",
# "supported_media_features":
# {
# "codecs": [
# "mp4a.40.2",
# "mp4a.40.42",
# "ec+3"
# ],
# "drm_types": [
# "Mpeg",
# "Adrm",
# "FairPlay"
# ]
# },
# "spatial": True
# }
lr = client.post(
f"content/{asin}/licenserequest",
body=body,
headers={"user-agent": "[REDACTED audible user agent of Android device with Dolby Atmos that has hardware support]"}
)
print(json.dumps(lr, indent=4))
mpd_url = lr['content_license']['license_response']
print('mpd_url: ' + mpd_url)
# mpd_url = 'https://[REDACTED].cloudfront.net/[REDACTED]-ec3_v8.master.mpd?ss_sec=10&iss_sec=10&isc=1&use_token_based_signing=true'
#lic_url = 'https://api.audible.fr/1.0/content/[REDACTED]/drmlicense'
pssh_raw = get_pssh(mpd_url)
# pssh = '[REDACTED base64]'
# params from mpd_url:
# ottsession=[REDACTED hexadecimal]&
# puid=[REDACTED integer]&
# video_content_id=[REDACTED hexadecimal]&
print(f'{chr(10)}PSSH obtained:\n{pssh_raw}')
pssh = PSSH(pssh_raw)
challenge = cdm.get_license_challenge(session_id, pssh, license_type="OFFLINE", privacy_mode=False)
challenge_b64 = b64encode(challenge)
print(f'Got challenge:\n{challenge_b64}')
response = client.post(
f"content/{asin}/drmlicense",
body={
"consumption_type": "Download",
"drm_type": "Widevine",
"licenseChallenge": str(challenge_b64, "utf-8" ),
},
headers={"user-agent": "[REDACTED audible user agent of Android device with Dolby Atmos that has hardware support]"}
)
print('response.license: ' + response['license'])
with open("widevine_license.txt", "w") as lic_file:
lic_file.write(response.license)
cdm.parse_license(session_id, response['license'])
for key in cdm.get_keys(session_id):
print(f"[{key.type}] {key.kid.hex}:{key.key.hex()}")
# close session, disposes of session data
cdm.close(session_id)
def widevine_get_keys(pssh, cert_b64=None):
"""main func, emulates license request and then decrypt obtained license
fileds that changes every new request is signature, expirationTimestamp, watchSessionId, puid, and rawLicenseRequestBase64 """
wvdecrypt = WvDecrypt(init_data_b64=pssh, cert_data_b64=cert_b64, device=deviceconfig.device_android_generic)
raw_request = wvdecrypt.get_challenge()
request = b64encode(raw_request)
signature = cdm.hash_object
# audible
response = client.post(
f"content/{asin}/drmlicense",
body={
"consumption_type": "download",
"drm_type": "widevine",
"licensechallenge": str(request, "utf-8" ),
},
headers={"user-agent": "[REDACTED audible user agent of Android device with Dolby Atmos that has hardware support]"}
)
print('response.license: ' + response['license'])
with open("widevine_license.txt", "w") as lic_file:
lic_file.write(response.license)
license_b64 = response.license
#
# license_b64 = "[REDACTED very long base64 string]"
wvdecrypt.update_license(license_b64)
Correct, keyswvdecrypt = wvdecrypt.start_process()
if Correct:
return Correct, keyswvdecrypt
# correct, keys = widevine_get_keys(pssh)
# print('is correct? ' + correct)
# for key in keys:
# print('KID:KEY -> ' + key)
Then some other files (I'm not sure anymore which is used/is useful/works but at least you have them at hand):
wvdecryptcustom.py:
# uncompyle6 version 3.7.3
# Python bytecode 3.6 (3379)
# Decompiled from: Python 3.7.8 (tags/v3.7.8:4b47a5b6ba, Jun 28 2020, 08:53:46) [MSC v.1916 64 bit (AMD64)]
# Embedded file name: pywidevine\decrypt\wvdecryptcustom.py
import logging, subprocess, re, base64
from cdm import cdm, deviceconfig
class WvDecrypt(object):
WV_SYSTEM_ID = [
237, 239, 139, 169, 121, 214, 74, 206, 163, 200, 39, 220, 213, 29, 33, 237]
def __init__(self, init_data_b64, cert_data_b64, device):
self.init_data_b64 = init_data_b64
self.cert_data_b64 = cert_data_b64
self.device = device
self.cdm = cdm.Cdm()
def check_pssh(pssh_b64):
pssh = base64.b64decode(pssh_b64)
if not pssh[12:28] == bytes(self.WV_SYSTEM_ID):
new_pssh = bytearray([0, 0, 0])
new_pssh.append(32 + len(pssh))
new_pssh[4:] = bytearray(b'pssh')
new_pssh[8:] = [0, 0, 0, 0]
new_pssh[13:] = self.WV_SYSTEM_ID
new_pssh[29:] = [0, 0, 0, 0]
new_pssh[31] = len(pssh)
new_pssh[32:] = pssh
return base64.b64encode(new_pssh)
else:
return pssh_b64
self.session = self.cdm.open_session(check_pssh(self.init_data_b64), deviceconfig.DeviceConfig(self.device), None, True)
if self.cert_data_b64:
self.cdm.set_service_certificate(self.session, self.cert_data_b64)
def log_message(self, msg):
return '{}'.format(msg)
def start_process(self):
keyswvdecrypt = []
try:
for key in self.cdm.get_keys(self.session):
if key.type == 'CONTENT':
keyswvdecrypt.append(self.log_message('{}:{}'.format(key.kid.hex(), key.key.hex())))
except Exception:
return (
False, keyswvdecrypt)
else:
return (
True, keyswvdecrypt)
def get_challenge(self):
return self.cdm.get_license_request(self.session)
def update_license(self, license_b64):
self.cdm.provide_license(self.session, license_b64)
return TruegetPSSH.py
import requests, xmltodict
def get_pssh(mpd_url):
# pssh = '[REDACTED]'
pssh = ''
try:
r = requests.get(url=mpd_url)
r.raise_for_status()
xml = xmltodict.parse(r.text)
mpd = xml
periods = mpd['MPD']['Period']
except Exception as e:
pssh = input(f'\nUnable to find PSSH in MPD: {e}. \nEdit getPSSH.py or enter PSSH manually: ')
return pssh
if isinstance(periods, list):
print('Period is a list')
for idx, period in enumerate(periods):
if isinstance(period['AdaptationSet'], list):
print('AdaptationSet is a list')
for ad_set in period['AdaptationSet']:
if True or ad_set['@mimeType'] in ['video/mp4', 'audio/mp4']:
print('Found appropriate mime')
# try:
for t in ad_set['ContentProtection']:
print(t['@schemeIdUri'].lower())
if t['@schemeIdUri'].lower() == "urn:uuid:edef8ba9-79d6-4ace-a3c8-27dcd51d21ed":
print('Found appropriate scheme id')
pssh = t["cenc:pssh"]
print('Found pssh: ' + pssh)
# except Exception:
# pass
else:
print('AdaptationSet is a single item')
if True or period['AdaptationSet']['@mimeType'] in ['video/mp4', 'audio/mp4']:
print('Found appropriate mime')
# try:
for t in period['AdaptationSet']['ContentProtection']:
print(t['@schemeIdUri'].lower())
if t['@schemeIdUri'].lower() == "urn:uuid:edef8ba9-79d6-4ace-a3c8-27dcd51d21ed":
print('Found appropriate scheme id')
pssh = t["cenc:pssh"]
print('Found pssh: ' + pssh)
# except Exception:
# pass
else:
print('Period is a single item')
for ad_set in periods['AdaptationSet']:
if True or ad_set['@mimeType'] in ['video/mp4', 'audio/mp4']:
print('Found appropriate mime')
# try:
for t in ad_set['ContentProtection']:
print(t['@schemeIdUri'].lower())
if t['@schemeIdUri'].lower() == "urn:uuid:edef8ba9-79d6-4ace-a3c8-27dcd51d21ed":
print('Found appropriate scheme id')
pssh = t["cenc:pssh"]
print('Found pssh: ' + pssh)
# except Exception:
# pass
if pssh == '':
pssh = input('Unable to find PSSH in mpd. Edit getPSSH.py or enter PSSH manually: ')
return pssh13.py:
# -*- coding: utf-8 -*-
# Module: widevine_keys
# Created on: 10.12.2021
# Authors: medvm
# Version: 2.1.0
import base64, requests, sys, xmltodict
# import headers
# import cookies
import json
from cdm import cdm, deviceconfig
from base64 import b64encode
from getPSSH import get_pssh
from wvdecryptcustom import WvDecrypt
from cdm.formats import wv_proto2_pb2 as wv_proto2
from urllib.parse import urlparse
import logging
# logging.basicConfig(level=logging.DEBUG)
# MDP_URL = input('\nInput MPD URL: ')
MDP_URL = 'https://[REDACTED].cloudfront.net/[REDACTED]-ec3_v8.master.mpd?ss_sec=10&iss_sec=10&isc=1&use_token_based_signing=true'
lic_url = 'https://api.audible.fr/1.0/content/[REDACTED]/drmlicense'
responses = []
license_b64 = ''
pssh = get_pssh(MDP_URL)
params = None
params = urlparse(lic_url).query
# pssh = '[REDACTED]'
# params from mdp_url:
# ottsession=[REDACTED hexadecimal number]&
# puid=[REDACTED integer number]&
# video_content_id=[REDACTED hexadecimal number]&
print(f'{chr(10)}PSSH obtained.\n{pssh}')
def widevine_get_keys(pssh, lic_url, cert_b64=None):
"""main func, emulates license request and then decrypt obtained license
fileds that changes every new request is signature, expirationTimestamp, watchSessionId, puid, and rawLicenseRequestBase64 """
wvdecrypt = WvDecrypt(init_data_b64=pssh, cert_data_b64=cert_b64, device=deviceconfig.device_android_generic)
raw_request = wvdecrypt.get_challenge()
request = b64encode(raw_request)
signature = cdm.hash_object
# audible
response = requests.post(url=lic_url, headers=headers.headers, params=params,
json={
"consumption_type": "Download",
"drm_type": "Widevine",
"licenseChallenge": str(request, "utf-8" ),
}
))
try:
str(response.content, "utf-8")
except UnicodeDecodeError:
widevine_license = response
print(f'{chr(10)}license response status: {widevine_license}{chr(10)}')
break
else:
if len(str(response.content, "utf-8")) > 500:
widevine_license = response
print(f'{chr(10)}license response status: {widevine_license}{chr(10)}')
break
if idx == len(responses) - 1:
print(f'{chr(10)}license response status: {response}')
print(f'server reports: {str(response.content, "utf-8")}')
print(f'server did not issue license, make sure you have correctly pasted all the required headers in the headers.py. Also check json/raw params of POST request.{chr(10)}')
exit()
lic_field_names = ['license', 'payload', 'getWidevineLicenseResponse']
lic_field_names2 = ['license']
open('license_content.bin', 'wb').write(widevine_license.content)
try:
if str(widevine_license.content, 'utf-8').find(':'):
for key in lic_field_names:
try:
license_b64 = json.loads(widevine_license.content.decode())[key]
except:
pass
else:
for key2 in lic_field_names2:
try:
license_b64 = json.loads(widevine_license.content.decode())[key][key2]
except:
pass
else:
license_b64 = widevine_license.content
except:
license_b64 = b64encode(widevine_license.content)
#
# license_b64 = "[REDACTED]"
wvdecrypt.update_license(license_b64)
Correct, keyswvdecrypt = wvdecrypt.start_process()
if Correct:
return Correct, keyswvdecrypt
correct, keys = widevine_get_keys(pssh, lic_url)
print('is correct? ' + correct)
for key in keys:
print('KID:KEY -> ' + key)
mkb79
szescxz
DrJapan
So it would be awesome to enable the Audible Dolby Atmos option for downloading I'm not sure if something would need to be added to the api or even what codec format they are using but it seems it can be downloaded in the iOS app and some select android devices.
Digging around the api found this in
/1.0/library/B0C66LN3JWbut it showed the standard stereoavailable_codecsThis is all new to me and I'm not even sure what I'm looking for.