from base64 import b64decode
from itertools import islice
from random import randbytes
from zlib import compress

import pandas as pd
import matplotlib.pyplot as plt

# From pyca/cryptography
from cryptography.hazmat.primitives import padding
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
import cryptography.hazmat.decrepit.ciphers.algorithms as decrepit_algorithms

def aes_128_ctr_crypt(text, key, nonce):
    # Symmetric encryption/decryption
    encryptor = Cipher(algorithms.AES128(key), modes.CTR(nonce)).encryptor()
    return encryptor.update(text) + encryptor.finalize()

def pad_pkcs7(text):
    padder = padding.PKCS7(128).padder()
    return padder.update(text) + padder.finalize()

def aes_128_cbc_encrypt(ptext, key, iv, padding=True):
    if padding:
        ptext = pad_pkcs7(ptext)
    encryptor = Cipher(algorithms.AES128(key), modes.CBC(iv)).encryptor()
    return encryptor.update(ptext) + encryptor.finalize()

def rc4_encrypt(ptext, key):
    encryptor = Cipher(decrepit_algorithms.ARC4(key), mode=None).encryptor()
    return encryptor.update(ptext) + encryptor.finalize()

def xor(x, y):
    return bytes(xb^yb for xb, yb in zip(x, y))

zero = bytes(16)

key = randbytes(16)

def cbc_mac(text, iv):
    return aes_128_cbc_encrypt(text, key, iv)[-16:]

def authenticate(text, iv, mac):
    return cbc_mac(text, iv) == mac

#                  |..............|  <-- first block
message =        b"from=Bob&to=Dave&amount=5"
forged_message = b"from=Bob&to=Greg&amount=5"

# The message and the following are intercepted...
iv = randbytes(16)
mac = cbc_mac(message, iv)

forged_iv = xor(xor(iv, message[:16]), forged_message[:16])

# And the server will accept...
authenticate(forged_message, forged_iv, mac)

True

message = b"from=Bob&tx_list=Dave:5"
mac = cbc_mac(message, iv=zero)

extension = b";Greg:1000000"

# The scenario calls for us to compute the CBC-MAC of the extension in
# which the first block has been XOR-ed with `mac` above, and using
# IV=0, but for brevity we note that that is mathematically equivalent
# to this call.
new_mac = cbc_mac(extension, iv=mac)

forged_message = pad_pkcs7(message) + extension

authenticate(forged_message, iv=zero, mac=new_mac)

True

print(forged_message)

b'from=Bob&tx_list=Dave:5\t\t\t\t\t\t\t\t\t;Greg:1000000'

def cbc_mac_hash(text):
    return aes_128_cbc_encrypt(text, b"YELLOW SUBMARINE", zero)[-16:]

original_code = b"alert('MZA who was that?');\n"
print(cbc_mac_hash(original_code).hex())

forged_code = b"alert('Ayo, the Wu is back!');\n" + b"//"
C_f = cbc_mac_hash(forged_code)
forged_code = (
    pad_pkcs7(forged_code) +
    xor(C_f, original_code[:16]) +
    original_code[16:]
)

print(cbc_mac_hash(forged_code).hex())

296b8d7cb78a243dda4d0a61d33bbdd1
296b8d7cb78a243dda4d0a61d33bbdd1

print(forged_code)

b'alert(\'Ayo, the Wu is back!\');\n//\x0f\x0f\x0f\x0f\x0f\x0f\x0f\x0f\x0f\x0f\x0f\x0f\x0f\x0f\x0f\xe3l\xa7\xc1BR\x0c\xa8Dy"8tA\xaa~as that?\');\n'

session_id = b"TmV2ZXIgcmV2ZWFsIHRoZSBXdS1UYW5nIFNlY3JldCE="

base64_alphabet = (
    b"ABCDEFGHIJKLMNOPQRSTUVWXYZ"
    b"abcdefghijklmnopqrstuvwxyz"
    b"0123456789/+="
)

def format_request(data):
    header = (
        "POST / HTTP/1.1\r\n"
        "Host: hapless.com\r\n"
        "Cookie: sessionid={session_id}\r\n"
        "Content-Length: {length}\r\n"
        "\r\n"
    )
    return header.format(
        session_id=session_id.decode(),
        length=len(data)
    ).encode() + data

def oracle(data):
    key = randbytes(16)
    nonce = randbytes(16)
    return len(aes_128_ctr_crypt(compress(format_request(data)), key, nonce))

def recover_session_id(add_padding=False):
    id = bytes()
    while not id.endswith(b"\r\n"):
        padding = b""
        while True:
            l = [
                [oracle(padding + b"Cookie: sessionid=" + id + bytes([b])), b]
                for b in base64_alphabet + b"\r\n"
            ]
            l.sort(key=lambda t: t[0])
            if l[0][0] < l[1][0]:
                # There is a unique shortest compressed output
                id += bytes([l[0][1]])
                break
            else:
                assert add_padding
                padding += randbytes(1)
    return id[:-2]

print(recover_session_id() == session_id)

True

def oracle(data):
    key = randbytes(16)
    iv = randbytes(16)
    return len(aes_128_cbc_encrypt(compress(format_request(data)), key, iv))

print(recover_session_id(add_padding=True) == session_id)

True

b = 16  # hash size in bits
B = b//8  # and bytes
key = b"YELLOW SUBMARINE"

def MD(message, hash, encryption_fn):
    # Assumes message is block-aligned.
    for i in range(0, len(message), 16):
        hash = encryption_fn(message[i:i+16], hash+bytes(16-B))[:B]
    return hash

def f(message, hash):
    return MD(
        message,
        hash,
        lambda b, h: aes_128_cbc_encrypt(b, key, h, padding=False)
    )

def g(message, hash):
    return MD(
        message,
        hash,
        lambda b, h: aes_128_ctr_crypt(b, key, h)
    )

def collisions_gen(message_gen, hash_fn, num_hashes):
    # Drawing messages from `message_gen`, yield collision pairs
    # (p_1, q_1), (p_2, q_2), ... as described above.  `num_hashes[0]`
    # is updated to count the number of hashes required to find the
    # collisions.
    seen = {}
    ih = bytes(B)  # input hash value
    while True:
        try:
            m = next(message_gen)
        except StopIteration:
            return
        h = hash_fn(m, ih)
        num_hashes[0] += 1
        if h in seen:
            yield seen[h], m
            seen.clear()
            ih = h
        else:
            seen[h] = m

def random_gen():
    while True:
        yield randbytes(16)

num_pairs = b//2 + 2  # see discussion below

num_hashes = [0]
pairs = list(islice(collisions_gen(random_gen(), f, num_hashes), num_pairs))

print("Number of hashes required to find one collision...")
print(f"    Estimated number, 50% probability: {2**(b//2)}")
print(f"    Average observed: {num_hashes[0]//num_pairs}")

Number of hashes required to find one collision...
    Estimated number, 50% probability: 256
    Average observed: 334

def combinatorial_gen(pairs):
    for i in range(2**len(pairs)):
        m = b""
        for j in range(len(pairs)):
            m += pairs[j][i&(2**j) != 0]
        yield m

num_hashes = [0]
found = any(collisions_gen(combinatorial_gen(pairs), g, num_hashes))

print(f"Found collision under g: {found}")
print(f"Number of messages hashed: {num_hashes[0]}")

Found collision under g: True
Number of messages hashed: 85

class ExpandableMessage:

    def __init__(self, k, ih):
        self.table = []
        self.num_hashes = 0  # number of hashes required to build the table
        h = ih
        for e in range(k-1, -1, -1):
            short, long, h = self.find_collision(2**e+1, h)
            self.table.append((short, long))
        self.last_hash = h

    def find_collision(self, l, ih):
        # Find a short message (block length 1) and long message
        # (block length l > 1) that collide.
        # Return (short, long, hash).
        long_base = randbytes((l-1)*16)
        long_base_h = f(long_base, ih)
        self.num_hashes += l-1
        seen = {}
        while True:
            short = randbytes(16)
            h = f(short, ih)
            self.num_hashes += 1
            if h in seen:
                if len(seen[h]) > 16:
                    return short, seen[h], h
            else:
                seen[h] = short
            long_tail = randbytes(16)
            h = f(long_tail, long_base_h)
            self.num_hashes += 1
            if h in seen:
                if len(seen[h]) == 16:
                    return seen[h], long_base+long_tail, h
            else:
                seen[h] = long_base+long_tail

    def construct(self, l):
        # Return message of length l.
        k = len(self.table)
        return b"".join(
            self.table[i][(l-k)&(2**(k-1-i)) != 0]
            for i in range(k)
        )

def construct_collision(message, k):
    # Return (colliding message, number of hashes to build expandable
    # message, number of hashes to find bridge block).
    # First, construct an expandable message.
    em = ExpandableMessage(k, zero[:B])
    # Incrementally hash the message.
    inc_hashes = {}
    h = zero[:B]
    num_hashes = 0
    for i in range(len(message)//16):
        h = f(message[i*16:(i+1)*16], h)
        num_hashes += 1
        if i >= k:
            inc_hashes[h] = i
    # Find a bridge block.
    while True:
        bridge = randbytes(16)
        h = f(bridge, em.last_hash)
        num_hashes += 1
        if h in inc_hashes:
            i = inc_hashes[h]
            break
    # Construct and return a colliding message.
    forged = em.construct(i) + bridge + message[(i+1)*16:]
    return forged, em.num_hashes, num_hashes

k = 7
M = randbytes(2**k * 16)
M2, num_hashes_em, num_hashes_bb = construct_collision(M, k)

print(len(M), f(M, zero[:B]).hex())
print(len(M2), f(M2, zero[:B]).hex())
print(M == M2)

2048 dc92
2048 dc92
False

print("Number of hashes required...")
print(f"    Desired minimum effort (2^b): {2**b}")
print(f"    Paper's estimate: {(2**k + k*2**(b//2+1)) + (2**k + 2**(b-k))}")
print(f"    Actual: {num_hashes_em + num_hashes_bb}")

Number of hashes required...
    Desired minimum effort (2^b): 65536
    Paper's estimate: 4352
    Actual: 3030

def run_experiments():
    num_iterations = 100
    df = pd.DataFrame({
        "k": pd.Series(dtype=int),
        "avg_hashes_em": pd.Series(dtype=float),
        "avg_hashes_bb": pd.Series(dtype=float)
    })
    for k in range(4, 9):
        M = randbytes(2**k * 16)
        sum_hashes_em = sum_hashes_bb = 0
        for _ in range(num_iterations):
            _, num_hashes_em, num_hashes_bb = construct_collision(M, k)
            sum_hashes_em += num_hashes_em
            sum_hashes_bb += num_hashes_bb
        df.loc[len(df)] = {
            "k": k,
            "avg_hashes_em": sum_hashes_em/num_iterations,
            "avg_hashes_bb": sum_hashes_bb/num_iterations
        }
    df.set_index("k", inplace=True)
    df["total"] = df.avg_hashes_em + df.avg_hashes_bb
    return df

df = run_experiments()

plt.plot(df.avg_hashes_em)
plt.plot(df.avg_hashes_bb)
plt.plot(df.total, linewidth=3)
plt.xlabel("k")
plt.xticks(range(4, 9))
plt.ylabel("number of hashes")
plt.legend(["build expandable message", "find bridge block", "total"])
None

k = 4
prediction_buffer_len = 6  # in blocks; roughly 100 bytes

class Funnel:

    class Node:
        def __init__(self, block, hash):
            self.block, self.hash = block, hash
            self.parent = None

    def __init__(self, k):
        # Create the leaf level of 2^k random messages with unique hash values.
        self.leaf_hashes = {}
        while len(self.leaf_hashes) < 2**k:
            m = randbytes(16)
            h = f(m, zero[:B])
            if h not in self.leaf_hashes:
                self.leaf_hashes[h] = Funnel.Node(m, h)
        level = list(self.leaf_hashes.values())
        # Now create the remaining levels of the tree up to the root.
        while len(level) > 1:
            next_level = []
            for i in range(0, len(level), 2):
                while True:
                    m = randbytes(16)
                    h1, h2 = f(m, level[i].hash), f(m, level[i+1].hash)
                    if h1 == h2:
                        n = Funnel.Node(m, h1)
                        level[i].parent = level[i+1].parent = n
                        next_level.append(n)
                        break
            level = next_level
        self.final_hash = h1

funnel = Funnel(k)

print(f"Pre-event prediction length: {(prediction_buffer_len+k+1)*16}")
print(f"Pre-event pediction hash: {funnel.final_hash.hex()}")

Pre-event prediction length: 176
Pre-event pediction hash: aacc

def attach_funnel(message, funnel):
    message_hash = f(message, zero[:B])
    while True:
        bridge = randbytes(16)
        h = f(bridge, message_hash)
        if h in funnel.leaf_hashes:
            message += bridge
            # Now walk up the tree to the root node.
            n = funnel.leaf_hashes[h].parent
            while n != None:
                message += n.block
                n = n.parent
            break
    return message

prediction = b"A second lamp in the belfry burns!"
prediction += b" "*(prediction_buffer_len*16-len(prediction))
prediction = attach_funnel(prediction, funnel)

print(f"Post-event prediction length: {len(prediction)}")
print(f"Post-event prediction hash: {f(prediction, zero[:B]).hex()}")

Post-event prediction length: 176
Post-event prediction hash: aacc

num_encryptions = 2**24
byte_15_bias = 240
byte_31_bias = 224

cookie = b64decode("QkUgU1VSRSBUTyBEUklOSyBZT1VSIE9WQUxUSU5F")

def rc4_byte_counts(ptext):
    n = len(ptext)
    a = [[0]*256 for _ in range(n)]
    for _ in range(num_encryptions):
        key = randbytes(16)
        ctext = rc4_encrypt(ptext, key)
        for i in range(n):
            a[i][ctext[i]] += 1
    return a

df = pd.DataFrame(
    { f"byte_{i}": a for i, a in enumerate(rc4_byte_counts(cookie)) }
)

plt.axhline(y=1/256, color="r", label=None)
plt.plot(df.byte_12/df.byte_12.sum(), label="byte 12")
plt.plot(df.byte_15/df.byte_15.sum(), label="byte 15")
plt.xlabel("byte values (0-255)")
plt.ylabel("frequency")
plt.xticks([])
plt.legend()
None

print(chr(cookie[15]))
print(chr(df.byte_15.argmax() ^ byte_15_bias))

K
K

recovered = bytearray(32)

assert len(cookie) == 30
prefix = b"AA"  # add two starting bytes for an even 32

for i in range(16):
    a = rc4_byte_counts(prefix + cookie)
    recovered[15-i] = a[15].index(max(a[15])) ^ byte_15_bias
    recovered[31-i] = a[31].index(max(a[31])) ^ byte_31_bias
    prefix += b"A"

print(bytes(recovered)[2:])

b'BE SURE TO DRINK YOUR OVALTINE'

bit position	short length	long length	input hash value	common output hash value
$k-1$	$1$	$1+2^{k-1}$	$h_0$	$h_1$
$k-2$	$1$	$1+2^{k-2}$	$h_1$	$h_2$
...	...	...	...	...
$1$	$1$	$1+2^1$	$h_{k-2}$	$h_{k-1}$
$0$	$1$	$1+2^0$	$h_{k-1}$	$h_k$

Set 7: Hashes¶

Preliminaries¶

Challenge 49: CBC-MAC message forgery¶

Challenge 50: Hashing with CBC-MAC¶

Challenge 51: Compression ratio side-channel attacks¶

Challenge 52: Iterated hash function multicollisions¶

Challenge 53: Kelsey and Schneier's expandable messages¶

Challenge 54: Kelsey and Kohno's Nostradamus attack¶

Challenge 56: RC4 single-byte biases¶