# 解班客项目安全文档
## 1. 安全概述
### 1.1 安全目标
- **数据保护**：确保用户数据和业务数据的机密性、完整性和可用性
- **系统安全**：防范各类网络攻击和安全威胁
- **合规要求**：满足相关法律法规和行业标准要求
- **隐私保护**：保护用户隐私信息不被泄露或滥用
### 1.2 安全架构
```mermaid
graph TB
A[用户] --> B[CDN/WAF]
B --> C[负载均衡器]
C --> D[Web应用防火墙]
D --> E[反向代理]
E --> F[应用服务器]
F --> G[数据库]
F --> H[缓存服务]
F --> I[文件存储]
J[安全监控] --> K[入侵检测]
J --> L[日志审计]
J --> M[漏洞扫描]
N[身份认证] --> O[JWT令牌]
N --> P[多因子认证]
N --> Q[权限控制]
subgraph "安全防护层"
B
D
E
end
subgraph "应用安全层"
F
N
end
subgraph "数据安全层"
G
H
I
end
subgraph "监控审计层"
J
K
L
M
end
```
### 1.3 安全原则
- **最小权限原则**：用户和系统组件只获得完成任务所需的最小权限
- **纵深防御**：多层安全防护，单点失效不影响整体安全
- **零信任架构**：不信任任何用户或设备，始终验证身份和权限
- **数据分类保护**：根据数据敏感性实施不同级别的保护措施
## 2. 威胁分析
### 2.1 威胁模型
```mermaid
graph LR
A[威胁来源] --> B[外部攻击者]
A --> C[内部威胁]
A --> D[第三方风险]
B --> B1[黑客攻击]
B --> B2[恶意软件]
B --> B3[DDoS攻击]
C --> C1[内部人员]
C --> C2[权限滥用]
C --> C3[数据泄露]
D --> D1[供应商风险]
D --> D2[第三方服务]
D --> D3[开源组件]
```
### 2.2 风险评估矩阵
| 威胁类型 | 可能性 | 影响程度 | 风险等级 | 防护措施 |
|---------|--------|----------|----------|----------|
| SQL注入 | 中 | 高 | 高 | 参数化查询、输入验证 |
| XSS攻击 | 高 | 中 | 高 | 输出编码、CSP策略 |
| CSRF攻击 | 中 | 中 | 中 | CSRF令牌、同源检查 |
| 暴力破解 | 高 | 中 | 高 | 账户锁定、验证码 |
| 数据泄露 | 低 | 高 | 中 | 数据加密、访问控制 |
| DDoS攻击 | 中 | 高 | 高 | 流量清洗、限流 |
| 内部威胁 | 低 | 高 | 中 | 权限管理、审计日志 |
### 2.3 攻击场景分析
#### 2.3.1 Web应用攻击
```javascript
// 常见攻击示例和防护
const securityExamples = {
// SQL注入攻击示例
sqlInjection: {
vulnerable: "SELECT * FROM users WHERE id = " + userId,
secure: "SELECT * FROM users WHERE id = ?", // 使用参数化查询
prevention: [
"使用ORM框架",
"参数化查询",
"输入验证",
"最小权限数据库用户"
]
},
// XSS攻击示例
xssAttack: {
vulnerable: `${userInput}
`,
secure: `${escapeHtml(userInput)}
`,
prevention: [
"输出编码",
"CSP策略",
"输入验证",
"使用安全的模板引擎"
]
},
// CSRF攻击示例
csrfAttack: {
vulnerable: "POST /api/transfer without token",
secure: "POST /api/transfer with CSRF token",
prevention: [
"CSRF令牌",
"SameSite Cookie",
"双重提交Cookie",
"验证Referer头"
]
}
};
```
## 3. 身份认证与授权
### 3.1 认证架构
```mermaid
sequenceDiagram
participant U as 用户
participant F as 前端
participant A as 认证服务
participant R as 资源服务
participant D as 数据库
U->>F: 登录请求
F->>A: 提交凭证
A->>D: 验证用户
D-->>A: 用户信息
A->>A: 生成JWT令牌
A-->>F: 返回令牌
F-->>U: 登录成功
U->>F: 访问资源
F->>R: 携带JWT令牌
R->>A: 验证令牌
A-->>R: 令牌有效
R->>D: 查询数据
D-->>R: 返回数据
R-->>F: 返回结果
F-->>U: 显示内容
```
### 3.2 JWT令牌安全配置
```javascript
// JWT配置
const jwtConfig = {
// 令牌配置
secret: process.env.JWT_SECRET, // 256位随机密钥
algorithm: 'HS256',
expiresIn: '2h', // 访问令牌2小时过期
refreshExpiresIn: '7d', // 刷新令牌7天过期
// 安全选项
issuer: 'jiebanke.com',
audience: 'jiebanke-api',
notBefore: 0,
// 令牌载荷
payload: {
userId: 'user.id',
username: 'user.username',
role: 'user.role',
permissions: 'user.permissions'
}
};
// JWT中间件
const jwtMiddleware = (req, res, next) => {
const token = req.headers.authorization?.replace('Bearer ', '');
if (!token) {
return res.status(401).json({ error: '缺少访问令牌' });
}
try {
const decoded = jwt.verify(token, jwtConfig.secret, {
issuer: jwtConfig.issuer,
audience: jwtConfig.audience,
algorithms: [jwtConfig.algorithm]
});
// 检查令牌黑名单
if (await isTokenBlacklisted(token)) {
return res.status(401).json({ error: '令牌已失效' });
}
req.user = decoded;
next();
} catch (error) {
return res.status(401).json({ error: '无效的访问令牌' });
}
};
```
### 3.3 权限控制系统
```javascript
// RBAC权限模型
const rbacModel = {
roles: {
admin: {
name: '管理员',
permissions: ['*'] // 所有权限
},
moderator: {
name: '版主',
permissions: [
'trip:read',
'trip:update',
'trip:delete',
'user:read',
'comment:moderate'
]
},
user: {
name: '普通用户',
permissions: [
'trip:read',
'trip:create',
'trip:update:own',
'comment:create',
'profile:update:own'
]
}
},
resources: {
trip: ['create', 'read', 'update', 'delete'],
user: ['create', 'read', 'update', 'delete'],
comment: ['create', 'read', 'update', 'delete', 'moderate'],
profile: ['read', 'update']
}
};
// 权限检查中间件
const checkPermission = (resource, action) => {
return (req, res, next) => {
const user = req.user;
const userRole = user.role;
const userPermissions = rbacModel.roles[userRole]?.permissions || [];
// 检查是否有通配符权限
if (userPermissions.includes('*')) {
return next();
}
// 检查具体权限
const requiredPermission = `${resource}:${action}`;
const hasPermission = userPermissions.some(permission => {
if (permission === requiredPermission) return true;
if (permission.endsWith(':*') && requiredPermission.startsWith(permission.slice(0, -1))) return true;
return false;
});
// 检查资源所有权
if (!hasPermission && action.endsWith(':own')) {
const basePermission = requiredPermission.replace(':own', '');
if (userPermissions.includes(basePermission + ':own')) {
// 需要验证资源所有权
return checkResourceOwnership(resource, req, res, next);
}
}
if (!hasPermission) {
return res.status(403).json({ error: '权限不足' });
}
next();
};
};
```
### 3.4 多因子认证
```javascript
// 多因子认证实现
const mfaService = {
// 生成TOTP密钥
generateSecret: (userId) => {
const secret = speakeasy.generateSecret({
name: `解班客 (${userId})`,
issuer: '解班客',
length: 32
});
return {
secret: secret.base32,
qrCode: qrcode.toDataURL(secret.otpauth_url)
};
},
// 验证TOTP令牌
verifyToken: (secret, token) => {
return speakeasy.totp.verify({
secret: secret,
encoding: 'base32',
token: token,
window: 2 // 允许时间偏差
});
},
// 生成备用码
generateBackupCodes: () => {
const codes = [];
for (let i = 0; i < 10; i++) {
codes.push(crypto.randomBytes(4).toString('hex').toUpperCase());
}
return codes;
}
};
// MFA中间件
const mfaMiddleware = async (req, res, next) => {
const user = req.user;
// 检查用户是否启用了MFA
const userMfa = await getUserMfaSettings(user.userId);
if (!userMfa.enabled) {
return next();
}
const mfaToken = req.headers['x-mfa-token'];
if (!mfaToken) {
return res.status(401).json({
error: '需要多因子认证',
mfaRequired: true
});
}
// 验证MFA令牌
const isValid = mfaService.verifyToken(userMfa.secret, mfaToken);
if (!isValid) {
// 尝试备用码
const backupCodeValid = await verifyBackupCode(user.userId, mfaToken);
if (!backupCodeValid) {
return res.status(401).json({ error: '多因子认证失败' });
}
}
next();
};
```
## 4. 数据安全
### 4.1 数据分类与保护
| 数据类型 | 敏感级别 | 保护措施 | 访问控制 |
|---------|----------|----------|----------|
| 用户密码 | 极高 | bcrypt加密、盐值 | 仅认证服务访问 |
| 身份证号 | 高 | AES-256加密 | 管理员+审计日志 |
| 手机号码 | 高 | AES-256加密 | 业务需要+脱敏显示 |
| 邮箱地址 | 中 | 明文存储 | 用户本人+管理员 |
| 旅行信息 | 中 | 明文存储 | 公开+隐私设置 |
| 系统日志 | 低 | 压缩存储 | 运维人员 |
### 4.2 数据加密实现
```javascript
// 数据加密服务
const encryptionService = {
// AES-256-GCM加密
encrypt: (plaintext, key = process.env.ENCRYPTION_KEY) => {
const iv = crypto.randomBytes(16);
const cipher = crypto.createCipher('aes-256-gcm', key, iv);
let encrypted = cipher.update(plaintext, 'utf8', 'hex');
encrypted += cipher.final('hex');
const authTag = cipher.getAuthTag();
return {
encrypted,
iv: iv.toString('hex'),
authTag: authTag.toString('hex')
};
},
// AES-256-GCM解密
decrypt: (encryptedData, key = process.env.ENCRYPTION_KEY) => {
const decipher = crypto.createDecipher('aes-256-gcm', key,
Buffer.from(encryptedData.iv, 'hex'));
decipher.setAuthTag(Buffer.from(encryptedData.authTag, 'hex'));
let decrypted = decipher.update(encryptedData.encrypted, 'hex', 'utf8');
decrypted += decipher.final('utf8');
return decrypted;
},
// 密码哈希
hashPassword: async (password) => {
const saltRounds = 12;
return await bcrypt.hash(password, saltRounds);
},
// 密码验证
verifyPassword: async (password, hash) => {
return await bcrypt.compare(password, hash);
}
};
// 敏感数据处理中间件
const sensitiveDataMiddleware = {
// 加密敏感字段
encryptSensitiveFields: (data, fields) => {
const result = { ...data };
fields.forEach(field => {
if (result[field]) {
result[field] = encryptionService.encrypt(result[field]);
}
});
return result;
},
// 解密敏感字段
decryptSensitiveFields: (data, fields) => {
const result = { ...data };
fields.forEach(field => {
if (result[field] && typeof result[field] === 'object') {
result[field] = encryptionService.decrypt(result[field]);
}
});
return result;
},
// 数据脱敏
maskSensitiveData: (data, field) => {
if (!data[field]) return data;
const value = data[field];
let masked;
switch (field) {
case 'phone':
masked = value.replace(/(\d{3})\d{4}(\d{4})/, '$1****$2');
break;
case 'email':
masked = value.replace(/(.{2}).*(@.*)/, '$1***$2');
break;
case 'idCard':
masked = value.replace(/(\d{6})\d{8}(\d{4})/, '$1********$2');
break;
default:
masked = '***';
}
return { ...data, [field]: masked };
}
};
```
### 4.3 数据库安全配置
```sql
-- 数据库安全配置
-- 1. 创建专用数据库用户
CREATE USER 'jiebanke_app'@'%' IDENTIFIED BY 'strong_password_here';
CREATE USER 'jiebanke_readonly'@'%' IDENTIFIED BY 'readonly_password_here';
-- 2. 授予最小权限
GRANT SELECT, INSERT, UPDATE, DELETE ON jiebanke.* TO 'jiebanke_app'@'%';
GRANT SELECT ON jiebanke.* TO 'jiebanke_readonly'@'%';
-- 3. 禁用危险权限
REVOKE FILE ON *.* FROM 'jiebanke_app'@'%';
REVOKE PROCESS ON *.* FROM 'jiebanke_app'@'%';
REVOKE SUPER ON *.* FROM 'jiebanke_app'@'%';
-- 4. 启用SSL连接
ALTER USER 'jiebanke_app'@'%' REQUIRE SSL;
-- 5. 设置连接限制
ALTER USER 'jiebanke_app'@'%' WITH MAX_CONNECTIONS_PER_HOUR 1000;
ALTER USER 'jiebanke_app'@'%' WITH MAX_QUERIES_PER_HOUR 10000;
-- 6. 创建审计表
CREATE TABLE security_audit_log (
id BIGINT AUTO_INCREMENT PRIMARY KEY,
user_id INT,
action VARCHAR(50),
resource VARCHAR(100),
ip_address VARCHAR(45),
user_agent TEXT,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
INDEX idx_user_id (user_id),
INDEX idx_action (action),
INDEX idx_created_at (created_at)
);
```
## 5. 网络安全
### 5.1 网络架构安全
```mermaid
graph TB
A[Internet] --> B[CDN]
B --> C[WAF]
C --> D[Load Balancer]
D --> E[DMZ]
E --> F[Web Server]
F --> G[Internal Network]
G --> H[App Server]
G --> I[Database Server]
J[VPN] --> G
K[Bastion Host] --> G
subgraph "Public Zone"
B
C
D
end
subgraph "DMZ Zone"
E
F
end
subgraph "Internal Zone"
G
H
I
end
subgraph "Management Zone"
J
K
end
```
### 5.2 防火墙配置
```bash
#!/bin/bash
# firewall-config.sh - 防火墙配置脚本
# 清空现有规则
iptables -F
iptables -X
iptables -t nat -F
iptables -t nat -X
# 设置默认策略
iptables -P INPUT DROP
iptables -P FORWARD DROP
iptables -P OUTPUT ACCEPT
# 允许本地回环
iptables -A INPUT -i lo -j ACCEPT
iptables -A OUTPUT -o lo -j ACCEPT
# 允许已建立的连接
iptables -A INPUT -m state --state ESTABLISHED,RELATED -j ACCEPT
# 允许SSH (限制IP)
iptables -A INPUT -p tcp --dport 22 -s 192.168.1.0/24 -j ACCEPT
# 允许HTTP/HTTPS
iptables -A INPUT -p tcp --dport 80 -j ACCEPT
iptables -A INPUT -p tcp --dport 443 -j ACCEPT
# 允许应用端口 (仅内网)
iptables -A INPUT -p tcp --dport 3000 -s 192.168.1.0/24 -j ACCEPT
# 防止DDoS攻击
iptables -A INPUT -p tcp --dport 80 -m limit --limit 25/minute --limit-burst 100 -j ACCEPT
iptables -A INPUT -p tcp --dport 443 -m limit --limit 25/minute --limit-burst 100 -j ACCEPT
# 防止端口扫描
iptables -A INPUT -m recent --name portscan --rcheck --seconds 86400 -j DROP
iptables -A INPUT -m recent --name portscan --remove
iptables -A INPUT -p tcp -m tcp --dport 139 -m recent --name portscan --set -j LOG --log-prefix "portscan:"
iptables -A INPUT -p tcp -m tcp --dport 139 -j DROP
# 保存规则
iptables-save > /etc/iptables/rules.v4
echo "防火墙配置完成"
```
### 5.3 WAF规则配置
```nginx
# WAF配置 - ModSecurity规则
# /etc/nginx/modsec/main.conf
# 基础配置
SecRuleEngine On
SecRequestBodyAccess On
SecResponseBodyAccess Off
SecRequestBodyLimit 13107200
SecRequestBodyNoFilesLimit 131072
# 日志配置
SecAuditEngine RelevantOnly
SecAuditLog /var/log/nginx/modsec_audit.log
SecAuditLogFormat JSON
# SQL注入防护
SecRule ARGS "@detectSQLi" \
"id:1001,\
phase:2,\
block,\
msg:'SQL Injection Attack Detected',\
logdata:'Matched Data: %{MATCHED_VAR} found within %{MATCHED_VAR_NAME}',\
tag:'application-multi',\
tag:'language-multi',\
tag:'platform-multi',\
tag:'attack-sqli'"
# XSS攻击防护
SecRule ARGS "@detectXSS" \
"id:1002,\
phase:2,\
block,\
msg:'XSS Attack Detected',\
logdata:'Matched Data: %{MATCHED_VAR} found within %{MATCHED_VAR_NAME}',\
tag:'application-multi',\
tag:'language-multi',\
tag:'platform-multi',\
tag:'attack-xss'"
# 文件上传限制
SecRule FILES_TMPNAMES "@inspectFile /etc/nginx/modsec/file_inspection.lua" \
"id:1003,\
phase:2,\
block,\
msg:'Malicious File Upload Detected'"
# 频率限制
SecRule IP:REQUEST_COUNT "@gt 100" \
"id:1004,\
phase:1,\
deny,\
msg:'Rate Limiting - Too Many Requests',\
setvar:ip.request_count=+1,\
expirevar:ip.request_count=60"
# 地理位置限制
SecRule REMOTE_ADDR "@geoLookup" \
"id:1005,\
phase:1,\
chain,\
msg:'Request from blocked country'"
SecRule GEO:COUNTRY_CODE "@within CN US JP KR" \
"t:none,\
deny"
```
### 5.4 SSL/TLS配置
```nginx
# SSL/TLS安全配置
server {
listen 443 ssl http2;
server_name api.jiebanke.com;
# SSL证书配置
ssl_certificate /etc/ssl/certs/jiebanke.crt;
ssl_certificate_key /etc/ssl/private/jiebanke.key;
# SSL协议和加密套件
ssl_protocols TLSv1.2 TLSv1.3;
ssl_ciphers ECDHE-RSA-AES128-GCM-SHA256:ECDHE-RSA-AES256-GCM-SHA384:ECDHE-RSA-CHACHA20-POLY1305;
ssl_prefer_server_ciphers off;
# SSL会话配置
ssl_session_cache shared:SSL:10m;
ssl_session_timeout 10m;
ssl_session_tickets off;
# OCSP装订
ssl_stapling on;
ssl_stapling_verify on;
ssl_trusted_certificate /etc/ssl/certs/ca-certificates.crt;
resolver 8.8.8.8 8.8.4.4 valid=300s;
resolver_timeout 5s;
# 安全头部
add_header Strict-Transport-Security "max-age=31536000; includeSubDomains; preload" always;
add_header X-Frame-Options "SAMEORIGIN" always;
add_header X-Content-Type-Options "nosniff" always;
add_header X-XSS-Protection "1; mode=block" always;
add_header Referrer-Policy "strict-origin-when-cross-origin" always;
add_header Content-Security-Policy "default-src 'self'; script-src 'self' 'unsafe-inline'; style-src 'self' 'unsafe-inline'; img-src 'self' data: https:; font-src 'self' data:; connect-src 'self'; frame-ancestors 'self';" always;
# 隐藏服务器信息
server_tokens off;
more_clear_headers Server;
location / {
proxy_pass http://backend;
proxy_set_header Host $host;
proxy_set_header X-Real-IP $remote_addr;
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
proxy_set_header X-Forwarded-Proto $scheme;
# 安全头部
proxy_hide_header X-Powered-By;
proxy_hide_header Server;
}
}
```
## 6. 应用安全
### 6.1 输入验证与过滤
```javascript
// 输入验证中间件
const inputValidation = {
// 通用验证规则
rules: {
username: {
type: 'string',
minLength: 3,
maxLength: 20,
pattern: /^[a-zA-Z0-9_]+$/,
sanitize: true
},
email: {
type: 'email',
maxLength: 100,
sanitize: true
},
password: {
type: 'string',
minLength: 8,
maxLength: 128,
pattern: /^(?=.*[a-z])(?=.*[A-Z])(?=.*\d)(?=.*[@$!%*?&])[A-Za-z\d@$!%*?&]/
},
phone: {
type: 'string',
pattern: /^1[3-9]\d{9}$/,
sanitize: true
}
},
// 验证函数
validate: (data, rules) => {
const errors = [];
for (const [field, rule] of Object.entries(rules)) {
const value = data[field];
// 必填检查
if (rule.required && (!value || value.trim() === '')) {
errors.push(`${field} 是必填字段`);
continue;
}
if (!value) continue;
// 类型检查
if (rule.type === 'email' && !validator.isEmail(value)) {
errors.push(`${field} 格式不正确`);
}
// 长度检查
if (rule.minLength && value.length < rule.minLength) {
errors.push(`${field} 长度不能少于 ${rule.minLength} 个字符`);
}
if (rule.maxLength && value.length > rule.maxLength) {
errors.push(`${field} 长度不能超过 ${rule.maxLength} 个字符`);
}
// 正则检查
if (rule.pattern && !rule.pattern.test(value)) {
errors.push(`${field} 格式不正确`);
}
}
return errors;
},
// 数据清理
sanitize: (data, rules) => {
const sanitized = {};
for (const [field, rule] of Object.entries(rules)) {
let value = data[field];
if (!value) {
sanitized[field] = value;
continue;
}
if (rule.sanitize) {
// HTML转义
value = validator.escape(value);
// 去除首尾空格
value = value.trim();
// SQL注入防护
value = value.replace(/['"\\]/g, '\\$&');
}
sanitized[field] = value;
}
return sanitized;
}
};
// 验证中间件
const validateInput = (rules) => {
return (req, res, next) => {
const errors = inputValidation.validate(req.body, rules);
if (errors.length > 0) {
return res.status(400).json({
error: '输入验证失败',
details: errors
});
}
// 清理输入数据
req.body = inputValidation.sanitize(req.body, rules);
next();
};
};
```
### 6.2 CSRF防护
```javascript
// CSRF防护中间件
const csrfProtection = {
// 生成CSRF令牌
generateToken: (req) => {
const token = crypto.randomBytes(32).toString('hex');
req.session.csrfToken = token;
return token;
},
// 验证CSRF令牌
verifyToken: (req) => {
const sessionToken = req.session.csrfToken;
const requestToken = req.headers['x-csrf-token'] || req.body._csrf;
return sessionToken && requestToken && sessionToken === requestToken;
},
// CSRF中间件
middleware: (req, res, next) => {
// GET请求不需要CSRF验证
if (req.method === 'GET') {
return next();
}
// API请求使用JWT，不需要CSRF验证
if (req.path.startsWith('/api/')) {
return next();
}
if (!csrfProtection.verifyToken(req)) {
return res.status(403).json({ error: 'CSRF令牌验证失败' });
}
next();
}
};
// 双重提交Cookie CSRF防护
const doubleSubmitCsrf = {
middleware: (req, res, next) => {
if (req.method === 'GET') {
// 设置CSRF Cookie
const token = crypto.randomBytes(32).toString('hex');
res.cookie('csrf-token', token, {
httpOnly: false,
secure: process.env.NODE_ENV === 'production',
sameSite: 'strict'
});
return next();
}
const cookieToken = req.cookies['csrf-token'];
const headerToken = req.headers['x-csrf-token'];
if (!cookieToken || !headerToken || cookieToken !== headerToken) {
return res.status(403).json({ error: 'CSRF验证失败' });
}
next();
}
};
```
### 6.3 会话安全
```javascript
// 会话安全配置
const sessionConfig = {
secret: process.env.SESSION_SECRET,
name: 'jiebanke.sid',
resave: false,
saveUninitialized: false,
rolling: true,
cookie: {
secure: process.env.NODE_ENV === 'production',
httpOnly: true,
maxAge: 2 * 60 * 60 * 1000, // 2小时
sameSite: 'strict'
},
store: new RedisStore({
client: redisClient,
prefix: 'sess:',
ttl: 2 * 60 * 60 // 2小时
})
};
// 会话安全中间件
const sessionSecurity = {
// 会话固定攻击防护
regenerateSession: (req, res, next) => {
if (req.session && req.session.regenerate) {
req.session.regenerate((err) => {
if (err) {
return next(err);
}
next();
});
} else {
next();
}
},
// 会话劫持防护
validateSession: (req, res, next) => {
if (!req.session.userAgent) {
req.session.userAgent = req.headers['user-agent'];
req.session.ipAddress = req.ip;
} else {
// 检查User-Agent和IP是否一致
if (req.session.userAgent !== req.headers['user-agent'] ||
req.session.ipAddress !== req.ip) {
req.session.destroy();
return res.status(401).json({ error: '会话已失效，请重新登录' });
}
}
next();
},
// 并发会话控制
limitConcurrentSessions: async (req, res, next) => {
if (!req.user) return next();
const userId = req.user.userId;
const currentSessionId = req.sessionID;
// 获取用户的所有活跃会话
const activeSessions = await redis.smembers(`user:${userId}:sessions`);
// 限制最多3个并发会话
if (activeSessions.length >= 3 && !activeSessions.includes(currentSessionId)) {
// 删除最旧的会话
const oldestSession = activeSessions[0];
await redis.srem(`user:${userId}:sessions`, oldestSession);
await redis.del(`sess:${oldestSession}`);
}
// 添加当前会话
await redis.sadd(`user:${userId}:sessions`, currentSessionId);
await redis.expire(`user:${userId}:sessions`, 2 * 60 * 60);
next();
}
};
```
## 7. 安全监控与审计
### 7.1 安全事件监控
```javascript
// 安全事件监控系统
const securityMonitor = {
// 事件类型定义
eventTypes: {
LOGIN_SUCCESS: 'login_success',
LOGIN_FAILURE: 'login_failure',
PASSWORD_CHANGE: 'password_change',
PERMISSION_DENIED: 'permission_denied',
SUSPICIOUS_ACTIVITY: 'suspicious_activity',
DATA_ACCESS: 'data_access',
ADMIN_ACTION: 'admin_action'
},
// 记录安全事件
logEvent: async (eventType, userId, details = {}) => {
const event = {
id: uuidv4(),
type: eventType,
userId: userId,
timestamp: new Date(),
ipAddress: details.ipAddress,
userAgent: details.userAgent,
resource: details.resource,
action: details.action,
result: details.result,
riskLevel: details.riskLevel || 'low',
metadata: details.metadata || {}
};
// 存储到数据库
await SecurityAuditLog.create(event);
// 发送到日志系统
logger.info('Security Event', event);
// 检查是否需要告警
await this.checkAlerts(event);
},
// 告警检查
checkAlerts: async (event) => {
const alerts = [];
// 登录失败次数检查
if (event.type === this.eventTypes.LOGIN_FAILURE) {
const recentFailures = await this.getRecentEvents(
event.userId,
this.eventTypes.LOGIN_FAILURE,
15 * 60 * 1000 // 15分钟内
);
if (recentFailures.length >= 5) {
alerts.push({
type: 'BRUTE_FORCE_ATTACK',
severity: 'high',
message: `用户 ${event.userId} 15分钟内登录失败${recentFailures.length}次`
});
}
}
// 异常IP访问检查
if (event.ipAddress) {
const userIPs = await this.getUserRecentIPs(event.userId, 24 * 60 * 60 * 1000);
if (userIPs.length > 1 && !userIPs.includes(event.ipAddress)) {
alerts.push({
type: 'UNUSUAL_IP_ACCESS',
severity: 'medium',
message: `用户 ${event.userId} 从新IP地址 ${event.ipAddress} 访问`
});
}
}
// 权限提升检查
if (event.type === this.eventTypes.ADMIN_ACTION && event.userId) {
alerts.push({
type: 'ADMIN_ACTION',
severity: 'medium',
message: `管理员 ${event.userId} 执行了操作: ${event.action}`
});
}
// 发送告警
for (const alert of alerts) {
await this.sendAlert(alert, event);
}
},
// 发送安全告警
sendAlert: async (alert, event) => {
const message = {
title: `安全告警: ${alert.type}`,
severity: alert.severity,
message: alert.message,
timestamp: event.timestamp,
details: event
};
// 发送到钉钉
await this.sendDingTalkAlert(message);
// 发送邮件
if (alert.severity === 'high') {
await this.sendEmailAlert(message);
}
// 记录告警
await SecurityAlert.create({
type: alert.type,
severity: alert.severity,
message: alert.message,
eventId: event.id,
status: 'open'
});
}
};
// 安全监控中间件
const securityMonitorMiddleware = (req, res, next) => {
// 记录请求信息
const originalSend = res.send;
res.send = function(data) {
const statusCode = res.statusCode;
// 记录安全相关事件
if (req.path.includes('/auth/login')) {
const eventType = statusCode === 200 ?
securityMonitor.eventTypes.LOGIN_SUCCESS :
securityMonitor.eventTypes.LOGIN_FAILURE;
securityMonitor.logEvent(eventType, req.body.username, {
ipAddress: req.ip,
userAgent: req.headers['user-agent'],
result: statusCode === 200 ? 'success' : 'failure'
});
}
if (statusCode === 403) {
securityMonitor.logEvent(securityMonitor.eventTypes.PERMISSION_DENIED, req.user?.userId, {
ipAddress: req.ip,
userAgent: req.headers['user-agent'],
resource: req.path,
action: req.method
});
}
originalSend.call(this, data);
};
next();
};
```
### 7.2 入侵检测系统
```python
#!/usr/bin/env python3
# intrusion_detection.py - 入侵检测系统
import re
import json
import time
from collections import defaultdict, deque
from datetime import datetime, timedelta
class IntrusionDetectionSystem:
def __init__(self):
self.attack_patterns = {
'sql_injection': [
r"(\bunion\b.*\bselect\b)",
r"(\bselect\b.*\bfrom\b.*\bwhere\b)",
r"(\bdrop\b.*\btable\b)",
r"(\binsert\b.*\binto\b)",
r"(\bdelete\b.*\bfrom\b)"
],
'xss_attack': [
r"]*>.*?",
r"javascript:",
r"on\w+\s*=",
r"]*>.*?"
],
'path_traversal': [
r"\.\./",
r"\.\.\\",
r"/etc/passwd",
r"/etc/shadow"
],
'command_injection': [
r";\s*(cat|ls|pwd|whoami)",
r"\|\s*(cat|ls|pwd|whoami)",
r"&&\s*(cat|ls|pwd|whoami)",
r"`.*`"
]
}
self.ip_requests = defaultdict(lambda: deque(maxlen=100))
self.failed_logins = defaultdict(lambda: deque(maxlen=50))
def analyze_request(self, request_data):
"""分析单个请求"""
alerts = []
# 检查攻击模式
for attack_type, patterns in self.attack_patterns.items():
for pattern in patterns:
if self._check_pattern(request_data, pattern):
alerts.append({
'type': attack_type,
'severity': 'high',
'pattern': pattern,
'timestamp': datetime.now(),
'source_ip': request_data.get('ip'),
'request_uri': request_data.get('uri'),
'user_agent': request_data.get('user_agent')
})
# 检查频率异常
ip = request_data.get('ip')
if ip:
self.ip_requests[ip].append(time.time())
if self._check_rate_limit(ip):
alerts.append({
'type': 'rate_limit_exceeded',
'severity': 'medium',
'source_ip': ip,
'request_count': len(self.ip_requests[ip]),
'timestamp': datetime.now()
})
# 检查登录失败
if request_data.get('path') == '/auth/login' and request_data.get('status') >= 400:
user = request_data.get('username', ip)
self.failed_logins[user].append(time.time())
if len(self.failed_logins[user]) >= 5:
alerts.append({
'type': 'brute_force_attack',
'severity': 'high',
'target_user': user,
'source_ip': ip,
'failed_attempts': len(self.failed_logins[user]),
'timestamp': datetime.now()
})
return alerts
def _check_pattern(self, request_data, pattern):
"""检查请求是否匹配攻击模式"""
text_fields = [
request_data.get('uri', ''),
request_data.get('query_string', ''),
request_data.get('post_data', ''),
request_data.get('headers', {}).get('user-agent', '')
]
for field in text_fields:
if re.search(pattern, field, re.IGNORECASE):
return True
return False
def _check_rate_limit(self, ip):
"""检查IP请求频率"""
now = time.time()
requests = self.ip_requests[ip]
# 1分钟内超过100个请求
recent_requests = [req for req in requests if now - req < 60]
return len(recent_requests) > 100
def generate_report(self, alerts):
"""生成入侵检测报告"""
if not alerts:
return None
report = {
'timestamp': datetime.now().isoformat(),
'total_alerts': len(alerts),
'severity_breakdown': defaultdict(int),
'attack_types': defaultdict(int),
'top_source_ips': defaultdict(int),
'alerts': alerts
}
for alert in alerts:
report['severity_breakdown'][alert['severity']] += 1
report['attack_types'][alert['type']] += 1
if 'source_ip' in alert:
report['top_source_ips'][alert['source_ip']] += 1
return report
def block_ip(self, ip, duration=3600):
"""阻止IP访问"""
# 这里应该调用防火墙API或更新IP黑名单
print(f"Blocking IP {ip} for {duration} seconds")
# 示例：更新iptables规则
import subprocess
try:
subprocess.run([
'iptables', '-A', 'INPUT',
'-s', ip, '-j', 'DROP'
], check=True)
# 设置定时解除阻止
subprocess.run([
'at', f'now + {duration} seconds'
], input=f'iptables -D INPUT -s {ip} -j DROP\n',
text=True, check=True)
except subprocess.CalledProcessError as e:
print(f"Failed to block IP {ip}: {e}")
# 使用示例
if __name__ == "__main__":
ids = IntrusionDetectionSystem()
# 模拟请求数据
request_data = {
'ip': '192.168.1.100',
'uri': '/api/users?id=1 UNION SELECT * FROM users',
'method': 'GET',
'status': 200,
'user_agent': 'Mozilla/5.0...'
}
alerts = ids.analyze_request(request_data)
if alerts:
report = ids.generate_report(alerts)
print(json.dumps(report, indent=2, default=str))
# 自动阻止高危IP
for alert in alerts:
if alert['severity'] == 'high' and 'source_ip' in alert:
ids.block_ip(alert['source_ip'])
```
## 8. 安全测试
### 8.1 自动化安全测试
```python
#!/usr/bin/env python3
# security_test.py - 自动化安全测试
import requests
import json
import time
from urllib.parse import urljoin
class SecurityTester:
def __init__(self, base_url):
self.base_url = base_url
self.session = requests.Session()
self.vulnerabilities = []
def test_sql_injection(self):
"""SQL注入测试"""
print("Testing SQL Injection...")
payloads = [
"' OR '1'='1",
"'; DROP TABLE users; --",
"' UNION SELECT * FROM users --",
"1' AND (SELECT COUNT(*) FROM users) > 0 --"
]
test_endpoints = [
"/api/users",
"/api/trips",
"/api/auth/login"
]
for endpoint in test_endpoints:
for payload in payloads:
params = {'id': payload}
try:
response = self.session.get(
urljoin(self.base_url, endpoint),
params=params,
timeout=10
)
# 检查是否存在SQL错误信息
error_indicators = [
'mysql_fetch_array',
'ORA-01756',
'Microsoft OLE DB Provider',
'SQLServer JDBC Driver',
'PostgreSQL query failed'
]
for indicator in error_indicators:
if indicator.lower() in response.text.lower():
self.vulnerabilities.append({
'type': 'SQL Injection',
'severity': 'High',
'endpoint': endpoint,
'payload': payload,
'evidence': indicator
})
break
except requests.RequestException as e:
print(f"Request failed: {e}")
def test_xss(self):
"""XSS测试"""
print("Testing XSS...")
payloads = [
"",
"
",
"javascript:alert('XSS')",
"