from lxml import etree import base64 #@@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 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 def buildFulfillRequest(acsm): adNS = lambda tag: '{%s}%s' % ('http://ns.adobe.com/adept', tag) activationxml = etree.parse(get_activation_xml_path()) devicexml = etree.parse(get_device_path()) user_uuid = activationxml.find("./%s/%s" % (adNS("credentials"), adNS("user"))).text device_uuid = activationxml.find("./%s/%s" % (adNS("activationToken"), adNS("device"))).text try: fingerprint = None device_type = None fingerprint = activationxml.find("./%s/%s" % (adNS("activationToken"), adNS("fingerprint"))).text device_type = activationxml.find("./%s/%s" % (adNS("activationToken"), adNS("deviceType"))).text except: pass if (fingerprint is None or fingerprint == "" or device_type is None or device_type == ""): # This should usually never happen with a proper activation, but just in case it does, # I'll leave this code in - it loads the fingerprint from the device data instead. fingerprint = devicexml.find("./%s" % (adNS("fingerprint"))).text device_type = devicexml.find("./%s" % (adNS("deviceType"))).text version = None clientOS = None clientLocale = None ver = devicexml.findall("./%s" % (adNS("version"))) for f in ver: if f.get("name") == "hobbes": version = f.get("value") elif f.get("name") == "clientOS": clientOS = f.get("value") elif f.get("name") == "clientLocale": clientLocale = f.get("value") # Find matching client version depending on the Hobbes version. # This way we don't need to store and re-load it for each fulfillment. try: v_idx = VAR_VER_HOBBES_VERSIONS.index(version) clientVersion = VAR_VER_SUPP_VERSIONS[v_idx] except: # Version not present, probably the "old" 10.0.4 entry. # As 10.X is in the 3.0 range, assume we're on ADE 3.0 clientVersion = "3.0.1.91394" if clientVersion == "ADE WIN 9,0,1131,27": # Ancient ADE 1.7.2 does this request differently request = "\n" request += "%s\n" % (user_uuid) request += "%s\n" % (device_uuid) request += "%s\n" % (device_type) request += etree.tostring(acsm, encoding="utf-8", pretty_print=True, xml_declaration=False).decode("utf-8") request += "" return request, False else: request = "" request += "" request += "" request += "%s" % (user_uuid) request += "%s" % (device_uuid) request += "%s" % (device_type) request += etree.tostring(acsm, encoding="utf-8", pretty_print=True, xml_declaration=False).decode("utf-8") request += "" request += "%s" % (version) request += "%s" % (clientOS) request += "%s" % (clientLocale) request += "%s" % (clientVersion) request += "%s" % (device_type) request += "%s" % ("ADOBE Digitial Editions") # YES, this typo ("Digitial" instead of "Digital") IS present in ADE!! request += "%s" % (fingerprint) request += "" request += "%s" % (user_uuid) request += "%s" % (device_uuid) request += "" request += "" request += "" return request, True def buildInitLicenseServiceRequest(authURL): # type: (str) -> str adNS = lambda tag: '{%s}%s' % ('http://ns.adobe.com/adept', tag) NSMAP = { "adept" : "http://ns.adobe.com/adept" } etree.register_namespace("adept", NSMAP["adept"]) activationxml = etree.parse(get_activation_xml_path()) user_uuid = activationxml.find("./%s/%s" % (adNS("credentials"), adNS("user"))).text ret = "" ret += "" ret += "" ret += "%s" % (authURL) ret += addNonce() ret += "%s" % (user_uuid) ret += "" NSMAP = { "adept" : "http://ns.adobe.com/adept" } etree.register_namespace("adept", NSMAP["adept"]) req_xml = etree.fromstring(ret) signature = sign_node(req_xml) if (signature is None): return None etree.SubElement(req_xml, etree.QName(NSMAP["adept"], "signature")).text = signature return "\n" + etree.tostring(req_xml, encoding="utf-8", pretty_print=True, xml_declaration=False).decode("utf-8") def buildAuthRequest(): activationxml = etree.parse(get_activation_xml_path()) adNS = lambda tag: '{%s}%s' % ('http://ns.adobe.com/adept', tag) user_pkcs12 = base64.b64decode(activationxml.find("./%s/%s" % (adNS("credentials"), adNS("pkcs12"))).text) try: from libadobe import devkey_bytes as devkey_adobe except: pass if devkey_adobe is not None: devkey_bytes = devkey_adobe else: f = open(get_devkey_path(), "rb") devkey_bytes = f.read() f.close() my_cert = get_cert_from_pkcs12(user_pkcs12, base64.b64encode(devkey_bytes)) ret = "\n" ret += "\n" ret += "%s\n" % (activationxml.find("./%s/%s" % (adNS("credentials"), adNS("user"))).text) ret += "%s\n" % (base64.b64encode(my_cert).decode("utf-8")) ret += "%s\n" % (activationxml.find("./%s/%s" % (adNS("credentials"), adNS("licenseCertificate"))).text) ret += "%s\n" % (activationxml.find("./%s/%s" % (adNS("credentials"), adNS("authenticationCertificate"))).text) ret += "" return ret def doOperatorAuth(operatorURL): # type: (str) -> str auth_req = buildAuthRequest() authURL = operatorURL if authURL.endswith("Fulfill"): authURL = authURL.replace("/Fulfill", "") replyData = sendRequestDocu(auth_req, authURL + "/Auth").decode("utf-8") if not " str adNS = lambda tag: '{%s}%s' % ('http://ns.adobe.com/adept', tag) NSMAP = { "adept" : "http://ns.adobe.com/adept" } etree.register_namespace("adept", NSMAP["adept"]) activationxml = etree.parse(get_activation_xml_path()) try: operator_url_list = activationxml.findall("./%s/%s" % (adNS("operatorURLList"), adNS("operatorURL"))) for member in operator_url_list: if member.text.strip() == operatorURL: #print("Already authenticated to operator") return None except: pass ret = doOperatorAuth(operatorURL) if (ret is not None): return "doOperatorAuth error: %s" % ret # Check if list exists: list = activationxml.find("./%s" % (adNS("operatorURLList"))) user_uuid = activationxml.find("./%s/%s" % (adNS("credentials"), adNS("user"))).text if list is None: x = etree.SubElement(activationxml.getroot(), etree.QName(NSMAP["adept"], "operatorURLList"), nsmap=NSMAP) etree.SubElement(x, etree.QName(NSMAP["adept"], "user")).text = user_uuid list = activationxml.find("./%s" % (adNS("operatorURLList"))) if list is None: return "Err, this list should not be none right now ..." etree.SubElement(list, etree.QName(NSMAP["adept"], "operatorURL")).text = operatorURL f = open(get_activation_xml_path(), "w") f.write("\n") f.write(etree.tostring(activationxml, encoding="utf-8", pretty_print=True, xml_declaration=False).decode("utf-8")) f.close() return None def buildRights(license_token_node): ret = "\n" ret += "\n" # Add license token ret += etree.tostring(license_token_node, encoding="utf-8", pretty_print=True, xml_declaration=False).decode("utf-8") ret += "\n" NSMAP = { "adept" : "http://ns.adobe.com/adept" } adNS = lambda tag: '{%s}%s' % ('http://ns.adobe.com/adept', tag) lic_token_url = license_token_node.find("./%s" % (adNS("licenseURL"))).text ret += "%s\n" % lic_token_url # Get cert for this license URL: activationxml = etree.parse(get_activation_xml_path()) try: licInfo = activationxml.findall("./%s/%s" % (adNS("licenseServices"), adNS("licenseServiceInfo"))) found = False for member in licInfo: if member.find("./%s" % (adNS("licenseURL"))).text == lic_token_url: ret += "%s\n" % (member.find("./%s" % (adNS("certificate"))).text) found = True break except: return None if not found: return None ret += "\n" ret += "\n" return ret def fulfill(acsm_file, do_notify = False): verbose_logging = False try: import calibre_plugins.deacsm.prefs as prefs deacsmprefs = prefs.DeACSM_Prefs() verbose_logging = deacsmprefs["detailed_logging"] except: pass # Get pkcs12: pkcs12 = None acsmxml = None try: activationxml = etree.parse(get_activation_xml_path()) adNS = lambda tag: '{%s}%s' % ('http://ns.adobe.com/adept', tag) pkcs12 = activationxml.find("./%s/%s" % (adNS("credentials"), adNS("pkcs12"))).text except: return False, "Activation not found or invalid" if pkcs12 is None or len(pkcs12) == 0: return False, "Activation missing" try: acsmxml = etree.parse(acsm_file) except: return False, "ACSM not found or invalid" #print(etree.tostring(acsmxml, encoding="utf-8", pretty_print=True, xml_declaration=False).decode("utf-8")) adNS = lambda tag: '{%s}%s' % ('http://ns.adobe.com/adept', tag) dcNS = lambda tag: '{%s}%s' % ('http://purl.org/dc/elements/1.1/', tag) try: mimetype = acsmxml.find("./%s/%s/%s" % (adNS("resourceItemInfo"), adNS("metadata"), dcNS("format"))).text if (mimetype == "application/pdf"): #print("You're trying to fulfill a PDF file.") pass elif (mimetype == "application/epub+zip"): #print("Trying to fulfill an EPUB file ...") pass else: print("Weird mimetype: %s" % (mimetype)) print("Continuing anyways ...") except: # Some books, like from Google Play books, use a different format and don't have that metadata tag. pass fulfill_request, adept_ns = buildFulfillRequest(acsmxml) if verbose_logging: print("Fulfill request:") print(fulfill_request) fulfill_request_xml = etree.fromstring(fulfill_request) # Sign the request: signature = sign_node(fulfill_request_xml) if (signature is None): return False, "Signing failed!" NSMAP = { "adept" : "http://ns.adobe.com/adept" } adNS = lambda tag: '{%s}%s' % ('http://ns.adobe.com/adept', tag) if adept_ns: # "new" ADE etree.SubElement(fulfill_request_xml, etree.QName(NSMAP["adept"], "signature")).text = signature else: # ADE 1.7.2 etree.SubElement(fulfill_request_xml, etree.QName("signature")).text = signature # Get operator URL: operatorURL = None try: operatorURL = acsmxml.find("./%s" % (adNS("operatorURL"))).text.strip() except: pass if (operatorURL is None or len(operatorURL) == 0): return False, "OperatorURL missing in ACSM" fulfillURL = operatorURL + "/Fulfill" ret = operatorAuth(fulfillURL) if (ret is not None): return False, "operatorAuth error: %s" % ret if adept_ns: # "new" ADE fulfill_req_signed = "\n" + etree.tostring(fulfill_request_xml, encoding="utf-8", pretty_print=True, xml_declaration=False).decode("utf-8") else: # ADE 1.7.2 fulfill_req_signed = etree.tostring(fulfill_request_xml, encoding="utf-8", pretty_print=True, xml_declaration=False).decode("utf-8") #print("will send:\n %s" % fulfill_req_signed) #print("Sending fulfill request to %s" % fulfillURL) # For debugging only # fulfillURL = fulfillURL.replace("https:", "http:") replyData = sendRequestDocu(fulfill_req_signed, fulfillURL).decode("utf-8") if "" req_data += "" req_data += "%s" % (user) req_data += "%s" % (device) req_data += "%s" % (loanID) req_data += addNonce() req_data += "" NSMAP = { "adept" : "http://ns.adobe.com/adept" } etree.register_namespace("adept", NSMAP["adept"]) full_text_xml = etree.fromstring(req_data) signature = sign_node(full_text_xml) if (signature is None): print("SIGN ERROR!") return False, "Sign error" etree.SubElement(full_text_xml, etree.QName(NSMAP["adept"], "signature")).text = signature print("Would notify server %s:" % (operatorURL + "/LoanReturn")) doc_send = "\n" + etree.tostring(full_text_xml, encoding="utf-8", pretty_print=True, xml_declaration=False).decode("utf-8") # print(doc_send) retval = sendRequestDocu(doc_send, operatorURL + "/LoanReturn").decode("utf-8") if " tag not found. Guess nobody wants to be notified.") #print(etree.tostring(fulfillmentResultToken, encoding="utf-8", pretty_print=True, xml_declaration=False).decode("utf-8")) return True, "" errmsg = "" errmsg_crit = "" for element in notifiers: url = element.find("./%s" % (adNS("notifyURL"))).text body = element.find("./%s" % (adNS("body"))) critical = True if element.get("critical", "yes") == "no": critical = False print("Notifying optional server %s" % (url)) else: print("Notifying server %s" % (url)) if (user is None): try: # "Normal" Adobe fulfillment user = fulfillmentResultToken.find("./%s/%s/%s/%s" % (adNS("fulfillmentResult"), adNS("resourceItemInfo"), adNS("licenseToken"), adNS("user"))).text except AttributeError: # B&N Adobe PassHash fulfillment user = body.find("./%s" % (adNS("user"))).text if (device is None): try: # "Normal" Adobe fulfillment device = fulfillmentResultToken.find("./%s/%s/%s/%s" % (adNS("fulfillmentResult"), adNS("resourceItemInfo"), adNS("licenseToken"), adNS("device"))).text except: # B&N Adobe PassHash fulfillment without device ID. # PassHash books aren't linked to a particular device, so there's no ID to send to Adobe. # If I understand Adobe's documentation correctly, PassHash books do not support notifications # and are not supposed to contain notify tags, so lets just skip if that's the case. print("Skipping notify due to passHash") continue full_text = "" full_text += "%s" % user full_text += "%s" % device # ADE 4.0 apparently changed the order of these two elements. # I still don't know exactly how this order is determined, but in most cases # ADE 4+ has the body first, then the nonce, while ADE 3 and lower usually has nonce first, then body. # It probably doesn't matter, but still, we want to behave exactly like ADE, so check the version number: devicexml = etree.parse(get_device_path()) for f in devicexml.findall("./%s" % (adNS("version"))): if f.get("name") == "hobbes": version = f.get("value") try: v_idx = VAR_VER_HOBBES_VERSIONS.index(version) clientVersion = VAR_VER_BUILD_IDS[v_idx] except: clientVersion = 0 if (clientVersion >= VAR_VER_USE_DIFFERENT_NOTIFICATION_XML_ORDER): full_text += etree.tostring(body, encoding="utf-8", pretty_print=True, xml_declaration=False).decode("utf-8") full_text += addNonce() else: full_text += addNonce() full_text += etree.tostring(body, encoding="utf-8", pretty_print=True, xml_declaration=False).decode("utf-8") full_text += "" NSMAP = { "adept" : "http://ns.adobe.com/adept" } etree.register_namespace("adept", NSMAP["adept"]) full_text_xml = etree.fromstring(full_text) signature = sign_node(full_text_xml) if (signature is None): print("SIGN ERROR!") continue etree.SubElement(full_text_xml, etree.QName(NSMAP["adept"], "signature")).text = signature doc_send = "\n" + etree.tostring(full_text_xml, encoding="utf-8", pretty_print=True, xml_declaration=False).decode("utf-8") # Debug: Print notify request #print(doc_send) try: code, msg = sendRequestDocuRC(doc_send, url) except: if not critical: print("There was an error during an optional fulfillment notification:") import traceback traceback.print_exc() print("Continuing execution ...") continue else: print("Error during critical notification:") raise try: msg = msg.decode("utf-8") except: pass if "\n") f.write(etree.tostring(activationxml, encoding="utf-8", pretty_print=True, xml_declaration=False).decode("utf-8")) f.close() return True, "Done"