权限管理与安全常见问题
权限控制相关
Q: 如何设置用户访问权限？
A: Sira AI 提供多层级的权限控制系统：
1. 角色权限设置：
json
{
"roles": {
"admin": {
"name": "系统管理员",
"permissions": [
"manage_assistants",
"manage_users",
"manage_models",
"view_analytics",
"system_settings"
]
},
"manager": {
"name": "部门经理",
"permissions": [
"manage_department_assistants",
"view_department_analytics",
"manage_department_users"
]
},
"user": {
"name": "普通用户",
"permissions": [
"use_assistants",
"view_own_conversations"
]
}
}
}2. 部门级权限控制：
- 进入「权限管理」→「部门权限」
- 选择部门并设置：
- 可访问的应用列表
- 可使用的 AI 模型
- 每日使用配额限制
- 功能权限开关
3. 用户级精细权限：
python
# 用户权限配置示例
user_permissions = {
"user_id": "zhang.san",
"department": "sales",
"role": "manager",
"custom_permissions": {
"ai_models": ["gpt-3.5-turbo", "qwen-turbo"],
"daily_quota": 1000, # tokens
"features": {
"image_analysis": True,
"file_upload": True,
"export_conversations": False
},
"time_restrictions": {
"allowed_hours": "09:00-18:00",
"weekdays_only": True
}
}
}4. 动态权限验证：
python
def check_permission(user_id: str, action: str, resource: str = None) -> bool:
"""权限检查函数"""
user = get_user_info(user_id)
# 检查基础角色权限
if not has_role_permission(user.role, action):
return False
# 检查时间限制
if not check_time_restrictions(user):
return False
# 检查配额限制
if not check_quota_limits(user, action):
return False
# 检查资源级权限
if resource and not check_resource_permission(user, resource):
return False
return TrueQ: 如何实现单点登录（SSO）集成？
A: Sira AI 支持多种 SSO 方案：
1. 企业微信SSO集成：
javascript
// 前端 SDK 配置
const weworkConfig = {
corpId: 'your_corp_id',
agentId: 'your_agent_id',
redirectUri: 'https://your-domain.com/auth/callback',
scope: 'snsapi_base',
state: 'random_state_string'
};
// 发起授权
function initiateWeworkAuth() {
const authUrl = `https://open.weixin.qq.com/connect/oauth2/authorize?` +
`appid=${weworkConfig.corpId}&` +
`redirect_uri=${encodeURIComponent(weworkConfig.redirectUri)}&` +
`response_type=code&` +
`scope=${weworkConfig.scope}&` +
`agentid=${weworkConfig.agentId}&` +
`state=${weworkConfig.state}`;
window.location.href = authUrl;
}2. LDAP/AD 集成：
python
import ldap3
class LDAPAuthenticator:
def __init__(self, server_url, base_dn, bind_user, bind_password):
self.server = ldap3.Server(server_url)
self.base_dn = base_dn
self.connection = ldap3.Connection(
self.server,
bind_user,
bind_password
)
def authenticate_user(self, username: str, password: str) -> dict:
"""LDAP用户认证"""
try:
# 搜索用户
user_dn = f"cn={username},{self.base_dn}"
# 尝试绑定用户
user_conn = ldap3.Connection(self.server, user_dn, password)
if user_conn.bind():
# 获取用户信息
user_info = self._get_user_info(username)
return {
"success": True,
"user_info": user_info
}
else:
return {"success": False, "error": "认证失败"}
except Exception as e:
return {"success": False, "error": str(e)}
def _get_user_info(self, username: str) -> dict:
"""获取用户详细信息"""
search_filter = f"(cn={username})"
self.connection.search(
self.base_dn,
search_filter,
attributes=['cn', 'mail', 'department', 'title']
)
if self.connection.entries:
entry = self.connection.entries[0]
return {
"username": str(entry.cn),
"email": str(entry.mail) if entry.mail else None,
"department": str(entry.department) if entry.department else None,
"title": str(entry.title) if entry.title else None
}
return {}3. OIDC/OAuth2 集成：
python
from authlib.integrations.flask_client import OAuth
class OIDCProvider:
def __init__(self, app, config):
self.oauth = OAuth(app)
# 注册OIDC提供商
self.client = self.oauth.register(
name='enterprise_sso',
client_id=config['CLIENT_ID'],
client_secret=config['CLIENT_SECRET'],
server_metadata_url=config['DISCOVERY_URL'],
client_kwargs={
'scope': 'openid email profile'
}
)
def get_authorization_url(self, redirect_uri: str) -> str:
"""获取授权URL"""
return self.client.authorize_redirect(redirect_uri).location
def handle_callback(self, code: str) -> dict:
"""处理回调并获取用户信息"""
try:
token = self.client.authorize_access_token()
user_info = self.client.parse_id_token(token)
return {
"success": True,
"user_info": {
"sub": user_info.get('sub'),
"email": user_info.get('email'),
"name": user_info.get('name'),
"department": user_info.get('department')
},
"access_token": token['access_token']
}
except Exception as e:
return {"success": False, "error": str(e)}4. 自动用户映射：
python
def map_sso_user_to_local(sso_user_info: dict) -> dict:
"""将SSO用户信息映射到本地用户"""
# 检查用户是否已存在
local_user = User.query.filter_by(email=sso_user_info['email']).first()
if not local_user:
# 创建新用户
local_user = User(
username=sso_user_info['email'],
email=sso_user_info['email'],
display_name=sso_user_info.get('name'),
department=sso_user_info.get('department'),
is_active=True,
auth_source='sso'
)
# 根据部门分配默认角色
local_user.role = determine_default_role(sso_user_info.get('department'))
db.session.add(local_user)
db.session.commit()
return {
"user_id": local_user.id,
"username": local_user.username,
"role": local_user.role,
"permissions": get_user_permissions(local_user.id)
}Q: 如何设置API访问权限？
A: API权限管控方案：
1. API密钥管理：
python
import secrets
import hashlib
from datetime import datetime, timedelta
class APIKeyManager:
def __init__(self):
self.keys = {} # 实际应存储在数据库中
def generate_api_key(self, user_id: str, permissions: list, expires_days: int = 365) -> dict:
"""生成API密钥"""
# 生成密钥
raw_key = secrets.token_urlsafe(32)
key_hash = hashlib.sha256(raw_key.encode()).hexdigest()
# 存储密钥信息
api_key_info = {
"key_hash": key_hash,
"user_id": user_id,
"permissions": permissions,
"created_at": datetime.now(),
"expires_at": datetime.now() + timedelta(days=expires_days),
"is_active": True,
"last_used": None,
"usage_count": 0
}
self.keys[key_hash] = api_key_info
return {
"api_key": f"tcf_{raw_key}", # 添加前缀便于识别
"key_id": key_hash[:8], # 用于显示的短ID
"expires_at": api_key_info["expires_at"],
"permissions": permissions
}
def validate_api_key(self, api_key: str) -> dict:
"""验证API密钥"""
if not api_key.startswith("tcf_"):
return {"valid": False, "error": "无效的密钥格式"}
raw_key = api_key[4:] # 移除前缀
key_hash = hashlib.sha256(raw_key.encode()).hexdigest()
key_info = self.keys.get(key_hash)
if not key_info:
return {"valid": False, "error": "密钥不存在"}
if not key_info["is_active"]:
return {"valid": False, "error": "密钥已禁用"}
if datetime.now() > key_info["expires_at"]:
return {"valid": False, "error": "密钥已过期"}
# 更新使用信息
key_info["last_used"] = datetime.now()
key_info["usage_count"] += 1
return {
"valid": True,
"user_id": key_info["user_id"],
"permissions": key_info["permissions"]
}2. API权限装饰器：
python
from functools import wraps
from flask import request, jsonify
def require_api_permission(permission: str):
"""API权限检查装饰器"""
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
# 获取API密钥
api_key = request.headers.get('Authorization', '')
if api_key.startswith('Bearer '):
api_key = api_key[7:]
# 验证密钥
key_validation = api_key_manager.validate_api_key(api_key)
if not key_validation["valid"]:
return jsonify({
"error": "API认证失败",
"message": key_validation["error"]
}), 401
# 检查权限
if permission not in key_validation["permissions"]:
return jsonify({
"error": "权限不足",
"required_permission": permission
}), 403
# 添加用户信息到请求上下文
request.current_user_id = key_validation["user_id"]
request.current_permissions = key_validation["permissions"]
return func(*args, **kwargs)
return wrapper
return decorator
# 使用示例
@app.route('/api/assistants', methods=['GET'])
@require_api_permission('read_assistants')
def list_assistants():
user_id = request.current_user_id
# 获取用户可访问的助手列表
assistants = get_user_assistants(user_id)
return jsonify(assistants)
@app.route('/api/assistants', methods=['POST'])
@require_api_permission('create_assistants')
def create_assistant():
# 创建助手逻辑
pass3. API限流控制：
python
from collections import defaultdict
import time
class APIRateLimiter:
def __init__(self):
self.requests = defaultdict(list)
self.limits = {
"basic": {"requests": 100, "window": 3600}, # 100请求/小时
"premium": {"requests": 1000, "window": 3600}, # 1000请求/小时
"enterprise": {"requests": 10000, "window": 3600} # 10000请求/小时
}
def is_allowed(self, user_id: str, tier: str = "basic") -> bool:
"""检查是否允许请求"""
current_time = time.time()
limit_info = self.limits.get(tier, self.limits["basic"])
# 清除过期记录
window_start = current_time - limit_info["window"]
self.requests[user_id] = [
req_time for req_time in self.requests[user_id]
if req_time > window_start
]
# 检查是否超过限制
if len(self.requests[user_id]) >= limit_info["requests"]:
return False
# 记录当前请求
self.requests[user_id].append(current_time)
return True
# 限流装饰器
def rate_limit(tier: str = "basic"):
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
user_id = request.current_user_id
if not rate_limiter.is_allowed(user_id, tier):
return jsonify({
"error": "请求频率过高",
"message": f"已超过 {tier} 级别的请求限制"
}), 429
return func(*args, **kwargs)
return wrapper
return decoratorQ: 如何实现数据权限隔离？
A: 数据隔离策略实现：
1. 多租户数据架构：
python
class MultiTenantQuery:
"""多租户查询基类"""
@classmethod
def get_tenant_filter(cls, user_id: str):
"""获取租户过滤条件"""
user = get_user_info(user_id)
if user.role == 'admin':
# 管理员可以看到所有数据
return {}
elif user.role == 'department_manager':
# 部门经理只能看到本部门数据
return {"department": user.department}
else:
# 普通用户只能看到自己的数据
return {"user_id": user_id}
@classmethod
def filter_by_tenant(cls, query, user_id: str):
"""应用租户过滤"""
tenant_filter = cls.get_tenant_filter(user_id)
for field, value in tenant_filter.items():
query = query.filter(getattr(cls, field) == value)
return query
# 使用示例
class Assistant(db.Model, MultiTenantQuery):
id = db.Column(db.Integer, primary_key=True)
name = db.Column(db.String(100))
user_id = db.Column(db.String(50))
department = db.Column(db.String(50))
@classmethod
def get_user_assistants(cls, user_id: str):
"""获取用户可访问的助手"""
query = cls.query
query = cls.filter_by_tenant(query, user_id)
return query.all()2. 行级安全策略：
sql
-- 数据库行级安全示例（PostgreSQL）
CREATE POLICY assistant_access_policy
ON assistants
FOR ALL
TO application_role
USING (
-- 管理员可以访问所有数据
current_setting('app.user_role') = 'admin'
OR
-- 部门经理可以访问本部门数据
(current_setting('app.user_role') = 'manager'
AND department = current_setting('app.user_department'))
OR
-- 用户只能访问自己的数据
user_id = current_setting('app.user_id')
);
-- 启用行级安全
ALTER TABLE assistants ENABLE ROW LEVEL SECURITY;3. 字段级权限控制：
python
class FieldLevelSecurity:
"""字段级安全控制"""
# 字段访问权限配置
FIELD_PERMISSIONS = {
"assistant": {
"basic_fields": ["id", "name", "description"],
"sensitive_fields": ["api_key", "config"],
"admin_fields": ["created_by", "system_config"]
},
"conversation": {
"basic_fields": ["id", "content", "timestamp"],
"private_fields": ["user_info", "ip_address"],
"analytics_fields": ["metrics", "performance_data"]
}
}
def filter_fields(self, data: dict, entity_type: str, user_role: str) -> dict:
"""根据用户角色过滤字段"""
permissions = self.FIELD_PERMISSIONS.get(entity_type, {})
allowed_fields = permissions.get("basic_fields", [])
if user_role == "admin":
# 管理员可以看到所有字段
allowed_fields.extend(permissions.get("sensitive_fields", []))
allowed_fields.extend(permissions.get("admin_fields", []))
elif user_role == "manager":
# 经理可以看到基础字段和分析字段
allowed_fields.extend(permissions.get("analytics_fields", []))
# 过滤数据
filtered_data = {
key: value for key, value in data.items()
if key in allowed_fields
}
return filtered_data4. 数据访问审计：
python
class DataAccessAuditor:
"""数据访问审计"""
def log_data_access(self, user_id: str, action: str, resource: str,
resource_id: str = None, result: str = "success"):
"""记录数据访问日志"""
audit_log = {
"timestamp": datetime.now(),
"user_id": user_id,
"action": action, # read, create, update, delete
"resource": resource, # assistant, conversation, model
"resource_id": resource_id,
"result": result, # success, denied, error
"ip_address": get_client_ip(),
"user_agent": get_user_agent()
}
# 存储到审计日志表
db.session.add(AuditLog(**audit_log))
db.session.commit()
# 异步发送到日志分析系统
self.send_to_log_system(audit_log)
def check_suspicious_activity(self, user_id: str):
"""检查可疑活动"""
# 检查短时间内大量访问
recent_accesses = AuditLog.query.filter(
AuditLog.user_id == user_id,
AuditLog.timestamp > datetime.now() - timedelta(minutes=10)
).count()
if recent_accesses > 100: # 10分钟内超过100次访问
self.trigger_security_alert(user_id, "高频访问")
# 检查异常访问模式
failed_accesses = AuditLog.query.filter(
AuditLog.user_id == user_id,
AuditLog.result == "denied",
AuditLog.timestamp > datetime.now() - timedelta(hours=1)
).count()
if failed_accesses > 10: # 1小时内超过10次被拒绝
self.trigger_security_alert(user_id, "频繁权限拒绝")
# 访问控制装饰器
def audit_data_access(resource: str):
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
user_id = getattr(request, 'current_user_id', 'anonymous')
action = request.method.lower()
try:
result = func(*args, **kwargs)
auditor.log_data_access(user_id, action, resource, result="success")
return result
except PermissionError:
auditor.log_data_access(user_id, action, resource, result="denied")
raise
except Exception as e:
auditor.log_data_access(user_id, action, resource, result="error")
raise
return wrapper
return decorator数据安全相关
Q: 如何确保对话数据的安全性？
A: 对话数据安全保护措施：
1. 数据加密存储：
python
from cryptography.fernet import Fernet
import base64
import hashlib
class ConversationEncryption:
def __init__(self, master_key: str):
# 基于主密钥生成加密密钥
key_hash = hashlib.sha256(master_key.encode()).digest()
self.cipher = Fernet(base64.urlsafe_b64encode(key_hash))
def encrypt_conversation(self, content: str) -> str:
"""加密对话内容"""
encrypted_data = self.cipher.encrypt(content.encode())
return base64.urlsafe_b64encode(encrypted_data).decode()
def decrypt_conversation(self, encrypted_content: str) -> str:
"""解密对话内容"""
try:
encrypted_data = base64.urlsafe_b64decode(encrypted_content.encode())
decrypted_data = self.cipher.decrypt(encrypted_data)
return decrypted_data.decode()
except Exception as e:
logger.error(f"解密失败: {e}")
return "[加密内容无法解密]"
def rotate_encryption_key(self, new_master_key: str):
"""密钥轮换"""
new_key_hash = hashlib.sha256(new_master_key.encode()).digest()
new_cipher = Fernet(base64.urlsafe_b64encode(new_key_hash))
# 重新加密所有对话数据
conversations = Conversation.query.all()
for conv in conversations:
# 用旧密钥解密
decrypted_content = self.decrypt_conversation(conv.content)
# 用新密钥加密
conv.content = new_cipher.encrypt(decrypted_content.encode())
db.session.commit()
self.cipher = new_cipher2. 敏感信息检测和脱敏：
python
import re
from typing import List, Dict
class SensitiveDataDetector:
def __init__(self):
# 敏感信息正则表达式
self.patterns = {
"phone": r'1[3-9]\d{9}', # 手机号
"email": r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b',
"id_card": r'\d{17}[\dXx]', # 身份证号
"bank_card": r'\d{16,19}', # 银行卡号
"ip_address": r'\b(?:\d{1,3}\.){3}\d{1,3}\b',
"credit_card": r'\b(?:\d{4}[-\s]?){3}\d{4}\b'
}
def detect_sensitive_data(self, text: str) -> List[Dict]:
"""检测敏感数据"""
detected = []
for data_type, pattern in self.patterns.items():
matches = re.finditer(pattern, text)
for match in matches:
detected.append({
"type": data_type,
"value": match.group(),
"start": match.start(),
"end": match.end(),
"confidence": self._calculate_confidence(data_type, match.group())
})
return detected
def mask_sensitive_data(self, text: str, mask_char: str = "*") -> str:
"""脱敏处理"""
masked_text = text
detected_data = self.detect_sensitive_data(text)
# 按位置倒序处理，避免位置偏移
for item in sorted(detected_data, key=lambda x: x['start'], reverse=True):
value = item['value']
if item['type'] == 'phone':
# 手机号：保留前3位和后4位
masked_value = value[:3] + mask_char * 4 + value[-4:]
elif item['type'] == 'email':
# 邮箱：保留域名，用户名部分脱敏
local, domain = value.split('@')
masked_local = local[:2] + mask_char * (len(local) - 2)
masked_value = f"{masked_local}@{domain}"
elif item['type'] == 'id_card':
# 身份证：保留前6位和后4位
masked_value = value[:6] + mask_char * 8 + value[-4:]
else:
# 其他类型：保留前后各2位
if len(value) > 4:
masked_value = value[:2] + mask_char * (len(value) - 4) + value[-2:]
else:
masked_value = mask_char * len(value)
masked_text = masked_text[:item['start']] + masked_value + masked_text[item['end']:]
return masked_text
def _calculate_confidence(self, data_type: str, value: str) -> float:
"""计算检测置信度"""
if data_type == "phone" and len(value) == 11:
return 0.95
elif data_type == "email" and "@" in value and "." in value:
return 0.90
elif data_type == "id_card" and len(value) == 18:
return 0.85
else:
return 0.703. 数据访问控制：
python
class ConversationAccessControl:
"""对话数据访问控制"""
def __init__(self, encryption_service, detector_service):
self.encryption = encryption_service
self.detector = detector_service
def get_conversation(self, conversation_id: str, user_id: str,
include_sensitive: bool = False) -> dict:
"""获取对话数据"""
# 权限检查
if not self._check_access_permission(conversation_id, user_id):
raise PermissionError("无权访问此对话")
# 获取加密对话
conversation = Conversation.query.get(conversation_id)
# 解密内容
decrypted_content = self.encryption.decrypt_conversation(conversation.content)
# 根据权限决定是否脱敏
if not include_sensitive:
decrypted_content = self.detector.mask_sensitive_data(decrypted_content)
return {
"id": conversation.id,
"content": decrypted_content,
"timestamp": conversation.timestamp,
"user_id": conversation.user_id
}
def _check_access_permission(self, conversation_id: str, user_id: str) -> bool:
"""检查访问权限"""
conversation = Conversation.query.get(conversation_id)
user = User.query.get(user_id)
# 用户只能访问自己的对话
if conversation.user_id == user_id:
return True
# 管理员可以访问所有对话
if user.role == 'admin':
return True
# 部门经理可以访问本部门的对话
if user.role == 'manager':
conversation_user = User.query.get(conversation.user_id)
return conversation_user.department == user.department
return False4. 数据保留和清理策略：
python
from datetime import datetime, timedelta
class DataRetentionManager:
"""数据保留管理"""
def __init__(self):
# 数据保留策略配置
self.retention_policies = {
"conversation": {
"basic_user": 90, # 普通用户90天
"premium_user": 365, # 高级用户1年
"admin": -1 # 管理员数据永久保留
},
"audit_log": {
"security_events": 2555, # 安全事件7年
"access_logs": 365, # 访问日志1年
"error_logs": 180 # 错误日志6个月
}
}
def cleanup_expired_data(self):
"""清理过期数据"""
# 清理过期对话
self._cleanup_expired_conversations()
# 清理过期日志
self._cleanup_expired_logs()
# 清理临时文件
self._cleanup_temporary_files()
def _cleanup_expired_conversations(self):
"""清理过期对话"""
for user_type, retention_days in self.retention_policies["conversation"].items():
if retention_days == -1: # 永久保留
continue
cutoff_date = datetime.now() - timedelta(days=retention_days)
# 查找过期对话
expired_conversations = Conversation.query.join(User).filter(
User.user_type == user_type,
Conversation.created_at < cutoff_date
).all()
for conversation in expired_conversations:
# 记录删除操作
self._log_data_deletion("conversation", conversation.id)
# 删除对话
db.session.delete(conversation)
db.session.commit()
logger.info(f"清理了 {len(expired_conversations)} 条 {user_type} 过期对话")
def _log_data_deletion(self, data_type: str, data_id: str):
"""记录数据删除日志"""
deletion_log = {
"timestamp": datetime.now(),
"data_type": data_type,
"data_id": data_id,
"reason": "retention_policy",
"operator": "system"
}
# 存储删除记录
db.session.add(DeletionLog(**deletion_log))Q: 如何防范数据泄露风险？
A: 数据防泄露综合方案：
1. DLP（数据丢失防护）系统：
python
class DataLeakagePreventionSystem:
def __init__(self):
self.risk_patterns = {
"high_risk": {
"patterns": [
r"(?i)密码.*[:：]\s*\w+", # 密码信息
r"(?i)secret.*[:：]\s*\w+", # 机密信息
r"(?i)token.*[:：]\s*[\w\-_]+", # API令牌
],
"action": "block"
},
"medium_risk": {
"patterns": [
r"\d{15,19}", # 银行卡号
r"\d{17}[\dXx]", # 身份证号
r"(?i)confidential|机密|内部使用" # 机密标识
],
"action": "alert"
},
"low_risk": {
"patterns": [
r"1[3-9]\d{9}", # 手机号
r"\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b" # 邮箱
],
"action": "log"
}
}
def scan_content(self, content: str, context: dict = None) -> dict:
"""扫描内容中的敏感信息"""
scan_result = {
"risk_level": "low",
"violations": [],
"action": "allow",
"recommendations": []
}
highest_risk = "low"
for risk_level, config in self.risk_patterns.items():
for pattern in config["patterns"]:
matches = re.finditer(pattern, content)
for match in matches:
violation = {
"risk_level": risk_level,
"pattern": pattern,
"match": match.group(),
"position": (match.start(), match.end()),
"context": content[max(0, match.start()-20):match.end()+20]
}
scan_result["violations"].append(violation)
# 更新最高风险等级
if self._get_risk_priority(risk_level) > self._get_risk_priority(highest_risk):
highest_risk = risk_level
scan_result["risk_level"] = highest_risk
scan_result["action"] = self.risk_patterns[highest_risk]["action"]
return scan_result
def _get_risk_priority(self, risk_level: str) -> int:
return {"low": 1, "medium": 2, "high": 3}.get(risk_level, 0)2. 水印和溯源技术：
python
class DigitalWatermarking:
"""数字水印系统"""
def __init__(self, secret_key: str):
self.secret_key = secret_key
def add_invisible_watermark(self, text: str, user_id: str, timestamp: str = None) -> str:
"""添加不可见水印"""
if not timestamp:
timestamp = datetime.now().isoformat()
# 创建水印信息
watermark_data = {
"user_id": user_id,
"timestamp": timestamp,
"hash": hashlib.md5(f"{text}{self.secret_key}".encode()).hexdigest()[:8]
}
# 将水印编码为不可见字符
watermark_string = self._encode_watermark(watermark_data)
# 在文本中随机位置插入水印
positions = self._calculate_insertion_positions(text, watermark_string)
watermarked_text = text
for pos, char in zip(positions, watermark_string):
watermarked_text = watermarked_text[:pos] + char + watermarked_text[pos:]
return watermarked_text
def extract_watermark(self, text: str) -> dict:
"""提取水印信息"""
try:
# 提取隐藏字符
hidden_chars = self._extract_hidden_chars(text)
# 解码水印
watermark_data = self._decode_watermark(hidden_chars)
return {
"found": True,
"user_id": watermark_data.get("user_id"),
"timestamp": watermark_data.get("timestamp"),
"hash": watermark_data.get("hash")
}
except Exception as e:
return {"found": False, "error": str(e)}
def _encode_watermark(self, data: dict) -> str:
"""编码水印为不可见字符"""
# 使用零宽字符编码
zero_width_chars = ['\u200B', '\u200C', '\u200D', '\u2060']
json_data = json.dumps(data)
encoded = ""
for char in json_data:
# 将字符转换为二进制
binary = format(ord(char), '08b')
# 用零宽字符表示二进制
for bit in binary:
encoded += zero_width_chars[int(bit)] if bit == '1' else zero_width_chars[0]
return encoded3. 异常行为检测：
python
class AnomalyDetectionSystem:
"""异常行为检测"""
def __init__(self):
self.user_baselines = {} # 用户基线行为
self.alert_thresholds = {
"bulk_download": 100, # 批量下载阈值
"off_hours_access": (22, 6), # 非工作时间访问
"rapid_requests": 50, # 快速请求阈值
"unusual_data_access": 0.8 # 异常数据访问阈值
}
def analyze_user_behavior(self, user_id: str, current_activity: dict) -> dict:
"""分析用户行为异常"""
# 获取用户历史基线
baseline = self._get_user_baseline(user_id)
anomalies = []
# 检测批量操作
if current_activity.get("download_count", 0) > self.alert_thresholds["bulk_download"]:
anomalies.append({
"type": "bulk_download",
"severity": "high",
"description": f"用户在短时间内下载了 {current_activity['download_count']} 个文件"
})
# 检测非工作时间访问
current_hour = datetime.now().hour
if (current_hour >= self.alert_thresholds["off_hours_access"][0] or
current_hour <= self.alert_thresholds["off_hours_access"][1]):
if not baseline.get("regular_off_hours_user", False):
anomalies.append({
"type": "off_hours_access",
"severity": "medium",
"description": f"用户在非工作时间 ({current_hour}:00) 访问系统"
})
# 检测数据访问模式异常
access_pattern_score = self._calculate_access_pattern_similarity(
current_activity.get("accessed_resources", []),
baseline.get("typical_resources", [])
)
if access_pattern_score < self.alert_thresholds["unusual_data_access"]:
anomalies.append({
"type": "unusual_data_access",
"severity": "medium",
"description": f"用户访问了异常的数据资源 (相似度: {access_pattern_score:.2f})"
})
return {
"user_id": user_id,
"anomaly_score": len(anomalies) / 3.0, # 归一化分数
"anomalies": anomalies,
"recommendation": self._generate_recommendations(anomalies)
}
def _generate_recommendations(self, anomalies: List[dict]) -> List[str]:
"""生成安全建议"""
recommendations = []
high_risk_count = sum(1 for a in anomalies if a["severity"] == "high")
medium_risk_count = sum(1 for a in anomalies if a["severity"] == "medium")
if high_risk_count > 0:
recommendations.append("立即暂停用户访问权限并通知安全团队")
elif medium_risk_count > 2:
recommendations.append("增强监控该用户的后续活动")
elif medium_risk_count > 0:
recommendations.append("向用户发送安全提醒")
return recommendationsQ: 如何实现合规性审计？
A: 合规性审计体系：
1. 审计日志标准化：
python
from enum import Enum
import json
class AuditEventType(Enum):
LOGIN = "login"
LOGOUT = "logout"
DATA_ACCESS = "data_access"
DATA_MODIFY = "data_modify"
DATA_EXPORT = "data_export"
PERMISSION_CHANGE = "permission_change"
CONFIG_CHANGE = "config_change"
SECURITY_EVENT = "security_event"
class ComplianceAuditor:
"""合规性审计器"""
def __init__(self, compliance_standards=None):
# 合规标准配置
self.standards = compliance_standards or {
"GDPR": {
"data_retention": 2555, # GDPR要求7年
"consent_tracking": True,
"right_to_be_forgotten": True
},
"SOX": {
"financial_data_access": "strict",
"change_approval": "required",
"audit_trail": "complete"
},
"ISO27001": {
"access_reviews": "quarterly",
"security_events": "immediate",
"risk_assessments": "annual"
}
}
def log_audit_event(self, event_type: AuditEventType, user_id: str,
details: dict, risk_level: str = "low"):
"""记录审计事件"""
audit_record = {
"timestamp": datetime.now().isoformat(),
"event_id": self._generate_event_id(),
"event_type": event_type.value,
"user_id": user_id,
"session_id": self._get_session_id(),
"ip_address": self._get_client_ip(),
"user_agent": self._get_user_agent(),
"risk_level": risk_level,
"details": details,
"compliance_flags": self._check_compliance_requirements(event_type, details)
}
# 存储审计记录
self._store_audit_record(audit_record)
# 实时合规性检查
self._check_real_time_compliance(audit_record)
return audit_record["event_id"]
def generate_compliance_report(self, standard: str, period: str = "monthly") -> dict:
"""生成合规性报告"""
if standard not in self.standards:
raise ValueError(f"不支持的合规标准: {standard}")
report = {
"standard": standard,
"period": period,
"generated_at": datetime.now().isoformat(),
"compliance_status": "compliant",
"findings": [],
"recommendations": [],
"metrics": {}
}
# 根据不同标准生成报告
if standard == "GDPR":
report.update(self._generate_gdpr_report(period))
elif standard == "SOX":
report.update(self._generate_sox_report(period))
elif standard == "ISO27001":
report.update(self._generate_iso27001_report(period))
return report
def _generate_gdpr_report(self, period: str) -> dict:
"""生成GDPR合规报告"""
gdpr_metrics = {
"data_subject_requests": self._count_data_subject_requests(period),
"consent_changes": self._count_consent_changes(period),
"data_breaches": self._count_security_incidents(period),
"data_retention_violations": self._check_retention_violations(period)
}
findings = []
recommendations = []
# 检查数据主体权利请求处理时间
overdue_requests = self._get_overdue_data_requests()
if overdue_requests:
findings.append({
"severity": "high",
"description": f"{len(overdue_requests)} 个数据主体请求超过30天处理期限",
"affected_items": overdue_requests
})
recommendations.append("建立自动化的数据主体请求处理流程")
return {
"metrics": gdpr_metrics,
"findings": findings,
"recommendations": recommendations
}
def _check_compliance_requirements(self, event_type: AuditEventType, details: dict) -> List[str]:
"""检查合规性要求"""
flags = []
# GDPR相关检查
if event_type == AuditEventType.DATA_ACCESS and "personal_data" in details:
flags.append("GDPR_PERSONAL_DATA_ACCESS")
if event_type == AuditEventType.DATA_EXPORT:
flags.append("GDPR_DATA_TRANSFER")
# SOX相关检查
if "financial" in details.get("data_category", ""):
flags.append("SOX_FINANCIAL_DATA")
# ISO27001相关检查
if event_type == AuditEventType.SECURITY_EVENT:
flags.append("ISO27001_SECURITY_INCIDENT")
return flags2. 自动化合规检查：
python
class AutomatedComplianceChecker:
"""自动化合规检查"""
def __init__(self):
self.compliance_rules = {
"access_review": {
"frequency": "quarterly",
"scope": "all_users",
"requirements": ["manager_approval", "documentation"]
},
"password_policy": {
"min_length": 8,
"complexity": True,
"expiry_days": 90,
"history_count": 5
},
"data_classification": {
"required_fields": ["classification_level", "data_owner"],
"retention_policies": True
}
}
def run_compliance_checks(self) -> dict:
"""运行合规性检查"""
check_results = {
"overall_status": "compliant",
"checks_performed": 0,
"violations_found": 0,
"details": []
}
# 访问权限审查
access_review_result = self._check_access_reviews()
check_results["details"].append(access_review_result)
# 密码策略检查
password_policy_result = self._check_password_policy()
check_results["details"].append(password_policy_result)
# 数据分类检查
data_classification_result = self._check_data_classification()
check_results["details"].append(data_classification_result)
# 汇总结果
total_violations = sum(
len(result.get("violations", []))
for result in check_results["details"]
)
check_results["checks_performed"] = len(check_results["details"])
check_results["violations_found"] = total_violations
check_results["overall_status"] = "non_compliant" if total_violations > 0 else "compliant"
return check_results
def _check_access_reviews(self) -> dict:
"""检查访问权限审查"""
result = {
"check_name": "access_review",
"status": "compliant",
"violations": [],
"recommendations": []
}
# 查找需要审查的用户
users_needing_review = User.query.filter(
User.last_access_review < datetime.now() - timedelta(days=90)
).all()
if users_needing_review:
result["status"] = "non_compliant"
result["violations"] = [
{
"user_id": user.id,
"last_review": user.last_access_review.isoformat() if user.last_access_review else None,
"days_overdue": (datetime.now() - user.last_access_review).days if user.last_access_review else "never"
}
for user in users_needing_review
]
result["recommendations"].append("执行季度访问权限审查")
return result3. 合规报告生成：
python
class ComplianceReportGenerator:
"""合规报告生成器"""
def generate_executive_summary(self, audit_data: dict) -> str:
"""生成执行摘要"""
template = """
# 合规性审计执行摘要
## 总体状况
- 审计期间：{period}
- 合规状态：{status}
- 发现问题：{total_findings}
- 高风险问题：{high_risk_findings}
## 关键发现
{key_findings}
## 改进建议
{recommendations}
## 下一步行动
{action_items}
"""
return template.format(
period=audit_data.get("period", "未指定"),
status=audit_data.get("compliance_status", "未知"),
total_findings=len(audit_data.get("findings", [])),
high_risk_findings=len([
f for f in audit_data.get("findings", [])
if f.get("severity") == "high"
]),
key_findings=self._format_findings(audit_data.get("findings", [])),
recommendations=self._format_recommendations(audit_data.get("recommendations", [])),
action_items=self._generate_action_items(audit_data)
)相关链接
安全咨询与支持：
我们的安全专家团队为您提供专业服务：
- 🔒 安全咨询：security@tcfusion.ai
- 📋 合规评估：compliance@tcfusion.ai
- 🚨 安全事件：incident@tcfusion.ai（24/7紧急响应）
- 📚 安全培训：training@tcfusion.ai