stress_ssh_channels.py

#!/usr/bin/python3
import argparse
import paramiko.client
import time

parser = argparse.ArgumentParser(description='stress test SSH channels')
parser.add_argument('-H', '--host', help='host to test')
parser.add_argument('-u', '--user', help='user to use SSH with')
parser.add_argument('-p', '--pwd', help='password to use SSH with')
parser.add_argument(
    '-P', '--port', type=int, default=22, help='SSH port to use')
parser.add_argument(
    '-N',
    '--num-channels',
    type=int,
    default=22,
    help='password to use SSH with')
args = parser.parse_args()


def main():
    c = paramiko.client.SSHClient()
    c.set_missing_host_key_policy(paramiko.client.WarningPolicy)
    c.connect(
        args.host,
        port=args.port,
        username=args.user,
        password=args.pwd,
        look_for_keys=False,
        allow_agent=False)

    sessions = []
    for i in range(args.num_channels):
        sessions.append(c._transport.open_session())

    for s in sessions:
        s._event_pending()
        s.transport._send_user_message(make_exec_msg(s, "hostname"))

    time.sleep(3)

    for s in sessions:
        s._wait_for_event()


def make_exec_msg(chan, cmd):
    m = paramiko.Message()
    m.add_byte(paramiko.common.cMSG_CHANNEL_REQUEST)
    m.add_int(chan.remote_chanid)
    m.add_string("exec")
    m.add_boolean(True)
    m.add_string(cmd)
    return m

if __name__ == '__main__':
    main()