设置账号 CAM 验证

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

<dependency>
<groupId>com.tencentcloudapi</groupId>
<artifactId>tencentcloud-dbauth-sdk-java</artifactId>
<version>1.0.4</version>
</dependency>

<dependency>
    <groupId>com.tencentcloudapi</groupId>
    <artifactId>tencentcloud-sdk-java</artifactId>
    <version>3.1.1039</version>
</dependency>

package com.example;
import com.tencentcloud.dbauth.DBAuthentication;
import com.tencentcloud.dbauth.model.GenerateAuthenticationTokenRequest;
import com.tencentcloudapi.common.Credential;
import com.tencentcloudapi.common.exception.TencentCloudSDKException;
import com.tencentcloudapi.common.profile.ClientProfile;
import com.tencentcloudapi.common.profile.HttpProfile;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.ResultSet;
import java.sql.Statement;

public class CAMDatabaseAuthenticationTester {
    public static void main(String[] args) throws Exception {
        // 定义连接所需的变量
        String region = "ap-guangzhou";
        String instanceId = "postgres-******";
        String userName = "dbadmin";
        String host = "10.0.0.*";
        int port = 5432;
        String dbName = "am";
        String secretId = "AKIDIa**********";
        String secretKey = "bEMGYZ**********";

        // 获取连接
        Connection connection = getDBConnectionUsingCAM(secretId, secretKey, region,
                instanceId, userName, host, port, dbName);

        // 验证连接是否成功
        Statement stmt = connection.createStatement();
         ResultSet rs = stmt.executeQuery("SELECT 'Success!';");
        while (rs.next()) {
            String id = rs.getString(1);
            System.out.println(id); // 应打印 "Success!"
        }

        // 关闭连接
        stmt.close();
        connection.close();
    }

    /**
     * 使用 CAM 数据库认证获取数据库连接
     *
     * @param secretId   密钥 ID
     * @param secretKey  密钥
     * @param region     地区
     * @param instanceId 实例 ID
     * @param userName   用户名
     * @param host       主机
     * @param port       端口
     * @param dbName     数据库名称
     * @return Connection 对象
     * @throws Exception 异常
     */
    private static Connection getDBConnectionUsingCAM(
            String secretId, String secretKey, String region, String instanceId, String userName,
            String host, int port, String dbName) throws Exception {

        // 从 secretId 和 secretKey 获取凭证
        Credential credential = new Credential(secretId, secretKey);

        // 定义最大尝试次数
        int maxAttempts = 3;
        Exception lastException = null;
        for (int attempt = 1; attempt <= maxAttempts; attempt++) {
            try {
                // 使用凭证获取认证令牌
                String authToken = getAuthToken(region, instanceId, userName, credential);

                String connectionUrl = String.format("jdbc:postgresql://%s:%d/%s", host, port, dbName);
                return DriverManager.getConnection(connectionUrl, userName, authToken);
            } catch (Exception e) {
                lastException = e;
                System.out.println("Attempt " + attempt + " failed.");
                Thread.sleep(5000);
            }
        }
        System.out.println("All attempts failed. error: " + lastException.getMessage());
        throw lastException;
    }

    /**
     * 获取认证令牌
     *
     * @param region     区域
     * @param instanceId 实例 ID
     * @param userName   用户名
     * @param credential 凭证
     * @return 认证令牌
     */
    private static String getAuthToken(String region, String instanceId, String userName, Credential credential) throws TencentCloudSDKException {
        // 实例化一个http选项，可选的，没有特殊需求可以跳过
        HttpProfile httpProfile = new HttpProfile();
        httpProfile.setEndpoint("cam.tencentcloudapi.com");
        // 实例化一个client选项，可选的，没有特殊需求可以跳过
        ClientProfile clientProfile = new ClientProfile();
        clientProfile.setHttpProfile(httpProfile);

        // 构建 GenerateAuthenticationTokenRequest
        GenerateAuthenticationTokenRequest tokenRequest = GenerateAuthenticationTokenRequest.builder()
                .region(region)
                .credential(credential)
                .userName(userName)
                .instanceId(instanceId)
                .clientProfile(clientProfile) // clientProfile是可选的
                .build();

        return DBAuthentication.generateAuthenticationToken(tokenRequest);
    }
}

pip install git+https://github.com/TencentCloud/dbauth-sdk-python.git

import logging
import os
import time

import psycopg2
from psycopg2 import extras
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" 
    # 实例id
    instance_id = "postgres-6yol****"
    # 用户名
    user_name = "dbadmin"
    # 实例vip
    host = "1*.*.*.*"
    # 实例端口
    port = 5432
    # database名
    db_name = "am"
    # secret_id和secret_key
    secret_id = "AKIDI***************"
    secret_key = "bEMGY***************"

    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 VERSION()")       
            result = cursor.fetchone()
            log.info(result[0])  # 应该打印 "Success!"
    except Exception as e:
        log.error(f"An error occurred: {e}")
    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)
            log.info(auth_token) 

            connection = psycopg2.connect(
                host=host,
                port=port,
                user=user_name,
                password=auth_token,
                dbname=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()