网络编程

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

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

data-ad-format="fluid" data-ad-layout-key="-7k+ex-4a-9w+4a">

✅ 功能说明

  • 提供一个 /scan 接口，支持文件上传

  • 使用预加载的 YARA 规则进行扫描

  • 返回 JSON 格式的匹配结果

  • 支持多规则、可扩展

📦 项目结构

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

  1. 安装依赖

创建 requirements.txt：

1
2
3
flask
yara-python

安装：

1
2
pip install -r requirements.txt

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

Ubuntu: sudo apt-get install yara libyara-dev

macOS: brew install yara

  1. 编写 YARA 规则

rules/hello.yar

1
2
3
4
5
6
7
8
rule ContainsHello
{
    strings:
        $hello = "Hello" ascii nocase
    condition:
        $hello
}

rules/suspicious_pe.yar

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
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
}

  1. Flask Web API 主程序 (app.py)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
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=&#91;'POST'])
def scan_file():
    if 'file' not in request.files:
        return jsonify({"error": "未提供文件字段 'file'"}), 400

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

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

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

            result = {
                "filename": filename,
                "matches": &#91;]
            }

            for match in matches:
                indicators = &#91;]
                for string in match.strings:
                    indicators.append({
                        "offset": f"0x{string&#91;0]:X}",
                        "identifier": string&#91;1],
                        "data": string&#91;2].decode('utf-8', errors='replace')
                    })
                result&#91;"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)

  1. 启动服务
1
2
python app.py

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

  1. 测试 API（使用 curl）

测试文本文件

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

✅ 预期输出（匹配 ContainsHello）：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
{
  "filename": "test.txt",
  "matches": &#91;
    {
      "rule": "ContainsHello",
      "tags": &#91;],
      "indicators": &#91;
        {
          "offset": "0x0",
          "identifier": "$hello",
          "data": "Hello"
        }
      ]
    }
  ]
}

测试 PE 文件（如 exe）

1
2
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 实现文件内容匹配技术

data-ad-format="auto" data-full-width-responsive="true">

📚 相关文章