janware/jw-pkg
1
mirror of ssh://git.janware.com/janware/proj/jw-pkg synced 2026-04-25 09:35:54 +02:00
Code Activity Actions

lib.util: Add ec: ExecContext to all subprocesses

Browse source 
Allow to pass an optional execution context to all functions spawning
a subprocess defined in lib.util.

Signed-off-by: Jan Lindemann <jan@janware.com>
This commit is contained in:
Jan Lindemann 2026-03-06 11:54:27 +01:00
commit 7fdba1b5db
1 changed files with 17 additions and 15 deletions
Download patch file Download diff file Expand all files Collapse all files

32
src/python/jw/pkg/lib/util.py
View file

@ -33,16 +33,17 @@ def pretty_cmd(cmd: list[str], wd=None):
# See ec.Local.run() for what this function does
async def run_cmd(*args, ec: ExecContext|None=None, **kwargs) -> tuple[str|bytes|None, str|bytes|None]:
if ec is not None:
if ec is None:
from .ec.Local import Local
ec = Local()
return await ec.run(*args, **kwargs)
async def run_curl(args: list[str], parse_json: bool=True, wd=None, throw=None, verbose=False, cmd_input=None) -> dict|str: # export
async def run_curl(args: list[str], parse_json: bool=True, wd=None, throw=None, verbose=False, cmd_input=None, ec: ExecContext|None=None) -> dict|str: # export
cmd = ['curl']
if not verbose:
cmd.append('-s')
cmd.extend(args)
ret, stderr, status = await run_cmd(cmd, wd=wd, throw=throw, verbose=verbose, cmd_input=cmd_input)
ret, stderr, status = await run_cmd(cmd, wd=wd, throw=throw, verbose=verbose, cmd_input=cmd_input, ec=ec)
if parse_json:
try:
ret = json.loads(ret)
@ -57,7 +58,7 @@ async def run_curl(args: list[str], parse_json: bool=True, wd=None, throw=None,
raise
return ret, stderr, status
async def run_askpass(askpass_env: list[str], key: AskpassKey, host: str|None=None):
async def run_askpass(askpass_env: list[str], key: AskpassKey, host: str|None=None, ec: ExecContext|None=None):
assert host is None # Currently unsupported
for var in askpass_env:
exe = os.getenv(var)
@ -77,17 +78,18 @@ async def run_askpass(askpass_env: list[str], key: AskpassKey, host: str|None=No
continue # Can't get user name from SSH_ASKPASS
case AskpassKey.Password:
exe_arg += 'Password'
ret, stderr, status = await run_cmd([exe, exe_arg], throw=False)
ret, stderr, status = await run_cmd([exe, exe_arg], throw=False, ec=ec)
if ret is not None:
return ret
return None
async def run_sudo(cmd: list[str], mod_env: dict[str, str] = {}, opts: list[str]=[], interactive: bool=True, verbose=True):
from .ec.Local import Local
ec = Local()
return await ec.sudo(cmd, mod_env, opts, interactive, verbose)
async def run_sudo(cmd: list[str], mod_env: dict[str, str] = {}, opts: list[str]=[], interactive: bool=True, verbose=True, ec: ExecContext|None=None):
if ec is None:
from .ec.Local import Local
ec = Local(interactive=interactive)
return await ec.sudo(cmd=cmd, mod_env=mod_env, opts=opts, verbose=verbose)
async def get_username(args: Namespace|None=None, url: str|None=None, askpass_env: list[str]=[]) -> str: # export
async def get_username(args: Namespace|None=None, url: str|None=None, askpass_env: list[str]=[], ec: ExecContext|None=None) -> str: # export
url_user = None if url is None else urlparse(url).username
if args is not None:
if args.username is not None:
@ -96,9 +98,9 @@ async def get_username(args: Namespace|None=None, url: str|None=None, askpass_en
return args.username
if url_user is not None:
return url_user
return await run_askpass(askpass_env, AskpassKey.Username)
return await run_askpass(askpass_env, AskpassKey.Username, ec=ec)
async def get_password(args: Namespace|None=None, url: str|None=None, askpass_env: list[str]=[]) -> str: # export
async def get_password(args: Namespace|None=None, url: str|None=None, askpass_env: list[str]=[], ec: ExecContext|None=None) -> str: # export
if args is None and url is None and not askpass_env:
raise Exception(f'Neither URL nor command-line arguments nor askpass environment variable available, can\'t get password')
if args is not None and hasattr(args, 'password'): # use getattr(), because we don't necessarily want to have insecure --password among options
@ -109,9 +111,9 @@ async def get_password(args: Namespace|None=None, url: str|None=None, askpass_en
parsed = urlparse(url)
if parsed.password is not None:
return parsed.password
return await run_askpass(askpass_env, AskpassKey.Password)
return await run_askpass(askpass_env, AskpassKey.Password, ec=ec)
async def get_profile_env(throw: bool=True, keep: Iterable[str]|bool=False) -> dict[str, str]: # export
async def get_profile_env(throw: bool=True, keep: Iterable[str]|bool=False, ec: ExecContext|None=None) -> dict[str, str]: # export
"""
Get a fresh environment from /etc/profile
@ -133,7 +135,7 @@ async def get_profile_env(throw: bool=True, keep: Iterable[str]|bool=False) -> d
}
# Run bash as a login shell, which sources /etc/profile, then print environment as NUL-separated key=value pairs
cmd = ['/usr/bin/env', '-i', '/bin/bash', '-lc', 'env -0']
stdout, stderr, status = await run_cmd(cmd, throw=throw, output_encoding="bytes", verbose=True, env=env)
stdout, stderr, status = await run_cmd(cmd, throw=throw, output_encoding="bytes", verbose=True, env=env, ec=ec)
ret: dict[str, str] = {}
for entry in stdout.split(b"\0"):
if not entry: