JMR-devandClaude Opus 4.8
bb96fe90ac
Move the copies of unpad, crc32, checksumPid and pidFromSerial that were scattered across the plugin into utilities.py, and add a shared safe_join that generalizes the Topaz extraction path-traversal fix. - unpad: adobekey, ineptepub and ineptpdf import the shared helper. The four bare-script key tools keep their local copies since they run without a package context. - checksumPid is type-preserving (bytes for kgenpids/kindlepid, str for mobidedrm) so every call site keeps its exact behavior. - kgenpids falls back to an absolute import because it is imported top-level by the worker modules. - topazextract.extractFiles uses safe_join; genbook is unchanged as it only handles already-sanitized names. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
124 lines
3.8 KiB
Python
124 lines
3.8 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
|
|
#@@CALIBRE_COMPAT_CODE@@
|
|
|
|
__license__ = 'GPL v3'
|
|
|
|
import binascii
|
|
import os
|
|
|
|
|
|
def uStrCmp (s1, s2, caseless=False):
|
|
import unicodedata as ud
|
|
str1 = s1 if isinstance(s1, str) else str(s1)
|
|
str2 = s2 if isinstance(s2, str) else str(s2)
|
|
|
|
if caseless:
|
|
return ud.normalize('NFC', str1.lower()) == ud.normalize('NFC', str2.lower())
|
|
else:
|
|
return ud.normalize('NFC', str1) == ud.normalize('NFC', str2)
|
|
|
|
|
|
|
|
# Wrap a stream so that output gets flushed immediately
|
|
# and also make sure that any unicode strings get safely
|
|
# encoded using "replace" before writing them.
|
|
class SafeUnbuffered:
|
|
def __init__(self, stream):
|
|
self.stream = stream
|
|
self.encoding = stream.encoding
|
|
if self.encoding == None:
|
|
self.encoding = "utf-8"
|
|
def write(self, data):
|
|
if isinstance(data,str):
|
|
data = data.encode(self.encoding,"replace")
|
|
try:
|
|
buffer = getattr(self.stream, 'buffer', self.stream)
|
|
# self.stream.buffer for Python3, self.stream for Python2
|
|
buffer.write(data)
|
|
buffer.flush()
|
|
except:
|
|
# We can do nothing if a write fails
|
|
raise
|
|
def __getattr__(self, attr):
|
|
return getattr(self.stream, attr)
|
|
|
|
|
|
def unpad(data, padding=16):
|
|
"""Strip PKCS#7-style padding by trusting the final byte as the pad length.
|
|
|
|
This matches the historical inline implementations across the plugin: it
|
|
does not validate that every pad byte is equal, it simply removes the
|
|
number of trailing bytes given by the last byte.
|
|
"""
|
|
pad_len = data[-1]
|
|
|
|
return data[:-pad_len]
|
|
|
|
|
|
# Alphabet used to encode Kindle/Mobipocket PID checksum characters.
|
|
PID_ALPHABET = b'ABCDEFGHIJKLMNPQRSTUVWXYZ123456789'
|
|
|
|
|
|
def crc32(s):
|
|
return (~binascii.crc32(s, -1)) & 0xFFFFFFFF
|
|
|
|
|
|
def checksumPid(s):
|
|
"""Convert an 8-digit PID into a 10-digit PID with a 2-character checksum.
|
|
|
|
The result is the same type (``str`` or ``bytes``) as the argument, which
|
|
preserves the historical contracts of every call site: kgenpids and
|
|
kindlepid pass and expect ``bytes`` while mobidedrm passes and expects
|
|
``str``.
|
|
"""
|
|
want_str = isinstance(s, str)
|
|
if want_str:
|
|
s = s.encode()
|
|
crc = crc32(s)
|
|
crc = crc ^ (crc >> 16)
|
|
res = s
|
|
length = len(PID_ALPHABET)
|
|
for _ in (0, 1):
|
|
b = crc & 0xff
|
|
pos = (b // length) ^ (b % length)
|
|
res += bytes(bytearray([PID_ALPHABET[pos % length]]))
|
|
crc >>= 8
|
|
return res.decode() if want_str else res
|
|
|
|
|
|
def pidFromSerial(s, l):
|
|
"""Convert an (old Kindle) serial number into a fixed-length PID.
|
|
|
|
Takes and returns ``bytes``.
|
|
"""
|
|
crc = crc32(s)
|
|
arr1 = [0] * l
|
|
for i in range(len(s)):
|
|
arr1[i % l] ^= s[i]
|
|
crc_bytes = [crc >> 24 & 0xff, crc >> 16 & 0xff, crc >> 8 & 0xff, crc & 0xff]
|
|
for i in range(l):
|
|
arr1[i] ^= crc_bytes[i & 3]
|
|
pid = b""
|
|
for i in range(l):
|
|
b = arr1[i] & 0xff
|
|
pid += bytes(bytearray([PID_ALPHABET[(b >> 7) + ((b >> 5 & 3) ^ (b & 0x1f))]]))
|
|
return pid
|
|
|
|
|
|
def safe_join(directory, filename):
|
|
"""Join *filename* onto *directory*, guaranteeing the result stays inside it.
|
|
|
|
Any directory components in *filename* are stripped (defeating ``../``
|
|
traversal) and the resolved path is verified to be contained within
|
|
*directory*. Raises ``ValueError`` for a name that would still escape.
|
|
"""
|
|
filename = os.path.basename(filename)
|
|
target = os.path.join(directory, filename)
|
|
real_target = os.path.abspath(target)
|
|
real_dir = os.path.abspath(directory)
|
|
if not (real_target == real_dir or real_target.startswith(real_dir + os.sep)):
|
|
raise ValueError("Unsafe path: {0!r} escapes {1!r}".format(filename, directory))
|
|
return target
|