1# This file is dual licensed under the terms of the Apache License, Version 2# 2.0, and the BSD License. See the LICENSE file in the root of this repository 3# for complete details. 4 5from __future__ import absolute_import, division, print_function 6 7from enum import Enum 8 9from six.moves import range 10 11from cryptography import utils 12from cryptography.exceptions import ( 13 AlreadyFinalized, InvalidKey, UnsupportedAlgorithm, _Reasons 14) 15from cryptography.hazmat.backends.interfaces import HMACBackend 16from cryptography.hazmat.primitives import constant_time, hashes, hmac 17from cryptography.hazmat.primitives.kdf import KeyDerivationFunction 18 19 20class Mode(Enum): 21 CounterMode = "ctr" 22 23 24class CounterLocation(Enum): 25 BeforeFixed = "before_fixed" 26 AfterFixed = "after_fixed" 27 28 29@utils.register_interface(KeyDerivationFunction) 30class KBKDFHMAC(object): 31 def __init__(self, algorithm, mode, length, rlen, llen, 32 location, label, context, fixed, backend): 33 if not isinstance(backend, HMACBackend): 34 raise UnsupportedAlgorithm( 35 "Backend object does not implement HMACBackend.", 36 _Reasons.BACKEND_MISSING_INTERFACE 37 ) 38 39 if not isinstance(algorithm, hashes.HashAlgorithm): 40 raise UnsupportedAlgorithm( 41 "Algorithm supplied is not a supported hash algorithm.", 42 _Reasons.UNSUPPORTED_HASH 43 ) 44 45 if not backend.hmac_supported(algorithm): 46 raise UnsupportedAlgorithm( 47 "Algorithm supplied is not a supported hmac algorithm.", 48 _Reasons.UNSUPPORTED_HASH 49 ) 50 51 if not isinstance(mode, Mode): 52 raise TypeError("mode must be of type Mode") 53 54 if not isinstance(location, CounterLocation): 55 raise TypeError("location must be of type CounterLocation") 56 57 if (label or context) and fixed: 58 raise ValueError("When supplying fixed data, " 59 "label and context are ignored.") 60 61 if rlen is None or not self._valid_byte_length(rlen): 62 raise ValueError("rlen must be between 1 and 4") 63 64 if llen is None and fixed is None: 65 raise ValueError("Please specify an llen") 66 67 if llen is not None and not isinstance(llen, int): 68 raise TypeError("llen must be an integer") 69 70 if label is None: 71 label = b'' 72 73 if context is None: 74 context = b'' 75 76 utils._check_bytes("label", label) 77 utils._check_bytes("context", context) 78 self._algorithm = algorithm 79 self._mode = mode 80 self._length = length 81 self._rlen = rlen 82 self._llen = llen 83 self._location = location 84 self._label = label 85 self._context = context 86 self._backend = backend 87 self._used = False 88 self._fixed_data = fixed 89 90 def _valid_byte_length(self, value): 91 if not isinstance(value, int): 92 raise TypeError('value must be of type int') 93 94 value_bin = utils.int_to_bytes(1, value) 95 if not 1 <= len(value_bin) <= 4: 96 return False 97 return True 98 99 def derive(self, key_material): 100 if self._used: 101 raise AlreadyFinalized 102 103 utils._check_byteslike("key_material", key_material) 104 self._used = True 105 106 # inverse floor division (equivalent to ceiling) 107 rounds = -(-self._length // self._algorithm.digest_size) 108 109 output = [b''] 110 111 # For counter mode, the number of iterations shall not be 112 # larger than 2^r-1, where r <= 32 is the binary length of the counter 113 # This ensures that the counter values used as an input to the 114 # PRF will not repeat during a particular call to the KDF function. 115 r_bin = utils.int_to_bytes(1, self._rlen) 116 if rounds > pow(2, len(r_bin) * 8) - 1: 117 raise ValueError('There are too many iterations.') 118 119 for i in range(1, rounds + 1): 120 h = hmac.HMAC(key_material, self._algorithm, backend=self._backend) 121 122 counter = utils.int_to_bytes(i, self._rlen) 123 if self._location == CounterLocation.BeforeFixed: 124 h.update(counter) 125 126 h.update(self._generate_fixed_input()) 127 128 if self._location == CounterLocation.AfterFixed: 129 h.update(counter) 130 131 output.append(h.finalize()) 132 133 return b''.join(output)[:self._length] 134 135 def _generate_fixed_input(self): 136 if self._fixed_data and isinstance(self._fixed_data, bytes): 137 return self._fixed_data 138 139 l_val = utils.int_to_bytes(self._length * 8, self._llen) 140 141 return b"".join([self._label, b"\x00", self._context, l_val]) 142 143 def verify(self, key_material, expected_key): 144 if not constant_time.bytes_eq(self.derive(key_material), expected_key): 145 raise InvalidKey 146