在企业网络安全建设中，日志分析是威胁检测的核心环节。传统的手工审计方式难以应对海量日志数据，据统计，中型企业每日产生的安全日志可达千万条级别。本文将分享一套基于Python与ELK Stack（Elasticsearch + Logstash + Kibana）构建的自动化日志分析系统方案，实现实时威胁检测与告警闭环。

一、系统架构与技术选型

网络安全日志来源广泛，包括防火墙、IDS/IPS、Web应用防火墙、服务器系统日志、应用访问日志等。系统架构采用分层设计：采集层用Filebeat轻量级Agent收集日志，处理层用Logstash做过滤和字段映射，存储层用Elasticsearch做全文索引，分析层用Python脚本定时检测异常模式，展示层用Kibana做可视化。

Logstash负责将多源异构日志归一化为统一JSON格式，通过Grok正则匹配解析关键字段。Elasticsearch采用ILM（Index Lifecycle Management）策略，热数据存SSD保留7天，温数据转HDD保留30天，冷数据归档存储90天，平衡检索性能与存储成本。

二、日志采集与Logstash配置

承恒信息科技在为某企业部署网络安全监控方案时，采用Filebeat+Logstash组合实现多源日志统一采集。以下是Logstash配置文件，解析Nginx访问日志并提取安全相关字段：

# logstash/conf.d/nginx_security.conf
input {
  beats {
    port => 5044
    codec => json
  }
}

filter {
  if [fileset][name] == "access" {
    grok {
      match => { "message" => "%{IPORHOST:client_ip} - %{DATA:user} \[%{HTTPDATE:timestamp}\] \"%{WORD:method} %{URIPATHPARAM:path} HTTP/%{NUMBER:http_version}\" %{NUMBER:status} %{NUMBER:bytes} \"%{DATA:referrer}\" \"%{DATA:agent}\"" }
    }
    # SQL注入特征检测
    if [path] =~ /(\bunion\b|\bselect\b|\binsert\b| ["sql_injection_suspect"] }
    }
    # XSS特征检测
    if [path] =~ /(%3Cscript|javascript:|onerror=)/i {
      mutate { add_tag => ["xss_suspect"] }
    }
    # 路径穿越检测
    if [path] =~ /(\.\.\/|\.\.%2f|%2e%2e)/i {
      mutate { add_tag => ["path_traversal_suspect"] }
    }
    # IP地理信息 enrichment
    geoip { source => "client_ip" }
  }
  date { match => ["timestamp", "dd/MMM/yyyy:HH:mm:ss Z"] }
}

output {
  elasticsearch {
    hosts => ["http://es-node1:9200", "http://es-node2:9200"]
    index => "security-logs-%{+YYYY.MM.dd}"
    template_name => "security-template"
  }
}

该配置在日志解析阶段即完成基础威胁标记，可疑请求自动打标签，后续在Elasticsearch中可通过标签快速筛选定位。Grok正则的解析性能约为每秒3万条日志，可满足中型企业实时采集需求。

三、Python异常检测与自动化告警

Logstash的规则检测属于静态特征匹配，对于未知的异常行为模式，需要Python脚本基于统计分析和阈值判断来动态识别。例如同一IP在短时间内大量请求不同URL、特定User-Agent异常高频访问等。

import requests
from datetime import datetime, timedelta
from elasticsearch import Elasticsearch

es = Elasticsearch(["http://es-node1:9200", "http://es-node2:9200"])
ALERT_WEBHOOK = "https://qyapi.weixin.com/cgi-bin/webhook/send?key=xxx"

def detect_brute_force():
    """检测SSH暴力破解：同一IP 5分钟内失败超过20次"""
    now = datetime.utcnow()
    query = {
        "size": 0,
        "query": {
            "bool": {
                "filter": [
                    {"range": {"@timestamp": {"gte": now - timedelta(minutes=5)}}},
                    {"match": {"message": "Failed password"}}
                ]
            }
        },
        "aggs": {
            "ip_count": {
                "terms": {"field": "client_ip", "size": 50, "min_doc_count": 20}
            }
        }
    }
    result = es.search(index="security-logs-*", body=query)
    buckets = result["aggregations"]["ip_count"]["buckets"]
    for bucket in buckets:
        ip = bucket["key"]
        count = bucket["doc_count"]
        if count >= 20:
            send_alert(f"SSH暴力破解告警", f"IP: {ip}, 5分钟内失败{count}次")
            auto_block_ip(ip)

def detect_cc_attack():
    """检测CC攻击：同一IP 1分钟内请求超过200次"""
    query = {
        "size": 0,
        "query": {"range": {"@timestamp": {"gte": "now-1m"}}},
        "aggs": {"ip_count": {"terms": {"field": "client_ip", "size": 20, "min_doc_count": 200}}}
    }
    result = es.search(index="security-logs-*", body=query)
    for bucket in result["aggregations"]["ip_count"]["buckets"]:
        send_alert(f"CC攻击告警", f"IP: {bucket['key']}, 1分钟请求{bucket['doc_count']}次")

def send_alert(title, content):
    payload = {"msgtype": "markdown", "markdown": {"content": f"## {title}\n> {content}\n> 时间: {datetime.now()}"}}
    requests.post(ALERT_WEBHOOK, json=payload, timeout=10)

if __name__ == "__main__":
    detect_brute_force()
    detect_cc_attack()

该脚本通过Elasticsearch聚合查询实现异常IP检测，结合企业微信Webhook实现实时告警推送。检测脚本通过crontab每分钟执行一次，暴力破解阈值可根据业务场景动态调整。在实际部署中，该方案将安全事件平均发现时间从原来的4小时缩短至1分钟以内。

四、Kibana可视化与运维指标

Kibana Dashboard用于安全态势可视化，核心面板包括：实时攻击地图（基于GeoIP展示攻击源地理分布）、Top10攻击源IP排行、攻击类型趋势图、威胁等级分布饼图等。通过Kibana的告警功能（Alerting）可设置基于ES查询条件的阈值告警，与Python脚本形成双重保障。

# Docker Compose 部署ELK集群
version: '3.8'
services:
  elasticsearch:
    image: docker.elastic.co/elasticsearch/elasticsearch:8.12.0
    environment:
      - cluster.name=security-cluster
      - discovery.type=single-node
      - xpack.security.enabled=true
      - ES_JAVA_OPTS=-Xms4g -Xmx4g
    volumes:
      - es_data:/usr/share/elasticsearch/data
    ports: ["9200:9200"]

  logstash:
    image: docker.elastic.co/logstash/logstash:8.12.0
    volumes:
      - ./logstash/conf.d:/usr/share/logstash/pipeline
    depends_on: [elasticsearch]
    ports: ["5044:5044"]

  kibana:
    image: docker.elastic.co/kibana/kibana:8.12.0
    environment:
      - ELASTICSEARCH_HOSTS=http://elasticsearch:9200
    ports: ["5601:5601"]
    depends_on: [elasticsearch]

volumes:
  es_data:

系统关键性能指标：Elasticsearch集群索引吞吐量稳定在每秒2.5万条日志，查询延迟P99 < 200ms，3节点集群存储容量可支撑30天日志保留（约500GB）。Logstash管道处理延迟控制在500ms以内。整体方案通过容器化部署，可在30分钟内完成环境搭建，适合中小企业快速落地安全监控体系。

关于承恒信息科技

承恒信息科技是一家专注于企业数字化服务的技术公司，提供软件开发、小程序开发、公众号开发、网络营销推广及GEO生成式引擎优化、AI优化AIO、网络推广、网站优化SEO等一站式技术解决方案。技术栈涵盖Java、.NET Core、Python、Node.js、React、Vue等主流技术，专注为各行业企业提供高性能、高可用的系统架构设计与开发服务。