ssh scanner

ssh_scanner.py

from argparse import ArgumentParser
from concurrent.futures import ThreadPoolExecutor
from ipaddress import ip_network
import socket

try:
    from paramiko import SSHClient
except ModuleNotFoundError:
    from subprocess import run
    import sys

    run(f"{sys.executable} -m pip install paramiko".split(" "))
    import site

    site.main()

import paramiko


def is_open(ip: str, port: int):
    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
        s.settimeout(1)
        res = s.connect_ex((ip, port))
        port_status = True if res == 0 else False
        return dict(ip=ip, port=port, status=port_status)


def scan_port(ip: str, port: int):
    with ThreadPoolExecutor(128) as e:
        futures = []
        for host in ip_network(ip).hosts():
            f = e.submit(is_open, host.compressed, port)
            futures.append(f)

        status = []
        for f in futures:
            res = f.result()
            status.append(res)

    return status


def check_ssh_connection(ip: str, port: int, user: str, passwd: str):
    if is_open(ip, port):
        status = False
        with paramiko.SSHClient() as ssh_client:
            try:
                ssh_client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
                ssh_client.connect(hostname=ip, port=port, username=user, password=passwd)
            except:
                status = False
            else:
                status = True

    return dict(ip=ip, port=port, status=status)


def scan_ssh(ip: str, port: int, user: str, password: str):
    with ThreadPoolExecutor(128) as e:
        futures = []
        for host in ip_network(ip).hosts():
            f = e.submit(check_ssh_connection, host.compressed, port, user, password)
            futures.append(f)

        status = []
        for f in futures:
            res = f.result()
            status.append(res)

    return status


if __name__ == '__main__':
    parser = ArgumentParser()
    parser.add_argument("--ip", type=str)
    parser.add_argument("--port", type=int)
    parser.add_argument("--user", type=str)
    parser.add_argument("--password", type=str)

    args = parser.parse_args()

    results = scan_ssh(args.ip, args.port, args.user, args.password)
    row = "{:<15} {:<8} {:<2}"
    print(row.format("IP", "PORT", "STATUS"))
    for res in results:
        if res["status"]:
            print(row.format(res["ip"], res["port"], "open" if res["status"] else "closed"))