设置账号 CAM 验证

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

String connectionUrl = "jdbc:sqlserver://localhost:1433;databaseName=<数据库名>;user=<账号名>;password=<密码>;encrypt=false;";  
Connection con = DriverManager.getConnection(connectionUrl);

import logging
import os
import time

import pymssql
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 = "mssql-123456"
    user_name = "camtest"
    host = "gz-mssql-123456.sql.tencentcdb.com"
    port = 24398
    db_name = "test"
    secret_id = os.environ['AK']
    secret_key = os.environ['SK']

    connection = None
    try:
        # 获取连接
        connection = get_db_connection_using_cam(secret_id, secret_key, region,
                                                 instance_id, user_name, host, port, db_name)

        # 验证连接是否成功
        with connection.cursor() as cursor:
            cursor.execute("SELECT 'Success!';")
            result = cursor.fetchone()
            log.info(result[0])  # 应该打印 "Success!"
    except Exception as e:
        log.error(f"An error occurred: {e}")
    finally:
        if connection:
            connection.close()


def get_db_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)

            connection = pymssql.connect(
                host=host,
                port=port,
                user=user_name,
                password=auth_token,
                database=db_name
            )
            return connection
        except Exception as e:
            last_exception = e
            log.info(f"Attempt {attempt} failed.")
            time.sleep(5)

    log.error(f"All attempts failed. error: {last_exception}")
    raise last_exception


def get_auth_token(region, instance_id, user_name, cred):
    try:
        # 实例化一个 http 选项，可选的，没有特殊需求可以跳过
        http_profile = HttpProfile()
        http_profile.endpoint = "cam.tencentcloudapi.com"

        # 实例化一个 client 选项，可选的，没有特殊需求可以跳过
        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()