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.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.
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()
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()