init

Containerfile

@@ -0,0 +1,29 @@
+# Low-interaction SSH honeypot image (Podman/Docker compatible).
+FROM python:3.12-slim
+
+# Don't write .pyc, flush stdout/stderr immediately so `podman logs` is live.
+ENV PYTHONDONTWRITEBYTECODE=1 \
+    PYTHONUNBUFFERED=1 \
+    HONEYPOT_HOST_KEY=/data/host_key \
+    HONEYPOT_LOG_DIR=/data/logs \
+    HONEYPOT_BIND=0.0.0.0 \
+    HONEYPOT_PORT=2222
+
+WORKDIR /app
+
+COPY requirements.txt .
+RUN pip install --no-cache-dir -r requirements.txt
+
+COPY ssh_honeypot.py .
+
+# Run as an unprivileged user. /data is where the host key and logs persist;
+# mount a volume here so they survive container restarts.
+RUN useradd --uid 1000 --create-home --shell /usr/sbin/nologin honeypot \
+    && mkdir -p /data/logs \
+    && chown -R honeypot:honeypot /data
+USER honeypot
+
+VOLUME ["/data"]
+EXPOSE 2222
+
+ENTRYPOINT ["python", "ssh_honeypot.py"]

requirements.txt

@@ -0,0 +1 @@
+paramiko==3.5.*

ssh_honeypot.py

@@ -0,0 +1,397 @@
+#!/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())