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 nocasecondition:$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 asciicondition:pe.is_pe andany 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}"] = filepathrules = yara.compile(filepaths=rule_files)print(f"[+] 成功加载 {len(rule_files)} 条 YARA 规则")return rulesexcept 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'"}), 400file = request.files['file']if file.filename == '':return jsonify({"error": "未选择文件"}), 400if 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), 200except Exception as e:os.remove(filepath)return jsonify({"error": f"扫描出错: {str(e)}"}), 500return 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 规则。

总结