中小企业零成本开源安全闭环｜全套可落地实操部署方案

适配前文博文架构｜100%开源｜低配服务器可跑｜全自动安全闭环｜自建IP信誉系统

本方案基于前文完整的开源安全体系，提供可直接复制部署、生产可用、无冗余、避坑优化的实操步骤。全程无需商业软件、无需大数据集群、无需专职安全运维，单台4C8G/4H8G虚拟机即可承载所有组件，完美适配中小企业预算少、人手缺、求稳定的核心需求。

部署整体顺序（核心原则：先采集、后研判、再处置）

系统环境准备 → 日志集中采集（Graylog）→ 全网风险感知（HFish+Suricata+Wazuh）→ 自建轻量化IP信誉系统 → 自动化封禁调度（SecAutoBan）→ 设备联动下发（防火墙/路由器ACL）→ 告警降噪+合规审计 → 常态化运维调优

---

一、基础环境标准化准备（生产必做）

1.1 硬件与系统要求

• 服务器配置：最低4C8G、40G固态磁盘（优先SSD，保障日志读写稳定）

• 操作系统：Ubuntu 22.04 / Debian 12（长期支持、兼容性最强）

• 环境要求：关闭Swap、开启时间同步、固定静态IP、关闭无用系统服务

1.2 系统初始化命令（可直接复制）

# 更新系统依赖
apt update && apt upgrade -y
# 安装基础必备组件
apt install -y python3 python3-pip git ipset iptables curl wget unzip supervisor cron openssh-server
# 关闭swap（永久生效）
swapoff -a
sed -i '/swap/d' /etc/fstab
# 时区同步
timedatectl set-timezone Asia/Shanghai
apt install -y systemd-timesyncd
systemctl enable --now systemd-timesyncd

---

二、第一层能力：全网日志归一化采集（Graylog）

核心作用：收敛防火墙、WAF、路由器、IDS、服务器、蜜罐所有日志，完成字段归一化、攻击IP聚合筛选，是整个安全体系的数据源底座。

2.1 极简部署方式（Docker一键部署）

采用官方稳定镜像，规避编译报错、依赖冲突问题，中小企业首选。

2.2 核心生产配置（关键避坑）

1. 开启字段归一化Pipeline：统一所有设备源IP为 src_ip，解决多设备字段不统一无法聚合的问题，Pipeline规则沿用博文标准规则；

2. 配置日志生命周期：设置日志留存6个月（满足等保），自动清理过期日志，防止磁盘占满；

3. 关闭无用日志采集：过滤内网正常访问、重复冗余日志，提前降噪，避免后续告警泛滥；

4. 创建专属安全数据流：单独创建Security数据流，收纳所有安全设备、蜜罐、主机安全日志，与业务日志隔离。

2.3 可疑IP筛选定时任务（准实时落地）

放弃Graylog原生不稳定Webhook，部署前文提供的 graylog_fetch_suspicious_ip.py定时脚本，配置5分钟执行一次。

筛选阈值（生产最优配置）：5分钟内同一src_ip攻击行为≥3次，判定为可疑IP，自动推送至后续处置服务，杜绝单条日志误判。

---

三、第二层能力：全网风险感知部署（三层主动探测）

补齐网络、主机、蜜罐三大感知维度，实现全方位攻击捕获，为IP信誉系统提供内部高权重真实攻击数据。

3.1 HFish开源蜜罐（最高可信度攻击源）

部署模式：公网节点+内网双节点部署

• 公网蜜罐：模拟SSH、FTP、MySQL、Web服务，捕获外网扫描、爆破攻击；

• 内网蜜罐：捕获内网横向移动、入侵渗透、内网爆破行为（内网告警优先级最高）。

关键配置：开启Webhook推送，将所有蜜罐攻击事件推送至本地统一调度接口，不直接封禁，进入信誉研判流水线。

3.2 Suricata开源IDS（旁路入侵检测）

核心优势：旁路镜像部署，不串接业务、无业务中断风险，零故障率。

• 加载ET Open免费开源规则集，检测端口扫描、漏洞探测、恶意请求、暴力破解；

• 所有告警日志以GELF格式推送至Graylog，统一聚合分析；

• 仅做检测告警，不主动阻断，保障业务稳定。

3.3 Wazuh主机安全监控（主机层防护）

全网所有Linux/Windows业务服务器部署Wazuh轻量Agent，核心采集：

• 系统账号变更、定时任务篡改、关键文件修改；

• SSH、RDP暴力破解、异常登录行为；

• 主机恶意进程、病毒木马行为。

所有主机安全日志统一上报Graylog，暴力破解IP纳入可疑IP筛选池，异常篡改行为仅钉钉告警、不自动处置。

---

四、第三层能力：自建轻量化IP信誉系统（体系核心）

替换单一Maltrail情报核验，实现内外结合打分、分级研判，彻底解决误封、漏封问题，中小企业专属轻量架构（Redis+SQLite，无需大数据组件）。

4.1 数据源接入配置

1. 外部开源情报定时同步：定时拉取Maltrail、FireHOL、Spamhaus恶意IP网段、AbuseIPDB免费情报，多源交叉验证；

2. 内部数据全量接入：同步HFish蜜罐记录、Suricata告警、Wazuh主机攻击、Graylog聚合攻击日志、历史封禁记录。

4.2 打分模型落地配置（直接套用）

• 内部攻击行为高权重加分，外部情报低权重辅助加分；

• 全局白名单强制优先：内网、运维IP、CDN、云厂商网段、合作客户IP永久可信；

• 时间衰减机制：7天分数衰减20%，30天无攻击自动清零，规避IP复用误判。

4.3 分级处置规则（生产强制标准）

• 0-20分（可信）：直接放行；

• 21-50分（可疑）：仅钉钉告警、日志记录，禁止自动封禁；

• 51-80分（高风险）：TTL限时8小时封禁；

• 81-100分（极高风险）：全网封禁+高危钉钉@全员告警。

4.4 接口对接改造

修改SecAutoBan钩子脚本，废弃单一Maltrail判定，优先调用自建IP信誉接口，实现精准分级研判。

---

五、第四层能力：自动化封禁调度与设备联动

以SecAutoBan为执行中枢，实现本机+硬件设备双层封禁、自动解封、无人工干预。

5.1 SecAutoBan+Maltrail基础部署

沿用前文Docker Compose一键部署方案，统一配置：

• 封禁默认TTL 8小时，禁止永久封禁；

• 全局唯一白名单，所有组件共用，避免策略冲突；

• 开启钉钉分级告警，关闭冗余日志推送。

5.2 统一调度服务改造（hfish-push）

改造Flask调度服务，双接口接收：HFish蜜罐事件、Graylog可疑IP事件，所有攻击IP统一送入SecAutoBan研判流水线。

5.3 硬件设备联动（核心落地点）

新增设备上下线钩子脚本，实现封禁/解封双向联动：

• 支持华为、H3C、普通路由器SSH协议下发ACL黑名单；

• 跟随SecAutoBan TTL策略，到期自动删除硬件设备黑名单，杜绝永久封禁；

• 设备联动异常降级：硬件设备离线时，仅保留本机ipset封禁，不影响整体体系运行。

---

六、第五层能力：告警降噪与合规审计闭环

6.1 告警分级降噪

• P0高危：内网蜜罐捕获、主机文件篡改、大规模爆破 → 钉钉@全员告警；

• P1中危：外网高风险IP封禁、多次Web攻击 → 普通钉钉推送；

• P2低危：单次扫描、偶发探测 → 仅日志留存，不推送告警。

6.2 全链路审计闭环

• 所有攻击、研判、封禁、解封、告警事件全部回写Graylog；

• 搭建Graylog安全态势仪表盘，直观展示TOP攻击IP、封禁统计、攻击类型分布；

• 配置定时脚本，每周自动生成安全简报，钉钉推送运维复盘。

---

七、事前加固配套部署（从源头减少攻击面）

自动化是事后处置，事前加固才是安全根本，中小企业轻量化落地：

1. 自动备份系统：部署BorgBackup定时增量备份，保障勒索病毒、入侵兜底恢复能力；

2. 基线定时检查：开源OpenSCAP定期扫描服务器配置、密码策略、权限合规性；

3. 资产监控：定时脚本探测公网端口、子域名变更、证书到期，异常自动告警。

---

八、生产核心防护机制（零误封、高稳定）

1. 全链路安全降级：任意组件故障（Graylog离线、情报服务异常、设备失联），系统自动停止自动封禁，仅保留日志，不影响业务；

2. 禁止盲自动化：仅网络攻击、扫描、爆破支持自动处置，漏洞、配置、账号风险仅人工修复；

3. 内部数据优先：自身环境真实攻击权重高于所有开源情报，适配自有业务场景；

4. 无永久黑名单：所有封禁均带TTL时效，配合时间衰减，彻底解决IP复用误判问题。

---

九、落地部署顺序（7天落地计划表）

• 第1天：系统初始化、部署Graylog、完成日志归一化配置；

• 第2天：部署HFish蜜罐、Suricata旁路IDS，打通日志上报；

• 第3天：全网部署Wazuh Agent，补齐主机安全采集；

• 第4天：部署SecAutoBan+Maltrail，调试基础封禁能力；

• 第5天：搭建自建IP信誉系统，调试打分、分级规则；

• 第6天：调试设备联动封禁、告警降噪、审计回写；

• 第7天：整体联调、模拟攻击测试、策略优化、常态化定时任务上线。

---

十、方案落地最终收益

1. 零成本替代商业安全体系：替代商业威胁情报、SOC、自动封禁平台、入侵检测设备，节省数万安全预算；

2. 无人值守自动化防护：90%的攻击研判、处置、日志统计全自动完成，解决中小企业无专职安全人员痛点；

3. 极低误封、高精准防护：内外双源研判+分级处置+时间衰减，兼顾业务稳定与安全防护；

4. 完全满足等保合规：日志留存、攻击溯源、处置审计、风险研判全流程可追溯；

5. 自适应迭代升级：自建信誉库持续沉淀业务专属攻击特征，越用越精准，远超通用商业情报。

---

十一、全套落地源码｜脚本 + DockerCompose + 定时任务（生产直接用）

本节为整套安全闭环最终可运行资产，包含：统一编排 docker-compose、Graylog 拉取可疑IP脚本、信誉系统对接钩子、设备联动封禁/解封脚本、hfish-push 完整服务、系统定时任务。全部兼容前文架构，无需二次修改。

11.1 整体目录结构（严格按此创建）

/opt/auto-sec-stack
├── docker-compose.yml
├── .env
├── maltrail/
│   └── maltrail.conf
├── secautoban/
│   ├── config.ini
│   ├── whitelist.conf
│   └── hook/
│       ├── maltrail_check_hook.py
│       ├── device_ban_hook.py
│       └── device_unban_hook.py
├── hfish/
│   └── hfish_push.py
├── scripts/
│   └── graylog_fetch_suspicious_ip.py
└── logs/

11.2 .env 全局环境变量

# Webhook 密钥（务必改为随机强密码）
HFISH_WEBHOOK_TOKEN=SecStack_2026_Random_Strong_Pwd
# 钉钉告警（无则留空）
DINGDING_WEBHOOK=
DINGDING_SECRET=
# 时区
TZ=Asia/Shanghai

11.3 maltrail/maltrail.conf

[server]
host = 127.0.0.1
port = 8337

[update]
interval = 3600

[sensor]
enabled = false

11.4 secautoban/whitelist.conf（全局唯一白名单）

# 内网网段
127.0.0.1
10.0.0.0/8
172.16.0.0/12
192.168.0.0/16
# 运维出口公网IP（自行补充）
# 业务、CDN、合作IP

11.5 secautoban/config.ini（开启执行钩子、适配信誉系统）

[global]
log_path = /var/log/SecAutoBan
ipset_name = SecAutoBan
ban_time = 28800
max_ban_count = 8192

[ssh]
enable = true
log_file = /var/log/auth.log
trigger_count = 5

[nginx]
enable = true
log_file = /var/log/nginx/access.log
trigger_count = 10

[hook]
enable = true
# 信誉复核钩子（替代原单纯 Maltrail 判断）
ban_hook_script = /etc/SecAutoBan/hook/maltrail_check_hook.py
# 执行多设备下发钩子
exec_ban_hook = /etc/SecAutoBan/hook/device_ban_hook.py
exec_unban_hook = /etc/SecAutoBan/hook/device_unban_hook.py

[alert]
enable = true
alert_type = dingtalk
dingtalk_webhook = ${DINGDING_WEBHOOK}
dingtalk_secret = ${DINGDING_SECRET}

[external_blacklist]
enable = false

11.6 secautoban/hook/maltrail_check_hook.py（【信誉系统复核钩子】核心改造）

不再只依赖 Maltrail，优先调用自建IP信誉服务，实现可信/可疑/高危分级，控制是否封禁。

#!/usr/bin/env python3
import sys
import requests
import logging

# 自建IP信誉服务接口
REPUTE_API = "http://reputation:9090/ip_score"
TIMEOUT = 5

logging.basicConfig(filename="/var/log/secautoban_repute.log", level=logging.INFO,
                    format="%(asctime)s %(levelname)s %(message)s")

def main():
    if len(sys.argv) < 2:
        sys.exit(1)
    ip = sys.argv[1]

    try:
        resp = requests.get(REPUTE_API, params={"ip": ip}, timeout=TIMEOUT)
        data = resp.json()
        score = data.get("score", 0)
        level = data.get("level", "trusted")
        reason = data.get("reason", "")

        logging.info(f"IP:{ip} Score:{score} Level:{level} Reason:{reason}")

        # 规则：仅高风险、极高风险才允许封禁
        if score > 50:
            sys.exit(0)  # 允许封禁
        else:
            sys.exit(1)  # 不封禁（可信/可疑只告警）

    except Exception as e:
        logging.error(f"Repute API error: {str(e)}")
        sys.exit(1)

if __name__ == "__main__":
    main()

11.7 secautoban/hook/device_ban_hook.py（硬件防火墙/路由器 ACL 封禁下发）

#!/usr/bin/env python3
import sys
import paramiko
import logging

logging.basicConfig(filename="/var/log/device_ban.log", level=logging.INFO)
BAN_IP = sys.argv[1]

# 根据自己设备修改：H3C/华为/思科 命令适配
DEV_LIST = [
    {
        "ip": "192.168.1.1",
        "user": "admin",
        "pwd": "your-pass",
        "cmd": "acl number 3999\n rule deny ip source {ip} 0\n save"
    }
]

def ban(ip):
    for d in DEV_LIST:
        try:
            ssh = paramiko.SSHClient()
            ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
            ssh.connect(d["ip"], username=d["user"], password=d["pwd"], timeout=8)
            for line in d["cmd"].format(ip=ip).splitlines():
                ssh.exec_command(line.strip())
            ssh.close()
            logging.info(f"SUCCESS ban {ip} on {d['ip']}")
        except Exception as e:
            logging.error(f"FAIL ban {ip} {d['ip']}: {str(e)}")

if __name__ == "__main__":
    ban(BAN_IP)

11.8 secautoban/hook/device_unban_hook.py（到期自动解封硬件ACL）

#!/usr/bin/env python3
import sys
import paramiko
import logging

logging.basicConfig(filename="/var/log/device_unban.log", level=logging.INFO)
BAN_IP = sys.argv[1]

DEV_LIST = [
    {
        "ip": "192.168.1.1",
        "user": "admin",
        "pwd": "your-pass",
        "cmd": "acl number 3999\n undo rule deny ip source {ip} 0\n save"
    }
]

def unban(ip):
    for d in DEV_LIST:
        try:
            ssh = paramiko.SSHClient()
            ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
            ssh.connect(d["ip"], username=d["user"], password=d["pwd"], timeout=8)
            for line in d["cmd"].format(ip=ip).splitlines():
                ssh.exec_command(line.strip())
            ssh.close()
            logging.info(f"SUCCESS unban {ip} on {d['ip']}")
        except Exception as e:
            logging.error(f"FAIL unban {ip} {d['ip']}: {str(e)}")

if __name__ == "__main__":
    unban(BAN_IP)

11.9 hfish/hfish_push.py（统一接收 HFish + Graylog 双源事件）

#!/usr/bin/env python3
from flask import Flask, request, jsonify
import subprocess
import logging
import os

app = Flask(__name__)
logging.basicConfig(filename="/var/log/hfish_push.log", level=logging.INFO)

API_TOKEN = os.getenv("HFISH_WEBHOOK_TOKEN", "")
SECAUTOBAN_BIN = "/usr/bin/SecAutoBan"

def submit_ip(ip, source, attack_type):
    logging.info(f"[PUSH] IP:{ip} SRC:{source} TYPE:{attack_type}")
    try:
        subprocess.run([SECAUTOBAN_BIN, "ban", "--ip", ip], timeout=10)
    except Exception as e:
        logging.error(f"[PUSH ERR] {ip} {str(e)}")

# 蜜罐事件
@app.route("/hfish/event", methods=["POST"])
def hfish_event():
    if request.headers.get("X-Token", "") != API_TOKEN:
        return jsonify({"code":403}),403
    d = request.get_json() or {}
    ip = d.get("src_ip")
    if ip:
        submit_ip(ip, "hfish", "honeypot_attack")
    return jsonify({"ok":1})

# Graylog 聚合可疑IP事件
@app.route("/graylog/event", methods=["POST"])
def graylog_event():
    if request.headers.get("X-Token", "") != API_TOKEN:
        return jsonify({"code":403}),403
    d = request.get_json() or {}
    ip = d.get("src_ip")
    if ip:
        submit_ip(ip, "graylog", "multi_device_attack")
    return jsonify({"ok":1})

if __name__ == "__main__":
    app.run("0.0.0.0", 8889, False)

11.10 scripts/graylog_fetch_suspicious_ip.py（5分钟聚合筛选可疑IP）

#!/usr/bin/env python3
import requests, json, logging, os

# ========== 配置区 ==========
GRAYLOG_API = "http://127.0.0.1:9000/api"
GRAYLOG_USER = "admin"
GRAYLOG_PASS = "你的密码"
STREAM_ID = "你的安全数据流ID"
TIME_WINDOW = 300
MIN_COUNT = 3
PUSH_URL = "http://hfish-push:8889/graylog/event"
TOKEN = os.getenv("HFISH_WEBHOOK_TOKEN","SecStack_2026_Random_Strong_Pwd")
WHITE_LIST_PATH = "/opt/auto-sec-stack/secautoban/whitelist.conf"
# ============================

logging.basicConfig(filename="/var/log/graylog_fetch.log", level=logging.INFO)

def load_white():
    wl = set()
    try:
        with open(WHITE_LIST_PATH,"r") as f:
            for line in f:
                line=line.strip()
                if line and not line.startswith("#"):
                    wl.add(line)
    except:
        pass
    return wl

def fetch():
    headers = {"Content-Type":"application/json"}
    query = '''
    (attack_type:SQLi OR attack_type:XSS OR attack_type:PathTraversal OR attack_type:RCE)
    OR (signature_id:>0)
    OR action:(drop reject)
    OR message:("authentication failed" "login fail")
    '''
    payload = {
        "query": query,
        "streams": [STREAM_ID],
        "range": {"type":"relative","from":TIME_WINDOW*1000},
        "aggregation":{"group_by":[{"type":"field","field":"src_ip"}],"metrics":[{"type":"count"}]}
    }
    res = requests.post(f"{GRAYLOG_API}/search/aggregation",
                        auth=(GRAYLOG_USER,GRAYLOG_PASS),headers=headers,json=payload,timeout=20)
    buckets = res.json().get("aggregation",{}).get("buckets",[])
    wl = load_white()
    out = []
    for b in buckets:
        ip = b.get("key")
        cnt = b.get("count",0)
        if ip and ip not in wl and cnt >= MIN_COUNT:
            out.append(ip)
    return out

def push(ip):
    headers = {"X-Token":TOKEN,"Content-Type":"application/json"}
    body = {"src_ip":ip,"attack_type":"multi_device_scan_brute"}
    requests.post(PUSH_URL,json=body,headers=headers,timeout=10)

def main():
    ips = fetch()
    logging.info(f"Fetch suspicious ips: {ips}")
    for ip in ips:
        push(ip)

if __name__=="__main__":
    main()

11.11 docker-compose.yml（整套三组件+调度一键编排）

version: '3.8'

services:
  maltrail:
    image: python:3.11-slim
    container_name: maltrail
    restart: always
    volumes:
      - ./maltrail/maltrail.conf:/opt/maltrail/conf/maltrail.conf
      - maltrail_data:/opt/maltrail
      - ./logs:/var/log
    working_dir: /opt/maltrail
    command: >
      bash -c "git clone https://github.com/stamparm/maltrail.git /opt/maltrail 2>/dev/null || true &&
               python3 install.py &&
               python3 maltrail-server.py"
    networks:
      - sec_net
    expose:
      - "8337"
    environment:
      TZ: ${TZ}

  secautoban:
    image: python:3.11-slim
    container_name: secautoban
    restart: always
    privileged: true
    cap_add:
      - NET_ADMIN
      - NET_RAW
    volumes:
      - ./secautoban/config.ini:/etc/SecAutoBan/config.ini
      - ./secautoban/whitelist.conf:/etc/SecAutoBan/whitelist.conf
      - ./secautoban/hook:/etc/SecAutoBan/hook
      - secautoban_data:/etc/SecAutoBan
      - ./logs:/var/log
      - /var/log:/host/var/log:ro
    working_dir: /opt/SecAutoBan
    command: >
      bash -c "git clone https://github.com/secauto/SecAutoBan.git /opt/SecAutoBan 2>/dev/null || true &&
               python3 install.py &&
               chmod +x /etc/SecAutoBan/hook/*.py &&
               python3 main.py"
    networks:
      - sec_net
    expose:
      - "9000"
    environment:
      TZ: ${TZ}
      DINGDING_WEBHOOK: ${DINGDING_WEBHOOK}
      DINGDING_SECRET: ${DINGDING_SECRET}

  hfish:
    image: debian:12-slim
    container_name: hfish
    restart: always
    volumes:
      - hfish_data:/opt/hfish
      - ./logs:/var/log
    working_dir: /opt/hfish
    command: >
      bash -c "apt update && apt install -y wget unzip &&
               wget -q https://github.com/hfish-security/HFish/releases/latest/download/hfish-linux-amd64.zip -O hfish.zip &&
               unzip -o hfish.zip &&
               ./hfish-server"
    networks:
      - sec_net
    ports:
      - "2222:2222"
      - "2121:2121"
      - "2323:2323"
      - "33060:33060"
      - "8080:8080"
    expose:
      - "9001"
    environment:
      TZ: ${TZ}

  hfish-push:
    image: python:3.11-slim
    container_name: hfish-push
    restart: always
    volumes:
      - ./hfish/hfish_push.py:/opt/hfish_push.py
      - ./logs:/var/log
    working_dir: /opt
    command: >
      bash -c "pip3 install flask requests && python3 hfish_push.py"
    networks:
      - sec_net
    expose:
      - "8889"
    environment:
      TZ: ${TZ}
      HFISH_WEBHOOK_TOKEN: ${HFISH_WEBHOOK_TOKEN}

networks:
  sec_net:
    driver: bridge

volumes:
  maltrail_data:
  secautoban_data:
  hfish_data:

11.12 系统定时任务（crontab）

# 每5分钟执行一次Graylog可疑IP抓取
*/5 * * * * /usr/bin/python3 /opt/auto-sec-stack/scripts/graylog_fetch_suspicious_ip.py >> /var/log/graylog_cron.log 2>&1

11.13 Graylog Pipeline 归一规则（直接导入）

rule normalize_src_ip
when true then
  set_field("src_ip", coalesce(
    $message.src,
    $message.source_ip,
    $message.srcaddr,
    $message."src-ip",
    $message.SourceIP,
    $message.src_ip
  ));
end

---

十二、上线最终检查清单（投产必做）

• 所有脚本已授权可执行：chmod +x *.py

• 修改设备登录密码、Graylog账号密码、Webhook密钥

• 白名单已填入内网、运维、CDN、业务IP

• 硬件ACL规则为空、避免冲突

• 先关闭自动封禁执行，观察3天日志无误判再开启

• 确认所有组件日志正常落盘、Graylog正常收录