摘要：在本文中，我们将为“Security Arsenal”平台构建一个企业级的、可扩展的结果存储与检索系统。我们将告别简单的SQLite文件数据库，转而集成一个强大的分布式搜索引擎——Elasticsearch。你将学习为何Elasticsearch是现代安全数据分析（如SIEM）的核心，如何为我们的扫描结果设计灵活的JSON文档模型，并使用elasticsearch-py库，将Celery后台任务产生的扫描结果实时地写入Elasticsearch索引。最后，我们将改造Flask应用，使其能够从Elasticsearch中进行快速、复杂的查询，为后续的数据分析和可视化打下坚实基础。

关键词：Python, Elasticsearch, ELK, 数据存储, 搜索引擎, 安全平台, Celery, 数据分析

---

正文

我们基于Celery的异步任务系统，已经让平台具备了高并发处理能力。但目前，任务的结果只是被临时存放在Redis中。这对于一个需要长期存储、管理和分析海量扫描数据的企业级平台来说，是远远不够的。

我们需要一个真正的“安全数据湖”——它必须具备高扩展性以存储数亿条结果，以及强大的检索能力，让分析师可以像使用Google一样，在秒级内从海量数据中找到他们需要的情报（例如，“找出过去一个月内，所有IP地址为10.x.x.x的主机上新开放的端口”）。

Elasticsearch正是为此而生的完美解决方案。

1. 为什么是Elasticsearch？

Elasticsearch（简称ES）是一个开源的、分布式的、基于JSON的搜索引擎。它不仅是一个数据库，更是一个强大的数据分析引擎。

• 为搜索而生：提供极其强大的全文搜索和复杂查询能力。

• 高扩展性：可以轻松地从单个节点扩展到数百个节点的集群，处理PB级的数据。

• 灵活的文档模型：基于JSON的无模式（Schemaless）特性，让我们在迭代扫描器功能时，可以轻松地增减结果字段，而无需修改数据库表结构。

• 强大的生态（ELK Stack）：ES是ELK（Elasticsearch, Logstash, Kibana）技术栈的核心。一旦数据进入ES，我们就可以立即使用Kibana这一世界级的可视化工具，创建各种炫酷的安全仪表盘和图表。

2. 环境准备：启动Elasticsearch

对于学习和开发，使用Docker是启动Elasticsearch最简单快捷的方式。

1. 拉取并运行Elasticsearch容器：

   Bash

   # 该命令会以后台模式启动一个单节点的Elasticsearch 8.x版本
   docker run -d --name es01 -p 9200:9200 -p 9300:9300 -e "discovery.type=single-node" -e "xpack.security.enabled=false" docker.elastic.co/elasticsearch/elasticsearch:8.11.1
   

   • xpack.security.enabled=false禁用了安全特性，使我们在开发环境中无需处理复杂的用户名密码认证。绝不能在生产环境中使用此配置！

2. 验证安装： 在浏览器或使用curl访问 http://localhost:9200。如果你能看到一个包含版本信息的JSON响应，说明ES已成功启动。

3. 安装Python库：

   Bash

   pip install elasticsearch
   

3. 代码实现：从写入到读取

a) 创建ES客户端模块 (es_client.py) 我们将所有与ES的交互逻辑，都封装在一个独立的模块中。

Python

# es_client.py
from elasticsearch import Elasticsearch
from datetime import datetime

# ES的地址和索引名
ES_HOSTS = ["http://localhost:9200"]
INDEX_NAME = "security-arsenal-findings"

try:
    es_client = Elasticsearch(hosts=ES_HOSTS)
    if not es_client.ping():
        raise ConnectionError("无法连接到Elasticsearch")
    print("[+] 成功连接到Elasticsearch。")
except Exception as e:
    print(f"[!] 连接Elasticsearch失败: {e}")
    es_client = None

def store_finding(finding_doc: dict):
    """将一个发现（finding）文档存入Elasticsearch。"""
    if not es_client: return False
    
    try:
        # 自动添加时间戳
        finding_doc['@timestamp'] = datetime.utcnow()
        es_client.index(index=INDEX_NAME, document=finding_doc)
        return True
    except Exception as e:
        print(f"[*] 写入ES时出错: {e}")
        return False

def search_findings_by_project(project_id):
    """根据项目ID搜索所有相关的发现。"""
    if not es_client: return []
    
    query = {
        "query": {
            "match": {
                "project_id": project_id
            }
        },
        "sort": [
            {"@timestamp": "desc"}
        ]
    }
    
    try:
        response = es_client.search(index=INDEX_NAME, body=query, size=100)
        return [hit['_source'] for hit in response['hits']['hits']]
    except Exception as e:
        print(f"[*] 从ES搜索时出错: {e}")
        return []

b) 改造Celery任务 (tasks.py) 现在，我们的Worker在完成扫描后，不再是简单地return结果，而是将每一条发现都格式化为一个JSON文档，并调用es_client将其存入ES。

Python

# tasks.py (部分修改)
# ... (celery_app的定义不变) ...
import es_client
from modules import port_scanner
from datetime import datetime

@celery_app.task
def run_port_scan_task(target, project_id):
    """
    执行端口扫描，并将每一条结果存入Elasticsearch。
    """
    open_ports = port_scanner.run(target, ports=list(range(1, 1025)))
    
    if open_ports is None:
        return f"扫描 {target} 失败。"
        
    for port in open_ports:
        # 为每一个发现创建一个结构化的文档
        finding = {
            "project_id": project_id,
            "target": target,
            "scan_type": "port_scan",
            "finding_type": "open_port",
            "details": {
                "port": port,
                "protocol": "tcp",
                "service": "unknown" # 可以后续增加服务识别
            },
            "severity": "Info"
        }
        es_client.store_finding(finding)
        
    return f"对 {target} 的端口扫描完成，发现 {len(open_ports)} 个开放端口。"

c) 改造Flask应用 (run.py) 我们将创建一个新的结果展示页面，它从ES中动态地拉取数据。

Python

# run.py (新增路由)
import es_client

@app.route('/project/<int:project_id>/results')
def show_results(project_id):
    """从Elasticsearch中查询并展示一个项目的所有扫描结果。"""
    # 假设我们有一个Project模型来获取项目名称
    # project = Project.query.get_or_404(project_id)
    project_name = f"Project-{project_id}" # 简化
    
    findings = es_client.search_findings_by_project(project_id)
    
    return render_template("es_results.html", project_name=project_name, findings=findings)

总结

通过将Elasticsearch集成到我们的平台中，我们完成了一次“鸟枪换炮”式的升级。我们的“Security Arsenal”现在拥有了一个专业级、高可扩展、强检索能力的数据中枢。这不仅仅是换了一个数据库那么简单，它为我们平台未来的所有高级功能——复杂的关联分析、机器学习异常检测、炫酷的可视化仪表盘（Kibana）——铺平了道路。

我们的任务可以在后台运行，结果可以被永久存储和检索。但在用户看来，从点击“开始扫描”到看到结果，中间的过程仍然是一个“黑盒子”。