+++ /dev/null
-#!/usr/bin/env python3
-#-------------------------------------------------------------------
-# aescrypto.py
-#-------------------------------------------------------------------
-# Copyright (C) 2013 Intra2net AG
-# All rights reserved.
-#
-# Permission is hereby granted, free of charge, to any person
-# obtaining a copy of this software and associated documentation
-# files (the "Software"), to deal in the Software without
-# restriction, including without limitation the rights to use,
-# copy, modify, merge, publish, distribute, sublicense, and/or sell
-# copies of the Software, and to permit persons to whom the
-# Software is furnished to do so, subject to the following
-# conditions:
-#
-# The above copyright notice and this permission notice shall be
-# included in all copies or substantial portions of the Software.
-#
-# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
-# EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES
-# OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
-# NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT
-# HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY,
-# WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
-# FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR
-# OTHER DEALINGS IN THE SOFTWARE.
-
-# Author: Daniel Garcia <danigm@wadobo.com>
-
-'''
-AES encryption and decryption lib.
-This is a simple utility lib over pycrypto to encrypt and decrypt using AES
-compatible with openssl command.
-'''
-
-
-from hashlib import md5
-
-# we ignore the PowmInsecureWarning warning given by libgmp4 because it doesn't
-# affect our code
-import warnings
-try:
- from Crypto.pct_warnings import PowmInsecureWarning
- warnings.simplefilter("ignore", PowmInsecureWarning)
-
- from Crypto.Cipher import AES
- from Crypto import Random
-except:
- pass
-
-
-class AESCrypt:
- '''
- This class provides a simple method to encrypt and decrypt text using
- AES.
- '''
- def __init__(self, password, salt=b'', key_length=128):
- self.bs = AES.block_size
- self.mode = AES.MODE_CBC
- if key_length not in [128, 256]:
- raise Exception('Invalid key_length, only 128 and 256 allowed')
- self.key_length = int(key_length/8)
- self.buf = b''
- if salt:
- self.salt = salt
- else:
- self.salt = Random.new().read(self.bs - len(b'Salted__'))
- if isinstance(password, str):
- password = bytes(password, 'UTF-8')
- self.password = password
-
- self.get_pad = self.get_random_pad
- self.split_pad = self.split_random_pad
-
- def init(self):
- '''
- Initialize the Crypto.AES object with the password provided and the
- salt calculated.
-
- For decrypt you should call to get_salt or get_salt_str before the
- decryption to get the correct salt
- '''
- self.derive_key_and_iv()
- self.cipher = AES.new(self.key, self.mode, self.iv)
- self.salt_str = b'Salted__' + self.salt
-
- def close_enc(self):
- '''
- Adds the needed padding to the chunk to be able to encrypt and
- encrypts the remaining buf
-
- returns the encrypted text
- '''
- chunk = self.buf
- self.buf = b''
- need_padding = len(chunk) % self.bs != 0
- padding_length = self.bs - len(chunk) % self.bs
- chunk += self.get_pad(padding_length)
- return self.cipher.encrypt(chunk)
-
- def encrypt(self, chunk):
- '''
- Encrypts the text chunk given. If it's not multiple of Block Size
- the chunk is buffered and '' is returned, in other case the chunk
- encrypted is returned.
- '''
-
- self.buf += chunk
- if len(self.buf) % self.bs == 0:
- cipher = self.cipher.encrypt(self.buf)
- self.buf = b''
- return cipher
-
- cipher = b''
- while len(self.buf) >= self.bs:
- chunk = self.buf[:self.bs]
- self.buf = self.buf[self.bs:]
- cipher += self.cipher.encrypt(chunk)
-
- return cipher
-
- def decrypt(self, buf, end=False):
- '''
- Decrypts the buf. If end is True this will split the encryption
- padding.
-
- Returns the decrypted text
- '''
-
- bs = self.bs
-
- # Adding pad, only needed when there's no pad, when using OFB
- #if len(buf) % bs != 0:
- # buf += self.get_pad(bs - len(buf) % bs)
-
- chunk = self.cipher.decrypt(buf)
- if end:
- chunk = self.split_pad(chunk)
- return chunk
-
- def get_salt(self, instream):
- '''
- Calculates the salt for an input encrypted file
- '''
- self.salt = instream.read(self.bs)[len(b'Salted__'):]
-
- def get_salt_str(self, instr):
- '''
- Calculates the salt for an input encrypted string
- '''
- self.salt = instr[len(b'Salted__'):self.bs]
-
- def derive_key_and_iv(self):
- '''
- Generates the key and iv using the password and salt as seed
- '''
- d = d_i = b''
- l = self.key_length + self.bs
- while len(d) < l:
- d_i = md5(d_i + self.password + self.salt).digest()
- d += d_i
- self.key = d[:self.key_length]
- self.iv = d[self.key_length:self.key_length + self.bs]
-
- def get_random_pad(self, padding_length):
- '''
- Returns an ISO_10126 pad, which is random
- '''
- return Random.new().read(padding_length - 1) + bytes([padding_length])
-
- def split_random_pad(self, chunk):
- '''
- Returns the chunk without the ISO_10126 pad
- '''
- return chunk[:-chunk[-1]]
-
- def get_pkcs5_pad(self, padding_length):
- '''
- Returns the PKCS pad
- '''
- return padding_length * bytes([padding_length])
-
- def split_pkcs5_pad(self, chunk):
- '''
- Returns the chunk without the PKCS pad
- '''
- return chunk.rstrip(chunk[-1])
-
-
-def encrypt(in_file, out_file, password):
- aes = AESCrypt(password)
- aes.init()
- out_file.write(aes.salt_str)
-
- finished = False
- while not finished:
- chunk = in_file.read(1024 * aes.bs)
- if not chunk or len(chunk) < 1024 * aes.bs:
- finished = True
-
- chunk = aes.encrypt(chunk)
- out_file.write(chunk)
- # adding padding
- out_file.write(aes.close_enc())
-
-
-def decrypt(in_file, out_file, password):
- aes = AESCrypt(password)
- salt = aes.get_salt(in_file)
- aes.init()
-
- next_chunk = b''
- finished = False
- while not finished:
- buf = in_file.read(1024 * aes.bs)
- if not buf:
- finished = True
- chunk = next_chunk
- next_chunk = buf
- out_file.write(aes.decrypt(chunk, finished))
-
-
-if __name__ == '__main__':
- from io import StringIO
- infile = StringIO('clear text')
- cipher = StringIO()
- out = StringIO()
- encrypt(infile, cipher, 'key')
- cipher.seek(0)
- decrypt(cipher, out, 'key')
- out.seek(0)
- print(out.read())
git://developer.intra2net.com / python-delta-tar / commitdiff