1

2

请求阶段：Mid 1 → Mid 2 → Agent Core

响应阶段：Agent Core → Mid 2 → Mid 1

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

# openclaw.yaml

middleware:

  enabled: true

  # 全局中间件（对所有渠道生效）

  global:

    - name: "auth-validator"      # 身份验证

      enabled: true

      order: 10                    # 执行顺序（数字越小越靠外）

    - name: "rate-limiter"         # 速率限制

      enabled: true

      order: 20

    - name: "request-logger"       # 请求日志

      enabled: true

      order: 30

  # 按渠道配置中间件

  channels:

    feishu:

      - name: "feishu-formatter"   # 飞书格式转换

        enabled: true

        order: 50

      - name: "sensitive-filter"   # 敏感信息过滤

        enabled: true

        order: 60

        config:

          keywords: ["密码", "token", "密钥"]

          mask_char: "*"

    wecom:

      - name: "wecom-formatter"    # 企微格式转换

        enabled: true

        order: 50

    discord:

      - name: "discord-sanitizer"  # Discord Markdown清洗

        enabled: true

        order: 50

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

97

98

99

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

119

120

121

122

123

124

125

126

127

128

129

130

131

132

133

134

135

136

137

138

139

140

141

142

143

144

145

146

147

148

149

150

151

152

153

154

155

156

157

158

159

160

161

162

163

164

165

166

167

168

169

170

171

172

173

174

175

176

177

178

179

180

181

182

183

184

185

186

187

188

189

190

191

192

193

194

195

196

197

198

199

200

201

202

203

204

205

206

207

208

209

210

211

212

213

214

215

216

217

218

219

220

221

222

223

224

225

226

227

228

229

230

231

232

233

234

235

"""

middleware_sensitive_filter.py

消息敏感词过滤中间件

功能：

1. 检测消息中的敏感信息（手机号、身份证、银行卡、API Key）

2. 自动脱敏替换（如 138****1234）

3. 记录脱敏日志供审计

4. 支持自定义规则和豁免名单

"""

import re

import json

import time

from typing import Dict, Any, Optional, List

from dataclasses import dataclass, field

# ============================================

# 1. 敏感信息检测规则定义

# ============================================

@dataclass

class DetectionRule:

    """检测规则定义"""

    name: str

    pattern: str

    mask_fn: callable  # 脱敏函数

    severity: str = "medium"  # low / medium / high

    enabled: bool = True

# 预置检测规则库

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. 自定义关键词匹配器

# ============================================

class KeywordMatcher:

    """基于关键词列表的匹配器

    支持：

    - 精确匹配

    - 正则模式匹配 

    - 自定义脱敏字符

    """

    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]:

        """检查文本是否包含关键词"""

        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. 关键词匹配

    3. 脱敏替换

    4. 记录审计日志

    5. 返回脱敏后的消息

    """

    def __init__(self, config: Optional[Dict] = None):

        cfg = config or {}

        # 加载规则

        self.rules = BUILTIN_RULES.copy()

        # 关键词过滤器

        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 = []

        # 统计

        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: 消息发送者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. 正则规则检测

        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)

                # 更新统计

                self.stats["by_rule"][rule.name] = \

                    self.stats["by_rule"].get(rule.name, 0) + len(matches)

        # 2. 关键词检测

        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. 更新统计

        self.stats["total_checked"] += 1

        if masked_count > 0:

            self.stats["total_masked"] += 1

        # 4. 审计日志（不记录原文内容，只记录元数据）

        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. 中间件钩子函数（OpenClaw 接口）

# ============================================

middleware_instance = None

def on_load(config: Dict = None):

    """中间件加载初始化"""

    global middleware_instance

    middleware_instance = SensitiveFilterMiddleware(config)

    print(f"? 敏感信息过滤中间件已加载 "

          f"(规则: {len(middleware_instance.rules)}, "

          f"关键词: {len(middleware_instance.keyword_matcher.keywords)})")

    return {"status": "loaded"}

def before_message(message: Dict) -> Dict:

    """消息发送前拦截"""

    if not middleware_instance:

        return message

    text = message.get("content", "")

    user_id = message.get("user_id")

    result = middleware_instance.check(text, user_id)

    # 替换为脱敏后的内容

    message["content"] = result["masked_text"]

    # 附加元数据（不影响消息内容，供下游使用）

    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"???? 敏感信息过滤统计: "

              f"检查 {middleware_instance.stats['total_checked']} 条, "

              f"脱敏 {middleware_instance.stats['total_masked']} 条")

    middleware_instance = None

    print("???? 敏感信息过滤中间件已卸载")

    return {"status": "unloaded"}

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

# 测试敏感词过滤

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()

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

???? 原文: 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处匹配

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

97

98

99

100

101

102

103

104

105

106

107

"""

middleware_latency_stats.py 

请求耗时统计仪表盘中间件

采用环绕模式（Around）实现对 Agent 处理全流程的耗时统计，

并聚合为小时级数据供监控面板查询。

"""

import time

import json

from collections import defaultdict

from typing import Dict, Any, Optional

from datetime import datetime

class LatencyStatsMiddleware:

    """环绕中间件——包裹Agent处理过程并记录耗时"""

    def __init__(self, config: Optional[Dict] = None):

        # 小时级聚合数据 { "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):

        """更新统计指标"""

        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 小时的统计数据，供监控面板 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]

# ============================================

# 环绕中间件——执行包裹

# ============================================

stats_middleware = None

def on_load(config: Dict = None):

    global stats_middleware

    stats_middleware = LatencyStatsMiddleware(config)

    print("? 请求耗时统计中间件已加载")

    return {"status": "loaded"}

# 环绕中间件需要一个包裹函数来同时处理请求前和请求后

# 在OpenClaw中，通过yield语法实现环绕

def around_message(message: Dict, next_handler: callable):

    """

    环绕中间件——包裹完整的消息处理流程

    prev_handler: 返回生成器（yield前是前置，yield后是后置）

    """

    channel = message.get("channel", "unknown")

    start = time.time()

    # ====== 前置：记录开始时间 ======

    # 可以在这里做限流判断

    print(f"???? [{channel}] 开始处理消息...")

    # ====== 调用下一层（Agent处理） ======

    result = next_handler(message)

    # ====== 后置：计算耗时并更新统计 ======

    duration_ms = (time.time() - start) * 1000

    stats_middleware.update_stats(channel, duration_ms)

    print(f"???? [{channel}] 处理完成, 耗时: {duration_ms:.2f}ms")

    # 将耗时信息附加到响应元数据

    if isinstance(result, dict):

        result["_meta"] = result.get("_meta", {})

        result["_meta"]["latency_ms"] = round(duration_ms, 2)

    return result

def get_dashboard_data():

    """查询仪表盘数据的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

    }

}

1

2

3

4

5

6

7

8

9

# 查询仪表盘数据

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"请求数={stats['count']}, "

              f"平均={stats['avg_ms']}ms, "

              f"P99={data['p99_ms']}ms")

1

2

3

4

5

6

最外层（先拦截、最后返回）

  ├── 安全类中间件（鉴权、限流、CORS）

  ├── 日志类中间件（请求日志）

  ├── 业务类中间件（格式化、脱敏）

  ├── 监控类中间件（耗时、错误率）

  └── 最内层：Agent 核心处理