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