ctypescrypto/cipher.py

   1 from ctypes import create_string_buffer,c_char_p,c_void_p,c_int,c_long,byref\r
   2 from ctypescrypto import libcrypto\r
   3 from ctypescryto.exception import LibCrytoError\r
   4 \r
   5 CIPHER_ALGORITHMS = ("DES", "DES-EDE3", "BF", "AES-128", "AES-192", "AES-256")\r
   6 CIPHER_MODES = ("STREAM","ECB","CBC", "CFB", "OFB", "CTR","GCM")\r
   7 \r
   8 #\r
   9 \r
  10 class CipherError(LibCryptoError):\r
  11     pass\r
  12 \r
  13 def new(algname,key,encrypt=True,iv=None):\r
  14         ct=CipherType(algname)\r
  15         return Cipher(ct,key,iv,encrypt)\r
  16 \r
  17 class CipherType:\r
  18 \r
  19     def __init__(self, cipher_name):\r
  20         self.cipher = libcrypto.EVP_get_cipherbyname(cipher_name)\r
  21         if self.cipher is None:\r
  22             raise CipherError, "Unknown cipher: %s" % cipher_name\r
  23 \r
  24     def __del__(self):\r
  25         pass\r
  26         def block_size(self):\r
  27                 return libcrypto.EVP_CIHPER_block_size(self.cipher)\r
  28         def key_length(self):\r
  29                 return libcrypto.EVP_CIPHER_key_length(self.cipher)\r
  30         def iv_length(self):\r
  31                 return libcrypto.EVP_CIPHER_iv_length(self.cipher)\r
  32         def flags(self):\r
  33                 return libcrypto.EVP_CIPHER_flags(self.cipher)\r
  34         def mode(self):\r
  35                 return CIPHER_MODES[self.flags & 0x7]\r
  36     def algo(self):\r
  37         return self.oid().short_name() \r
  38     def mode(self):\r
  39         return self.cipher_mode\r
  40         def oid(self):\r
  41                 return Oid(libcrypto.EVP_CIPHER_nid(self.cipher))\r
  42 \r
  43 class Cipher:\r
  44 \r
  45     def __init__(self,  cipher_type, key, iv, encrypt=True):\r
  46         self._clean_ctx()\r
  47         key_ptr = c_char_p(key)\r
  48         iv_ptr = c_char_p(iv)\r
  49         self.ctx = libcrypto.EVP_CIPHER_CTX_new(cipher_type.cipher, None, key_ptr, iv_ptr)\r
  50         if self.ctx == 0:\r
  51             raise CipherError, "Unable to create cipher context"\r
  52         self.encrypt = encrypt\r
  53         if encrypt: \r
  54             enc = 1\r
  55         else: \r
  56             enc = 0\r
  57         result = libcrypto.EVP_CipherInit_ex(self.ctx, cipher_type.cipher, None, key_ptr, iv_ptr, c_int(enc))\r
  58         self.cipher_type = cipher_type\r
  59                 self.block_size = self.cipher_type.block_size()\r
  60         if result == 0:\r
  61             self._clean_ctx()\r
  62             raise CipherError, "Unable to initialize cipher"\r
  63 \r
  64     def __del__(self):\r
  65         self._clean_ctx()\r
  66 \r
  67     def enable_padding(self, padding=True):\r
  68         if padding:\r
  69             padding_flag = 1\r
  70         else:\r
  71             padding_flag = 0\r
  72         libcrypto.EVP_CIPHER_CTX_set_padding(self.ctx, padding_flag)\r
  73 \r
  74     def update(self, data):\r
  75         if self.cipher_finalized :\r
  76             raise CipherError, "No updates allowed"\r
  77         if type(data) != type(""):\r
  78             raise TypeError, "A string is expected"\r
  79         if len(data) <= 0:\r
  80             return ""\r
  81                 outbuf=string_buffer_create(self.blocsize+len(data))\r
  82                 outlen=c_int(0)\r
  83                 ret=libcrypto.EVP_CipherUpdate(self.ctx,outbuf,byref(outlen),\r
  84                         data,len(data))\r
  85                 if ret <=0:\r
  86                         self._clean_ctx()\r
  87                         self.cipher_finalized=True\r
  88                         del self.ctx\r
  89                         raise CipherError("problem processing data")\r
  90                 return outbuf.raw[:outlen]\r
  91     \r
  92     def finish(self):\r
  93         if self.cipher_finalized :\r
  94             raise CipherError, "Cipher operation is already completed"\r
  95                 outbuf=create_string_buffer(self.block_size)\r
  96         self.cipher_finalized = True\r
  97         result = self.libcrypto.EVP_CipherFinal_ex(self.ctx,outbuf , byref(outlen))\r
  98         if result == 0:\r
  99             self._clean_ctx()\r
 100             raise CipherError, "Unable to finalize cipher"\r
 101                 if outlen>0:\r
 102                 return outbuf.raw[:outlen]\r
 103                 else\r
 104                         return ""\r
 105         \r
 106     def _clean_ctx(self):\r
 107         try:\r
 108             if self.ctx is not None:\r
 109                 self.libcrypto.EVP_CIPHER_CTX_cleanup(self.ctx)\r
 110                 self.libcrypto.EVP_CIPHER_CTX_free(self.ctx)\r
 111                 del(self.ctx)\r
 112         except AttributeError:\r
 113             pass\r
 114         self.cipher_finalized = True\r