别让探针白嫖带宽：公网 VPS 低成本轻量级诱捕与恶意流量画像实战

#!/usr/bin/env python3
import asyncio
import logging
import subprocess

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
    handlers=[logging.FileHandler("/var/log/honeypot.log"), logging.StreamHandler()]
)

NFT_SET_NAME = "bad_guys"
NFT_TIMEOUT = "86400s"  # 封禁 24 小时

def ban_ip_kernel(ip: str):
    """通过 nftables 动态集合注入内核黑名单"""
    cmd = ["nft", "add", "element", "ip", "filter", NFT_SET_NAME, f"{{ {ip} timeout {NFT_TIMEOUT} }}"]
    try:
        subprocess.run(cmd, check=True, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
        logging.info(f"[BANNED] IP {ip} 已打入内核黑洞集")
    except Exception as e:
        logging.error(f"nftables 下发失败: {e}")

async def handle_ssh(reader: asyncio.StreamReader, writer: asyncio.StreamWriter):
    peer = writer.get_extra_info("peername")
    ip = peer[0] if peer else "unknown"
    try:
        # 伪装 Ubuntu 官方 OpenSSH Banner
        writer.write(b"SSH-2.0-OpenSSH_8.9p1 Ubuntu-3ubuntu0.1\r\n")
        await writer.drain()
        
        # 严格 5 秒超时，防 Slowloris 悬挂
        client_banner = await asyncio.wait_for(reader.readline(), timeout=5.0)
        logging.info(f"[SSH-PROBE] 来自 {ip} | 指纹: {client_banner.strip().decode('latin1', 'ignore')}")
        ban_ip_kernel(ip)
    except Exception:
        pass
    finally:
        writer.close()
        await writer.wait_closed()

async def handle_http(reader: asyncio.StreamReader, writer: asyncio.StreamWriter):
    peer = writer.get_extra_info("peername")
    ip = peer[0] if peer else "unknown"
    try:
        # 限制读取前 4KB，防止恶意大报文消耗缓冲区
        raw_request = await asyncio.wait_for(reader.read(4096), timeout=5.0)
        if raw_request:
            first_line = raw_request.split(b"\r\n")[0].decode("latin1", "ignore")
            logging.info(f"[HTTP-PROBE] 来自 {ip} | 探针载荷: {first_line}")
            ban_ip_kernel(ip)
        
        response = b"HTTP/1.1 404 Not Found\r\nContent-Length: 0\r\nConnection: close\r\n\r\n"
        writer.write(response)
        await writer.drain()
    except Exception:
        pass
    finally:
        writer.close()
        await writer.wait_closed()

async def main():
    # 生产中可用高位端口或直接监听对外伪装端口
    server_ssh = await asyncio.start_server(handle_ssh, "0.0.0.0", 2222)
    server_http = await asyncio.start_server(handle_http, "0.0.0.0", 8080)
    logging.info("诱捕 Facade 服务已启动，开始捕获恶意探针...")
    async with server_ssh, server_http:
        await asyncio.gather(server_ssh.serve_forever(), server_http.serve_forever())

if __name__ == "__main__":
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        pass

# 1. 确保基础 filter 表与 input 链就位 [S1][S3]
sudo nft add table ip filter
sudo nft add chain ip filter input { type filter hook input priority 0 \; policy accept \; }

# 2. 创建带自动超时老化机制的动态集合 [S2][S4]
sudo nft add set ip filter bad_guys '{ type ipv4_addr; flags timeout; timeout 86400s; }'

# 3. 将黑名单封禁规则插入到 input 链的最前列 [S3]
sudo nft insert rule ip filter input ip saddr @bad_guys drop

sudo nft list set ip filter bad_guys

/var/log/honeypot.log {
    daily
    rotate 7
    missingok
    notifempty
    compress
    maxsize 50M
}