3
\$\begingroup\$

I've been porting python Keyczar to work under 2to3 conversion (Github). I wanted to consolidate its streaming AES encrypt/decrypt backend interface with its string decrypt/encrypt. So I wrote a new implementation using io.BufferReader and io.BufferedWriter. In my construction, I decrypt up until the last block and then check the MAC before decrypting that block and unpadding.

I feel a bit unsure of whether this is the right thing to do, as it's not my typical construction, and whether instead I should just buffer the whole thing and not decrypt at all until the MAC is checked.

I'm not a Python developer so any API or idiomatic suggestions are very welcome too.

These snippets are covered under the Apache license, follow the links to see whole code.

Encrypt

 def EncryptIO(self, reader , writer):
    """
    Write ciphertext byte stream containing Header|IV|Ciph|Sig.
    """    
    mac = self.hmac_key.CreateStreamable()

    #Write and update mac for header
    header = self.Header()
    writer.write(header)
    mac.Update(header)

    #Generate, write, and update mac for IV
    iv_bytes = util.RandBytes(self.block_size)
    writer.write(iv_bytes)
    mac.Update(iv_bytes)

    buffer_size = self.block_size * 4
    cipher = self.__CreateCipher(self.key_bytes, iv_bytes)
    more = True
    while more:
      data = util.ReadLength(reader, buffer_size)
      more = len(data) == buffer_size

      if not more:
        data = self._Pad(data)

      #encrypt, update mac and write out ciphertext
      ciph_bytes = cipher.encrypt(data)
      mac.Update(ciph_bytes)
      writer.write(ciph_bytes)

    #call final mac and write whatever is returned
    ciph_bytes = cipher.final()
    mac.Update(ciph_bytes)
    writer.write(ciph_bytes)

    #write mac and flush
    writer.write(mac.Sign())
    writer.flush()

Decrypt

  def DecryptIO(self, header, reader, writer):
    mac = self.hmac_key.CreateStreamable()

    #update mac of header
    mac.Update(header)
    #read out iv and update mac
    iv_bytes = util.ReadLength(reader, self.block_size)
    mac.Update(iv_bytes)

    cipher = self.__CreateCipher(self.key_bytes, iv_bytes)

    more = True
    # buffer_size is a block multiple -- req'd
    buffer_size = self.block_size * 4 
    # read out one more byte into starting tag buff
    # so on a valid ciphertext we will know we are on 
    # the last block to unpad
    initial_len = util.HLEN + 1
    tagbuff = util.ReadLength(reader, initial_len)

    while more:
      buff = util.ReadLength(reader, buffer_size)
      more = len(buff) == buffer_size
      tempbuff = tagbuff + buff
      if more:
        buff = tempbuff[:buffer_size]
        tagbuff = tempbuff[buffer_size:]
      else:
        buff = tempbuff[:-util.HLEN]
        tagbuff = tempbuff[-util.HLEN:]
      mac.Update(buff)
      if not more:
        #check the mac before we try and finish decryption
        calctag = mac.Sign()
        if not util.ConstantTimeCompare(tagbuff, calctag):
          raise errors.InvalidSignatureError()
      ptext = cipher.decrypt(buff)
      if not more:
        #remove padding now that we know the ciphertext is valid
        ptext += cipher.final()
        ptext = self._UnPad(ptext)
      writer.write(ptext)
      writer.flush()
\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Browse other questions tagged or ask your own question.