Hermes 五层安全防线：从用户授权到供应链安全

┌─────────────────────────────────────────────────────┐
│            第五层：供应链安全                          │
│   技能市场审核、依赖校验、签名验证                     │
├─────────────────────────────────────────────────────┤
│            第四层：行为审计                            │
│   操作日志、异常检测、告警通知                         │
├─────────────────────────────────────────────────────┤
│            第三层：容器隔离                            │
│   Docker 沙箱、资源限制、网络策略                      │
├─────────────────────────────────────────────────────┤
│            第二层：危险命令审批                        │
│   命令分类、风险评分、人工确认                         │
├─────────────────────────────────────────────────────┤
│            第一层：用户授权                            │
│   身份认证、权限分级、操作范围限制                      │
└─────────────────────────────────────────────────────┘

# config.yaml - 认证配置
auth:
  mode: token              # token | oauth | multi
  
  # Token 认证
  tokens:
    - token: "hermes-xxxx"
      user: "alice"
      role: admin           # admin | user | readonly
    
  # 多用户支持
  users:
    alice:
      role: admin
      allowed_tools: ["*"]
      allowed_paths: ["/"]
    
    bob:
      role: user
      allowed_tools: ["shell", "file_read", "file_write"]
      allowed_paths: ["/home/bob", "/tmp"]
    
    guest:
      role: readonly
      allowed_tools: ["file_read"]
      allowed_paths: ["/public"]

class AuthManager:
    """用户授权管理"""
    
    def check_permission(self, user, action, resource):
        role = self.get_role(user)
        
        # 检查工具权限
        if action.tool not in role.allowed_tools:
            return False, "tool_not_allowed"
        
        # 检查路径权限
        if not self._check_path(resource.path, role.allowed_paths):
            return False, "path_not_allowed"
        
        # 检查危险操作
        if action.is_dangerous and not role.can_dangerous:
            return False, "dangerous_not_allowed"
        
        return True, "ok"

class CommandClassifier:
    """命令风险分类"""
    
    RISK_LEVELS = {
        "safe": 0,        # 安全：读取、查询
        "moderate": 1,    # 中等：写入、创建
        "dangerous": 2,   # 危险：删除、修改系统
        "critical": 3,    # 极危：网络、权限变更
    }
    
    # 危险命令模式
    DANGEROUS_PATTERNS = [
        (r"rm\s+(-rf?)\s+/", "critical", "递归删除根目录"),
        (r"chmod\s+777", "dangerous", "设置全权限"),
        (r"curl.*\|\s*sh", "critical", "远程执行脚本"),
        (r"docker\s+rm\s+-f", "dangerous", "强制删除容器"),
        (r"git\s+push\s+--force", "moderate", "强制推送"),
        (r"DROP\s+TABLE", "critical", "删除数据库表"),
        (r"DELETE\s+FROM\s+\w+\s*;(?!\s*WHERE)", "critical", "无条件删除"),
        (r"format\s+[A-Z]:", "critical", "格式化磁盘"),
    ]
    
    def classify(self, command):
        for pattern, level, reason in self.DANGEROUS_PATTERNS:
            if re.search(pattern, command, re.IGNORECASE):
                return level, reason
        
        # LLM 辅助分类（对未匹配的命令）
        return self._llm_classify(command)

class ApprovalSystem:
    """危险命令审批系统"""
    
    async def execute(self, command, user):
        level, reason = self.classifier.classify(command)
        
        if level == "safe":
            return await self._run(command)
        
        if level == "moderate":
            if user.role == "admin":
                return await self._run(command)
            return await self._request_approval(command, reason, user)
        
        if level in ("dangerous", "critical"):
            # 必须人工确认
            return await self._request_approval(
                command, reason, user,
                require_explicit=True,
                timeout=300  # 5 分钟超时
            )
    
    async def _request_approval(self, command, reason, user, require_explicit=False, timeout=60):
        """发送审批请求"""
        approval_id = generate_id()
        
        # 发送到用户的消息渠道
        await self.notify(user, f"""
        ⚠️ 审批请求 [{approval_id}]
        
        命令：{command}
        风险：{reason}
        
        回复 /approve {approval_id} 确认执行
        回复 /reject {approval_id} 拒绝执行
        超时 {timeout}s 自动拒绝
        """)
        
        # 等待用户确认
        result = await self._wait_for_approval(approval_id, timeout)
        
        if result == "approved":
            return await self._run(command)
        else:
            return {"status": "rejected", "reason": "user_rejected"}

用户：帮我清理 /var/log 下的旧日志
Hermes：我需要执行以下命令：
  rm -rf /var/log/*.log.202[0-4]*

⚠️ 危险命令审批
风险等级：dangerous
原因：递归删除文件

回复 /approve 确认，/reject 拒绝

用户：/approve
Hermes：已执行，删除了 237 个旧日志文件，释放 4.2GB 空间。

class SandboxExecutor:
    """Docker 沙箱执行器"""
    
    SANDBOX_IMAGE = "hermes-sandbox:latest"
    
    async def execute(self, command, workspace):
        container = await docker.containers.create(
            image=self.SANDBOX_IMAGE,
            command=command,
            
            # 文件系统隔离
            volumes={
                workspace: {"bind": "/workspace", "mode": "rw"},
            },
            
            # 网络限制
            network_mode="none",  # 默认无网络
            
            # 资源限制
            mem_limit="512m",
            cpu_quota=50000,      # 0.5 CPU
            pids_limit=100,
            
            # 权限限制
            cap_drop=["ALL"],
            cap_add=["CHOWN", "SETUID"],  # 仅必要权限
            read_only=False,
            no_new_privileges=True,
            
            # 超时
            timeout=30,
        )
        
        try:
            result = await container.start()
            return result
        finally:
            await container.remove(force=True)

# 网络访问策略
network_policy:
  default: deny           # 默认拒绝所有网络
  
  allow:
    # 仅允许特定域名
    - domain: "api.openai.com"
      port: 443
      reason: "LLM API"
    
    - domain: "api.github.com"
      port: 443
      reason: "代码操作"
    
    - domain: "pypi.org"
      port: 443
      reason: "包安装"
      require_approval: true
    
  deny:
    - domain: "*"          # 拒绝其他所有
    - ip: "10.0.0.0/8"    # 拒绝内网
    - ip: "172.16.0.0/12"
    - ip: "192.168.0.0/16"

class AuditLogger:
    """行为审计日志"""
    
    async def log_action(self, action):
        entry = {
            "timestamp": datetime.utcnow().isoformat(),
            "user": action.user,
            "session_id": action.session_id,
            "action_type": action.type,       # tool_call / file_op / network
            "tool": action.tool,
            "command": action.command,
            "args": action.args,
            "result": action.result_summary,
            "risk_level": action.risk_level,
            "approved_by": action.approved_by,
            "duration_ms": action.duration,
            "tokens_used": action.tokens,
        }
        
        # 写入 SQLite（结构化查询）
        await self.db.insert(entry)
        
        # 写入日志文件（审计追溯）
        with open("audit.log", "a") as f:
            f.write(json.dumps(entry) + "\n")
        
        # 异常检测
        await self.anomaly_check(entry)

class AnomalyDetector:
    """异常行为检测"""
    
    RULES = [
        # 规则 1：短时间内大量删除
        {
            "name": "mass_deletion",
            "condition": "COUNT(delete_ops) > 10 IN 60s",
            "action": "block_and_alert"
        },
        # 规则 2：非工作时间执行危险操作
        {
            "name": "off_hours_dangerous",
            "condition": "risk_level == 'critical' AND hour NOT IN (9-18)",
            "action": "require_extra_approval"
        },
        # 规则 3：异常网络访问
        {
            "name": "unusual_network",
            "condition": "network_access AND domain NOT IN allowlist",
            "action": "block_and_alert"
        },
        # 规则 4：Token 消耗异常
        {
            "name": "token_spike",
            "condition": "tokens_in_5min > 100000",
            "action": "rate_limit_and_alert"
        },
    ]
    
    async def check(self, action):
        for rule in self.RULES:
            if await self._matches(action, rule):
                await self._take_action(action, rule)

class SkillSecurityScanner:
    """技能安全扫描"""
    
    async def scan(self, skill_content):
        issues = []
        
        # 1. 恶意代码检测
        issues += self._scan_malware(skill_content)
        
        # 2. 敏感信息泄露检测
        issues += self._scan_secrets(skill_content)
        
        # 3. 危险命令检测
        issues += self._scan_dangerous_commands(skill_content)
        
        # 4. 依赖安全检测
        issues += await self._scan_dependencies(skill_content)
        
        # 5. 签名验证
        issues += self._verify_signature(skill_content)
        
        return issues
    
    def _scan_dangerous_commands(self, content):
        """检测技能中的危险命令"""
        dangerous = []
        for match in re.finditer(r'```(?:bash|sh|shell)\n(.*?)```', content, re.DOTALL):
            code = match.group(1)
            for pattern, level, reason in CommandClassifier.DANGEROUS_PATTERNS:
                if re.search(pattern, code):
                    dangerous.append({
                        "level": level,
                        "reason": reason,
                        "code": code[:100]
                    })
        return dangerous

async def verify_dependencies(skill):
    """验证技能依赖的安全性"""
    for dep in skill.dependencies:
        # 检查包是否存在
        if not await package_exists(dep.name):
            return False, f"Package {dep.name} not found"
        
        # 检查版本是否有已知漏洞
        vulns = await check_vulnerabilities(dep.name, dep.version)
        if vulns:
            return False, f"Vulnerable: {vulns}"
        
        # 检查下载量（低下载量可能是恶意包）
        downloads = await get_download_count(dep.name)
        if downloads < 100:
            return False, f"Suspicious: low downloads ({downloads})"
    
    return True, "ok"

# config.yaml - 安全配置
security:
  # 审批
  approval:
    auto_approve_safe: true
    moderate_timeout: 60
    dangerous_timeout: 300
    critical_timeout: 600
  
  # 沙箱
  sandbox:
    enabled: true
    memory_limit: 512m
    cpu_limit: 0.5
    network: restricted
    timeout: 30
  
  # 审计
  audit:
    log_level: info          # debug | info | warn | error
    retention_days: 90
    alert_webhook: ""        # Slack/DingTalk webhook
  
  # 供应链
  supply_chain:
    scan_on_install: true
    block_unsigned: false    # 开发阶段可放宽
    allow_local_only: false
  
  # 速率限制
  rate_limit:
    requests_per_minute: 60
    dangerous_per_hour: 10
    tokens_per_hour: 500000