SECCON 2022 Qual - witches_symmetric_exam

solve.py

from pwn import *
from enum import Enum
from Crypto.Util import Counter
from Crypto.Util.Padding import pad, unpad
from Crypto.Util.number import long_to_bytes, bytes_to_long
from Crypto.Cipher._mode_gcm import _GHASH, _ghash_portable

BIN_NAME = ["python", './problem.py']
REMOTE_ADDR = 'witches-symmetric-exam.seccon.games'
REMOTE_PORT = 8080
LOCAL = False

if LOCAL: stream = process(BIN_NAME)
else: stream = remote(REMOTE_ADDR, REMOTE_PORT)

print(stream.recvuntil(b'ciphertext: ', drop=True).decode())
ct = bytes.fromhex(stream.recvline(keepends=False).decode())

class Result(Enum):
  OFBError = 1
  GCMError = 2
  OK = 3
  SPELL_PROMPT = 4

def _decrypt_response():
  l = stream.recvuntil(b'\n', b':', timeout=5)
  if b'ofb error' in l: return Result.OFBError
  if b'gcm error' in l: return Result.GCMError
  if b'secret spell:' in l: return Result.SPELL_PROMPT
  if b'ok' in l: return Result.OK
  return Result.SPELL_PROMPT
  assert False, l
def test_decs(cts):
  for ct in cts:
    stream.sendline(ct.hex().encode())
  res = []
  for _ in cts:
    stream.recvuntil(b'ciphertext: ')
    res.append(_decrypt_response())
  return res
def test_dec(ct):
  return test_decs([ct])[0]


def encrypt(plain):
  assert len(plain) == 16
  known_suff = b''
  while len(known_suff) < 16:
    cts = []
    for i in range(256):
      known_suff_cand = bytes([i]) + known_suff
      padded = xor(known_suff_cand, bytes([len(known_suff_cand)])).rjust(16, b'\x00')
      cts.append(plain + padded)
    # print(cts)
    res = test_decs(cts)
    # print(res)
    assert Result.GCMError in res
    known_suff = bytes([res.index(Result.GCMError)]) + known_suff
  # print(known_suff)
  return known_suff

ofb_iv = ct[:16]
ofb_ct = ct[16:]

xor_in_ofb = encrypt(ofb_iv)

def extend_xor_in_ofb(length):
  global xor_in_ofb
  while len(xor_in_ofb) < length:
    xor_in_ofb += encrypt(xor_in_ofb[-16:])

extend_xor_in_ofb(len(ofb_ct))

# print(f'[+] {xor_in_ofb.hex()=}')

before_ofb = xor(ofb_ct, xor_in_ofb)
# print(f'[+] {before_ofb.hex()=}')
unpadded = unpad(before_ofb, 16)
# print(f'[+] {unpadded.hex()=}')

gcm_tag = unpadded[:16]
gcm_nonce = unpadded[16:32]
gcm_ct = unpadded[32:]

print(f'[+]   {gcm_tag.hex()=}')
print(f'[+] {gcm_nonce.hex()=}')
print(f'[+]    {gcm_ct.hex()=}')

hash_subkey = encrypt(b'\x00' * 16)
def compute_ctr_nonce(gcm_nonce):
  fill = (16 - (len(gcm_nonce) % 16)) % 16 + 8
  ghash_in = gcm_nonce + b'\x00' * fill + long_to_bytes(8 * len(gcm_nonce), 8)
  j0 = _GHASH(hash_subkey, _ghash_portable).update(ghash_in).digest()
  return j0

initial_nonce_ctr = compute_ctr_nonce(gcm_nonce)
nonce_ctr = initial_nonce_ctr

print(f'[+] {nonce_ctr.hex()=}')
xor_in_gcm = b''
def extend_xor_in_gcm(length):
  global xor_in_gcm, nonce_ctr
  while len(xor_in_gcm) < length:
    nonce_ctr = (int.from_bytes(nonce_ctr, "big") + 1).to_bytes(16, "big")
    xor_in_gcm += encrypt(nonce_ctr)

extend_xor_in_gcm(len(gcm_ct))
print(f'[+] {xor_in_gcm.hex()=}')

spell = xor(gcm_ct, xor_in_gcm[:len(gcm_ct)])
print(f'[+] {spell=}')

print(" === enc('give me key') === ")

plain = b'give me key'
extend_xor_in_gcm(len(plain))
gcm_ct = xor(plain, xor_in_gcm[:len(plain)])

print(f'[+] {gcm_ct.hex()=}')

hasher = _GHASH(hash_subkey, _ghash_portable)
pad_len = (16 - len(gcm_ct)) % 16
hasher.update(gcm_ct + b'\x00' * pad_len)

print(f'[+] ct all consumed')
print(f'[+] {hasher.digest().hex()=}')

hasher.update(long_to_bytes(0, 8) + long_to_bytes(8 * len(plain), 8))
print(f'[+] length consumed')
print(f'[+] {hasher.digest().hex()=}')

gcm_tag = xor(hasher.digest(), encrypt(initial_nonce_ctr))
print(f'[+] {gcm_tag.hex()=}')

before_ofb = pad(gcm_tag + gcm_nonce + gcm_ct, 16)

extend_xor_in_ofb(len(before_ofb))
res = ofb_iv + xor(before_ofb, xor_in_ofb[:len(before_ofb)])

assert test_dec(res) == Result.SPELL_PROMPT
print("[+] query accepted")
stream.sendline(spell)
stream.interactive()