ssh-handler 1.4.1

Extensive SSH/SFTP/SCP/FTP handler built on Paramiko, for test automation, CLIs and PyQt5 tools.

ssh-handler

An extensive SSH / SFTP / SCP / FTP automation handler built on Paramiko. One package, three ways to use it:

• Test-automation framework — raise-on-error API, pytest fixtures, parallel fleet ops.
• Standalone CLI — python -m ssh_handler …, fully argument-driven.
• PyQt5 tool — safe mode + log streaming over Qt signals, runs off the GUI thread.

Passwords are wrapped in a Secret and stored in the OS credential vault — they never appear in logs, reprs, tracebacks, or on disk in plaintext.

pip install ssh-handler

---

Table of contents

• Why this package
• Install
• Quick start
• What you can do — full capability list
• Domain / Windows (RDP) hosts
• Confidential credentials
• Performance
• Error handling: two styles
• Result objects
• CLI reference
• PyQt5 integration
• Parallel fleet operations
• FTP / FTPS
• API map
• Releasing

---

Why this package

Paramiko is powerful but low-level: you manage clients, transports, channels, SFTP sessions, timeouts, retries, host-key policies and error handling yourself, and you repeat that boilerplate in every project. ssh-handler wraps all of it behind one object that:

• auto-selects the right authentication strategy (password, key, agent, empty password),
• retries connections and transparently reconnects dropped sessions,
• returns structured results for every action instead of raw strings,
• keeps passwords confidential end-to-end,
• and exposes the same surface whether you're in a test, a CLI, or a GUI.

Install

pip install ssh-handler           # everything: SSH, SFTP, SCP, FTP, serial,
                                  # credential vault, WinRM bootstrap
pip install "ssh-handler[gui]"    # also installs PyQt5 for the GUI worker

Batteries included. A plain pip install ssh-handler pulls in paramiko, scp, pyserial, keyring, and pywinrm, so SSH, SFTP/SCP/FTP transfers, serial/COM ports, confidential credential storage, and the WinRM bootstrap all work out of the box. Only PyQt5 is optional ([gui]), because it's a large GUI toolkit you only need when building a GUI — forcing it would bloat headless and CI installs.

Quick start

from ssh_handler import SSHHandler, SSHConfig

with SSHHandler(SSHConfig(host="10.0.0.5", username="root", password="pw")) as ssh:
    print(ssh.run("uptime").stdout)
    ssh.run("systemctl restart nginx", check=True)      # raises on non-zero exit
    ssh.push("local.txt", "/tmp/remote.txt")            # SFTP upload
    ssh.pull("/etc/nginx", "./backup", recursive=True)  # recursive download

What you can do

Connection & session

• Connect with password, private key (+ passphrase), SSH agent, auto-discovered keys, or an empty-password account — all auto-tried in a smart order.
• Auto-retry connects with backoff; auto-reconnect if a session drops.
• Keepalives, per-command and connection timeouts, optional compression.
• Jump host / bastion chaining (ProxyJump-style) via SSHConfig(jump_host=…).
• Host-key policy: auto / reject / warn, with an optional known_hosts file.
• Remote-OS awareness (detect_os(), is_windows) for Linux and Windows targets.
• Raw escape hatch: ssh.client and ssh.transport expose the underlying Paramiko objects.

Command execution

• run() — timeout, check (raise on non-zero), PTY allocation, custom environment.
• run_many() — batch with stop-on-error.
• sudo() — runs sudo -S and feeds the password on stdin.
• open_shell() — a persistent interactive ShellSession with send / read_until (send-expect) / read_available.

Continuous / streaming output (logs)

• iter_lines(cmd) — generator yielding a never-ending command's output line by line, live (slog2info -w, tail -f, journalctl -f, dmesg -w).
• stream(cmd, on_line=, match=, save_to=, stop_on_match=, stop_event=) — stream with live regex matching, a per-line/per-match callback, and tee to a local file, all built in.

Serial / COM ports (SerialHandler, included by default)

• list_serial_ports(), open/close, write / write_line.
• iter_lines() and stream(...) — same live streaming + match + save-to-file model as SSH, for device consoles.

File operations (SFTP) — full Paramiko parity

• Transfers: push / pull (single file or recursive directory, with progress callbacks and transfer statistics), plus scp_push / scp_pull (SCP protocol).
• Listing & metadata: listdir, listdir_attr, stat, lstat, exists, isdir, walk.
• Directories: mkdir, makedirs (recursive mkdir -p), rmdir.
• Files: remove, rename, open (remote file object), read_text, write_text.
• Permissions & links: chmod, chown, symlink, readlink.

Other protocols

• FTP / FTPS via FTPHandler (standard-library ftplib, no extra dependency): connect, login, TLS, push, pull, listdir, cwd, pwd, mkdir, rmdir, remove, rename, size, exists.

Scale & integration

• SSHPool — run the same command/transfer across many hosts in parallel threads.
• Safe mode + log callback for GUIs; structured result objects everywhere.
• Confidential credential storage in the OS vault (CredentialStore, Secret, mask).

Domain / Windows (RDP) hosts

The Windows machines you normally RDP into can be driven over SSH once OpenSSH Server is enabled on them:

Add-WindowsCapability -Online -Name OpenSSH.Server~~~~0.0.1.0
Start-Service sshd ; Set-Service -Name sshd -StartupType Automatic

Then log in with your normal domain credentials. Pass domain and username separately — never hard-code a single "DOMAIN\user" Python string, because a backslash escape (e.g. \n, \t) silently becomes a control character. The handler builds the DOMAIN\user login string for you:

from ssh_handler import SSHHandler, SSHConfig, CredentialStore

store = CredentialStore(service="my_test_lab")
cfg = SSHConfig(
    host="10.20.30.40",
    domain="CORP", username="myuser",        # -> login "CORP\myuser"
    password=store.get("CORP\\myuser"),       # a Secret pulled from the OS vault
    remote_os="windows",                      # skip the OS probe
    fast_auth=True,                           # skip key probing -> faster login
)
with SSHHandler(cfg) as ssh:
    print(ssh.run("whoami").stdout)                       # CORP\myuser
    print(ssh.run("powershell Get-Service sshd").stdout)
    ssh.push("report.xlsx", "C:/Users/myuser/Desktop/report.xlsx")

Store the password once so it never appears in code again:

from ssh_handler import CredentialStore, prompt_password
CredentialStore("my_test_lab").set("CORP\\myuser", prompt_password())

# …or from the CLI:
python -m ssh_handler store-credential --user myuser --domain CORP --service my_test_lab

RDP-only Windows hosts: auto-enable SSH once (WinRM bootstrap)

A freshly imaged corporate Windows box often has RDP and WinRM open but no SSH server (port 22 closed). You can't start sshd over SSH when SSH is down — but if WinRM is reachable, the handler can use it as a one-time bootstrap channel.

Set one flag and connect normally:

from ssh_handler import SSHHandler, SSHConfig

cfg = SSHConfig(
    host="10.232.9.22", domain="CORP", username="myuser", password="pw",
    auto_bootstrap_via_winrm=True,   # if SSH is down but WinRM is up, enable sshd, then retry
)
with SSHHandler(cfg) as ssh:         # 1st run: enables sshd over WinRM, then connects
    print(ssh.run("whoami").stdout)  # every later run: connects straight over SSH

It's genuinely one-time. The bootstrap installs the OpenSSH Server capability, starts sshd with Automatic startup, and adds a persistent firewall rule — so it survives reboots. After the first run, port 22 is already open and the handler connects directly over SSH; WinRM is never touched again.

Do it explicitly instead of automatically if you prefer:

ssh = SSHHandler(cfg)
ssh.bootstrap_sshd_via_winrm()       # one-time setup
ssh.connect()

Requirements: pip install "ssh-handler[winrm]" (pulls in pywinrm; uses NTLM so domain creds work without Kerberos), and the account must be a local administrator on the target. If SSH already works, this code path never runs.

Set up OpenSSH Server on a machine (bundled installer)

Installing the package also gives you a one-command, fully offline setup for the local Windows machine. The OpenSSH ZIPs (ARM64 / Win64 / Win32) ship inside the wheel, so the installer needs no internet and no Windows Update — it picks the ZIP matching the CPU architecture, self-elevates to Administrator, installs & starts OpenSSH Server, and opens the firewall. (This avoids the Add-WindowsCapability hang common on locked-down corporate networks.)

pip install ssh-handler
ssh-handler-setup                 # offline install + start sshd + firewall (self-elevates)
ssh-handler-setup --install-pip   # also (re)install the package as admin
ssh-handler-setup --force         # reinstall even if sshd already exists
# equivalent: python -m ssh_handler setup-server

Run it on whichever machine you want to reach over SSH (e.g. the RDP jump box).

When a connection just fails, the error now self-diagnoses — it probes the SSH and RDP ports and tells you why (e.g. "Port 22 is closed but RDP (3389) is open … no SSH server listening"). Call ssh.diagnose() for a pre-flight reachability check.

Continuous logs & live pattern matching

Stream a long-running remote command and react to lines as they arrive — match a pattern, save to a file, or stop when something appears. Works through the jump host too.

from ssh_handler import SSHHandler, SSHConfig

with SSHHandler(SSHConfig(host="10.120.1.91", username="root", password="pw",
                          jump_host=rdp_box), quiet=True) as ssh:

    # (a) simplest: iterate lines live
    for line in ssh.iter_lines("slog2info -w"):
        print(line)
        if "FATAL" in line:
            break

    # (b) full: match + tee to a local file + callback, stop on a pattern
    result = ssh.stream(
        "tail -f /var/log/messages",
        on_line=print,                 # called for every line
        match=r"error|fail",           # regex; matching lines collected
        on_match=lambda l: print("HIT:", l),
        save_to="device.log",          # tee every line to this local file
        stop_on_match=False,           # set True to stop at the first match
        timeout=60,                    # optional overall time limit
    )
    print(result["lines"], "lines,", len(result["matches"]), "matched")

To stop a stream from another thread (e.g. a GUI Stop button), pass a threading.Event as stop_event= and .set() it.

Serial / COM ports

Same streaming + match + save model for device serial consoles (included by default — no extra install).

Local serial (device plugged into the machine running the code)

from ssh_handler import SerialHandler, list_serial_ports
print(list_serial_ports())
with SerialHandler("COM5", baudrate=115200, quiet=True) as ser:
    ser.write_line("version")
    ser.stream(on_line=print, match=r"login:", stop_on_match=True, save_to="console.log")

Serial via RDP / SSH (port on a remote machine)

pyserial only opens a local port, so when the serial port is on a remote machine, stream it over SSH with serial_stream() — same live match + save. It auto-detects the OS from the device name: COM* → Windows (PowerShell SerialPort reader), /dev/tty* → Linux (stty + cat).

Windows COM port on the remote machine (connect SSH straight to that machine — it has sshd from ssh-handler-setup):

cfg = SSHConfig(host="10.232.9.22", domain="CORP", username="myuser",
                password="pw", host_key_policy="ignore")
with SSHHandler(cfg, quiet=True) as ssh:
    ssh.serial_write("COM5", "version", baudrate=115200)      # write a line
    ssh.serial_stream("COM5", baudrate=115200,                # read it live
                      on_line=print, match=r"login:|ERROR",
                      save_to="com5.log", timeout=120)

Linux device file on a target reached through the jump:

target = SSHConfig(host="10.120.1.91", username="root", password="pw",
                   jump_host=rdp_box, host_key_policy="ignore")
with SSHHandler(target, quiet=True) as ssh:
    ssh.serial_stream("/dev/ttyUSB0", baudrate=115200,
                      on_line=print, match=r"login:", save_to="ttyusb0.log")

Note: on Windows a COM port can't be shared — don't run serial_write while a serial_stream on the same port is open (serial_write opens/writes/closes). If the port is on your own laptop, use the local SerialHandler("COM5") above instead — no SSH needed.

File transfer (SFTP / SCP / FTP) via RDP

SFTP and SCP already work through the jump host — no special setup. Once you pass jump_host=, every transfer runs over that tunnel (laptop → RDP → target):

with SSHHandler(target, quiet=True) as ssh:        # target has jump_host=rdp_box
    ssh.push("firmware.bin", "/tmp/firmware.bin")  # SFTP, through the jump
    ssh.pull("/var/log/messages", "messages.log")  # SFTP, through the jump
    ssh.scp_push("img.tar", "/tmp/img.tar")        # SCP, through the jump
    print(ssh.read_text("/etc/os-release"))

FTP via RDP: FTP is a separate protocol (its data channel can't ride an SSH tunnel cleanly), so prefer SFTP through the jump as shown above — it does the same job better and is already routed via RDP. If you specifically need a real FTP server on the target, run FTPHandler on the RDP machine itself (where it can reach that server directly).

from ssh_handler import SerialHandler, list_serial_ports

print(list_serial_ports())            # [{'device':'COM5','description':...}, ...]

with SerialHandler("COM5", baudrate=115200, quiet=True) as ser:
    ser.write_line("version")                       # send a command
    res = ser.stream(
        on_line=print,
        match=r"login:",                            # wait for the login prompt
        stop_on_match=True,
        save_to="serial_console.log",               # tee to file
        timeout=120,
    )
    print("matched:", res["matched"])

write_line(..., eol="\r\n") for consoles that need CRLF. Everything returns the same OperationResult in safe mode and raises SerialError otherwise.

Confidential credentials

Mechanism	What it does
Secret	wraps a password; str()/repr()/logs show ********; only .reveal() exposes it
mask()	redacts secret values from any string (applied automatically to all logging)
CredentialStore	stores/reads passwords in the OS vault (Windows Credential Manager / macOS Keychain / Secret Service) via keyring — no plaintext on disk
prompt_password()	hidden terminal input, returns a Secret

Pass a Secret (or a plain string, which is wrapped automatically) anywhere a password is accepted. It stays redacted across the whole stack.

Performance

• fast_auth (default on): when a password is supplied, the slow key/agent probing is skipped — faster logins and no "Too many authentication failures" from the server's MaxAuthTries.
• One SFTP channel is opened lazily and reused across operations.
• SFTP downloads use Paramiko prefetch for high throughput.
• remote_os="windows"|"linux" skips the one-time OS-detection probe.
• compress=True for slow/high-latency links; keepalives keep long sessions alive.
• SSHPool parallelizes across hosts with a thread pool.

Error handling: two styles

Raise (default) — best for tests/scripts. Typed exceptions, all subclasses of SSHError:

SSHConnectionError  SSHAuthenticationError  SSHTimeoutError
SSHCommandError     SSHTransferError        SSHNotConnectedError
FTPError            CredentialError

Safe mode (SSHHandler(cfg, safe=True)) — best for GUIs. Every call returns an OperationResult instead of raising, so an event loop never dies:

res = ssh.connect()
if not res:                 # OperationResult is falsy on failure
    show_error(res.error)
else:
    data = res.value        # or res.unwrap() to re-raise on failure

Override per call with safe=True / safe=False.

Result objects

Every action returns structured data, not bare strings:

• CommandResult — exit_code, stdout, stderr, duration, host, .ok, .as_dict()
• TransferResult — size_bytes, duration, speed_bps, human_speed, human_size, files
• ShellResult — output, matched, timed_out, duration
• OperationResult — safe-mode wrapper: bool(res), res.value, res.error, res.unwrap()

CLI reference

python -m ssh_handler run    --host H --user U --domain CORP uptime
python -m ssh_handler push   --host H --user U ./build /tmp/build --recursive
python -m ssh_handler pull   --host H --user U /var/log ./logs --recursive
python -m ssh_handler info   --host H --user U --json
python -m ssh_handler store-credential --user U --domain CORP --service my_test_lab

# continuous logs over SSH, with live matching + save:
python -m ssh_handler stream --host H --user U --match "error|fail" \
                             --save run.log -- slog2info -w

# serial / COM ports:
python -m ssh_handler list-serial
python -m ssh_handler serial-monitor --port COM5 --baud 115200 \
                             --match "login:" --stop-on-match --save console.log

# install OpenSSH Server on THIS Windows machine (offline, self-elevates):
ssh-handler-setup

Password options: --password (hidden prompt), --use-stored (read from the OS vault), --key FILE (private key). Add --json for machine-readable output. Put --match/--save before the streamed command. After pip install, the ssh-handler and ssh-handler-setup console scripts are also available.

PyQt5 integration

ssh_handler.pyqt_worker.SSHWorker is a QObject wrapping the handler in safe mode. Move it to a QThread, connect its signals, and drive it from the GUI:

from PyQt5.QtCore import QThread
from ssh_handler import SSHConfig
from ssh_handler.pyqt_worker import SSHWorker

worker = SSHWorker(SSHConfig(host="10.0.0.5", username="root", password=secret))
thread = QThread(); worker.moveToThread(thread)
worker.log.connect(text_edit.append)            # live, secret-masked log lines
worker.command_done.connect(on_command_done)
worker.progress.connect(progress_bar.setValue)  # bytes_done, bytes_total
thread.started.connect(lambda: worker.run_command("uptime"))
thread.start()

Signals: log, connected, command_done, transfer_done, progress, stream_line, stream_match, stream_done, error, finished. The import is lazy, so the rest of the package works where PyQt5 isn't installed.

Streaming logs into the GUI — drive start_stream in the worker thread and wire the per-line signals to your widgets; stop_stream() ends it cleanly:

worker.stream_line.connect(log_view.append)         # every live line
worker.stream_match.connect(lambda l: alerts.append(l))   # only matching lines
thread.started.connect(lambda: worker.start_stream("slog2info -w",
                                                    match="error|fail",
                                                    save_to="device.log"))
# later, from a Stop button:
worker.stop_stream()

Parallel fleet operations

from ssh_handler import SSHPool, SSHConfig

configs = [SSHConfig(host=h, username="root", password="pw")
           for h in ("10.0.0.1", "10.0.0.2", "10.0.0.3")]

with SSHPool(configs, max_workers=8) as pool:
    for host, res in pool.run("uptime").items():
        print(host, res.value.stdout.strip() if res else res.error)
    pool.pull("/var/log/syslog", "logs/{host}_syslog.txt")   # {host} avoids collisions

FTP / FTPS

from ssh_handler import FTPHandler, FTPConfig

with FTPHandler(FTPConfig(host="ftp.example.com", username="u",
                          password="p", use_tls=True)) as ftp:
    ftp.push("local.txt", "remote.txt")
    ftp.pull("remote.txt", "copy.txt")
    print(ftp.listdir("/"))

API map

ssh_handler/
  config.py       SSHConfig, FTPConfig
  credentials.py  Secret, CredentialStore, mask, prompt_password
  core.py         SSHHandler, ShellSession   (SSH + SFTP + SCP + stream + diagnose)
  ftp.py          FTPHandler                 (FTP / FTPS)
  serial_handler.py   SerialHandler, list_serial_ports   (serial / COM ports)
  winrm_bootstrap.py  enable_openssh_via_winrm   (one-time sshd enable over WinRM)
  pool.py         SSHPool                    (parallel multi-host)
  cli.py          argparse entry point       (python -m ssh_handler / ssh-handler)
  pyqt_worker.py  SSHWorker                  (PyQt5, lazy import)
  results.py      CommandResult, TransferResult, ShellResult, OperationResult
  exceptions.py   SSHError hierarchy
examples/examples.py        copy-paste recipes
tests/test_offline.py       offline checks (no network needed)

Releasing

Maintainers: use the helper to build and publish a new version.

python scripts/release.py 1.0.1            # bump -> build -> twine check -> upload
python scripts/release.py 1.0.1 --dry-run  # build + check only, no upload
python scripts/release.py patch            # auto-bump patch/minor/major

The token is read from the TWINE_PASSWORD environment variable (username __token__), never hard-coded. See scripts/release.py and the optional GitHub Actions workflow (publishes on a v* tag). PyPI permanently forbids re-uploading an existing version, so each release must use a new version number.

License

MIT