# SPDX-License-Identifier: GPL-3.0-or-later from collections.abc import Iterable from collections.abc import Iterator from contextlib import contextmanager from os import chdir from os import getcwd from pathlib import Path from re import split from subprocess import STDOUT from subprocess import CalledProcessError from subprocess import check_output from sys import exit from sys import stderr from tempfile import mkstemp from traceback import print_stack from typing import IO from typing import AnyStr from typing import Dict from typing import List from typing import Optional from typing import Set from typing import Union from libkeyringctl.types import Fingerprint from libkeyringctl.types import Trust @contextmanager def cwd(new_dir: Path) -> Iterator[None]: """Change to a new current working directory in a context and go back to the previous dir after the context is done Parameters ---------- new_dir: A path to change to """ previous_dir = getcwd() chdir(new_dir) try: yield finally: chdir(previous_dir) def natural_sort_path(_list: Iterable[Path]) -> Iterable[Path]: """Sort an Iterable of Paths naturally Parameters ---------- _list: An iterable containing paths to be sorted Return ------ An Iterable of paths that are naturally sorted """ def convert_text_chunk(text: str) -> Union[int, str]: """Convert input text to int or str Parameters ---------- text: An input string Returns ------- Either an integer if text is a digit, else text in lower-case representation """ return int(text) if text.isdigit() else text.lower() def alphanum_key(key: Path) -> List[Union[int, str]]: """Retrieve an alphanumeric key from a Path, that can be used in sorted() Parameters ---------- key: A path for which to create a key Returns ------- A list of either int or str objects that may serve as 'key' argument for sorted() """ return [convert_text_chunk(c) for c in split("([0-9]+)", str(key.name))] return sorted(_list, key=alphanum_key) def system(cmd: List[str], _stdin: Optional[IO[AnyStr]] = None, exit_on_error: bool = False) -> str: """Execute a command using check_output Parameters ---------- cmd: A list of strings to be fed to check_output _stdin: input fd used for the spawned process exit_on_error: Whether to exit the script when encountering an error (defaults to False) Raises ------ CalledProcessError: If not exit_on_error and `check_output()` encounters an error Returns ------- The output of cmd """ try: return check_output(cmd, stderr=STDOUT, stdin=_stdin).decode() except CalledProcessError as e: stderr.buffer.write(bytes(e.stdout, encoding="utf8")) print_stack() if exit_on_error: exit(e.returncode) raise e def absolute_path(path: str) -> Path: """Return the absolute path of a given str Parameters ---------- path: A string representing a path Returns ------- The absolute path representation of path """ return Path(path).absolute() def transform_fd_to_tmpfile(working_dir: Path, sources: List[Path]) -> None: """Transforms an input list of paths from any file descriptor of the current process to a tempfile in working_dir. Using this function on fd inputs allow to pass the content to another process while hidepid is active and /proc not visible for the other process. Parameters ---------- working_dir: A directory to use for temporary files sources: Paths that should be iterated and all fd's transformed to tmpfiles """ for index, source in enumerate(sources): if str(source).startswith("/proc/self/fd"): file = mkstemp(dir=working_dir, prefix=f"{source.name}", suffix=".fd")[1] with open(file, mode="wb") as f: f.write(source.read_bytes()) f.flush() sources[index] = Path(file) def get_cert_paths(paths: Iterable[Path]) -> Set[Path]: """Walks a list of paths and resolves all discovered certificate paths Parameters ---------- paths: A list of paths to walk and resolve to certificate paths. Returns ------- A set of paths to certificates """ # depth first search certificate paths cert_paths: Set[Path] = set() visit: List[Path] = list(paths) while visit: path = visit.pop() # this level contains a certificate, abort depth search if list(path.glob("*.asc")): cert_paths.add(path) continue visit.extend([path for path in path.iterdir() if path.is_dir()]) return cert_paths def get_parent_cert_paths(paths: Iterable[Path]) -> Set[Path]: """Walks a list of paths upwards and resolves all discovered parent certificate paths Parameters ---------- paths: A list of paths to walk and resolve to certificate paths. Returns ------- A set of paths to certificates """ # depth first search certificate paths cert_paths: Set[Path] = set() visit: List[Path] = list(paths) while visit: node = visit.pop().parent # this level contains a certificate, abort depth search if "keyring" == node.parent.parent.parent.name: cert_paths.add(node) continue visit.append(node) return cert_paths def contains_fingerprint(fingerprints: Iterable[Fingerprint], fingerprint: Fingerprint) -> bool: """Returns weather an iterable structure of fingerprints contains a specific fingerprint Parameters ---------- fingerprints: Iteratable structure of fingerprints that should be searched fingerprint: Fingerprint to search for Returns ------- Weather an iterable structure of fingerprints contains a specific fingerprint """ return any(filter(lambda e: str(e).endswith(fingerprint), fingerprints)) def filter_fingerprints_by_trust(trusts: Dict[Fingerprint, Trust], trust: Trust) -> List[Fingerprint]: """Filters a dict of Fingerprint to Trust by a passed Trust parameter and returns the matching fingerprints. Parameters ---------- trusts: Dict of Fingerprint to Trust that should be filtered based on the trust parameter trust: Trust that should be used to filter the trusts dict Returns ------- The matching fingerprints of the dict filtered by trust """ return list( map( lambda item: item[0], filter(lambda item: trust == item[1], trusts.items()), ) )