1

2

3

4

5

6

7

8

9

model:

  providers:

    openai:

      api_key: sk-proj-abc123def456...

channels:

  feishu:

    app_secret: x8Kj2mP9qR5v...

  wecom:

    corp_secret: wL7nQ3tY6bV1...

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

# openclaw.yaml - TLS 配置段

gateway:

  port: 18789

  # TLS 配置

  tls:

    enabled: true

    cert_file: /etc/openclaw/tls/cert.pem

    key_file: /etc/openclaw/tls/key.pem

    # 最低 TLS 版本

    min_version: "1.2"

    # 支持的加密套件（白名单模式）

    cipher_suites:

      - TLS_ECDHE_RSA_WITH_AES_256_GCM_SHA384

      - TLS_ECDHE_RSA_WITH_AES_128_GCM_SHA256

    # 证书自动续期

    auto_renew:

      enabled: true

      provider: letsencrypt

      domains:

        - openclaw.your-domain.com

      email: admin@your-domain.com

      renew_before_days: 30

1

2

3

4

5

6

7

8

# 生成自签名证书（仅开发/测试用）

openssl req -x509 -nodes -days 365 -newkey rsa:4096 \

  -keyout /etc/openclaw/tls/key.pem \

  -out /etc/openclaw/tls/cert.pem \

  -subj "/CN=localhost" \

  -addext "subjectAltName=DNS:localhost,IP:127.0.0.1"

# 验证证书

openssl x509 -in /etc/openclaw/tls/cert.pem -text -noout | head -20

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

"""

secure_storage.py

OpenClaw 安全存储模块 - AES-256-GCM 加密

设计原则：

1. AES-256-GCM 提供认证加密（同时保证机密性和完整性）

2. 每次加密生成随机 IV（初始化向量），防止相同明文产生相同密文

3. 密钥从环境变量读取，绝不硬编码

4. 支持密钥版本化，方便轮转

"""

import os

import json

import base64

import hashlib

from typing import Dict, Optional, Any

from cryptography.hazmat.primitives.ciphers.aead import AESGCM

class SecureStorage:

    """

    敏感数据加密存储管理器

    使用 AES-256-GCM 认证加密算法：

    - GCM 模式同时提供加密和完整性验证

    - 256 位密钥，当前计算能力下暴力 破解不可行

    - 每次加密使用随机 Nonce（12字节）

    """

    def __init__(self, key_version: str = "v1"):

        """

        初始化加密管理器

        密钥从环境变量 OPENCLAW_ENCRYPTION_KEY 读取

        支持多版本密钥（key_version）用于密钥轮转

        """

        # 从环境变量读取主密钥

        master_key_hex = os.environ.get("OPENCLAW_ENCRYPTION_KEY")

        if not master_key_hex:

            raise ValueError(

                "? 未设置 OPENCLAW_ENCRYPTION_KEY 环境变量\n"

                "???? 生成密钥: python3 -c \"import os; print(os.urandom(32).hex())\""

            )

        # 多版本密钥派生

        self.key_version = key_version

        self._keys = self._derive_keys(bytes.fromhex(master_key_hex))

        self._aesgcm = AESGCM(self._keys[key_version])

    def _derive_keys(self, master_key: bytes) -> Dict[str, bytes]:

        """从主密钥派生多版本密钥（支持密钥轮转）"""

        keys = {}

        for version in ["v1", "v2", "v3"]:

            salt = f"openclaw-key-{version}".encode()

            derived = hashlib.pbkdf2_hmac(

                "sha256", master_key, salt, 100000, dklen=32

            )

            keys[version] = derived

        return keys

    def encrypt(self, plaintext: str) -> str:

        """

        加密字符串

        返回格式: base64(version || nonce || ciphertext)

        前缀 version 用于解密时选择正确的密钥版本

        """

        # 生成 12 字节随机 Nonce

        nonce = os.urandom(12)

        # 加密（AESGCM 自动附加认证标签）

        plaintext_bytes = plaintext.encode("utf-8")

        ciphertext = self._aesgcm.encrypt(nonce, plaintext_bytes, None)

        # 打包：版本(2字节) + Nonce(12字节) + 密文

        packed = (

            self.key_version.encode() +

            nonce +

            ciphertext

        )

        return base64.b64encode(packed).decode("ascii")

    def decrypt(self, encrypted: str) -> str:

        """

        解密密文

        自动识别密钥版本，支持密钥轮转期间的平滑过渡

        """

        try:

            packed = base64.b64decode(encrypted)

            # 解析版本（前2字节）

            version = packed[:2].decode()

            if version not in self._keys:

                raise ValueError(f"不支持的密钥版本: {version}")

            nonce = packed[2:14]

            ciphertext = packed[14:]

            aesgcm = AESGCM(self._keys[version])

            plaintext = aesgcm.decrypt(nonce, ciphertext, None)

            return plaintext.decode("utf-8")

        except Exception as e:

            raise DecryptionError(f"解密失败: {e}")

    def encrypt_dict(self, data: Dict[str, Any]) -> str:

        """加密字典（自动序列化为 JSON）"""

        return self.encrypt(json.dumps(data, ensure_ascii=False))

    def decrypt_dict(self, encrypted: str) -> Dict[str, Any]:

        """解密密文并反序列化为字典"""

        return json.loads(self.decrypt(encrypted))

class DecryptionError(Exception):

    """解密异常"""

    pass

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

# 使用示例

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

if __name__ == "__main__":

    # 1. 生成主密钥（一次性操作）

    # python3 -c "import os; print(os.urandom(32).hex())"

    store = SecureStorage()

    # 2. 加密敏感数据

    api_key = "sk-proj-this-is-a-secret-key-12345"

    encrypted = store.encrypt(api_key)

    print(f"???? 加密后: {encrypted[:50]}...")

    # 3. 解密验证

    decrypted = store.decrypt(encrypted)

    print(f"???? 解密后: {decrypted}")

    assert decrypted == api_key, "加解密不匹配！"

    print("? 加解密验证通过")

    # 4. 加密结构化数据

    config = {

        "openai_key": "sk-xxx",

        "feishu_secret": "abc123",

        "db_password": "p@ssw0rd"

    }

    encrypted_config = store.encrypt_dict(config)

    print(f"???? 加密配置: {encrypted_config[:50]}...")

    # 5. 相同明文产生不同密文（随机Nonce保证）

    encrypted2 = store.encrypt(api_key)

    print(f"???? 再次加密: {encrypted2[:50]}...")

    print(f"   与上次不同: {encrypted != encrypted2}")  # True

1

2

3

4

5

6

7

8

9

# openclaw.yaml - 使用加密占位符

model:

  providers:

    openai:

      # 使用 ENC() 标记加密值

      api_key: "ENC(A8f3kL9mX2...加密后的密文...)"

channels:

  feishu:

    app_secret: "ENC(pQ7wR2sY5b...)"

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

"""

config_decrypt.py

配置文件解密加载器

在 Gateway 启动时自动解密 ENC() 占位符

"""

import re

import os

from secure_storage import SecureStorage

ENC_PATTERN = re.compile(r'^ENC\((.*)\)$')

def load_config_with_decrypt(config_path: str) -> dict:

    """

    加载配置文件并自动解密集 ENC() 标记的值

    支持递归解密嵌套结构

    """

    import yaml

    with open(config_path, "r") as f:

        config = yaml.safe_load(f)

    store = SecureStorage()

    def decrypt_values(obj):

        """递归解密所有 ENC() 标记的值"""

        if isinstance(obj, dict):

            return {k: decrypt_values(v) for k, v in obj.items()}

        elif isinstance(obj, list):

            return [decrypt_values(v) for v in obj]

        elif isinstance(obj, str):

            match = ENC_PATTERN.match(obj)

            if match:

                return store.decrypt(match.group(1))

            return obj

        else:

            return obj

    return decrypt_values(config)

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

"""

conversation_anonymizer.py

对话记录脱敏归档模块

支持的脱敏类型：

- 手机号: 138****1234

- 身份证: 3101**********1234

- 邮箱: ab****@domain.com

- API Key: sk-****...****

- IP地址: 192.168.*.*

- 自定义关键字列表

"""

import re

import json

import hashlib

from typing import List, Dict, Optional

from datetime import datetime

class ConversationAnonymizer:

    """

    对话脱敏器——归档前自动清理敏感信息

    设计原则：

    1. 不可逆脱敏（脱敏后无法恢复原文）

    2. 保留语义（脱敏后的内容仍可用于分析和搜索）

    3. 可配置规则（不同业务场景使用不同规则）

    """

    DEFAULT_RULES = [

        # (名称, 正则, 替换函数)

        ("手机号", r'1[3-9]\d{9}',

         lambda m: m.group()[:3] + "****" + m.group()[-4:]),

        ("身份证", r'\d{17}[\dXx]',

         lambda m: m.group()[:4] + "**********" + m.group()[-4:]),

        ("邮箱", r'[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}',

         lambda m: m.group().split("@")[0][:2] + "****@" + m.group().split("@")[1]),

        ("API Key", r'(?:sk|api[_-]?key|token)[-=:]\s*["\']?([\w-]{20,})["\']?',

         lambda m: m.group(0).replace(m.group(1), m.group(1)[:3] + "***..." + m.group(1)[-4:])),

        ("IPv4", r'\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}',

         lambda m: ".".join(m.group().split(".")[:2]) + ".*.*"),

    ]

    def __init__(self, extra_keywords: Optional[List[str]] = None):

        self.rules = self.DEFAULT_RULES.copy()

        self.keywords = set(extra_keywords or [])

        self.stats = {"total_processed": 0, "total_redacted": 0}

    def anonymize(self, text: str) -> Dict:

        """对单条文本进行脱敏"""

        clean = text

        redactions = []

        # 正则规则脱敏

        for rule_name, pattern, mask_fn in self.rules:

            for match in re.finditer(pattern, clean):

                original = match.group()

                masked = mask_fn(match)

                clean = clean.replace(original, masked)

                redactions.append({

                    "type": rule_name,

                    "position": match.start(),

                    "length": len(original)

                })

        # 关键词脱敏

        for kw in self.keywords:

            if kw in clean:

                clean = clean.replace(kw, "*" * len(kw))

                redactions.append({"type": "keyword", "keyword": kw})

        self.stats["total_processed"] += 1

        if redactions:

            self.stats["total_redacted"] += 1

        return {

            "original_length": len(text),

            "anonymized_text": clean,

            "redaction_count": len(redactions),

            "redaction_types": list(set(r["type"] for r in redactions))

        }

    def anonymize_conversation(

        self, messages: List[Dict], archive_path: str

    ) -> str:

        """

        脱敏并归档完整对话

        输出 JSONL 格式，每条一行，方便后续分析

        """

        anonymized = []

        for msg in messages:

            content = msg.get("content", "")

            result = self.anonymize(content)

            anonymized.append({

                "role": msg.get("role", "unknown"),

                "content": result["anonymized_text"],

                "redacted": result["redaction_count"] > 0,

                "timestamp": msg.get("timestamp", datetime.now().isoformat())

            })

        # 写入归档文件

        with open(archive_path, "w", encoding="utf-8") as f:

            for record in anonymized:

                f.write(json.dumps(record, ensure_ascii=False) + "\n")

        return archive_path

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

# 测试脱敏效果

anonymizer = ConversationAnonymizer(

    extra_keywords=["内部机密", "工资"]

)

test_conversation = [

    {"role": "user", "content": "我的手机号是13812345678，帮我查下订单"},

    {"role": "assistant", "content": "好的，已查到您的订单。发货地址是北京市朝阳区xx路xx号。"},

    {"role": "user", "content": "我的身份证310101199001011234，帮我验证下"},

    {"role": "user", "content": "API Key 是 sk-abcdefghijklmn1234567890"},

    {"role": "assistant", "content": "您的token已激活，这是内部机密请勿外传"},

]

for msg in test_conversation:

    result = anonymizer.anonymize(msg["content"])

    status = "????" if result["redaction_count"] > 0 else "????"

    print(f"{status} [{msg['role']}] {result['anonymized_text']}")

print(f"\n???? 脱敏统计: 处理{anonymizer.stats['total_processed']}条, "

      f"脱敏{anonymizer.stats['total_redacted']}条")

1

2

3

4

5

6

7

???? [user] 我的手机号是138****5678，帮我查下订单

???? [assistant] 好的，已查到您的订单。发货地址是北京市朝阳区xx路xx号。

???? [user] 我的身份证3101**********1234，帮我验证下

???? [user] API Key 是 sk-***...7890

???? [assistant] 您的token已激活，这是****请勿外传

???? 脱敏统计: 处理5条, 脱敏4条

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

"""

key_rotation.py

密钥轮转工具——平滑切换加密密钥

"""

import os

import json

import glob

from datetime import datetime, timedelta

from secure_storage import SecureStorage

def rotate_keys(data_dir: str = "/data/openclaw/encrypted"):

    """

    密钥轮转主流程

    1. 用旧密钥解密所有数据

    2. 用新密钥重新加密

    3. 验证新密钥可解密

    4. 保留旧密钥24小时（应急回滚）

    """

    print("???? 开始密钥轮转...")

    # 使用旧密钥解密

    old_store = SecureStorage(key_version="v1")

    new_store = SecureStorage(key_version="v2")

    encrypted_files = glob.glob(f"{data_dir}/**/*.enc", recursive=True)

    print(f"???? 找到 {len(encrypted_files)} 个加密文件")

    rotated = 0

    errors = []

    for filepath in encrypted_files:

        try:

            # 读入 → 解密 → 重新加密

            with open(filepath, "r") as f:

                old_cipher = f.read()

            plaintext = old_store.decrypt(old_cipher)

            new_cipher = new_store.encrypt(plaintext)

            # 验证

            verified = new_store.decrypt(new_cipher)

            assert verified == plaintext, f"验证失败: {filepath}"

            # 写入（原子替换）

            tmp_path = filepath + ".tmp"

            with open(tmp_path, "w") as f:

                f.write(new_cipher)

            os.replace(tmp_path, filepath)

            rotated += 1

            print(f"  ? {os.path.basename(filepath)}")

        except Exception as e:

            errors.append((filepath, str(e)))

            print(f"  ? {os.path.basename(filepath)}: {e}")

    # 报告

    print(f"\n???? 轮转完成: 成功 {rotated}/{len(encrypted_files)}")

    if errors:

        print(f"?? 失败 {len(errors)} 个文件:")

        for f, e in errors:

            print(f"   - {f}: {e}")

    return {"rotated": rotated, "errors": len(errors)}