DeDRM_tools/DeDRM_plugin/ion.py

1041 lines
32 KiB
Python

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# ion.py
# Copyright © 2013-2020 Apprentice Harper et al.
__license__ = 'GPL v3'
__version__ = '3.0'
# Revision history:
# Pascal implementation by lulzkabulz.
# BinaryIon.pas + DrmIon.pas + IonSymbols.pas
# 1.0 - Python translation by apprenticenaomi.
# 1.1 - DeDRM integration by anon.
# 1.2 - Added pylzma import fallback
# 1.3 - Fixed lzma support for calibre 4.6+
# 2.0 - VoucherEnvelope v2/v3 support by apprenticesakuya.
# 3.0 - Added Python 3 compatibility for calibre 5.0
"""
Decrypt Kindle KFX files.
"""
import collections
import hashlib
import hmac
import os
import os.path
import struct
from io import BytesIO
from Crypto.Cipher import AES
from Crypto.Util.py3compat import bchr
try:
# lzma library from calibre 4.6.0 or later
import calibre_lzma.lzma1 as calibre_lzma
except ImportError:
calibre_lzma = None
# lzma library from calibre 2.35.0 or later
try:
import lzma.lzma1 as calibre_lzma
except ImportError:
calibre_lzma = None
try:
import lzma
except ImportError:
# Need pip backports.lzma on Python <3.3
try:
from backports import lzma
except ImportError:
# Windows-friendly choice: pylzma wheels
import pylzma as lzma
TID_NULL = 0
TID_BOOLEAN = 1
TID_POSINT = 2
TID_NEGINT = 3
TID_FLOAT = 4
TID_DECIMAL = 5
TID_TIMESTAMP = 6
TID_SYMBOL = 7
TID_STRING = 8
TID_CLOB = 9
TID_BLOB = 0xA
TID_LIST = 0xB
TID_SEXP = 0xC
TID_STRUCT = 0xD
TID_TYPEDECL = 0xE
TID_UNUSED = 0xF
SID_UNKNOWN = -1
SID_ION = 1
SID_ION_1_0 = 2
SID_ION_SYMBOL_TABLE = 3
SID_NAME = 4
SID_VERSION = 5
SID_IMPORTS = 6
SID_SYMBOLS = 7
SID_MAX_ID = 8
SID_ION_SHARED_SYMBOL_TABLE = 9
SID_ION_1_0_MAX = 10
LEN_IS_VAR_LEN = 0xE
LEN_IS_NULL = 0xF
VERSION_MARKER = [b"\x01", b"\x00", b"\xEA"]
# asserts must always raise exceptions for proper functioning
def _assert(test, msg="Exception"):
if not test:
raise Exception(msg)
class SystemSymbols(object):
ION = '$ion'
ION_1_0 = '$ion_1_0'
ION_SYMBOL_TABLE = '$ion_symbol_table'
NAME = 'name'
VERSION = 'version'
IMPORTS = 'imports'
SYMBOLS = 'symbols'
MAX_ID = 'max_id'
ION_SHARED_SYMBOL_TABLE = '$ion_shared_symbol_table'
class IonCatalogItem(object):
name = ""
version = 0
symnames = []
def __init__(self, name, version, symnames):
self.name = name
self.version = version
self.symnames = symnames
class SymbolToken(object):
text = ""
sid = 0
def __init__(self, text, sid):
if text == "" and sid == 0:
raise ValueError("Symbol token must have Text or SID")
self.text = text
self.sid = sid
class SymbolTable(object):
table = None
def __init__(self):
self.table = [None] * SID_ION_1_0_MAX
self.table[SID_ION] = SystemSymbols.ION
self.table[SID_ION_1_0] = SystemSymbols.ION_1_0
self.table[SID_ION_SYMBOL_TABLE] = SystemSymbols.ION_SYMBOL_TABLE
self.table[SID_NAME] = SystemSymbols.NAME
self.table[SID_VERSION] = SystemSymbols.VERSION
self.table[SID_IMPORTS] = SystemSymbols.IMPORTS
self.table[SID_SYMBOLS] = SystemSymbols.SYMBOLS
self.table[SID_MAX_ID] = SystemSymbols.MAX_ID
self.table[SID_ION_SHARED_SYMBOL_TABLE] = SystemSymbols.ION_SHARED_SYMBOL_TABLE
def findbyid(self, sid):
if sid < 1:
raise ValueError("Invalid symbol id")
if sid < len(self.table):
return self.table[sid]
else:
return ""
def import_(self, table, maxid):
for i in range(maxid):
self.table.append(table.symnames[i])
def importunknown(self, name, maxid):
for i in range(maxid):
self.table.append("%s#%d" % (name, i + 1))
class ParserState:
Invalid,BeforeField,BeforeTID,BeforeValue,AfterValue,EOF = 1,2,3,4,5,6
ContainerRec = collections.namedtuple("ContainerRec", "nextpos, tid, remaining")
class BinaryIonParser(object):
eof = False
state = None
localremaining = 0
needhasnext = False
isinstruct = False
valuetid = 0
valuefieldid = 0
parenttid = 0
valuelen = 0
valueisnull = False
valueistrue = False
value = None
didimports = False
def __init__(self, stream):
self.annotations = []
self.catalog = []
self.stream = stream
self.initpos = stream.tell()
self.reset()
self.symbols = SymbolTable()
def reset(self):
self.state = ParserState.BeforeTID
self.needhasnext = True
self.localremaining = -1
self.eof = False
self.isinstruct = False
self.containerstack = []
self.stream.seek(self.initpos)
def addtocatalog(self, name, version, symbols):
self.catalog.append(IonCatalogItem(name, version, symbols))
def hasnext(self):
while self.needhasnext and not self.eof:
self.hasnextraw()
if len(self.containerstack) == 0 and not self.valueisnull:
if self.valuetid == TID_SYMBOL:
if self.value == SID_ION_1_0:
self.needhasnext = True
elif self.valuetid == TID_STRUCT:
for a in self.annotations:
if a == SID_ION_SYMBOL_TABLE:
self.parsesymboltable()
self.needhasnext = True
break
return not self.eof
def hasnextraw(self):
self.clearvalue()
while self.valuetid == -1 and not self.eof:
self.needhasnext = False
if self.state == ParserState.BeforeField:
_assert(self.valuefieldid == SID_UNKNOWN)
self.valuefieldid = self.readfieldid()
if self.valuefieldid != SID_UNKNOWN:
self.state = ParserState.BeforeTID
else:
self.eof = True
elif self.state == ParserState.BeforeTID:
self.state = ParserState.BeforeValue
self.valuetid = self.readtypeid()
if self.valuetid == -1:
self.state = ParserState.EOF
self.eof = True
break
if self.valuetid == TID_TYPEDECL:
if self.valuelen == 0:
self.checkversionmarker()
else:
self.loadannotations()
elif self.state == ParserState.BeforeValue:
self.skip(self.valuelen)
self.state = ParserState.AfterValue
elif self.state == ParserState.AfterValue:
if self.isinstruct:
self.state = ParserState.BeforeField
else:
self.state = ParserState.BeforeTID
else:
_assert(self.state == ParserState.EOF)
def next(self):
if self.hasnext():
self.needhasnext = True
return self.valuetid
else:
return -1
def push(self, typeid, nextposition, nextremaining):
self.containerstack.append(ContainerRec(nextpos=nextposition, tid=typeid, remaining=nextremaining))
def stepin(self):
_assert(self.valuetid in [TID_STRUCT, TID_LIST, TID_SEXP] and not self.eof,
"valuetid=%s eof=%s" % (self.valuetid, self.eof))
_assert((not self.valueisnull or self.state == ParserState.AfterValue) and
(self.valueisnull or self.state == ParserState.BeforeValue))
nextrem = self.localremaining
if nextrem != -1:
nextrem -= self.valuelen
if nextrem < 0:
nextrem = 0
self.push(self.parenttid, self.stream.tell() + self.valuelen, nextrem)
self.isinstruct = (self.valuetid == TID_STRUCT)
if self.isinstruct:
self.state = ParserState.BeforeField
else:
self.state = ParserState.BeforeTID
self.localremaining = self.valuelen
self.parenttid = self.valuetid
self.clearvalue()
self.needhasnext = True
def stepout(self):
rec = self.containerstack.pop()
self.eof = False
self.parenttid = rec.tid
if self.parenttid == TID_STRUCT:
self.isinstruct = True
self.state = ParserState.BeforeField
else:
self.isinstruct = False
self.state = ParserState.BeforeTID
self.needhasnext = True
self.clearvalue()
curpos = self.stream.tell()
if rec.nextpos > curpos:
self.skip(rec.nextpos - curpos)
else:
_assert(rec.nextpos == curpos)
self.localremaining = rec.remaining
def read(self, count=1):
if self.localremaining != -1:
self.localremaining -= count
_assert(self.localremaining >= 0)
result = self.stream.read(count)
if len(result) == 0:
raise EOFError()
return result
def readfieldid(self):
if self.localremaining != -1 and self.localremaining < 1:
return -1
try:
return self.readvaruint()
except EOFError:
return -1
def readtypeid(self):
if self.localremaining != -1:
if self.localremaining < 1:
return -1
self.localremaining -= 1
b = self.stream.read(1)
if len(b) < 1:
return -1
b = ord(b)
result = b >> 4
ln = b & 0xF
if ln == LEN_IS_VAR_LEN:
ln = self.readvaruint()
elif ln == LEN_IS_NULL:
ln = 0
self.state = ParserState.AfterValue
elif result == TID_NULL:
# Must have LEN_IS_NULL
_assert(False)
elif result == TID_BOOLEAN:
_assert(ln <= 1)
self.valueistrue = (ln == 1)
ln = 0
self.state = ParserState.AfterValue
elif result == TID_STRUCT:
if ln == 1:
ln = self.readvaruint()
self.valuelen = ln
return result
def readvarint(self):
b = ord(self.read())
negative = ((b & 0x40) != 0)
result = (b & 0x3F)
i = 0
while (b & 0x80) == 0 and i < 4:
b = ord(self.read())
result = (result << 7) | (b & 0x7F)
i += 1
_assert(i < 4 or (b & 0x80) != 0, "int overflow")
if negative:
return -result
return result
def readvaruint(self):
b = ord(self.read())
result = (b & 0x7F)
i = 0
while (b & 0x80) == 0 and i < 4:
b = ord(self.read())
result = (result << 7) | (b & 0x7F)
i += 1
_assert(i < 4 or (b & 0x80) != 0, "int overflow")
return result
def readdecimal(self):
if self.valuelen == 0:
return 0.
rem = self.localremaining - self.valuelen
self.localremaining = self.valuelen
exponent = self.readvarint()
_assert(self.localremaining > 0, "Only exponent in ReadDecimal")
_assert(self.localremaining <= 8, "Decimal overflow")
signed = False
b = [ord(x) for x in self.read(self.localremaining)]
if (b[0] & 0x80) != 0:
b[0] = b[0] & 0x7F
signed = True
# Convert variably sized network order integer into 64-bit little endian
j = 0
vb = [0] * 8
for i in range(len(b), -1, -1):
vb[i] = b[j]
j += 1
v = struct.unpack("<Q", b"".join(bchr(x) for x in vb))[0]
result = v * (10 ** exponent)
if signed:
result = -result
self.localremaining = rem
return result
def skip(self, count):
if self.localremaining != -1:
self.localremaining -= count
if self.localremaining < 0:
raise EOFError()
self.stream.seek(count, os.SEEK_CUR)
def parsesymboltable(self):
self.next() # shouldn't do anything?
_assert(self.valuetid == TID_STRUCT)
if self.didimports:
return
self.stepin()
fieldtype = self.next()
while fieldtype != -1:
if not self.valueisnull:
_assert(self.valuefieldid == SID_IMPORTS, "Unsupported symbol table field id")
if fieldtype == TID_LIST:
self.gatherimports()
fieldtype = self.next()
self.stepout()
self.didimports = True
def gatherimports(self):
self.stepin()
t = self.next()
while t != -1:
if not self.valueisnull and t == TID_STRUCT:
self.readimport()
t = self.next()
self.stepout()
def readimport(self):
version = -1
maxid = -1
name = ""
self.stepin()
t = self.next()
while t != -1:
if not self.valueisnull and self.valuefieldid != SID_UNKNOWN:
if self.valuefieldid == SID_NAME:
name = self.stringvalue()
elif self.valuefieldid == SID_VERSION:
version = self.intvalue()
elif self.valuefieldid == SID_MAX_ID:
maxid = self.intvalue()
t = self.next()
self.stepout()
if name == "" or name == SystemSymbols.ION:
return
if version < 1:
version = 1
table = self.findcatalogitem(name)
if maxid < 0:
_assert(table is not None and version == table.version, "Import %s lacks maxid" % name)
maxid = len(table.symnames)
if table is not None:
self.symbols.import_(table, min(maxid, len(table.symnames)))
else:
self.symbols.importunknown(name, maxid)
def intvalue(self):
_assert(self.valuetid in [TID_POSINT, TID_NEGINT], "Not an int")
self.preparevalue()
return self.value
def stringvalue(self):
_assert(self.valuetid == TID_STRING, "Not a string")
if self.valueisnull:
return ""
self.preparevalue()
return self.value
def symbolvalue(self):
_assert(self.valuetid == TID_SYMBOL, "Not a symbol")
self.preparevalue()
result = self.symbols.findbyid(self.value)
if result == "":
result = "SYMBOL#%d" % self.value
return result
def lobvalue(self):
_assert(self.valuetid in [TID_CLOB, TID_BLOB], "Not a LOB type: %s" % self.getfieldname())
if self.valueisnull:
return None
result = self.read(self.valuelen)
self.state = ParserState.AfterValue
return result
def decimalvalue(self):
_assert(self.valuetid == TID_DECIMAL, "Not a decimal")
self.preparevalue()
return self.value
def preparevalue(self):
if self.value is None:
self.loadscalarvalue()
def loadscalarvalue(self):
if self.valuetid not in [TID_NULL, TID_BOOLEAN, TID_POSINT, TID_NEGINT,
TID_FLOAT, TID_DECIMAL, TID_TIMESTAMP,
TID_SYMBOL, TID_STRING]:
return
if self.valueisnull:
self.value = None
return
if self.valuetid == TID_STRING:
self.value = self.read(self.valuelen).decode("UTF-8")
elif self.valuetid in (TID_POSINT, TID_NEGINT, TID_SYMBOL):
if self.valuelen == 0:
self.value = 0
else:
_assert(self.valuelen <= 4, "int too long: %d" % self.valuelen)
v = 0
for i in range(self.valuelen - 1, -1, -1):
v = (v | (ord(self.read()) << (i * 8)))
if self.valuetid == TID_NEGINT:
self.value = -v
else:
self.value = v
elif self.valuetid == TID_DECIMAL:
self.value = self.readdecimal()
#else:
# _assert(False, "Unhandled scalar type %d" % self.valuetid)
self.state = ParserState.AfterValue
def clearvalue(self):
self.valuetid = -1
self.value = None
self.valueisnull = False
self.valuefieldid = SID_UNKNOWN
self.annotations = []
def loadannotations(self):
ln = self.readvaruint()
maxpos = self.stream.tell() + ln
while self.stream.tell() < maxpos:
self.annotations.append(self.readvaruint())
self.valuetid = self.readtypeid()
def checkversionmarker(self):
for i in VERSION_MARKER:
_assert(self.read() == i, "Unknown version marker")
self.valuelen = 0
self.valuetid = TID_SYMBOL
self.value = SID_ION_1_0
self.valueisnull = False
self.valuefieldid = SID_UNKNOWN
self.state = ParserState.AfterValue
def findcatalogitem(self, name):
for result in self.catalog:
if result.name == name:
return result
def forceimport(self, symbols):
item = IonCatalogItem("Forced", 1, symbols)
self.symbols.import_(item, len(symbols))
def getfieldname(self):
if self.valuefieldid == SID_UNKNOWN:
return ""
return self.symbols.findbyid(self.valuefieldid)
def getfieldnamesymbol(self):
return SymbolToken(self.getfieldname(), self.valuefieldid)
def gettypename(self):
if len(self.annotations) == 0:
return ""
return self.symbols.findbyid(self.annotations[0])
@staticmethod
def printlob(b):
if b is None:
return "null"
result = ""
for i in b:
result += ("%02x " % ord(i))
if len(result) > 0:
result = result[:-1]
return result
def ionwalk(self, supert, indent, lst):
while self.hasnext():
if supert == TID_STRUCT:
L = self.getfieldname() + ":"
else:
L = ""
t = self.next()
if t in [TID_STRUCT, TID_LIST]:
if L != "":
lst.append(indent + L)
L = self.gettypename()
if L != "":
lst.append(indent + L + "::")
if t == TID_STRUCT:
lst.append(indent + "{")
else:
lst.append(indent + "[")
self.stepin()
self.ionwalk(t, indent + " ", lst)
self.stepout()
if t == TID_STRUCT:
lst.append(indent + "}")
else:
lst.append(indent + "]")
else:
if t == TID_STRING:
L += ('"%s"' % self.stringvalue())
elif t in [TID_CLOB, TID_BLOB]:
L += ("{%s}" % self.printlob(self.lobvalue()))
elif t == TID_POSINT:
L += str(self.intvalue())
elif t == TID_SYMBOL:
tn = self.gettypename()
if tn != "":
tn += "::"
L += tn + self.symbolvalue()
elif t == TID_DECIMAL:
L += str(self.decimalvalue())
else:
L += ("TID %d" % t)
lst.append(indent + L)
def print_(self, lst):
self.reset()
self.ionwalk(-1, "", lst)
SYM_NAMES = [ 'com.amazon.drm.Envelope@1.0',
'com.amazon.drm.EnvelopeMetadata@1.0', 'size', 'page_size',
'encryption_key', 'encryption_transformation',
'encryption_voucher', 'signing_key', 'signing_algorithm',
'signing_voucher', 'com.amazon.drm.EncryptedPage@1.0',
'cipher_text', 'cipher_iv', 'com.amazon.drm.Signature@1.0',
'data', 'com.amazon.drm.EnvelopeIndexTable@1.0', 'length',
'offset', 'algorithm', 'encoded', 'encryption_algorithm',
'hashing_algorithm', 'expires', 'format', 'id',
'lock_parameters', 'strategy', 'com.amazon.drm.Key@1.0',
'com.amazon.drm.KeySet@1.0', 'com.amazon.drm.PIDv3@1.0',
'com.amazon.drm.PlainTextPage@1.0',
'com.amazon.drm.PlainText@1.0', 'com.amazon.drm.PrivateKey@1.0',
'com.amazon.drm.PublicKey@1.0', 'com.amazon.drm.SecretKey@1.0',
'com.amazon.drm.Voucher@1.0', 'public_key', 'private_key',
'com.amazon.drm.KeyPair@1.0', 'com.amazon.drm.ProtectedData@1.0',
'doctype', 'com.amazon.drm.EnvelopeIndexTableOffset@1.0',
'enddoc', 'license_type', 'license', 'watermark', 'key', 'value',
'com.amazon.drm.License@1.0', 'category', 'metadata',
'categorized_metadata', 'com.amazon.drm.CategorizedMetadata@1.0',
'com.amazon.drm.VoucherEnvelope@1.0', 'mac', 'voucher',
'com.amazon.drm.ProtectedData@2.0',
'com.amazon.drm.Envelope@2.0',
'com.amazon.drm.EnvelopeMetadata@2.0',
'com.amazon.drm.EncryptedPage@2.0',
'com.amazon.drm.PlainText@2.0', 'compression_algorithm',
'com.amazon.drm.Compressed@1.0', 'page_index_table',
'com.amazon.drm.VoucherEnvelope@2.0', 'com.amazon.drm.VoucherEnvelope@3.0' ]
def addprottable(ion):
ion.addtocatalog("ProtectedData", 1, SYM_NAMES)
def pkcs7pad(msg, blocklen):
paddinglen = blocklen - len(msg) % blocklen
padding = bchr(paddinglen) * paddinglen
return msg + padding
def pkcs7unpad(msg, blocklen):
_assert(len(msg) % blocklen == 0)
paddinglen = msg[-1]
_assert(paddinglen > 0 and paddinglen <= blocklen, "Incorrect padding - Wrong key")
_assert(msg[-paddinglen:] == bchr(paddinglen) * paddinglen, "Incorrect padding - Wrong key")
return msg[:-paddinglen]
# every VoucherEnvelope version has a corresponding "word" and magic number, used in obfuscating the shared secret
VOUCHER_VERSION_INFOS = {
2: [b'Antidisestablishmentarianism', 5],
3: [b'Floccinaucinihilipilification', 8]
}
# obfuscate shared secret according to the VoucherEnvelope version
def obfuscate(secret, version):
if version == 1: # v1 does not use obfuscation
return secret
params = VOUCHER_VERSION_INFOS[version]
word = params[0]
magic = params[1]
# extend secret so that its length is divisible by the magic number
if len(secret) % magic != 0:
secret = secret + b'\x00' * (magic - len(secret) % magic)
secret = bytearray(secret)
obfuscated = bytearray(len(secret))
wordhash = bytearray(hashlib.sha256(word).digest())
# shuffle secret and xor it with the first half of the word hash
for i in range(0, len(secret)):
index = i // (len(secret) // magic) + magic * (i % (len(secret) // magic))
obfuscated[index] = secret[i] ^ wordhash[index % 16]
return obfuscated
class DrmIonVoucher(object):
envelope = None
version = None
voucher = None
drmkey = None
license_type = "Unknown"
encalgorithm = ""
enctransformation = ""
hashalgorithm = ""
lockparams = None
ciphertext = b""
cipheriv = b""
secretkey = b""
def __init__(self, voucherenv, dsn, secret):
self.dsn, self.secret = dsn, secret
self.lockparams = []
self.envelope = BinaryIonParser(voucherenv)
addprottable(self.envelope)
def decryptvoucher(self):
shared = "PIDv3" + self.encalgorithm + self.enctransformation + self.hashalgorithm
self.lockparams.sort()
for param in self.lockparams:
if param == "ACCOUNT_SECRET":
shared += param + self.secret
elif param == "CLIENT_ID":
shared += param + self.dsn
else:
_assert(False, "Unknown lock parameter: %s" % param)
sharedsecret = obfuscate(shared.encode('ASCII'), self.version)
key = hmac.new(sharedsecret, b"PIDv3", digestmod=hashlib.sha256).digest()
aes = AES.new(key[:32], AES.MODE_CBC, self.cipheriv[:16])
b = aes.decrypt(self.ciphertext)
b = pkcs7unpad(b, 16)
self.drmkey = BinaryIonParser(BytesIO(b))
addprottable(self.drmkey)
_assert(self.drmkey.hasnext() and self.drmkey.next() == TID_LIST and self.drmkey.gettypename() == "com.amazon.drm.KeySet@1.0",
"Expected KeySet, got %s" % self.drmkey.gettypename())
self.drmkey.stepin()
while self.drmkey.hasnext():
self.drmkey.next()
if self.drmkey.gettypename() != "com.amazon.drm.SecretKey@1.0":
continue
self.drmkey.stepin()
while self.drmkey.hasnext():
self.drmkey.next()
if self.drmkey.getfieldname() == "algorithm":
_assert(self.drmkey.stringvalue() == "AES", "Unknown cipher algorithm: %s" % self.drmkey.stringvalue())
elif self.drmkey.getfieldname() == "format":
_assert(self.drmkey.stringvalue() == "RAW", "Unknown key format: %s" % self.drmkey.stringvalue())
elif self.drmkey.getfieldname() == "encoded":
self.secretkey = self.drmkey.lobvalue()
self.drmkey.stepout()
break
self.drmkey.stepout()
def parse(self):
self.envelope.reset()
_assert(self.envelope.hasnext(), "Envelope is empty")
_assert(self.envelope.next() == TID_STRUCT and str.startswith(self.envelope.gettypename(), "com.amazon.drm.VoucherEnvelope@"),
"Unknown type encountered in envelope, expected VoucherEnvelope")
self.version = int(self.envelope.gettypename().split('@')[1][:-2])
self.envelope.stepin()
while self.envelope.hasnext():
self.envelope.next()
field = self.envelope.getfieldname()
if field == "voucher":
self.voucher = BinaryIonParser(BytesIO(self.envelope.lobvalue()))
addprottable(self.voucher)
continue
elif field != "strategy":
continue
_assert(self.envelope.gettypename() == "com.amazon.drm.PIDv3@1.0", "Unknown strategy: %s" % self.envelope.gettypename())
self.envelope.stepin()
while self.envelope.hasnext():
self.envelope.next()
field = self.envelope.getfieldname()
if field == "encryption_algorithm":
self.encalgorithm = self.envelope.stringvalue()
elif field == "encryption_transformation":
self.enctransformation = self.envelope.stringvalue()
elif field == "hashing_algorithm":
self.hashalgorithm = self.envelope.stringvalue()
elif field == "lock_parameters":
self.envelope.stepin()
while self.envelope.hasnext():
_assert(self.envelope.next() == TID_STRING, "Expected string list for lock_parameters")
self.lockparams.append(self.envelope.stringvalue())
self.envelope.stepout()
self.envelope.stepout()
self.parsevoucher()
def parsevoucher(self):
_assert(self.voucher.hasnext(), "Voucher is empty")
_assert(self.voucher.next() == TID_STRUCT and self.voucher.gettypename() == "com.amazon.drm.Voucher@1.0",
"Unknown type, expected Voucher")
self.voucher.stepin()
while self.voucher.hasnext():
self.voucher.next()
if self.voucher.getfieldname() == "cipher_iv":
self.cipheriv = self.voucher.lobvalue()
elif self.voucher.getfieldname() == "cipher_text":
self.ciphertext = self.voucher.lobvalue()
elif self.voucher.getfieldname() == "license":
_assert(self.voucher.gettypename() == "com.amazon.drm.License@1.0",
"Unknown license: %s" % self.voucher.gettypename())
self.voucher.stepin()
while self.voucher.hasnext():
self.voucher.next()
if self.voucher.getfieldname() == "license_type":
self.license_type = self.voucher.stringvalue()
self.voucher.stepout()
def printenvelope(self, lst):
self.envelope.print_(lst)
def printkey(self, lst):
if self.voucher is None:
self.parse()
if self.drmkey is None:
self.decryptvoucher()
self.drmkey.print_(lst)
def printvoucher(self, lst):
if self.voucher is None:
self.parse()
self.voucher.print_(lst)
def getlicensetype(self):
return self.license_type
class DrmIon(object):
ion = None
voucher = None
vouchername = ""
key = b""
onvoucherrequired = None
def __init__(self, ionstream, onvoucherrequired):
self.ion = BinaryIonParser(ionstream)
addprottable(self.ion)
self.onvoucherrequired = onvoucherrequired
def parse(self, outpages):
self.ion.reset()
_assert(self.ion.hasnext(), "DRMION envelope is empty")
_assert(self.ion.next() == TID_SYMBOL and self.ion.gettypename() == "doctype", "Expected doctype symbol")
_assert(self.ion.next() == TID_LIST and self.ion.gettypename() in ["com.amazon.drm.Envelope@1.0", "com.amazon.drm.Envelope@2.0"],
"Unknown type encountered in DRMION envelope, expected Envelope, got %s" % self.ion.gettypename())
while True:
if self.ion.gettypename() == "enddoc":
break
self.ion.stepin()
while self.ion.hasnext():
self.ion.next()
if self.ion.gettypename() in ["com.amazon.drm.EnvelopeMetadata@1.0", "com.amazon.drm.EnvelopeMetadata@2.0"]:
self.ion.stepin()
while self.ion.hasnext():
self.ion.next()
if self.ion.getfieldname() != "encryption_voucher":
continue
if self.vouchername == "":
self.vouchername = self.ion.stringvalue()
self.voucher = self.onvoucherrequired(self.vouchername)
self.key = self.voucher.secretkey
_assert(self.key is not None, "Unable to obtain secret key from voucher")
else:
_assert(self.vouchername == self.ion.stringvalue(),
"Unexpected: Different vouchers required for same file?")
self.ion.stepout()
elif self.ion.gettypename() in ["com.amazon.drm.EncryptedPage@1.0", "com.amazon.drm.EncryptedPage@2.0"]:
decompress = False
ct = None
civ = None
self.ion.stepin()
while self.ion.hasnext():
self.ion.next()
if self.ion.gettypename() == "com.amazon.drm.Compressed@1.0":
decompress = True
if self.ion.getfieldname() == "cipher_text":
ct = self.ion.lobvalue()
elif self.ion.getfieldname() == "cipher_iv":
civ = self.ion.lobvalue()
if ct is not None and civ is not None:
self.processpage(ct, civ, outpages, decompress)
self.ion.stepout()
self.ion.stepout()
if not self.ion.hasnext():
break
self.ion.next()
def print_(self, lst):
self.ion.print_(lst)
def processpage(self, ct, civ, outpages, decompress):
aes = AES.new(self.key[:16], AES.MODE_CBC, civ[:16])
msg = pkcs7unpad(aes.decrypt(ct), 16)
if not decompress:
outpages.write(msg)
return
_assert(msg[0] == 0, "LZMA UseFilter not supported")
if calibre_lzma is not None:
with calibre_lzma.decompress(msg[1:], bufsize=0x1000000) as f:
f.seek(0)
outpages.write(f.read())
return
decomp = lzma.LZMADecompressor(format=lzma.FORMAT_ALONE)
while not decomp.eof:
segment = decomp.decompress(msg[1:])
msg = b"" # Contents were internally buffered after the first call
outpages.write(segment)