sshscan.py

#!/usr/bin/env python3
# SPDX-License-Identifier: MIT
# Copyright (c) 2017 Vincent Ruijter
# Copyright (c) 2020-2025 Babak Farrokhi
#
# Algorithm classifications based on algorithm_guidance.json

import argparse
import os
import socket
import struct
import sys
from typing import Any, Optional, Tuple, List, Dict


# Terminal Colors
class TerminalColors:
    """ANSI color codes for terminal output."""
    RED = '\033[91m'
    GREEN = '\033[92m'
    RESET = '\033[0m'


def supports_color() -> bool:
    """
    Detect if the terminal supports color output.

    Checks for:
    - NO_COLOR environment variable (https://no-color.org/)
    - FORCE_COLOR environment variable
    - Output is to a TTY (not redirected)
    - TERM environment variable is set and not 'dumb'
    """
    if os.environ.get('NO_COLOR'):
        return False

    if os.environ.get('FORCE_COLOR'):
        return True

    if not hasattr(sys.stdout, 'isatty') or not sys.stdout.isatty():
        return False

    term = os.environ.get('TERM', '')
    if term == 'dumb':
        return False

    return True


# Global flag to control color output
USE_COLOR = supports_color()


def colorize(text: str, color: str) -> str:
    """Apply color to text if colors are enabled."""
    if USE_COLOR:
        return f"{color}{text}{TerminalColors.RESET}"
    return text


def red(text: str) -> str:
    """Return text in red (for not recommended or insecure algorithms)."""
    return colorize(text, TerminalColors.RED)


def green(text: str) -> str:
    """Return text in green (for strong/secure algorithms)."""
    return colorize(text, TerminalColors.GREEN)


# SSH Protocol Constants
SSH_MSG_KEXINIT = 20
MAX_PACKET_LENGTH = 1024 * 1024
SSH_HEADER_LENGTH = 5
KEXINIT_COOKIE_LENGTH = 16
VERSION_STRING_MAX_LENGTH = 255
MIN_PADDING_LENGTH = 4
MAX_PADDING_LENGTH = 255

# Strong algorithm lists based on algorithm_guidance.json (canonical source)
# See algorithm_guidance.json for detailed rationale and references
STRONG_CIPHERS = [
    'chacha20-poly1305@openssh.com',
    'aes256-gcm@openssh.com',
    'aes128-gcm@openssh.com',
    'aes256-ctr',
    'aes192-ctr',
    'aes128-ctr'
]

STRONG_MACS = [
    'hmac-sha2-512-etm@openssh.com',
    'hmac-sha2-256-etm@openssh.com',
    'umac-128',
    'umac-128-etm@openssh.com',
    'hmac-sha2-512',
    'hmac-sha2-256',
    'umac-128@openssh.com'
]

STRONG_KEX = [
    'curve25519-sha256',
    'curve25519-sha256@libssh.org',
    'diffie-hellman-group-exchange-sha256',
    'diffie-hellman-group14-sha256',
    'diffie-hellman-group16-sha512',
    'diffie-hellman-group18-sha512',
    'sntrup761x25519-sha512@openssh.com',
    'sntrup761x25519-sha512',
    'mlkem768x25519-sha256',
    'kex-strict-s-v00@openssh.com',
    'ext-info-s'
]

STRONG_HOST_KEY_ALGORITHMS = [
    'ssh-ed25519',
    'ssh-ed25519-cert-v01@openssh.com',
    'rsa-sha2-256',
    'rsa-sha2-512',
    'ssh-rsa-cert-v01@openssh.com'
]


def parse_uint32(data: bytes, offset: int) -> Tuple[int, int]:
    """Parse a 4-byte big-endian unsigned integer. Returns (value, new_offset)."""
    if offset + 4 > len(data):
        raise ValueError("Insufficient data to parse uint32")
    value = struct.unpack('>I', data[offset:offset + 4])[0]
    return value, offset + 4


def parse_byte(data: bytes, offset: int) -> Tuple[int, int]:
    """Parse a single byte. Returns (value, new_offset)."""
    if offset >= len(data):
        raise ValueError("Insufficient data to parse byte")
    return data[offset], offset + 1


def parse_string(data: bytes, offset: int) -> Tuple[bytes, int]:
    """Parse a length-prefixed string. Returns (string_bytes, new_offset)."""
    length, offset = parse_uint32(data, offset)
    if offset + length > len(data):
        raise ValueError(f"Insufficient data to parse string of length {length}")
    string_data = data[offset:offset + length]
    return string_data, offset + length


def parse_name_list(data: bytes, offset: int) -> Tuple[List[str], int]:
    """Parse a name-list (comma-separated algorithm names). Returns (list, new_offset)."""
    name_list_bytes, offset = parse_string(data, offset)
    if len(name_list_bytes) == 0:
        return [], offset
    try:
        name_list_str = name_list_bytes.decode('ascii')
    except UnicodeDecodeError as e:
        raise ValueError(f"Invalid ASCII data in name-list: {e}")
    names = name_list_str.split(',')
    return names, offset


def parse_boolean(data: bytes, offset: int) -> Tuple[bool, int]:
    """Parse a boolean byte. Returns (bool_value, new_offset)."""
    value, offset = parse_byte(data, offset)
    return value != 0, offset


def parse_ssh_packet(conn: socket.socket) -> bytes:
    """
    Read and parse an SSH binary packet from the connection.
    Returns the payload bytes (without padding).
    """
    header = conn.recv(SSH_HEADER_LENGTH)
    if len(header) < SSH_HEADER_LENGTH:
        raise ValueError("Failed to read SSH packet header")

    packet_length = struct.unpack('>I', header[0:4])[0]
    padding_length = header[4]

    if packet_length < 1 or packet_length > MAX_PACKET_LENGTH:
        raise ValueError(f"Invalid packet length: {packet_length}")

    if padding_length < MIN_PADDING_LENGTH or padding_length > MAX_PADDING_LENGTH:
        raise ValueError(f"Invalid padding length: {padding_length} (must be {MIN_PADDING_LENGTH}-{MAX_PADDING_LENGTH})")

    payload_length = packet_length - padding_length - 1
    if payload_length < 0:
        raise ValueError(f"Invalid packet: padding_length {padding_length} exceeds packet_length {packet_length}")

    remaining = packet_length - 1
    data = b''
    while len(data) < remaining:
        chunk = conn.recv(remaining - len(data))
        if not chunk:
            raise ValueError("Connection closed while reading packet")
        data += chunk

    payload = data[0:payload_length]

    return payload


def parse_kexinit(payload: bytes) -> Dict[str, Any]:
    """
    Parse SSH_MSG_KEXINIT message payload.
    Returns a dictionary with all the algorithm lists.
    """
    offset = 0

    msg_type, offset = parse_byte(payload, offset)
    if msg_type != SSH_MSG_KEXINIT:
        raise ValueError(f"Expected SSH_MSG_KEXINIT (20), got {msg_type}")

    if offset + KEXINIT_COOKIE_LENGTH > len(payload):
        raise ValueError("Insufficient data for KEXINIT cookie")
    offset += KEXINIT_COOKIE_LENGTH

    kex_algorithms, offset = parse_name_list(payload, offset)
    server_host_key_algorithms, offset = parse_name_list(payload, offset)
    encryption_algorithms_c2s, offset = parse_name_list(payload, offset)
    encryption_algorithms_s2c, offset = parse_name_list(payload, offset)
    mac_algorithms_c2s, offset = parse_name_list(payload, offset)
    mac_algorithms_s2c, offset = parse_name_list(payload, offset)
    compression_algorithms_c2s, offset = parse_name_list(payload, offset)
    compression_algorithms_s2c, offset = parse_name_list(payload, offset)
    languages_c2s, offset = parse_name_list(payload, offset)
    languages_s2c, offset = parse_name_list(payload, offset)

    first_kex_packet_follows, offset = parse_boolean(payload, offset)
    reserved, offset = parse_uint32(payload, offset)

    return {
        'kex_algorithms': kex_algorithms,
        'server_host_key_algorithms': server_host_key_algorithms,
        'encryption_algorithms_client_to_server': encryption_algorithms_c2s,
        'encryption_algorithms_server_to_client': encryption_algorithms_s2c,
        'mac_algorithms_client_to_server': mac_algorithms_c2s,
        'mac_algorithms_server_to_client': mac_algorithms_s2c,
        'compression_algorithms_client_to_server': compression_algorithms_c2s,
        'compression_algorithms_server_to_client': compression_algorithms_s2c,
        'languages_client_to_server': languages_c2s,
        'languages_server_to_client': languages_s2c,
        'first_kex_packet_follows': first_kex_packet_follows,
        'reserved': reserved
    }


def exchange(ip: str, port: int) -> Optional[Dict[str, Any]]:
    """
    Connect to SSH server and retrieve KEXINIT data.
    Returns a dictionary with algorithm lists, or None on failure.
    """
    kexinit_data = None
    conn = None
    try:
        conn = socket.create_connection((ip, port), timeout=5)
        print(f"[*] Connected to {ip} on port {port}...")

        version_data = conn.recv(VERSION_STRING_MAX_LENGTH)
        if not version_data or b'\n' not in version_data:
            raise ValueError("Failed to receive SSH version string")

        version = version_data.decode('ascii', errors='ignore').split('\n')[0].strip()
        print(f"    [+] Target SSH version is: {version}")

        conn.send(b'SSH-2.0-OpenSSH_6.0p1\r\n')
        print("    [+] Retrieving algorithm information...")

        payload = parse_ssh_packet(conn)
        kexinit_data = parse_kexinit(payload)

    except Exception as e:
        print(f"[-] Error while connecting to {ip} on port {port}: {e}")
    finally:
        if conn:
            conn.close()

    return kexinit_data


def validate_port(port_str: str) -> Tuple[Optional[int], Optional[str]]:
    """Validate that port is a valid integer in range 1-65535."""
    try:
        port = int(port_str)
        if port < 1 or port > 65535:
            return None, "Port must be between 1 and 65535"
        return port, None
    except ValueError:
        return None, "Port must be a valid integer"


def parse_target(target: str) -> Tuple[Optional[str], Optional[int], Optional[str]]:
    """Parse target string to extract host and port, handling IPv6 addresses."""
    port: Optional[int] = 22
    host: Optional[str] = target

    if target.startswith('['):
        bracket_end = target.find(']')
        if bracket_end == -1:
            return None, None, "Invalid format: missing closing bracket"

        host = target[1:bracket_end]

        if bracket_end + 1 < len(target):
            if target[bracket_end + 1] != ':':
                return None, None, "Invalid format: expected ':' after bracket"
            port_str = target[bracket_end + 2:]
            if not port_str:
                return None, None, "Invalid format: missing port after ':'"
            port, error = validate_port(port_str)
            if error:
                return None, None, error
    elif ':' in target:
        colon_count = target.count(':')
        if colon_count > 1:
            return None, None, "Invalid format: IPv6 addresses must be enclosed in brackets [host]:port"

        parts = target.split(':')
        host = parts[0]
        port, error = validate_port(parts[1])
        if error:
            return None, None, error

    return host, port, None


def scan_target(target: str) -> int:
    """
    Scan target SSH server and display results.
    Returns 0 on success, 1 on failure.
    """
    host, port, error = parse_target(target)

    if error:
        print(f"[-] Error: {error}")
        return 1

    if not host or not host.strip():
        print("[-] Error: Hostname cannot be empty")
        return 1

    if port is None:
        print("[-] Error: Invalid port")
        return 1

    print(f"[*] Initiating scan for {host} on port {port}")
    kexinit_data = exchange(host, port)
    if kexinit_data:
        display_result(kexinit_data)
        return 0

    return 1


def print_algo_list(algo_list: List[str], title: str, strong_list: Optional[List[str]] = None) -> None:
    """Print a formatted list of algorithms in two columns with optional color coding."""
    if algo_list:
        print(f'    [+] Detected {title}: ')
        display_list = algo_list.copy()
        cols = 2
        while len(display_list) % cols != 0:
            display_list.append('')

        split = [display_list[i:i + len(display_list) // cols] for i in
                 range(0, len(display_list), len(display_list) // cols)]

        for row in zip(*split):
            formatted_row = []
            for algo in row:
                if algo:
                    if strong_list is not None:
                        if algo in strong_list:
                            colored = green(algo)
                        else:
                            colored = red(algo)
                        formatted_row.append(str.ljust(colored, 37 + len(colored) - len(algo)))
                    else:
                        formatted_row.append(str.ljust(algo, 37))
                else:
                    formatted_row.append(' ' * 37)
            print("          " + "".join(formatted_row))
    else:
        print(f'    [-] No {title} detected!')


def detect_not_recommended_algo(detected_list: List[str], strong_list: List[str]) -> List[str]:
    """Identify algorithms not recommended by comparing detected against strong list."""
    return [algo for algo in detected_list if algo not in strong_list]


def display_result(kexinit_data: Dict[str, Any]) -> None:
    """Display KEXINIT algorithm information and identify not recommended algorithms."""
    detected_ciphers = kexinit_data['encryption_algorithms_server_to_client']
    detected_kex = kexinit_data['kex_algorithms']
    detected_macs = kexinit_data['mac_algorithms_server_to_client']
    detected_hka = kexinit_data['server_host_key_algorithms']

    not_recommended_ciphers = detect_not_recommended_algo(detected_ciphers, STRONG_CIPHERS)
    not_recommended_kex = detect_not_recommended_algo(detected_kex, STRONG_KEX)
    not_recommended_macs = detect_not_recommended_algo(detected_macs, STRONG_MACS)
    not_recommended_hka = detect_not_recommended_algo(detected_hka, STRONG_HOST_KEY_ALGORITHMS)

    print_algo_list(detected_ciphers, 'ciphers', STRONG_CIPHERS)
    print_algo_list(detected_kex, 'KEX algorithms', STRONG_KEX)
    print_algo_list(detected_macs, 'MACs', STRONG_MACS)
    print_algo_list(detected_hka, 'HostKey algorithms', STRONG_HOST_KEY_ALGORITHMS)

    print_algo_list(not_recommended_ciphers, 'not recommended ciphers')
    print_algo_list(not_recommended_kex, 'not recommended KEX algorithms')
    print_algo_list(not_recommended_macs, 'not recommended MACs')
    print_algo_list(not_recommended_hka, 'not recommended HostKey algorithms')

    compression_algos = kexinit_data['compression_algorithms_server_to_client']
    if 'zlib@openssh.com' in compression_algos or 'zlib' in compression_algos:
        print('    [+] Compression is enabled')
    else:
        print('    [-] Compression is *not* enabled')


def main() -> None:
    """Main entry point for the SSH scanner."""
    global USE_COLOR

    parser = argparse.ArgumentParser(
        description='SSH server cipher and algorithm scanner',
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
Examples:
  %(prog)s github.com
  %(prog)s example.com:22
  %(prog)s [::1]:22
  %(prog)s [2001:db8::1]:8022
        """
    )
    parser.add_argument(
        'target',
        help='target SSH server (format: host[:port] or [ipv6]:port)'
    )
    parser.add_argument(
        '-v', '--version',
        action='version',
        version='%(prog)s 2.0'
    )
    parser.add_argument(
        '--no-color',
        action='store_true',
        help='disable colored output'
    )
    parser.add_argument(
        '--color',
        action='store_true',
        help='force colored output even when not in a TTY'
    )

    args = parser.parse_args()

    # Handle color flags
    if args.no_color:
        USE_COLOR = False
    elif args.color:
        USE_COLOR = True

    exit_code = scan_target(args.target)
    sys.exit(exit_code)


if __name__ == '__main__':
    main()