請求階段：Mid 1 → Mid 2 → Agent Core
響應階段：Agent Core → Mid 2 → Mid 1

# openclaw.yaml
middleware:
  enabled: true
  # 全局中間件（對所有渠道生效）
  global:
    - name: "auth-validator"      # 身份驗證
      enabled: true
      order: 10                    # 執(zhí)行順序（數(shù)字越小越靠外）
    - name: "rate-limiter"         # 速率限制
      enabled: true
      order: 20
    - name: "request-logger"       # 請求日志
      enabled: true
      order: 30
  # 按渠道配置中間件
  channels:
    feishu:
      - name: "feishu-formatter"   # 飛書格式轉(zhuǎn)換
        enabled: true
        order: 50
      - name: "sensitive-filter"   # 敏感信息過濾
        enabled: true
        order: 60
        config:
          keywords: ["密碼", "token", "密鑰"]
          mask_char: "*"
    wecom:
      - name: "wecom-formatter"    # 企微格式轉(zhuǎn)換
        enabled: true
        order: 50
    discord:
      - name: "discord-sanitizer"  # Discord Markdown清洗
        enabled: true
        order: 50

"""
middleware_sensitive_filter.py
消息敏感詞過濾中間件
功能：
1. 檢測消息中的敏感信息（手機號、身份證、銀行卡、API Key）
2. 自動脫敏替換（如 138****1234）
3. 記錄脫敏日志供審計
4. 支持自定義規(guī)則和豁免名單
"""
import re
import json
import time
from typing import Dict, Any, Optional, List
from dataclasses import dataclass, field
# ============================================
# 1. 敏感信息檢測規(guī)則定義
# ============================================
@dataclass
class DetectionRule:
    """檢測規(guī)則定義"""
    name: str
    pattern: str
    mask_fn: callable  # 脫敏函數(shù)
    severity: str = "medium"  # low / medium / high
    enabled: bool = True
# 預置檢測規(guī)則庫
BUILTIN_RULES: List[DetectionRule] = [
    # 中國大陸手機號
    DetectionRule(
        name="phone_cn",
        pattern=r'1[3-9]\d{9}',
        mask_fn=lambda m: m.group()[:3] + "****" + m.group()[-4:],
        severity="high"
    ),
    # 身份證號碼
    DetectionRule(
        name="id_card_cn",
        pattern=r'\d{17}[\dXx]',
        mask_fn=lambda m: m.group()[:4] + "**********" + m.group()[-4:],
        severity="high"
    ),
    # 銀行卡號（16-19位）
    DetectionRule(
        name="bank_card",
        pattern=r'\d{16,19}',
        mask_fn=lambda m: m.group()[:4] + " **** **** " + m.group()[-4:],
        severity="high"
    ),
    # API Key（常見格式）
    DetectionRule(
        name="api_key",
        pattern=r'(?:api[_-]?key|apikey|token|secret)[:=]\s*["\']?([\w-]{20,})["\']?',
        mask_fn=lambda m: m.group(1)[:4] + "***..." + m.group(1)[-4:] 
            if len(m.group(1)) > 8 else "***",
        severity="high"
    ),
    # 郵箱地址
    DetectionRule(
        name="email",
        pattern=r'[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}',
        mask_fn=lambda m: m.group()[:2] + "****@" + m.group().split("@")[1],
        severity="medium"
    ),
]
# ============================================
# 2. 自定義關(guān)鍵詞匹配器
# ============================================
class KeywordMatcher:
    """基于關(guān)鍵詞列表的匹配器
    支持：
    - 精確匹配
    - 正則模式匹配  
    - 自定義脫敏字符
    """
    def __init__(self, keywords: List[str], mask_char: str = "*"):
        self.keywords = set(keywords)
        self.mask_char = mask_char
    def check(self, text: str) -> Dict[str, Any]:
        """檢查文本是否包含關(guān)鍵詞"""
        found = []
        clean_text = text
        for kw in self.keywords:
            if kw.lower() in clean_text.lower():
                found.append(kw)
                # 替換為脫敏字符
                masked = self.mask_char * len(kw)
                clean_text = clean_text.replace(kw, masked)
        return {
            "has_sensitive": len(found) > 0,
            "matched_keywords": found,
            "count": len(found),
            "masked_text": clean_text
        }
# ============================================
# 3. 核心中間件邏輯
# ============================================
class SensitiveFilterMiddleware:
    """敏感信息過濾中間件
    工作流程：
    1. 接收消息 → 正則匹配敏感模式
    2. 關(guān)鍵詞匹配
    3. 脫敏替換
    4. 記錄審計日志
    5. 返回脫敏后的消息
    """
    def __init__(self, config: Optional[Dict] = None):
        cfg = config or {}
        # 加載規(guī)則
        self.rules = BUILTIN_RULES.copy()
        # 關(guān)鍵詞過濾器
        keywords = cfg.get("keywords", [])
        mask_char = cfg.get("mask_char", "*")
        self.keyword_matcher = KeywordMatcher(keywords, mask_char)
        # 豁免名單（不對特定用戶做過濾）
        self.whitelist = set(cfg.get("whitelist", []))
        # 審計日志
        self.audit_log = []
        # 統(tǒng)計
        self.stats = {
            "total_checked": 0,
            "total_masked": 0,
            "by_rule": {}
        }
    def check(self, text: str, user_id: Optional[str] = None) -> Dict[str, Any]:
        """
        檢查并脫敏一條消息
        Args:
            text: 待檢查的文本
            user_id: 消息發(fā)送者ID（用于豁免檢查）
        Returns:
            包含脫敏文本和檢測報告的字典
        """
        # 豁免檢查
        if user_id and user_id in self.whitelist:
            return {
                "original": text,
                "masked_text": text,
                "masked": False,
                "reason": "whitelist",
                "details": []
            }
        clean_text = text
        details = []
        masked_count = 0
        # 1. 正則規(guī)則檢測
        for rule in self.rules:
            if not rule.enabled:
                continue
            matches = list(re.finditer(rule.pattern, clean_text, re.IGNORECASE))
            if matches:
                # 應用脫敏
                for match in matches:
                    mask_result = rule.mask_fn(match)
                    clean_text = clean_text.replace(match.group(), mask_result)
                details.append({
                    "rule": rule.name,
                    "severity": rule.severity,
                    "matches": len(matches),
                    "samples": [m.group()[:20] for m in matches[:3]]
                })
                masked_count += len(matches)
                # 更新統(tǒng)計
                self.stats["by_rule"][rule.name] = \
                    self.stats["by_rule"].get(rule.name, 0) + len(matches)
        # 2. 關(guān)鍵詞檢測
        kw_result = self.keyword_matcher.check(clean_text)
        if kw_result["has_sensitive"]:
            clean_text = kw_result["masked_text"]
            details.append({
                "rule": "keyword_match",
                "severity": "low",
                "matches": kw_result["count"],
                "samples": kw_result["matched_keywords"][:3]
            })
            masked_count += kw_result["count"]
        # 3. 更新統(tǒng)計
        self.stats["total_checked"] += 1
        if masked_count > 0:
            self.stats["total_masked"] += 1
        # 4. 審計日志（不記錄原文內(nèi)容，只記錄元數(shù)據(jù)）
        audit_entry = {
            "timestamp": time.time(),
            "user_id": user_id,
            "text_length": len(text),
            "masked": masked_count > 0,
            "match_count": masked_count,
            "rule_names": [d["rule"] for d in details]
        }
        self.audit_log.append(audit_entry)
        return {
            "original_length": len(text),
            "masked_text": clean_text,
            "masked": masked_count > 0,
            "total_matched": masked_count,
            "details": details,
            "audit_id": len(self.audit_log)
        }
# ============================================
# 4. 中間件鉤子函數(shù)（OpenClaw 接口）
# ============================================
middleware_instance = None
def on_load(config: Dict = None):
    """中間件加載初始化"""
    global middleware_instance
    middleware_instance = SensitiveFilterMiddleware(config)
    print(f"? 敏感信息過濾中間件已加載 "
          f"(規(guī)則: {len(middleware_instance.rules)}, "
          f"關(guān)鍵詞: {len(middleware_instance.keyword_matcher.keywords)})")
    return {"status": "loaded"}
def before_message(message: Dict) -> Dict:
    """消息發(fā)送前攔截"""
    if not middleware_instance:
        return message
    text = message.get("content", "")
    user_id = message.get("user_id")
    result = middleware_instance.check(text, user_id)
    # 替換為脫敏后的內(nèi)容
    message["content"] = result["masked_text"]
    # 附加元數(shù)據(jù)（不影響消息內(nèi)容，供下游使用）
    message["_meta"] = message.get("_meta", {})
    message["_meta"]["sensitive_filter"] = {
        "masked": result["masked"],
        "count": result["total_matched"]
    }
    return message
def on_unload():
    """卸載前清理"""
    global middleware_instance
    if middleware_instance:
        print(f"?? 敏感信息過濾統(tǒng)計: "
              f"檢查 {middleware_instance.stats['total_checked']} 條, "
              f"脫敏 {middleware_instance.stats['total_masked']} 條")
    middleware_instance = None
    print("?? 敏感信息過濾中間件已卸載")
    return {"status": "unloaded"}

# 測試敏感詞過濾
from middleware_sensitive_filter import SensitiveFilterMiddleware
middleware = SensitiveFilterMiddleware({
    "keywords": ["密碼", "密鑰", "secret"],
    "mask_char": "*"
})
test_messages = [
    "您的手機號是13812341234，請記錄",
    "我的身份證號是310101199001011234",
    "API密鑰是sk-abcdefghijklmnopqrstuvw",
    "請幫我重置一下登錄密碼",
]
for msg in test_messages:
    result = middleware.check(msg)
    status = "??" if not result["masked"] else "??"
    print(f"{status} 原文: {result['original_length']}字")
    print(f"   脫敏后: {result['masked_text']}")
    if result["details"]:
        for d in result["details"]:
            print(f"   └─ {d['rule']}: {d['matches']}處匹配")
    print()

?? 原文: 16字
   脫敏后: 您的手機號是138****1234，請記錄
   └─ phone_cn: 1處匹配

?? 原文: 19字
   脫敏后: 我的身份證號是3101**********1234
   └─ id_card_cn: 1處匹配

?? 原文: 18字
   脫敏后: API密鑰是sk-a***...tuvw
   └─ api_key: 1處匹配

?? 原文: 11字
   脫敏后: 請幫我重置一下登錄**
   └─ keyword_match: 1處匹配

"""
middleware_latency_stats.py  
請求耗時統(tǒng)計儀表盤中間件
采用環(huán)繞模式（Around）實現(xiàn)對 Agent 處理全流程的耗時統(tǒng)計，
并聚合為小時級數(shù)據(jù)供監(jiān)控面板查詢。
"""
import time
import json
from collections import defaultdict
from typing import Dict, Any, Optional
from datetime import datetime
class LatencyStatsMiddleware:
    """環(huán)繞中間件——包裹Agent處理過程并記錄耗時"""
    def __init__(self, config: Optional[Dict] = None):
        # 小時級聚合數(shù)據(jù) { "2026-06-21T12": { "feishu": {count, total, min, max} } }
        self.hourly_stats = defaultdict(lambda: defaultdict(
            lambda: {"count": 0, "total_ms": 0, "min": float("inf"), "max": 0, "p99": 0}
        ))
        self.all_durations = []  # 用于計算P99
    def get_hour_key(self) -> str:
        """獲取當前小時標識"""
        return datetime.now().strftime("%Y-%m-%dT%H")
    def update_stats(self, channel: str, duration_ms: float):
        """更新統(tǒng)計指標"""
        hour = self.get_hour_key()
        stats = self.hourly_stats[hour][channel]
        stats["count"] += 1
        stats["total_ms"] += duration_ms
        stats["min"] = min(stats["min"], duration_ms)
        stats["max"] = max(stats["max"], duration_ms)
        # 保留最近1000條用于P99計算
        self.all_durations.append(duration_ms)
        if len(self.all_durations) > 1000:
            self.all_durations.pop(0)
    def get_stats(self, hours: int = 24) -> Dict:
        """獲取最近 N 小時的統(tǒng)計數(shù)據(jù)，供監(jiān)控面板 API 查詢"""
        result = {}
        now = datetime.now()
        for h in range(hours):
            key = (now.replace(hour=now.hour - h)).strftime("%Y-%m-%dT%H")
            if key in self.hourly_stats:
                hour_data = {}
                for channel, stats in self.hourly_stats[key].items():
                    avg = stats["total_ms"] / stats["count"] if stats["count"] > 0 else 0
                    hour_data[channel] = {
                        **stats,
                        "avg_ms": round(avg, 2),
                        "min": round(stats["min"], 2) if stats["min"] != float("inf") else 0,
                        "max": round(stats["max"], 2)
                    }
                result[key] = hour_data
        return result
    def calculate_p99(self) -> float:
        """計算P99延遲"""
        if not self.all_durations:
            return 0
        sorted_durations = sorted(self.all_durations)
        idx = int(len(sorted_durations) * 0.99)
        return sorted_durations[idx]
# ============================================
# 環(huán)繞中間件——執(zhí)行包裹
# ============================================
stats_middleware = None
def on_load(config: Dict = None):
    global stats_middleware
    stats_middleware = LatencyStatsMiddleware(config)
    print("? 請求耗時統(tǒng)計中間件已加載")
    return {"status": "loaded"}
# 環(huán)繞中間件需要一個包裹函數(shù)來同時處理請求前和請求后
# 在OpenClaw中，通過yield語法實現(xiàn)環(huán)繞
def around_message(message: Dict, next_handler: callable):
    """
    環(huán)繞中間件——包裹完整的消息處理流程
    prev_handler: 返回生成器（yield前是前置，yield后是后置）
    """
    channel = message.get("channel", "unknown")
    start = time.time()
    # ====== 前置：記錄開始時間 ======
    # 可以在這里做限流判斷
    print(f"?? [{channel}] 開始處理消息...")
    # ====== 調(diào)用下一層（Agent處理） ======
    result = next_handler(message)
    # ====== 后置：計算耗時并更新統(tǒng)計 ======
    duration_ms = (time.time() - start) * 1000
    stats_middleware.update_stats(channel, duration_ms)
    print(f"?? [{channel}] 處理完成, 耗時: {duration_ms:.2f}ms")
    # 將耗時信息附加到響應元數(shù)據(jù)
    if isinstance(result, dict):
        result["_meta"] = result.get("_meta", {})
        result["_meta"]["latency_ms"] = round(duration_ms, 2)
    return result
def get_dashboard_data():
    """查詢儀表盤數(shù)據(jù)的API"""
    if not stats_middleware:
        return {"error": "中間件未初始化"}
    return {
        "p99_ms": round(stats_middleware.calculate_p99(), 2),
        "hourly": stats_middleware.get_stats(24)
    }
EXPORTS = {
    "middleware": {
        "around_message": around_message
    },
    "api": {
        "get_dashboard_data": get_dashboard_data
    }
}

# 查詢儀表盤數(shù)據(jù)
data = get_dashboard_data()
for hour_key, channels in sorted(data["hourly"].items()):
    print(f"\n? {hour_key}")
    for channel, stats in channels.items():
        print(f"  ?? {channel}: "
              f"請求數(shù)={stats['count']}, "
              f"平均={stats['avg_ms']}ms, "
              f"P99={data['p99_ms']}ms")

最外層（先攔截、最后返回）
  ├── 安全類中間件（鑒權(quán)、限流、CORS）
  ├── 日志類中間件（請求日志）
  ├── 業(yè)務類中間件（格式化、脫敏）
  ├── 監(jiān)控類中間件（耗時、錯誤率）
  └── 最內(nèi)層：Agent 核心處理