Skip to main content

Execution helpers for simplified usage of subprocess and ssh.

Project description

exec-helpers

https://github.com/python-useful-helpers/exec-helpers/workflows/Python%20package/badge.svg Documentation Status https://img.shields.io/pypi/v/exec-helpers.svg https://img.shields.io/pypi/pyversions/exec-helpers.svg https://img.shields.io/pypi/status/exec-helpers.svg https://img.shields.io/github/license/python-useful-helpers/exec-helpers.svg https://img.shields.io/badge/code%20style-black-000000.svg

Execution helpers for simplified usage of subprocess and ssh. Why another subprocess wrapper and why no clear paramiko?

Historically paramiko offers good ssh client, but with specific limitations: you can call command with timeout, but without receiving return code, or call command and wait for return code, but without timeout processing.

In the most cases, we are need just simple SSH client with comfortable API for calls, calls via SSH proxy and checking return code/stderr. This library offers this functionality with deadlock free polling and friendly result objects (with inline decoding of XML Element tree, YAML, JSON, binary or just strings). In addition this library offers the same API for subprocess calls, but with specific limitation: no parallel calls (for protection from race conditions).

Pros:

Python 3.8
Python 3.9
Python 3.10
Python 3.11
Python 3.12

This package includes:

  • SSHClient - historically the first one helper, which used for SSH connections. Several API calls for sFTP also presents.

  • SSHAuth - class for credentials storage. SSHClient does not store credentials as-is, but uses SSHAuth for it. Objects of this class can be copied between ssh connection objects, also it used for execute_through_host.

  • Subprocess - subprocess.Popen wrapper with timeouts, polling and almost the same API, as SSHClient (except specific flags, like cwd for subprocess and get_tty for ssh).

  • async_api.Subprocess - the same, as Subprocess helper, but works with asyncio. .. note:: for Windows ProactorEventLoop or another non-standard event loop should be used!

  • ExecResult - class for execution results storage. Contains exit code, stdout, stderr and getters for decoding as JSON, YAML, XML (and LXML) element tree, string, bytearray and brief strings (up to 7 lines).

  • ExitCodes - enumerator for standard Linux exit codes. BASH return codes (produced from signal codes) also available.

Installation

Standard: pip install exec-helpers Extras:

  • yaml - install PyYaml for yaml decoding (PyYAML is main decoder, ruamel.YAML also supported as fallback.)

  • xml - install defusedxml for safe XML parsing to xml.etree.ElementTree.Element.

  • lxml - install lxml for advanced XML parsing. Can be unsafe.

  • ALL_FORMATS (all-formats) - install all parsers. When new parsers will be added, it will ne also supported.

Usage

SSHClient

Basic initialization of SSHClient can be done without construction of specific objects:

client = exec_helpers.SSHClient(host, username="username", password="password")

If ssh agent is running - keys will be collected by paramiko automatically, but if keys are in specific location - it should be loaded manually and provided as iterable object of paramiko.PKey.

For advanced cases or re-use of credentials, SSHAuth object should be used. It can be collected from connection object via property auth.

Creation from scratch:

auth = exec_helpers.SSHAuth(
    username='username',  # str | None
    password='password',  # str | None
    key=None,  # type: paramiko.PKey | None
    keys=None,  # type: Iterable[paramiko.PKey] | None
    key_filename=None,  # type: list[str] | None
    passphrase=None,  # str | None
)

Key is a main connection key (always tried first) and keys are alternate keys. Key filename is a filename or list of filenames with keys, which should be loaded. Passphrase is an alternate password for keys, if it differs from main password. If main key now correct for username - alternate keys tried, if correct key found - it became main. If no working key - password is used and None is set as main key.

Context manager is available, connection is closed and lock is released on exit from context.

Subprocess

Context manager is available, subprocess is killed and lock is released on exit from context.

Base methods

Main methods are execute, check_call and check_stderr for simple executing, executing and checking return code and executing, checking return code and checking for empty stderr output. This methods are almost the same for SSHClient and Subprocess, except specific flags.

result: ExecResult = helper.execute(
    command,  # type: str | Iterable[str]
    verbose=False,  # type: bool
    timeout=1 * 60 * 60,  # type: int | float | None
    # Keyword only:
    log_mask_re=None,  # str | None
    stdin=None,  # type: bytes | str | bytearray | None
    open_stdout=True,  # type: bool
    log_stdout=True,  # type: bool
    open_stderr=True,  # type: bool
    log_stderr=True,  # type: bool
    **kwargs
)
result: ExecResult = helper.check_call(
    command,  # type: str | Iterable[str]
    verbose=False,  # type: bool
    timeout=1 * 60 * 60,  # type: type: int | float | None
    error_info=None,  # str | None
    expected=(0,),  # type: Iterable[int | ExitCodes]
    raise_on_err=True,  # type: bool
    # Keyword only:
    log_mask_re=None,  # str | None
    stdin=None,  # type: bytes | str | bytearray | None
    open_stdout=True,  # type: bool
    log_stdout=True,  # type: bool
    open_stderr=True,  # type: bool
    log_stderr=True,  # type: bool
    exception_class=CalledProcessError,  # type[CalledProcessError]
    **kwargs
)
result: ExecResult = helper.check_stderr(
    command,  # type: str | Iterable[str]
    verbose=False,  # type: bool
    timeout=1 * 60 * 60,  # type: type: int | float | None
    error_info=None,  # str | None
    raise_on_err=True,  # type: bool
    # Keyword only:
    expected=(0,),  # Iterable[int | ExitCodes]
    log_mask_re=None,  # str | None
    stdin=None,  # type: bytes | str | bytearray | None
    open_stdout=True,  # type: bool
    log_stdout=True,  # type: bool
    open_stderr=True,  # type: bool
    log_stderr=True,  # type: bool
    exception_class=CalledProcessError,  # type[CalledProcessError]
)
result: ExecResult = helper(  # Lazy way: instances are callable and uses `execute`.
    command,  # type: str | Iterable[str]
    verbose=False,  # type: bool
    timeout=1 * 60 * 60,  # type: int | float | None
    # Keyword only:
    log_mask_re=None,  # str | None
    stdin=None,  # type: bytes | str | bytearray | None
    open_stdout=True,  # type: bool
    log_stdout=True,  # type: bool
    open_stderr=True,  # type: bool
    log_stderr=True,  # type: bool
    **kwargs
)

If no STDOUT or STDERR required, it is possible to disable this FIFO pipes via **kwargs with flags open_stdout=False and open_stderr=False.

The next command level uses lower level and kwargs are forwarded, so expected exit codes are forwarded from check_stderr. Implementation specific flags are always set via kwargs.

If required to mask part of command from logging, log_mask_re attribute can be set global over instance or provided with command. All regex matched groups will be replaced by ‘<*masked*>’.

result: ExecResult = helper.execute(
    command="AUTH='top_secret_key'; run command",  # type: str | Iterable[str]
    verbose=False,  # type: bool
    timeout=1 * 60 * 60,  # type: Optional[int]
    log_mask_re=r"AUTH\s*=\s*'(\w+)'"  # str | None
)

result.cmd will be equal to AUTH=’<*masked*>’; run command

ExecResult

Execution result object has a set of useful properties:

  • cmd - Command

  • exit_code - Command return code. If possible to decode using enumerators for Linux -> it used.

  • ok -> bool. Command return code is 0 (EX_OK).

  • stdin -> str. Text representation of stdin.

  • stdout -> tuple[bytes]. Raw stdout output.

  • stderr -> tuple[bytes]. Raw stderr output.

  • stdout_bin -> bytearray. Binary stdout output.

  • stderr_bin -> bytearray. Binary stderr output.

  • stdout_str -> str. Text representation of output.

  • stderr_str -> str. Text representation of output.

  • stdout_brief -> str. Up to 7 lines from stdout (3 first and 3 last if >7 lines).

  • stderr_brief -> str. Up to 7 lines from stderr (3 first and 3 last if >7 lines).

  • stdout_json - STDOUT decoded as JSON.

  • stdout_yaml - STDOUT decoded as YAML. Accessible only if PyYAML or ruamel.YAML library installed. (Extras: yaml)

  • stdout_xml - STDOUT decoded as XML to ElementTree using defusedxml library. Accessible only if defusedxml library installed. (Extras: xml)

  • stdout_lxml - STDOUT decoded as XML to ElementTree using lxml library. Accessible only if lxml library installed. (Extras: lxml) Can be insecure.

  • timestamp -> Optional(datetime.datetime). Timestamp for received exit code.

SSHClient specific

SSHClient commands support get_pty flag, which enables PTY open on remote side. PTY width and height can be set via keyword arguments, dimensions in pixels are always 0x0.

Possible to call commands in parallel on multiple hosts if it’s not produce huge output:

results: dict[tuple[str, int], ExecResult] = SSHClient.execute_together(
    remotes,  # type: Iterable[SSHClient]
    command,  # type: str | Iterable[str]
    timeout=1 * 60 * 60,  # type: type: int | float | None
    expected=(0,),  # type: Iterable[int | ExitCodes]
    raise_on_err=True,  # type: bool
    # Keyword only:
    stdin=None,  # type: bytes | str | bytearray | None
    open_stdout=True,  # type: bool
    open_stderr=True,  # type: bool
    log_mask_re=None,  # str | None
    exception_class=ParallelCallProcessError  # type[ParallelCallProcessError]
)
results  # type: dict[tuple[str, int], exec_result.ExecResult]

Results is a dict with keys = (hostname, port) and and results in values. By default execute_together raises exception if unexpected return code on any remote.

To open new connection using current as proxy is accessible method proxy_to. Basic usage example:

conn: SSHClient = client.proxy_to(host, username="username", password="password")

For execute through SSH host can be used execute_through_host method:

result: ExecResult = client.execute_through_host(
    hostname,  # type: str
    command,  # type: str | Iterable[str]
    # Keyword only:
    auth=None,  # type: SSHAuth | None
    port=22,  # type: int
    timeout=1 * 60 * 60,  # type: type: int | float | None
    verbose=False,  # type: bool
    stdin=None,  # type: bytes | str | bytearray | None
    open_stdout=True,  # type: bool
    log_stdout=True,  # type: bool
    open_stderr=True,  # type: bool
    log_stderr=True,  # type: bool
    log_mask_re=None,  # str | None
    get_pty=False,  # type: bool
    width=80,  # type: int
    height=24  # type: int
)

Where hostname is a target hostname, auth is an alternate credentials for target host.

SSH client implements fast sudo support via context manager:

Commands will be run with sudo enforced independently from client settings for normal usage:

with client.sudo(enforce=True):
    ...

Commands will be run without sudo independently from client settings for normal usage:

with client.sudo(enforce=False):
    ...

“Permanent client setting”:

client.sudo_mode = mode  # where mode is True or False

SSH Client supports sFTP for working with remote files:

with client.open(path, mode='r') as f:
    ...

For fast remote paths checks available methods:

  • exists(path) -> bool

>>> conn.exists('/etc/passwd')
True
  • stat(path) -> paramiko.sftp_attr.SFTPAttributes

>>> conn.stat('/etc/passwd')
<SFTPAttributes: [ size=1882 uid=0 gid=0 mode=0o100644 atime=1521618061 mtime=1449733241 ]>
>>> str(conn.stat('/etc/passwd'))
'-rw-r--r--   1 0        0            1882 10 Dec 2015  ?'
  • isfile(path) -> bool

>>> conn.isfile('/etc/passwd')
True
  • isdir(path) -> bool

>>> conn.isdir('/etc/passwd')
False

Additional (non-standard) helpers:

  • mkdir(path: str) - execute mkdir -p path

  • rm_rf(path: str) - execute rm -rf path

  • upload(source: str, target: str) - upload file or from source to target using sFTP.

  • download(destination: str, target: str) - download file from target to destination using sFTP.

Subprocess specific

Keyword arguments:

  • cwd - working directory.

  • env - environment variables dict.

async_api.Subprocess specific

All standard methods are coroutines. Async context manager also available.

Example:

async with helper:
  result: ExecResult = await helper.execute(
      command,  # type: str | Iterable[str]
      verbose=False,  # type: bool
      timeout=1 * 60 * 60,  # type: int | float | None
      **kwargs
  )

Testing

The main test mechanism for the package exec-helpers is using tox. Available environments can be collected via tox -l

CI systems

For code checking several CI systems is used in parallel:

  1. GitHub actions: is used for checking: PEP8, pylint, bandit, installation possibility and unit tests.

Project details


Release history Release notifications | RSS feed

This version

8.1.2

Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

exec_helpers-8.1.2.tar.gz (70.7 kB view details)

Uploaded Source

Built Distribution

exec_helpers-8.1.2-py3-none-any.whl (76.1 kB view details)

Uploaded Python 3

File details

Details for the file exec_helpers-8.1.2.tar.gz.

File metadata

  • Download URL: exec_helpers-8.1.2.tar.gz
  • Upload date:
  • Size: 70.7 kB
  • Tags: Source
  • Uploaded using Trusted Publishing? Yes
  • Uploaded via: twine/5.1.1 CPython/3.12.5

File hashes

Hashes for exec_helpers-8.1.2.tar.gz
Algorithm Hash digest
SHA256 d72dcc310d5e7c4e3391f4353b15bdf3c44c0705c1e2dc4bf327591993c14685
MD5 54dfde5182dcbbfaef2c30f318f83a51
BLAKE2b-256 4ba241a8a1fa570fde9050d7c78ded98946b09eea0c40534ec143f033c747512

See more details on using hashes here.

File details

Details for the file exec_helpers-8.1.2-py3-none-any.whl.

File metadata

  • Download URL: exec_helpers-8.1.2-py3-none-any.whl
  • Upload date:
  • Size: 76.1 kB
  • Tags: Python 3
  • Uploaded using Trusted Publishing? Yes
  • Uploaded via: twine/5.1.1 CPython/3.12.5

File hashes

Hashes for exec_helpers-8.1.2-py3-none-any.whl
Algorithm Hash digest
SHA256 9167b20278094a7b120df8af9c1ceb84b3bd23684c7fd40391ed48aeba98d6ab
MD5 7892428f348c066ce90b0de517cab243
BLAKE2b-256 7a11d6c862db230c6d40f23af839f1611d61d53a071ae0eb464eaa558de57a03

See more details on using hashes here.

Supported by

AWS AWS Cloud computing and Security Sponsor Datadog Datadog Monitoring Fastly Fastly CDN Google Google Download Analytics Microsoft Microsoft PSF Sponsor Pingdom Pingdom Monitoring Sentry Sentry Error logging StatusPage StatusPage Status page