wecom_it_smart_desk/backend/app/services/matchers/regex_matcher.py

# =============================================================================
# 企微IT智能服务台 — 正则匹配器
# =============================================================================
# 说明：使用 Python re.search 进行正则匹配，带编译缓存和 ReDoS 超时保护。
#   编译缓存：同一 pattern 只编译一次，缓存在类变量 _compile_cache 中。
#   ReDoS 保护：使用 signal.alarm 超时机制，防止恶意正则导致 CPU 打满。
# =============================================================================

import logging
import re
import signal
from typing import Optional

from app.services.matchers.base import BaseMatcher, MatchResult

logger = logging.getLogger(__name__)

# 正则匹配超时时间（秒），防止 ReDoS
_REGEX_TIMEOUT_SEC: float = 2.0

# 正则编译缓存最大条目数
_MAX_CACHE_SIZE: int = 200


class RegexMatcher(BaseMatcher):
    """正则匹配器。

    匹配逻辑：
    1. 编译 condition 为正则 Pattern（带缓存）
    2. 在消息文本中搜索匹配
    3. 命中则返回匹配详情和位置

    安全措施：
    - 正则编译缓存：同一 pattern 只编译一次
    - ReDoS 超时保护：匹配超过 2 秒自动中断，返回未命中

    Example:
        condition = "密码.*过期"
        message = "我的密码好像过期了"
        → 命中，matched_detail="密码好像过期"
    """

    # 类级正则编译缓存：pattern_str → compiled Pattern
    _compile_cache: dict[str, re.Pattern] = {}

    def _get_compiled(self, pattern_str: str) -> Optional[re.Pattern]:
        """获取编译后的正则 Pattern（带缓存）。

        Args:
            pattern_str: 正则表达式字符串

        Returns:
            Optional[re.Pattern]: 编译后的 Pattern，编译失败返回 None
        """
        # 缓存命中
        if pattern_str in self._compile_cache:
            return self._compile_cache[pattern_str]

        # 缓存清理：超过上限时清空（简单 LRU 策略）
        if len(self._compile_cache) >= _MAX_CACHE_SIZE:
            self._compile_cache.clear()

        # 编译正则
        try:
            compiled = re.compile(pattern_str, re.IGNORECASE | re.MULTILINE)
            self._compile_cache[pattern_str] = compiled
            return compiled
        except re.error as e:
            logger.warning("正则编译失败: pattern=%s, error=%s", pattern_str, e)
            return None

    @staticmethod
    def _timeout_handler(signum, frame):
        """正则匹配超时信号处理器。"""
        raise TimeoutError("Regex matching timed out (possible ReDoS)")

    async def match(
        self,
        message: str,
        condition: str,
        context: Optional[dict] = None,
    ) -> MatchResult:
        """检查消息是否匹配正则表达式。

        Args:
            message: 用户消息文本
            condition: 正则表达式字符串
            context: 上下文（本匹配器不需要）

        Returns:
            MatchResult: 命中时 matched=True, matched_detail=匹配到的文本
        """
        if not message or not condition:
            return MatchResult(matched=False)

        compiled = self._get_compiled(condition)
        if compiled is None:
            return MatchResult(matched=False)

        # 使用 signal 超时保护（仅 Unix 平台可用，Windows 降级为无超时）
        try:
            # 设置超时信号
            old_handler = signal.signal(signal.SIGALRM, self._timeout_handler)
            signal.setitimer(signal.ITIMER_REAL, _REGEX_TIMEOUT_SEC)
            try:
                m = compiled.search(message)
            finally:
                signal.setitimer(signal.ITIMER_REAL, 0)
                signal.signal(signal.SIGALRM, old_handler)
        except (TimeoutError, OSError):
            logger.warning("正则匹配超时（ReDoS 保护）: pattern=%s", condition)
            return MatchResult(matched=False)
        except Exception as e:
            logger.error("正则匹配异常: pattern=%s, error=%s", condition, e)
            return MatchResult(matched=False)

        if m:
            matched_text = m.group(0)
            return MatchResult(
                matched=True,
                matched_detail=f"正则匹配: {matched_text}",
                match_position=f"位置 {m.start()}-{m.end()}",
            )

        return MatchResult(matched=False)