model:
  providers:
    openai:
      api_key: sk-proj-abc123def456...
channels:
  feishu:
    app_secret: x8Kj2mP9qR5v...
  wecom:
    corp_secret: wL7nQ3tY6bV1...

# 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
    # 證書自動續(xù)期
    auto_renew:
      enabled: true
      provider: letsencrypt
      domains:
        - openclaw.your-domain.com
      email: admin@your-domain.com
      renew_before_days: 30

# 生成自簽名證書（僅開發(fā)/測試用）
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"
# 驗(yàn)證證書
openssl x509 -in /etc/openclaw/tls/cert.pem -text -noout | head -20

"""
secure_storage.py
OpenClaw 安全存儲模塊 - AES-256-GCM 加密
設(shè)計(jì)原則：
1. AES-256-GCM 提供認(rèn)證加密（同時保證機(jī)密性和完整性）
2. 每次加密生成隨機(jī) IV（初始化向量），防止相同明文產(chǎn)生相同密文
3. 密鑰從環(huán)境變量讀取，絕不硬編碼
4. 支持密鑰版本化，方便輪轉(zhuǎn)
"""
import os
import json
import base64
import hashlib
from typing import Dict, Optional, Any
from cryptography.hazmat.primitives.ciphers.aead import AESGCM
class SecureStorage:
    """
    敏感數(shù)據(jù)加密存儲管理器
    使用 AES-256-GCM 認(rèn)證加密算法：
    - GCM 模式同時提供加密和完整性驗(yàn)證
    - 256 位密鑰，當(dāng)前計(jì)算能力下暴力 破解不可行
    - 每次加密使用隨機(jī) Nonce（12字節(jié)）
    """
    def __init__(self, key_version: str = "v1"):
        """
        初始化加密管理器
        密鑰從環(huán)境變量 OPENCLAW_ENCRYPTION_KEY 讀取
        支持多版本密鑰（key_version）用于密鑰輪轉(zhuǎn)
        """
        # 從環(huán)境變量讀取主密鑰
        master_key_hex = os.environ.get("OPENCLAW_ENCRYPTION_KEY")
        if not master_key_hex:
            raise ValueError(
                "? 未設(shè)置 OPENCLAW_ENCRYPTION_KEY 環(huán)境變量\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]:
        """從主密鑰派生多版本密鑰（支持密鑰輪轉(zhuǎn)）"""
        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 字節(jié)隨機(jī) Nonce
        nonce = os.urandom(12)
        # 加密（AESGCM 自動附加認(rèn)證標(biāo)簽）
        plaintext_bytes = plaintext.encode("utf-8")
        ciphertext = self._aesgcm.encrypt(nonce, plaintext_bytes, None)
        # 打包：版本(2字節(jié)) + Nonce(12字節(jié)) + 密文
        packed = (
            self.key_version.encode() +
            nonce +
            ciphertext
        )
        return base64.b64encode(packed).decode("ascii")
    def decrypt(self, encrypted: str) -> str:
        """
        解密密文
        自動識別密鑰版本，支持密鑰輪轉(zhuǎn)期間的平滑過渡
        """
        try:
            packed = base64.b64decode(encrypted)
            # 解析版本（前2字節(jié)）
            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. 加密敏感數(shù)據(jù)
    api_key = "sk-proj-this-is-a-secret-key-12345"
    encrypted = store.encrypt(api_key)
    print(f"?? 加密后: {encrypted[:50]}...")
    # 3. 解密驗(yàn)證
    decrypted = store.decrypt(encrypted)
    print(f"?? 解密后: {decrypted}")
    assert decrypted == api_key, "加解密不匹配！"
    print("? 加解密驗(yàn)證通過")
    # 4. 加密結(jié)構(gòu)化數(shù)據(jù)
    config = {
        "openai_key": "sk-xxx",
        "feishu_secret": "abc123",
        "db_password": "p@ssw0rd"
    }
    encrypted_config = store.encrypt_dict(config)
    print(f"?? 加密配置: {encrypted_config[:50]}...")
    # 5. 相同明文產(chǎn)生不同密文（隨機(jī)Nonce保證）
    encrypted2 = store.encrypt(api_key)
    print(f"?? 再次加密: {encrypted2[:50]}...")
    print(f"   與上次不同: {encrypted != encrypted2}")  # True

# openclaw.yaml - 使用加密占位符
model:
  providers:
    openai:
      # 使用 ENC() 標(biāo)記加密值
      api_key: "ENC(A8f3kL9mX2...加密后的密文...)"
channels:
  feishu:
    app_secret: "ENC(pQ7wR2sY5b...)"

"""
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() 標(biāo)記的值
    支持遞歸解密嵌套結(jié)構(gòu)
    """
    import yaml
    with open(config_path, "r") as f:
        config = yaml.safe_load(f)
    store = SecureStorage()
    def decrypt_values(obj):
        """遞歸解密所有 ENC() 標(biāo)記的值"""
        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)

"""
conversation_anonymizer.py
對話記錄脫敏歸檔模塊
支持的脫敏類型：
- 手機(jī)號: 138****1234
- 身份證: 3101**********1234
- 郵箱: ab****@domain.com
- API Key: sk-****...****
- IP地址: 192.168.*.*
- 自定義關(guān)鍵字列表
"""
import re
import json
import hashlib
from typing import List, Dict, Optional
from datetime import datetime
class ConversationAnonymizer:
    """
    對話脫敏器——?dú)w檔前自動清理敏感信息
    設(shè)計(jì)原則：
    1. 不可逆脫敏（脫敏后無法恢復(fù)原文）
    2. 保留語義（脫敏后的內(nèi)容仍可用于分析和搜索）
    3. 可配置規(guī)則（不同業(yè)務(wù)場景使用不同規(guī)則）
    """
    DEFAULT_RULES = [
        # (名稱, 正則, 替換函數(shù))
        ("手機(jī)號", 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:
        """對單條文本進(jìn)行脫敏"""
        clean = text
        redactions = []
        # 正則規(guī)則脫敏
        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)
                })
        # 關(guān)鍵詞脫敏
        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 格式，每條一行，方便后續(xù)分析
        """
        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

# 測試脫敏效果
anonymizer = ConversationAnonymizer(
    extra_keywords=["內(nèi)部機(jī)密", "工資"]
)
test_conversation = [
    {"role": "user", "content": "我的手機(jī)號是13812345678，幫我查下訂單"},
    {"role": "assistant", "content": "好的，已查到您的訂單。發(fā)貨地址是北京市朝陽區(qū)xx路xx號。"},
    {"role": "user", "content": "我的身份證310101199001011234，幫我驗(yàn)證下"},
    {"role": "user", "content": "API Key 是 sk-abcdefghijklmn1234567890"},
    {"role": "assistant", "content": "您的token已激活，這是內(nèi)部機(jī)密請勿外傳"},
]
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?? 脫敏統(tǒng)計(jì): 處理{anonymizer.stats['total_processed']}條, "
      f"脫敏{anonymizer.stats['total_redacted']}條")

?? [user] 我的手機(jī)號是138****5678，幫我查下訂單
?? [assistant] 好的，已查到您的訂單。發(fā)貨地址是北京市朝陽區(qū)xx路xx號。
?? [user] 我的身份證3101**********1234，幫我驗(yàn)證下
?? [user] API Key 是 sk-***...7890
?? [assistant] 您的token已激活，這是****請勿外傳

?? 脫敏統(tǒng)計(jì): 處理5條, 脫敏4條

"""
key_rotation.py
密鑰輪轉(zhuǎn)工具——平滑切換加密密鑰
"""
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"):
    """
    密鑰輪轉(zhuǎn)主流程
    1. 用舊密鑰解密所有數(shù)據(jù)
    2. 用新密鑰重新加密
    3. 驗(yàn)證新密鑰可解密
    4. 保留舊密鑰24小時（應(yīng)急回滾）
    """
    print("?? 開始密鑰輪轉(zhuǎn)...")
    # 使用舊密鑰解密
    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)
            # 驗(yàn)證
            verified = new_store.decrypt(new_cipher)
            assert verified == plaintext, f"驗(yàn)證失敗: {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}")
    # 報(bào)告
    print(f"\n?? 輪轉(zhuǎ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)}