Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

ENH: Support R6 decrypting #1015

Merged
merged 7 commits into from Jun 26, 2022
Merged

ENH: Support R6 decrypting #1015

Show file tree
Hide file tree
Changes from 3 commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
c992563
support R6 decrypting
exiledkingcc Jun 20, 2022
b4b74a4
make encryption test file names clear
exiledkingcc Jun 20, 2022
6cbf6e9
make decrypt clear
exiledkingcc Jun 20, 2022
532930b
make Encryption private for PdfReader
exiledkingcc Jun 26, 2022
5d076fe
add more tests
exiledkingcc Jun 26, 2022
10ff49d
Merge branch 'master' into encryption
exiledkingcc Jun 26, 2022
dfaa735
fix test with issue 327
exiledkingcc Jun 26, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
149 changes: 75 additions & 74 deletions PyPDF2/_encryption.py
View file
Open in desktop
Expand Up @@ -28,7 +28,7 @@
import hashlib
import random
import struct
from typing import Dict, Optional, Tuple, Union, cast
from typing import Optional, Tuple, Union, cast

from PyPDF2.errors import DependencyError
from PyPDF2.generic import (
Expand Down Expand Up @@ -226,16 +226,7 @@ def _padding(data: bytes) -> bytes:
return (data + _PADDING)[:32]


def _bytes(value: Union[bytes, str]) -> bytes:
if isinstance(value, bytes):
return value
try:
return value.encode("latin-1")
except Exception: # noqa
return value.encode("utf-8")


class AlgR4:
class AlgV4:
@staticmethod
def compute_key(
password: bytes,
Expand Down Expand Up @@ -397,10 +388,10 @@ def verify_user_password(
encryption dictionary’s U (user password) value (Security handlers of revision 3 or greater)") shall be used
to decrypt the document.
"""
key = AlgR4.compute_key(
key = AlgV4.compute_key(
user_pwd, rev, key_size, o_entry, P, id1_entry, metadata_encrypted
)
u_value = AlgR4.compute_U_value(key, rev, id1_entry)
u_value = AlgV4.compute_U_value(key, rev, id1_entry)
if rev >= 3:
u_value = u_value[:16]
u_entry = u_entry[:16]
Expand Down Expand Up @@ -434,7 +425,7 @@ def verify_owner_password(
c) The result of step (b) purports to be the user password. Authenticate this user password using "Algorithm 6:
Authenticating the user password". If it is correct, the password supplied is the correct owner password.
"""
rc4_key = AlgR4.compute_O_value_key(owner_pwd, rev, key_size)
rc4_key = AlgV4.compute_O_value_key(owner_pwd, rev, key_size)

if rev <= 2:
u_pwd = RC4_decrypt(rc4_key, o_entry)
Expand All @@ -443,15 +434,15 @@ def verify_owner_password(
for i in range(19, -1, -1):
key = bytes(bytearray(x ^ i for x in rc4_key))
u_pwd = RC4_decrypt(key, u_pwd)
return AlgR4.verify_user_password(
return AlgV4.verify_user_password(
u_pwd, rev, key_size, o_entry, u_entry, P, id1_entry, metadata_encrypted
)


class AlgR5:
class AlgV5:
@staticmethod
def verify_owner_password(
password: bytes, o_value: bytes, oe_value: bytes, u_value: bytes
R: int, password: bytes, o_value: bytes, oe_value: bytes, u_value: bytes
) -> bytes:
"""
Algorithm 3.2a Computing an encryption key
Expand Down Expand Up @@ -483,23 +474,45 @@ def verify_owner_password(
should match the value in the P key.
"""
password = password[:127]
if hashlib.sha256(password + o_value[32:40] + u_value).digest() != o_value[:32]:
if AlgV5.calculate_hash(R, password, o_value[32:40], u_value) != o_value[:32]:
return b""
iv = bytes(0 for _ in range(16))
tmp_key = hashlib.sha256(password + o_value[40:] + u_value).digest()
tmp_key = AlgV5.calculate_hash(R, password, o_value[40:], u_value)
key = AES_CBC_decrypt(tmp_key, iv, oe_value)
return key

@staticmethod
def verify_user_password(password: bytes, u_value: bytes, ue_value: bytes) -> bytes:
def verify_user_password(R: int, password: bytes, u_value: bytes, ue_value: bytes) -> bytes:
"""see :func:`verify_owner_password`"""
password = password[:127]
if hashlib.sha256(password + u_value[32:40]).digest() != u_value[:32]:
if AlgV5.calculate_hash(R, password, u_value[32:40], b"") != u_value[:32]:
return b""
iv = bytes(0 for _ in range(16))
tmp_key = hashlib.sha256(password + u_value[40:]).digest()
tmp_key = AlgV5.calculate_hash(R, password, u_value[40:], b"")
return AES_CBC_decrypt(tmp_key, iv, ue_value)

@staticmethod
def calculate_hash(R: int, password: bytes, salt: bytes, udata: bytes) -> bytes:
# from https://github.com/qpdf/qpdf/blob/main/libqpdf/QPDF_encryption.cc
K = hashlib.sha256(password + salt + udata).digest()
if R < 6:
return K
count = 0
while True:
count += 1
K1 = password + K + udata
E = AES_CBC_encrypt(K[:16], K[16:32], K1 * 64)
hash_fn = (
hashlib.sha256,
hashlib.sha384,
hashlib.sha512,
)[sum(E[:16]) % 3]
K = hash_fn(E).digest()
if count >= 64 and E[-1] <= count - 32:
print(count, E[-1])
MartinThoma marked this conversation as resolved.
Show resolved Hide resolved
break
return K[:32]

@staticmethod
def verify_perms(
key: bytes, perms: bytes, p: int, metadata_encrypted: bool
Expand All @@ -514,9 +527,9 @@ def verify_perms(
def generate_values(
user_pwd: bytes, owner_pwd: bytes, key: bytes, p: int, metadata_encrypted: bool
) -> dict:
u_value, ue_value = AlgR5.compute_U_value(user_pwd, key)
o_value, oe_value = AlgR5.compute_O_value(owner_pwd, key, u_value)
perms = AlgR5.compute_Perms_value(key, p, metadata_encrypted)
u_value, ue_value = AlgV5.compute_U_value(user_pwd, key)
o_value, oe_value = AlgV5.compute_O_value(owner_pwd, key, u_value)
perms = AlgV5.compute_Perms_value(key, p, metadata_encrypted)
return {
"/U": u_value,
"/UE": ue_value,
Expand Down Expand Up @@ -603,6 +616,7 @@ class Encryption:
def __init__(
self,
algV: int,
algR: int,
entry: DictionaryObject,
first_id_entry: bytes,
StmF: str,
Expand All @@ -611,17 +625,21 @@ def __init__(
) -> None:
# See TABLE 3.18 Entries common to all encryption dictionaries
self.algV = algV
self.algR = algR
self.entry = entry
self.key_size = entry.get("/Length", 40)
self.id1_entry = first_id_entry
self.StmF = StmF
self.StrF = StrF
self.EFF = EFF

# 1 => owner password
# 2 => user password
self._password_type = 0
MartinThoma marked this conversation as resolved.
Show resolved Hide resolved
self._key: Optional[bytes] = None
# keep key
self._user_keys: Dict = {}
self._owner_keys: Dict = {}

def verified(self) -> bool:
Copy link
Member

Choose a reason for hiding this comment

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

I guess verified means that the file was decrypted? Would is_decrypted be better?

Copy link
Member

Choose a reason for hiding this comment

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

Also, I think this should rather be _verified if the user does not need it.

return self._password_type != 0

def decrypt_object(self, obj: PdfObject, idnum: int, generation: int) -> PdfObject:
"""
Expand Down Expand Up @@ -666,7 +684,7 @@ def decrypt_object(self, obj: PdfObject, idnum: int, generation: int) -> PdfObje
key_hash.update(b"sAlT")
aes128_key = key_hash.digest()[: min(n + 5, 16)]

# for V=5 use AES-256
# for AES-256
aes256_key = key

stmCrypt = self._get_crypt(self.StmF, rc4_key, aes128_key, aes256_key)
Expand All @@ -689,45 +707,32 @@ def _get_crypt(
else:
return CryptRC4(rc4_key)

def verify(self, user_pwd: Union[bytes, str], owner_pwd: Union[bytes, str]) -> int:
up_bytes = _bytes(user_pwd)
op_bytes = _bytes(owner_pwd)

key = self._user_keys.get(up_bytes)
if key:
self._key = key
return 1

key = self._owner_keys.get(op_bytes)
if key:
self._key = key
return 2

rc = 0
if self.algV <= 4:
key, rc = self.verify_r4(up_bytes, op_bytes)
def verify(self, password: Union[bytes, str]) -> int:
if isinstance(password, str):
try:
pwd = password.encode("latin-1")
except Exception: # noqa
pwd = password.encode("utf-8")
Comment on lines +718 to +721
Copy link
Member

Choose a reason for hiding this comment

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

Why did you choose latin-1 as the default?

Copy link
Contributor Author

Choose a reason for hiding this comment

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

to be honest, i don't think too much about it, just copy it from previous code. 😀

Copy link
Member

Choose a reason for hiding this comment

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

Hehe, ok. Thanks for the honesty ❤️

else:
key, rc = self.verify_r5(up_bytes, op_bytes)
pwd = password

if rc == 1:
self._key = key
self._user_keys[up_bytes] = key
elif rc == 2:
key, rc = self.verify_v4(pwd) if self.algV <= 4 else self.verify_v5(pwd)
if rc != 0:
self._password_type = rc
self._key = key
self._owner_keys[op_bytes] = key

return rc

def verify_r4(self, user_pwd: bytes, owner_pwd: bytes) -> Tuple[bytes, int]:
def verify_v4(self, password: bytes) -> Tuple[bytes, int]:
R = cast(int, self.entry["/R"])
P = cast(int, self.entry["/P"])
P = (P + 0x100000000) % 0x100000000 # maybe < 0
metadata_encrypted = self.entry.get("/EncryptMetadata", True)
o_entry = cast(ByteStringObject, self.entry["/O"].get_object()).original_bytes
u_entry = cast(ByteStringObject, self.entry["/U"].get_object()).original_bytes

key = AlgR4.verify_user_password(
user_pwd,
# verify owner password first
key = AlgV4.verify_owner_password(
password,
R,
self.key_size,
o_entry,
Expand All @@ -737,9 +742,9 @@ def verify_r4(self, user_pwd: bytes, owner_pwd: bytes) -> Tuple[bytes, int]:
metadata_encrypted,
)
if key:
return key, 1
key = AlgR4.verify_owner_password(
owner_pwd,
return key, 2
key = AlgV4.verify_user_password(
password,
R,
self.key_size,
o_entry,
Expand All @@ -749,32 +754,31 @@ def verify_r4(self, user_pwd: bytes, owner_pwd: bytes) -> Tuple[bytes, int]:
metadata_encrypted,
)
if key:
return key, 2
return key, 1
return b"", 0

def verify_r5(self, user_pwd: bytes, owner_pwd: bytes) -> Tuple[bytes, int]:
def verify_v5(self, password: bytes) -> Tuple[bytes, int]:
# TODO: use SASLprep process
o_entry = cast(ByteStringObject, self.entry["/O"].get_object()).original_bytes
u_entry = cast(ByteStringObject, self.entry["/U"].get_object()).original_bytes
oe_entry = cast(ByteStringObject, self.entry["/OE"].get_object()).original_bytes
ue_entry = cast(ByteStringObject, self.entry["/UE"].get_object()).original_bytes

rc = 0
key = AlgR5.verify_user_password(user_pwd, u_entry, ue_entry)
if key:
# verify owner password first
key = AlgV5.verify_owner_password(self.algR, password, o_entry, oe_entry, u_entry)
rc = 2
if not key:
key = AlgV5.verify_user_password(self.algR, password, u_entry, ue_entry)
rc = 1
else:
key = AlgR5.verify_owner_password(owner_pwd, o_entry, oe_entry, u_entry)
if key:
rc = 2
if rc == 0:
if not key:
return b"", 0

# verify Perms
perms = cast(ByteStringObject, self.entry["/Perms"].get_object()).original_bytes
P = cast(int, self.entry["/P"])
P = (P + 0x100000000) % 0x100000000 # maybe < 0
metadata_encrypted = self.entry.get("/EncryptMetadata", True)
if not AlgR5.verify_perms(key, perms, P, metadata_encrypted):
if not AlgV5.verify_perms(key, perms, P, metadata_encrypted):
return b"", 0
return key, rc

Expand Down Expand Up @@ -818,7 +822,4 @@ def read(encryption_entry: DictionaryObject, first_id_entry: bytes) -> "Encrypti
raise NotImplementedError(f"EFF Method {EFF} NOT supported!")

R = cast(int, encryption_entry["/R"])
if R > 5:
raise NotImplementedError(f"encryption R={R} NOT supported!")

return Encryption(V, encryption_entry, first_id_entry, StmF, StrF, EFF)
return Encryption(V, R, encryption_entry, first_id_entry, StmF, StrF, EFF)
6 changes: 3 additions & 3 deletions PyPDF2/_merger.py
View file
Open in desktop
Expand Up @@ -137,7 +137,7 @@ def merge(
reader = PdfReader(stream, strict=self.strict) # type: ignore[arg-type]
self.inputs.append((stream, reader, my_file))
if encryption_obj is not None:
reader._encryption = encryption_obj
reader.encryption = encryption_obj
Copy link
Member

Choose a reason for hiding this comment

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

Every attribute that does not start with an underscore is part of the public interface. That means we cannot change the behavior / name without deprecation warnings.

Is there a reason why a user would want to access the encryption attribute directly?


# Find the range of pages to merge.
if pages is None:
Expand Down Expand Up @@ -204,8 +204,8 @@ def _create_stream(
stream = FileIO(fileobj, "rb")
my_file = True
elif isinstance(fileobj, PdfReader):
if hasattr(fileobj, "_encryption"):
encryption_obj = fileobj._encryption
if fileobj.encryption:
encryption_obj = fileobj.encryption
orig_tell = fileobj.stream.tell()
fileobj.stream.seek(0)
stream = BytesIO(fileobj.stream.read())
Expand Down