BKP 2017 Sponge Challenge

hash.py

rom Crypto.Cipher import AES
from SocketServer import ThreadingMixIn
from BaseHTTPServer import HTTPServer, BaseHTTPRequestHandler
import sys

class Hasher:
  def __init__(self):
    self.aes = AES.new('\x00'*16)

  def reset(self):
    self.state = '\x00'*16

  def ingest(self, block):
    """Ingest a block of 10 characters """
    block += '\x00'*6
    state = ""
    for i in range(16):
      state += chr(ord(self.state[i]) ^ ord(block[i]))
    self.state = self.aes.encrypt(state)

  def final_ingest(self, block):
    """Call this for the final ingestion.
    Calling this with a 0 length block is the same as calling it one round
    earlier with a 10 length block.
    """
    if len(block) == 10:
      self.ingest(block)
      self.ingest('\x80' + '\x00'*8 + '\x01')
    elif len(block) == 9:
      self.ingest(block + '\x81')
    else:
      self.ingest(block + '\x80' + '\x00'*(8-len(block)) + '\x01')

  def squeeze(self):
    """Output a block of hash information"""
    result = self.state[:10]
    self.state = self.aes.encrypt(self.state)
    return result

  def hash(self, s):
    """Hash an input of any length of bytes.  Return a 160-bit digest."""
    self.reset()
    blocks = len(s) // 10
    for i in range(blocks):
      self.ingest(s[10*i:10*(i+1)])
    self.final_ingest(s[blocks*10:])

    return self.squeeze() + self.squeeze()

class HashHandler(BaseHTTPRequestHandler):
  def do_GET(self):
    if self.path in ['/favicon.ico', '/index.html']:
      # Stop.
      self.send_response(409)
      return

    try:
      to_hash = self.path[1:].decode('hex')
    except TypeError:
      # Bad hex.
      self.send_response(418)
      return

    if to_hash == GIVEN:
      # Nice try.
      self.send_response(451)
      return

    result = HASHER.hash(to_hash)
    if result != TARGET:
      # Wrong
      self.send_response(400)
      return
    self.send_response(200)
    self.end_headers()
    self.wfile.write(FLAG)

class ThreadedHTTPServer(ThreadingMixIn, HTTPServer):
  pass

if __name__=='__main__':
  assert(len(sys.argv) >= 3)
  HASHER = Hasher()
  with open('FLAG.txt') as f:
    FLAG = f.read()
  GIVEN = 'I love using sponges for crypto'
  TARGET = HASHER.hash(GIVEN)
  server = ThreadedHTTPServer((sys.argv[1], int(sys.argv[2])), HashHandler)
  server.serve_forever()

spongecrack.py

from Crypto.Cipher import AES
from itertools import *
from hash import Hasher

GIVEN = 'I love using sponges for crypto'

class SpongeBreaker(object):

    def __init__(self, target, key='\x00'*16):
        self.target = target
        self.targets = {}
        self.aes = AES.new(key)
        self.iterspace = permutations(range(255),10)
        self.preimage  = '00010203040507cc7033000000000000'.decode('hex')
        self.collision = '4961e4542e04e19932eb0ee589b64127'.decode('hex')
        
    def precompute(self, n):
        j = 0
        for i in islice(self.iterspace, n):
            prefix = str(bytearray(i))
            if j % 1000000 == 0:
                print(j)
            dec = self.aes.decrypt(prefix+'\x00'*6)
            self.targets[dec[10:]] = dec
            j+=1

    def collide(self):
        j = 0
        for i in self.iterspace:
            s = bytes(bytearray(i))+'\x00'*6
            h = self.aes.encrypt(s)
            if j % 1000000 == 0:
                print(j)
            j+=1
            if h[10:] in self.targets:
                self.preimage = s
                self.collision = self.targets[h[10:]]
                print("Preimage: %s\nCollides with: %s" % (self.preimage.encode('hex'), self.collision.encode('hex')))
                return (self.preimage, self.collision)

    def compute_preimage(self):
        state = '\x00'*16
        state = self.aes.encrypt(xor_bytes(state, self.preimage))
        diff = xor_bytes(self.aes.encrypt(self.preimage),self.collision)
        state = self.aes.encrypt(xor_bytes(state,diff)) # Capacity is now 0
        b0 = self.preimage[:10]
        b1 = diff[:10]
        b2 = xor_bytes(state[:10], self.target[:10])
        s = (b0+b1+b2+self.target[10:]).encode('hex')
        return s

    def fast_preimage(self):
	return "00010203040507cc7033865e7ffd7c215c84c2bb"+ (
    	xor_bytes(self.target[:10], "\x00\x01\x02\x03\x04\x05\x07\x6f\x60\xcc")+self.target[10:]
    	).encode('hex')  

def xor_bytes(a,b):
    return bytes(bytearray([chr(ord(i)^ord(j)) for (i,j) in zip(a,b)]))

def test(msg, find_collisions=False, n=25000000):
    s = SpongeBreaker(msg)
    if find_collisions:
        s.precompute(n)
        s.collide()
    pre = s.compute_preimage()
    print("Preimage: "+pre)
    h = Hasher()
    correct = h.hash(msg)
    return h.hash(pre.decode('hex')) == correct