#!/usr/bin/env python3 """Low-interaction SSH honeypot. Accepts SSH connections, lets clients *attempt* to authenticate, logs the source address and the credentials they submit, then always rejects the auth and closes the connection. It never grants a shell or runs commands. This is a defensive / threat-intelligence tool. Only run it on systems and networks you are authorized to monitor. """ from __future__ import annotations import argparse import json import logging import os import signal import socket import threading from datetime import datetime, timezone from logging.handlers import RotatingFileHandler import paramiko # Spoofed version string presented to clients. Looks like a stock Ubuntu # OpenSSH so the honeypot blends in with real hosts. paramiko requires the # string to start with "SSH-2.0-". DEFAULT_BANNER = "SSH-2.0-OpenSSH_8.9p1 Ubuntu-3ubuntu0.4" # --------------------------------------------------------------------------- # # Logging # --------------------------------------------------------------------------- # # Dedicated loggers so we can route structured vs. human-readable records to # different handlers without interfering with paramiko's own logging. _event_logger = logging.getLogger("honeypot.events") # -> human-readable _json_logger = logging.getLogger("honeypot.json") # -> JSON Lines def setup_logging(log_dir: str) -> None: """Configure three sinks: JSON Lines file, text file, and stdout. Both ``_event_logger`` and ``_json_logger`` are configured here. They do not propagate to the root logger so paramiko's transport chatter stays out of our event logs. """ os.makedirs(log_dir, exist_ok=True) # JSON Lines: one json.dumps() record per line, no extra formatting. json_handler = RotatingFileHandler( os.path.join(log_dir, "honeypot.jsonl"), maxBytes=10 * 1024 * 1024, backupCount=5, encoding="utf-8", ) json_handler.setFormatter(logging.Formatter("%(message)s")) _json_logger.addHandler(json_handler) _json_logger.setLevel(logging.INFO) _json_logger.propagate = False # Human-readable: rotating file + stdout (stdout is what `podman logs` sees). # Local-time, space-separated, no timezone clutter — easy to scan in a terminal. text_fmt = logging.Formatter("%(asctime)s %(message)s", "%Y-%m-%d %H:%M:%S") text_handler = RotatingFileHandler( os.path.join(log_dir, "honeypot.log"), maxBytes=10 * 1024 * 1024, backupCount=5, encoding="utf-8", ) text_handler.setFormatter(text_fmt) stdout_handler = logging.StreamHandler() # defaults to stderr/stdout stream stdout_handler.setFormatter(text_fmt) _event_logger.addHandler(text_handler) _event_logger.addHandler(stdout_handler) _event_logger.setLevel(logging.INFO) _event_logger.propagate = False # Keep paramiko's internal logging quiet but capture genuine errors. logging.getLogger("paramiko").setLevel(logging.WARNING) def _sanitize(value) -> str: """Escape control / non-printable characters in an attacker-controlled string. The human-readable log is meant to be viewed in a terminal (``cat``, ``tail``, ``podman logs``). Without this, a hostile value — e.g. a crafted SSH client version banner — could carry ANSI escape sequences (move the cursor, clear the screen, recolour) or embedded newlines that forge extra log lines. The JSON sink is already safe because ``json.dumps`` escapes these, so this only guards the human sink. ``repr``-rendered fields (username/password) are likewise safe. """ return "".join(ch if ch.isprintable() or ch == " " else repr(ch)[1:-1] for ch in str(value)) def _human_message(event: str, fields: dict) -> str: """Render a concise, terminal-friendly one-liner for an event. Each event type gets a purpose-built layout (aligned source column, only the fields that matter) instead of a generic key=value dump, so a stream of attempts is easy to read at a glance. """ src = f"{fields['src_ip']}:{fields.get('src_port', '?')}" if "src_ip" in fields else "" if event == "login_attempt": if fields.get("auth_method") == "password": cred = f"user={fields.get('username')!r} pass={fields.get('password')!r}" else: cred = (f"user={fields.get('username')!r} " f"key={fields.get('key_type')}/{fields.get('key_fingerprint')}") client = _sanitize(fields.get("client_version", "")) return f"{src:<21} {cred} [{fields.get('auth_method')}] {client}".rstrip() if event == "connection": return f"{src:<21} new connection" if event == "disconnect": return f"{src:<21} disconnected" if event == "connection_error": return f"{src:<21} error: {_sanitize(fields.get('error'))}" if event == "server_start": return (f"listening on {fields.get('bind')}:{fields.get('port')} " f"banner={fields.get('banner')!r} fingerprint={fields.get('host_key_fingerprint')}") if event == "server_stop": return "shutting down" if event == "host_key_generated": return f"generated host key {fields.get('fingerprint')} at {fields.get('path')}" # Fallback for any event without a custom layout. return " ".join(f"{k}={fields[k]!r}" for k in fields) def log_event(event: str, **fields) -> None: """Emit one event to both the JSON Lines sink and the human-readable sink.""" # RFC3339 / ISO-8601 UTC with millisecond precision and a trailing 'Z'. # This is what Grafana Loki / Promtail parse natively, and millisecond # precision keeps bursty attempts correctly ordered. ts = datetime.now(timezone.utc).isoformat(timespec="milliseconds").replace("+00:00", "Z") record = {"timestamp": ts, "event": event, **fields} # JSON Lines for automation/Loki: one object per line. ensure_ascii=False # keeps non-ASCII credentials readable; sort_keys gives stable field order. _json_logger.info(json.dumps(record, ensure_ascii=False, sort_keys=True)) # Aligned, human-readable rendering of the same event for the terminal. _event_logger.info(f"{event:<18} {_human_message(event, fields)}") # --------------------------------------------------------------------------- # # Host key # --------------------------------------------------------------------------- # def load_or_create_host_key(path: str) -> paramiko.PKey: """Load the persisted RSA host key, generating one on first run. A stable host key keeps the server's fingerprint constant across restarts, so returning scanners see a consistent host. RSA is used because paramiko can both generate and persist it directly. """ if os.path.exists(path): return paramiko.RSAKey(filename=path) os.makedirs(os.path.dirname(os.path.abspath(path)), exist_ok=True) key = paramiko.RSAKey.generate(2048) # Write with 0600 perms (the key file controls the host identity). fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600) os.close(fd) key.write_private_key_file(path) os.chmod(path, 0o600) log_event("host_key_generated", path=path, fingerprint=key.fingerprint) return key # --------------------------------------------------------------------------- # # SSH server logic # --------------------------------------------------------------------------- # class HoneypotServer(paramiko.ServerInterface): """Logs every auth attempt and refuses all of them (low-interaction).""" def __init__(self, peer: tuple[str, int]) -> None: self.peer_ip, self.peer_port = peer self.client_version = "" # filled in once the transport handshake runs # Offer password + publickey so clients reveal which they intend to use. def get_allowed_auths(self, username: str) -> str: return "password,publickey" def check_auth_password(self, username: str, password: str) -> int: log_event( "login_attempt", src_ip=self.peer_ip, src_port=self.peer_port, username=username, password=password, auth_method="password", client_version=self.client_version, ) return paramiko.common.AUTH_FAILED def check_auth_publickey(self, username: str, key: paramiko.PKey) -> int: log_event( "login_attempt", src_ip=self.peer_ip, src_port=self.peer_port, username=username, auth_method="publickey", key_type=key.get_name(), key_fingerprint=key.fingerprint, client_version=self.client_version, ) return paramiko.common.AUTH_FAILED # Reject anything that would constitute interaction. We never get here with # a successful auth, but be explicit and safe regardless. def check_channel_request(self, kind: str, chanid: int) -> int: return paramiko.common.OPEN_FAILED_ADMINISTRATIVELY_PROHIBITED def check_channel_shell_request(self, channel) -> bool: return False def check_channel_exec_request(self, channel, command) -> bool: return False def check_channel_pty_request(self, *args, **kwargs) -> bool: return False def handle_connection( client_sock: socket.socket, addr: tuple[str, int], host_key: paramiko.PKey, banner: str, max_attempts: int, timeout: int, ) -> None: """Run the SSH handshake for one client and log its auth attempts.""" peer_ip, peer_port = addr log_event("connection", src_ip=peer_ip, src_port=peer_port) client_sock.settimeout(timeout) transport = paramiko.Transport(client_sock) transport.local_version = banner transport.add_server_key(host_key) # Limit auth tries so a single client can't hold a thread open forever. transport.auth_timeout = timeout transport.banner_timeout = timeout server = HoneypotServer(addr) try: transport.start_server(server=server) server.client_version = transport.remote_version or "" # The client authenticates against HoneypotServer (which always fails). # We just wait for the transport to be torn down or the attempts to run # out. paramiko enforces its own internal auth-attempt limit; we bound # the wall-clock time via the join timeout. transport.join(timeout=timeout) except (paramiko.SSHException, EOFError, socket.error, OSError) as exc: log_event( "connection_error", src_ip=peer_ip, src_port=peer_port, error=f"{type(exc).__name__}: {exc}", ) except Exception as exc: # never let one client kill the accept loop log_event( "connection_error", src_ip=peer_ip, src_port=peer_port, error=f"unexpected {type(exc).__name__}: {exc}", ) finally: try: transport.close() except Exception: pass try: client_sock.close() except Exception: pass log_event("disconnect", src_ip=peer_ip, src_port=peer_port) # --------------------------------------------------------------------------- # # Server bootstrap # --------------------------------------------------------------------------- # class Honeypot: def __init__(self, args: argparse.Namespace, host_key: paramiko.PKey) -> None: self.args = args self.host_key = host_key self._stop = threading.Event() self._sock: socket.socket | None = None def stop(self, *_) -> None: self._stop.set() if self._sock is not None: try: self._sock.close() # unblocks accept() except Exception: pass def serve(self) -> None: sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) sock.bind((self.args.bind, self.args.port)) sock.listen(100) self._sock = sock log_event( "server_start", bind=self.args.bind, port=self.args.port, banner=self.args.banner, host_key_fingerprint=self.host_key.fingerprint, ) while not self._stop.is_set(): try: client_sock, addr = sock.accept() except OSError: break # socket closed by stop() thread = threading.Thread( target=handle_connection, args=( client_sock, addr, self.host_key, self.args.banner, self.args.max_attempts, self.args.timeout, ), daemon=True, ) thread.start() log_event("server_stop") # --------------------------------------------------------------------------- # # CLI # --------------------------------------------------------------------------- # def _env_default(name: str, fallback: str) -> str: return os.environ.get(name, fallback) def parse_args(argv=None) -> argparse.Namespace: p = argparse.ArgumentParser(description="Low-interaction SSH honeypot.") p.add_argument("--bind", default=_env_default("HONEYPOT_BIND", "0.0.0.0"), help="Listen address (env HONEYPOT_BIND, default 0.0.0.0)") p.add_argument("--port", type=int, default=int(_env_default("HONEYPOT_PORT", "2222")), help="Listen port (env HONEYPOT_PORT, default 2222)") p.add_argument("--host-key", default=_env_default("HONEYPOT_HOST_KEY", "./host_key"), help="Path to the persisted host key (env HONEYPOT_HOST_KEY)") p.add_argument("--log-dir", default=_env_default("HONEYPOT_LOG_DIR", "./logs"), help="Directory for log files (env HONEYPOT_LOG_DIR)") p.add_argument("--banner", default=_env_default("HONEYPOT_BANNER", DEFAULT_BANNER), help="Spoofed SSH version string (env HONEYPOT_BANNER)") p.add_argument("--max-attempts", type=int, default=int(_env_default("HONEYPOT_MAX_ATTEMPTS", "3")), help="Auth attempts before disconnect (env HONEYPOT_MAX_ATTEMPTS)") p.add_argument("--timeout", type=int, default=int(_env_default("HONEYPOT_TIMEOUT", "15")), help="Per-connection timeout in seconds (env HONEYPOT_TIMEOUT)") return p.parse_args(argv) def main(argv=None) -> int: args = parse_args(argv) setup_logging(args.log_dir) if not args.banner.startswith("SSH-2.0-"): log_event("config_error", error="banner must start with 'SSH-2.0-'") return 2 host_key = load_or_create_host_key(args.host_key) honeypot = Honeypot(args, host_key) # Clean shutdown on Ctrl-C / container stop. signal.signal(signal.SIGINT, honeypot.stop) signal.signal(signal.SIGTERM, honeypot.stop) try: honeypot.serve() except OSError as exc: log_event("fatal", error=f"{type(exc).__name__}: {exc}") return 1 return 0 if __name__ == "__main__": raise SystemExit(main())