1. 概要
生成 AI アプリケーションの構築を急ぐあまり、最も重要なコンポーネントである安全性を見落としがちです。
人事 chatbot を構築することを考えてみましょう。「私の給与はいくらですか？」や「チームのパフォーマンスはどうですか？」などの質問に回答できるようにします。
- Alice（一般社員）がリクエストした場合は、自分のデータのみが表示される必要があります。
- Bob（マネージャー）がリクエストした場合は、チームのデータが表示されます。
問題点
ほとんどの RAG（検索拡張生成）アーキテクチャは、アプリケーション レイヤでこの処理を行おうとします。チャンクを取得した後でフィルタするか、LLM の動作に依存します。これは壊れやすいものです。アプリのロジックが失敗すると、データが漏洩します。
ソリューション
セキュリティを Database Layer までプッシュします。AlloyDB で PostgreSQL の行レベル セキュリティ（RLS）を使用することで、AI が何を要求しても、ユーザーが閲覧できないデータをデータベースが物理的に拒否するようにします。
このガイドでは、ログインしているユーザーに基づいて回答を動的に変更する安全な HR アシスタント「The Private Vault」を構築します。
アーキテクチャ
Python で複雑な権限ロジックを構築していません。データベース エンジン自体を使用しています。
- インターフェース: ログインをシミュレートするシンプルな Streamlit アプリ。
- The Brain: AlloyDB AI（PostgreSQL 互換）。
- メカニズム: すべてのトランザクションの開始時にセッション変数（
app.active_user）を設定します。データベース ポリシーは、user_rolesテーブル（ID プロバイダとして機能）を自動的にチェックして行をフィルタします。
作成するアプリの概要
安全な HR アシスタント アプリケーション。アプリケーション ロジックに依存して機密データをフィルタするのではなく、AlloyDB データベース エンジンに直接行レベル セキュリティ（RLS）を実装します。これにより、AI モデルが「ハルシネーション」を起こしたり、不正なデータにアクセスしようとしたりしても、データベースが物理的にそのデータを返さないようにすることができます。
学習内容
学習内容:
- RLS のスキーマを設計する方法（データと ID の分離）。
- PostgreSQL ポリシー（
CREATE POLICY）の作成方法。 FORCE ROW LEVEL SECURITYを使用して「テーブル オーナー」の例外をバイパスする方法。- ユーザーの「コンテキスト切り替え」を行う Python アプリを構築する方法。
要件
2. 始める前に
プロジェクトを作成する
- Google Cloud コンソールのプロジェクト選択ページで、Google Cloud プロジェクトを選択または作成します。
- Cloud プロジェクトに対して課金が有効になっていることを確認します。詳しくは、プロジェクトで課金が有効になっているかどうかを確認する方法をご覧ください。
- Google Cloud 上で動作するコマンドライン環境の Cloud Shell を使用します。Google Cloud コンソールの上部にある [Cloud Shell をアクティブにする] をクリックします。
![[Cloud Shell をアクティブにする] ボタンの画像](https://codelabs.developers.google.com/static/zero-trust-agents-with-alloydb/img/91567e2f55467574.png?hl=ja)
- Cloud Shell に接続したら、次のコマンドを使用して、すでに認証済みであることと、プロジェクトがプロジェクト ID に設定されていることを確認します。
gcloud auth list
- Cloud Shell で次のコマンドを実行して、gcloud コマンドがプロジェクトを認識していることを確認します。
gcloud config list project
- プロジェクトが設定されていない場合は、次のコマンドを使用して設定します。
gcloud config set project <YOUR_PROJECT_ID>
- 必要な API を有効にする: リンクにアクセスして、API を有効にします。
または、この操作に gcloud コマンドを使用することもできます。gcloud コマンドとその使用方法については、ドキュメントをご覧ください。
gcloud services enable \
alloydb.googleapis.com \
compute.googleapis.com \
cloudresourcemanager.googleapis.com \
servicenetworking.googleapis.com \
aiplatform.googleapis.com
注意点とトラブルシューティング
「ゴースト プロジェクト」症候群 |
|
請求のバリケード | プロジェクトを有効にしたが、請求先アカウントを忘れた。AlloyDB は高性能エンジンです。ガソリン タンク（課金）が空の場合、起動しません。 |
API 伝播の遅延 | [API を有効にする] をクリックしたのに、コマンドラインに |
割り当て Quags | 新しいトライアル アカウントを使用している場合は、AlloyDB インスタンスのリージョン割り当てに達する可能性があります。 |
3. データベースの設定
このラボでは、テストデータのデータベースとして AlloyDB を使用します。クラスタを使用して、データベースやログなどのすべてのリソースを保持します。各クラスタには、データへのアクセス ポイントを提供するプライマリ インスタンスがあります。テーブルには実際のデータが格納されます。
テスト データセットを読み込む AlloyDB クラスタ、インスタンス、テーブルを作成しましょう。
- ボタンをクリックするか、以下のリンクを Google Cloud コンソールのユーザーがログインしているブラウザにコピーします。
- この手順が完了すると、リポジトリがローカルの Cloud Shell エディタにクローンされ、プロジェクト フォルダから次のコマンドを実行できるようになります（プロジェクト ディレクトリにいることを確認することが重要です）。
sh run.sh
- UI を使用します（ターミナルのリンクをクリックするか、ターミナルの [ウェブでプレビュー] リンクをクリックします）。
- プロジェクト ID、クラスタ名、インスタンス名の詳細を入力して、開始します。
- ログがスクロールしている間にコーヒーを飲んで、裏側で何が行われているかについてはこちらをご覧ください。10 ～ 15 分ほどかかることがあります。
注意点とトラブルシューティング
「忍耐」の問題 | データベース クラスタは重いインフラストラクチャです。ページを更新したり、「フリーズした」ように見える Cloud Shell セッションを強制終了したりすると、部分的にプロビジョニングされた「ゴースト」インスタンスが作成され、手動で介入しないと削除できなくなる可能性があります。 |
リージョンが一致しない |
|
ゾンビ クラスタ | 以前にクラスタに同じ名前を使用し、削除していない場合、スクリプトでクラスタ名がすでに存在すると表示されることがあります。クラスタ名はプロジェクト内で一意にする必要があります。 |
Cloud Shell のタイムアウト | コーヒー ブレイクに 30 分かかると、Cloud Shell がスリープ状態になり、 |
4. スキーマのプロビジョニング
このステップでは、次の内容について説明します。
詳しい手順は次のとおりです。
AlloyDB クラスタとインスタンスが実行されたら、AlloyDB Studio の SQL エディタに移動して、AI 拡張機能を有効にしてスキーマをプロビジョニングします。
インスタンスの作成が完了するまで待つ必要がある場合があります。完了したら、クラスタの作成時に作成した認証情報を使用して AlloyDB にログインします。PostgreSQL の認証には次のデータを使用します。
- ユーザー名: 「
postgres」 - データベース: 「
postgres」 - パスワード: 「
alloydb」（作成時に設定したパスワード）
AlloyDB Studio への認証が成功すると、エディタに SQL コマンドが入力されます。最後のウィンドウの右にあるプラス記号を使用して、複数のエディタ ウィンドウを追加できます。
必要に応じて [実行]、[形式]、[クリア] オプションを使用して、エディタ ウィンドウに AlloyDB のコマンドを入力します。
テーブルを作成する
機密データ（従業員）用のテーブルと、ID ルール（user_roles）用のテーブルの 2 つが必要です。これらを分離することは、ポリシーで「無限再帰」エラーを回避するために不可欠です。
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 はテーブル オーナーまたはスーパーユーザーには適用されません。すべてのポリシーをバイパスします。トラブルシューティング: ポリシーが「壊れている」（すべてを許可している）ように見える場合は、 |
6. アクセス ポリシーを作成する
ここでは、後でアプリケーション コードから設定するセッション変数（app.active_user）を使用して 2 つのルールを定義します。
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'
)
);
注意点とトラブルシューティング
app.active_user の代わりに current_user を使用します。 | 問題: Current_user は、データベース ロール（app_user など）を返す予約済みの SQL キーワードです。アプリケーション ユーザー（Alice）。修正: 常に app.variable_name などのカスタム名前空間を使用します。 |
| 問題: 変数が設定されていない場合、クエリがエラーでクラッシュします。修正: current_setting('...', true) はクラッシュする代わりに NULL を返し、安全に 0 行が返されます。 |
7. 「カメレオン」アプリをビルドする
Python と Streamlit を使用して、アプリケーション ロジックをシミュレートします。
エディタ モードで Cloud Shell ターミナルを開き、ルートフォルダまたはこのアプリケーションを作成するディレクトリに移動します。新しいフォルダを作成する。
1. 依存関係をインストールする:
新しいプロジェクト ディレクトリ内から、Cloud Shell ターミナルで次のコマンドを実行します。
pip install streamlit psycopg2-binary
2. app.py を作成します。
app.py という名前の新しいファイルを作成し、リポジトリ ファイルからコンテンツをコピーします。
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 の [ウェブでプレビュー] ボタンを使用します。ターミナルで localhost リンクをクリックしないでください。 |
8. ゼロトラストを検証する
アプリを試して、ゼロトラストの実装を確認します。
「Alice」 を選択: 1 行（自分自身）が表示されます。
「Bob」を選択します。3 行（全員）が表示されます。
AI エージェントにとって重要な理由
このデータベースにモデルを接続することを考えてみましょう。ユーザーがモデルに「すべてのパフォーマンス レビューを要約して」と尋ねると、SELECT performance_review FROM employees が生成されます。
- RLS なし: モデルがすべてのユーザーの非公開レビューを取得し、Alice に漏洩します。
- RLS を使用した場合: モデルはまったく同じクエリを実行しますが、データベースは Alice のレビューのみを返します。
これがゼロトラスト AI です。モデルがデータをフィルタリングすることを信頼せず、データベースに非表示を強制します。
本番環境への移行
ここで説明するアーキテクチャは本番環境グレードですが、具体的な実装は学習用に簡略化されています。実際のエンタープライズ環境に安全にデプロイするには、次の拡張機能を実装する必要があります。
- 実際の認証: [Identity Switcher] プルダウンを、Google Identity Platform、Okta、Auth0 などの堅牢な ID プロバイダ（IDP）に置き換えます。アプリケーションは、データベース セッション変数を設定する前に、ユーザーのトークンを検証して ID を安全に抽出する必要があります。これにより、ユーザーが ID をスプーフィングできないようにします。
- 接続プーリングの安全性: 接続プールを使用する場合、正しく処理しないと、セッション変数が異なるユーザー リクエスト間で保持されることがあります。アプリケーションがセッション変数（RESET app.active_user）または、接続をプールに戻すときに接続状態をクリアして、ユーザー間のデータ漏洩を防ぎます。
- シークレット管理: データベース認証情報をハードコードすると、セキュリティ リスクが発生します。Google Secret Manager などの専用のシークレット管理サービスを使用して、データベースのパスワードと接続文字列を安全に保存し、実行時に取得します。
9. クリーンアップ
このラボが完了したら、必ず AlloyDB クラスタとインスタンスを削除してください。
クラスタとそのインスタンスをクリーンアップする必要があります。
10. 完了
これで完了です。セキュリティがデータレイヤに正常にプッシュされました。Python コードに print(all_salaries) を試みるバグがあったとしても、データベースは Alice に何も返しません。
次のステップ
- ご自身のデータセットで試してみてください。
- AlloyDB AI のドキュメントをご覧ください。
- 他のワークショップについては、Code Vipassana のウェブサイトをご覧ください。