私密保险库：利用 AlloyDB 行级安全性构建“零信任智能”

1. 概览

在匆忙构建生成式 AI 应用时，我们往往会忘记最关键的组件：安全。

假设您要构建一个人力资源聊天机器人。您希望它能回答“我的薪资是多少？”或“我们团队的业绩如何？”之类的问题。

• 如果 Alice（一名普通员工）提出请求，她应该只能看到自己的数据。
• 如果Bob（一位经理）提出要求，他应该能看到自己团队的数据。

问题

大多数 RAG（检索增强生成）架构都尝试在应用层中处理此问题。它们在检索到块后对其进行过滤，或者依赖 LLM 来“正常运行”。这很脆弱。如果应用逻辑失败，则会发生数据泄露。

解决方案

将安全性下推到数据库层。通过在 AlloyDB 中使用 PostgreSQL 行级安全性 (RLS)，我们确保数据库会拒绝返回用户无权查看的数据，无论 AI 请求什么数据。

在本指南中，我们将构建“私密保险库”：一个安全的人力资源助理，可根据登录的用户动态更改其回答。

架构

我们不会在 Python 中构建复杂的权限逻辑。我们使用的是数据库引擎本身。

1. 界面：一个模拟登录的简单 Streamlit 应用。
2. 大脑：AlloyDB AI（与 PostgreSQL 兼容）。
3. 机制：我们在每次交易开始时设置一个会话变量 (app.active_user)。数据库政策会自动检查 user_roles 表（充当我们的身份提供方）以过滤行。

构建内容

一款安全的人力资源助理应用。您将直接在 AlloyDB 数据库引擎中实现行级安全性 (RLS)，而不是依赖于应用逻辑来过滤敏感数据。这样一来，即使 AI 模型“产生幻觉”或尝试访问未经授权的数据，数据库也会拒绝返回相应数据。

学习内容

您会了解到以下内容：

• 如何为 RLS 设计架构（分离数据与身份）。
• 如何编写 PostgreSQL 政策 (CREATE POLICY)。
• 如何使用 FORCE ROW LEVEL SECURITY 绕过“表所有者”豁免。
• 如何构建可为用户执行“上下文切换”的 Python 应用。

要求

• 一个浏览器，例如 Chrome 或 Firefox。
• 启用了结算功能的 Google Cloud 项目。
• 对 Cloud Shell 或安装了 gcloud 和 psql 的终端的访问权限。

2. 准备工作

创建项目

1. 在 Google Cloud Console 的项目选择器页面上，选择或创建一个 Google Cloud 项目。
2. 确保您的 Cloud 项目已启用结算功能。了解如何检查项目是否已启用结算功能。

如需试用此 Codelab，您可以使用 GCP 试用结算赠金链接：https://bit.ly/ez-alloydb-5

如何使用此链接兑换积分？推荐。

3. 您将使用 Cloud Shell，它是在 Google Cloud 中运行的命令行环境。点击 Google Cloud 控制台顶部的“激活 Cloud Shell”。

4. 连接到 Cloud Shell 后，您可以使用以下命令检查自己是否已通过身份验证，以及项目是否已设置为您的项目 ID：

gcloud auth list

5. 在 Cloud Shell 中运行以下命令，以确认 gcloud 命令了解您的项目。

gcloud config list project

6. 如果项目未设置，请使用以下命令进行设置：

gcloud config set project <YOUR_PROJECT_ID>

7. 启用必需的 API：点击链接并启用 API。

或者，您也可以使用 gcloud 命令来完成此操作。如需了解 gcloud 命令和用法，请参阅文档。

gcloud services enable \
  alloydb.googleapis.com \
  compute.googleapis.com \
  cloudresourcemanager.googleapis.com \
  servicenetworking.googleapis.com \
  aiplatform.googleapis.com

注意事项和问题排查

“幽灵项目” 综合征	您运行了 gcloud config set project，但实际上在控制台界面中查看的是另一个项目。检查左上角下拉菜单中的项目 ID！
结算 路障	您已启用项目，但忘记了结算账号。AlloyDB 是一款高性能引擎；如果“油箱”（结算）为空，它将无法启动。
API 传播 延迟	您点击了“启用 API”，但命令行仍显示 Service Not Enabled。等待 60 秒。云需要一些时间来唤醒其神经元。
配额 Quags	如果您使用的是全新试用账号，则可能会达到 AlloyDB 实例的区域配额。如果 us-central1 失败，则尝试 us-east1。

3. 数据库设置

在本实验中，我们将使用 AlloyDB 作为测试数据的数据库。它使用集群来保存所有资源，例如数据库和日志。每个集群都有一个主实例，可提供对数据的访问点。表将包含实际数据。

我们来创建 AlloyDB 集群、实例和表，以便加载测试数据集。

1. 点击相应按钮，或将下方的链接复制到已登录 Google Cloud 控制台用户的浏览器中。

一键运行

2. 完成此步骤后，代码库将克隆到本地 Cloud Shell 编辑器，您将能够从项目文件夹中运行以下命令（请务必确保您位于项目目录中）：

sh run.sh

3. 现在，使用界面（点击终端中的链接或点击终端中的“在网页上预览”链接）。
4. 输入项目 ID、集群名称和实例名称等详细信息，即可开始使用。
5. 在日志滚动时，您可以去喝杯咖啡，然后点击此处了解该功能在后台的运作方式。此过程可能需要大约 10-15 分钟。

重要提示：如需在本地或从任何位置进行测试，请前往 AlloyDB 实例，点击“修改”，然后点击“启用公共 IP”或“公共 IP 连接”（不是出站公共 IP），并在“授权的外部网络”中输入“0.0.0.0/0”（仅限开发用途），完成后请移除并停用公共 IP 连接。

注意事项和问题排查

“耐心”问题	数据库集群是重型基础架构。如果您因 Cloud Shell 会话“看起来卡住了”而刷新页面或终止会话，最终可能会得到一个“幽灵”实例，该实例已部分完成预配，但无法在不进行人工干预的情况下删除。
地区不匹配	如果您在 us-central1 中启用了 API，但尝试在 asia-south1 中预配集群，则可能会遇到配额问题或服务账号权限延迟。在整个实验过程中，请坚持使用一个区域！
僵尸集群	如果您之前曾使用过某个集群名称，但未删除该集群，脚本可能会提示该集群名称已存在。集群名称在项目中必须是唯一的。
Cloud Shell 超时	如果您的咖啡休息时间为 30 分钟，Cloud Shell 可能会进入休眠状态并断开 sh run.sh 进程。让标签页保持活跃状态！

4. 架构配置

在此步骤中，我们将介绍以下内容：

以下是详细的分步操作：

在 AlloyDB 集群和实例运行后，前往 AlloyDB Studio SQL 编辑器，启用 AI 扩展程序并预配架构。

您可能需要等待实例完成创建。完成后，使用您在创建集群时创建的凭据登录 AlloyDB。使用以下数据向 PostgreSQL 进行身份验证：

• 用户名：“postgres”
• 数据库：“postgres”
• 密码：“alloydb”（或您在创建时设置的任何密码）

成功通过身份验证进入 AlloyDB Studio 后，您可以在编辑器中输入 SQL 命令。您可以使用最后一个窗口右侧的加号添加多个编辑器窗口。

您将在编辑器窗口中输入 AlloyDB 命令，并根据需要使用“运行”“格式”和“清除”选项。

创建表

我们需要两个表：一个用于存储敏感数据（员工），另一个用于存储身份规则 (user_roles)。将它们分开对于避免政策中的“无限递归”错误至关重要。

您可以在 AlloyDB Studio 中使用以下 DDL 语句创建表：

-- 1. Create User Roles (The Identity Provider)
CREATE TABLE user_roles (
    username TEXT PRIMARY KEY,
    role TEXT -- 'employee', 'manager', 'admin'
);

INSERT INTO user_roles (username, role) VALUES
('Alice', 'employee'),
('Bob', 'manager'),
('Charlie', 'employee');

-- 2. Create the Data Table
CREATE TABLE employees (
    id SERIAL PRIMARY KEY,
    name TEXT,
    salary INTEGER,
    performance_review TEXT
);

INSERT INTO employees (name, salary, performance_review) VALUES
('Alice', 80000, 'Alice meets expectations but needs to improve punctuality.'),
('Bob', 120000, 'Bob is a strong leader. Team morale is high.'),
('Charlie', 85000, 'Charlie exceeds expectations. Ready for promotion.');

注意事项和问题排查

在员工表中定义角色时检测到无限递归

失败原因：如果您的政策规定“检查员工表以查看我是否是经理”，数据库必须查询该表才能检查该政策，这会再次触发该政策。结果：检测到无限递归。修复：始终保留单独的查找表 (user_roles)，或使用实际的数据库用户来表示角色。

验证数据：

SELECT count(*) FROM employees;
-- Output: 3

5. 启用和强制执行安全设置

现在，我们开启护盾。我们还将创建一个通用“应用用户”，供我们的 Python 代码用于连接。

在 AlloyDB 查询编辑器中运行以下 SQL 语句：

-- 1. Activate RLS
ALTER TABLE employees ENABLE ROW LEVEL SECURITY;


-- 2. CRITICAL: Force RLS for Table Owners
ALTER TABLE employees FORCE ROW LEVEL SECURITY;


-- 3. Create the Application User
DO
$do$
BEGIN
   IF NOT EXISTS (SELECT FROM pg_catalog.pg_roles WHERE rolname = 'app_user') THEN
      CREATE ROLE app_user LOGIN PASSWORD 'password';
   END IF;
END
$do$;


-- 4. Grant Access
GRANT SELECT ON employees TO app_user;
GRANT SELECT ON user_roles TO app_user;

注意事项和问题排查

以 postgres（超级用户）身份进行测试，并查看所有数据。

失败原因： 默认情况下，RLS 不适用于表所有者或超级用户。它们会绕过所有政策。问题排查：如果您的政策似乎“失效”（允许所有操作），请检查您是否以 postgres 身份登录。修复：FORCE ROW LEVEL SECURITY 命令可确保即使是所有者也要遵守规则。这对于测试至关重要。

6. 创建访问权限政策

我们将使用会话变量 (app.active_user) 定义两条规则，稍后将通过应用代码设置该变量。

在 AlloyDB 查询编辑器中运行以下 SQL 语句：

-- Policy 1: Self-View
-- Users can see rows where their name matches the session variable
CREATE POLICY "view_own_data" ON employees
FOR SELECT
USING (name = current_setting('app.active_user', true));

-- Policy 2: Manager-View
-- Managers can see ALL rows.
CREATE POLICY "manager_view_all" ON employees
FOR SELECT
USING (
    EXISTS (
        SELECT 1 FROM user_roles
        WHERE username = current_setting('app.active_user', true)
        AND role = 'manager'
    )
);

注意事项和问题排查

使用 current_user 而不是 app.active_user。	问题： current_user 是一个保留的 SQL 关键字，用于返回数据库角色（例如 app_user）。我们需要应用用户（例如，Alice）。修复：始终使用自定义命名空间，例如 app.variable_name。
忘记在 current_setting 中添加 true 参数。	问题：如果未设置变量，查询会因错误而崩溃。修复：current_setting('...', true) 会返回 NULL 而不是崩溃，从而安全地返回 0 行。

7. 构建“变色龙”应用

我们将使用 Python 和 Streamlit 来模拟应用逻辑。

在编辑器模式下打开 Cloud Shell 终端，然后前往根文件夹或要在其中创建此应用的目录。创建新文件夹。

1. 安装依赖项：

在 Cloud Shell 终端中从新项目目录中运行以下命令：

pip install streamlit psycopg2-binary

2. 创建 app.py：

新建一个名为 app.py 的文件，然后复制 repo 文件中的内容。

import streamlit as st
import psycopg2

# CONFIGURATION (Replace with your IP)
DB_HOST = "10.x.x.x"
DB_NAME = "postgres"
DB_USER = "postgres"
DB_PASS = "alloydb"

def get_db_connection():
    return psycopg2.connect(
        host=DB_HOST, database=DB_NAME, user=DB_USER, password=DB_PASS
    )

def query_database(user_name):
    conn = get_db_connection()
    try:
        with conn.cursor() as cur:
            # THE SECURITY HANDSHAKE
            # We tell the database: "For this transaction, I am acting as..."
            cur.execute(f"SET app.active_user = '{user_name}';")
            
            # THE BLIND QUERY
            # We ask for EVERYTHING. The database silently filters it.
            cur.execute("SELECT name, role, salary, performance_review FROM employees;")
            return cur.fetchall()
    finally:
        conn.close()

# UI
st.title("🛡️ The Private Vault")
user = st.sidebar.radio("Act as User:", ["Alice", "Bob", "Charlie", "Eve"])

if st.button("Access Data"):
    results = query_database(user)
    if not results:
        st.error("🚫 Access Denied.")
    else:
        st.success(f"Viewing data as {user}")
        for row in results:
            st.write(row)

3. 运行应用：

在 Cloud Shell 终端中从新项目目录中运行以下命令：

streamlit run app.py --server.port 8080 --server.enableCORS false

注意事项和问题排查

连接池。	风险：如果您使用连接池，会话变量 SET app.active_user 可能会保留在连接中，并“泄露”给获取该连接的下一位用户。修复：在生产环境中，将连接返回到池时，请务必使用 RESET app.active_user 或 DISCARD ALL。
Cloud Shell 中的空白屏幕。	修复： 使用端口 8080 上的“网页预览”按钮。请勿点击终端中的本地主机链接。

8. 验证零信任

尝试使用该应用，确保零信任实现：

选择 Alice：她应该会看到 1 行（她自己）。

选择 “Bob”：他应该会看到 3 行（所有人）。

这对 AI 代理有何重要意义

不妨想象一下将模型连接到此数据库。如果用户向模型提出“总结所有绩效考核”这一问题，模型将生成 SELECT performance_review FROM employees。

1. 不使用 RLS：模型会检索所有人的私密评价，并将其泄露给 Alice。
2. 使用 RLS：您的模型运行的查询完全相同，但数据库仅返回 Alice 的评价。

这就是零信任 AI。您不信任模型来过滤数据，而是强制数据库隐藏数据。

将此功能投入生产

此处展示的架构是生产级架构，但为了便于学习，我们对具体实现进行了简化。如需在实际的企业环境中安全部署此功能，您应实现以下增强功能：

1. 真实身份验证：使用 Google Identity Platform、Okta 或 Auth0 等强大的身份提供方 (IDP) 替换“身份切换器”下拉菜单。您的应用应验证用户的令牌并安全地提取其身份，然后再设置数据库会话变量，以确保用户无法伪造自己的身份。
2. 连接池安全性：使用连接池时，如果处理不当，会话变量有时会在不同的用户请求之间保持不变。确保您的应用重置会话变量（例如，RESET app.active_user) 或在将连接返回到池时清除连接状态，以防止用户之间发生数据泄露。
3. Secret 管理：对数据库凭据进行硬编码会带来安全风险。使用 Google Secret Manager 等专用密钥管理服务在运行时安全地存储和检索数据库密码和连接字符串。

9. 清理

完成本实验后，请务必删除 AlloyDB 集群和实例。

前往 https://console.cloud.google.com/alloydb/clusters。点击要删除的集群旁边的竖向省略号，然后点击“删除”，即可选择要删除的集群。

它应清理集群及其实例。

10. 恭喜

恭喜！您已成功将安全性下推到数据层。即使您的 Python 代码存在尝试 print(all_salaries) 的 bug，数据库也不会向 Alice 返回任何内容。

后续步骤

• 不妨使用您自己的数据集试一下。
• 探索 AlloyDB AI 文档。
• 如需了解更多工作坊，请访问 Code Vipassana 网站。