|
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
"""
|
|
复现「Zeek 有攻击 HTTP 记录但 Suricata 无告警」问题 —— 真实扫描器的攻击请求形态
|
|
|
|
背景（2026-08-24 通过 ES 日志 + Zeek 离线分析确认）：
|
|
真实扫描器（如源 1.13.9.36）的攻击请求，是【完整正常的 HTTP 流】：
|
|
完整三次握手 + 正常 seq + 服务器回 404 响应（如 /.git/index 等路径）。
|
|
之前「半开/seq 错乱状态下发 HTTP 数据」的认知是过拟合的畸形流，已纠正。
|
|
|
|
本脚本用真实客户端（内核 TCP 栈）发送攻击请求，完整握手、正常 seq、
|
|
收到服务器响应 —— 精确复现真实扫描器攻击请求的流量形态。
|
|
|
|
用法：
|
|
# 默认 6 个攻击路径轮流发送
|
|
sudo python3 repro_http_stream.py -i 10.10.168.197
|
|
|
|
# 指定端口 / 单个路径 / 次数
|
|
sudo python3 repro_http_stream.py -i 10.10.168.197 -p 8888 -P "/v1/api-docs" -c 3
|
|
|
|
参数说明：
|
|
-i, --ip 目标 IP（受害主机），必填
|
|
-p, --port 目标端口，默认 80
|
|
-P, --path 攻击 URI 路径，指定后只发该路径；不指定则默认 6 个路径轮流发送
|
|
--host 请求 Host 头（默认用 --ip）
|
|
-c, --count 每个路径发送次数，默认 1
|
|
|
|
源 IP 自动探测本机出口 IP。
|
|
|
|
默认路径列表（不指定 -P 时轮流发送）：
|
|
/.git/index /.svn/entries /WEB-INF/web.xml
|
|
/druid/index.html /v1/api-docs /prod-api/druid/index.html
|
|
"""
|
|
|
|
DEFAULT_PATHS = [
|
|
"/.git/index",
|
|
"/.svn/entries",
|
|
"/WEB-INF/web.xml",
|
|
"/druid/index.html",
|
|
"/v1/api-docs",
|
|
"/prod-api/druid/index.html",
|
|
]
|
|
|
|
import argparse
|
|
import socket
|
|
import time
|
|
|
|
|
|
def get_local_ip(dst_ip: str) -> str:
|
|
"""获取本机到目标的路由出口 IP"""
|
|
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
|
|
try:
|
|
s.connect((dst_ip, 9))
|
|
return s.getsockname()[0]
|
|
except Exception:
|
|
return "127.0.0.1"
|
|
finally:
|
|
s.close()
|
|
|
|
|
|
def build_http_request(path: str, host: str, dport: int) -> bytes:
|
|
if ":" in host and not host.startswith("["):
|
|
host_header = f"{host}:{dport}"
|
|
else:
|
|
host_header = host
|
|
req = (
|
|
f"GET {path} HTTP/1.1\r\n"
|
|
f"Host: {host_header}\r\n"
|
|
"User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
|
|
"AppleWebKit/537.36 (KHTML, like Gecko) Chrome/148.0.0.0 Safari/537.36\r\n"
|
|
"Accept: */*\r\n"
|
|
"Connection: close\r\n"
|
|
"\r\n"
|
|
)
|
|
return req.encode("latin-1")
|
|
|
|
|
|
def send_http(dst_ip: str, dport: int, path: str, host: str) -> str:
|
|
"""发送完整 HTTP 请求（走内核 TCP 栈），返回响应状态码或错误信息。"""
|
|
req = build_http_request(path, host, dport)
|
|
try:
|
|
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
|
s.settimeout(3)
|
|
s.connect((dst_ip, dport))
|
|
s.sendall(req)
|
|
status = None
|
|
try:
|
|
resp = s.recv(4096)
|
|
if resp:
|
|
first_line = resp.split(b"\r\n", 1)[0].decode("latin-1", "ignore")
|
|
parts = first_line.split(" ", 2)
|
|
if len(parts) >= 2:
|
|
status = parts[1]
|
|
except socket.timeout:
|
|
pass
|
|
s.close()
|
|
return status or "(无响应)"
|
|
except OSError as e:
|
|
return f"ERR:{e}"
|
|
|
|
|
|
def main():
|
|
ap = argparse.ArgumentParser(description="复现真实扫描器攻击请求（完整 HTTP 流）")
|
|
ap.add_argument("-i", "--ip", required=True, help="目标 IP（受害主机）")
|
|
ap.add_argument("-p", "--port", type=int, default=80, help="目标端口，默认 80")
|
|
ap.add_argument("-P", "--path", default=None,
|
|
help="攻击 URI 路径，指定后只发该路径；不指定则默认 6 个路径轮流发送")
|
|
ap.add_argument("--host", default=None, help="Host 头（默认用 --ip）")
|
|
ap.add_argument("-c", "--count", type=int, default=1, help="每个路径发送次数，默认 1")
|
|
args = ap.parse_args()
|
|
|
|
dst_ip = args.ip
|
|
dport = args.port
|
|
src_ip = get_local_ip(dst_ip)
|
|
host = args.host or dst_ip
|
|
|
|
paths = [args.path] if args.path else DEFAULT_PATHS
|
|
|
|
print(f"[*] 源 IP : {src_ip}")
|
|
print(f"[*] 目标 : {dst_ip}:{dport}")
|
|
print(f"[*] 攻击URI : {', '.join(paths)}")
|
|
print(f"[*] 次数 : x{args.count}")
|
|
print()
|
|
|
|
for i in range(args.count):
|
|
for path in paths:
|
|
status = send_http(dst_ip, dport, path, host)
|
|
print(f" -> {path} (resp={status})")
|
|
time.sleep(0.1)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|