前言
最近这段时间家里的飞牛服务器被人疯狂的试SSH密码，看了下系统日志，是用撞库方式试密码，发现SSH密码已经连续试了半个月了，很是烦人，正好今天没事，就做个SSH蜜罐玩玩吧，随便还能收集一本密码库。
服务器配置：飞牛，用frpc做内网穿透
文件目录结构
/opt/ssh-honeypot/
├── ssh-honeypot.py # 蜜罐主程序
├── install.sh # 一键部署
├── ssh-honeypot.service # systemd 服务（参考，install.sh 会自动生成）
├── host_key_rsa # 主机密钥（运行时自动生成）
└── .venv/ # Python 虚拟环境/var/log/ssh-honeypot/
├── ssh-honeypot.log # 主日志（完整事件流）
└── ssh-honeypot-creds.log # 凭据 CSV（只含 IP + 账号 + 密码）
ssh-honeypot.py
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
SSH 蜜罐服务 - 捕获暴力破解的 IP、用户名、密码
原理：在独立端口启动假 SSH 服务，攻击者尝试登录时完整记录其
IP、账号、密码，然后一律拒绝认证，诱使其继续尝试。
标准 SSH 日志只记录用户名不记录密码，只有蜜罐才能抓到密码。
网络架构：
蜜罐监听 2222，由网络前端把外部 22 端口的流量转发进来。
常见前端：frp（内网穿透）、iptables DNAT、HAProxy、nginx stream。
若前端做了 SNAT（源 IP 被改写），需开启 PROXY protocol 还原真实 IP。
依赖：pip install paramiko
"""
import csv
import io
import logging
import os
import signal
import socket
import sys
import threading
import time
from logging.handlers import RotatingFileHandler
try:
import paramiko
except ImportError:
sys.stderr.write("缺少依赖，请先安装: pip install paramiko\n")
sys.exit(1)
# ======================== 配置 ========================
LISTEN_IP = "0.0.0.0"
LISTEN_PORT = 2222 # 蜜罐监听端口；外部 22 由网络前端转发到此
LOG_DIR = "/var/log/ssh-honeypot"
LOG_FILE = f"{LOG_DIR}/ssh-honeypot.log"
CREDS_LOG_FILE = f"{LOG_DIR}/ssh-honeypot-creds.log"
LOG_MAX_BYTES = 10 * 1024 * 1024 # 单文件 10MB
LOG_BACKUP_COUNT = 5
HOST_KEY_DIR = os.path.dirname(os.path.abspath(__file__))
HOST_KEY_FILE = f"{HOST_KEY_DIR}/host_key_rsa"
USE_PROXY_PROTOCOL = True # frp/HAProxy 等做 SNAT 时设 True
PROXY_HEADER_TIMEOUT = 5 # 等待 PROXY 头部的超时（秒）
PROXY_V2_SIG = b"\x0d\x0a\x0d\x0a\x00\x0d\x0a\x51\x55\x49\x54\x0a"
# ======================== 日志 ========================
class CsvCredsFormatter(logging.Formatter):
"""凭据记录 -> CSV 行（自动转义含逗号/引号的密码）"""
def format(self, record):
ts = self.formatTime(record, self.datefmt)
row = [ts, getattr(record, "ip", ""),
getattr(record, "user", ""),
getattr(record, "pass", "")]
buf = io.StringIO()
csv.writer(buf, quoting=csv.QUOTE_MINIMAL, lineterminator="").writerow(row)
return buf.getvalue()
class CsvCredsHandler(RotatingFileHandler):
"""轮转 CSV 文件；新建/轮转后的空文件自动写入表头"""
HEADER = "timestamp,ip,user,password"
def _open(self):
stream = super()._open()
try:
if stream.tell() == 0:
stream.write(self.HEADER + "\n")
except Exception:
pass
return stream
def setup_loggers():
os.makedirs(LOG_DIR, exist_ok=True)
lg = logging.getLogger("ssh-honeypot")
lg.setLevel(logging.INFO)
lg.propagate = False
if not lg.handlers:
fmt = logging.Formatter("%(asctime)s | %(message)s",
datefmt="%Y-%m-%d %H:%M:%S")
handler = RotatingFileHandler(
LOG_FILE, maxBytes=LOG_MAX_BYTES,
backupCount=LOG_BACKUP_COUNT, encoding="utf-8")
handler.setFormatter(fmt)
lg.addHandler(handler)
sh = logging.StreamHandler(sys.stderr)
sh.setFormatter(fmt)
lg.addHandler(sh)
creds = logging.getLogger("ssh-honeypot-creds")
creds.setLevel(logging.INFO)
creds.propagate = False
if not creds.handlers:
ch = CsvCredsHandler(
CREDS_LOG_FILE, maxBytes=LOG_MAX_BYTES,
backupCount=LOG_BACKUP_COUNT, encoding="utf-8")
ch.setFormatter(CsvCredsFormatter(datefmt="%Y-%m-%d %H:%M:%S"))
creds.addHandler(ch)
return lg, creds
logger, creds_logger = setup_loggers()
# ======================== 主机密钥 ========================
def load_or_create_host_key():
"""加载或生成 RSA 主机密钥（蜜罐自用，不影响真实 sshd）"""
os.makedirs(HOST_KEY_DIR, exist_ok=True)
if os.path.exists(HOST_KEY_FILE):
return paramiko.RSAKey(filename=HOST_KEY_FILE)
key = paramiko.RSAKey.generate(2048)
key.write_private_key_file(HOST_KEY_FILE)
os.chmod(HOST_KEY_FILE, 0o600)
logger.info("HOST_KEY | 生成新蜜罐主机密钥 path=%s", HOST_KEY_FILE)
return key
HOST_KEY = load_or_create_host_key()
# ======================== PROXY protocol 解析 ========================
# 核心设计：用 MSG_PEEK 偷看前几个字节，只在确认是 PROXY 头时才消费。
# 不是 PROXY 头时一个字节都不读，paramiko 直接从原始 socket 读取。
# =======================================================================
def _recv_exact(sock, n):
"""精确读取并消费 n 字节"""
data = b""
while len(data) < n:
try:
chunk = sock.recv(n - len(data))
if not chunk:
break
data += chunk
except socket.timeout:
break
return data
def _parse_proxy_v1(data):
"""解析 PROXY protocol v1 文本头部 -> (src_ip, src_port) 或 None"""
try:
line = data.decode("ascii", "replace").strip()
if not line.startswith("PROXY "):
return None
parts = line.split()
if len(parts) < 6 or parts[1] == "UNKNOWN":
return None
return (parts[2], int(parts[4]))
except (ValueError, UnicodeDecodeError):
return None
def _parse_proxy_v2(data):
"""解析 PROXY protocol v2 二进制头部 -> (src_ip, src_port) 或 None"""
if len(data) < 16 or not data.startswith(PROXY_V2_SIG):
return None
if data[12] >> 4 != 2:
return None
af = data[13] >> 4
length = (data[14] << 8) | data[15]
if len(data) < 16 + length:
return None
payload = data[16:16 + length]
if af == 1: # IPv4: src(4)+dst(4)+src_port(2)+dst_port(2)
if len(payload) >= 12:
return (socket.inet_ntoa(payload[:4]),
(payload[8] << 8) | payload[9])
elif af == 2: # IPv6: src(16)+dst(16)+src_port(2)+dst_port(2)
if len(payload) >= 36:
return (socket.inet_ntop(socket.AF_INET6, payload[:16]),
(payload[32] << 8) | payload[33])
return None
def read_proxy_header(client, fallback_addr):
"""
用 MSG_PEEK 偷看 socket 前几个字节判断是否为 PROXY protocol 头部。
- 是 PROXY 头：精确消费头部字节（不多读一个 SSH 字节），返回真实 IP。
- 不是 PROXY 头：一个字节都不消费，返回 socket 对端 IP。
返回 (real_ip, real_port)
"""
fb_ip, fb_port = fallback_addr
client.settimeout(PROXY_HEADER_TIMEOUT)
# 偷看前 12 字节（不消费）
peek = b""
while len(peek) < 12:
try:
chunk = client.recv(12 - len(peek), socket.MSG_PEEK)
if not chunk:
client.settimeout(None)
return (fb_ip, fb_port)
peek += chunk
except (socket.timeout, OSError):
client.settimeout(None)
return (fb_ip, fb_port)
# ---- PROXY v2（二进制签名）----
if peek == PROXY_V2_SIG:
client.settimeout(PROXY_HEADER_TIMEOUT)
header = _recv_exact(client, 16)
if len(header) < 16:
logger.warning("PROXY_V2_READ_FAIL | ip=%s", fb_ip)
client.settimeout(None)
return (fb_ip, fb_port)
length = (header[14] << 8) | header[15]
payload = _recv_exact(client, length) if length > 0 else b""
result = _parse_proxy_v2(header + payload)
if result:
#logger.info("PROXY_V2_OK | real_ip=%s real_port=%s sock_ip=%s",
# result[0], result[1], fb_ip)
client.settimeout(None)
return result
logger.warning("PROXY_V2_PARSE_FAIL | ip=%s", fb_ip)
client.settimeout(None)
return (fb_ip, fb_port)
# ---- PROXY v1（文本 "PROXY ...\r\n"）----
if peek.startswith(b"PROXY "):
# 用 deadline 循环 peek，直到找到 \r\n 或超时
deadline = time.monotonic() + PROXY_HEADER_TIMEOUT
while b"\r\n" not in peek:
remaining = deadline - time.monotonic()
if remaining <= 0:
break
client.settimeout(remaining)
try:
new_peek = client.recv(256, socket.MSG_PEEK)
if not new_peek or len(new_peek) <= len(peek):
# 没有新数据，短暂等待后重试
time.sleep(0.1)
continue
peek = new_peek
if len(peek) >= 256:
break
except (socket.timeout, OSError):
break
if b"\r\n" in peek:
idx = peek.index(b"\r\n") + 2
client.settimeout(PROXY_HEADER_TIMEOUT)
header_data = _recv_exact(client, idx)
result = _parse_proxy_v1(header_data)
if result:
logger.info("PROXY_V1_OK | real_ip=%s real_port=%s sock_ip=%s",
result[0], result[1], fb_ip)
client.settimeout(None)
return result
logger.warning("PROXY_V1_PARSE_FAIL | ip=%s", fb_ip)
client.settimeout(None)
return (fb_ip, fb_port)
# ---- 不是 PROXY 头部：不消费任何字节 ----
client.settimeout(None)
return (fb_ip, fb_port)
# ======================== 蜜罐 SSH 服务端 ========================
class HoneypotServer(paramiko.ServerInterface):
"""paramiko 服务端回调：在认证阶段捕获账号密码"""
def __init__(self, client_addr):
self.client_addr = client_addr
self.event = threading.Event()
def check_auth_password(self, username, password):
ip, port = self.client_addr
logger.info("AUTH_PASSWORD_FAIL | ip=%s port=%s user=%s pass=%s",
ip, port, username, password)
creds_logger.info("", extra={"ip": ip, "user": username, "pass": password})
return paramiko.AUTH_FAILED
def check_auth_publickey(self, username, key):
ip, port = self.client_addr
logger.info("AUTH_PUBKEY_FAIL | ip=%s port=%s user=%s key=%s fp=%s",
ip, port, username, key.get_name(),
":".join(f"{b:02x}" for b in key.get_fingerprint()))
return paramiko.AUTH_FAILED
def get_allowed_auths(self, username):
return "password,publickey"
def check_channel_request(self, kind, chanid):
return paramiko.OPEN_SUCCEEDED
def check_channel_shell_request(self, channel):
return False
def check_channel_pty_request(self, channel, term, width, height,
pixwidth, pixheight, modes):
return False
def check_channel_exec_request(self, channel, command):
ip, port = self.client_addr
logger.info("EXEC_BLOCK | ip=%s port=%s cmd=%s",
ip, port, command.decode("utf-8", "replace"))
return False
# ======================== 连接处理 ========================
def handle_client(client, addr):
"""处理单个连接的完整生命周期"""
transport = None
try:
real_addr = addr
if USE_PROXY_PROTOCOL:
real_ip, real_port = read_proxy_header(client, addr)
real_addr = (real_ip, real_port)
if real_ip != addr[0]:
logger.info("PROXY_PROTOCOL | real_ip=%s real_port=%s sock_ip=%s",
real_ip, real_port, addr[0])
transport = paramiko.Transport(client)
transport.add_server_key(HOST_KEY)
transport.banner_timeout = 30
transport.auth_timeout = 60
transport.start_server(server=HoneypotServer(real_addr))
chan = transport.accept(timeout=20)
if chan is not None:
chan.close()
except paramiko.SSHException as e:
logger.debug("SSH_PROTOCOL_ERR | ip=%s err=%s", addr[0], e)
except Exception as e:
logger.debug("CONN_ERR | ip=%s err=%s", addr[0], e)
finally:
if transport is not None:
try:
transport.close()
except Exception:
pass
try:
client.close()
except Exception:
pass
# ======================== 主入口 ========================
def main():
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
try:
sock.bind((LISTEN_IP, LISTEN_PORT))
except PermissionError:
sys.stderr.write(f"无权限绑定端口 {LISTEN_PORT}，<1024 需 root。\n")
sys.exit(1)
except OSError as e:
sys.stderr.write(f"绑定 {LISTEN_IP}:{LISTEN_PORT} 失败：{e}\n")
sys.exit(1)
sock.listen(200)
logger.info("HONEYPOT_START | listen=%s:%s log=%s creds_log=%s proxy=%s",
LISTEN_IP, LISTEN_PORT, LOG_FILE, CREDS_LOG_FILE, USE_PROXY_PROTOCOL)
sys.stderr.write(
f"\n[*] SSH 蜜罐已启动 listen={LISTEN_IP}:{LISTEN_PORT}\n"
f"[*] 主日志: {LOG_FILE}\n"
f"[*] 凭据日志: {CREDS_LOG_FILE}\n"
f"[*] PROXY protocol: {'开启' if USE_PROXY_PROTOCOL else '关闭'}\n"
f"[*] 按 Ctrl+C 停止\n\n"
)
def graceful_shutdown(signum, frame):
logger.info("HONEYPOT_STOP | signal=%s", signum)
sock.close()
sys.exit(0)
signal.signal(signal.SIGTERM, graceful_shutdown)
signal.signal(signal.SIGINT, graceful_shutdown)
try:
while True:
try:
client, addr = sock.accept()
except OSError:
break
#logger.info("NEW_CONN | ip=%s port=%s", addr[0], addr[1])
threading.Thread(target=handle_client, args=(client, addr),
daemon=True).start()
finally:
sock.close()
if __name__ == "__main__":
main()
ssh-honeypot.service
[Unit]
Description=SSH Honeypot (capture brute-force credentials)
After=network.target
[Service]
Type=simple
ExecStart=/opt/ssh-honeypot/.venv/bin/python /opt/ssh-honeypot/ssh-honeypot.py
Restart=on-failure
RestartSec=5
User=root
NoNewPrivileges=true
ProtectSystem=strict
ProtectHome=true
PrivateTmp=true
ReadWritePaths=/var/log/ssh-honeypot /opt/ssh-honeypot
[Install]
WantedBy=multi-user.target
install.sh
#!/bin/bash
# ===================================================================
# SSH 蜜罐一键部署脚本
#
# 前提：ssh-honeypot.py 和本脚本放在同一目录（默认 /opt/ssh-honeypot）
# 功能：创建虚拟环境 + 安装 paramiko + 创建日志目录 + 安装 systemd 服务
# 用法：sudo bash install.sh
# ===================================================================
set -e
# ---------- 配置 ----------
DEPLOY_DIR="$(cd "$(dirname "$0")" && pwd)"
VENV_DIR="${VENV_DIR:-$DEPLOY_DIR/.venv}"
HONEYPOT_SCRIPT="$DEPLOY_DIR/ssh-honeypot.py"
SERVICE_FILE="/etc/systemd/system/ssh-honeypot.service"
LOG_DIR="/var/log/ssh-honeypot"
PYTHON_BIN="${PYTHON_BIN:-python3}"
# --------------------------
err() { echo "[错误] $*" >&2; exit 1; }
info() { echo "[*] $*"; }
[[ "$(id -u)" -eq 0 ]] || err "请用 root 或 sudo 运行"
[[ -f "$HONEYPOT_SCRIPT" ]] || err "找不到 $HONEYPOT_SCRIPT（请与本脚本放同一目录）"
info "部署目录: $DEPLOY_DIR"
# ---- Python ----
command -v "$PYTHON_BIN" >/dev/null 2>&1 || err "找不到 $PYTHON_BIN"
info "Python: $($PYTHON_BIN -c 'import sys; print("%d.%d" % sys.version_info[:2])')"
if ! "$PYTHON_BIN" -c "import venv" 2>/dev/null; then
info "安装 venv 模块..."
if command -v apt-get >/dev/null 2>&1; then
apt-get update -qq && apt-get install -y -qq python3-venv
elif command -v dnf >/dev/null 2>&1; then
dnf install -y -q python3-devel
elif command -v yum >/dev/null 2>&1; then
yum install -y -q python3-devel
else
err "请手动安装 python3-venv"
fi
fi
# ---- 虚拟环境 ----
if [[ -d "$VENV_DIR" ]]; then
info "虚拟环境已存在: $VENV_DIR"
else
info "创建虚拟环境: $VENV_DIR"
"$PYTHON_BIN" -m venv "$VENV_DIR"
fi
VENV_PY="$VENV_DIR/bin/python"
[[ -x "$VENV_PY" ]] || err "虚拟环境创建失败"
# ---- 依赖 ----
info "安装 paramiko..."
"$VENV_DIR/bin/pip" install --quiet --upgrade pip
"$VENV_DIR/bin/pip" install --quiet paramiko
"$VENV_PY" -c "import paramiko; print(' paramiko', paramiko.__version__)" || err "paramiko 安装失败"
# ---- 日志目录 ----
mkdir -p "$LOG_DIR"
# ---- systemd 服务 ----
info "安装 systemd 服务"
cat > "$SERVICE_FILE" << EOF
[Unit]
Description=SSH Honeypot (capture brute-force credentials)
After=network.target
[Service]
Type=simple
ExecStart=$VENV_PY $HONEYPOT_SCRIPT
Restart=on-failure
RestartSec=5
User=root
NoNewPrivileges=true
ProtectSystem=strict
ProtectHome=true
PrivateTmp=true
ReadWritePaths=$LOG_DIR $DEPLOY_DIR
[Install]
WantedBy=multi-user.target
EOF
systemctl daemon-reload
# ---- 完成提示 ----
cat << EOF
============================================================
部署完成
============================================================
程序: $HONEYPOT_SCRIPT
虚拟环境: $VENV_DIR
日志目录: $LOG_DIR
服务: $SERVICE_FILE
后续操作:
1) frp 用户（获取真实 IP）:
# frpc.toml 加: transport.proxyProtocolVersion = "v2"
sudo sed -i 's/USE_PROXY_PROTOCOL = False/USE_PROXY_PROTOCOL = True/' $HONEYPOT_SCRIPT
sudo systemctl restart ssh-honeypot
2) 启动服务:
sudo systemctl enable --now ssh-honeypot
sudo systemctl status ssh-honeypot
3) 查看日志:
sudo tail -f $LOG_DIR/ssh-honeypot.log
sudo tail -f $LOG_DIR/ssh-honeypot-creds.log
4) 前台调试:
sudo $VENV_PY $HONEYPOT_SCRIPT
卸载:
sudo systemctl disable --now ssh-honeypot
sudo rm -f $SERVICE_FILE && sudo systemctl daemon-reload
sudo rm -rf $VENV_DIR $DEPLOY_DIR/host_key_rsa $LOG_DIR
============================================================
EOF部署流程
# 一条命令搞定：建虚拟环境 + 装依赖 + 装服务
sudo chmod +x install.sh
sudo bash install.sh
# 启动
sudo systemctl enable --now ssh-honeypot
日志样式
root@DiskStation:/var/log/ssh-honeypot# tail -n 20 ssh-honeypot.log2026-08-19 07:42:54 | PROXY_PROTOCOL | real_ip=45.148.10.195 real_port=38046 sock_ip=10.0.0.5
2026-08-19 07:42:55 | AUTH_PASSWORD_FAIL | ip=45.148.10.195 port=38046 user=root pass=admin@!QAZ2wsx
2026-08-19 07:42:56 | PROXY_PROTOCOL | real_ip=193.32.162.46 real_port=40370 sock_ip=10.0.0.5
2026-08-19 07:42:57 | AUTH_PASSWORD_FAIL | ip=193.32.162.46 port=40370 user=cancer pass=123456
2026-08-19 07:42:58 | PROXY_PROTOCOL | real_ip=193.32.162.45 real_port=41582 sock_ip=10.0.0.5
2026-08-19 07:42:59 | AUTH_PASSWORD_FAIL | ip=193.32.162.45 port=41582 user=root pass=taiji123root
2026-08-19 07:43:33 | PROXY_PROTOCOL | real_ip=193.32.162.45 real_port=49960 sock_ip=10.0.0.5
2026-08-19 07:43:33 | AUTH_PASSWORD_FAIL | ip=193.32.162.45 port=49960 user=root pass=gpuadmin123
2026-08-19 07:43:36 | PROXY_PROTOCOL | real_ip=45.148.10.195 real_port=59454 sock_ip=10.0.0.5
2026-08-19 07:43:37 | AUTH_PASSWORD_FAIL | ip=45.148.10.195 port=59454 user=root pass=Pa55word88
2026-08-19 07:43:48 | PROXY_PROTOCOL | real_ip=193.32.162.46 real_port=36986 sock_ip=10.0.0.5
2026-08-19 07:43:49 | AUTH_PASSWORD_FAIL | ip=193.32.162.46 port=36986 user=liyang.zhang pass=123456
2026-08-19 07:44:07 | PROXY_PROTOCOL | real_ip=193.32.162.45 real_port=53160 sock_ip=10.0.0.5
2026-08-19 07:44:08 | AUTH_PASSWORD_FAIL | ip=193.32.162.45 port=53160 user=root pass=1qaz2wsx3edc$
2026-08-19 07:44:19 | PROXY_PROTOCOL | real_ip=45.148.10.195 real_port=39230 sock_ip=10.0.0.5
2026-08-19 07:44:20 | AUTH_PASSWORD_FAIL | ip=45.148.10.195 port=39230 user=root pass=Unicom123!@#
2026-08-19 07:44:40 | PROXY_PROTOCOL | real_ip=193.32.162.46 real_port=36432 sock_ip=10.0.0.5
2026-08-19 07:44:41 | AUTH_PASSWORD_FAIL | ip=193.32.162.46 port=36432 user=chentianshuo pass=123456
2026-08-19 07:44:43 | PROXY_PROTOCOL | real_ip=193.32.162.45 real_port=37162 sock_ip=10.0.0.5
2026-08-19 07:44:44 | AUTH_PASSWORD_FAIL | ip=193.32.162.45 port=37162 user=root pass=A2forever
root@DiskStation:/var/log/ssh-honeypot$ tail -n 20 ssh-honeypot-creds.log2026-08-19 07:40:26,193.32.162.46,qianyuzhuang,123456
2026-08-19 07:40:42,193.32.162.45,ubuntu,changeme
2026-08-19 07:40:50,45.148.10.195,root,Cmadaas#2019
2026-08-19 07:41:16,193.32.162.45,root,admin2015
2026-08-19 07:41:17,193.32.162.46,kang,123456
2026-08-19 07:41:32,45.148.10.195,root,yaoyong@1978
2026-08-19 07:41:50,193.32.162.45,root,Aa123321
2026-08-19 07:42:06,193.32.162.46,xichaowen,123456
2026-08-19 07:42:13,45.148.10.195,root,root@2016!@#$
2026-08-19 07:42:24,193.32.162.45,root,buzhidao
2026-08-19 07:42:55,45.148.10.195,root,admin@!QAZ2wsx
2026-08-19 07:42:57,193.32.162.46,cancer,123456
2026-08-19 07:42:59,193.32.162.45,root,taiji123root
2026-08-19 07:43:33,193.32.162.45,root,gpuadmin123
2026-08-19 07:43:37,45.148.10.195,root,Pa55word88
2026-08-19 07:43:49,193.32.162.46,liyang.zhang,123456
2026-08-19 07:44:08,193.32.162.45,root,1qaz2wsx3edc$
2026-08-19 07:44:20,45.148.10.195,root,Unicom123!@#
2026-08-19 07:44:41,193.32.162.46,chentianshuo,123456
2026-08-19 07:44:44,193.32.162.45,root,A2forever
挺不错，一天就收集成1万多条密码本，真希望多来几个人试我的SSH密码 
:bi...
文章评论