"""
Utilities to perform encryption following the PDF standards.
The contents of this module are internal to fpdf2, and not part of the public API.
They may change at any time without prior warning or any deprecation period,
in non-backward-compatible ways.
"""
import hashlib
import logging
import math
import stringprep
import unicodedata
from binascii import hexlify
from codecs import BOM_UTF16_BE
from os import urandom
from typing import Callable, Iterable, Type, Union
from .enums import AccessPermission, EncryptionMethod
from .errors import FPDFException
from .syntax import Name, PDFObject, PDFString, build_obj_dict
from .syntax import create_dictionary_string as pdf_dict
# try to use cryptography for AES encryption
try:
from cryptography.hazmat.primitives.ciphers import Cipher, modes
from cryptography.hazmat.primitives.ciphers.algorithms import AES128, AES256
from cryptography.hazmat.primitives.padding import PKCS7
import_error = None
except ImportError as error:
import_error = error
LOGGER = logging.getLogger(__name__)
class ARC4:
"""
This is a simplified version of the ARC4 (alleged RC4) algorithm,
created based on the following sources:
* Wikipedia article on RC4
* github.com/manojpandey/rc4 (MIT License)
* http://people.csail.mit.edu/rivest/pubs/RS14.pdf
Having this ARC4 implementation makes it possible to have basic
encryption functions without additional dependencies
"""
MOD = 256
def KSA(self, key: bytes) -> list:
key_length = len(key)
S = list(range(self.MOD))
j = 0
for i in range(self.MOD):
j = (j + S[i] + key[i % key_length]) % self.MOD
S[i], S[j] = S[j], S[i]
return S
def PRGA(self, S: list) -> Iterable[int]:
i = 0
j = 0
while True:
i = (i + 1) % self.MOD
j = (j + S[i]) % self.MOD
S[i], S[j] = S[j], S[i]
K = S[(S[i] + S[j]) % self.MOD]
yield K
def encrypt(self, key: bytes, text: Union[bytes, bytearray]) -> list:
keystream = self.PRGA(self.KSA(key))
res = []
for c in text:
res.append(c ^ next(keystream))
return res
class CryptFilter:
"""Represents one crypt filter, listed under CF inside the encryption dictionary"""
def __init__(self, mode: str, length: int) -> None:
super().__init__()
self.type = Name("CryptFilter")
self.c_f_m = Name(mode)
self.length = int(length / 8)
def serialize(self) -> str:
obj_dict = build_obj_dict({key: getattr(self, key) for key in dir(self)})
return pdf_dict(obj_dict)
class EncryptionDictionary(PDFObject):
"""
This class represents an encryption dictionary
PDF 32000 reference - Table 20
The PDF trailer must reference this object (/Encrypt)
"""
def __init__(self, security_handler: "StandardSecurityHandler") -> None:
super().__init__()
self.filter = Name("Standard")
self.length = security_handler.key_length
self.r = security_handler.revision
self.o = f"<{security_handler.o.upper()}>"
self.u = f"<{security_handler.u.upper()}>"
if security_handler.revision == 6:
self.o_e = f"<{security_handler.oe.upper()}>"
self.u_e = f"<{security_handler.ue.upper()}>"
self.perms = f"<{security_handler.perms.upper()}>"
self.v = security_handler.version
self.p = int32(security_handler.access_permission)
if not security_handler.encrypt_metadata:
self.encrypt_metadata = "false"
if security_handler.cf:
self.c_f = pdf_dict({"/StdCF": security_handler.cf.serialize()})
if security_handler.encryption_method == EncryptionMethod.NO_ENCRYPTION:
self.stm_f = Name("Identity") # crypt filter for streams
self.str_f = Name("Identity") # crypt filter for strings
else:
self.stm_f = Name("StdCF") # crypt filter for streams
self.str_f = Name("StdCF") # crypt filter for strings
class StandardSecurityHandler:
"""
This class is referenced in the main PDF class and is used to handle all encryption functions
* Calculate password and hashes
* Provide encrypt method to be called by stream and strings
* Set the access permissions on the document
"""
DEFAULT_PADDING = (
b"(\xbfN^Nu\x8aAd\x00NV\xff\xfa\x01\x08..\x00\xb6\xd0h>\x80/\x0c\xa9\xfedSiz"
)
def __init__(
self,
fpdf,
owner_password: str,
user_password: Union[str, None] = None,
permission: AccessPermission = AccessPermission.all(),
encryption_method: EncryptionMethod = EncryptionMethod.RC4,
encrypt_metadata: bool = False,
):
self.fpdf = fpdf
self.access_permission = (
0b11111111111111111111000011000000
if permission is None
else (0b11111111111111111111000011000000 | permission)
)
self.owner_password = owner_password
self.user_password = user_password if user_password else ""
self.encryption_method = encryption_method
self.cf = None
self.key_length = 128
if import_error and self.encryption_method in (
EncryptionMethod.AES_128,
EncryptionMethod.AES_256,
):
raise EnvironmentError(
"cryptography module not available"
" - Try: 'pip install cryptography' or use RC4 encryption method"
f" - Import error was: {import_error}"
)
if self.encryption_method == EncryptionMethod.AES_128:
self.version = 4
self.revision = 4
fpdf._set_min_pdf_version("1.6")
self.cf = CryptFilter(mode="AESV2", length=self.key_length)
elif self.encryption_method == EncryptionMethod.AES_256:
self.version = 5
self.revision = 6
fpdf._set_min_pdf_version("2.0")
self.key_length = 256
self.cf = CryptFilter(mode="AESV3", length=self.key_length)
elif self.encryption_method == EncryptionMethod.NO_ENCRYPTION:
self.version = 4
self.revision = 4
fpdf._set_min_pdf_version("1.6")
self.cf = CryptFilter(mode="V2", length=self.key_length)
else:
self.version = 2
self.revision = 3
fpdf._set_min_pdf_version("1.5")
# not including crypt filter because it's only required on V=4
# if needed, it would be CryptFilter(mode=V2)
self.encrypt_metadata = encrypt_metadata
def generate_passwords(self, file_id: str) -> None:
"""File_id is the first hash of the PDF file id"""
self.file_id = file_id
self.info_id = file_id[1:33]
if self.revision == 6:
self.k = self.get_random_bytes(32)
self.generate_user_password_rev6()
self.generate_owner_password_rev6()
self.generate_perms_rev6()
else:
self.o = self.generate_owner_password()
self.k = self.generate_encryption_key()
self.u = self.generate_user_password()
def get_encryption_obj(self) -> EncryptionDictionary:
"""Return an encryption dictionary"""
return EncryptionDictionary(self)
def encrypt(
self, text: Union[str, bytearray, bytes], obj_id: int
) -> Union[str, bytes]:
"""Method invoked by PDFObject and PDFContentStream to encrypt strings and streams"""
LOGGER.debug("Encrypting %s", text)
return (
self.encrypt_stream(text, obj_id)
if isinstance(text, (bytearray, bytes))
else self.encrypt_string(text, obj_id)
)
def encrypt_string(self, string: str, obj_id: int) -> str:
if self.encryption_method == EncryptionMethod.NO_ENCRYPTION:
return PDFString(string, encrypt=False).serialize()
LOGGER.debug("Encrypting string: %s", string)
try:
string.encode("latin-1")
return f"<{bytes(self.encrypt_bytes(string.encode('latin-1'), obj_id)).hex().upper()}>"
except UnicodeEncodeError:
return f'<{hexlify(bytearray(self.encrypt_bytes(BOM_UTF16_BE + string.encode("utf-16-be"), obj_id))).decode("latin-1")}>'
def encrypt_stream(self, stream: bytes, obj_id: int) -> bytes:
if self.encryption_method == EncryptionMethod.NO_ENCRYPTION:
return stream
return bytes(self.encrypt_bytes(stream, obj_id))
def is_aes_algorithm(self) -> bool:
return self.encryption_method in (
EncryptionMethod.AES_128,
EncryptionMethod.AES_256,
)
def encrypt_bytes(self, data: bytes, obj_id: int):
"""
PDF32000 reference - Algorithm 1: Encryption of data using the RC4 or AES algorithms
Append object ID and generation ID to the key and encrypt the data
Generation ID is fixed as 0. Will need to revisit if the application start changing generation ID
"""
h = hashlib.new("md5", usedforsecurity=False)
h.update(self.k)
h.update(
(obj_id & 0xFFFFFF).to_bytes(3, byteorder="little", signed=False)
) # object id
h.update(
(0 & 0xFFFF).to_bytes(2, byteorder="little", signed=False)
) # generation id
if self.is_aes_algorithm():
h.update(bytes([0x73, 0x41, 0x6C, 0x54])) # add salt (sAlT) for AES
key = h.digest()
if self.is_aes_algorithm():
return self.encrypt_AES_cryptography(key, data)
return ARC4().encrypt(key, data)
def encrypt_AES_cryptography(self, key: bytes, data: bytes) -> bytes:
"""Encrypts an array of bytes using AES algorithms (AES 128 or AES 256)"""
iv = bytearray(self.get_random_bytes(16))
padder = PKCS7(128).padder()
padded_data = padder.update(data)
padded_data += padder.finalize()
cipher = (
Cipher(AES128(key), modes.CBC(iv))
if self.encryption_method == EncryptionMethod.AES_128
else Cipher(AES256(self.k), modes.CBC(iv))
)
encryptor = cipher.encryptor()
data = encryptor.update(padded_data) + encryptor.finalize()
iv.extend(data)
return iv
@classmethod
def get_random_bytes(cls: Type["StandardSecurityHandler"], size: int) -> bytes:
"""
https://docs.python.org/3/library/os.html#os.urandom
os.urandom will use OS-specific sources to generate random bytes
suitable for cryptographic use
"""
return urandom(size)
@classmethod
def prepare_string(cls: Type["StandardSecurityHandler"], string: str) -> bytes:
"""
PDF2.0 - ISO 32000-2:2020
All passwords for revision 6 shall be based on Unicode. Preprocessing of a user-provided password
consists first of normalizing its representation by applying the "SASLPrep" profile (Internet RFC 4013)
of the "stringprep" algorithm (Internet RFC 3454) to the supplied password using the Normalize and BiDi
options. Next, the password string shall be converted to UTF-8 encoding, and then truncated to the
first 127 bytes if the string is longer than 127 bytes
Python offers a stringprep module with the tables mapped in methods
"""
# Mapping
def char_map(char: str) -> str:
if not char:
return ""
# Commonly mapped to nothing
if stringprep.in_table_b1(char):
return ""
# Map non-ascii space characters to space
if stringprep.in_table_c12(char):
return "\u0020"
return char
if len(string) < 1:
return bytes()
prepared_string = "".join(char_map(c) for c in string)
# Normalization - applies Unicode normalization form KC
prepared_string = unicodedata.ucd_3_2_0.normalize("NFKC", prepared_string)
# Prohibited output - RCF4013 2.3
def is_prohibited(char: str) -> bool:
return (
stringprep.in_table_c12(char) # Non-ASCII space characters
or stringprep.in_table_c21_c22(char) # Control characters
or stringprep.in_table_c3(char) # Private use
or stringprep.in_table_c4(char) # Non-character code points
or stringprep.in_table_c5(char) # Surrogate codes
or stringprep.in_table_c6(char) # Inappropriate for plain text
or stringprep.in_table_c7(
char
) # Inappropriate for canonical representation
or stringprep.in_table_c8(
char
) # Change display properties or are deprecated
or stringprep.in_table_c9(char) # Tagging characters
)
for char in prepared_string:
if is_prohibited(char):
raise FPDFException(
f"The password {string} contains prohibited characters"
)
# Bidirectional characters
def has_character(string: str, fun: Callable) -> bool:
return any(fun(char) for char in string)
if has_character(prepared_string, stringprep.in_table_d1):
# If a string contains any RandALCat character, the string MUST NOT contain any LCat character.
if has_character(prepared_string, stringprep.in_table_d2):
raise FPDFException(
f"The password {string} contains invalid bidirectional characters."
)
# If a string contains any RandALCat character, a RandALCat character MUST be the first character
# of the string, and a RandALCat character MUST be the last character of the string.
if not (
stringprep.in_table_d1(prepared_string[0])
and stringprep.in_table_d1(prepared_string[-1])
):
raise FPDFException(
f"The password {string} contains invalid bidirectional characters."
)
if len(prepared_string) > 127:
prepared_string = prepared_string[:127]
return prepared_string.encode("UTF-8")
def padded_password(self, password: str) -> bytearray:
"""
PDF32000 reference - Algorithm 2: Computing an encryption key
Step (a) - Add the default padding at the end of provided password to make it 32 bit long
"""
if len(password) > 32:
password = password[:32]
p = bytearray(password.encode("latin1"))
p.extend(self.DEFAULT_PADDING[: (32 - len(p))])
return p
def generate_owner_password(self) -> str:
"""
PDF32000 reference - Algorithm 3: Computing the encryption dictionary's O (owner password) value
The security handler is only using revision 3 or 4, so the legacy r2 version is not implemented here
"""
m = self.padded_password(self.owner_password)
for _ in range(51):
m = md5(m)
rc4key = m[: (math.ceil(self.key_length / 8))]
result = self.padded_password(self.user_password)
for i in range(20):
new_key = []
for k in rc4key:
new_key.append(k ^ i)
result = ARC4().encrypt(bytes(new_key), result)
return bytes(result).hex()
def generate_user_password(self) -> str:
"""
PDF32000 reference - Algorithm 5: Computing the encryption dictionary's U (user password) value
The security handler is only using revision 3 or 4, so the legacy r2 version is not implemented here
"""
m = hashlib.new("md5", usedforsecurity=False)
m.update(bytearray(self.DEFAULT_PADDING))
m.update(bytes.fromhex(self.info_id))
result = bytearray(m.digest())
key = self.k
for i in range(20):
new_key = []
for k in key:
new_key.append(k ^ i)
result = ARC4().encrypt(bytes(new_key), result)
result.extend(
(result[x] ^ self.DEFAULT_PADDING[x]) for x in range(16)
) # add 16 bytes of random padding
return bytes(result).hex()
@classmethod
def compute_hash(
cls: Type["StandardSecurityHandler"],
input_password: bytes,
salt: bytes,
user_key: bytes = bytearray(),
) -> bytes:
"""
Algorithm 2B - section 7.6.4.3.4 of the ISO 32000-2:2020
Applied on Security handlers revision 6
"""
k = hashlib.sha256(input_password + salt + user_key).digest()
round_number = 0
while True:
round_number += 1
k1 = input_password + k + user_key
# Step (a + b)
cipher = Cipher(AES128(k[:16]), modes.CBC(k[16:32]))
encryptor = cipher.encryptor()
e = encryptor.update(k1 * 64) + encryptor.finalize()
# Step (c)
# remainder = int.from_bytes(e[:16], byteorder="big") % 3
remainder = sum(e[:16]) % 3
# Step (d)
if remainder == 0:
k = hashlib.sha256(e).digest()
elif remainder == 1:
k = hashlib.sha384(e).digest()
else:
k = hashlib.sha512(e).digest()
# Step (e)
if round_number >= 64 and e[-1] <= round_number - 32:
break
return k[:32]
def generate_user_password_rev6(self) -> None:
"""
Generating the U (user password) and UE (user encryption)
for security handlers of revision 6
Algorithm 8 - Section 7.6.4.4.7 of the ISO 32000-2:2020
"""
user_password = self.prepare_string(self.user_password)
if not user_password:
user_password = bytearray()
user_validation_salt = self.get_random_bytes(8)
user_key_salt = self.get_random_bytes(8)
u = (
self.compute_hash(input_password=user_password, salt=user_validation_salt)
+ user_validation_salt
+ user_key_salt
)
self.u = u.hex()
key = self.compute_hash(input_password=user_password, salt=user_key_salt)
cipher = Cipher(AES256(key), modes.CBC(b"\x00" * 16))
encryptor = cipher.encryptor()
ue = encryptor.update(self.k) + encryptor.finalize()
self.ue = ue.hex()
def generate_owner_password_rev6(self) -> None:
"""
Generating the O (owner password) and OE (owner encryption)
for security handlers of revision 6
Algorithm 9 - Section 7.6.4.4.8 of the ISO 32000-2:2020
"""
owner_password = self.prepare_string(self.owner_password)
if not owner_password:
raise FPDFException(f"Invalid owner password {self.owner_password}")
owner_validation_salt = self.get_random_bytes(8)
owner_key_salt = self.get_random_bytes(8)
o = (
self.compute_hash(
input_password=owner_password,
salt=owner_validation_salt,
user_key=bytes.fromhex(self.u),
)
+ owner_validation_salt
+ owner_key_salt
)
self.o = o.hex()
key = self.compute_hash(
input_password=owner_password,
salt=owner_key_salt,
user_key=bytes.fromhex(self.u),
)
cipher = Cipher(AES256(key), modes.CBC(b"\x00" * 16))
encryptor = cipher.encryptor()
oe = encryptor.update(self.k) + encryptor.finalize()
self.oe = oe.hex()
def generate_perms_rev6(self) -> None:
"""
7.6.4.4.9 Algorithm 10: Computing the encryption dictionary’s Perms (permissions) value
(Security handlers of revision 6) of the ISO 32000-2:2020
"""
perms64b = 0xFFFFFFFF00000000 | self.access_permission
encrypt_metadata = b"T" if self.encrypt_metadata else b"F"
perms_input = (
perms64b.to_bytes(8, byteorder="little", signed=False)
+ encrypt_metadata
+ b"adb"
+ self.get_random_bytes(4)
)
# nosemgrep: python.cryptography.security.insecure-cipher-mode-ecb.insecure-cipher-mode-ecb
cipher = Cipher(AES256(self.k), modes.ECB())
encryptor = cipher.encryptor()
perms = encryptor.update(perms_input) + encryptor.finalize()
self.perms = perms.hex()
def generate_encryption_key(self) -> bytes:
"""
PDF32000 reference
Algorithm 2: Computing an encryption key
"""
m = hashlib.new("md5", usedforsecurity=False)
m.update(self.padded_password(self.user_password))
m.update(bytes.fromhex(self.o))
m.update(
(self.access_permission & 0xFFFFFFFF).to_bytes(
4, byteorder="little", signed=False
)
)
m.update(bytes.fromhex(self.info_id))
if self.encrypt_metadata is False and self.version == 4:
m.update(bytes([0xFF, 0xFF, 0xFF, 0xFF]))
result = m.digest()[: (math.ceil(self.key_length / 8))]
for _ in range(50):
result = md5(result)[: (math.ceil(self.key_length / 8))]
return result
def md5(data: Union[bytes, bytearray]) -> bytes:
h = hashlib.new("md5", usedforsecurity=False)
h.update(data)
return h.digest()
def int32(n: int) -> int:
"""convert long to signed 32 bit integer"""
n = n & 0xFFFFFFFF
return (n ^ 0x80000000) - 0x80000000