From c8b94594cbb9a80e876706621b3bb7bf7bf13125 Mon Sep 17 00:00:00 2001 From: Teemu Rytilahti Date: Tue, 24 Oct 2017 20:39:51 +0200 Subject: [PATCH 1/3] use upstream android_backup --- miio/android_backup.py | 226 ----------------------------------------- miio/extract_tokens.py | 2 +- 2 files changed, 1 insertion(+), 227 deletions(-) delete mode 100755 miio/android_backup.py diff --git a/miio/android_backup.py b/miio/android_backup.py deleted file mode 100755 index d32bb16b9..000000000 --- a/miio/android_backup.py +++ /dev/null @@ -1,226 +0,0 @@ -#!/usr/bin/env python -# -*- coding: utf-8 -*- -# This is a fork from original work by BlueC0re -# https://github.com/bluec0re/android-backup-tools -# License: GPLv3 -import tarfile -import zlib -import enum -import io -import pickle -import os -import binascii - -try: - from Crypto.Cipher import AES - from Crypto.Protocol.KDF import PBKDF2 - from Crypto import Random -except ImportError: - AES = None - - -class CompressionType(enum.IntEnum): - NONE = 0 - ZLIB = 1 - - -class EncryptionType(enum.Enum): - NONE = 'none' - AES256 = 'AES-256' - - -class AndroidBackup: - def __init__(self, fname=None): - if fname: - self.open(fname) - - def open(self, fname, mode='rb'): - self.fp = open(fname, mode) - - def close(self): - self.fp.close() - - def parse(self): - self.fp.seek(0) - magic = self.fp.readline() - assert magic == b'ANDROID BACKUP\n' - self.version = int(self.fp.readline().strip()) - self.compression = CompressionType(int(self.fp.readline().strip())) - self.encryption = EncryptionType(self.fp.readline().strip().decode()) - - def is_encrypted(self): - return self.encryption == EncryptionType.AES256 - - def _decrypt(self, enc, password): - if AES is None: - raise ImportError("PyCrypto required") - - user_salt, enc = enc.split(b'\n', 1) - user_salt = binascii.a2b_hex(user_salt) - ck_salt, enc = enc.split(b'\n', 1) - ck_salt = binascii.a2b_hex(ck_salt) - rounds, enc = enc.split(b'\n', 1) - rounds = int(rounds) - iv, enc = enc.split(b'\n', 1) - iv = binascii.a2b_hex(iv) - master_key, enc = enc.split(b'\n', 1) - master_key = binascii.a2b_hex(master_key) - - user_key = PBKDF2(password, user_salt, dkLen=256//8, count=rounds) - cipher = AES.new(user_key, - mode=AES.MODE_CBC, - IV=iv) - - master_key = list(cipher.decrypt(master_key)) - l = master_key.pop(0) - master_iv = bytes(master_key[:l]) - master_key = master_key[l:] - l = master_key.pop(0) - mk = bytes(master_key[:l]) - master_key = master_key[l:] - l = master_key.pop(0) - master_ck = bytes(master_key[:l]) - - # gen checksum - - # double encode utf8 - utf8mk = self.encode_utf8(mk) - calc_ck = PBKDF2(utf8mk, ck_salt, dkLen=256//8, count=rounds) - assert calc_ck == master_ck - - cipher = AES.new(mk, - mode=AES.MODE_CBC, - IV=master_iv) - - dec = cipher.decrypt(enc) - pad = dec[-1] - - return dec[:-pad] - - @staticmethod - def encode_utf8(mk): - utf8mk = mk.decode('raw_unicode_escape') - utf8mk = list(utf8mk) - for i in range(len(utf8mk)): - c = ord(utf8mk[i]) - # fix java encoding (add 0xFF00 to non ascii chars) - if 0x7f < c < 0x100: - c += 0xff00 - utf8mk[i] = chr(c) - return ''.join(utf8mk).encode('utf-8') - - def _encrypt(self, dec, password): - if AES is None: - raise ImportError("PyCrypto required") - - master_key = Random.get_random_bytes(32) - master_salt = Random.get_random_bytes(64) - user_salt = Random.get_random_bytes(64) - master_iv = Random.get_random_bytes(16) - user_iv = Random.get_random_bytes(16) - rounds = 10000 - - l = len(dec) - pad = 16 - (l % 16) - dec += bytes([pad] * pad) - cipher = AES.new(master_key, IV=master_iv, mode=AES.MODE_CBC) - enc = cipher.encrypt(dec) - - master_ck = PBKDF2(self.encode_utf8(master_key), - master_salt, dkLen=256//8, count=rounds) - - user_key = PBKDF2(password, - user_salt, dkLen=256//8, count=rounds) - - master_dec = b"\x10" + master_iv + b"\x20" + master_key + b"\x20" + master_ck - l = len(master_dec) - pad = 16 - (l % 16) - master_dec += bytes([pad] * pad) - cipher = AES.new(user_key, IV=user_iv, mode=AES.MODE_CBC) - master_enc = cipher.encrypt(master_dec) - - enc = binascii.b2a_hex(user_salt).upper() + b"\n" + \ - binascii.b2a_hex(master_salt).upper() + b"\n" + \ - str(rounds).encode() + b"\n" + \ - binascii.b2a_hex(user_iv).upper() + b"\n" + \ - binascii.b2a_hex(master_enc).upper() + b"\n" + enc - - return enc - - def read_data(self, password): - """Reads from the file and returns a TarFile object.""" - data = self.fp.read() - - if self.encryption == EncryptionType.AES256: - if password is None: - raise Exception("Password need to be provided to extract encrypted archives") - data = self._decrypt(data, password) - - if self.compression == CompressionType.ZLIB: - data = zlib.decompress(data, zlib.MAX_WBITS) - - tar = tarfile.TarFile(fileobj=io.BytesIO(data)) - return tar - - def unpack(self, target_dir=None, password=None): - tar = self.read_data(password) - - members = tar.getmembers() - - if target_dir is None: - target_dir = os.path.basename(self.fp.name) + '_unpacked' - pickle_fname = os.path.basename(self.fp.name) + '.pickle' - if not os.path.exists(target_dir): - os.mkdir(target_dir) - - tar.extractall(path=target_dir) - - with open(pickle_fname, 'wb') as fp: - pickle.dump(members, fp) - - def list(self, password=None): - tar = self.read_tar(password) - return tar.list() - - def pack(self, fname, password=None): - target_dir = os.path.basename(fname) + '_unpacked' - pickle_fname = os.path.basename(fname) + '.pickle' - - data = io.BytesIO() - tar = tarfile.TarFile(name=fname, - fileobj=data, - mode='w', - format=tarfile.PAX_FORMAT) - - with open(pickle_fname, 'rb') as fp: - members = pickle.load(fp) - - os.chdir(target_dir) - for member in members: - if member.isreg(): - tar.addfile(member, open(member.name, 'rb')) - else: - tar.addfile(member) - - tar.close() - - data.seek(0) - if self.compression == CompressionType.ZLIB: - data = zlib.compress(data.read()) - if self.encryption == EncryptionType.AES256: - data = self._encrypt(data, password) - - with open(fname, 'wb') as fp: - fp.write(b'ANDROID BACKUP\n') - fp.write('{}\n'.format(self.version).encode()) - fp.write('{:d}\n'.format(self.compression).encode()) - fp.write('{}\n'.format(self.encryption.value).encode()) - - fp.write(data) - - def __exit__(self, *args, **kwargs): - self.close() - - def __enter__(self): - self.parse() - return self diff --git a/miio/extract_tokens.py b/miio/extract_tokens.py index f917f82a5..bd62c2723 100644 --- a/miio/extract_tokens.py +++ b/miio/extract_tokens.py @@ -5,7 +5,7 @@ from Crypto.Cipher import AES from pprint import pformat as pf import attr -from .android_backup import AndroidBackup +from android_backup import AndroidBackup logging.basicConfig(level=logging.INFO) _LOGGER = logging.getLogger(__name__) From a4e9bf0fa3ab570e918f7fd085cfebce069e9bd2 Mon Sep 17 00:00:00 2001 From: Teemu Rytilahti Date: Thu, 26 Oct 2017 22:43:27 +0200 Subject: [PATCH 2/3] use non-streaming mode for reading just the database file --- miio/extract_tokens.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/miio/extract_tokens.py b/miio/extract_tokens.py index bd62c2723..491c7ec3e 100644 --- a/miio/extract_tokens.py +++ b/miio/extract_tokens.py @@ -111,7 +111,7 @@ def main(backup, write_to_disk, password, dump_all, dump_raw): reader = BackupDatabaseReader(dump_raw) if backup.endswith(".ab"): DBFILE = "apps/com.xiaomi.smarthome/db/miio2.db" - with AndroidBackup(backup) as f: + with AndroidBackup(backup, stream=False) as f: tar = f.read_data(password) try: db = tar.extractfile(DBFILE) From 97718b43086304f206175a7f3da043c1daf40a90 Mon Sep 17 00:00:00 2001 From: Teemu Rytilahti Date: Thu, 26 Oct 2017 22:44:51 +0200 Subject: [PATCH 3/3] fix install_requires to be saner --- setup.py | 11 +++++++++-- 1 file changed, 9 insertions(+), 2 deletions(-) diff --git a/setup.py b/setup.py index 9f2709e26..7a9795765 100644 --- a/setup.py +++ b/setup.py @@ -26,8 +26,15 @@ packages=["miio", "mirobo"], python_requires='>=3.4', - install_requires=['construct', 'click', 'cryptography', 'pretty_cron', - 'typing', 'zeroconf', 'pycrypto', 'attrs', 'pytz'], + install_requires=['construct', + 'click', + 'cryptography', + 'pretty_cron', + 'typing', + 'zeroconf', + 'pycrypto', + 'attrs', + 'android_backup'], entry_points={ 'console_scripts': [