# -*- Mode:Python; indent-tabs-mode:nil; tab-width:4; encoding:utf-8 -*- # # Copyright 2002 Ben Escoto # Copyright 2007 Kenneth Loafman # # This file is part of duplicity. # # Duplicity is free software; you can redistribute it and/or modify it # under the terms of the GNU General Public License as published by the # Free Software Foundation; either version 2 of the License, or (at your # option) any later version. # # Duplicity is distributed in the hope that it will be useful, but # WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU # General Public License for more details. # # You should have received a copy of the GNU General Public License # along with duplicity; if not, write to the Free Software Foundation, # Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA import os import platform import random import unittest import pytest from duplicity import config from duplicity import gpg from duplicity import path from duplicity import util from testing import _runtest_dir from . import UnitTestCase @pytest.mark.usefixtures("redirect_stdin") class GPGTest(UnitTestCase): """Test GPGFile""" def setUp(self): super().setUp() self.unpack_testfiles() self.default_profile = gpg.GPGProfile(passphrase="foobar") def gpg_cycle(self, s, profile=None): """Test encryption/decryption cycle on string s""" epath = path.Path(f"{_runtest_dir}/testfiles/output/encrypted_file") if not profile: profile = self.default_profile encrypted_file = gpg.GPGFile(1, epath, profile) encrypted_file.write(s) encrypted_file.close() epath2 = path.Path(f"{_runtest_dir}/testfiles/output/encrypted_file") decrypted_file = gpg.GPGFile(0, epath2, profile) dec_buf = decrypted_file.read() decrypted_file.close() assert s == dec_buf, (len(s), len(dec_buf)) def test_gpg1(self): """Test gpg short strings""" if config.use_gpgsm: pytest.skip("gpgsm does not support symmetric (password) encryption") self.gpg_cycle(b"hello, world") self.gpg_cycle(b"ansoetuh aoetnuh aoenstuh aoetnuh asoetuh saoteuh ") def test_gpg2(self): """Test gpg long strings easily compressed""" if config.use_gpgsm: pytest.skip("gpgsm does not support symmetric (password) encryption") self.gpg_cycle(b" " * 50000) self.gpg_cycle(b"aoeu" * 1000000) def test_gpg3(self): """Test on random data - must have /dev/urandom device""" if config.use_gpgsm: pytest.skip("gpgsm does not support symmetric (password) encryption") infp = open("/dev/urandom", "rb") rand_buf = infp.read(120000) infp.close() self.gpg_cycle(rand_buf) def test_gpg_asym(self): """Test GPG asymmetric encryption""" profile = gpg.GPGProfile( passphrase=self.sign_passphrase, recipients=[self.encrypt_key1, self.encrypt_key2], ) if config.use_gpgsm and profile.gpg_version < (2, 2, 27): pytest.skip(f"Version {profile.gpg_version} of gpgsm is not supported. Minimum version is 2.2.27") self.gpg_cycle(b"aoensutha aonetuh saoe", profile) profile2 = gpg.GPGProfile(passphrase=self.sign_passphrase, recipients=[self.encrypt_key1]) self.gpg_cycle(b"aoeu" * 10000, profile2) def test_gpg_hidden_asym(self): """Test GPG asymmetric encryption with hidden key id""" if config.use_gpgsm: pytest.skip("gpgsm does not support hidden recipient") profile = gpg.GPGProfile( passphrase=self.sign_passphrase, hidden_recipients=[self.encrypt_key1, self.encrypt_key2], ) self.gpg_cycle(b"aoensutha aonetuh saoe", profile) profile2 = gpg.GPGProfile(passphrase=self.sign_passphrase, hidden_recipients=[self.encrypt_key1]) self.gpg_cycle(b"aoeu" * 10000, profile2) def test_gpg_signing(self): """Test to make sure GPG reports the proper signature key""" plaintext = b"hello" * 50000 signing_profile = gpg.GPGProfile( passphrase=self.sign_passphrase, sign_key=self.sign_key, recipients=[self.encrypt_key1], ) if config.use_gpgsm and signing_profile.gpg_version <= (2, 2, 27): pytest.skip(f"Version {signing_profile.gpg_version} of gpgsm is not supported. Minimum version is 2.2.27") epath = path.Path(f"{_runtest_dir}/testfiles/output/encrypted_file") encrypted_signed_file = gpg.GPGFile(1, epath, signing_profile) encrypted_signed_file.write(plaintext) encrypted_signed_file.close() decrypted_file = gpg.GPGFile(0, epath, signing_profile) assert decrypted_file.read() == plaintext decrypted_file.close() sig = decrypted_file.get_signature() assert sig == self.sign_key, sig def test_gpg_signing_and_hidden_encryption(self): """Test to make sure GPG reports the proper signature key even with hidden encryption key id""" if config.use_gpgsm: pytest.skip("gpgsm does not support hidden recipient") plaintext = b"hello" * 50000 signing_profile = gpg.GPGProfile( passphrase=self.sign_passphrase, sign_key=self.sign_key, hidden_recipients=[self.encrypt_key1], ) epath = path.Path(f"{_runtest_dir}/testfiles/output/encrypted_file") encrypted_signed_file = gpg.GPGFile(1, epath, signing_profile) encrypted_signed_file.write(plaintext) encrypted_signed_file.close() decrypted_file = gpg.GPGFile(0, epath, signing_profile) assert decrypted_file.read() == plaintext decrypted_file.close() sig = decrypted_file.get_signature() assert sig == self.sign_key, sig @pytest.mark.xfail def test_GPGWriteFile(self): """Test GPGWriteFile""" size = 400 * 1000 gwfh = GPGWriteFile_Helper() profile = gpg.GPGProfile(passphrase="foobar") for i in range(10): gpg.GPGWriteFile( gwfh, f"{_runtest_dir}/testfiles/output/gpgwrite.gpg", profile, size=size, ) assert ( size - 64 * 1024 <= os.stat(f"{_runtest_dir}/testfiles/output/gpgwrite.gpg").st_size <= size + 64 * 1024 # noqs ), ( f"{size - 64 * 1024}" f" <= {os.stat(f'{_runtest_dir}/testfiles/output/gpgwrite.gpg').st_size}" f" <= {size + 64 * 1024} Failed." # noqs ) gwfh.set_at_end() gpg.GPGWriteFile(gwfh, f"{_runtest_dir}/testfiles/output/gpgwrite.gpg", profile, size=size) @unittest.skip("Flaky test because it relies on compressed size of random bytes") def test_GzipWriteFile(self): """Test GzipWriteFile""" size = 400 * 1000 gwfh = GPGWriteFile_Helper() for i in range(10): gpg.GzipWriteFile(gwfh, f"{_runtest_dir}/testfiles/output/gzwrite.gz", size=size) assert ( size - 64 * 1024 <= os.stat(f"{_runtest_dir}/testfiles/output/gzwrite.gz").st_size <= size + 64 * 1024 # noqa ), ( f"{size - 64 * 1024}" f" <= {os.stat(f'{_runtest_dir}/testfiles/output/gzwrite.gz').st_size}" f" <= {size + 64 * 1024} Failed." # noqs ) gwfh.set_at_end() gpg.GzipWriteFile(gwfh, f"{_runtest_dir}/testfiles/output/gzwrite.gz", size=size) @pytest.mark.usefixtures("redirect_stdin") class GPGSMTest(GPGTest): """Test compatibility with 'gpgsm' the GnuPG tool for S/MIME""" sign_key = None sign_passphrase = None encrypt_key1 = "0x0F1ABB99" encrypt_key2 = "0xBEC52982" def setUp(self): super().setUp() self.old_use_gpgsm = config.use_gpgsm self.old_gpg_binary = config.gpg_binary config.use_gpgsm = True config.gpg_binary = util.which("gpgsm") self.default_profile = gpg.GPGProfile(passphrase="foobar") def tearDown(self): config.use_gpgsm = self.old_use_gpgsm config.gpg_binary = self.old_gpg_binary # Inherited test_gpg_asym class GPGWriteHelper2(object): def __init__(self, data): self.data = data class GPGWriteFile_Helper(object): """Used in test_GPGWriteFile above""" def __init__(self): self.from_random_fp = open("/dev/urandom", "rb") self.at_end = False def set_at_end(self): """Iterator stops when you call this""" self.at_end = True def get_buffer(self, size): """Return buffer of size size, consisting of half random data""" s1 = size // 2 s2 = size - s1 return b"a" * s1 + self.from_random_fp.read(s2) def __next__(self): if self.at_end: raise StopIteration block_data = self.get_buffer(self.get_read_size()) return GPGWriteHelper2(block_data) def get_read_size(self): size = 64 * 1024 if random.randrange(2): return size else: return random.randrange(1, size) def get_footer(self): return b"e" * random.randrange(0, 15000) class SHATest(UnitTestCase): """Test making sha signatures""" def setUp(self): super().setUp() self.unpack_testfiles() def test_sha(self): testhash = gpg.get_hash( "SHA1", path.Path(f"{_runtest_dir}/testfiles/various_file_types/regular_file"), ) # noqa assert testhash == "886d722999862724e1e62d0ac51c468ee336ef8e", testhash if __name__ == "__main__": unittest.main()