使用 Tink 与 Cloud KMS 在 Cloud SQL for SQL Server 中实现客户端字段级加密（Envelope AEAD 实战）

示例工程【免费下载链接】python-docs-samplesCode samples used on cloud.google.com项目地址https://gitcode.com/GitHub_Trending/py/python-docs-samples点击查看免费下载导读本指南基于当前仓库 cloud-sql/sql-server/client-side-encryption 目录下的完整示例讲解如何在 Cloud SQL for SQL Server 中借助 Tink 的 Envelope AEAD 原语与 Cloud KMS 密钥在应用侧完成敏感字段如用户邮箱的加密写入与解密读取。读完本文你将掌握从创建 KMS 密钥、配置环境变量、通过 Cloud SQL Auth Proxy 建连到运行 encrypt_and_insert_data.py 与 query_and_decrypt_data.py 的完整链路并理解 Tink 信封加密的底层原理与测试验证方式。说明该目录的 README.md 标题仍沿用早期文案Encrypting fields in Cloud SQL - MySQL with Tink但目录实际面向 SQL Server 场景本文章节均以仓库内的 SQL Server 实现为准。核心思路客户端加密解决什么问题传统的数据加密通常停留在传输层TLS与存储层Cloud SQL 自带加密数据库管理员、备份文件读取者仍可能看到明文敏感字段。客户端加密Client-Side Encryption, CSE把加密动作前移到应用进程内明文只在应用内存中短暂存在落库的始终是密文。本示例以“投票应用”为场景将voter_email字段加密后存入 SQL Server 的varbinary(max)列而team与time_cast保持明文以便查询与排序。该方案不依赖数据库内置加密功能因此密文对 DBA、备份与迁移过程不可见密钥由 Cloud KMS 托管可审计、可轮换使用 Tink 的标准 AEAD 原语避免手写密码学实现。一、开始前的准备1. 环境与项目按官方指引配置 Python 开发环境并创建 Google Cloud 项目。若尚未安装依赖参考仓库的 requirements.txtpip install -r requirements.txt该文件锁定的核心依赖如下依赖版本作用SQLAlchemy2.0.40数据库 ORM 与连接池python-tds1.16.0SQL Server 的 TDS 协议客户端sqlalchemy-pytds1.0.2SQLAlchemy 的mssqlpytds方言tink1.9.0加密原语库AEAD / KMS 集成2. 创建 Cloud SQL for SQL Server 实例与数据库创建一个 2nd Gen Cloud SQL for SQL Server 实例记录连接字符串、数据库用户名和密码为应用创建数据库记录数据库名创建具备 Cloud SQL Client 权限的服务账号并下载JSON 密钥文件用于后续认证。3. 创建 Cloud KMS 密钥按 KMS 官方指引创建密钥并复制其资源名Resource Name。资源名形如projects/project-id/locations/location/keyRings/key-ring/cryptoKeys/key在代码中该资源名需要加上gcp-kms://前缀构成 Tink 使用的密钥 URI。源码 encrypt_and_insert_data.py 中有明确注释key_uri gcp-kms:// os.environ[GCP_KMS_URI] # e.g. gcp-kms://projects/...path/to/key # Tink uses the gcp-kms:// prefix for paths to keys stored in Google Cloud KMS4. macOS / Windows 专用配置 gRPC Root Certificates部分平台上Tink 访问 Google KMS 时需要通过 gRPC 校验 Google 服务器证书。若遇到证书校验失败需按指引配置根证书这一点在 README.md 的 macOS / Windows only 步骤中有明确说明。二、运行环境变量TCP 与 Unix Socket 两种方式Linux / Mac OSbashexport GOOGLE_APPLICATION_CREDENTIALS/path/to/service/account/key.json export DB_HOST127.0.0.1:1433 export DB_USERDB_USER_NAME export DB_PASSDB_PASSWORD export DB_NAMEDB_NAME export GCP_KMS_URIGCP_KMS_URIWindows / PowerShell$env:GOOGLE_APPLICATION_CREDENTIALSCREDENTIALS_JSON_FILE $env:DB_HOST127.0.0.1:1433 $env:DB_USERDB_USER_NAME $env:DB_PASSDB_PASSWORD $env:DB_NAMEDB_NAME $env:GCP_KMS_URIGCP_KMS_URI安全提醒README 原文亦强调把凭据写进环境变量虽然方便但不够安全建议改用 Secret Manager 等方案托管密钥。除上述变量外源码还支持两个可选变量见 encrypt_and_insert_data.py变量默认值说明DB_SOCKET_DIR/cloudsql使用 Unix Socket 时的目录INSTANCE_CONNECTION_NAME无必填形如project-name:region:instance-name用于定位实例若通过INSTANCE_CONNECTION_NAME连接还需在代码路径中传递db_socket_dir与instance_connection_name给init_db。三、通过 Cloud SQL Auth Proxy 建立连接在本地运行时先下载并安装cloud_sql_proxy。README 提供了 TCP 与 Unix Domain Socket 两种方式Linux / macOS 二者皆可Windows 目前仅支持 TCP。TCP 方式Linux / macOS 后台启动./cloud_sql_proxy -instancesproject-id:region:instance-nametcp:1433 -credential_file$GOOGLE_APPLICATION_CREDENTIALS PowerShell 独立会话启动Start-Process -filepath C:\path to proxy exe -ArgumentList -instancesproject-id:region:instance-nametcp:1433 -credential_fileCREDENTIALS_JSON_FILE启动后DB_HOST127.0.0.1:1433即指向代理的本地端口应用无需直接暴露实例公网地址。依赖安装与虚拟环境virtualenv --python python3 env source env/bin/activate pip install -r requirements.txt运行演示python snippets/query_and_decrypt_data.py该脚本会依次完成「加密插入一条投票记录」与「解密查询最近 5 条投票」并打印类似输出Team Email Time Cast TABS helloexample.com 2026-10-02 01:54:1900:00四、源码拆解加密写入链路1. 初始化 Envelope AEAD 原语核心逻辑在 cloud_kms_env_aead.pydef init_tink_env_aead(key_uri: str, credentials: str) - tink.aead.KmsEnvelopeAead: aead.register() gcp_client gcpkms.GcpKmsClient(key_uri, credentials) gcp_aead gcp_client.get_aead(key_uri) # Create envelope AEAD primitive using AES256 GCM for encrypting the data key_template aead.aead_key_templates.AES256_GCM env_aead aead.KmsEnvelopeAead(key_template, gcp_aead) print(fCreated envelope AEAD Primitive using KMS URI: {key_uri}) return env_aead关键点GcpKmsClient通过GOOGLE_APPLICATION_CREDENTIALS或显式 credentials 参数完成对 KMS 的认证Envelope AEAD信封加密的原理是为每一条数据生成一个随机的数据加密密钥DEK用 Cloud KMS 中的密钥加密密钥KEK去加密 DEK形成「信封」数据本身用 DEK 按AES256_GCM加密。这样既避免了逐条调用 KMS性能友好又保证了每条的密钥独立性数据加密密钥永不落盘明文明文 DEK 只存在于内存中由 KMS 解密信封获得。2. 加密并插入数据核心逻辑在 encrypt_and_insert_data.pytime_cast datetime.datetime.now(tzdatetime.timezone.utc) # Use the envelope AEAD primitive to encrypt the email, using the team name as # associated data. Encryption with associated data ensures authenticity # (who the sender is) and integrity (the data has not been tampered with) of that # data, but not its secrecy. (see RFC 5116 for more info) encrypted_email env_aead.encrypt(email.encode(), team.encode()) # Verify that the team is one of the allowed options if team ! TABS and team ! SPACES: logger.error(fInvalid team specified: {team}) return stmt sqlalchemy.text( fINSERT INTO {table_name} (time_cast, team, voter_email) VALUES (:time_cast, :team, CONVERT(varbinary(max), :voter_email, 0)) ) with db.connect() as conn: conn.execute(stmt, time_casttime_cast, teamteam, voter_emailencrypted_email) print(fVote successfully cast for {team} at time {time_cast}!)值得学习的工程细节关联数据Associated Dataenv_aead.encrypt(email.encode(), team.encode())把team作为 AAD。解密时必须传入相同 AAD否则失败。这样可防止「把 A 行的密文搬到 B 行」的篡改攻击同时保证数据的真实性与完整性见 RFC 5116参数化 SQL使用sqlalchemy.text的绑定参数:time_cast等有效抵御 SQL 注入voter_email通过CONVERT(varbinary(max), ..., 0)显式转成二进制列类型连接生命周期with db.connect() as conn确保语句执行后连接一定归还连接池即使出错也不泄漏。3. 建表与连接池cloud_sql_connection_pool.py 承担连接池与建表职责def connect_with_pytds() - pytds.Connection: return pytds.connect( db_hostname, userdb_user, passworddb_pass, databasedb_name, portdb_port, bytes_to_unicodeFalse, # disables automatic decoding of bytes ) pool sqlalchemy.create_engine( # This allows us to use the pytds sqlalchemy dialect, but also set the # bytes_to_unicode flag to False, which is not supported by the dialect mssqlpytds://, creatorconnect_with_pytds, )两个关键点bytes_to_unicodeFalse默认驱动会把varbinary字节自动解码为字符串这会破坏密文必须关闭自动解码让密文以bytes原样读写mssqlpytds://加自定义creator由于该方言不支持直接设置bytes_to_unicode示例通过creator注入自定义的pytds.connect参数绕过方言限制。init_db还会在表不存在时自动建表列定义与加密需求严格对应Table( table_name, metadata, Column(vote_id, Integer, primary_keyTrue, nullableFalse), Column(voter_email, sqlalchemy.types.VARBINARY, nullableFalse), Column(time_cast, DateTime, nullableFalse), Column(team, sqlalchemy.types.VARCHAR(6), nullableFalse), )voter_email使用VARBINARY类型存放密文而team、time_cast保持明文便于ORDER BY time_cast等常规 SQL 操作。五、源码拆解解密查询链路query_and_decrypt_data.py 演示了「读取密文 → 解密」的完整流程def query_and_decrypt_data(db, env_aead, table_name) - None: with db.connect() as conn: recent_votes conn.execute( fSELECT TOP(5) team, time_cast, voter_email FROM {table_name} ORDER BY time_cast DESC ).fetchall() print(Team\tEmail\tTime Cast) for row in recent_votes: team row[0] # Use the envelope AEAD primitive to decrypt the email, using the team name as # associated data. email env_aead.decrypt(row[2], team).decode() time_cast row[1] print(f{team}\t{email}\t{time_cast})要点查询在数据库侧完全以密文形态进行SELECT TOP(5) ... ORDER BY time_cast DESC是普通 SQL不会触发任何解密逻辑解密必须携带与加密时相同的 AAD即该行的team值env_aead.decrypt(row[2], team)若 AAD 不匹配或数据被篡改会抛出异常从而暴露数据完整性问题.decode()将解密后的bytes还原为原始字符串例如邮箱地址。六、测试与验证仓库内的可信证据该目录包含完整的 pytest 测试可在具备真实云资源的环境下验证加密闭环1. KMS 原语初始化测试cloud_kms_env_aead_test.py 需要环境变量CLOUD_KMS_KEY含gcp-kms://前缀验证init_tink_env_aead能成功创建 Envelope AEAD 原语并断言输出包含Created envelope AEAD Primitive using KMS URI: ...。2. 加密插入测试encrypt_and_insert_data_test.py 的测试流程值得关注从SQLSERVER_USER、SQLSERVER_PASSWORD、SQLSERVER_DATABASE、SQLSERVER_HOST读取连接信息使用随机表名votes_uuid建表避免测试间互相污染调用encrypt_and_insert_data插入一条SPACES / helloexample.com记录直接查询库中密文并用env_aead.decrypt(row[2], team)解密断言helloexample.com可被还原——这从测试层面证明了「库中存密文、应用可解密」的闭环。3. 解密查询测试query_and_decrypt_data_test.py 在插入后调用query_and_decrypt_data断言输出包含表头Team\tEmail\tTime Cast与还原出的邮箱helloexample.com。测试用例中以gcp-kms://前缀拼接CLOUD_KMS_KEY与主程序保持一致见测试文件第 60 行。此外 noxfile_config.py 声明该样例在 Python 3.10 上运行测试其余版本被忽略并支持通过GOOGLE_CLOUD_PROJECT指定测试项目。七、实战要点与注意事项密文列类型必须是二进制SQL Server 侧使用varbinary(max)且连接必须关闭bytes_to_unicode否则密文会被驱动解码破坏AAD 的一致性加密与解密必须使用相同的关联数据本示例为team这是防篡改、防密文搬移的关键密钥前缀Cloud KMS 资源名必须加gcp-kms://前缀才能被 Tink 识别两者拼接时不要重复或遗漏凭据安全环境变量存放凭据仅适合本地演示生产环境应使用 Secret Manager 或工作负载身份联邦性能与 KMS 配额Envelope 加密模式下每条记录仅需一次 KMS 信封解密数据加密走本地 AES-256-GCM避免高频调用 KMS 产生配额瓶颈本地连接Windows 仅支持 TCP 代理Linux / macOS 可选用 Unix Socket运行前确保cloud_sql_proxy已指向tcp:1433。八、延伸阅读连接池与建表实现cloud_sql_connection_pool.py加密插入实现encrypt_and_insert_data.py解密查询实现query_and_decrypt_data.pyKMS Envelope AEAD 初始化cloud_kms_env_aead.py依赖清单requirements.txt原始操作指南README.md赞分享示例工程【免费下载链接】python-docs-samplesCode samples used on cloud.google.com项目地址https://gitcode.com/GitHub_Trending/py/python-docs-samples点击查看免费下载相关推荐Cloud SQL PostgreSQL 客户端字段加密实战基于 Tink 与 Cloud KMS 的信封加密方案Cloud SQL PostgreSQL 客户端字段加密实战基于 Tink 与 Cloud KMS 的信封加密方案 本指南以 cloud sql/postgr示例工程Tink Java 与 Google Cloud Storage基于 Cloud KMS 信封加密Envelope Encryption实现 GCS Blob 客户端加密实战Tink Java 与 Google Cloud Storage基于 Cloud KMS 信封加密Envelope Encryption实现 GCS Bl密码学应用安全Tink Java 信封加密Envelope Encryption实战使用 Cloud KMS 密钥包装 DEK 的 AEAD 加密方案Tink Java 信封加密Envelope Encryption实战使用 Cloud KMS 密钥包装 DEK 的 AEAD 加密方案 本篇技术指南围绕密码学应用安全上一篇JavaQuestPlayer一站式解决QSP游戏兼容性难题的三大核心功能下一篇Akagi麻将AI助手5分钟免费搭建你的智能麻将教练创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考