RISC-V架构在Web应用防火墙中的性能瓶颈与优化策略

riscv-waf-perf-demo/
├── config/
│   └── waf_rules.yaml          # WAF规则配置文件
├── src/
│   ├── __init__.py
│   ├── config_loader.py       # 配置加载器
│   ├── rule_engine.py         # 规则引擎（核心匹配逻辑）
│   ├── packet_processor.py    # 数据包处理器
│   └── perf_simulator.py      # 性能瓶颈模拟与优化器
├── tests/
│   ├── __init__.py
│   └── test_rule_engine.py    # 单元测试
├── main.py                    # 主入口文件
├── requirements.txt           # Python依赖
└── README.md                  # 项目说明（此处仅示意，输出中不包含）

# 简化的WAF规则配置示例
rules:

  - id: 1001
    name: "SQL Injection Detection - Basic"
    severity: "HIGH"
    patterns:

      - "' OR '1'='1"
      - " UNION SELECT "
      - ";\-\-"
    match_field: ["request_uri", "request_body"]

  - id: 1002
    name: "Cross-Site Scripting - Script Tags"
    severity: "MEDIUM"
    patterns:

      - "<script>"
      - "javascript:"
    match_field: ["request_uri", "request_body", "response_body"]

  - id: 1003
    name: "Path Traversal"
    severity: "HIGH"
    patterns:

      - "\.\./"
      - "\.\.\\"
    match_field: ["request_uri"]

"""
WAF规则配置加载器。
负责从YAML文件加载规则，并转换为内部数据结构。
"""
import yaml
import logging
from typing import List, Dict, Any

logger = logging.getLogger(__name__)

class WAFRule:
    """单条WAF规则的内部表示"""
    __slots__ = ('id', 'name', 'severity', 'patterns', 'match_field') # 优化：使用__slots__减少内存开销

    def __init__(self, rule_dict: Dict[str, Any]):
        self.id = rule_dict.get('id')
        self.name = rule_dict.get('name', '')
        self.severity = rule_dict.get('severity', 'LOW')
        self.patterns = rule_dict.get('patterns', [])
        self.match_field = rule_dict.get('match_field', [])

    def __repr__(self):
        return f"<WAFRule id={self.id}, name='{self.name}'>"

class ConfigLoader:
    @staticmethod
    def load_rules(file_path: str) -> List[WAFRule]:
        """
        从YAML文件加载规则。
        模拟潜在瓶颈：文件I/O和复杂的字典解析。
        """
        try:
            with open(file_path, 'r', encoding='utf-8') as f:
                data = yaml.safe_load(f)
        except (IOError, yaml.YAMLError) as e:
            logger.error(f"Failed to load rule config {file_path}: {e}")
            return []

        rules_data = data.get('rules', [])
        # 列表推导式，相对高效
        rules = [WAFRule(rule) for rule in rules_data]
        logger.info(f"Loaded {len(rules)} rules from {file_path}")
        return rules

"""
性能瓶颈模拟与优化策略演示。
此模块中的函数专门设计用来模拟RISC-V环境下的性能问题，并展示优化版本。
"""
import time
import struct
import random
from typing import List

# 模拟一个不对齐的数据结构 (模拟RISC-V上非对齐访问的惩罚)
class UnalignedPacket:
    """故意使用不对齐的字段排列，模拟糟糕的内存布局"""
    def __init__(self):
        self.tiny: int = 0          # 1字节
        # 这里可能因编译器/解释器插入填充字节，但在CPython中对象模型不同。
        # 我们通过`struct`打包来模拟。
        self.medium: int = 0        # 4字节
        self.small: int = 0         # 2字节
        self.data: bytes = b''      # 变长

def process_packets_unaligned(packets: List[UnalignedPacket]) -> int:
    """
    模拟非对齐内存访问和分支密集型处理。
    瓶颈：频繁的成员访问、条件分支、非连续的字节处理。
    """
    total = 0
    for p in packets: # 外层循环
        # 模拟一些基于字段值的条件操作
        if p.tiny > 128:
            total += p.medium
        else:
            total -= p.small

        # 模拟对data字段的逐字节扫描（低效）
        for byte in p.data: # 内层循环，双重循环放大瓶颈
            if byte == 0xFF: # 频繁的分支预测
                total += 1
            elif byte == 0x00:
                total -= 1
    return total

def process_packets_aligned(packets: List['AlignedPacket']) -> int:
    """
    优化版本：使用对齐的数据结构，减少分支，循环展开。
    """
    total = 0
    # 使用局部变量绑定方法，减少属性查找开销 (Python微观优化)
    # 这模拟了编译器将常用数据放入寄存器的行为。
    for p in packets:
        tiny = p.tiny
        medium = p.medium
        small = p.small
        data = p.data
        data_len = len(data)

        # 优化分支：使用算术运算替代部分条件分支
        # 如果 tiny > 128, mask = 0xFFFFFFFF...; 否则 mask = 0
        # 但这在Python中可能不直接高效，这里简化演示逻辑。
        total += medium if tiny > 128 else -small

        # 对内部循环进行部分展开，并减少分支
        i = 0
        # 处理4字节的块（模拟利用更宽的寄存器/指令）
        while i + 3 < data_len:
            # 一次性读取4字节（模拟字访问）
            # 注意：`int.from_bytes` 在Python中开销大，此处仅为逻辑演示。
            chunk = data[i] << 24 | data[i+1] << 16 | data[i+2] << 8 | data[i+3]
            # 简化操作：统计0xFF字节数 (非常粗略的模拟)
            # 实际中会用SIMD或位操作技巧。
            total += (chunk & 0xFF000000) == 0xFF000000
            total += (chunk & 0x00FF0000) == 0x00FF0000
            total += (chunk & 0x0000FF00) == 0x0000FF00
            total += (chunk & 0x000000FF) == 0x000000FF
            i += 4
        # 处理剩余字节
        while i < data_len:
            total += 1 if data[i] == 0xFF else 0
            i += 1
    return total

class AlignedPacket:
    """优化后的数据结构，假设字段已对齐（在Python中更多是概念性）"""
    __slots__ = ('tiny', 'medium', 'small', 'data') # 明确使用__slots__
    def __init__(self, tiny, medium, small, data):
        self.tiny = tiny
        self.medium = medium
        self.small = small
        self.data = data

def benchmark_bottleneck():
    """运行瓶颈与优化的对比基准测试"""
    print("=== 性能瓶颈模拟与优化对比 ===")
    num_packets = 5000
    data_size = 200

    # 生成测试数据
    unaligned_list = []
    aligned_list = []
    random.seed(42)
    for _ in range(num_packets):
        tiny = random.randint(0, 255)
        medium = random.randint(0, 10000)
        small = random.randint(0, 1000)
        data = bytes(random.randint(0, 255) for _ in range(data_size))

        up = UnalignedPacket()
        up.tiny, up.medium, up.small, up.data = tiny, medium, small, data
        unaligned_list.append(up)

        ap = AlignedPacket(tiny, medium, small, data)
        aligned_list.append(ap)

    # 测试未优化版本
    start = time.perf_counter()
    result_unaligned = process_packets_unaligned(unaligned_list)
    time_unaligned = time.perf_counter() - start
    print(f"非对齐/分支密集型处理耗时: {time_unaligned:.4f} 秒, 结果: {result_unaligned}")

    # 测试优化版本
    start = time.perf_counter()
    result_aligned = process_packets_aligned(aligned_list)
    time_aligned = time.perf_counter() - start
    print(f"  对齐/循环展开处理耗时: {time_aligned:.4f} 秒, 结果: {result_aligned}")

    if time_unaligned > 0:
        speedup = time_unaligned / time_aligned
        print(f"  加速比: {speedup:.2f}x")
    print()

"""
WAF规则引擎核心。
实现两种模式匹配算法：基础的逐字节扫描和优化的Aho-Corasick多模式匹配。
"""
from typing import List, Dict, Any
import ahocorasick  # 使用pyahocorasick库，它是高效的Aho-Corasick实现
from .config_loader import WAFRule

class BaseRuleEngine:
    """基础规则引擎，使用朴素的逐规则逐模式扫描"""
    def __init__(self, rules: List[WAFRule]):
        self.rules = rules
        print(f"[BaseRuleEngine] 初始化，加载 {len(rules)} 条规则，共 {sum(len(r.patterns) for r in rules)} 个模式。")

    def scan(self, text: str, fields_to_scan: List[str]) -> List[Dict[str, Any]]:
        """
        扫描给定文本。
        瓶颈：O(n * m * p) 复杂度，n为文本长度，m为规则数，p为平均模式长度。
        大量分支（每个字符的循环，每个模式的起始比较）。
        """
        matches = []
        if not text:
            return matches

        text_lower = text.lower()  # 简单的大小写不敏感处理，增加开销

        for rule in self.rules:
            for pattern in rule.patterns:
                pattern_lower = pattern.lower()
                # 朴素的子串搜索 (模拟最坏情况，未使用KMP等)
                if pattern_lower in text_lower: # Python的`in`操作符使用高效算法，但这里我们用它代表一次扫描。
                    matches.append({
                        'rule_id': rule.id,
                        'rule_name': rule.name,
                        'severity': rule.severity,
                        'matched_pattern': pattern,
                        'matched_field': fields_to_scan # 简化为传入的字段
                    })
                    break  # 一条规则匹配一个模式即触发
        return matches

class OptimizedRuleEngine:
    """优化规则引擎，使用Aho-Corasick自动机进行多模式匹配"""
    def __init__(self, rules: List[WAFRule]):
        self.rules = rules
        self._automaton = None
        self._pattern_to_rule_map = {}  # 映射模式到所属规则列表（一个模式可能属于多条规则）
        self._build_automaton()

    def _build_automaton(self):
        """构建Aho-Corasick自动机，初始化开销较大，但查询极快。"""
        A = ahocorasick.Automaton()
        rule_pattern_counts = {}

        for rule in self.rules:
            for pattern in rule.patterns:
                pattern_lower = pattern.lower()
                # 将模式添加到自动机
                if not A.exists(pattern_lower):
                    A.add_word(pattern_lower, pattern_lower) # 值存储为模式本身
                # 更新映射
                if pattern_lower not in self._pattern_to_rule_map:
                    self._pattern_to_rule_map[pattern_lower] = []
                self._pattern_to_rule_map[pattern_lower].append(rule)

        A.make_automaton() # 构建失败指针，完成自动机构建
        self._automaton = A
        print(f"[OptimizedRuleEngine] 自动机构建完成，{len(self._pattern_to_rule_map)} 个唯一模式。")

    def scan(self, text: str, fields_to_scan: List[str]) -> List[Dict[str, Any]]:
        """
        使用Aho-Corasick自动机扫描文本。
        优势：O(n + z) 复杂度，n为文本长度，z为匹配数。对文本只遍历一次。
        非常适合RISC-V场景：减少指令数，降低分支预测压力。
        """
        matches = []
        if not text or not self._automaton:
            return matches

        text_lower = text.lower()
        matched_rules = set()  # 避免同一规则因多个模式重复报告

        # 核心：遍历文本一次，找出所有出现的模式
        for end_index, original_pattern in self._automaton.iter(text_lower):
            # original_pattern 是add_word时存储的值
            matched_pattern = original_pattern
            # 查找匹配该模式的所有规则
            rules_for_pattern = self._pattern_to_rule_map.get(matched_pattern, [])
            for rule in rules_for_pattern:
                if rule.id not in matched_rules:
                    matched_rules.add(rule.id)
                    matches.append({
                        'rule_id': rule.id,
                        'rule_name': rule.name,
                        'severity': rule.severity,
                        'matched_pattern': matched_pattern,
                        'matched_field': fields_to_scan
                    })
        return matches

"""
模拟HTTP数据包生成与处理。
"""
import random
from typing import Dict, List

class HTTPPacket:
    """模拟一个HTTP请求或响应数据包"""
    __slots__ = ('method', 'uri', 'headers', 'body', 'is_request', 'direction')

    def __init__(self, is_request: bool = True):
        self.is_request = is_request
        self.method = "GET"
        self.uri = "/"
        self.headers = {}
        self.body = ""
        self.direction = "INBOUND" if is_request else "OUTBOUND"
        self._generate_random_content()

    def _generate_random_content(self):
        """生成模拟的恶意/正常流量"""
        safe_uris = ["/index.html", "/about", "/api/data", "/static/image.jpg"]
        malicious_uris = ["/index.php?id=1' OR '1'='1", "/search?q=<script>alert(1)</script>", "/download/../../../etc/passwd"]

        self.method = random.choice(["GET", "POST", "PUT"])
        # 20%的概率生成恶意URI
        if random.random() < 0.2:
            self.uri = random.choice(malicious_uris)
        else:
            self.uri = random.choice(safe_uris)

        self.headers = {
            "User-Agent": "Mozilla/5.0 (Demo-Bot)",
            "Host": "example.com",
            "Content-Type": "application/x-www-form-urlencoded" if self.method == "POST" else "text/html"
        }

        if self.method == "POST" and random.random() < 0.3:
            # 模拟POST数据，可能包含恶意负载
            malicious_body_snippets = ["username=admin'--", "password=UNION SELECT credit_card FROM users", "comment=<img src=x onerror=alert('XSS')>"]
            safe_body_snippets = ["username=john&password=doe", "search_query=hello+world"]
            self.body = random.choice(malicious_body_snippets) if random.random() < 0.25 else random.choice(safe_body_snippets)
        else:
            self.body = ""

        # 如果是响应，模拟响应体
        if not self.is_request:
            self.body = "<html>... some content ... <script>console.log('legit')</script> ...</html>"
            if random.random() < 0.1:
                # 响应中注入恶意内容
                self.body += "<iframe src='malicious-site.com' style='display:none'></iframe>"

    def to_scan_text(self, fields: List[str]) -> str:
        """根据指定字段，将包的相关部分拼接成待扫描的文本字符串"""
        text_parts = []
        if "request_uri" in fields or "response_uri" in fields:
            text_parts.append(self.uri)
        if "request_body" in fields or "response_body" in fields:
            text_parts.append(self.body)
        # 头部信息通常也需扫描，此处简化
        if "request_headers" in fields or "response_headers" in fields:
            text_parts.append(str(self.headers))
        return " ".join(text_parts)

    def __repr__(self):
        return f"<HTTPPacket {self.direction} {self.method} {self.uri}>"

class PacketProcessor:
    """数据包处理器，负责生成流量并调用引擎扫描"""
    def __init__(self, base_engine, optimized_engine):
        self.base_engine = base_engine
        self.optimized_engine = optimized_engine

    def generate_and_process(self, num_packets: int = 50):
        """生成一批数据包并用两个引擎分别扫描，对比结果和时间"""
        print(f"\n=== 生成并处理 {num_packets} 个模拟数据包 ===")
        packets = [HTTPPacket(is_request=(i % 3 != 0)) for i in range(num_packets)] # 混合请求和响应

        all_matches_base = []
        all_matches_opt = []

        # 使用基础引擎扫描
        import time
        start = time.perf_counter()
        for p in packets:
            fields = ['request_uri', 'request_body'] if p.is_request else ['response_body']
            text_to_scan = p.to_scan_text(fields)
            matches = self.base_engine.scan(text_to_scan, fields)
            if matches:
                all_matches_base.extend(matches)
        time_base = time.perf_counter() - start

        # 使用优化引擎扫描
        start = time.perf_counter()
        for p in packets:
            fields = ['request_uri', 'request_body'] if p.is_request else ['response_body']
            text_to_scan = p.to_scan_text(fields)
            matches = self.optimized_engine.scan(text_to_scan, fields)
            if matches:
                all_matches_opt.extend(matches)
        time_opt = time.perf_counter() - start

        # 结果报告
        print(f"基础引擎扫描耗时: {time_base:.4f} 秒，检测到 {len(all_matches_base)} 个事件。")
        for match in all_matches_base[:3]: # 只显示前3个
            print(f"  - [规则 {match['rule_id']}] {match['rule_name']} -> 模式: '{match['matched_pattern']}'")
        if len(all_matches_base) > 3:
            print(f"  ... 以及 {len(all_matches_base) - 3} 个其他事件")

        print(f"优化引擎扫描耗时: {time_opt:.4f} 秒，检测到 {len(all_matches_opt)} 个事件。")
        for match in all_matches_opt[:3]:
            print(f"  - [规则 {match['rule_id']}] {match['rule_name']} -> 模式: '{match['matched_pattern']}'")
        if len(all_matches_opt) > 3:
            print(f"  ... 以及 {len(all_matches_opt) - 3} 个其他事件")

        # 简单验证结果一致性
        base_ids = {(m['rule_id'], m['matched_pattern']) for m in all_matches_base}
        opt_ids = {(m['rule_id'], m['matched_pattern']) for m in all_matches_opt}
        if base_ids == opt_ids:
            print("结果一致性检查: ✓ 两个引擎检测到相同的事件集合。")
        else:
            print(f"结果一致性检查: ! 结果有差异。基础引擎唯一事件数: {len(base_ids)}，优化引擎: {len(opt_ids)}")
            # 可选：打印差异
            # print("仅在基础引擎中:", base_ids - opt_ids)
            # print("仅在优化引擎中:", opt_ids - base_ids)

        if time_base > 0:
            speedup = time_base / time_opt
            print(f"引擎扫描加速比: {speedup:.2f}x")
        print()

"""
规则引擎单元测试。
"""
import sys
import os
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..'))

from src.config_loader import ConfigLoader, WAFRule
from src.rule_engine import BaseRuleEngine, OptimizedRuleEngine

def test_rule_loading():
    """测试规则加载"""
    # 使用内联规则数据，避免依赖外部文件
    dummy_rules = [
        WAFRule({'id': 1, 'name': 'Test1', 'patterns': ['evil', 'bad'], 'match_field': ['uri']}),
        WAFRule({'id': 2, 'name': 'Test2', 'patterns': ['<script>'], 'match_field': ['body']}),
    ]
    assert len(dummy_rules) == 2
    assert dummy_rules[0].patterns == ['evil', 'bad']
    print("✓ test_rule_loading passed")

def test_base_engine_scan():
    """测试基础引擎扫描"""
    rules = [
        WAFRule({'id': 1, 'name': 'Test', 'patterns': ['apple', 'banana'], 'match_field': ['any']}),
    ]
    engine = BaseRuleEngine(rules)
    # 匹配测试
    matches = engine.scan("I have an apple and a banana.", ['any'])
    assert len(matches) == 1  # 一条规则，匹配到即停
    assert matches[0]['rule_id'] == 1
    # 不匹配测试
    matches = engine.scan("I have grapes.", ['any'])
    assert len(matches) == 0
    print("✓ test_base_engine_scan passed")

def test_optimized_engine_scan():
    """测试优化引擎扫描"""
    rules = [
        WAFRule({'id': 1, 'name': 'Test1', 'patterns': ['apple', 'orange'], 'match_field': ['any']}),
        WAFRule({'id': 2, 'name': 'Test2', 'patterns': ['banana'], 'match_field': ['any']}),
        WAFRule({'id': 3, 'name': 'Test3', 'patterns': ['orange'], 'match_field': ['any']}), # 共享模式'orange'
    ]
    engine = OptimizedRuleEngine(rules)
    # 匹配多个规则（包括共享模式）
    matches = engine.scan("apple banana orange", ['any'])
    # 应匹配到规则1（apple或orange）、规则2（banana）、规则3（orange）
    matched_ids = {m['rule_id'] for m in matches}
    assert matched_ids == {1, 2, 3}
    # 不匹配测试
    matches = engine.scan("grapefruit", ['any'])
    assert len(matches) == 0
    print("✓ test_optimized_engine_scan passed")

def run_all_tests():
    """运行所有测试"""
    print("运行单元测试...")
    test_rule_loading()
    test_base_engine_scan()
    test_optimized_engine_scan()
    print("所有单元测试通过！\n")

if __name__ == '__main__':
    run_all_tests()

#!/usr/bin/env python3
"""
RISC-V WAF 性能瓶颈与优化策略演示 - 主程序
"""
import sys
import os
import logging

# 添加src目录到路径
sys.path.insert(0, os.path.join(os.path.dirname(__file__), 'src'))

from config_loader import ConfigLoader
from rule_engine import BaseRuleEngine, OptimizedRuleEngine
from packet_processor import PacketProcessor
from perf_simulator import benchmark_bottleneck

# 设置日志
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')

def main():
    print("=" * 70)
    print("    RISC-V 架构 WAF 性能瓶颈与优化策略演示项目")
    print("=" * 70)

    # 1. 加载规则
    config_path = os.path.join('config', 'waf_rules.yaml')
    if not os.path.exists(config_path):
        # 如果配置文件不存在，尝试创建示例（仅用于演示）
        print(f"警告: 配置文件 {config_path} 不存在。将使用内置示例规则。")
        # 这里可以动态创建，为简洁起见，我们假设文件存在。
        print("请确保 config/waf_rules.yaml 文件存在。")
        return

    rules = ConfigLoader.load_rules(config_path)
    if not rules:
        print("无法加载规则，程序退出。")
        return

    # 2. 初始化引擎
    print("\n>>> 初始化规则引擎...")
    base_engine = BaseRuleEngine(rules)
    optimized_engine = OptimizedRuleEngine(rules)

    # 3. 运行数据包处理演示（核心WAF功能）
    processor = PacketProcessor(base_engine, optimized_engine)
    processor.generate_and_process(num_packets=50) # 处理50个模拟包

    # 4. 运行性能瓶颈微观模拟
    benchmark_bottleneck()

    # 5. （可选）运行单元测试
    run_tests = input("\n是否运行单元测试? (y/N): ").strip().lower()
    if run_tests == 'y':
        from tests.test_rule_engine import run_all_tests
        run_all_tests()

    print("\n演示完成。")
    print("关键结论：")
    print("1. 在RISC-V等架构上，内存访问模式（对齐/非对齐）对性能影响显著。")
    print("2. 算法选择（如Aho-Corasick vs 朴素扫描）带来的性能提升远大于微观优化。")
    print("3. 减少分支、利用数据局部性是通用且有效的软件优化策略。")

if __name__ == '__main__':
    main()

python -m venv venv
    # Linux/macOS
    source venv/bin/activate
    # Windows
    venv\Scripts\activate

pip install -r requirements.txt

PyYAML>=5.4
pyahocorasick>=1.4

python main.py

python -m pytest tests/ -v

python tests/test_rule_engine.py