如何在IAM认证的Cloud SQL pg8000+SQLAlchemy连接中强制SSL模式

当你使用Cloud SQL Python Connector搭配pg8000和IAM认证时，要强制启用SSL（sslmode=require），可以通过以下两种方式实现：

方案一：通过Connector传递SSL上下文（推荐） #

Cloud SQL Connector可以自动管理连接Cloud SQL所需的SSL证书，你可以显式创建SSL上下文并强制验证要求，确保连接必须加密：

import ssl
from google.cloud.sql.connector import Connector, IPTypes
import sqlalchemy

def test_iam_connection():
    connector = Connector()

    def getconn():
        # 创建强制SSL验证的上下文
        ssl_context = ssl.create_default_context()
        ssl_context.check_hostname = True
        ssl_context.verify_mode = ssl.CERT_REQUIRED

        conn = connector.connect(
            INSTANCE_CONNECTION_NAME,
            "pg8000",
            user=IAM_USER,
            db=DB_NAME,
            enable_iam_auth=True,
            ip_type=IPTypes.PRIVATE,
            ssl_context=ssl_context  # 传递SSL上下文
        )
        return conn

    pool = sqlalchemy.create_engine(
        "postgresql+pg8000://",
        creator=getconn
    )

    with pool.connect() as db_conn:
        # 验证SSL连接状态
        ssl_status = db_conn.execute(sqlalchemy.text("SELECT ssl FROM pg_stat_ssl WHERE pid = pg_backend_pid()")).fetchone()
        print("SSL连接已启用: ", ssl_status[0])
        
        # 原查询逻辑
        results = db_conn.execute(sqlalchemy.text("SELECT NOW()")).fetchone()
        results2 = db_conn.execute(sqlalchemy.text("SELECT * from public.test")).fetchone()
        print("Current time: ", results[0])
        print("results: ", results2[0])

    connector.close()

if __name__ == "__main__":
    test_iam_connection()

方案二：直接传递pg8000的sslmode参数 #

Cloud SQL Connector的connect方法会将额外参数透传给pg8000的连接函数，因此可以直接添加sslmode="require"参数：

from google.cloud.sql.connector import Connector, IPTypes
import sqlalchemy

def test_iam_connection():
    connector = Connector()

    def getconn():
        conn = connector.connect(
            INSTANCE_CONNECTION_NAME,
            "pg8000",
            user=IAM_USER,
            db=DB_NAME,
            enable_iam_auth=True,
            ip_type=IPTypes.PRIVATE,
            sslmode="require"  # 直接指定sslmode
        )
        return conn

    pool = sqlalchemy.create_engine(
        "postgresql+pg8000://",
        creator=getconn
    )

    with pool.connect() as db_conn:
        # 验证SSL连接状态
        ssl_status = db_conn.execute(sqlalchemy.text("SELECT ssl FROM pg_stat_ssl WHERE pid = pg_backend_pid()")).fetchone()
        print("SSL连接已启用: ", ssl_status[0])
        
        # 原查询逻辑
        results = db_conn.execute(sqlalchemy.text("SELECT NOW()")).fetchone()
        results2 = db_conn.execute(sqlalchemy.text("SELECT * from public.test")).fetchone()
        print("Current time: ", results[0])
        print("results: ", results2[0])

    connector.close()

if __name__ == "__main__":
    test_iam_connection()

关键说明 #

• 为什么之前用SQLAlchemy的connect_args无效？因为你使用了creator参数，此时SQLAlchemy会直接调用你提供的getconn函数获取连接，不会处理connect_args里的参数，必须将SSL相关配置直接传递给Connector的connect方法。
• 验证SSL是否生效：通过查询pg_stat_ssl视图可以确认当前连接的SSL状态，返回True则表示连接已加密。

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