diff --git a/kiwi_scp/executable.py b/kiwi_scp/executable.py index f2646cc..48168d1 100644 --- a/kiwi_scp/executable.py +++ b/kiwi_scp/executable.py @@ -28,7 +28,7 @@ class Executable: def exe_file(self) -> Optional[Path]: return self.__find_exe_file(self.exe_name) - def __build_cmd(self, args, kwargs) -> List[Path, Any, ...]: + def __build_cmd(self, args, kwargs) -> List: cmd = [self.exe_file, *args] _logger.debug(f"Executable cmd{cmd}, kwargs{kwargs}") @@ -40,7 +40,7 @@ class Executable: **kwargs ) - def Popen(self, process_args, **kwargs) -> subprocess.Popen[str]: + def Popen(self, process_args, **kwargs) -> subprocess.Popen: return subprocess.Popen( self.__build_cmd(process_args, kwargs), **kwargs diff --git a/kiwi_scp/rootkit.py b/kiwi_scp/rootkit.py index a3c7937..7332c43 100644 --- a/kiwi_scp/rootkit.py +++ b/kiwi_scp/rootkit.py @@ -1,86 +1,78 @@ -# system +import functools import logging -import os import subprocess +from pathlib import Path +from typing import Optional, TypeVar, Union, List + +import attr -# local from ._constants import IMAGES_DIRECTORY_NAME, LOCAL_IMAGES_NAME, DEFAULT_IMAGE_NAME from .executable import Executable +_logger = logging.getLogger(__name__) + +PSL = TypeVar("PSL", Union[Path, str], List[Union[Path, str]]) + + +def prefix_path(path: PSL, prefix: Path = Path("/mnt")) -> PSL: + if isinstance(path, Path): + return prefix.joinpath(path.absolute()) -def _prefix_path(prefix, path): if isinstance(path, str): - abs_path = os.path.abspath(path) - return os.path.realpath(f"{prefix}/{abs_path}") + return prefix_path(Path(path), prefix) elif isinstance(path, list): - return [_prefix_path(prefix, p) for p in path] - - -def prefix_path_mnt(path): - return _prefix_path('/mnt/', path) - - -def _image_name(image_tag): - if image_tag is not None: - return f"{LOCAL_IMAGES_NAME}:{image_tag}" - else: - return DEFAULT_IMAGE_NAME + return [prefix_path(prefix, p) for p in path] +@attr.s class Rootkit: - class __Rootkit: - __image_tag = None + image_tag: str = attr.ib() - def __init__(self, image_tag=None): - self.__image_tag = image_tag + @staticmethod + @functools.lru_cache(maxsize=None) + def __image_name(image_tag: Optional[str]) -> str: + if image_tag is not None: + return f"{LOCAL_IMAGES_NAME}:{image_tag}" + else: + return DEFAULT_IMAGE_NAME - def __exists(self): - ps = Executable('docker').run([ - 'images', - '--filter', f"reference={_image_name(self.__image_tag)}", - '--format', '{{.Repository}}:{{.Tag}}' - ], stdout=subprocess.PIPE) + @staticmethod + @functools.lru_cache(maxsize=None) + def __exists(image_tag: str) -> bool: + ps = Executable('docker').run([ + 'images', + '--filter', f"reference={Rootkit.__image_name(image_tag)}", + '--format', '{{.Repository}}:{{.Tag}}' + ], stdout=subprocess.PIPE) - return str(ps.stdout, 'utf-8').strip() == _image_name(self.__image_tag) + return str(ps.stdout, 'utf-8').strip() == Rootkit.__image_name(image_tag) + + def __build_image(self) -> None: + if Rootkit.__exists(self.image_tag): + _logger.info(f"Using image {Rootkit.__image_name(self.image_tag)}") + else: + if self.image_tag is None: + _logger.info(f"Pulling image {Rootkit.__image_name(self.image_tag)}") + Executable('docker').run([ + 'pull', Rootkit.__image_name(self.image_tag) + ], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) - def __build_image(self): - if self.__exists(): - logging.info(f"Using image {_image_name(self.__image_tag)}") else: - if self.__image_tag is None: - logging.info(f"Pulling image {_image_name(self.__image_tag)}") - Executable('docker').run([ - 'pull', _image_name(self.__image_tag) - ], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) + _logger.info(f"Building image {Rootkit.__image_name(self.image_tag)}") + Executable('docker').run([ + 'build', + '-t', Rootkit.__image_name(self.image_tag), + '-f', f"{IMAGES_DIRECTORY_NAME}/{self.image_tag}.Dockerfile", + f"{IMAGES_DIRECTORY_NAME}" + ], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) - else: - logging.info(f"Building image {_image_name(self.__image_tag)}") - Executable('docker').run([ - 'build', - '-t', _image_name(self.__image_tag), - '-f', f"{IMAGES_DIRECTORY_NAME}/{self.__image_tag}.Dockerfile", - f"{IMAGES_DIRECTORY_NAME}" - ], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) - - def run(self, process_args, **kwargs): - self.__build_image() - Executable('docker').run([ - 'run', '--rm', - '-v', '/:/mnt', - '-u', 'root', - _image_name(self.__image_tag), - *process_args - ], **kwargs) - - __image_tag = None - __instances = {} - - def __init__(self, image_tag=None): - self.__image_tag = image_tag - - if _image_name(self.__image_tag) not in Rootkit.__instances: - Rootkit.__instances[_image_name(self.__image_tag)] = Rootkit.__Rootkit(image_tag) - - def __getattr__(self, item): - return getattr(self.__instances[_image_name(self.__image_tag)], item) + def run(self, process_args, **kwargs): + self.__build_image() + Executable('docker').run([ + 'run', '--rm', + '-v', '/:/mnt', + '-u', 'root', + Rootkit.__image_name(self.image_tag), + *process_args + ], **kwargs)