ctypescrypto/cipher.py

   1 """
   2 access to symmetric ciphers from libcrypto
   3
   4 """
   5 from ctypes import create_string_buffer, c_char_p, c_void_p, c_int
   6 from ctypes import byref, POINTER
   7 from ctypescrypto import libcrypto, pyver, bintype
   8 from ctypescrypto.exception import LibCryptoError
   9 from ctypescrypto.oid import Oid
  10
  11 CIPHER_ALGORITHMS = ("DES", "DES-EDE3", "BF", "AES-128", "AES-192", "AES-256")
  12 CIPHER_MODES = ("STREAM", "ECB", "CBC", "CFB", "OFB", "CTR", "GCM")
  13
  14 #
  15
  16 __all__ = ['CipherError', 'new', 'Cipher', 'CipherType']
  17
  18 class CipherError(LibCryptoError):
  19     """
  20     Exception raise when OpenSSL function returns error
  21     """
  22     pass
  23
  24 def new(algname, key, encrypt=True, iv=None):
  25     """
  26     Returns new cipher object ready to encrypt-decrypt data
  27
  28     @param algname - string algorithm name like in opemssl command
  29                      line
  30     @param key - binary string representing ciher key
  31     @param encrypt - if True (default) cipher would be initialized
  32                     for encryption, otherwise - for decrypton
  33     @param iv - initialization vector
  34     """
  35     ciph_type = CipherType(algname)
  36     return Cipher(ciph_type, key, iv, encrypt)
  37
  38 class CipherType(object):
  39     """
  40     Describes cihper algorihm. Can be used to produce cipher
  41     instance and to get various information about cihper
  42     """
  43
  44     def __init__(self, cipher_name):
  45         """
  46         Constructs cipher algortihm using textual name as in openssl
  47         command line
  48         """
  49         if pyver > 2:
  50             cipher_name = cipher_name.encode('utf-8')
  51         self.cipher = libcrypto.EVP_get_cipherbyname(cipher_name)
  52         if self.cipher is None:
  53             raise CipherError("Unknown cipher: %s" % cipher_name)
  54
  55     def __del__(self):
  56         """
  57         It is constant object with do-nothing del
  58         """
  59         pass
  60
  61     def block_size(self):
  62         """
  63         Returns block size of the cipher
  64         """
  65         return libcrypto.EVP_CIPHER_block_size(self.cipher)
  66
  67     def key_length(self):
  68         """
  69         Returns key length of the cipher
  70         """
  71         return libcrypto.EVP_CIPHER_key_length(self.cipher)
  72
  73     def iv_length(self):
  74         """
  75         Returns initialization vector length of the cipher
  76         """
  77         return libcrypto.EVP_CIPHER_iv_length(self.cipher)
  78
  79     def flags(self):
  80         """
  81         Return cipher flags. Low three bits of the flags encode
  82         cipher mode (see mode). Higher bits  is combinatuon of
  83         EVP_CIPH* constants defined in the <openssl/evp.h>
  84         """
  85         return libcrypto.EVP_CIPHER_flags(self.cipher)
  86
  87     def mode(self):
  88         """
  89         Returns cipher mode as string constant like CBC, OFB etc.
  90         """
  91         return CIPHER_MODES[self.flags() & 0x7]
  92
  93     def algo(self):
  94         """
  95         Return cipher's algorithm name, derived from OID
  96         """
  97         return self.oid().shortname()
  98
  99     def oid(self):
 100         """
 101         Returns ASN.1 object identifier of the cipher as
 102         ctypescrypto.oid.Oid object
 103         """
 104         return Oid(libcrypto.EVP_CIPHER_nid(self.cipher))
 105
 106 class Cipher(object):
 107     """
 108     Performs actual encrypton decryption
 109     Note that object keeps some internal state.
 110     To obtain full ciphertext (or plaintext during decihpering)
 111     user should concatenate results of all calls of update with
 112     result of finish
 113     """
 114     def __init__(self, cipher_type, key, iv, encrypt=True):
 115         """
 116         Initializing cipher instance.
 117
 118         @param cipher_type - CipherType object
 119         @param key = binary string representing the key
 120         @param iv - binary string representing initializtion vector
 121         @param encrypt - if True(default) we ere encrypting.
 122                Otherwise decrypting
 123
 124         """
 125         self._clean_ctx()
 126         # Check key and iv length
 127         if key is None:
 128             raise ValueError("No key specified")
 129
 130         key_ptr = c_char_p(key)
 131         iv_ptr = c_char_p(iv)
 132         self.ctx = libcrypto.EVP_CIPHER_CTX_new()
 133         if self.ctx == 0:
 134             raise CipherError("Unable to create cipher context")
 135         self.encrypt = encrypt
 136         enc = 1 if encrypt else 0
 137         if not iv is None and len(iv) != cipher_type.iv_length():
 138             raise ValueError("Invalid IV length for this algorithm")
 139
 140         if len(key) != cipher_type.key_length():
 141             if (cipher_type.flags() & 8) != 0:
 142                 # Variable key length cipher.
 143                 result = libcrypto.EVP_CipherInit_ex(self.ctx,
 144                                                      cipher_type.cipher,
 145                                                      None, None, None,
 146                                                      c_int(enc))
 147                 result = libcrypto.EVP_CIPHER_CTX_set_key_length(self.ctx,
 148                                                                  len(key))
 149                 if result == 0:
 150                     self._clean_ctx()
 151                     raise CipherError("Unable to set key length")
 152                 result = libcrypto.EVP_CipherInit_ex(self.ctx, None, None,
 153                                                      key_ptr, iv_ptr,
 154                                                      c_int(enc))
 155             else:
 156                 raise ValueError("Invalid key length for this algorithm")
 157         else:
 158             result = libcrypto.EVP_CipherInit_ex(self.ctx, cipher_type.cipher,
 159                                                  None, key_ptr, iv_ptr,
 160                                                  c_int(enc))
 161         if result == 0:
 162             self._clean_ctx()
 163             raise CipherError("Unable to initialize cipher")
 164         self.cipher_type = cipher_type
 165         self.block_size = self.cipher_type.block_size()
 166         self.cipher_finalized = False
 167
 168     def __del__(self):
 169         """
 170         We define _clean_ctx() to do all the cleanup
 171         """
 172         self._clean_ctx()
 173
 174     def padding(self, padding=True):
 175         """
 176         Sets padding mode of the cipher
 177         """
 178         padding_flag = 1 if padding else 0
 179         libcrypto.EVP_CIPHER_CTX_set_padding(self.ctx, padding_flag)
 180
 181     def update(self, data):
 182         """
 183         Performs actual encrypton/decrypion
 184
 185         @param data - part of the plain text/ciphertext to process
 186         @returns - part of ciphercext/plain text
 187
 188         Passed chunk of text doesn't need to contain full ciher
 189         blocks. If neccessery, part of passed data would be kept
 190         internally until next data would be received or finish
 191         called
 192         """
 193         if self.cipher_finalized:
 194             raise CipherError("No updates allowed")
 195         if not isinstance(data, bintype):
 196             raise TypeError("A byte string is expected")
 197         if len(data) == 0:
 198             return ""
 199         outbuf = create_string_buffer(self.block_size+len(data))
 200         outlen = c_int(0)
 201         ret = libcrypto.EVP_CipherUpdate(self.ctx, outbuf, byref(outlen),
 202                                          data, len(data))
 203         if ret <= 0:
 204             self._clean_ctx()
 205             self.cipher_finalized = True
 206             raise CipherError("problem processing data")
 207         return outbuf.raw[:int(outlen.value)]
 208
 209     def finish(self):
 210         """
 211         Finalizes processing. If some data are kept in the internal
 212         state, they would be processed and returned.
 213         """
 214         if self.cipher_finalized:
 215             raise CipherError("Cipher operation is already completed")
 216         outbuf = create_string_buffer(self.block_size)
 217         self.cipher_finalized = True
 218         outlen = c_int(0)
 219         result = libcrypto.EVP_CipherFinal_ex(self.ctx, outbuf, byref(outlen))
 220         if result == 0:
 221             self._clean_ctx()
 222             raise CipherError("Unable to finalize cipher")
 223         if outlen.value > 0:
 224             return outbuf.raw[:int(outlen.value)]
 225         else:
 226             return b""
 227
 228     def _clean_ctx(self):
 229         """
 230         Cleans up cipher ctx and deallocates it
 231         """
 232         try:
 233             if self.ctx is not None:
 234                 self.__ctxcleanup(self.ctx)
 235                 libcrypto.EVP_CIPHER_CTX_free(self.ctx)
 236                 del self.ctx
 237         except AttributeError:
 238             pass
 239         self.cipher_finalized = True
 240
 241
 242 #
 243 # Used C function block_size
 244 #
 245 libcrypto.EVP_CIPHER_block_size.argtypes = (c_void_p, )
 246
 247 #Function EVP_CIPHER_CTX_cleanup renamed to EVP_CIPHER_CTX_reset
 248 # in the OpenSSL 1.1.0
 249 if hasattr(libcrypto,"EVP_CIPHER_CTX_cleanup"):
 250     Cipher.__ctxcleanup = libcrypto.EVP_CIPHER_CTX_cleanup
 251 else:
 252     Cipher.__ctxcleanup = libcrypto.EVP_CIPHER_CTX_reset
 253 Cipher.__ctxcleanup.argtypes  = (c_void_p, )
 254 libcrypto.EVP_CIPHER_CTX_free.argtypes = (c_void_p, )
 255 libcrypto.EVP_CIPHER_CTX_new.restype = c_void_p
 256 libcrypto.EVP_CIPHER_CTX_set_padding.argtypes = (c_void_p, c_int)
 257 libcrypto.EVP_CipherFinal_ex.argtypes = (c_void_p, c_char_p, POINTER(c_int))
 258 libcrypto.EVP_CIPHER_flags.argtypes = (c_void_p, )
 259 libcrypto.EVP_CipherInit_ex.argtypes = (c_void_p, c_void_p, c_void_p, c_char_p,
 260                                         c_char_p, c_int)
 261 libcrypto.EVP_CIPHER_iv_length.argtypes = (c_void_p, )
 262 libcrypto.EVP_CIPHER_key_length.argtypes = (c_void_p, )
 263 libcrypto.EVP_CIPHER_nid.argtypes = (c_void_p, )
 264 libcrypto.EVP_CipherUpdate.argtypes = (c_void_p, c_char_p, POINTER(c_int),
 265                                        c_char_p, c_int)
 266 libcrypto.EVP_get_cipherbyname.restype = c_void_p
 267 libcrypto.EVP_get_cipherbyname.argtypes = (c_char_p, )
 268 libcrypto.EVP_CIPHER_CTX_set_key_length.argtypes = (c_void_p, c_int)