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