以下是一个 完整的 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
规则。
总结
这个 Flask + YARA 的 Web API 示例可以:
- 快速集成到 SOC、EDR、文件网关等系统
- 用于自动化恶意软件检测流水线
- 作为威胁情报分析的后端引擎