import io import logging import shutil import tempfile from collections import namedtuple from struct import pack, unpack import olefile from msoffcrypto import exceptions from msoffcrypto.format import base from msoffcrypto.format.common import _parse_header_RC4CryptoAPI from msoffcrypto.method.rc4_cryptoapi import DocumentRC4CryptoAPI logger = logging.getLogger(__name__) logger.addHandler(logging.NullHandler()) RecordHeader = namedtuple( "RecordHeader", [ "recVer", "recInstance", "recType", "recLen", ], ) def _parseRecordHeader(blob): # RecordHeader: https://docs.microsoft.com/en-us/openspecs/office_file_formats/ms-ppt/df201194-0cd0-4dfb-bf10-eea353d8eabc getBitSlice = lambda bits, i, w: (bits & (2**w - 1 << i)) >> i blob.seek(0) (buf,) = unpack("> i (buf,) = unpack(" 0: persistdirectoryatom = persistdirectoryatom_stack.pop() for entry in persistdirectoryatom.rgPersistDirEntry: # logger.debug("persistId: %d" % entry.persistId) for i, offset in enumerate(entry.rgPersistOffset): persistobjectdirectory[entry.persistId + i] = offset return persistobjectdirectory class Ppt97File(base.BaseOfficeFile): """Return a MS-PPT file object. Examples: >>> with open("tests/inputs/rc4cryptoapi_password.ppt", "rb") as f: ... officefile = Ppt97File(f) ... officefile.load_key(password="Password1234_") >>> with open("tests/inputs/rc4cryptoapi_password.ppt", "rb") as f: ... officefile = Ppt97File(f) ... officefile.load_key(password="0000") Traceback (most recent call last): ... msoffcrypto.exceptions.InvalidKeyError: ... """ def __init__(self, file): self.file = file ole = olefile.OleFileIO(file) # do not close this, would close file self.ole = ole self.format = "ppt97" self.keyTypes = ["password"] self.key = None self.salt = None # streams closed in destructor: currentuser = ole.openstream("Current User") powerpointdocument = ole.openstream("PowerPoint Document") Data = namedtuple("Data", ["currentuser", "powerpointdocument"]) self.data = Data( currentuser=currentuser, powerpointdocument=powerpointdocument, ) def __del__(self): """Destructor, closes opened streams.""" if hasattr(self, "data") and self.data: if self.data.currentuser: self.data.currentuser.close() if self.data.powerpointdocument: self.data.powerpointdocument.close() def load_key(self, password=None): persistobjectdirectory = construct_persistobjectdirectory(self.data) logger.debug("[*] persistobjectdirectory: {}".format(persistobjectdirectory)) self.data.currentuser.seek(0) currentuser = _parseCurrentUser(self.data.currentuser) logger.debug("[*] currentuser: {}".format(currentuser)) self.data.powerpointdocument.seek( currentuser.currentuseratom.offsetToCurrentEdit ) usereditatom = _parseUserEditAtom(self.data.powerpointdocument) logger.debug("[*] usereditatom: {}".format(usereditatom)) # cf. Part 2 in https://docs.microsoft.com/en-us/openspecs/office_file_formats/ms-ppt/1fc22d56-28f9-4818-bd45-67c2bf721ccf cryptsession10container_offset = persistobjectdirectory[ usereditatom.encryptSessionPersistIdRef ] logger.debug( "[*] cryptsession10container_offset: {}".format( cryptsession10container_offset ) ) self.data.powerpointdocument.seek(cryptsession10container_offset) cryptsession10container = _parseCryptSession10Container( self.data.powerpointdocument ) logger.debug("[*] cryptsession10container: {}".format(cryptsession10container)) encryptionInfo = io.BytesIO(cryptsession10container.data) encryptionVersionInfo = encryptionInfo.read(4) vMajor, vMinor = unpack(" be an encrypted document. headerToken=0xE391C05F, offsetToCurrentEdit=cuatom.offsetToCurrentEdit, lenUserName=cuatom.lenUserName, docFileVersion=cuatom.docFileVersion, majorVersion=cuatom.majorVersion, minorVersion=cuatom.minorVersion, unused=cuatom.unused, ansiUserName=cuatom.ansiUserName, relVersion=cuatom.relVersion, unicodeUserName=cuatom.unicodeUserName, ) ) buf = _packCurrentUser(currentuser_new) buf.seek(0) currentuser_buf = buf # List of encrypted parts: https://docs.microsoft.com/en-us/openspecs/office_file_formats/ms-ppt/b0963334-4408-4621-879a-ef9c54551fd8 # PowerPoint Document Stream self.data.powerpointdocument.seek(0) powerpointdocument_size = len(self.data.powerpointdocument.read()) logger.debug("[*] powerpointdocument_size: {}".format(powerpointdocument_size)) self.data.powerpointdocument.seek(0) dec_bytearray = bytearray(self.data.powerpointdocument.read()) # UserEditAtom self.data.powerpointdocument.seek( currentuser.currentuseratom.offsetToCurrentEdit ) # currentuseratom_raw = self.data.powerpointdocument.read(40) self.data.powerpointdocument.seek( currentuser.currentuseratom.offsetToCurrentEdit ) usereditatom = _parseUserEditAtom(self.data.powerpointdocument) # logger.debug(usereditatom) # logger.debug(["offsetToCurrentEdit", currentuser.currentuseratom.offsetToCurrentEdit]) rh_new = RecordHeader( recVer=usereditatom.rh.recVer, recInstance=usereditatom.rh.recInstance, recType=usereditatom.rh.recType, recLen=usereditatom.rh.recLen - 4, # Omit encryptSessionPersistIdRef field ) # logger.debug([_packRecordHeader(usereditatom.rh).read(), _packRecordHeader(rh_new).read()]) usereditatom_new = UserEditAtom( rh=rh_new, lastSlideIdRef=usereditatom.lastSlideIdRef, version=usereditatom.version, minorVersion=usereditatom.minorVersion, majorVersion=usereditatom.majorVersion, offsetLastEdit=usereditatom.offsetLastEdit, offsetPersistDirectory=usereditatom.offsetPersistDirectory, docPersistIdRef=usereditatom.docPersistIdRef, persistIdSeed=usereditatom.persistIdSeed, lastView=usereditatom.lastView, unused=usereditatom.unused, encryptSessionPersistIdRef=0x00000000, # Clear ) # logger.debug(currentuseratom_raw) # logger.debug(_packUserEditAtom(usereditatom).read()) # logger.debug(_packUserEditAtom(usereditatom_new).read()) buf = _packUserEditAtom(usereditatom_new) buf.seek(0) buf_bytes = bytearray(buf.read()) offset = currentuser.currentuseratom.offsetToCurrentEdit dec_bytearray[offset : offset + len(buf_bytes)] = buf_bytes # PersistDirectoryAtom self.data.powerpointdocument.seek( currentuser.currentuseratom.offsetToCurrentEdit ) usereditatom = _parseUserEditAtom(self.data.powerpointdocument) # logger.debug(usereditatom) self.data.powerpointdocument.seek(usereditatom.offsetPersistDirectory) persistdirectoryatom = _parsePersistDirectoryAtom(self.data.powerpointdocument) # logger.debug(persistdirectoryatom) persistdirectoryatom_new = PersistDirectoryAtom( rh=persistdirectoryatom.rh, rgPersistDirEntry=[ PersistDirectoryEntry( persistId=persistdirectoryatom.rgPersistDirEntry[0].persistId, # Omit CryptSession10Container cPersist=persistdirectoryatom.rgPersistDirEntry[0].cPersist - 1, rgPersistOffset=persistdirectoryatom.rgPersistDirEntry[ 0 ].rgPersistOffset, ), ], ) self.data.powerpointdocument.seek(usereditatom.offsetPersistDirectory) buf = _packPersistDirectoryAtom(persistdirectoryatom_new) buf_bytes = bytearray(buf.read()) offset = usereditatom.offsetPersistDirectory dec_bytearray[offset : offset + len(buf_bytes)] = buf_bytes # Persist Objects self.data.powerpointdocument.seek(0) persistobjectdirectory = construct_persistobjectdirectory(self.data) directory_items = list(persistobjectdirectory.items()) for i, (persistId, offset) in enumerate(directory_items): self.data.powerpointdocument.seek(offset) buf = self.data.powerpointdocument.read(8) rh = _parseRecordHeader(io.BytesIO(buf)) logger.debug("[*] rh: {}".format(rh)) # CryptSession10Container if rh.recType == 0x2F14: logger.debug("[*] CryptSession10Container found") # Remove encryption, pad by zero to preserve stream size dec_bytearray[offset : offset + (8 + rh.recLen)] = b"\x00" * ( 8 + rh.recLen ) continue # The UserEditAtom record (section 2.3.3) and the PersistDirectoryAtom record (section 2.3.4) MUST NOT be encrypted. if rh.recType in [0x0FF5, 0x1772]: logger.debug("[*] UserEditAtom/PersistDirectoryAtom found") continue # TODO: Fix here recLen = directory_items[i + 1][1] - offset - 8 logger.debug("[*] recLen: {}".format(recLen)) self.data.powerpointdocument.seek(offset) enc_buf = io.BytesIO(self.data.powerpointdocument.read(8 + recLen)) blocksize = self.keySize * ( (8 + recLen) // self.keySize + 1 ) # Undocumented dec = DocumentRC4CryptoAPI.decrypt( self.key, self.salt, self.keySize, enc_buf, blocksize=blocksize, block=persistId, ) dec_bytes = bytearray(dec.read()) dec_bytearray[offset : offset + len(dec_bytes)] = dec_bytes # To BytesIO dec_buf = io.BytesIO(dec_bytearray) dec_buf.seek(0) for i, (persistId, offset) in enumerate(directory_items): dec_buf.seek(offset) buf = dec_buf.read(8) rh = _parseRecordHeader(io.BytesIO(buf)) logger.debug("[*] rh: {}".format(rh)) dec_buf.seek(0) logger.debug( "[*] powerpointdocument_size={}, len(dec_buf.read())={}".format( powerpointdocument_size, len(dec_buf.read()) ) ) dec_buf.seek(0) powerpointdocument_dec_buf = dec_buf # TODO: Pictures Stream # TODO: Encrypted Summary Info Stream with tempfile.TemporaryFile() as _outfile: self.file.seek(0) shutil.copyfileobj(self.file, _outfile) outole = olefile.OleFileIO(_outfile, write_mode=True) outole.write_stream("Current User", currentuser_buf.read()) outole.write_stream( "PowerPoint Document", powerpointdocument_dec_buf.read() ) # Finalize _outfile.seek(0) shutil.copyfileobj(_outfile, outfile) return def is_encrypted(self): r""" Test if the file is encrypted. >>> f = open("tests/inputs/plain.ppt", "rb") >>> file = Ppt97File(f) >>> file.is_encrypted() False >>> f = open("tests/inputs/rc4cryptoapi_password.ppt", "rb") >>> file = Ppt97File(f) >>> file.is_encrypted() True """ self.data.currentuser.seek(0) currentuser = _parseCurrentUser(self.data.currentuser) logger.debug("[*] currentuser: {}".format(currentuser)) self.data.powerpointdocument.seek( currentuser.currentuseratom.offsetToCurrentEdit ) usereditatom = _parseUserEditAtom(self.data.powerpointdocument) logger.debug("[*] usereditatom: {}".format(usereditatom)) if usereditatom.rh.recLen == 0x00000020: # Cf. _parseUserEditAtom return True else: return False