killvxk 实施API密钥安全控制
适用场景
- 设计具有足够熵和可识别前缀的安全API密钥生成机制，用于泄露检测
- 实施服务端API密钥哈希存储（永不以明文存储密钥），使用SHA-256或bcrypt
- 构建支持零停机密钥替换的密钥轮换工作流
- 配置按密钥范围限制，将每个API密钥限制到特定端点、IP范围和速率限制
- 设置自动化监控，检测GitHub仓库、日志和客户端代码中的API密钥泄露
不适用将API密钥作为面向用户应用的唯一认证机制。API密钥最适合服务间通信和开发者访问。
前置条件
- 安全随机数生成器（os.urandom、secrets模块）用于密钥生成
- 具有静态加密功能的数据库用于存储哈希后的API密钥
- Redis或类似存储用于密钥到元数据的缓存和速率限制
- 密钥扫描工具（GitHub secret scanning、truffleHog、gitleaks）
- 用于密钥使用异常的监控和告警基础设施
工作流程
步骤1：安全API密钥生成
import secrets
import hashlib
import hmac
import time
import json
from datetime import datetime, timedelta
class APIKeyManager:
"""管理安全API密钥生命周期：生成、存储、验证、轮换。"""
# 密钥格式：prefix_base64random（例如 sk_live_a1b2c3d4e5f6...）
# 前缀用于标识密钥类型和环境，便于泄露检测
KEY_PREFIXES = {
"live_secret": "sk_live_",
"test_secret": "sk_test_",
"live_public": "pk_live_",
"test_public": "pk_test_",
}
def __init__(self, db_connection, redis_connection):
self.db = db_connection
self.redis = redis_connection
def generate_key(self, key_type="live_secret", owner_id=None, scopes=None,
rate_limit=None, ip_allowlist=None, expires_days=365):
"""生成带元数据的新API密钥。"""
prefix = self.KEY_PREFIXES.get(key_type, "sk_live_")
# 生成32字节（256位）随机数
random_bytes = secrets.token_bytes(32)
key_body = secrets.token_urlsafe(32) # Base64url编码
# 客户端收到的完整API密钥（仅显示一次）
full_key = f"{prefix}{key_body}"
# 对密钥进行哈希处理用于存储（永不存储原始密钥）
key_hash = hashlib.sha256(full_key.encode()).hexdigest()
# 创建用于引用的短密钥ID（前8个字符）
key_id = f"{prefix}{key_body[:8]}..."
# 存储哈希密钥及元数据
key_metadata = {
"key_hash": key_hash,
"key_id": key_id,
"key_type": key_type,
"owner_id": owner_id,
"scopes": scopes or ["read"],
"rate_limit": rate_limit or {"requests": 1000, "window": 3600},
"ip_allowlist": ip_allowlist or [],
"created_at": datetime.utcnow().isoformat(),
"expires_at": (datetime.utcnow() + timedelta(days=expires_days)).isoformat(),
"last_used": None,
"is_active": True,
"usage_count": 0,
}
# 存入数据库
self.db.execute(
"INSERT INTO api_keys (key_hash, key_id, metadata) VALUES (?, ?, ?)",
(key_hash, key_id, json.dumps(key_metadata))
)
# 缓存到Redis用于快速验证
self.redis.setex(
f"apikey:{key_hash}",
86400, # 24小时缓存TTL
json.dumps(key_metadata)
)
return {
"api_key": full_key, # 仅向用户展示一次
"key_id": key_id, # 用于引用/管理
"scopes": key_metadata["scopes"],
"expires_at": key_metadata["expires_at"],
}
def validate_key(self, api_key):
"""验证API密钥并返回其元数据。"""
key_hash = hashlib.sha256(api_key.encode()).hexdigest()
# 优先检查Redis缓存
cached = self.redis.get(f"apikey:{key_hash}")
if cached:
metadata = json.loads(cached)
else:
# 回退到数据库
row = self.db.execute(
"SELECT metadata FROM api_keys WHERE key_hash = ?",
(key_hash,)
).fetchone()
if not row:
return None, "invalid_key"
metadata = json.loads(row[0])
# 刷新缓存
self.redis.setex(f"apikey:{key_hash}", 86400, row[0])
# 验证检查
if not metadata.get("is_active"):
return None, "key_revoked"
if metadata.get("expires_at"):
if datetime.fromisoformat(metadata["expires_at"]) < datetime.utcnow():
return None, "key_expired"
# 更新最后使用时间
metadata["last_used"] = datetime.utcnow().isoformat()
metadata["usage_count"] = metadata.get("usage_count", 0) + 1
self.redis.setex(f"apikey:{key_hash}", 86400, json.dumps(metadata))
return metadata, "valid"
def revoke_key(self, key_id):
"""立即吊销API密钥。"""
row = self.db.execute(
"SELECT key_hash, metadata FROM api_keys WHERE key_id = ?",
(key_id,)
).fetchone()
if row:
key_hash = row[0]
metadata = json.loads(row[1])
metadata["is_active"] = False
metadata["revoked_at"] = datetime.utcnow().isoformat()
self.db.execute(
"UPDATE api_keys SET metadata = ? WHERE key_id = ?",
(json.dumps(metadata), key_id)
)
# 立即使缓存失效
self.redis.delete(f"apikey:{key_hash}")
return True
return False
def rotate_key(self, old_key_id, grace_period_hours=24):
"""轮换API密钥，提供新旧密钥同时生效的宽限期。"""
old_row = self.db.execute(
"SELECT key_hash, metadata FROM api_keys WHERE key_id = ?",
(old_key_id,)
).fetchone()
if not old_row:
return None, "key_not_found"
old_metadata = json.loads(old_row[1])
# 使用相同设置生成新密钥
new_key_data = self.generate_key(
key_type=old_metadata["key_type"],
owner_id=old_metadata["owner_id"],
scopes=old_metadata["scopes"],
rate_limit=old_metadata["rate_limit"],
ip_allowlist=old_metadata["ip_allowlist"],
)
# 在宽限期后安排旧密钥吊销
revoke_at = datetime.utcnow() + timedelta(hours=grace_period_hours)
old_metadata["scheduled_revocation"] = revoke_at.isoformat()
self.db.execute(
"UPDATE api_keys SET metadata = ? WHERE key_id = ?",
(json.dumps(old_metadata), old_key_id)
)
return {
"new_key": new_key_data,
"old_key_id": old_key_id,
"old_key_revokes_at": revoke_at.isoformat(),
"message": f"旧密钥将在 {grace_period_hours} 小时后吊销"
}, "success"
步骤2：API密钥验证中间件
from flask import Flask, request, jsonify, g
from functools import wraps
app = Flask(__name__)
def require_api_key(required_scopes=None):
"""验证API密钥并检查范围的中间件。"""
def decorator(f):
@wraps(f)
def wrapped(*args, **kwargs):
# 从请求头提取API密钥
api_key = request.headers.get("X-API-Key")
if not api_key:
# 也检查 Authorization: Bearer <key>
auth_header = request.headers.get("Authorization", "")
if auth_header.startswith("Bearer "):
api_key = auth_header[7:]
if not api_key:
return jsonify({"error": "missing_api_key"}), 401
# 验证密钥
metadata, status = key_manager.validate_key(api_key)
if status != "valid":
return jsonify({"error": status}), 401
# 检查IP白名单
if metadata.get("ip_allowlist"):
client_ip = request.remote_addr
if client_ip not in metadata["ip_allowlist"]:
return jsonify({"error": "ip_not_allowed"}), 403
# 检查范围
if required_scopes:
key_scopes = set(metadata.get("scopes", []))
if not key_scopes.intersection(required_scopes):
return jsonify({"error": "insufficient_scope"}), 403
# 将元数据附加到请求上下文
g.api_key_metadata = metadata
return f(*args, **kwargs)
return wrapped
return decorator
@app.route('/api/v1/data', methods=['GET'])
@require_api_key(required_scopes=["read", "admin"])
def get_data():
return jsonify({"data": "敏感信息"})
@app.route('/api/v1/data', methods=['POST'])
@require_api_key(required_scopes=["write", "admin"])
def create_data():
return jsonify({"created": True})
步骤3：自动化密钥泄露检测
# 使用gitleaks扫描GitHub仓库中的泄露API密钥
gitleaks detect --source=/path/to/repo --config=gitleaks.toml --report-path=leaks.json
# 用于API密钥前缀检测的自定义gitleaks配置
# gitleaks.toml
cat <<'EOF'
[[rules]]
id = "company-api-key-live"
description = "Company Live API Key"
regex = '''sk_live_[A-Za-z0-9_-]{32,}'''
tags = ["api-key", "live", "critical"]
[[rules]]
id = "company-api-key-test"
description = "Company Test API Key"
regex = '''sk_test_[A-Za-z0-9_-]{32,}'''
tags = ["api-key", "test"]
[[rules]]
id = "company-public-key"
description = "Company Public API Key"
regex = '''pk_live_[A-Za-z0-9_-]{32,}'''
tags = ["api-key", "public"]
EOF
# 自动化泄露密钥吊销
import json
def process_leaked_keys(leaks_file):
"""自动吊销在公共仓库中检测到的API密钥。"""
with open(leaks_file) as f:
leaks = json.load(f)
for leak in leaks:
key_match = leak.get("match", "")
# 从匹配中提取密钥
for prefix in ["sk_live_", "sk_test_", "pk_live_"]:
if prefix in key_match:
start = key_match.index(prefix)
potential_key = key_match[start:start+50] # 最大密钥长度
# 验证并吊销
metadata, status = key_manager.validate_key(potential_key)
if status == "valid":
key_manager.revoke_key(metadata["key_id"])
print(f"[已吊销] 密钥 {metadata['key_id']} 在 {leak.get('file')} 中泄露")
# 通知密钥所有者
notify_owner(metadata["owner_id"], metadata["key_id"], leak)
核心概念
| 术语 | 定义 |
|---|---|
| API密钥（API Key） | 用于认证API请求的密钥字符串，通常通过请求头或查询参数传递 |
| 密钥哈希（Key Hashing） | 在数据库中只存储API密钥的哈希值（SHA-256），永不存储明文密钥，类似于密码哈希 |
| 密钥轮换（Key Rotation） | 在维持宽限期（新旧密钥均有效）的情况下将API密钥替换为新密钥，确保零停机过渡 |
| 密钥范围限制（Key Scoping） | 将每个API密钥限制到特定端点、HTTP方法、IP范围和速率限制，以最小化影响范围 |
| 密钥前缀（Key Prefix） | 可识别的前缀（如 sk_live_），使自动化扫描能够在日志、代码和公共仓库中检测泄露密钥 |
| 密钥扫描（Secret Scanning） | 自动监控仓库、日志和公共来源中暴露的API密钥和凭据 |
工具和系统
- GitHub Secret Scanning：内置的GitHub功能，检测仓库中暴露的密钥并提醒密钥提供者
- gitleaks：使用可定制正则模式检测git仓库中密钥的开源工具
- truffleHog：密钥扫描工具，搜索整个git历史中的高熵字符串和已知密钥模式
- HashiCorp Vault：企业级密钥管理系统，用于API密钥存储、轮换和动态凭据生成
- AWS Secrets Manager：托管的密钥存储服务，支持API密钥和凭据的自动轮换
常见场景
场景：开发者平台的API密钥安全程序
背景：开发者平台提供以API密钥认证的公共API，平台有10,000+个API消费者，每天产生50M+个请求，密钥频繁在公共GitHub仓库中泄露。
方法：
- 实施带前缀的API密钥（sk_live_、sk_test_），使用256位熵用于泄露检测
- 在数据库中只存储密钥的SHA-256哈希值，在Redis中缓存已验证的密钥
- 实施按密钥范围限制：每个密钥限定到特定端点、速率限制和可选的IP白名单
- 构建带24小时宽限期的密钥轮换API，实现无缝过渡
- 与GitHub Secret Scanning集成，在数分钟内自动检测和吊销泄露的密钥
- 在CI/CD管道中运行gitleaks，防止密钥首先被提交
- 实施异常检测：对从异常IP使用或具有异常流量模式的密钥发出告警
- 添加密钥过期策略：所有密钥在365天后过期，并提前30天通知
注意事项：
- 在数据库中以明文存储API密钥（应使用SHA-256哈希）
- 使用可预测或低熵的密钥生成（应使用加密安全随机生成器）
- 未实施密钥前缀，导致无法在自动化扫描中识别泄露密钥
- 允许API密钥出现在URL查询参数中，导致密钥泄露在日志、浏览器历史和Referer头中
- 未实施按密钥速率限制，允许单个受损密钥滥用整个API
输出格式
## API密钥安全实施报告
**平台**: Developer API v3
**总活跃密钥数**: 12,450
**日密钥验证次数**: 5200万
### 安全控制
| 控制措施 | 实施方式 | 状态 |
|---------|---------------|--------|
| 密钥熵 | 256位（secrets.token_urlsafe(32)） | 已实施 |
| 密钥格式 | sk_live_/sk_test_ 前缀 | 已实施 |
| 存储 | SHA-256哈希，Redis缓存 | 已实施 |
| 范围限制 | 按密钥的端点/IP/速率限制 | 已实施 |
| 轮换 | 24小时宽限期API | 已实施 |
| 过期 | 最大TTL 365天 | 已实施 |
| 泄露检测 | GitHub Secret Scanning + gitleaks | 活跃 |
| 自动吊销 | 泄露密钥5分钟内吊销 | 活跃 |
### 密钥泄露统计（过去30天）
- 公共仓库中检测到的密钥: 23个
- 平均吊销时间: 3.2分钟
- CI/CD预提交检测到的密钥: 7个（已阻断）