diff --git a/bunker/backends/__init__.py b/bunker/backends/__init__.py index 80ca188..c935f92 100644 --- a/bunker/backends/__init__.py +++ b/bunker/backends/__init__.py @@ -1,3 +1,7 @@ from .kvs import KeyValueStore +from .directory_store import DirectoryStore -components = {KeyValueStore.component_type: KeyValueStore} +components = { + KeyValueStore.component_type: KeyValueStore + , DirectoryStore.component_type: DirectoryStore + } diff --git a/bunker/backends/directory_store.py b/bunker/backends/directory_store.py new file mode 100644 index 0000000..5073915 --- /dev/null +++ b/bunker/backends/directory_store.py @@ -0,0 +1,84 @@ +import os +import tarfile +import shutil +import tempfile + +from .component import AbstractComponent + +class DirectoryStore(AbstractComponent): + """ + This is a store for encrypting the content of a + whole directory. Encrypted are all files, ie. ``os.path.isfile`` + returns true. All other files are ignored. + + Note that if ``delete`` is true all the ignored will be deleted! + This should not be a problem, as ``DirectoryStore`` does usually + not operate on true directories but on temporary directories that + are filled with content by anonther layer. + """ + component_type = "dir" + def __init__(self, bunker, name, password, directory=None, delete=None): + AbstractComponent.__init__(self, bunker, name, password) + + if(directory is None): + directory = tempfile.mkdtemp() + delete = True + + self._directory = directory + self._delete = delete + + file_, type_ = bunker._load_component(name, password) + if(type_ != self.component_type): + raise TypeError( + "loaded component type ({}) != required component type({})".format( + type_ + , self.component_type)) + + tar = tarfile.TarFile(fileobj=file_, mode="r") + print(tar.getmembers()) + tar.extractall(self._directory) + + + @classmethod + def new(cls, bunker, name, password): + directory = tempfile.mkdtemp() + delete = True + bunker._add_component(name, password, DirectoryStore.component_type) + + file_, type_ = bunker._load_component(name, password) + tar = tarfile.TarFile(fileobj=file_, mode="w") + + for fname in os.listdir(directory): + if(os.path.isfile(os.path.join(directory, fname))): + tar.add(os.path.join(directory, fname), arcname=fname) + + tar.close() + bunker._save_component(name, password, file_) + + return cls(bunker, name, password, directory, delete) + + + def write_back(self): + file_, type_ = self._bunker._load_component(self._name, self._password) + tar = tarfile.TarFile(fileobj=file_, mode="w") + for fname in os.listdir(self._directory): + if(os.path.isfile(os.path.join(self._directory, fname))): + tar.add(os.path.join(self._directory, fname), arcname=fname) + + tar.close() + self._bunker._save_component(self._name, self._password, file_) + + def close(self, notify_bunker=True): + self.write_back() + self._name = "" + self._password = b"" + + if(self._delete): + shutil.rmtree(self._directory) + + if(notify_bunker): + self._bunker.notify_component_is_closed(self) + + def get_directory(self): + return self._directory + diff --git a/test/test_backends_directory_store.py b/test/test_backends_directory_store.py new file mode 100644 index 0000000..031d96f --- /dev/null +++ b/test/test_backends_directory_store.py @@ -0,0 +1,55 @@ +import os +import pytest + +from bunker.bunker import Bunker +from bunker.backends.directory_store import DirectoryStore + +@pytest.fixture +def bunker_and_password(tmpdir): + path = os.path.join(str(tmpdir), "test.bunker") + bunker = Bunker.open(path) + + return bunker, b"H6ihKLXV8HMQWbJs" + +def test_add_directory_store(bunker_and_password): + bunker, password = bunker_and_password + + ds = bunker.add_component("dir", "test.dir", password) + + assert isinstance(ds, DirectoryStore) + +@pytest.fixture +def bunker_ds_password_name(tmpdir): + path = os.path.join(str(tmpdir), "test.bunker") + password = b"H6ihKLXV8HMQWbJs" + bunker = Bunker.open(path) + + ds = bunker.add_component("dir", "test.dir", password) + + return bunker, ds, password, "test.dir" + +@pytest.fixture +def file_contents(): + return { + "foo.tx": "test 123" + , "bar.tx": "foobar" + } + +def test_directory_store_add_get_files(bunker_ds_password_name, file_contents): + bunker, ds, password, name = bunker_ds_password_name + directory = ds.get_directory() + for fname, content in file_contents.items(): + with open(os.path.join(directory, fname), "w") as fout: + fout.write(content) + ds.close() + ds = bunker.get_component(name, password) + directory = ds.get_directory() + + for fname, content in file_contents.items(): + with open(os.path.join(directory, fname), "r") as fin: + assert fin.read() == content + + + + +