Flask+YARA-Python实现文件扫描功能

以下是一个 完整的 Web API 示例，使用 Flask + YARA-Python 实现文件扫描功能，支持上传文件并返回 YARA 规则匹配结果。

---

✅ 功能说明

• 提供一个 /scan 接口，支持文件上传
• 使用预加载的 YARA 规则进行扫描
• 返回 JSON 格式的匹配结果
• 支持多规则、可扩展

---

📦 项目结构

yara-flask-api/
├── app.py                  # Flask 主程序
├── rules/                  # YARA 规则目录
│   ├── hello.yar
│   └── suspicious_pe.yar
├── uploads/                # 临时存储上传文件（可选）
└── requirements.txt

---

1. 安装依赖

创建 requirements.txt：

flask
yara-python

安装：

pip install -r requirements.txt

确保系统已安装 YARA 开发库：

• Ubuntu: sudo apt-get install yara libyara-dev
• macOS: brew install yara

---

2. 编写 YARA 规则

rules/hello.yar

rule ContainsHello
{
    strings:
        $hello = "Hello" ascii nocase
    condition:
        $hello
}

rules/suspicious_pe.yar

import "pe"

rule SuspiciousPEScan
{
    meta:
        description = "Detects common suspicious PE imports"

    strings:
        $create_remote_thread = "CreateRemoteThread" fullword ascii
        $write_process_memory = "WriteProcessMemory" fullword ascii

    condition:
        pe.is_pe and
        any of them
}

---

3. Flask Web API 主程序 (app.py)

import os
import yara
from flask import Flask, request, jsonify
from werkzeug.utils import secure_filename

# 初始化 Flask 应用
app = Flask(__name__)
app.config['UPLOAD_FOLDER'] = 'uploads'
app.config['MAX_CONTENT_LENGTH'] = 10 * 1024 * 1024  # 10MB 限制

# 确保目录存在
os.makedirs(app.config['UPLOAD_FOLDER'], exist_ok=True)
os.makedirs('rules', exist_ok=True)

# 编译所有 .yar 规则
def load_yara_rules():
    try:
        rule_files = {}
        for filename in os.listdir('rules'):
            if filename.endswith('.yar'):
                filepath = os.path.join('rules', filename)
                rule_files[f"rule_{filename}"] = filepath
        rules = yara.compile(filepaths=rule_files)
        print(f"[+] 成功加载 {len(rule_files)} 条 YARA 规则")
        return rules
    except yara.Error as e:
        print(f"[-] YARA 规则编译失败: {e}")
        return None

# 全局加载规则
yara_rules = load_yara_rules()

if not yara_rules:
    print("[-] 无法启动：YARA 规则加载失败")
    exit(1)

# 根路径
@app.route('/')
def index():
    return '''
    <h3>YARA 扫描 API 服务</h3>
    <p>使用 POST /scan 上传文件进行扫描</p>
    '''

# 扫描接口
@app.route('/scan', methods=['POST'])
def scan_file():
    if 'file' not in request.files:
        return jsonify({"error": "未提供文件字段 'file'"}), 400

    file = request.files['file']
    if file.filename == '':
        return jsonify({"error": "未选择文件"}), 400

    if file:
        filename = secure_filename(file.filename)
        filepath = os.path.join(app.config['UPLOAD_FOLDER'], filename)
        file.save(filepath)

        try:
            # 执行 YARA 扫描
            matches = yara_rules.match(filepath)

            result = {
                "filename": filename,
                "matches": []
            }

            for match in matches:
                indicators = []
                for string in match.strings:
                    indicators.append({
                        "offset": f"0x{string[0]:X}",
                        "identifier": string[1],
                        "data": string[2].decode('utf-8', errors='replace')
                    })
                result["matches"].append({
                    "rule": match.rule,
                    "tags": match.tags,
                    "indicators": indicators
                })

            os.remove(filepath)  # 扫描后删除文件（可选）
            return jsonify(result), 200

        except Exception as e:
            os.remove(filepath)
            return jsonify({"error": f"扫描出错: {str(e)}"}), 500

    return jsonify({"error": "未知错误"}), 500


# 启动服务
if __name__ == '__main__':
    print("🚀 启动 YARA 扫描服务 http://127.0.0.1:5000")
    app.run(host='0.0.0.0', port=5000, debug=False)

---

4. 启动服务

python app.py

服务将运行在：http://127.0.0.1:5000

---

5. 测试 API（使用 curl）

测试文本文件

echo "Hello, this is a test." > test.txt
curl -X POST -F "file=@test.txt" http://127.0.0.1:5000/scan

✅ 预期输出（匹配 ContainsHello）：

{
  "filename": "test.txt",
  "matches": [
    {
      "rule": "ContainsHello",
      "tags": [],
      "indicators": [
        {
          "offset": "0x0",
          "identifier": "$hello",
          "data": "Hello"
        }
      ]
    }
  ]
}

测试 PE 文件（如 exe）

curl -X POST -F "file=@malware.exe" http://127.0.0.1:5000/scan

如果该 PE 文件调用了 CreateRemoteThread，会触发 SuspiciousPEScan 规则。

总结

这个 Flask + YARA 的 Web API 示例可以：

• 快速集成到 SOC、EDR、文件网关等系统
• 用于自动化恶意软件检测流水线
• 作为威胁情报分析的后端引擎

Flask YARA 文件扫描实现, Flask Web API 文件检测示例, YARA-Python 文件分析教程, 使用 Flask 和 YARA 开发扫描工具, Flask 集成 YARA 实现恶意文件检测, Python YARA 文件扫描代码, Web API 文件扫描 Flask 实现, YARA-Python 恶意软件检测方法, Flask 构建文件扫描服务教程, Python 实现文件内容匹配技术