mirror of https://github.com/borgbackup/borg.git
376 lines
13 KiB
Python
376 lines
13 KiB
Python
from __future__ import with_statement
|
|
from getpass import getpass
|
|
import os
|
|
import msgpack
|
|
import shutil
|
|
import tempfile
|
|
import unittest
|
|
import zlib
|
|
|
|
from Crypto.Cipher import AES
|
|
from Crypto.Hash import SHA256, HMAC
|
|
from Crypto.Util import Counter
|
|
from Crypto.Util.number import bytes_to_long, long_to_bytes
|
|
from Crypto.Random import get_random_bytes
|
|
from Crypto.Protocol.KDF import PBKDF2
|
|
|
|
from .helpers import IntegrityError, get_keys_dir, Location
|
|
|
|
PREFIX = '\0' * 8
|
|
|
|
KEYFILE = '\0'
|
|
PASSPHRASE = '\1'
|
|
PLAINTEXT = '\2'
|
|
|
|
|
|
def key_creator(store, args):
|
|
if args.keyfile:
|
|
return KeyfileKey.create(store, args)
|
|
elif args.passphrase:
|
|
return PassphraseKey.create(store, args)
|
|
else:
|
|
return PlaintextKey.create(store, args)
|
|
|
|
|
|
def key_factory(store, manifest_data):
|
|
if manifest_data[0] == KEYFILE:
|
|
return KeyfileKey.detect(store, manifest_data)
|
|
elif manifest_data[0] == PASSPHRASE:
|
|
return PassphraseKey.detect(store, manifest_data)
|
|
elif manifest_data[0] == PLAINTEXT:
|
|
return PlaintextKey.detect(store, manifest_data)
|
|
else:
|
|
raise Exception('Unkown Key type %d' % ord(manifest_data[0]))
|
|
|
|
|
|
def SHA256_PDF(p, s):
|
|
return HMAC.new(p, s, SHA256).digest()
|
|
|
|
|
|
class KeyBase(object):
|
|
|
|
def id_hash(self, data):
|
|
"""Return HMAC hash using the "id" HMAC key
|
|
"""
|
|
|
|
def encrypt(self, data):
|
|
pass
|
|
|
|
def decrypt(self, id, data):
|
|
pass
|
|
|
|
|
|
class PlaintextKey(KeyBase):
|
|
TYPE = PLAINTEXT
|
|
|
|
chunk_seed = 0
|
|
|
|
@classmethod
|
|
def create(cls, store, args):
|
|
print 'Encryption NOT enabled.\nUse the --key-file or --passphrase options to enable encryption.'
|
|
return cls()
|
|
|
|
@classmethod
|
|
def detect(cls, store, manifest_data):
|
|
return cls()
|
|
|
|
def id_hash(self, data):
|
|
return SHA256.new(data).digest()
|
|
|
|
def encrypt(self, data):
|
|
return ''.join([self.TYPE, zlib.compress(data)])
|
|
|
|
def decrypt(self, id, data):
|
|
if data[0] != self.TYPE:
|
|
raise IntegrityError('Invalid encryption envelope')
|
|
data = zlib.decompress(data[1:])
|
|
if id and SHA256.new(data).digest() != id:
|
|
raise IntegrityError('Chunk id verification failed')
|
|
return data
|
|
|
|
|
|
class AESKeyBase(KeyBase):
|
|
|
|
def id_hash(self, data):
|
|
"""Return HMAC hash using the "id" HMAC key
|
|
"""
|
|
return HMAC.new(self.id_key, data, SHA256).digest()
|
|
|
|
def encrypt(self, data):
|
|
data = zlib.compress(data)
|
|
nonce = long_to_bytes(self.counter.next_value(), 8)
|
|
data = ''.join((nonce, AES.new(self.enc_key, AES.MODE_CTR, '',
|
|
counter=self.counter).encrypt(data)))
|
|
hash = HMAC.new(self.enc_hmac_key, data, SHA256).digest()
|
|
return ''.join((self.TYPE, hash, data))
|
|
|
|
def decrypt(self, id, data):
|
|
if data[0] != self.TYPE:
|
|
raise IntegrityError('Invalid encryption envelope')
|
|
hash = data[1:33]
|
|
if HMAC.new(self.enc_hmac_key, data[33:], SHA256).digest() != hash:
|
|
raise IntegrityError('Encryption envelope checksum mismatch')
|
|
nonce = bytes_to_long(data[33:41])
|
|
counter = Counter.new(64, initial_value=nonce, prefix=PREFIX)
|
|
data = zlib.decompress(AES.new(self.enc_key, AES.MODE_CTR, counter=counter).decrypt(data[41:]))
|
|
if id and HMAC.new(self.id_key, data, SHA256).digest() != id:
|
|
raise IntegrityError('Chunk id verification failed')
|
|
return data
|
|
|
|
def extract_iv(self, payload):
|
|
if payload[0] != self.TYPE:
|
|
raise IntegrityError('Invalid encryption envelope')
|
|
nonce = bytes_to_long(payload[33:41])
|
|
return nonce
|
|
|
|
def init_from_random_data(self, data):
|
|
self.enc_key = data[0:32]
|
|
self.enc_hmac_key = data[32:64]
|
|
self.id_key = data[64:96]
|
|
self.chunk_seed = bytes_to_long(data[96:100])
|
|
# Convert to signed int32
|
|
if self.chunk_seed & 0x80000000:
|
|
self.chunk_seed = self.chunk_seed - 0xffffffff - 1
|
|
self.counter = Counter.new(64, initial_value=1, prefix=PREFIX)
|
|
|
|
|
|
class PassphraseKey(AESKeyBase):
|
|
TYPE = PASSPHRASE
|
|
iterations = 10000
|
|
|
|
@classmethod
|
|
def create(cls, store, args):
|
|
key = cls()
|
|
passphrase = os.environ.get('DARC_PASSPHRASE')
|
|
if passphrase is not None:
|
|
passphrase2 = passphrase
|
|
else:
|
|
passphrase, passphrase2 = 1, 2
|
|
while passphrase != passphrase2:
|
|
passphrase = getpass('Enter passphrase: ')
|
|
if not passphrase:
|
|
print 'Passphrase must not be blank'
|
|
continue
|
|
passphrase2 = getpass('Enter same passphrase again: ')
|
|
if passphrase != passphrase2:
|
|
print 'Passphrases do not match'
|
|
key.init(store, passphrase)
|
|
if passphrase:
|
|
print 'Remember your passphrase. Your data will be inaccessible without it.'
|
|
return key
|
|
|
|
@classmethod
|
|
def detect(cls, store, manifest_data):
|
|
prompt = 'Enter passphrase for %s: ' % store._location.orig
|
|
key = cls()
|
|
passphrase = os.environ.get('DARC_PASSPHRASE')
|
|
if passphrase is None:
|
|
passphrase = getpass(prompt)
|
|
while True:
|
|
key.init(store, passphrase)
|
|
try:
|
|
key.decrypt(None, manifest_data)
|
|
iv = key.extract_iv(manifest_data)
|
|
key.counter = Counter.new(64, initial_value=iv + 1000, prefix=PREFIX)
|
|
return key
|
|
except IntegrityError:
|
|
passphrase = getpass(prompt)
|
|
|
|
def init(self, store, passphrase):
|
|
self.init_from_random_data(PBKDF2(passphrase, store.id, 100, self.iterations, SHA256_PDF))
|
|
|
|
|
|
class KeyfileKey(AESKeyBase):
|
|
FILE_ID = 'DARC KEY'
|
|
TYPE = KEYFILE
|
|
|
|
@classmethod
|
|
def detect(cls, store, manifest_data):
|
|
key = cls()
|
|
path = cls.find_key_file(store)
|
|
prompt = 'Enter passphrase for key file %s: ' % path
|
|
passphrase = os.environ.get('DARC_PASSPHRASE', '')
|
|
while not key.load(path, passphrase):
|
|
passphrase = getpass(prompt)
|
|
iv = key.extract_iv(manifest_data)
|
|
key.counter = Counter.new(64, initial_value=iv + 1000, prefix=PREFIX)
|
|
return key
|
|
|
|
@classmethod
|
|
def find_key_file(cls, store):
|
|
id = store.id.encode('hex')
|
|
keys_dir = get_keys_dir()
|
|
for name in os.listdir(keys_dir):
|
|
filename = os.path.join(keys_dir, name)
|
|
with open(filename, 'rb') as fd:
|
|
line = fd.readline().strip()
|
|
if line and line.startswith(cls.FILE_ID) and line[9:] == id:
|
|
return filename
|
|
raise Exception('Key file for store with ID %s not found' % id)
|
|
|
|
def load(self, filename, passphrase):
|
|
with open(filename, 'rb') as fd:
|
|
cdata = (''.join(fd.readlines()[1:])).decode('base64')
|
|
data = self.decrypt_key_file(cdata, passphrase)
|
|
if data:
|
|
key = msgpack.unpackb(data)
|
|
if key['version'] != 1:
|
|
raise IntegrityError('Invalid key file header')
|
|
self.store_id = key['store_id']
|
|
self.enc_key = key['enc_key']
|
|
self.enc_hmac_key = key['enc_hmac_key']
|
|
self.id_key = key['id_key']
|
|
self.chunk_seed = key['chunk_seed']
|
|
self.counter = Counter.new(64, initial_value=1, prefix=PREFIX)
|
|
self.path = filename
|
|
return True
|
|
|
|
def decrypt_key_file(self, data, passphrase):
|
|
d = msgpack.unpackb(data)
|
|
assert d['version'] == 1
|
|
assert d['algorithm'] == 'SHA256'
|
|
key = PBKDF2(passphrase, d['salt'], 32, d['iterations'], SHA256_PDF)
|
|
data = AES.new(key, AES.MODE_CTR, counter=Counter.new(128)).decrypt(d['data'])
|
|
if HMAC.new(key, data, SHA256).digest() != d['hash']:
|
|
return None
|
|
return data
|
|
|
|
def encrypt_key_file(self, data, passphrase):
|
|
salt = get_random_bytes(32)
|
|
iterations = 10000
|
|
key = PBKDF2(passphrase, salt, 32, iterations, SHA256_PDF)
|
|
hash = HMAC.new(key, data, SHA256).digest()
|
|
cdata = AES.new(key, AES.MODE_CTR, counter=Counter.new(128)).encrypt(data)
|
|
d = {
|
|
'version': 1,
|
|
'salt': salt,
|
|
'iterations': iterations,
|
|
'algorithm': 'SHA256',
|
|
'hash': hash,
|
|
'data': cdata,
|
|
}
|
|
return msgpack.packb(d)
|
|
|
|
def save(self, path, passphrase):
|
|
key = {
|
|
'version': 1,
|
|
'store_id': self.store_id,
|
|
'enc_key': self.enc_key,
|
|
'enc_hmac_key': self.enc_hmac_key,
|
|
'id_key': self.id_key,
|
|
'chunk_seed': self.chunk_seed,
|
|
}
|
|
data = self.encrypt_key_file(msgpack.packb(key), passphrase)
|
|
with open(path, 'wb') as fd:
|
|
fd.write('%s %s\n' % (self.FILE_ID, self.store_id.encode('hex')))
|
|
fd.write(data.encode('base64'))
|
|
self.path = path
|
|
|
|
def change_passphrase(self):
|
|
passphrase, passphrase2 = 1, 2
|
|
while passphrase != passphrase2:
|
|
passphrase = getpass('New passphrase: ')
|
|
passphrase2 = getpass('Enter same passphrase again: ')
|
|
if passphrase != passphrase2:
|
|
print 'Passphrases do not match'
|
|
self.save(self.path, passphrase)
|
|
print 'Key file "%s" updated' % self.path
|
|
|
|
@classmethod
|
|
def create(cls, store, args):
|
|
filename = args.store.to_key_filename()
|
|
path = filename
|
|
i = 1
|
|
while os.path.exists(path):
|
|
i += 1
|
|
path = filename + '.%d' % i
|
|
passphrase = os.environ.get('DARC_PASSPHRASE')
|
|
if passphrase is not None:
|
|
passphrase2 = passphrase
|
|
else:
|
|
passphrase, passphrase2 = 1, 2
|
|
while passphrase != passphrase2:
|
|
passphrase = getpass('Enter passphrase (empty for no passphrase):')
|
|
passphrase2 = getpass('Enter same passphrase again: ')
|
|
if passphrase != passphrase2:
|
|
print 'Passphrases do not match'
|
|
key = cls()
|
|
key.store_id = store.id
|
|
key.init_from_random_data(get_random_bytes(100))
|
|
key.save(path, passphrase)
|
|
print 'Key file "%s" created.' % key.path
|
|
print 'Keep this file safe. Your data will be inaccessible without it.'
|
|
return key
|
|
|
|
|
|
class KeyTestCase(unittest.TestCase):
|
|
|
|
def setUp(self):
|
|
self.tmppath = tempfile.mkdtemp()
|
|
os.environ['DARC_KEYS_DIR'] = self.tmppath
|
|
|
|
def tearDown(self):
|
|
shutil.rmtree(self.tmppath)
|
|
|
|
class MockStore(object):
|
|
class _Location(object):
|
|
orig = '/some/place'
|
|
|
|
_location = _Location()
|
|
id = '\0' * 32
|
|
|
|
def setUp(self):
|
|
self.tmpdir = tempfile.mkdtemp()
|
|
os.environ['DARC_KEYS_DIR'] = self.tmpdir
|
|
|
|
def tearDown(self):
|
|
shutil.rmtree(self.tmpdir)
|
|
|
|
def test_plaintext(self):
|
|
key = PlaintextKey.create(None, None)
|
|
data = 'foo'
|
|
self.assertEqual(key.id_hash(data).encode('hex'), '2c26b46b68ffc68ff99b453c1d30413413422d706483bfa0f98a5e886266e7ae')
|
|
self.assertEqual(data, key.decrypt(key.id_hash(data), key.encrypt(data)))
|
|
|
|
def test_keyfile(self):
|
|
class MockArgs(object):
|
|
store = Location(tempfile.mkstemp()[1])
|
|
os.environ['DARC_PASSPHRASE'] = 'test'
|
|
key = KeyfileKey.create(self.MockStore(), MockArgs())
|
|
self.assertEqual(bytes_to_long(key.counter()), 1)
|
|
manifest = key.encrypt('')
|
|
iv = key.extract_iv(manifest)
|
|
key2 = KeyfileKey.detect(self.MockStore(), manifest)
|
|
self.assertEqual(bytes_to_long(key2.counter()), iv + 1000)
|
|
# Key data sanity check
|
|
self.assertEqual(len(set([key2.id_key, key2.enc_key, key2.enc_hmac_key])), 3)
|
|
self.assertEqual(key2.chunk_seed == 0, False)
|
|
data = 'foo'
|
|
self.assertEqual(data, key2.decrypt(key.id_hash(data), key.encrypt(data)))
|
|
|
|
def test_passphrase(self):
|
|
os.environ['DARC_PASSPHRASE'] = 'test'
|
|
key = PassphraseKey.create(self.MockStore(), None)
|
|
self.assertEqual(bytes_to_long(key.counter()), 1)
|
|
self.assertEqual(key.id_key.encode('hex'), 'f28e915da78a972786da47fee6c4bd2960a421b9bdbdb35a7942eb82552e9a72')
|
|
self.assertEqual(key.enc_hmac_key.encode('hex'), '169c6082f209e524ea97e2c75318936f6e93c101b9345942a95491e9ae1738ca')
|
|
self.assertEqual(key.enc_key.encode('hex'), 'c05dd423843d4dd32a52e4dc07bb11acabe215917fc5cf3a3df6c92b47af79ba')
|
|
self.assertEqual(key.chunk_seed, -324662077)
|
|
manifest = key.encrypt('')
|
|
iv = key.extract_iv(manifest)
|
|
key2 = PassphraseKey.detect(self.MockStore(), manifest)
|
|
self.assertEqual(bytes_to_long(key2.counter()), iv + 1000)
|
|
self.assertEqual(key.id_key, key2.id_key)
|
|
self.assertEqual(key.enc_hmac_key, key2.enc_hmac_key)
|
|
self.assertEqual(key.enc_key, key2.enc_key)
|
|
self.assertEqual(key.chunk_seed, key2.chunk_seed)
|
|
data = 'foo'
|
|
self.assertEqual(key.id_hash(data).encode('hex'), '016c27cd40dc8e84f196f3b43a9424e8472897e09f6935d0d3a82fb41664bad7')
|
|
self.assertEqual(data, key2.decrypt(key2.id_hash(data), key.encrypt(data)))
|
|
|
|
|
|
def suite():
|
|
return unittest.TestLoader().loadTestsFromTestCase(KeyTestCase)
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main() |