add purge_extract_dir option to Client

.gitignore

@@ -7,6 +7,7 @@ examples/client/cache/
 examples/client/programs/
 examples/repo/keystore/
 examples/repo/repository/
+.ruff_cache/
 
 # files
 .tufup-repo-config

src/tufup/client.py

@@ -1,8 +1,10 @@
 import bsdiff4
+import json
 import logging
 import pathlib
 import shutil
 import sys
+import tarfile
 import tempfile
 from typing import Callable, Dict, Iterator, List, Optional, Tuple, Union
 from urllib import parse
@@ -13,11 +15,14 @@
 import tuf.ngclient
 
 from tufup.common import TargetMeta
+from tufup.utils import remove_path
 from tufup.utils.platform_specific import install_update
 
 logger = logging.getLogger(__name__)
 
-DEFAULT_EXTRACT_DIR = pathlib.Path(tempfile.gettempdir()) / 'tufup'
+# dir name must be unequivocally linked to tufup, as dir contents are removed
+EXTRACT_DIR_NAME = 'tufup_extract_dir'
+DEFAULT_EXTRACT_DIR = pathlib.Path(tempfile.gettempdir()) / EXTRACT_DIR_NAME
 
 
 class Client(tuf.ngclient.Updater):
@@ -33,6 +38,7 @@ def __init__(
         extract_dir: Optional[pathlib.Path] = None,
         refresh_required: bool = False,
         session_auth: Optional[Dict[str, Union[Tuple[str, str], AuthBase]]] = None,
+        purge_extract_dir: bool = False,
         **kwargs,
     ):
         # tuf.ngclient.Updater.__init__ loads local root metadata automatically
@@ -46,6 +52,7 @@ def __init__(
         )
         self.app_install_dir = app_install_dir
         self.extract_dir = extract_dir
+        self.purge_extract_dir = purge_extract_dir
         self.refresh_required = refresh_required
         self.current_archive = TargetMeta(name=app_name, version=current_version)
         self.current_archive_local_path = target_dir / self.current_archive.path
@@ -257,17 +264,9 @@ def _apply_updates(
             self.new_archive_info.verify_length_and_hashes(data=archive_bytes)
             # write the patched new archive
             self.new_archive_local_path.write_bytes(archive_bytes)
-        # extract archive to temporary directory
-        if self.extract_dir is None:
-            self.extract_dir = DEFAULT_EXTRACT_DIR
-            self.extract_dir.mkdir(exist_ok=True)
-            logger.debug(f'default extract dir created: {self.extract_dir}')
-        # extract
-        shutil.unpack_archive(
-            filename=self.new_archive_local_path, extract_dir=self.extract_dir
-        )
-        logger.debug(f'files extracted to {self.extract_dir}')
-        # install
+        # extract archive to "temporary" directory
+        self._extract_archive()
+        # install, i.e. move extracted files from "temporary" dir to final "install" dir
         confirmation_message = f'Install update in {self.app_install_dir}? [y]/n'
         if skip_confirmation or input(confirmation_message) in ['y', '']:
             # start a script that moves the extracted files to the app install
@@ -279,7 +278,24 @@ def _apply_updates(
             )
         else:
             logger.warning('Installation aborted.')
-        # todo: clean up deprecated local archive
+
+    def _extract_archive(self):
+        # ensure extract_dir exists
+        if self.extract_dir is None:
+            self.extract_dir = DEFAULT_EXTRACT_DIR
+            logger.debug(f'using default extract_dir: {self.extract_dir}')
+        self.extract_dir.mkdir(exist_ok=True)
+        # purge extract_dir
+        purge_manifest = PurgeManifest(dir_to_purge=self.extract_dir)
+        if self.purge_extract_dir:
+            purge_manifest.purge()
+        # extract the new archive into the extract_dir
+        shutil.unpack_archive(
+            filename=self.new_archive_local_path, extract_dir=self.extract_dir
+        )
+        logger.debug(f'files extracted to {self.extract_dir}')
+        # create new purge manifest in extract_dir
+        purge_manifest.create_from_archive(archive=self.new_archive_local_path)
 
 
 class AuthRequestsFetcher(tuf.ngclient.RequestsFetcher):
@@ -356,3 +372,70 @@ def _chunks(self, response: 'requests.Response') -> Iterator[bytes]:
         for data in super()._chunks(response=response):
             self._progress(bytes_new=len(data))
             yield data
+
+
+class PurgeManifest(object):
+    _filename = 'purge.manifest'
+
+    def __init__(self, dir_to_purge: Union[pathlib.Path, str]):
+        """
+        a purge manifest is used to remove files and subdirectories from the
+        specified directory (`dir_to_purge`) in a safe manner, i.e. minimizing the
+        risk of accidental removal of pre-existing files/dirs that are unrelated to
+        the present application
+
+        this is achieved by removing only those items listed in a purge.manifest file
+        in the specified directory
+
+        if there is no purge.manifest file, nothing is removed
+
+        the purge.manifest file should be created (or updated) whenever files are
+        added to the `dir_to_purge`
+
+        the purge manifest can be used to clear the extract_dir or the
+        app_install_dir, for example
+        """
+        self.dir_to_purge = pathlib.Path(dir_to_purge)
+
+    @property
+    def file_path(self) -> pathlib.Path:
+        return self.dir_to_purge / self._filename
+
+    def read_from_file(self) -> Optional[List[str]]:
+        if not self.file_path.exists():
+            logger.warning(f'cannot read manifest: {self.file_path} not found')
+            return
+        return json.loads(self.file_path.read_text())
+
+    def write_to_file(self, manifest: List[str]):
+        self.file_path.write_text(json.dumps(manifest))
+        logger.debug(f'purge manifest created: {self.file_path}')
+
+    def purge(self) -> None:
+        """
+        remove all files/folders listed in the purge manifest from the specified
+        directory (`dir_to_purge`)
+        """
+        manifest = self.read_from_file()
+        if manifest:
+            for path in self.dir_to_purge.iterdir():
+                relative_path = path.relative_to(self.dir_to_purge)
+                if str(relative_path) in manifest:
+                    remove_path(path=path, remove_readonly=True)
+                else:
+                    logger.warning(f'{path} remains: not in {self._filename}')
+            logger.info(f'directory purged: {self.dir_to_purge}')
+
+    def create_from_archive(self, archive: Union[pathlib.Path, str]) -> None:
+        """create/overwrite purge manifest in extract_dir"""
+        # note that `shutil.make_archive` puts everything in a dir called "." (for
+        # cwd) and prefixes items with "./" (if this ever poses a problem, we could
+        # switch to using `tarfile.TarFile.add()` with the `arcname` argument) see
+        # e.g https://stackoverflow.com/a/70081512
+        with tarfile.open(archive) as tar:
+            manifest = [self._filename] + [
+                tarinfo.name[2:] if tarinfo.name.startswith('./') else tarinfo.name
+                for tarinfo in tar
+                if tarinfo.name != '.'
+            ]
+        self.write_to_file(manifest=manifest)

src/tufup/repo/__init__.py

@@ -97,7 +97,7 @@ def make_gztar_archive(
         if input(f'Found existing archive: {archive_path}.\nOverwrite? [n]/y') != 'y':
             print('Using existing archive.')
             return TargetMeta(archive_path)
-    # make archive
+    # make archive (see https://stackoverflow.com/a/70081512 about "." directory)
     base_name = str(dst_dir / archive_filename.replace(SUFFIX_ARCHIVE, ''))
     archive_path_str = shutil.make_archive(
         base_name=base_name,  # archive file path, no suffix

src/tufup/utils/__init__.py

@@ -1,4 +1,5 @@
 import logging
+import os
 import pathlib
 import shutil
 import sys
@@ -9,7 +10,7 @@
 _INPUT_SEPARATOR = ' '
 
 
-def remove_path(path: Union[pathlib.Path, str]) -> bool:
+def remove_path(path: Union[pathlib.Path, str], remove_readonly=False) -> bool:
     """
     Recursively remove directory or file at specified path.
 
@@ -18,11 +19,27 @@ def remove_path(path: Union[pathlib.Path, str]) -> bool:
         for path in my_dir_path.iterdir():
             remove_path(path)
     """
+    def _remove_readonly(failed_function, failed_path, _):
+        """
+        clear the readonly bit and reattempt the removal
+
+        copied from https://docs.python.org/3/library/shutil.html#rmtree-example
+        """
+        os.chmod(failed_path, 0o777)
+        failed_function(failed_path)
+
     # enforce pathlib.Path
     path = pathlib.Path(path)
     try:
+        if remove_readonly:
+            # override any read-only permissions (note this will fail when dealing
+            # with a file in a readonly directory on linux, but the alternative would
+            # be to (temporarily) change permissions on the parent directory)
+            path.chmod(0o777)
         if path.is_dir():
-            shutil.rmtree(path=path)
+            shutil.rmtree(
+                path=path, onerror=_remove_readonly if remove_readonly else None
+            )
             utils_logger.debug(f'Removed directory {path}')
         elif path.is_file():
             path.unlink()

src/tufup/utils/platform_specific.py

@@ -128,7 +128,7 @@ def _install_update_win(
     batch_template_extra_kwargs: Optional[dict] = None,
     log_file_name: Optional[str] = None,
     robocopy_options_override: Optional[List[str]] = None,
-    process_creation_flags = None,
+    process_creation_flags=None,
 ):
     """
     Create a batch script that moves files from src to dst, then run the

tests/data/README.md

@@ -1,3 +1,5 @@
 Large parts of the test data were copied verbatim from the `python-tuf` [repository_data][1] folder.
 
+The repository data were created using `examples/repo_workflow_example.py`.
+
 [1]: https://github.com/theupdateframework/python-tuf/tree/develop/tests/repository_data