a script for applying MS patch deltas

delta_patch.py

import base64
import hashlib
import zlib
from ctypes import (
    CDLL,
    POINTER,
    LittleEndianStructure,
    c_size_t,
    c_ubyte,
    c_uint64,
    cast,
    windll,
    wintypes,
)
from ctypes import (
    Union as CUnion,
)
from pathlib import Path
from typing import List, Optional, Union

# types and flags
DELTA_FLAG_TYPE = c_uint64
DELTA_FLAG_NONE = 0x00000000
DELTA_APPLY_FLAG_ALLOW_PA19 = 0x00000001

# magic values
MAGIC_PA19 = b"PA19"
MAGIC_PA30 = b"PA30"
MAGIC_PA31 = b"PA31"


class DeltaPatchException(Exception):
    pass


# structures
class DELTA_INPUT(LittleEndianStructure):
    class U1(CUnion):
        _fields_ = [("lpcStart", wintypes.LPVOID), ("lpStart", wintypes.LPVOID)]

    _anonymous_ = ("u1",)
    _fields_ = [("u1", U1), ("uSize", c_size_t), ("Editable", wintypes.BOOL)]


class DELTA_OUTPUT(LittleEndianStructure):
    _fields_ = [("lpStart", wintypes.LPVOID), ("uSize", c_size_t)]


class DeltaPatcher(object):
    class DeltaFuncs:
        def __init__(self, msdelta: CDLL):
            def _apply_delta_errcheck(res: wintypes.BOOL, func, args):
                if not res:
                    last_err = windll.kernel32.GetLastError()
                    raise DeltaPatchException(
                        f"ApplyDeltaB failed with GLE = {last_err}"
                    )
                output: DELTA_OUTPUT = args[3]
                # cast the void pointer output to a correctly sized byte array pointer
                # then get the contents and initialize a bytes object
                # this should copy the bytes
                patchbuf = bytes(
                    cast(output.lpStart, POINTER(c_ubyte * output.uSize)).contents
                )
                self.DeltaFree(output.lpStart)
                return patchbuf

            self.ApplyDeltaB = msdelta.ApplyDeltaB
            self.DeltaFree = msdelta.DeltaFree
            self.ApplyDeltaB.argtypes = [
                DELTA_FLAG_TYPE,
                DELTA_INPUT,
                DELTA_INPUT,
                POINTER(DELTA_OUTPUT),
            ]
            self.ApplyDeltaB.restype = wintypes.BOOL
            self.ApplyDeltaB.errcheck = _apply_delta_errcheck
            self.DeltaFree.argtypes = [wintypes.LPVOID]
            self.DeltaFree.restype = wintypes.BOOL

    def __init__(
        self,
        patcher_dll_path: Union[Path, str] = "msdelta.dll",
        buffer: Optional[Union[str, bytes, Path]] = None,
        allow_legacy=False,
    ):
        self.__msdelta = CDLL(patcher_dll_path)
        self.funcs = DeltaPatcher.DeltaFuncs(self.__msdelta)
        self.flags = DELTA_APPLY_FLAG_ALLOW_PA19 if allow_legacy else DELTA_FLAG_NONE
        if isinstance(buffer, (str, Path)):
            buffer = open(buffer, "rb").read()
        self.buffer = buffer or b""

    def apply_all(self, patches: List[Union[str, bytes, Path]]):
        for patch in patches:
            self.apply_delta(patch)

    def apply_delta(self, patch: Union[str, bytes, Path]):
        if isinstance(patch, (str, Path)):
            patch = open(patch, "rb").read()
        # check for the CRC, strip if required
        magics = [MAGIC_PA19, MAGIC_PA30, MAGIC_PA31]
        if (
            zlib.crc32(patch[4:]) == int.from_bytes(patch[:4], "little")
            and patch[4:8] in magics
        ):
            patch = patch[4:]
        if patch[:4] not in magics:
            raise DeltaPatchException(
                "Invalid patch file. Starts with {} instead of acceptable magic values",
                patch[:4].hex(),
            )
        buffer_input = DELTA_INPUT(
            DELTA_INPUT.U1(lpcStart=cast(self.buffer, wintypes.LPVOID)),
            len(self.buffer),
            False,
        )
        patch_input = DELTA_INPUT(
            DELTA_INPUT.U1(lpcStart=cast(patch, wintypes.LPVOID)), len(patch), False
        )
        output = DELTA_OUTPUT()
        self.buffer = self.funcs.ApplyDeltaB(
            self.flags, buffer_input, patch_input, output
        )

    def checksum(self) -> str:
        return base64.b64encode(hashlib.sha256(self.buffer).digest()).decode()

    def __bytes__(self) -> bytes:
        return self.buffer


if __name__ == "__main__":
    import argparse
    import sys

    ap = argparse.ArgumentParser()
    mode = ap.add_mutually_exclusive_group(required=True)
    output = ap.add_mutually_exclusive_group(required=True)
    mode.add_argument(
        "-i", "--input-file", type=Path, help="File to patch (forward or reverse)"
    )
    mode.add_argument(
        "-n",
        "--null",
        action="store_true",
        default=False,
        help="Create the output file from a null diff "
        "(null diff must be the first one specified)",
    )
    output.add_argument(
        "-o", "--output-file", type=Path, help="Destination to write patched file to"
    )
    output.add_argument(
        "-d",
        "--dry-run",
        action="store_true",
        help="Don't write patch, just see if it would patch"
        "correctly and get the resulting hash",
    )
    ap.add_argument(
        "-l",
        "--legacy",
        action="store_true",
        default=False,
        help="Let the API use the PA19 legacy API (if required)",
    )
    ap.add_argument(
        "-D",
        "--patcher-dll",
        type=Path,
        default="msdelta.dll",
        help="DLL to load and use for patch delta API",
    )
    ap.add_argument("patches", nargs="+", type=Path, help="Patches to apply")
    args = ap.parse_args()

    if not args.dry_run and not args.output_file:
        print("Either specify -d or -o", file=sys.stderr)
        ap.print_help()
        sys.exit(1)

    patcher = DeltaPatcher(args.patcher_dll, args.input_file, args.legacy)
    patcher.apply_all(args.patches)

    print(
        "Applied {} patch{} successfully".format(
            len(args.patches), "es" if len(args.patches) > 1 else ""
        )
    )
    print("Final hash: {}".format(patcher.checksum()))

    if not args.dry_run:
        open(args.output_file, "wb").write(bytes(patcher))
        print(f"Wrote {len(bytes(patcher))} bytes to {args.output_file.resolve()}")

Thanks for sharing the code. I added it to Winbindex to unpack null differential files, which allows me to index extra data for these files.

I stumbled upon a bug, though. If the CRC happens to contain the "PA" bytes, the script breaks. For example (that's a real file from an update package):

I fixed it by assuming the CRC is always there, which works for my use case. For a more general fix, you might want to calculate the CRC checksum before checking the header, and remove it only if it matches.

@m417z woah thanks. Love your project and glad I could contribute, if only accidentally :)
Good catch on that bug, I'll fix it. I use gist too much, I should have this in a repo so it's easier for people to help fix stuff with PRs.
Cheers.

GE_RELEASE just added support for PA31 deltas, any chance on updating that script when possible?

For PA31 support you have to use a recent version of UpdateCompression.dll instead of msdelta.dll. You can download it here, place it in the script's folder, and apply this change:

diff --git a/delta_patch.py b/delta_patch.py
index fe04da2..69c9049 100644
--- a/delta_patch.py
+++ b/delta_patch.py
@@ -1,5 +1,5 @@
 from ctypes import (windll, wintypes, c_uint64, cast, POINTER, Union, c_ubyte,
-                    LittleEndianStructure, byref, c_size_t)
+                    LittleEndianStructure, byref, c_size_t, CDLL)
 import zlib
 
 
@@ -26,11 +26,12 @@ class DELTA_OUTPUT(LittleEndianStructure):
 
 
 # functions
-ApplyDeltaB = windll.msdelta.ApplyDeltaB
+msdelta = CDLL('UpdateCompression.dll')
+ApplyDeltaB = msdelta.ApplyDeltaB
 ApplyDeltaB.argtypes = [DELTA_FLAG_TYPE, DELTA_INPUT, DELTA_INPUT,
                         POINTER(DELTA_OUTPUT)]
 ApplyDeltaB.rettype = wintypes.BOOL
-DeltaFree = windll.msdelta.DeltaFree
+DeltaFree = msdelta.DeltaFree
 DeltaFree.argtypes = [wintypes.LPVOID]
 DeltaFree.rettype = wintypes.BOOL
 gle = windll.kernel32.GetLastError

Thanks for the update @m417z
I'm going to push a more considerable update soon that allows you to specify the DLL to use.

Do you happen to know why there is only a forward differential on Windows11 patches ? The script fails with error 13 when I try to use to forward delta (same versions). I extracted the files with PsfExtractor
The forward diff is described as raw instead of PA30 in the .manifest :

    <File id="25666" name="amd64_microsoft-windows-win32k_31bf3856ad364e35_10.0.22621.3085_none_990c09ec15f6744a\f\win32kfull.sys" length="323070" time="133519934640000000" attr="128">
      <Hash alg="SHA256" value="104C7682E1A4857C5E3BE0369680717A1F6DFDFFBA1FDDF5A5BBE25E617189FA" />
      <Delta>
        <**Source type="RAW"** offset="257107653" length="323070">
          <Hash alg="SHA256" value="104C7682E1A4857C5E3BE0369680717A1F6DFDFFBA1FDDF5A5BBE25E617189FA" />
        </Source>
      </Delta>

My best guess is that the reverse diffs are now generated when the patch is applied by using the API on the fly to diff the .1 and new version. I noticed this as well while ripping apart the new patch format.

GE_RELEASE just added support for PA31 deltas, any chance on updating that script when possible?

Yea, it's updated now. Provide your own UpdateCompression.dll and use the -D option to the script to load it.