cmds.secrets: Make commands work remotely
The "secrets" class of commands currently only works on the host it's invoked on. Use the current FileContext to allow using the existing commands on a target host.
Signed-off-by: Jan Lindemann <jan@janware.com>
This commit is contained in:
parent
3cf5b2264e
commit
1b821f3b3f
5 changed files with 299 additions and 287 deletions
73
src/python/jw/pkg/cmds/secrets/lib/DistroContext.py
Normal file
73
src/python/jw/pkg/cmds/secrets/lib/DistroContext.py
Normal file
|
|
@ -0,0 +1,73 @@
|
|||
# -*- coding: utf-8 -*-
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
|
||||
from pathlib import Path
|
||||
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from typing import Iterable
|
||||
from ....lib.FileContext import FileContext
|
||||
from ...CmdDistro import CmdDistro
|
||||
|
||||
from ....lib.log import *
|
||||
from ....lib.util import run_cmd
|
||||
|
||||
from .base import Attrs
|
||||
from .FilesContext import FilesContext
|
||||
|
||||
class DistroContext(FilesContext):
|
||||
|
||||
def __init__(self, distro: Distro) -> None:
|
||||
super().__init__(distro.ctx)
|
||||
self.__distro = distro
|
||||
|
||||
async def match_files(self, packages: Iterable[str], pattern: str) -> list[str]:
|
||||
ret: list[str] = []
|
||||
for package_name in packages:
|
||||
for path in await self.__distro.pkg_files(package_name):
|
||||
if re.match(pattern, path):
|
||||
ret.append(path)
|
||||
return ret
|
||||
|
||||
async def list_template_files(self, packages: Iterable[str]) -> list[str]:
|
||||
return await self.match_files(packages, pattern=r'.*\.jw-tmpl$')
|
||||
|
||||
async def list_secret_paths(self, packages: Iterable[str], ignore_missing: bool=False) -> list[str]:
|
||||
ret = []
|
||||
for tmpl in await self.list_template_files(packages):
|
||||
path = str(Path(tmpl).with_suffix(".jw-secret"))
|
||||
if ignore_missing and not self.ctx.file_exists(path):
|
||||
continue
|
||||
ret.append(path)
|
||||
return ret
|
||||
|
||||
async def list_compilation_targets(self, packages: Iterable[str], ignore_missing: bool=False) -> list[str]:
|
||||
ret = []
|
||||
for tmpl in await self.list_template_files(packages):
|
||||
path = tmpl.removesuffix('.jw-tmpl')
|
||||
if ignore_missing and not self.ctx.file_exists(path):
|
||||
continue
|
||||
ret.append(path)
|
||||
return ret
|
||||
|
||||
async def remove_compilation_targets(self, packages: Iterable[str]) -> list[str]:
|
||||
for path in await self.list_compilation_targets(packages):
|
||||
try:
|
||||
self.ctx.stat(path)
|
||||
log(NOTICE, f'Removing {path}')
|
||||
await self.ctx.unlink(path)
|
||||
except FileNotFoundError as e:
|
||||
log(DEBUG, f'Compilation target {path} doesn\'t exist (ignored)')
|
||||
continue
|
||||
|
||||
async def compile_template_files(self, packages: Iterable[str], default_attrs: Attrs) -> list[str]:
|
||||
missing = 0
|
||||
for target in await self.list_compilation_targets(packages):
|
||||
if not await self.compile_template_file(target, default_attrs):
|
||||
missing += 1
|
||||
if missing > 0:
|
||||
log(WARNING, f'{missing} missing secrets found. You might want to add them and run sudo {app.cmdline} again')
|
||||
Loading…
Add table
Add a link
Reference in a new issue