55,9 → 55,6 |
# <http://www.py2exe.org/index.cgi/WorkingWithVariousPackagesAndModules> |
import lxml._elementpath as _dummy |
import gzip # py2exe again |
import hashlib # required by newer format |
# PBKDF2 stolen from Revelation |
import PBKDF2 |
|
USE_PYCRYPTO = True |
|
109,18 → 106,6 |
MODE_AND='and' |
MODE_OR='or' |
|
# Errors |
class Errors(object): |
class Error(Exception): |
def __str__(self): |
return self.msg |
class DecryptError(Error): |
def __init__(self, msg = 'Failed to decrypt data. Wrong password?'): |
self.msg = msg |
class DataFormatError(Error): |
def __init__(self, msg = 'Incorrect data format'): |
self.msg = msg |
|
def printe(s): |
' Print to stderr ' |
sys.stderr.write(s+'\n') |
324,134 → 309,46 |
printe('Configuration file (~/.relevation.conf) is not readable!') |
return ( fl, pw, mode ) |
|
class _DataReaderBase(object): |
' Common methods for reading data files ' |
def validate_compressed_padding(self, data): |
''' Checks that the gzip-compressed 'data' is padded correctly. |
validate_compressed_padding(str) -> bool |
''' |
padlen = ord(data[-1]) |
for i in data[-padlen:]: |
if ord(i) != padlen: |
return False |
return True |
def validate_cipher_length(self, data): |
''' Checks that encrypted 'data' has an appropriate length. |
validate_cipher_length(str) -> bool |
Encrypted data length must be a multiple of 16 |
''' |
return ( len(data) % 16 == 0 ) |
def _aes_decrypt_ecb(self, key, data): |
''' Decrypt AES cipher text in ECB mode |
_aes_decrypt_ecb(str, str) -> str |
class DataReaderV1(object): |
def _decrypt_compressed_data(self, key, cipher_text): |
''' Decrypt cipher_text using key. |
_decrypt_compressed_data(str, str) -> cleartext (gzipped xml) |
|
This function will use the underlying, available, cipher module. |
''' |
# Old format: |
# [0:12) 12B header: "rvl" 0x00, 0x01, 0x00 |
# [12:28) 16B ECB encrypted CBC IV |
# [28:] encrypted data |
if USE_PYCRYPTO: |
# Extract IV |
c = AES.new(key) |
cleardata = c.decrypt(data) |
iv = c.decrypt(cipher_text[12:28]) |
# Decrypt data, CBC mode |
c = AES.new(key, AES.MODE_CBC, iv) |
ct = c.decrypt(cipher_text[28:]) |
else: |
# Extract IV |
c = rijndael.Rijndael(key, keySize=len(key), padding=noPadding()) |
cleardata = c.decrypt(data) |
return cleardata |
def _aes_decrypt_cbc(self, key, iv, data): |
''' Decrypt AES cipher text in CBC mode |
_aes_decrypt_ecb(str, str, str) -> str |
|
This function will use the underlying, available, cipher module. |
''' |
if USE_PYCRYPTO: |
c = AES.new(key, AES.MODE_CBC, iv) |
cleardata = c.decrypt(data) |
else: |
iv = c.decrypt(cipher_text[12:28]) |
# Decrypt data, CBC mode |
bc = rijndael.Rijndael(key, keySize=len(key), padding=noPadding()) |
c = cbc.CBC(bc, padding=noPadding()) |
cleardata = c.decrypt(data, iv=iv) |
return cleardata |
ct = c.decrypt(cipher_text[28:], iv=iv) |
return ct |
def get_xml(self, data, password): |
''' Extract the XML contents from the encrypted and compressed input. |
get_xml(str, str) -> str |
''' |
pass |
|
class DataReaderV1(_DataReaderBase): |
''' Data reading for Revelation data files in the original format. |
Old format header: |
[0:12) 12B header: "rvl" 0x00, 0x01, 0x00 |
[12:28) 16B ECB encrypted IV (for CBC-encrypted data) |
[28:] CBC encrypted data |
''' |
def _decrypt_compressed_data(self, password, cipher_text): |
''' Decrypt cipher_text using password. |
_decrypt_compressed_data(str, str) -> cleartext (gzipped xml) |
''' |
# Minimum length of header |
if len(cipher_text) < 28: |
raise Errors.DataFormatError |
# Key <= Padded password |
key = password |
key += (chr(0) * (32 - len(password))) |
# Extract IV |
iv = self._aes_decrypt_ecb(key, cipher_text[12:28]) |
# Skip IV |
cipher_text = cipher_text[28:] |
# Input strings for decrypt must be a multiple of 16 in length |
if not self.validate_cipher_length(cipher_text): |
raise Errors.DataFormatError |
# Decrypt data, CBC mode |
return self._aes_decrypt_cbc(key, iv, cipher_text) |
|
def get_xml(self, data, password): |
# Pad password |
password += (chr(0) * (32 - len(password))) |
# Decrypt. Decrypted data is compressed |
cleardata_gz = self._decrypt_compressed_data(password, data) |
# Validate padding for decompression |
if not self.validate_compressed_padding(cleardata_gz): |
raise Errors.DataFormatError |
# Length of data padding |
padlen = ord(cleardata_gz[-1]) |
# Decompress actual data (15 is wbits [ref3] DON'T CHANGE, 2**15 is the (initial) buf size) |
padlen = ord(cleardata_gz[-1]) |
return zlib.decompress(cleardata_gz[:-padlen], 15, 2**15) |
|
class DataReaderV2(_DataReaderBase): |
''' Data reading for Revelation data files in the new format. |
New format header: |
[0:12) 12B header: "rvl" 0x00, 0x02, 0x00 |
[12:20) 8B salt |
[20:36) 16B IV (for CBC-encrypted data) |
[36:] CBC encrypted data |
The encryption key is derived from the password and salt through the |
PBKDF2 module. |
''' |
def _decrypt_compressed_data(self, password, cipher_text): |
# Minimum length of header |
if len(cipher_text) < 36: |
raise Errors.DataFormatError |
salt = cipher_text[12:20] |
iv = cipher_text[20:36] |
key = PBKDF2.PBKDF2(password, salt, iterations=12000).read(32) |
# Skip encryption header |
cipher_text = cipher_text[36:] |
if not self.validate_cipher_length(cipher_text): |
raise Errors.DataFormatError |
# Decrypt data (CBC) |
decrypted = self._aes_decrypt_cbc(key, iv, cipher_text) |
sha256hash = decrypted[0:32] |
# Skip hash. decrypted <= Decrypted, Compressed data |
decrypted = decrypted[32:] |
# Validate hash |
if sha256hash != hashlib.sha256(decrypted).digest(): |
raise Errors.DecryptError |
return decrypted |
|
def get_xml(self, data, password): |
# Decrypt... |
cleardata_gz = self._decrypt_compressed_data(password, data) |
# Validate padding for decompression |
if not self.validate_compressed_padding(cleardata_gz): |
raise Errors.DataFormatError |
# Decompress |
padlen = ord(cleardata_gz[-1]) |
return zlib.decompress(cleardata_gz[:-padlen]) |
|
class DataReader(object): |
''' Interface to read Revelation's data files ''' |
def __init__(self, filename): |
488,8 → 385,6 |
app_version = header[6:9] |
if data_version == '\x01': |
self._impl = DataReaderV1() |
elif data_version == '\x02': |
self._impl = DataReaderV2() |
else: |
raise IOError('File \'%s\' is in a newer, unsupported, data format' % self._filename) |
def get_xml(self, password): |