Nano-Banana Studio安全加固：基于JWT的API身份认证方案-程序员充电站

Nano-Banana Studio安全加固：基于JWT的API身份认证方案

最近Nano-Banana在服装设计、电商展示这些领域火得不行，很多企业都想把它用起来。但问题来了——当你把这么厉害的AI工具放到公司内部用，怎么保证数据安全？特别是服装设计这种行业，设计稿、客户数据都是商业机密，可不能随便让人访问。

我最近帮几家服装公司部署Nano-Banana Studio，发现很多团队都卡在安全这一关。他们要么直接用公开的API，要么就是简单设个密码，结果数据泄露的风险特别大。今天我就来聊聊，怎么用JWT给Nano-Banana Studio做个靠谱的身份认证，让企业用起来既方便又安全。

1. 为什么企业部署需要安全加固？

先说说我遇到的实际案例。有家做高端定制服装的公司，他们用Nano-Banana做客户形象分析和服装搭配推荐。刚开始就是简单部署，谁都能访问，结果差点出大事——有个实习生不小心把测试环境的链接发到了外部群，虽然及时发现没造成实际损失，但把老板吓得不轻。

服装行业的数据有多敏感？我简单列几个：

设计图纸和款式：这是公司的核心资产，泄露了竞争对手马上就能仿
客户身材数据和偏好：涉及隐私，处理不好可能违法
供应链和成本信息：商业机密，关系到公司利润
营销策略和客户画像：市场竞争力所在

这些数据要是随便谁都能访问，那风险可就太大了。所以企业部署Nano-Banana Studio，安全加固不是“最好有”，而是“必须有”。

2. JWT认证是什么？为什么选它？

可能有些朋友对JWT不太熟，我用人话解释一下。JWT就像你去游乐场买的门票——你买票的时候验明身份（登录），拿到一张印着你信息的票（Token），之后进每个项目都不用再查身份证，刷票就行。

传统的Session认证是啥样呢？就像游乐场给你个手环，所有信息都存在游乐场的电脑里。JWT的好处很明显：

无状态：服务器不用存会话信息，减轻负担
跨域方便：适合微服务架构，各个服务都能验证
适合API：RESTful API天然适配
信息自包含：Token里就带着用户信息，不用老查数据库

对于Nano-Banana Studio这种AI服务来说，JWT特别合适。因为AI生成通常需要多次调用，每次都要验证身份，JWT验证速度快，不会成为性能瓶颈。

3. 实战部署：四步搭建安全认证系统

说了这么多理论，咱们来点实际的。下面我手把手带你搭建一套完整的JWT认证系统，围绕Nano-Banana Studio展开。

3.1 环境准备与基础架构

首先得把环境搭好。我建议用Docker部署，这样环境隔离，也好管理。

# docker-compose.yml version: '3.8' services: # Nano-Banana Studio服务 nano-banana: image: nano-banana-studio:latest container_name: nano-banana-studio ports: - "8000:8000" environment: - AUTH_ENABLED=true - JWT_SECRET=${JWT_SECRET} depends_on: - auth-service networks: - ai-network # 认证服务 auth-service: build: ./auth-service container_name: auth-service ports: - "3000:3000" environment: - JWT_SECRET=${JWT_SECRET} - DB_HOST=postgres - DB_PASSWORD=${DB_PASSWORD} depends_on: - postgres networks: - ai-network # 数据库 postgres: image: postgres:15 container_name: postgres-db environment: - POSTGRES_PASSWORD=${DB_PASSWORD} - POSTGRES_DB=auth_db volumes: - postgres_data:/var/lib/postgresql/data networks: - ai-network # 反向代理（可选，增强安全） nginx: image: nginx:alpine container_name: nginx-proxy ports: - "443:443" - "80:80" volumes: - ./nginx.conf:/etc/nginx/nginx.conf - ./ssl:/etc/nginx/ssl depends_on: - nano-banana - auth-service networks: - ai-network networks: ai-network: driver: bridge volumes: postgres_data:

这个架构有几个关键点：

服务分离：认证服务和AI服务分开，安全边界清晰
网络隔离：内部通信走内部网络，外部只能访问指定端口
数据库独立：用户数据单独存放，不和业务数据混在一起

3.2 JWT认证服务核心实现

认证服务是核心，咱们看看关键代码怎么写。我用Node.js + Express实现，你也可以用其他语言。

// auth-service/src/server.js const express = require('express'); const jwt = require('jsonwebtoken'); const bcrypt = require('bcrypt'); const { body, validationResult } = require('express-validator'); const rateLimit = require('express-rate-limit'); const app = express(); app.use(express.json()); // 环境变量 const JWT_SECRET = process.env.JWT_SECRET || 'your-secret-key-change-in-production'; const JWT_EXPIRES_IN = '24h'; // Token 24小时过期 // 限流：防止暴力破解 const loginLimiter = rateLimit({ windowMs: 15 * 60 * 1000, // 15分钟 max: 5, // 每个IP最多5次登录尝试 message: '登录尝试过多，请15分钟后再试' }); // 用户模型（简化版） class User { constructor(id, username, passwordHash, role, department) { this.id = id; this.username = username; this.passwordHash = passwordHash; this.role = role; // 'admin', 'designer', 'viewer' this.department = department; // 部门信息，用于数据隔离 } static async findByUsername(username) { // 这里应该查数据库，为了示例简化 const users = { 'admin': new User(1, 'admin', await bcrypt.hash('admin123', 10), 'admin', 'all'), 'designer1': new User(2, 'designer1', await bcrypt.hash('design123', 10), 'designer', 'design'), 'viewer1': new User(3, 'viewer1', await bcrypt.hash('view123', 10), 'viewer', 'marketing') }; return users[username] || null; } } // 1. 登录接口 app.post('/api/auth/login', loginLimiter, [ body('username').notEmpty().withMessage('用户名不能为空'), body('password').notEmpty().withMessage('密码不能为空') ], async (req, res) => { // 验证输入 const errors = validationResult(req); if (!errors.isEmpty()) { return res.status(400).json({ errors: errors.array() }); } const { username, password } = req.body; try { // 查找用户 const user = await User.findByUsername(username); if (!user) { return res.status(401).json({ error: '认证失败', message: '用户名或密码错误' }); } // 验证密码（实际应该用数据库存储的hash） const isValid = await bcrypt.compare(password, user.passwordHash); if (!isValid) { return res.status(401).json({ error: '认证失败', message: '用户名或密码错误' }); } // 生成JWT Token const tokenPayload = { userId: user.id, username: user.username, role: user.role, department: user.department, permissions: getPermissionsByRole(user.role) }; const token = jwt.sign(tokenPayload, JWT_SECRET, { expiresIn: JWT_EXPIRES_IN }); // 返回Token和用户信息 res.json({ success: true, token, user: { id: user.id, username: user.username, role: user.role, department: user.department }, expiresIn: JWT_EXPIRES_IN }); } catch (error) { console.error('登录错误:', error); res.status(500).json({ error: '服务器错误', message: '登录过程出现异常' }); } }); // 2. Token验证中间件 const authenticateToken = (req, res, next) => { const authHeader = req.headers['authorization']; const token = authHeader && authHeader.split(' ')[1]; // Bearer TOKEN if (!token) { return res.status(401).json({ error: '未授权', message: '需要提供认证Token' }); } jwt.verify(token, JWT_SECRET, (err, user) => { if (err) { if (err.name === 'TokenExpiredError') { return res.status(401).json({ error: 'Token过期', message: '请重新登录' }); } return res.status(403).json({ error: 'Token无效', message: '认证信息不正确' }); } req.user = user; next(); }); }; // 3. 权限检查中间件 const requireRole = (...roles) => { return (req, res, next) => { if (!req.user) { return res.status(401).json({ error: '未认证' }); } if (!roles.includes(req.user.role)) { return res.status(403).json({ error: '权限不足', message: `需要角色: ${roles.join(', ')}` }); } next(); }; }; // 4. Token刷新接口 app.post('/api/auth/refresh', authenticateToken, (req, res) => { const user = req.user; // 移除过期时间，生成新payload const { exp, iat, ...payload } = user; const newToken = jwt.sign(payload, JWT_SECRET, { expiresIn: JWT_EXPIRES_IN }); res.json({ success: true, token: newToken, expiresIn: JWT_EXPIRES_IN }); }); // 5. 用户信息接口 app.get('/api/auth/me', authenticateToken, (req, res) => { res.json({ success: true, user: req.user }); }); // 权限配置函数 function getPermissionsByRole(role) { const permissions = { admin: ['read', 'write', 'delete', 'manage_users', 'view_logs'], designer: ['read', 'write', 'view_own_designs'], viewer: ['read'] }; return permissions[role] || []; } // 启动服务 const PORT = process.env.PORT || 3000; app.listen(PORT, () => { console.log(`认证服务运行在端口 ${PORT}`); });

3.3 Nano-Banana Studio集成JWT验证

认证服务做好了，接下来要让Nano-Banana Studio能验证Token。这里我用Python的FastAPI做示例。

# nano-banana-studio/app/main.py from fastapi import FastAPI, Depends, HTTPException, status from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials from pydantic import BaseModel import jwt from typing import Optional, List import os app = FastAPI(title="Nano-Banana Studio API", version="1.0.0") security = HTTPBearer() # JWT配置 JWT_SECRET = os.getenv("JWT_SECRET", "your-secret-key-change-in-production") ALGORITHM = "HS256" # 用户权限模型 class UserPermission(BaseModel): user_id: str username: str role: str department: str permissions: List[str] # JWT验证依赖 async def verify_token(credentials: HTTPAuthorizationCredentials = Depends(security)): token = credentials.credentials try: # 验证Token payload = jwt.decode( token, JWT_SECRET, algorithms=[ALGORITHM] ) # 检查必要字段 required_fields = ["userId", "username", "role", "department", "permissions"] for field in required_fields: if field not in payload: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail=f"Token缺少必要字段: {field}" ) return UserPermission( user_id=payload["userId"], username=payload["username"], role=payload["role"], department=payload["department"], permissions=payload["permissions"] ) except jwt.ExpiredSignatureError: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Token已过期，请重新登录" ) except jwt.InvalidTokenError: raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="无效的Token" ) # 权限检查装饰器 def check_permission(required_permission: str): def permission_dependency(user: UserPermission = Depends(verify_token)): if required_permission not in user.permissions: raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail=f"需要权限: {required_permission}" ) return user return permission_dependency # 部门数据隔离检查 def check_department_access(target_department: str, user: UserPermission): """检查用户是否有权限访问指定部门的数据""" if user.role == "admin": return True # 管理员可以访问所有部门 if user.department == target_department: return True # 同部门可以访问 # 可以在这里添加部门间协作的逻辑 allowed_cross_departments = { "design": ["marketing", "production"], "marketing": ["design"], "production": ["design", "quality"] } if user.department in allowed_cross_departments: if target_department in allowed_cross_departments[user.department]: return True return False # 服装设计相关API class DesignRequest(BaseModel): prompt: str style: str = "modern" reference_images: Optional[List[str]] = None department: str # 数据所属部门 class DesignResponse(BaseModel): design_id: str image_url: str prompt: str department: str created_by: str created_at: str @app.post("/api/designs/generate", response_model=DesignResponse) async def generate_design( request: DesignRequest, user: UserPermission = Depends(check_permission("write")) ): """生成服装设计图""" # 检查部门访问权限 if not check_department_access(request.department, user): raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="无权访问该部门的数据" ) # 调用Nano-Banana生成图片 # 这里简化处理，实际应该调用Nano-Banana的API design_data = { "design_id": "design_12345", "image_url": "https://example.com/designs/12345.png", "prompt": request.prompt, "department": request.department, "created_by": user.username, "created_at": "2024-01-15T10:30:00Z" } # 记录操作日志（安全审计） log_operation( user_id=user.user_id, action="generate_design", resource=f"design:{design_data['design_id']}", department=request.department ) return DesignResponse(**design_data) @app.get("/api/designs/{design_id}") async def get_design( design_id: str, user: UserPermission = Depends(check_permission("read")) ): """获取设计图详情""" # 这里应该从数据库查询设计图信息 # 包括设计图所属部门 design_info = { "design_id": design_id, "department": "design", # 假设这是设计部门的图 "created_by": "designer1" } # 检查部门访问权限 if not check_department_access(design_info["department"], user): raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="无权访问该设计图" ) # 记录查看日志 log_operation( user_id=user.user_id, action="view_design", resource=f"design:{design_id}", department=design_info["department"] ) return {"design_id": design_id, "access": "granted"} @app.delete("/api/designs/{design_id}") async def delete_design( design_id: str, user: UserPermission = Depends(check_permission("delete")) ): """删除设计图""" # 检查权限和部门 # 这里需要先查询设计图信息，再检查权限 # 简化处理，直接返回成功 log_operation( user_id=user.user_id, action="delete_design", resource=f"design:{design_id}", department="design" ) return {"message": "设计图已删除", "design_id": design_id} def log_operation(user_id: str, action: str, resource: str, department: str): """记录操作日志，用于安全审计""" # 实际应该写入数据库或日志系统 print(f"[AUDIT] user:{user_id} action:{action} resource:{resource} department:{department}") # 用户管理API（仅管理员） @app.get("/api/admin/users") async def list_users( user: UserPermission = Depends(check_permission("manage_users")) ): """列出所有用户（仅管理员）""" if user.role != "admin": raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="需要管理员权限" ) # 返回用户列表 return {"users": ["admin", "designer1", "viewer1"]} if __name__ == "__main__": import uvicorn uvicorn.run(app, host="0.0.0.0", port=8000)

3.4 前端集成示例

前端怎么用这个认证系统呢？我以React为例，展示关键部分。

// frontend/src/services/auth.js import axios from 'axios'; const API_BASE = process.env.REACT_APP_API_BASE || 'http://localhost:3000'; const AI_BASE = process.env.REACT_APP_AI_BASE || 'http://localhost:8000'; class AuthService { constructor() { this.token = localStorage.getItem('jwt_token'); this.user = JSON.parse(localStorage.getItem('user') || 'null'); // 配置axios默认headers if (this.token) { this.setAuthHeader(this.token); } } setAuthHeader(token) { axios.defaults.headers.common['Authorization'] = `Bearer ${token}`; } clearAuthHeader() { delete axios.defaults.headers.common['Authorization']; } async login(username, password) { try { const response = await axios.post(`${API_BASE}/api/auth/login`, { username, password }); if (response.data.success) { const { token, user } = response.data; // 保存到localStorage localStorage.setItem('jwt_token', token); localStorage.setItem('user', JSON.stringify(user)); // 设置axios header this.setAuthHeader(token); this.token = token; this.user = user; // 设置自动刷新token this.scheduleTokenRefresh(); return { success: true, user }; } } catch (error) { console.error('登录失败:', error); return { success: false, error: error.response?.data?.message || '登录失败' }; } } logout() { localStorage.removeItem('jwt_token'); localStorage.removeItem('user'); this.clearAuthHeader(); this.token = null; this.user = null; // 清除定时刷新 if (this.refreshTimer) { clearTimeout(this.refreshTimer); } } async refreshToken() { try { const response = await axios.post(`${API_BASE}/api/auth/refresh`); if (response.data.success) { const newToken = response.data.token; localStorage.setItem('jwt_token', newToken); this.setAuthHeader(newToken); this.token = newToken; // 重新安排刷新 this.scheduleTokenRefresh(); return true; } } catch (error) { console.error('刷新Token失败:', error); // Token刷新失败，可能需要重新登录 this.logout(); return false; } } scheduleTokenRefresh() { // Token 24小时过期，我们在23小时后刷新 const refreshTime = 23 * 60 * 60 * 1000; // 23小时 if (this.refreshTimer) { clearTimeout(this.refreshTimer); } this.refreshTimer = setTimeout(() => { this.refreshToken(); }, refreshTime); } // 检查用户是否有特定权限 hasPermission(permission) { if (!this.user) return false; // 从Token中解析权限（实际应该从API获取） try { const tokenData = JSON.parse(atob(this.token.split('.')[1])); return tokenData.permissions?.includes(permission) || false; } catch (error) { console.error('解析Token权限失败:', error); return false; } } // 检查用户是否属于某个部门 isInDepartment(department) { return this.user?.department === department || this.user?.role === 'admin'; } } // AI服务调用示例 class AIService { constructor(authService) { this.auth = authService; } async generateDesign(prompt, style, department) { try { const response = await axios.post(`${AI_BASE}/api/designs/generate`, { prompt, style, department }); return response.data; } catch (error) { if (error.response?.status === 401) { // Token过期，尝试刷新 const refreshed = await this.auth.refreshToken(); if (refreshed) { // 重试请求 return this.generateDesign(prompt, style, department); } else { throw new Error('请重新登录'); } } throw error; } } async getDesign(designId) { try { const response = await axios.get(`${AI_BASE}/api/designs/${designId}`); return response.data; } catch (error) { if (error.response?.status === 403) { throw new Error('无权访问该设计图'); } throw error; } } } export { AuthService, AIService };

4. 高级安全策略与最佳实践

基础认证做好了，咱们再聊聊更高级的安全策略。企业级应用可不能只靠一个Token。

4.1 多因素认证增强

对于特别敏感的操作，比如删除设计图、导出所有数据，可以加上多因素认证。

// auth-service/src/mfa.js const speakeasy = require('speakeasy'); const QRCode = require('qrcode'); class MFAService { // 为用户生成MFA密钥 static generateSecret(username) { const secret = speakeasy.generateSecret({ name: `Nano-Banana Studio (${username})` }); return { secret: secret.base32, otpauth_url: secret.otpauth_url }; } // 生成QR码（用于扫码绑定） static async generateQRCode(otpauthUrl) { try { const qrCodeUrl = await QRCode.toDataURL(otpauthUrl); return qrCodeUrl; } catch (error) { console.error('生成QR码失败:', error); return null; } } // 验证TOTP代码 static verifyToken(secret, token) { return speakeasy.totp.verify({ secret: secret, encoding: 'base32', token: token, window: 1 // 允许前后1个时间窗口的偏差 }); } // 高风险操作需要MFA验证 static requireMFA(req, res, next) { const user = req.user; const { mfaToken } = req.body; if (!user.mfaEnabled) { // 用户未启用MFA，直接通过 return next(); } if (!mfaToken) { return res.status(403).json({ error: '需要MFA验证', message: '此操作需要多因素认证' }); } // 从数据库获取用户的MFA密钥 const userSecret = getUserMfaSecret(user.id); if (!MFAService.verifyToken(userSecret, mfaToken)) { return res.status(403).json({ error: 'MFA验证失败', message: '验证码不正确' }); } next(); } }

4.2 操作审计与日志

安全不只是防外部攻击，内部操作也要可追溯。完善的审计日志很重要。

# nano-banana-studio/app/audit.py from datetime import datetime import json from typing import Dict, Any import redis import os class AuditLogger: def __init__(self): self.redis_client = redis.Redis( host=os.getenv('REDIS_HOST', 'localhost'), port=int(os.getenv('REDIS_PORT', 6379)), db=0 ) self.log_key = "audit_logs" def log_operation(self, operation: Dict[str, Any]): """记录操作日志""" # 添加时间戳和IP地址 operation['timestamp'] = datetime.utcnow().isoformat() operation['ip_address'] = self._get_client_ip() # 敏感信息脱敏 if 'password' in operation: operation['password'] = '***REDACTED***' if 'token' in operation: operation['token'] = operation['token'][:10] + '...' # 存储到Redis（实际生产环境应该用专门的日志系统） log_entry = json.dumps(operation) self.redis_client.rpush(self.log_key, log_entry) # 同时输出到控制台（开发环境） print(f"[AUDIT] {log_entry}") def query_logs(self, user_id: str = None, action: str = None, start_time: str = None, end_time: str = None, limit: int = 100): """查询审计日志""" all_logs = self.redis_client.lrange(self.log_key, 0, -1) logs = [] for log_bytes in all_logs[-limit:]: # 取最新的limit条 log = json.loads(log_bytes.decode('utf-8')) # 过滤条件 if user_id and log.get('user_id') != user_id: continue if action and log.get('action') != action: continue if start_time and log.get('timestamp') < start_time: continue if end_time and log.get('timestamp') > end_time: continue logs.append(log) return logs def _get_client_ip(self): """获取客户端IP（简化实现）""" # 实际应该从请求头获取 return "127.0.0.1" def cleanup_old_logs(self, days_to_keep: int = 90): """清理旧日志""" cutoff_time = datetime.utcnow().timestamp() - (days_to_keep * 86400) logs = self.redis_client.lrange(self.log_key, 0, -1) keep_logs = [] for log_bytes in logs: log = json.loads(log_bytes.decode('utf-8')) log_time = datetime.fromisoformat(log['timestamp']).timestamp() if log_time > cutoff_time: keep_logs.append(log_bytes) # 重新存储 self.redis_client.delete(self.log_key) if keep_logs: self.redis_client.rpush(self.log_key, *keep_logs) # 使用示例 audit_logger = AuditLogger() # 在需要记录的操作中调用 def generate_design_with_audit(user_id, prompt, department): # ... 生成设计的逻辑 ... audit_logger.log_operation({ 'user_id': user_id, 'action': 'generate_design', 'resource_type': 'design', 'resource_id': design_id, 'department': department, 'details': { 'prompt_length': len(prompt), 'style': style }, 'result': 'success' })

4.3 速率限制与防滥用

AI生成很耗资源，得防止被滥用。

// auth-service/src/rateLimit.js const rateLimit = require('express-rate-limit'); const RedisStore = require('rate-limit-redis'); const redis = require('redis'); // 创建Redis客户端 const redisClient = redis.createClient({ host: process.env.REDIS_HOST || 'localhost', port: process.env.REDIS_PORT || 6379 }); redisClient.on('error', (err) => { console.error('Redis连接错误:', err); }); // 不同端点的速率限制配置 const rateLimitConfigs = { // 登录接口：严格限制 login: rateLimit({ store: new RedisStore({ client: redisClient, prefix: 'rate_limit:login:' }), windowMs: 15 * 60 * 1000, // 15分钟 max: 5, // 最多5次尝试 message: { error: '请求过于频繁', message: '登录尝试过多，请15分钟后再试' }, standardHeaders: true, legacyHeaders: false }), // AI生成接口：按用户限制 aiGenerate: rateLimit({ store: new RedisStore({ client: redisClient, prefix: 'rate_limit:generate:' }), windowMs: 60 * 60 * 1000, // 1小时 max: (req) => { // 根据用户角色设置不同限制 const user = req.user; if (!user) return 10; // 未认证用户 switch(user.role) { case 'admin': return 1000; // 管理员 case 'designer': return 100; // 设计师 case 'viewer': return 20; // 查看者 default: return 10; } }, message: { error: '请求过于频繁', message: '生成次数超限，请稍后再试' }, standardHeaders: true, legacyHeaders: false, keyGenerator: (req) => { // 按用户ID限流 return req.user ? req.user.userId : req.ip; } }), // API总体限制 apiGlobal: rateLimit({ store: new RedisStore({ client: redisClient, prefix: 'rate_limit:global:' }), windowMs: 1 * 60 * 1000, // 1分钟 max: 100, // 每分钟最多100次请求 message: { error: '请求过于频繁', message: 'API调用过于频繁，请稍后再试' }, standardHeaders: true, legacyHeaders: false }) }; // 智能限流：根据系统负载动态调整 class AdaptiveRateLimiter { constructor(baseLimit = 100) { this.baseLimit = baseLimit; this.currentLimit = baseLimit; this.systemLoad = 0; // 监控系统负载 setInterval(() => this.monitorSystemLoad(), 60000); // 每分钟检查一次 } async monitorSystemLoad() { // 获取系统负载（CPU、内存等） // 这里简化处理，实际应该监控真实指标 const load = await this.getSystemLoad(); this.systemLoad = load; // 根据负载调整限流 if (load > 0.8) { this.currentLimit = Math.floor(this.baseLimit * 0.5); // 负载高，减半 } else if (load > 0.6) { this.currentLimit = Math.floor(this.baseLimit * 0.8); // 负载中等，减少20% } else { this.currentLimit = this.baseLimit; // 负载正常，恢复原限制 } console.log(`系统负载: ${load.toFixed(2)}, 当前限流: ${this.currentLimit}`); } async getSystemLoad() { // 获取系统负载的简化实现 // 实际应该使用os模块或监控系统API return Math.random(); // 模拟负载 } getMiddleware() { return rateLimit({ store: new RedisStore({ client: redisClient, prefix: 'rate_limit:adaptive:' }), windowMs: 1 * 60 * 1000, // 1分钟 max: (req) => this.currentLimit, message: { error: '系统繁忙', message: '当前系统负载较高，请稍后再试' }, standardHeaders: true, legacyHeaders: false }); } } module.exports = { rateLimitConfigs, AdaptiveRateLimiter };

5. 部署与监控

系统做好了，部署和监控也很关键。这里给几个实用建议。

5.1 安全配置检查清单

部署前先对照这个清单检查一遍：

# security-checklist.yaml security_config: # 1. 认证相关 authentication: jwt_secret: "已设置强密码（长度>32，包含大小写数字符号）" token_expiry: "已设置合理过期时间（如24小时）" refresh_token: "已实现Token刷新机制" password_policy: "已启用强密码策略" # 2. 网络与通信 network: https: "已启用HTTPS" cors: "已正确配置CORS（仅允许信任域名）" internal_network: "服务间通信使用内部网络" firewall: "已配置防火墙规则" # 3. 数据安全 data: encryption_at_rest: "静态数据已加密" encryption_in_transit: "传输数据使用TLS" data_isolation: "已实现部门数据隔离" backup: "已设置定期备份" # 4. 监控与审计 monitoring: audit_logs: "已启用操作审计" access_logs: "已记录所有访问" anomaly_detection: "已设置异常检测" alerting: "已配置安全告警" # 5. 防攻击措施 protection: rate_limiting: "已启用速率限制" sql_injection: "已防护SQL注入" xss_protection: "已防护XSS攻击" csrf_protection: "已防护CSRF攻击" # 6. 合规性 compliance: data_retention: "符合数据保留政策" privacy_policy: "已实施隐私政策" user_consent: "已获取用户同意" logging_policy: "日志记录符合法规"

5.2 监控仪表板示例

用Prometheus + Grafana做个监控面板，随时掌握系统状态。

# prometheus/prometheus.yml global: scrape_interval: 15s evaluation_interval: 15s scrape_configs: - job_name: 'nano-banana-studio' static_configs: - targets: ['nano-banana:8000'] metrics_path: '/metrics' - job_name: 'auth-service' static_configs: - targets: ['auth-service:3000'] metrics_path: '/metrics' - job_name: 'redis' static_configs: - targets: ['redis:6379'] - job_name: 'postgres' static_configs: - targets: ['postgres:5432'] # 自定义指标 rule_files: - "alert.rules.yml"

# prometheus/alert.rules.yml groups: - name: nano-banana-alerts rules: # 认证失败告警 - alert: HighAuthFailureRate expr: rate(auth_failures_total[5m]) > 0.1 for: 2m labels: severity: warning annotations: summary: "认证失败率过高" description: "过去5分钟认证失败率超过10%" # 异常访问模式 - alert: AbnormalAccessPattern expr: rate(http_requests_total{handler="/api/designs/generate"}[5m]) > 100 for: 1m labels: severity: critical annotations: summary: "设计生成接口访问异常" description: "设计生成接口访问频率异常升高" # Token滥用检测 - alert: TokenAbuseDetected expr: sum by (user_id) (rate(token_refreshes_total[10m])) > 10 for: 1m labels: severity: warning annotations: summary: "Token刷新频率异常" description: "用户 {{ $labels.user_id }} Token刷新频率异常"