Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
53 changes: 53 additions & 0 deletions calibre-plugin/libadobe.py
Original file line number Diff line number Diff line change
Expand Up @@ -142,6 +142,59 @@ def update_account_path(folder_path):
FILE_ACTIVATIONXML = os.path.join(folder_path, "activation.xml")


# Loan tokens are stored in the account's activation.xml, next to the operatorURLList and the
# licenseServices, because they are per-operator data just like those two. Each operator signs
# its own loanToken and that token only covers the loans that operator issued, so one token is
# kept per operator - keeping just the most recent one would silently break the loans of every
# other operator the user has borrowed from.

def get_loan_tokens():
# type: () -> list

adNS = lambda tag: '{%s}%s' % ('http://ns.adobe.com/adept', tag)

try:
activationxml = etree.parse(get_activation_xml_path())
except:
return []

return activationxml.findall("./%s" % (adNS("loanToken")))


def store_loan_token(loanToken):
# type: (etree._Element) -> None

adNS = lambda tag: '{%s}%s' % ('http://ns.adobe.com/adept', tag)
NSMAP = { "adept" : "http://ns.adobe.com/adept" }
etree.register_namespace("adept", NSMAP["adept"])

try:
operatorURL = loanToken.find("./%s" % (adNS("operatorURL"))).text.strip()
except:
# Without an operator there's no way to tell which token this one replaces.
return

activationxml = etree.parse(get_activation_xml_path())

# A loanToken covers all of that operator's currently-active loans, so a new one from the
# same operator supersedes the old one. Tokens are signed blobs: they are stored verbatim
# and never edited.
for oldToken in activationxml.findall("./%s" % (adNS("loanToken"))):
try:
if oldToken.find("./%s" % (adNS("operatorURL"))).text.strip() == operatorURL:
activationxml.getroot().remove(oldToken)
except:
pass

# Detach the node from the response document before adding it to the activation.
activationxml.getroot().append(etree.fromstring(etree.tostring(loanToken)))

f = open(get_activation_xml_path(), "w")
f.write("<?xml version=\"1.0\"?>\n")
f.write(etree.tostring(activationxml, encoding="utf-8", pretty_print=True, xml_declaration=False).decode("utf-8"))
f.close()


def createDeviceKeyFile():
# Original implementation: Device::createDeviceKeyFile()

Expand Down
28 changes: 27 additions & 1 deletion calibre-plugin/libadobeAccount.py
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,7 @@

from lxml import etree
import base64
import re
import locale, platform

try:
Expand All @@ -27,7 +28,7 @@

from libadobe import addNonce, sign_node, sendRequestDocu, sendHTTPRequest
from libadobe import makeFingerprint, makeSerial, encrypt_with_device_key, decrypt_with_device_key
from libadobe import get_devkey_path, get_device_path, get_activation_xml_path
from libadobe import get_devkey_path, get_device_path, get_activation_xml_path, get_loan_tokens
from libadobe import VAR_VER_SUPP_CONFIG_NAMES, VAR_VER_HOBBES_VERSIONS, VAR_VER_OS_IDENTIFIERS
from libadobe import VAR_VER_ALLOWED_BUILD_IDS_SWITCH_TO, VAR_VER_SUPP_VERSIONS, VAR_ACS_SERVER_HTTP
from libadobe import VAR_ACS_SERVER_HTTPS, VAR_VER_BUILD_IDS, VAR_VER_NEED_HTTPS_BUILD_ID_LIMIT, VAR_VER_ALLOWED_BUILD_IDS_AUTHORIZE
Expand Down Expand Up @@ -551,6 +552,31 @@ def exportProxyAuth(act_xml_path, activationToken):

ret += activationToken

# A device authorization (activationToken) on its own is enough to open *purchased* ADEPT
# books, but a library LOAN additionally requires a signed loanToken (bound to user+loan,
# NOT to a device) plus the operator's licenseServices to be present in the target device's
# activation.xml. Without them the reader refuses loaned content (e.g. a Kobo shows
# "Your eReader is not authorized to open this book") even though the device is authorized.
# Every operator the user has borrowed from has its own loanToken, so they all get written.
def _embed_adept(node):

@bjmc bjmc Aug 1, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Nitpick but instead of doing raw string manipulation with regex, if we need to remove this adobe adept namespace (are we sure we do? why?) I think it would be cleaner to make the changes as XML and keep working with the data as XML.

AI suggested this but I haven't verified it and it might be overkill:

def _embed_adept(node):
    ADEPT_NS = "http://ns.adobe.com/adept"
    
    # Find all elements with the Adobe namespace
    for elem in node.xpath(f'.//*[namespace-uri()="{ADEPT_NS}"]'):
        # Replace the tag with its local name
        elem.tag = etree.QName(elem).localname
    
    # Clean up namespace declarations for the Adobe namespace
    for elem in node.iter():
        if elem.nsmap:
            # Remove the Adobe namespace from the mapping
            elem.nsmap = {k: v for k, v in elem.nsmap.items() if v != ADEPT_NS}
    
    return node

s = etree.tostring(node, encoding="unicode")
s = re.sub(r'\s+xmlns(:ns\d+)?="http://ns\.adobe\.com/adept"', '', s)
s = re.sub(r'(</?)ns\d+:', r'\1', s)
return s

try:
for loanToken in get_loan_tokens():
ret += _embed_adept(loanToken)
except Exception:
pass

try:
account_ls = etree.parse(get_activation_xml_path()).find("./{http://ns.adobe.com/adept}licenseServices")
if account_ls is not None:
ret += _embed_adept(account_ls)
except Exception:
pass

ret += "</activationInfo>"

# Okay, now we can finally write this to the device.
Expand Down
13 changes: 12 additions & 1 deletion calibre-plugin/libadobeFulfill.py
Original file line number Diff line number Diff line change
Expand Up @@ -15,7 +15,7 @@
#@@CALIBRE_COMPAT_CODE@@

from libadobe import addNonce, sign_node, get_cert_from_pkcs12, sendRequestDocu, sendRequestDocuRC, sendHTTPRequest
from libadobe import get_devkey_path, get_device_path, get_activation_xml_path
from libadobe import get_devkey_path, get_device_path, get_activation_xml_path, store_loan_token
from libadobe import VAR_VER_SUPP_VERSIONS, VAR_VER_SUPP_CONFIG_NAMES, VAR_VER_HOBBES_VERSIONS
from libadobe import VAR_VER_BUILD_IDS, VAR_VER_USE_DIFFERENT_NOTIFICATION_XML_ORDER

Expand Down Expand Up @@ -458,6 +458,17 @@ def fulfill(acsm_file, do_notify = False):
NSMAP = { "adept" : "http://ns.adobe.com/adept" }
adNS = lambda tag: '{%s}%s' % ('http://ns.adobe.com/adept', tag)

# Persist the signed loanToken returned by the operator so it can later be embedded into a
# tethered eReader's activation.xml (see exportProxyAuth). A device authorization alone is
# enough for purchased books, but library loans additionally require this token on the device
# or the reader shows "not authorized to open this book".
try:
loanTokenNode = adobe_fulfill_response.find("./%s" % (adNS("loanToken")))
if loanTokenNode is not None:
store_loan_token(loanTokenNode)
except Exception:
pass

licenseURL = adobe_fulfill_response.find("./%s/%s/%s/%s" % (adNS("fulfillmentResult"), adNS("resourceItemInfo"), adNS("licenseToken"), adNS("licenseURL"))).text

if adept_ns:
Expand Down
187 changes: 186 additions & 1 deletion tests/main.py
Original file line number Diff line number Diff line change
Expand Up @@ -528,8 +528,193 @@ def test_loanReturnFulfillmentID(self):
self.assertEqual(extracted_token, expected_token, "Loan record generator broken")


def test_loanTokenStorageIsPerOperator(self):
'''Check if loan tokens are stored one per operator'''

class TestOther(unittest.TestCase):
# Each operator signs its own loanToken, covering only the loans that operator issued.
# Storing just the most recent one would silently break the loans of every other operator
# the user has borrowed from, which is exactly what happens once a single library platform
# routes different books through different operators.

import tempfile, shutil

def make_token(operator, loans):
return etree.fromstring("""
<loanToken xmlns="http://ns.adobe.com/adept">
<time>2022-07-03T01:14:42Z</time>
<user>urn:uuid:2bd57a81-6192-4a1b-8eb2-64e2d197f9fa</user>
<operatorURL>%s</operatorURL>
<licenseURL>https://acs.example.com/licensesign</licenseURL>
%s
<signature>c2lnbmF0dXJl</signature>
</loanToken>
""" % (operator, "".join(["<loan>%s</loan>" % (l) for l in loans])))

def stored():
return [(t.find("{http://ns.adobe.com/adept}operatorURL").text,
[l.text for l in t.iter("{http://ns.adobe.com/adept}loan")])
for t in libadobe.get_loan_tokens()]

old_account_path = os.path.dirname(libadobe.get_activation_xml_path())
tmpdir = tempfile.mkdtemp()

try:
libadobe.update_account_path(tmpdir)

f = open(libadobe.get_activation_xml_path(), "w")
f.write('<?xml version="1.0"?>\n')
f.write('<activationInfo xmlns="http://ns.adobe.com/adept"></activationInfo>\n')
f.close()

libadobe.store_loan_token(make_token("https://acs-a.example.com/fulfillment", ["a-1"]))
self.assertEqual(stored(),
[("https://acs-a.example.com/fulfillment", ["a-1"])],
"First loan token wasn't stored")

# A second operator must not evict the first one.
libadobe.store_loan_token(make_token("https://acs-b.example.com/fulfillment", ["b-1"]))
self.assertEqual(sorted(stored()),
[("https://acs-a.example.com/fulfillment", ["a-1"]),
("https://acs-b.example.com/fulfillment", ["b-1"])],
"Second operator's loan token evicted the first operator's")

# A newer token from an operator we already know replaces only that operator's entry.
libadobe.store_loan_token(make_token("https://acs-a.example.com/fulfillment", ["a-1", "a-2"]))
self.assertEqual(sorted(stored()),
[("https://acs-a.example.com/fulfillment", ["a-1", "a-2"]),
("https://acs-b.example.com/fulfillment", ["b-1"])],
"Updating one operator's loan token didn't leave the other operator alone")

finally:
libadobe.update_account_path(old_account_path)
shutil.rmtree(tmpdir, ignore_errors=True)


def test_tetheredDeviceAuthorizationIncludesLoanData(self):
'''Check if authorizing a tethered eReader includes the loan data'''

# An activationToken on its own is enough for a purchased book, but a loaned book also
# needs the operator's signed loanToken and its licenseServices entry on the device -
# without them the reader refuses loaned content even though it is properly authorized.

import tempfile, shutil

adNS = lambda tag: '{%s}%s' % ('http://ns.adobe.com/adept', tag)

account_xml = """<?xml version="1.0"?>
<activationInfo xmlns="http://ns.adobe.com/adept">
<activationServiceInfo>
<authURL>https://adeactivate.example.com/adept</authURL>
<userInfoURL>https://adeactivate.example.com/adept</userInfoURL>
<activationURL>https://adeactivate.example.com/adept</activationURL>
<certificate>Y2VydGlmaWNhdGU=</certificate>
</activationServiceInfo>
<credentials>
<user>urn:uuid:2bd57a81-6192-4a1b-8eb2-64e2d197f9fa</user>
<username method="AdobeID">test@example.com</username>
<licenseCertificate>bGljZW5zZUNlcnQ=</licenseCertificate>
<privateLicenseKey>cHJpdmF0ZUtleQ==</privateLicenseKey>
<authenticationCertificate>YXV0aENlcnQ=</authenticationCertificate>
</credentials>
<licenseServices>
<licenseServiceInfo>
<licenseURL>https://acs.example.com/licensesign</licenseURL>
<certificate>bGljZW5zZVNlcnZpY2VDZXJ0</certificate>
</licenseServiceInfo>
</licenseServices>
<loanToken>
<time>2022-07-03T01:14:42Z</time>
<user>urn:uuid:2bd57a81-6192-4a1b-8eb2-64e2d197f9fa</user>
<operatorURL>https://acs-a.example.com/fulfillment</operatorURL>
<licenseURL>https://acs.example.com/licensesign</licenseURL>
<loan>a-1</loan>
<signature>c2lnbmF0dXJlQQ==</signature>
</loanToken>
<loanToken>
<time>2022-07-03T01:14:42Z</time>
<user>urn:uuid:2bd57a81-6192-4a1b-8eb2-64e2d197f9fa</user>
<operatorURL>https://acs-b.example.com/fulfillment</operatorURL>
<licenseURL>https://acs.example.com/licensesign</licenseURL>
<loan>b-1</loan>
<signature>c2lnbmF0dXJlQg==</signature>
</loanToken>
</activationInfo>
"""

activation_token = b"""<activationToken xmlns="http://ns.adobe.com/adept">
<device>urn:uuid:83681cbb-b6df-44a3-a423-c2b37ba66e84</device>
<fingerprint>ZmluZ2VycHJpbnQ=</fingerprint>
<deviceType>mobile</deviceType>
<activationURL>https://adeactivate.example.com/adept</activationURL>
<user>urn:uuid:2bd57a81-6192-4a1b-8eb2-64e2d197f9fa</user>
<signature>c2lnbmF0dXJl</signature>
</activationToken>"""

old_account_path = os.path.dirname(libadobe.get_activation_xml_path())
tmpdir = tempfile.mkdtemp()

try:
libadobe.update_account_path(tmpdir)

f = open(libadobe.get_activation_xml_path(), "w")
f.write(account_xml)
f.close()

device_xml_path = os.path.join(tmpdir, "device_activation.xml")
ret, msg = libadobeAccount.exportProxyAuth(device_xml_path, activation_token)
self.assertTrue(ret, "Writing the device authorization failed: %s" % (msg))

written = etree.parse(device_xml_path)

# The device authorization itself - this already worked before.
self.assertIsNotNone(written.find("./%s" % (adNS("activationToken"))),
"No activationToken written to the device")
self.assertIsNotNone(written.find("./%s" % (adNS("credentials"))),
"No credentials written to the device")

# The loan data - this is what loaned books need.
written_loans = dict([
(t.find("./%s" % (adNS("operatorURL"))).text,
[l.text for l in t.iter(adNS("loan"))])
for t in written.findall("./%s" % (adNS("loanToken")))])

self.assertEqual(written_loans, {
"https://acs-a.example.com/fulfillment": ["a-1"],
"https://acs-b.example.com/fulfillment": ["b-1"],
}, "Loan tokens missing from the device authorization")

written_license_services = written.findall("./%s/%s/%s" % (
adNS("licenseServices"), adNS("licenseServiceInfo"), adNS("licenseURL")))

self.assertEqual([e.text for e in written_license_services],
["https://acs.example.com/licensesign"],
"licenseServices missing from the device authorization")

# An account that has never borrowed anything (only purchased books) has no loan
# tokens at all - that has to keep producing a valid device authorization.
no_loans = etree.parse(libadobe.get_activation_xml_path())
for token in no_loans.findall("./%s" % (adNS("loanToken"))):
no_loans.getroot().remove(token)
f = open(libadobe.get_activation_xml_path(), "wb")
f.write(etree.tostring(no_loans, encoding="utf-8", xml_declaration=True))
f.close()

ret, msg = libadobeAccount.exportProxyAuth(device_xml_path, activation_token)
self.assertTrue(ret, "Writing the device authorization without loans failed: %s" % (msg))

written = etree.parse(device_xml_path)
self.assertIsNotNone(written.find("./%s" % (adNS("activationToken"))),
"Device authorization broken for an account with no loans")
self.assertEqual(written.findall("./%s" % (adNS("loanToken"))), [],
"Loan tokens written for an account that has none")

finally:
libadobe.update_account_path(old_account_path)
shutil.rmtree(tmpdir, ignore_errors=True)



class TestOther(unittest.TestCase):

def setUp(self):
pass
Expand Down