第20讲：安全与权限管理

掌握 Skill 的安全与权限管理，保护用户数据安全，防范常见安全风险。

一、安全风险分析

1.1 常见安全风险

风险类型	描述	危害
数据泄露	用户敏感信息被未授权访问	隐私泄露、财产损失
注入攻击	恶意代码通过输入执行	系统被控制、数据被破坏
越权访问	用户访问超出权限的资源	数据被篡改、删除
会话劫持	攻击者窃取用户会话	冒充用户操作
API 滥用	接口被恶意调用	资源耗尽、服务中断

1.2 安全原则

安全设计原则
├── 最小权限原则
│   └── 只授予必要的权限
├── 纵深防御原则
│   └── 多层安全防护
├── 失败安全原则
│   └── 出错时默认拒绝
├── 完全仲裁原则
│   └── 每次访问都验证权限
└── 开放设计原则
    └── 不依赖隐藏实现的安全

二、身份认证

2.1 认证方式

from abc import ABC, abstractmethod
from typing import Optional
import hashlib
import secrets
import time

class Authenticator(ABC):
    """认证器基类"""
    
    @abstractmethod
    def authenticate(self, credentials: dict) -> Optional[str]:
        """认证，返回用户ID或None"""
        pass
    
    @abstractmethod
    def verify_token(self, token: str) -> Optional[str]:
        """验证令牌，返回用户ID或None"""
        pass

class APIKeyAuthenticator(Authenticator):
    """API Key 认证"""
    
    def __init__(self, storage):
        self.storage = storage
    
    def generate_api_key(self, user_id: str) -> str:
        """生成 API Key"""
        api_key = secrets.token_urlsafe(32)
        hashed_key = hashlib.sha256(api_key.encode()).hexdigest()
        
        # 存储哈希值
        self.storage.save(f"apikey:{hashed_key}", {
            'user_id': user_id,
            'created_at': time.time(),
            'last_used': time.time()
        })
        
        return api_key
    
    def authenticate(self, credentials: dict) -> Optional[str]:
        """验证 API Key"""
        api_key = credentials.get('api_key')
        if not api_key:
            return None
        
        hashed_key = hashlib.sha256(api_key.encode()).hexdigest()
        key_data = self.storage.load(f"apikey:{hashed_key}")
        
        if key_data:
            # 更新最后使用时间
            key_data['last_used'] = time.time()
            self.storage.save(f"apikey:{hashed_key}", key_data)
            return key_data['user_id']
        
        return None
    
    def verify_token(self, token: str) -> Optional[str]:
        """API Key 即令牌"""
        return self.authenticate({'api_key': token})
    
    def revoke_api_key(self, api_key: str):
        """吊销 API Key"""
        hashed_key = hashlib.sha256(api_key.encode()).hexdigest()
        self.storage.delete(f"apikey:{hashed_key}")

class JWTAuthenticator(Authenticator):
    """JWT 认证"""
    
    def __init__(self, secret_key: str, storage):
        self.secret_key = secret_key
        self.storage = storage
    
    def generate_token(self, user_id: str, expires_in: int = 3600) -> str:
        """生成 JWT 令牌"""
        import jwt
        
        payload = {
            'user_id': user_id,
            'exp': time.time() + expires_in,
            'iat': time.time()
        }
        
        return jwt.encode(payload, self.secret_key, algorithm='HS256')
    
    def authenticate(self, credentials: dict) -> Optional[str]:
        """验证用户名密码（示例）"""
        username = credentials.get('username')
        password = credentials.get('password')
        
        # 从存储验证
        user_data = self.storage.load(f"user:{username}")
        if user_data and self._verify_password(password, user_data['password_hash']):
            return user_data['user_id']
        
        return None
    
    def verify_token(self, token: str) -> Optional[str]:
        """验证 JWT 令牌"""
        import jwt
        
        try:
            payload = jwt.decode(token, self.secret_key, algorithms=['HS256'])
            return payload.get('user_id')
        except jwt.ExpiredSignatureError:
            return None
        except jwt.InvalidTokenError:
            return None
    
    def _verify_password(self, password: str, password_hash: str) -> bool:
        """验证密码"""
        import bcrypt
        return bcrypt.checkpw(password.encode(), password_hash.encode())

2.2 多因素认证

class MFAAuthenticator:
    """多因素认证"""
    
    def __init__(self, storage):
        self.storage = storage
    
    def setup_totp(self, user_id: str) -> str:
        """设置 TOTP（基于时间的一次性密码）"""
        import pyotp
        
        # 生成密钥
        secret = pyotp.random_base32()
        
        # 存储密钥
        self.storage.save(f"mfa:{user_id}", {
            'secret': secret,
            'enabled': False  # 需要验证一次后才启用
        })
        
        # 生成二维码 URI
        totp = pyotp.TOTP(secret)
        provisioning_uri = totp.provisioning_uri(
            name=user_id,
            issuer_name="AI Skill"
        )
        
        return provisioning_uri
    
    def verify_totp(self, user_id: str, code: str) -> bool:
        """验证 TOTP 码"""
        import pyotp
        
        mfa_data = self.storage.load(f"mfa:{user_id}")
        if not mfa_data:
            return False
        
        totp = pyotp.TOTP(mfa_data['secret'])
        
        if totp.verify(code):
            # 首次验证成功后启用 MFA
            if not mfa_data['enabled']:
                mfa_data['enabled'] = True
                self.storage.save(f"mfa:{user_id}", mfa_data)
            return True
        
        return False
    
    def generate_backup_codes(self, user_id: str, count: int = 10) -> list:
        """生成备用验证码"""
        import secrets
        
        codes = [secrets.token_hex(4) for _ in range(count)]
        
        # 存储哈希后的备用码
        hashed_codes = [hashlib.sha256(c.encode()).hexdigest() for c in codes]
        self.storage.save(f"mfa:{user_id}:backup", hashed_codes)
        
        return codes

三、权限控制

3.1 RBAC 权限模型

from enum import Enum
from typing import List, Set

class Permission(Enum):
    """权限枚举"""
    SKILL_USE = "skill:use"
    SKILL_ADMIN = "skill:admin"
    DATA_READ = "data:read"
    DATA_WRITE = "data:write"
    DATA_DELETE = "data:delete"
    USER_MANAGE = "user:manage"
    CONFIG_READ = "config:read"
    CONFIG_WRITE = "config:write"

class Role:
    """角色"""
    
    def __init__(self, name: str, permissions: Set[Permission]):
        self.name = name
        self.permissions = permissions

# 预定义角色
ROLES = {
    'admin': Role('admin', {
        Permission.SKILL_USE,
        Permission.SKILL_ADMIN,
        Permission.DATA_READ,
        Permission.DATA_WRITE,
        Permission.DATA_DELETE,
        Permission.USER_MANAGE,
        Permission.CONFIG_READ,
        Permission.CONFIG_WRITE
    }),
    'user': Role('user', {
        Permission.SKILL_USE,
        Permission.DATA_READ,
        Permission.DATA_WRITE
    }),
    'guest': Role('guest', {
        Permission.SKILL_USE
    })
}

class RBACManager:
    """RBAC 权限管理器"""
    
    def __init__(self, storage):
        self.storage = storage
    
    def assign_role(self, user_id: str, role_name: str):
        """分配角色"""
        if role_name not in ROLES:
            raise ValueError(f"未知角色: {role_name}")
        
        self.storage.save(f"role:{user_id}", role_name)
    
    def get_user_role(self, user_id: str) -> Role:
        """获取用户角色"""
        role_name = self.storage.load(f"role:{user_id}") or 'guest'
        return ROLES.get(role_name, ROLES['guest'])
    
    def check_permission(self, user_id: str, permission: Permission) -> bool:
        """检查权限"""
        role = self.get_user_role(user_id)
        return permission in role.permissions
    
    def require_permission(self, permission: Permission):
        """权限装饰器"""
        def decorator(func):
            def wrapper(self, user_id: str, *args, **kwargs):
                if not self.check_permission(user_id, permission):
                    raise PermissionError(f"缺少权限: {permission.value}")
                return func(self, user_id, *args, **kwargs)
            return wrapper
        return decorator

3.2 资源级权限

class ResourcePermissionManager:
    """资源级权限管理"""
    
    def __init__(self, storage):
        self.storage = storage
    
    def grant_access(self, resource_id: str, user_id: str, 
                     permissions: List[str]):
        """授予资源访问权限"""
        key = f"acl:{resource_id}"
        acl = self.storage.load(key) or {}
        
        acl[user_id] = {
            'permissions': permissions,
            'granted_at': time.time()
        }
        
        self.storage.save(key, acl)
    
    def revoke_access(self, resource_id: str, user_id: str):
        """撤销资源访问权限"""
        key = f"acl:{resource_id}"
        acl = self.storage.load(key) or {}
        
        if user_id in acl:
            del acl[user_id]
            self.storage.save(key, acl)
    
    def check_resource_access(self, resource_id: str, user_id: str, 
                              action: str) -> bool:
        """检查资源访问权限"""
        # 1. 检查资源所有者
        owner = self.storage.load(f"owner:{resource_id}")
        if owner == user_id:
            return True
        
        # 2. 检查 ACL
        acl = self.storage.load(f"acl:{resource_id}") or {}
        user_acl = acl.get(user_id, {})
        
        return action in user_acl.get('permissions', [])
    
    def set_owner(self, resource_id: str, user_id: str):
        """设置资源所有者"""
        self.storage.save(f"owner:{resource_id}", user_id)

四、输入安全

4.1 输入验证

import re
from typing import Any, Dict

class InputValidator:
    """输入验证器"""
    
    @staticmethod
    def validate_email(email: str) -> bool:
        """验证邮箱格式"""
        pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$'
        return bool(re.match(pattern, email))
    
    @staticmethod
    def validate_phone(phone: str) -> bool:
        """验证手机号"""
        pattern = r'^1[3-9]\d{9}$'
        return bool(re.match(pattern, phone))
    
    @staticmethod
    def validate_file_path(path: str) -> bool:
        """验证文件路径（防止目录遍历）"""
        # 检查是否包含 .. 或绝对路径
        dangerous_patterns = ['..', '~', '//']
        return not any(p in path for p in dangerous_patterns)
    
    @staticmethod
    def sanitize_html(html: str) -> str:
        """清理 HTML（防止 XSS）"""
        import bleach
        
        allowed_tags = ['p', 'br', 'strong', 'em', 'u', 'h1', 'h2', 'h3']
        allowed_attrs = {}
        
        return bleach.clean(html, tags=allowed_tags, attributes=allowed_attrs)
    
    @staticmethod
    def validate_json(data: str) -> Dict:
        """验证并解析 JSON"""
        import json
        
        try:
            parsed = json.loads(data)
            if not isinstance(parsed, dict):
                raise ValueError("JSON 必须是对象")
            return parsed
        except json.JSONDecodeError as e:
            raise ValueError(f"无效的 JSON: {e}")
    
    @staticmethod
    def limit_string_length(text: str, max_length: int = 10000) -> str:
        """限制字符串长度"""
        if len(text) > max_length:
            raise ValueError(f"输入超过最大长度 {max_length}")
        return text
    
    @staticmethod
    def validate_sql_injection(text: str) -> bool:
        """检查 SQL 注入风险"""
        dangerous_keywords = [
            'SELECT', 'INSERT', 'UPDATE', 'DELETE', 'DROP',
            'UNION', 'EXEC', 'EXECUTE', 'SCRIPT'
        ]
        upper_text = text.upper()
        return not any(kw in upper_text for kw in dangerous_keywords)

4.2 命令注入防护

import subprocess
import shlex

class SafeCommandExecutor:
    """安全命令执行器"""
    
    ALLOWED_COMMANDS = {
        'ls': ['ls', '-la'],
        'cat': ['cat'],
        'grep': ['grep'],
        'wc': ['wc', '-l']
    }
    
    @classmethod
    def execute(cls, command: str, args: list = None) -> str:
        """安全执行命令"""
        if command not in cls.ALLOWED_COMMANDS:
            raise ValueError(f"不允许的命令: {command}")
        
        # 构建命令
        cmd = cls.ALLOWED_COMMANDS[command].copy()
        
        if args:
            # 验证参数
            for arg in args:
                if not cls._is_safe_arg(arg):
                    raise ValueError(f"不安全的参数: {arg}")
            cmd.extend(args)
        
        # 执行命令
        try:
            result = subprocess.run(
                cmd,
                capture_output=True,
                text=True,
                timeout=30
            )
            return result.stdout
        except subprocess.TimeoutExpired:
            raise TimeoutError("命令执行超时")
    
    @staticmethod
    def _is_safe_arg(arg: str) -> bool:
        """检查参数是否安全"""
        dangerous_chars = [';', '&', '|', '`', '$', '(', ')', '{', '}']
        return not any(c in arg for c in dangerous_chars)

五、数据安全

5.1 加密存储

from cryptography.fernet import Fernet
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC
import base64
import os

class DataEncryption:
    """数据加密"""
    
    def __init__(self, master_key: str):
        """初始化加密器"""
        kdf = PBKDF2HMAC(
            algorithm=hashes.SHA256(),
            length=32,
            salt=os.urandom(16),
            iterations=100000
        )
        key = base64.urlsafe_b64encode(kdf.derive(master_key.encode()))
        self.cipher = Fernet(key)
    
    def encrypt(self, data: str) -> str:
        """加密数据"""
        return self.cipher.encrypt(data.encode()).decode()
    
    def decrypt(self, encrypted: str) -> str:
        """解密数据"""
        return self.cipher.decrypt(encrypted.encode()).decode()
    
    def hash_sensitive(self, data: str) -> str:
        """哈希敏感数据（不可逆）"""
        import hashlib
        return hashlib.sha256(data.encode()).hexdigest()

class SecureDataStore:
    """安全数据存储"""
    
    def __init__(self, storage, encryption: DataEncryption):
        self.storage = storage
        self.encryption = encryption
    
    def save_sensitive(self, key: str, data: str):
        """安全保存敏感数据"""
        encrypted = self.encryption.encrypt(data)
        self.storage.save(f"enc:{key}", encrypted)
    
    def load_sensitive(self, key: str) -> str:
        """加载敏感数据"""
        encrypted = self.storage.load(f"enc:{key}")
        if encrypted:
            return self.encryption.decrypt(encrypted)
        return None
    
    def save_hash(self, key: str, data: str):
        """保存哈希值（用于验证）"""
        hashed = self.encryption.hash_sensitive(data)
        self.storage.save(f"hash:{key}", hashed)
    
    def verify_hash(self, key: str, data: str) -> bool:
        """验证哈希值"""
        stored_hash = self.storage.load(f"hash:{key}")
        if not stored_hash:
            return False
        
        data_hash = self.encryption.hash_sensitive(data)
        return stored_hash == data_hash

六、安全审计

class SecurityAuditor:
    """安全审计"""
    
    def __init__(self, storage):
        self.storage = storage
    
    def log_security_event(self, event_type: str, user_id: str, 
                          details: dict, risk_level: str = 'low'):
        """记录安全事件"""
        event = {
            'timestamp': time.time(),
            'type': event_type,
            'user_id': user_id,
            'details': details,
            'risk_level': risk_level
        }
        
        # 存储事件
        date = datetime.now().strftime('%Y-%m-%d')
        key = f"security:{date}"
        
        events = self.storage.load(key) or []
        events.append(event)
        self.storage.save(key, events)
        
        # 高风险事件立即告警
        if risk_level == 'high':
            self._alert_security_team(event)
    
    def _alert_security_team(self, event: dict):
        """告警安全团队"""
        # 发送告警通知
        pass
    
    def detect_anomaly(self, user_id: str, action: str) -> bool:
        """检测异常行为"""
        # 检查短时间内大量操作
        recent_actions = self._get_recent_actions(user_id, minutes=5)
        
        if len(recent_actions) > 100:  # 5分钟内超过100次操作
            self.log_security_event(
                'anomaly_detected',
                user_id,
                {'action': action, 'count': len(recent_actions)},
                'high'
            )
            return True
        
        # 检查异地登录
        # 检查非常规时间操作
        # ...
        
        return False
    
    def _get_recent_actions(self, user_id: str, minutes: int) -> list:
        """获取最近操作"""
        # 实现查询逻辑
        pass