开启 CAM 验证

{
    "statement": [
        {
            "action": [
                "cam:GetMongoDBPassword"
            ],
            "effect": "allow",
            "resource": [
                "qcs::cam::uin/<用户 uin>:resourceUser/<实例 ID>/<账号名>",
            ]
        }
    ],
    "version": "2.0"
}

<dependency>
  <groupId>org.mongodb</groupId>
  <artifactId>mongodb-driver-sync</artifactId>
  <version>4.11.0</version>
</dependency>

import logging
import os
import time
 
from pymongo import MongoClient
 
from dbauth.db_authentication import DBAuthentication
from dbauth.model.generate_authentication_token_request import GenerateAuthenticationTokenRequest
from tencentcloud.common import credential
from tencentcloud.common.exception.tencent_cloud_sdk_exception import TencentCloudSDKException
from tencentcloud.common.profile.client_profile import ClientProfile
from tencentcloud.common.profile.http_profile import HttpProfile
 
# 配置root logger
logging.basicConfig(
    level=logging.INFO,
    format='[%(asctime)s] - [%(threadName)s] - {%(module)s:%(funcName)s:%(lineno)d} %(levelname)s - %(message)s',
    datefmt='%Y-%m-%d %H:%M:%S'
)
log = logging.getLogger(__name__)
 
def main():
    region = "ap-guangzhou"
    instance_id = "cmgo-xxxx"   # MongoDB实例ID
    user_name = "camtest"
    host = "10.0.0.1"
    port = 27017
    db_name = "test"
    secret_id = os.environ['AK']
    secret_key = os.environ['SK']
 
    client = None
    try:
        # 获取MongoDB连接
        client = get_mongo_connection_using_cam(secret_id, secret_key, region,
                                                instance_id, user_name, host, port, db_name)
 
        # 验证连接是否成功
        db = client[db_name]
 
        # 查询测试
        dummy_collections = db.list_collection_names()
        log.info(f"Collections: {dummy_collections}")
        log.info("Success!")
    except Exception as e:
        log.error(f"An error occurred: {e}")
    finally:
        if client:
            client.close()
 
def get_mongo_connection_using_cam(secret_id, secret_key, region, instance_id, user_name, host, port, db_name):
    cred = credential.Credential(secret_id, secret_key)
 
    max_attempts = 3
    last_exception = None
    for attempt in range(1, max_attempts + 1):
        try:
            auth_token = get_auth_token(region, instance_id, user_name, cred)
 
            # MongoDB连接字符串
            mongo_uri = (
                f"mongodb://{user_name}:{auth_token}@{host}:{port}/{db_name}?authSource={db_name}"
            )
 
            client = MongoClient(mongo_uri, serverSelectionTimeoutMS=5000)
            # 测试连接
            client.admin.command("ping")
            return client
        except Exception as e:
            last_exception = e
            log.info(f"Attempt {attempt} failed: {e}")
            time.sleep(5)
 
    log.error(f"All attempts failed: {last_exception}")
    raise last_exception
 
def get_auth_token(region, instance_id, user_name, cred):
    try:
        http_profile = HttpProfile()
        http_profile.endpoint = "cam.tencentcloudapi.com"
 
        client_profile = ClientProfile()
        client_profile.httpProfile = http_profile
 
        request = GenerateAuthenticationTokenRequest(
            region=region,
            instance_id=instance_id,
            user_name=user_name,
            credential=cred,
            client_profile=client_profile
        )
        return DBAuthentication.generate_authentication_token(request)
    except TencentCloudSDKException as err:
        log.error(err)
        raise
 
if __name__ == "__main__":
    main()