From b88a8ee299eab0e14d51fde714f1827322e29cae Mon Sep 17 00:00:00 2001 From: Jan Lindemann Date: Fri, 24 Apr 2026 16:29:14 +0200 Subject: [PATCH] cmds.secrets.lib.DistroContext.install(): Add method Add DistroContext.install(). It takes a tar file containing secrets, decrypts it, and installs all secrets needed on target and present in that file. For every file that should be extracted, it logs if it acutally did something or didn't. It also features an only_missing argument, which is just a stub for "allow to define somy extraction policy with respect to replacing all / some / not replace / whatever. Not thought through. Signed-off-by: Jan Lindemann --- .../jw/pkg/cmds/secrets/lib/DistroContext.py | 20 +++++++++++++++++++ 1 file changed, 20 insertions(+) diff --git a/src/python/jw/pkg/cmds/secrets/lib/DistroContext.py b/src/python/jw/pkg/cmds/secrets/lib/DistroContext.py index 7628fbb1..93985bc3 100644 --- a/src/python/jw/pkg/cmds/secrets/lib/DistroContext.py +++ b/src/python/jw/pkg/cmds/secrets/lib/DistroContext.py @@ -14,6 +14,8 @@ if TYPE_CHECKING: from ....lib.log import * from ....lib.util import run_cmd +from ....lib.TarIo import TarIo +from ....lib.ProcFilterGpg import ProcFilterGpg from .base import Attrs from .FilesContext import FilesContext @@ -72,3 +74,21 @@ class DistroContext(FilesContext): missing += 1 if missing > 0: log(WARNING, f'{missing} missing secrets found. You might want to add them and run sudo {app.cmdline} again') + + async def install(self, src_uri: str, pkg_names: Iterable[str], only_missing: bool=False) -> None: + if only_missing: + raise NotImplementedError('--only-missing is not yet implemented') + secret_paths = await self.list_secret_paths(pkg_names) + ec = self.ctx + from ....lib.ec.Local import Local + from ....lib.FileContext import FileContext + if not isinstance(ec, Local): + ec = Local() # Security: Use a local exec context for decrypting and filtering secrets + async with TarIo.create(src=src_uri, dst=self.ctx) as tio: + tio.src.add_proc_filter(FileContext.Direction.In, ProcFilterGpg(ec=ec)) + extracted_paths = await tio.extract(path_filter=secret_paths) + for path in secret_paths: + if not path in extracted_paths: + log(NOTICE, f'not extracted: {path}') + else: + log(NOTICE, f'extracted: {path}')