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