AI に本番 DB を安全に接続するには？ ― 読み取り専用ユーザーだけでは足りない理由

「先月の売上が落ちた理由を教えて」と聞くだけで、AI が社内データを調べて答えてくれる。これは便利です。ただ、実装を急いで本番 DB の接続文字列と run_sql のような機能を渡してしまうと、モデルが出した文字列を本番環境で実行することになります。かなり怖い構成です。

モデルの出力は、信頼済みの命令ではありません。ユーザーが入力した文章はもちろん、DB に保存されている問い合わせ内容やメモに、AI への指示が紛れ込むこともあります。OWASP はこれを「間接プロンプトインジェクション」として整理しています。

この記事では PostgreSQL を例に、AI に必要なデータだけを読ませる構成を作ります。書き込みを禁止するだけでなく、別テナントのデータや個人情報を読ませないこと、本番 DB に重いクエリを投げさせないことまで扱います。

前提バージョン: この記事の SQL と Python コードは PostgreSQL 18、Python 3.12、Psycopg 3.3 系を前提にしています。既存 DB の PUBLIC 権限や RLS（Row-Level Security）の構成は環境ごとに異なるため、本番へ適用する前にステージング環境で実効権限を確認してください。

AI には用途別のツールを渡す

私なら、モデルには run_sql(sql: string) を渡しません。代わりに、アプリケーション側で用途の決まったツールを用意します。

• get_daily_sales(date_from, date_to)
• compare_plan_usage(plan, period)
• find_failed_imports(source_id, limit)

AI が決めるのは「どの機能を、どの引数で呼ぶか」までです。SQL の組み立てやテナント ID の付与、認可、実行時間の制限は通常のアプリケーションコードに任せます。

AI と DB の間には Query Broker を置きます。

認証済みユーザー
      ↓
AI オーケストレーター
      ↓
Query Broker  ──→  監査ログ
      ↓
閲覧用ビュー
      ↓
読み取りレプリカ

Query Broker は、認証済みユーザーの権限を引き継ぎ、許可されたクエリだけを実行する小さなバックエンドです。接続先も、できればプライマリではなく読み取りレプリカか分析用 DB にします。

ただし、Query Broker を置くだけでは足りません。アプリケーションにバグがあっても DB 側で止まるように、制限を重ねます。

境界	防ぐこと	主な対策
AI ツール	想定外の操作	用途別の関数、構造化された引数
アプリケーション	なりすまし、過剰取得	認可、引数検証、件数上限、マスキング
DB	書き込み、権限外の参照	専用ロール、ビュー、RLS、読み取り専用トランザクション
インフラ	本番負荷、侵入経路	読み取りレプリカ、閉域接続、接続数制限
運用	異常の見逃し	監査ログ、アラート、失効手順

「読み取り専用ユーザー」だけでは足りない

「読み取り専用なら壊されないので安心」と考えたくなります。確かに、書き込み事故は減ります。しかし SELECT が許可されている以上、まだできることはかなりあります。

1. 同じテーブルに含まれる他テナントの行を取得できる
2. メールアドレスや住所など、分析に不要な列を取得できる
3. 巨大な結合や集計によってプライマリ DB の CPU と I/O を消費できる
4. 大量の結果をモデルへ送り、コスト増加や情報漏えいを起こす
5. PUBLIC 経由で関数やスキーマの権限を持っている可能性がある

しかも PostgreSQL の実効権限は、ロールに直接付与したものだけではありません。所属ロールと PUBLIC の権限も加わります。関数とプロシージャには、デフォルトで PUBLIC の EXECUTE 権限が付くため、既存 DB では一度棚卸しが必要です。詳しくは PostgreSQL の権限に関するドキュメントにまとまっています。

「書き込めない」と「見せてよいデータしか読めない」は別の話です。

DB 側に閲覧専用の窓口を作る

例として、AI には日次売上だけを見せます。元の orders テーブルにはメールアドレス、配送先、備考も入っていますが、売上傾向の説明には不要です。そこで、日付・注文数・売上額に集計したビューを別スキーマに作ります。app_owner は orders を参照できる既存の非ログインロールです。

-- db/ai_reader.sql
BEGIN;

CREATE SCHEMA ai_read AUTHORIZATION app_owner;

CREATE VIEW ai_read.daily_sales
WITH (security_barrier = true)
AS
SELECT
    tenant_id,
    ordered_at::date AS sales_date,
    count(*) AS order_count,
    sum(total_amount) AS revenue
FROM app.orders
WHERE status = 'paid'
GROUP BY tenant_id, ordered_at::date;

ALTER VIEW ai_read.daily_sales OWNER TO app_owner;

CREATE ROLE ai_reader
    LOGIN
    NOSUPERUSER
    NOCREATEDB
    NOCREATEROLE
    NOINHERIT
    NOREPLICATION
    NOBYPASSRLS
    CONNECTION LIMIT 5;

-- パスワード認証を使う場合は、Secret Manager などから別途設定します。

GRANT CONNECT ON DATABASE app TO ai_reader;
GRANT USAGE ON SCHEMA ai_read TO ai_reader;
GRANT SELECT ON ai_read.daily_sales TO ai_reader;

ALTER ROLE ai_reader IN DATABASE app
    SET default_transaction_read_only = on;
ALTER ROLE ai_reader IN DATABASE app
    SET statement_timeout = '3s';
ALTER ROLE ai_reader IN DATABASE app
    SET lock_timeout = '500ms';
ALTER ROLE ai_reader IN DATABASE app
    SET idle_in_transaction_session_timeout = '5s';
ALTER ROLE ai_reader IN DATABASE app
    SET search_path = ai_read, pg_catalog;

COMMIT;

ai_reader は元の orders を読めません。参照できるのは ai_read.daily_sales だけです。アプリケーションで取得してから列を消すのではなく、そもそも生データを取得できない形にしておくほうが安全です。

security_barrier は、悪意のある関数や演算子がビューのフィルタより先に評価されるのを防ぎます。ビューをセキュリティ境界にするなら外せない設定です。詳しい動作は PostgreSQL の Rules and Privilegesを参照してください。

広すぎる権限に注意: GRANT pg_read_all_data TO ai_reader は簡単ですが、AI 用ロールには広すぎます。この定義済みロールはすべてのテーブル、ビュー、シーケンスに対する読み取り権限と、すべてのスキーマに対する USAGE を与えます。AI が必要とするビューへ個別に GRANT SELECT してください。

RLS を使う場合の注意

マルチテナント DB では、RLS で見える行を絞れます。RLS を有効にしたテーブルに適用可能なポリシーがない場合、PostgreSQL はそのアクセスを拒否します。

落とし穴は、スーパーユーザー、BYPASSRLS を持つロール、通常のテーブル所有者なら RLS を迂回できることです。テーブル所有者にも適用したい場合は FORCE ROW LEVEL SECURITY が必要です。PostgreSQL の Row Security Policiesにも、この例外が明記されています。

ビューと RLS の組み合わせも少し厄介です。誰の権限でビューが動くかによって、使われる RLS ポリシーが変わります。security_invoker を付けると呼び出し元の権限で評価されますが、その呼び出し元には基底テーブルの権限も必要です。オプションを 1 つ足せば終わり、とはいきません。

本番では次のどれかを選び、別テナントの行が見えないことをテストします。

• 認可済みの tenant_id を必ず条件に含める用途別クエリ
• RLS と、RLS を迂回しない専用のビュー所有ロール
• テナント単位または顧客単位で分離した分析用データストア

Query Broker を Python で実装する

Query Broker の仕事は地味です。AI から受け取った引数を検証し、決めておいた SQL を実行します。ここでは Psycopg 3.3.4 と Pydantic 2.13.4 を使います。

python -m venv .venv
source .venv/bin/activate
pip install "psycopg[binary,pool]==3.3.4" "pydantic==2.13.4"

日付範囲は最大 31 日にします。tenant_id は AI に選ばせず、認証済みユーザーのコンテキストから渡します。

# src/query_broker.py
import os
from datetime import date
from decimal import Decimal
from typing import Any
from uuid import UUID

from psycopg.rows import dict_row
from psycopg_pool import ConnectionPool
from pydantic import BaseModel, model_validator


class DailySalesArgs(BaseModel):
    date_from: date
    date_to: date

    @model_validator(mode="after")
    def validate_period(self) -> "DailySalesArgs":
        days = (self.date_to - self.date_from).days
        if days < 0:
            raise ValueError("date_to must be on or after date_from")
        if days > 30:
            raise ValueError("the maximum period is 31 days")
        return self


pool = ConnectionPool(
    conninfo=os.environ["AI_DATABASE_URL"],
    min_size=1,
    max_size=5,
    open=True,
    kwargs={"row_factory": dict_row},
)


QUERY = """
    SELECT sales_date, order_count, revenue
    FROM ai_read.daily_sales
    WHERE tenant_id = %s
      AND sales_date BETWEEN %s AND %s
    ORDER BY sales_date
"""


def get_daily_sales(
    authenticated_tenant_id: UUID,
    raw_args: dict[str, Any],
) -> list[dict[str, date | int | Decimal]]:
    args = DailySalesArgs.model_validate(raw_args)

    with pool.connection() as conn:
        with conn.transaction():
            # DB ロールのデフォルトに加え、実行単位でも読み取り専用を強制します。
            conn.execute("SET TRANSACTION READ ONLY")
            conn.execute("SET LOCAL statement_timeout = '3s'")
            conn.execute("SET LOCAL lock_timeout = '500ms'")

            rows = conn.execute(
                QUERY,
                (
                    authenticated_tenant_id,
                    args.date_from,
                    args.date_to,
                ),
            ).fetchall()

    return rows

このコードで一番大事なのは、tenant_id が raw_args に入っていないことです。JWT などを検証して確定したテナント ID を、サーバー側から authenticated_tenant_id に渡します。モデルが別会社の ID を出力しても、クエリには入りません。

日付は Pydantic で検証します。SQL には Psycopg のパラメータとして渡すため、文字列を連結する処理もありません。

PostgreSQL の読み取り専用トランザクションでは、非一時テーブルへの INSERT、UPDATE、DELETE と、多くの DDL が禁止されます。対象となるコマンドは SET TRANSACTION のドキュメントで確認できます。

もちろん、読み取り専用でも重い SELECT は動きます。そこで statement_timeout を設定し、接続プールの max_size と日付範囲にも上限を置きました。Psycopg の ConnectionPoolを使えば、アプリケーションが同時に開く接続数も抑えられます。

モデルに送る前にデータを減らす

DB から取得できたデータを、そのままモデルへ送ってよいとは限りません。DB への接続は安全でも、モデルプロバイダーへ不要な個人情報を送れば、別の場所で問題が起きます。

データはモデルへ届く前に減らします。

1. ビューで不要な列を公開しない
2. SQL で行を絞り、DB 内で集計する
3. アプリケーションで件数とサイズを検査し、必要ならマスキングする
4. 残った結果だけをモデルへ渡す

売上傾向を説明するだけなら、顧客の氏名やメールアドレス、注文メモはいりません。日付、注文数、売上額で十分です。

モデルプロバイダーの確認項目は、学習利用の有無だけではありません。保持期間と保存場所、削除方法、ログ、リージョン、契約上の扱いも確認します。OWASP の Sensitive Information Disclosureでも、データのサニタイズやアクセス制御、参照できるデータソースの制限が対策に挙げられています。

それでも生成 SQL が必要な場合

BI では、質問の種類をすべて事前に用意できないことがあります。Text-to-SQL の出番です。デモでは見栄えがしますが、本番で生成 SQL をそのまま実行するのは避けてください。

先ほどの daily_sales ビューには全テナントの集計行があります。安全なのは、認証済みの tenant_id を固定 SQL が必ず付けるからです。生成 SQL を許可するなら、その前に RLS を適用したビューや、テナントごとの分析ストアを用意します。WHERE tenant_id = ... が生成されているかを認可の根拠にするのは危険です。

生成 SQL には、少なくとも以下の制御が必要です。

1. SQL を構文木として検査する

正規表現では、コメント、CTE、サブクエリ、引用符、方言の違いを正しく扱えません。PostgreSQL 対応の SQL パーサーで構文木に変換し、次を検査します。

• 文が 1 つだけである
• 最上位が SELECT または許可した WITH ... SELECT である
• 許可したスキーマとビューだけを参照している
• INSERT、UPDATE、DELETE、MERGE、COPY、DDL を含まない
• 許可していない関数、システムカタログ、外部接続機能を参照しない
• 結果件数の上限がある

AST を検査しても DB 権限は省けません。パーサーの不具合や未対応の構文はあり得るので、検査を通った SQL も ai_reader と読み取り専用トランザクションで実行します。

2. 実行前にコストを判定する

EXPLAIN (FORMAT JSON) を実行すると、推定コストと推定行数、対象テーブルが分かります。基準を超えたクエリはここで拒否します。統計情報が古ければ推定は外れるので、絶対的な防御ではありません。なお、EXPLAIN ANALYZE はクエリを実際に動かすため、事前審査には向きません。

クエリ負荷をプライマリから確実に切り離したいなら、読み取りレプリカか分析基盤へ接続します。レプリカも無限に耐えられるわけではないので、CPU、I/O、レプリケーション遅延と同時実行数は監視します。

3. 最初は人の承認を入れる

新しいデータソースやビュー、高コストと判定されたクエリ、個人情報を含む結果には、人の承認を挟みます。しばらく運用し、安全だと確認できたパターンから自動化すれば十分です。

OWASP の Excessive Agencyも、自由度の高い汎用ツールを避けること、下流システムの権限を絞ること、影響の大きい処理には人の承認を入れることを勧めています。

接続情報をモデルに見せない

DB の接続文字列を、モデルのコンテキストやツール引数に入れてはいけません。会話履歴にも残さないようにします。接続するのは Query Broker だけで、資格情報は Secret Manager などから実行時に取得します。

DB が IAM や Managed Identity に対応しているなら、長期間有効なパスワードを置くより、短命な認証情報を使うほうが扱いやすいです。例えば Amazon RDS の IAM DB 認証が生成するトークンの有効期間は 15 分です。アプリケーションに DB パスワードを保存する必要もありません。

短命なトークンが解決するのは認証だけです。どのテーブルや行を読めるかは、引き続き DB ロール、ビュー、RLS で決まります。

ネットワークも同じです。DB をインターネットへ公開して IP 制限だけに頼らず、同一 VPC、Private Link、Cloud SQL Connector などを使います。TLS の検証は無効にせず、接続元を Query Broker に絞ります。

監査ログには「誰が、なぜ、何を」を残す

DB の SQL ログだけを見ても、そのクエリが誰のどんな質問から生まれたのかは分かりません。Query Broker には次の情報を残します。

• 認証済みユーザー ID とテナント ID
• 会話 ID、リクエスト ID、ツール名
• 実行したクエリのテンプレート ID または正規化 SQL
• 引数のハッシュまたはマスキング済みの値
• 参照したビュー
• 実行時間、返却行数、返却バイト数
• 許可、拒否、タイムアウト、エラーの結果

一方、接続文字列やアクセストークン、取得した個人情報、モデルへ渡した全文は記録しません。何でもログに入れると、今度は監査ログが情報漏えい経路になります。

DB 側では pg_stat_statementsから、SQL の計画と実行に関する統計を確認できます。アプリケーションのリクエスト ID と結び付けておくと、重いクエリが見つかったときに追いやすくなります。

本番導入前のテスト

正常系の確認だけでは不十分です。モデルが攻撃者の指示どおりに動いている、と仮定して試します。

• UPDATE、DELETE、DROP TABLE が DB 権限で拒否される
• 許可していないテーブルと information_schema を参照できない
• 別テナントの ID を指定してもデータが返らない
• 31 日を超える期間や不正な日付が Query Broker で拒否される
• pg_sleep() や巨大な直積を含むクエリが許可されない、またはタイムアウトする
• 同時リクエストが増えても接続数上限を超えない
• タイムアウト後にトランザクションと接続が正常に回収される
• DB のエラーメッセージやスキーマ情報がユーザーへそのまま返らない
• 監査ログから、実行者、ツール、対象ビュー、行数を追跡できる
• 資格情報を失効すると、新しい接続が直ちに失敗する

システムプロンプトに「秘密情報を表示しない」と書いても、これらのテストは通りません。OWASP も、プロンプトインジェクションを完全に防ぐ方法は明確ではないとしています。ここはプロンプトの工夫ではなく、権限と通常のコード、人の承認、攻撃を想定したテストで止める部分です。

固定クエリから始める

私は、最初から自由な Text-to-SQL を目指さないほうがよいと考えています。固定クエリから始め、必要になった分だけ自由度を上げます。

1. 開発用の匿名化データで回答品質を確認する
2. 分析用 DB に、用途別の固定クエリだけを接続する
3. 閲覧用ビューを増やし、構造化引数の範囲を広げる
4. 読み取りレプリカで、許可したビューに対する生成 SQL を試す
5. 監査と承認の実績を基に、一部のクエリを自動化する

固定クエリで足りるなら、生成 SQL へ進まなくても構いません。ユーザーの質問を安全な処理に変換し、結果を分かりやすく説明できれば、AI を使う価値はあります。

Phaide AI の場合

ここまでの構成を自前で用意するのは、それなりの仕事です。閲覧用ビューと専用ロールを設計し、Query Broker を書き、取り込み経路を運用し、監査ログを整える。分析を始める手前に、これだけ積み上がります。

Phaide AI は、この層をマネージドで提供します。

• 接続はマネージドコネクタ経由。 PostgreSQL や MySQL のほか、Snowflake、BigQuery、Salesforce、HubSpot など700 を超えるデータソースに対応しています。AI が本番のプライマリへ直接クエリを投げることはなく、参照するのは取り込み済みの分析用データベースです。
• 個人情報は取り込み時に落とす。 氏名・メールアドレス・顧客 ID などをマスク対象のカラムに指定すると、分析用データベースへ書き込まれる前に決定論的なトークンへ置き換わります。チャットに実名を打ち込んだ場合も、送信前に同じトークンへ変換されます。
• AI とサンドボックスが見るのはトークンだけ。 決定論的な変換なので、マスクしたカラムでも件数・結合・グルーピングはこれまで通り動きます。元の値に戻るのは、あなたの画面に描画される瞬間だけです。
• 設定は一度きり。 一度保護したカラムは、チャット・ダッシュボード・自動探索エージェントのすべてで保護され続けます。生データを読む画面は用意されていません。

この記事で書いた「見せてよいデータしか読めない状態を、AI の手前で作る」という考え方を、製品として実装したものだと考えていただければと思います。マスキングの仕組みはデータマスキングの記事で詳しく説明しています。プラットフォーム全体は Phaide AI でご覧いただけます。

まとめ

• AI には自由な SQL 実行機能を渡さず、まず用途別の読み取りツールを用意します。
• 専用ロールや閲覧用ビュー、タイムアウトを DB 側にも設定します。
• テナント ID はモデルに選ばせず、認証結果から付与します。
• DB から取得する時点でデータを絞り、必要な結果だけをモデルへ送ります。
• 生成 SQL を使うなら、モデルが攻撃者の指示どおりに動いても被害が広がらないかを基準に設計します。

参考

• OWASP LLM01:2025 Prompt Injection
• OWASP LLM02:2025 Sensitive Information Disclosure
• OWASP LLM06:2025 Excessive Agency
• NIST AI 600-1: Artificial Intelligence Risk Management Framework: Generative Artificial Intelligence Profile
• PostgreSQL: Privileges
• PostgreSQL: Row Security Policies
• PostgreSQL: SET TRANSACTION
• PostgreSQL: Client Connection Defaults
• PostgreSQL: CREATE VIEW
• PostgreSQL: pg_stat_statements
• Psycopg: Connection pools
• Amazon RDS: IAM database authentication