ssh-run.py

#!/usr/bin/env python3
"""Run one command on one or more SSH hosts without invoking a local shell."""

import argparse
import concurrent.futures
import subprocess
import sys


def parse_args():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("hosts", nargs="+", help="USER@HOST targets")
    parser.add_argument("--command", required=True, help="Command interpreted by the remote login shell")
    parser.add_argument("--identity", help="SSH private-key path")
    parser.add_argument("--port", type=int, default=22)
    parser.add_argument("--timeout", type=int, default=10)
    parser.add_argument("--parallel", type=int, default=4)
    return parser.parse_args()


def run(host, args):
    command = [
        "ssh",
        "-o", "BatchMode=yes",
        "-o", f"ConnectTimeout={args.timeout}",
        "-o", "StrictHostKeyChecking=yes",
        "-p", str(args.port),
    ]
    if args.identity:
        command.extend(["-i", args.identity])
    command.extend(["--", host, args.command])
    result = subprocess.run(command, text=True, capture_output=True, check=False)
    return host, result


def main():
    args = parse_args()
    if not 1 <= args.port <= 65535 or args.timeout < 1 or args.parallel < 1:
        raise SystemExit("Port, timeout, and parallelism must be positive and valid")
    status = 0
    workers = min(args.parallel, len(args.hosts))
    with concurrent.futures.ThreadPoolExecutor(max_workers=workers) as executor:
        futures = [executor.submit(run, host, args) for host in args.hosts]
        for future in concurrent.futures.as_completed(futures):
            host, result = future.result()
            print(f"== {host} (exit {result.returncode}) ==")
            if result.stdout:
                print(result.stdout, end="" if result.stdout.endswith("\n") else "\n")
            if result.stderr:
                print(result.stderr, file=sys.stderr, end="" if result.stderr.endswith("\n") else "\n")
            status = max(status, 1 if result.returncode else 0)
    return status


if __name__ == "__main__":
    raise SystemExit(main())