Source code for vespa.aes_gcm

#!/usr/bin/env python2
#
#    Copyright (C) 2013 Bo Zhu http://about.bozhu.me
#
#    Permission is hereby granted, free of charge, to any person obtaining a
#    copy of this software and associated documentation files (the "Software"),
#    to deal in the Software without restriction, including without limitation
#    the rights to use, copy, modify, merge, publish, distribute, sublicense,
#    and/or sell copies of the Software, and to permit persons to whom the
#    Software is furnished to do so, subject to the following conditions:
#
#    The above copyright notice and this permission notice shall be included in
#    all copies or substantial portions of the Software.
#
#    THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
#    IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
#    FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL
#    THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
#    LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
#    FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
#    DEALINGS IN THE SOFTWARE.

from Crypto.Cipher import AES
from Crypto.Util import Counter
from Crypto.Util.number import long_to_bytes, bytes_to_long


# GF(2^128) defined by 1 + a + a^2 + a^7 + a^128
# Please note the MSB is x0 and LSB is x127
[docs]def gf_2_128_mul(x, y): assert x < (1 << 128) assert y < (1 << 128) res = 0 for i in range(127, -1, -1): res ^= x * ((y >> i) & 1) # branchless x = (x >> 1) ^ ((x & 1) * 0xE1000000000000000000000000000000) assert res < 1 << 128 return res
[docs]class InvalidInputException(Exception): def __init__(self, msg): self.msg = msg def __str__(self): return str(self.msg)
[docs]class InvalidTagException(Exception): def __str__(self): return 'The authenticaiton tag is invalid.' # Galois/Counter Mode with AES-128 and 96-bit IV
[docs]class AES_GCM: def __init__(self, master_key): self.change_key(master_key)
[docs] def change_key(self, master_key): if master_key >= (1 << 128): raise InvalidInputException('Master key should be 128-bit') self.__master_key = long_to_bytes(master_key, 16) self.__aes_ecb = AES.new(self.__master_key, AES.MODE_ECB) self.__auth_key = bytes_to_long(self.__aes_ecb.encrypt(b'\x00' * 16)) # precompute the table for multiplication in finite field table = [] # for 8-bit for i in range(16): row = [] for j in range(256): row.append(gf_2_128_mul(self.__auth_key, j << (8 * i))) table.append(tuple(row)) self.__pre_table = tuple(table) self.prev_init_value = None # reset
def __times_auth_key(self, val): res = 0 for i in range(16): res ^= self.__pre_table[i][val & 0xFF] val >>= 8 return res def __ghash(self, aad, txt): len_aad = len(aad) len_txt = len(txt) # padding if 0 == len_aad % 16: data = aad else: data = aad + b'\x00' * (16 - len_aad % 16) if 0 == len_txt % 16: data += txt else: data += txt + b'\x00' * (16 - len_txt % 16) tag = 0 assert len(data) % 16 == 0 for i in range(len(data) / 16): tag ^= bytes_to_long(data[i * 16: (i + 1) * 16]) tag = self.__times_auth_key(tag) # print 'X\t', hex(tag) tag ^= ((8 * len_aad) << 64) | (8 * len_txt) tag = self.__times_auth_key(tag) return tag
[docs] def encrypt(self, init_value, plaintext, auth_data=b''): if init_value >= (1 << 96): raise InvalidInputException('IV should be 96-bit') # a naive checking for IV reuse if init_value == self.prev_init_value: raise InvalidInputException('IV must not be reused!') self.prev_init_value = init_value len_plaintext = len(plaintext) # len_auth_data = len(auth_data) if len_plaintext > 0: counter = Counter.new( nbits=32, prefix=long_to_bytes(init_value, 12), initial_value=2, # notice this allow_wraparound=True) aes_ctr = AES.new(self.__master_key, AES.MODE_CTR, counter=counter) if 0 != len_plaintext % 16: padded_plaintext = plaintext + \ b'\x00' * (16 - len_plaintext % 16) else: padded_plaintext = plaintext ciphertext = aes_ctr.encrypt(padded_plaintext)[:len_plaintext] else: ciphertext = b'' auth_tag = self.__ghash(auth_data, ciphertext) # print 'GHASH\t', hex(auth_tag) auth_tag ^= bytes_to_long(self.__aes_ecb.encrypt( long_to_bytes((init_value << 32) | 1, 16))) # assert len(ciphertext) == len(plaintext) assert auth_tag < (1 << 128) return ciphertext, auth_tag
[docs] def decrypt(self, init_value, ciphertext, auth_tag, auth_data=b''): if init_value >= (1 << 96): raise InvalidInputException('IV should be 96-bit') if auth_tag >= (1 << 128): raise InvalidInputException('Tag should be 128-bit') if auth_tag != self.__ghash(auth_data, ciphertext) ^ \ bytes_to_long(self.__aes_ecb.encrypt( long_to_bytes((init_value << 32) | 1, 16))): raise InvalidTagException len_ciphertext = len(ciphertext) if len_ciphertext > 0: counter = Counter.new( nbits=32, prefix=long_to_bytes(init_value, 12), initial_value=2, allow_wraparound=True) aes_ctr = AES.new(self.__master_key, AES.MODE_CTR, counter=counter) if 0 != len_ciphertext % 16: padded_ciphertext = ciphertext + \ b'\x00' * (16 - len_ciphertext % 16) else: padded_ciphertext = ciphertext plaintext = aes_ctr.decrypt(padded_ciphertext)[:len_ciphertext] else: plaintext = b'' return plaintext