Google CTF 2023 mytls

client.py

# Google CTF 2023
# mytls challenge: https://capturetheflag.withgoogle.com/challenges/crypto-mytls

import binascii
from cryptography import x509
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives import hmac
from cryptography.hazmat.primitives.asymmetric import ec
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
from cryptography.hazmat.primitives.kdf.hkdf import HKDF
from secrets import token_hex
from pwn import *


def encrypt_message(message, iv, key):
    cipher = Cipher(
        algorithms.AES(key),
        modes.CBC(binascii.unhexlify(iv)))
    encryptor = cipher.encryptor()
    payload = encryptor.update(
        message + b'\x00' * (16 - len(message) % 16)) + encryptor.finalize()
    return binascii.hexlify(payload)


def decrypt_message(message, iv, key):
    cipher = Cipher(
        algorithms.AES(key),
        modes.CBC(binascii.unhexlify(iv)))
    decryptor = cipher.decryptor()
    message = binascii.unhexlify(message)
    res = decryptor.update(message)
    return res.strip(b'\x00')


hash_priv_key = ""


def brute_force(pad_length: int, known_bytes: bytes):
    with open("guest-ecdhkey.pem", "rb") as f:
        client_key = serialization.load_pem_private_key(
            f.read(), None)

    with open("server-ecdhcert.pem", "rb") as f:
        server_cert = x509.load_pem_x509_certificate(f.read())
    client_cert_bytes = open("guest-ecdhcert.pem", "rb").read()
    client_cert = x509.load_pem_x509_certificate(client_cert_bytes)

    io = remote("mytls.2023.ctfcompetition.com", 1337)

    io.recvuntil(b'Please provide the client certificate in PEM format:')
    io.sendline(client_cert_bytes)

    io.recvuntil(b'Please provide the ephemeral client random:\n')
    client_ephemeral_random = token_hex(16)
    io.sendline(client_ephemeral_random.encode())

    io.recvuntil(b'Please provide the ephemeral client key:\n')
    client_ephemeral_key = ec.generate_private_key(ec.SECP256R1())
    client_ephemeral_public_key = client_ephemeral_key.public_key()
    io.sendline(client_ephemeral_public_key.public_bytes(
        serialization.Encoding.PEM, format=serialization.PublicFormat.SubjectPublicKeyInfo))

    io.recvuntil(b'Server ephemeral random:\n')
    server_ephemeral_random = io.recvline(False)
    io.recvline_contains(b'Server ephemeral key:')
    server_ephemeral_key_bytes = b''.join(io.recvlines(4, True))
    server_ephemeral_public_key = serialization.load_pem_public_key(
        server_ephemeral_key_bytes)

    io.recvlines(2)

    client_ephemeral_secret = client_ephemeral_key.exchange(
        ec.ECDH(), server_ephemeral_public_key)
    client_secret = client_key.exchange(ec.ECDH(), server_cert.public_key())
    derived_key = HKDF(algorithm=hashes.SHA256(),
                       length=32,
                       salt=b'SaltyMcSaltFace',
                       info=b'mytls').derive(
                           client_ephemeral_secret +
                           client_secret +
                           client_ephemeral_random.encode('utf-8') +
                           server_ephemeral_random)
    client_hmac = hmac.HMAC(derived_key, hashes.SHA256())
    client_hmac.update(b'client myTLS successful!')
    io.sendline(binascii.hexlify(client_hmac.finalize()))

    def decrypt(msg): return decrypt_message(
        msg, server_ephemeral_random, derived_key)

    def encrypt(msg): return encrypt_message(
        msg, server_ephemeral_random, derived_key)

    def decrypt_line(): return decrypt(io.recvline(False))

    def print_decrypt(): print(decrypt(io.recvline(False)))

    io.recvlines(2)
    message = decrypt(io.recvline(False))
    assert (message == b'Hello guest!')

    for _i in range(0, 2):
        decrypt_line()
        path_encrypted = encrypt(b'../../app/server-ecdhkey.pem')
        io.sendline(path_encrypted)
        decrypt_line()

        padding = b'A' * pad_length
        assert (len(padding + known_bytes) == 240)
        pad_encrypted = encrypt(padding)
        io.sendline(pad_encrypted)

        msg = decrypt_line()
        hash_prev = msg.decode()[34:]
        if (pad_length == 240):
            hash_priv_key = hash_prev
        # print("previous hash is: ", hash_prev)
        assert (len(hash_prev) == 64)

        if _i == 1:
            found = False
            # print("padding size: ", len(padding))
            print("known_bytes size: ", len(known_bytes))
            for i in range(0, 256):
                msg = padding + i.to_bytes() + known_bytes
                assert (len(msg) == 241)
                hash_candidate = sha256sumhex(msg)
                assert (len(hash_candidate) == len(hash_prev))
                if hash_candidate == hash_prev:
                    print("byte found!: ", i.to_bytes())
                    found = True
                    known_bytes = i.to_bytes() + known_bytes
                    break
            assert (found)

    io.close()
    return known_bytes


def get_priv_key():
    known_bytes = b''
    for i in range(240, -1, -1):
        known_bytes = brute_force(i, known_bytes)
        print("known bytes = ", known_bytes)
    assert (len(known_bytes) == 241)
    with open("server-ecdhkey.pem", "wb") as f:
        f.write(known_bytes)


def exploit():

    with open("server-ecdhcert.pem", "rb") as f:
        server_cert = x509.load_pem_x509_certificate(f.read())

    client_cert_bytes = open("admin-ecdhcert.pem", "rb").read()
    client_cert = x509.load_pem_x509_certificate(client_cert_bytes)

    with open("server-ecdhkey.pem", "rb") as f:
        server_key = serialization.load_pem_private_key(
            f.read(), None)

    io = remote("mytls.2023.ctfcompetition.com", 1337)

    io.recvuntil(b'Please provide the client certificate in PEM format:')
    io.sendline(client_cert_bytes)

    io.recvuntil(b'Please provide the ephemeral client random:\n')
    client_ephemeral_random = token_hex(16)
    io.sendline(client_ephemeral_random.encode())

    io.recvuntil(b'Please provide the ephemeral client key:\n')
    client_ephemeral_key = ec.generate_private_key(ec.SECP256R1())
    client_ephemeral_public_key = client_ephemeral_key.public_key()
    io.sendline(client_ephemeral_public_key.public_bytes(
        serialization.Encoding.PEM, format=serialization.PublicFormat.SubjectPublicKeyInfo))

    io.recvuntil(b'Server ephemeral random:\n')
    server_ephemeral_random = io.recvline(False)
    io.recvline_contains(b'Server ephemeral key:')
    server_ephemeral_key_bytes = b''.join(io.recvlines(4, True))
    server_ephemeral_public_key = serialization.load_pem_public_key(
        server_ephemeral_key_bytes)

    io.recvlines(2)

    client_ephemeral_secret = client_ephemeral_key.exchange(
        ec.ECDH(), server_ephemeral_public_key)
    client_secret = server_key.exchange(ec.ECDH(), client_cert.public_key())
    derived_key = HKDF(algorithm=hashes.SHA256(),
                       length=32,
                       salt=b'SaltyMcSaltFace',
                       info=b'mytls').derive(
                           client_ephemeral_secret +
                           client_secret +
                           client_ephemeral_random.encode('utf-8') +
                           server_ephemeral_random)
    client_hmac = hmac.HMAC(derived_key, hashes.SHA256())
    client_hmac.update(b'client myTLS successful!')
    io.sendline(binascii.hexlify(client_hmac.finalize()))

    def decrypt(msg): return decrypt_message(
        msg, server_ephemeral_random, derived_key)

    def encrypt(msg): return encrypt_message(
        msg, server_ephemeral_random, derived_key)

    def decrypt_line(): return decrypt(io.recvline(False))

    def print_decrypt(): print(decrypt(io.recvline(False)))

    io.recvlines(2)
    print_decrypt()


if __name__ == '__main__':
    get_priv_key()
    exploit()