_subprocess_run_as_user

Run simple commands as another user

This module provides a set of tools similar to subprocess.run, glob.glob or os.path.exists, but extended so that they can be executed on behalf of another user. Note that this requires appropriate sudo permissions.

_mkdir_as_user(*, folder, user)

Create a folder as a different user

PARAMETER	DESCRIPTION
folder	Absolute path to the folder TYPE: str
user	User to be impersonated TYPE: str

RAISES	DESCRIPTION
RuntimeError	if user is not correctly defined, or if subprocess returncode is not 0.

Source code in fractal_server/runner/executors/slurm_sudo/_subprocess_run_as_user.py

def _mkdir_as_user(*, folder: str, user: str) -> None:
    """
    Create a folder as a different user

    Args:
        folder: Absolute path to the folder
        user: User to be impersonated

    Raises:
        RuntimeError: if `user` is not correctly defined, or if subprocess
                      returncode is not 0.
    """
    if not user:
        raise RuntimeError(f"{user=} not allowed in _mkdir_as_user")

    cmd = f"mkdir -p {folder}"
    _run_command_as_user(cmd=cmd, user=user, check=True)

_run_command_as_user(*, cmd, user=None, check=False)

Use sudo -u to impersonate another user and run a command

PARAMETER	DESCRIPTION
cmd	Command to be run TYPE: str
user	User to be impersonated TYPE: str | None DEFAULT: None
check	If True, check that returncode=0 and fail otherwise. TYPE: bool DEFAULT: False

RAISES	DESCRIPTION
RuntimeError	if check=True and returncode is non-zero.

RETURNS	DESCRIPTION
res	The return value from subprocess.run. TYPE: CompletedProcess

Source code in fractal_server/runner/executors/slurm_sudo/_subprocess_run_as_user.py

def _run_command_as_user(
    *,
    cmd: str,
    user: str | None = None,
    check: bool = False,
) -> subprocess.CompletedProcess:
    """
    Use `sudo -u` to impersonate another user and run a command

    Args:
        cmd: Command to be run
        user: User to be impersonated
        check: If `True`, check that `returncode=0` and fail otherwise.

    Raises:
        RuntimeError: if `check=True` and returncode is non-zero.

    Returns:
        res: The return value from `subprocess.run`.
    """
    validate_cmd(cmd)
    logger.debug(f'[_run_command_as_user] {user=}, cmd="{cmd}"')
    if user:
        new_cmd = f"sudo --set-home --non-interactive -u {user} {cmd}"
    else:
        new_cmd = cmd
    res = subprocess.run(  # nosec
        shlex.split(new_cmd),
        capture_output=True,
        encoding="utf-8",
    )
    logger.debug(f"[_run_command_as_user] {res.returncode=}")
    logger.debug(f"[_run_command_as_user] {res.stdout=}")
    logger.debug(f"[_run_command_as_user] {res.stderr=}")

    if check and not res.returncode == 0:
        raise RuntimeError(
            f"{cmd=}\n\n{res.returncode=}\n\n{res.stdout=}\n\n{res.stderr=}\n"
        )

    return res