如何使用API公钥与私钥替代连接字符串连接MongoDB（Python）

1. 明确API密钥的适用场景 #

MongoDB原生连接协议不支持直接用「公钥+私钥」格式的API密钥作为连接认证凭证，需根据你的API密钥类型选择适配方案：

若使用MongoDB Atlas的Programmatic API密钥 #

这类密钥（公钥为PublicKey，私钥为PrivateKey）仅用于调用Atlas管理API，无法直接连接数据库。你需要通过它生成临时数据库用户凭证，再用凭证连接：

• 确保API密钥拥有创建数据库用户的权限（如atlasAdmin）
• 调用Atlas Admin API创建带过期时间的临时数据库用户
• 用临时用户名密码构造连接字符串，通过pymongo.MongoClient连接

示例代码：

import requests
import json
from pymongo import MongoClient
import secrets

# 替换为你的Atlas信息
PUBLIC_KEY = "你的公钥"
PRIVATE_KEY = "你的私钥"
PROJECT_ID = "你的项目ID"
CLUSTER_NAME = "你的集群名称"
TARGET_DB = "目标数据库名"

# 生成随机临时密码
temp_password = secrets.token_urlsafe(16)

# 1. 创建临时数据库用户
create_user_url = f"https://cloud.mongodb.com/api/atlas/v1.0/groups/{PROJECT_ID}/databaseUsers"
auth = (PUBLIC_KEY, PRIVATE_KEY)
headers = {"Content-Type": "application/json"}
user_payload = {
    "databaseName": "admin",
    "username": f"temp_app_user_{secrets.token_hex(4)}",
    "password": temp_password,
    "roles": [{"roleName": "readWrite", "databaseName": TARGET_DB}],
    "expireAt": "2024-12-31T23:59:59Z"  # 设置过期时间，建议短周期
}

response = requests.post(create_user_url, auth=auth, headers=headers, data=json.dumps(user_payload))
if response.status_code != 201:
    raise Exception(f"创建临时用户失败: {response.text}")

# 2. 连接数据库
connection_str = f"mongodb+srv://{user_payload['username']}:{temp_password}@{CLUSTER_NAME}.mongodb.net/{TARGET_DB}?retryWrites=true&w=majority"
client = MongoClient(connection_str)

# 验证连接
try:
    client.admin.command('ping')
    print("数据库连接成功")
except Exception as e:
    print(f"连接失败: {e}")

若使用应用自定义的API密钥体系 #

需要在应用层完成API密钥验证，通过后再用原有的用户名密码连接数据库：

• 先验证客户端传入的公钥/私钥合法性（比如校验签名、查询密钥白名单）
• 验证通过后，初始化pymongo.MongoClient执行数据库操作

示例代码：

from pymongo import MongoClient
import hmac
import hashlib

def verify_api_key(public_key, signed_data, client_signature):
    # 从数据库/配置中获取对应公钥的私钥（或预设密钥）
    stored_private_key = "存储的对应私钥".encode()
    # 验证签名（示例用HMAC-SHA256）
    computed_signature = hmac.new(stored_private_key, signed_data.encode(), hashlib.sha256).hexdigest()
    return hmac.compare_digest(computed_signature, client_signature)

# 客户端传入的API密钥信息
client_public_key = "客户端公钥"
client_signed_data = "客户端签名的请求数据"
client_signature = "客户端签名结果"

# 验证API密钥
if verify_api_key(client_public_key, client_signed_data, client_signature):
    # 验证通过，连接数据库
    connection_str = "mongodb+srv://固定用户名:固定密码@集群地址/目标数据库?retryWrites=true&w=majority"
    client = MongoClient(connection_str)
    # 执行数据库操作
else:
    raise PermissionError("无效的API密钥")

2. 关键注意事项 #

• MongoDB原生连接字符串不支持直接嵌入公钥/私钥，必须通过上述两种方式适配
• 使用Atlas Programmatic API密钥时，最小化权限范围，避免过度授权
• 临时数据库用户务必设置合理的过期时间，降低泄露风险

内容的提问来源于stack exchange，提问作者Achintya Agarwal