Vector Store汚染を防ぐRAG運用フロー サムネイル

Vector Store 汚染 対策 RAG を検討している運用担当者向けに、取り込み前の属性検査・RBAC設計・差分チェックスクリプトの3点を軸とした実践的な防御フローを解説する。RAGシステムの知識ベースは一度汚染されると発見が困難なため、登録ライフサイクル全体を設計段階から管理する必要がある。

この記事で分かること

  • Vector Storeへの汚染攻撃シナリオと、OWASP LLM08:2025が定義するリスク分類
  • ソース収集から削除・ロールバックまでの知識ベース更新ライフサイクルと承認フロー
  • 実運用で即使えるPython差分チェックスクリプト・RBAC設計パターン・週次/月次プレイブック

Vector Store汚染とは何か

Vector Storeの「汚染」とは、RAGシステムが参照する知識ベース（ベクトルDB）に、悪意のあるまたは不正なドキュメントが混入し、LLMの回答を意図的に歪める状態を指す。通常のプロンプトインジェクションと異なり、汚染はデータ層で発生するため、モデル側の防御だけでは検知も対処もできない。

OWASP LLM08:2025（Vector and Embedding Weaknesses）は、ベクトルストアへの不正データ注入・アクセス制御の欠如・テナント分離の失敗を主要なリスク要因として定義している。代表的な攻撃シナリオは以下の3つだ：

  • 直接ドキュメント注入：登録APIやファイルアップロード機能に対して、正規ユーザーを装って誤情報・有害指示を含むドキュメントを投入する。埋め込み後はベクトル空間で正規コンテンツと区別がつかなくなるため、セマンティック検索で高頻度に召喚され続ける
  • ベクトルストア改ざん検知を回避した間接汚染：外部ソース（Webクロール・RSS・API連携）経由で汚染コンテンツを混入させる。ソース検証が甘い自動取り込みパイプラインが標的になる。RAGデータ属性検査が未整備だと発見まで数週間を要することがある
  • クロステナント漏洩と権限昇格：マルチテナント構成でテナント間の名前空間分離が不十分な場合、他テナントのドキュメントが検索結果に混入する。これは情報漏洩と同時に、汚染済みテナントの知識が別テナントの回答を汚染する二次攻撃にも発展する
⚠️ 注意：PoisonedRAGの研究（USENIX Security 2025）では、100万件規模の知識DBに5件の悪意ある文書を挿入するだけで攻撃成功率90%を達成している。数量的な少数注入でも致命的な影響が出る点を前提として対策を設計すること。

ここまでのまとめ：Vector Store汚染はデータ層の問題であり、モデル単独では防御できない。直接注入・外部ソース経由の間接汚染・クロステナント漏洩の3シナリオを前提とした多層防御が必要となる。

RAG知識ベースの更新ライフサイクルを整理する

汚染を防ぐには、知識ベースの「登録→監視→削除」全工程に管理ゲートを設ける必要がある。以下のASCII図でライフサイクル全体を把握してから、各ステップの詳細に進む。


┌─────────────────────────────────────────────────────────────────────┐
│             RAG 知識ベース更新ライフサイクル                         │
└─────────────────────────────────────────────────────────────────────┘

  [1. ソース収集]
     社内文書 / 外部URL / APIフィード
          │
          ▼
  [2. 取り込み前検査]  ◄── ホワイトリスト照合 / 属性フィルタ / ハッシュ検証
          │
     NG ──┤──── 隔離キュー → セキュリティ担当レビュー → 棄却 or 修正
          │ OK
          ▼
  [3. 承認ゲート]  ◄── RBAC: Reviewer承認 (staging環境でテスト)
          │
     差し戻し ─┤
          │ 承認
          ▼
  [4. インデックス登録]  ◄── メタデータ付与 (source_hash, registered_at, owner)
          │
          ▼
  [5. 継続監視]  ◄── 差分チェック / ログ監査 / 異常検知
          │
     汚染検知 ─┤──── [6. 削除 / ロールバック / 再構築]
          │ 正常
          ▼
       (継続運用)

取り込み前チェック（ソース・属性検査）

ソース収集の段階では、コンテンツを埋め込む前に必ず検査パイプラインを通す。外部コンテンツ ホワイトリストに未登録のドメインからのコンテンツは即時隔離する。社内文書であっても、部署ラベル・機密レベル・フォーマット（PDF/Markdown/HTML）を属性フィールドとして記録し、後続フィルタの根拠にする。

ハッシュ（SHA-256）はドキュメント単位で計算し、登録DB（SQLiteで十分）に保存する。同一ハッシュが既存エントリと一致した場合は重複登録をスキップ、変更があれば差分を検出して再承認フローに回す。

インデックス登録・更新フロー（承認・ロール分担）

取り込み前検査を通過したドキュメントは、本番インデックスに直接登録せずstaging環境でベクトル化・検索テストを行う。Reviewerロール（後述のRBAC設計参照）が承認した時点ではじめてproductionインデックスへ反映する。この承認ログはRAGデータ属性検査の監査証跡として保存する。

更新の単位はドキュメント全体の置き換えを基本とし、パッチ更新は認めない。部分的な書き換えはハッシュ管理を複雑化し、改ざん検知の精度を下げる。

削除・ロールバック・再構築の手順

汚染を検知した場合の手順は以下の通り固定する：

  1. 汚染ドキュメントのIDを特定し、ベクトルDBから即時削除（論理削除フラグを立て物理削除は後回しにしない）
  2. 同一ソース・同一バッチ由来のドキュメントを一括で隔離対象にする（バッチIDのメタデータが必須）
  3. 汚染ドキュメントが登録された時点より前のスナップショットからインデックスを再構築する（スナップショットは最低直近3世代保持）
  4. インシデントログに影響範囲・発見時刻・対応者を記録し、原因分析を完了してから運用再開する
⚠️ 注意：インデックスを再構築する際、同じ汚染ソースが自動取り込みパイプラインに残っていると再汚染する。削除と並行してソースのブラックリスト登録を行うこと。

ここまでのまとめ：ライフサイクルの各ステップにゲートを設けることで汚染の混入経路を限定できる。承認・ハッシュ管理・スナップショットの3機構がロールバック対応の前提条件となる。

属性検査とホワイトリスト設計の実装

外部コンテンツのドメインホワイトリスト設計（Pythonサンプル）

外部ソースからのコンテンツを取り込む際は、許可ドメインリストとの照合を必須とする。ホワイトリスト管理はYAMLで外部化し、GitリポジトリでPRレビューを経て変更する運用とする。

# whitelist.yaml
allowed_domains:
  - docs.python.org
  - owasp.org
  - arxiv.org
blocked_patterns:
  - "*.blogspot.com"
  - "*.medium.com"   # vendor blog exclusion per policy

# ファイル名: source_validator.py
import re
import hashlib
from urllib.parse import urlparse
from pathlib import Path
import yaml


def load_whitelist(path: str = "whitelist.yaml") -> dict:
    with open(path, "r") as f:
        return yaml.safe_load(f)


def is_domain_allowed(url: str, whitelist: dict) -> bool:
    hostname = urlparse(url).netloc.lower()

    # Exact match against allowed_domains
    if hostname in whitelist.get("allowed_domains", []):
        return True

    # Reject if matches any blocked pattern
    for pattern in whitelist.get("blocked_patterns", []):
        regex = re.escape(pattern).replace(r"\*", ".*")
        if re.fullmatch(regex, hostname):
            return False

    return False


def compute_sha256(content: bytes) -> str:
    return hashlib.sha256(content).hexdigest()


def validate_external_source(url: str, content: bytes, whitelist: dict) -> dict:
    """
    Returns a validation result dict.
    status: "approved" | "rejected"
    reason: human-readable explanation
    hash: SHA-256 of content
    """
    result = {
        "url": url,
        "hash": compute_sha256(content),
        "status": "rejected",
        "reason": "",
    }

    if not is_domain_allowed(url, whitelist):
        result["reason"] = f"Domain not in whitelist: {urlparse(url).netloc}"
        return result

    # Minimum content size guard (avoid empty/stub pages)
    if len(content) < 512:
        result["reason"] = "Content too short (< 512 bytes), possible error page"
        return result

    result["status"] = "approved"
    result["reason"] = "Passed whitelist and size checks"
    return result


# --- Usage example ---
if __name__ == "__main__":
    wl = load_whitelist()
    test_url = "https://owasp.org/www-project-top-10-for-large-language-model-applications/"
    dummy_content = b"x" * 1024  # replace with real fetched bytes
    print(validate_external_source(test_url, dummy_content, wl))

社内文書の属性ベースフィルタ（部署・ラベル・フォーマット）

社内文書は「部署」「機密ラベル」「ファイルフォーマット」の3属性をメタデータとして付与し、RAGの検索クエリ実行時にも属性でフィルタリングする。これにより、人事部のドキュメントが営業部のRAGクエリに引っかかるクロス汚染を防ぐ。

# ファイル名: attribute_filter.py
from dataclasses import dataclass, field
from typing import Literal


ALLOWED_FORMATS = {"pdf", "md", "txt", "docx"}
SENSITIVITY_LEVELS = {"public", "internal", "confidential", "restricted"}


@dataclass
class DocumentMetadata:
    doc_id: str
    source_url: str
    department: str
    sensitivity: Literal["public", "internal", "confidential", "restricted"]
    file_format: str
    registered_by: str
    batch_id: str
    source_hash: str
    tags: list[str] = field(default_factory=list)


def validate_metadata(meta: DocumentMetadata) -> tuple[bool, str]:
    if meta.file_format.lower() not in ALLOWED_FORMATS:
        return False, f"Unsupported format: {meta.file_format}"

    if meta.sensitivity not in SENSITIVITY_LEVELS:
        return False, f"Unknown sensitivity level: {meta.sensitivity}"

    if not meta.department or not meta.registered_by:
        return False, "department and registered_by are required fields"

    if not meta.batch_id:
        return False, "batch_id is required for rollback traceability"

    return True, "OK"


def build_query_filter(department: str, max_sensitivity: str) -> dict:
    """
    Build a metadata filter dict for vector store query (e.g., Qdrant / Weaviate filter).
    max_sensitivity: highest sensitivity level the requester is allowed to see.
    """
    level_order = ["public", "internal", "confidential", "restricted"]
    allowed = level_order[: level_order.index(max_sensitivity) + 1]
    return {
        "must": [
            {"key": "department", "match": {"value": department}},
            {"key": "sensitivity", "match": {"any": allowed}},
        ]
    }

ここまでのまとめ：外部コンテンツはドメインホワイトリスト＋コンテンツサイズのダブルチェックで弾く。社内文書は部署・機密ラベル・フォーマットの3属性をメタデータとして強制し、クエリ時にも同属性フィルタを適用することで情報境界を維持する。

Vector Store改ざん検知のためのログ＆クエリ設計

必須ログ項目一覧

イベント名 記録すべきフィールド 検知ルール例
doc.ingest.requested doc_id, source_url, source_hash, requester_id, batch_id, timestamp 同一source_hashが24h以内に3回以上リクエストされたら警告
doc.ingest.approved doc_id, reviewer_id, batch_id, approved_at, staging_test_result reviewer_idがrequester_idと同一なら自己承認アラート
doc.ingest.rejected doc_id, reason_code, rejected_by, timestamp 同一ユーザーの棄却率が急増したら権限昇格試行の疑い
doc.delete doc_id, deleted_by, reason, batch_id, physical_delete_at 批准なしの削除（deleted_by がAdminロール以外）はアラート必須
index.snapshot.created snapshot_id, index_version, triggered_by, size_bytes, timestamp 前回スナップショットから72h以上経過したら作成漏れアラート
query.retrieval query_id, user_id, top_k_doc_ids, similarity_scores, latency_ms 特定doc_idが上位1位を連続30回以上占めたらリトリーバル偏り検知
index.hash_mismatch doc_id, expected_hash, actual_hash, detected_at 発生した瞬間にP1インシデントとして即時通知

差分チェックスクリプト（Python）

以下のスクリプトは、登録時に保存したSHA-256ハッシュとベクトルDBから再取得したドキュメントコンテンツのハッシュを比較し、改ざんされたエントリを検出する。定期実行（cron / Airflow DAG）に組み込む想定で設計している。

# ファイル名: diff_checker.py
"""
Vector Store integrity diff checker.
Compares stored source_hash metadata against recomputed hashes of raw content.
Run this on a schedule (e.g., every 6 hours) to detect tampering.
"""
import hashlib
import json
import logging
import sys
from dataclasses import dataclass, asdict
from datetime import datetime, timezone
from typing import Protocol

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
logger = logging.getLogger(__name__)


# --- Protocol for vector store client (implement per your DB: Qdrant/Weaviate/Pinecone) ---
class VectorStoreClient(Protocol):
    def scroll_all(self, collection: str) -> list[dict]:
        """Yield all documents with id, content bytes, and metadata."""
        ...


@dataclass
class IntegrityViolation:
    doc_id: str
    expected_hash: str
    actual_hash: str
    detected_at: str
    source_url: str
    severity: str  # "CRITICAL" | "WARNING"


def compute_sha256(content: bytes) -> str:
    return hashlib.sha256(content).hexdigest()


def run_diff_check(
    client: VectorStoreClient,
    collection: str,
    output_report: str = "integrity_report.json",
) -> list[IntegrityViolation]:
    violations: list[IntegrityViolation] = []
    docs = client.scroll_all(collection)

    logger.info(f"Checking {len(docs)} documents in collection '{collection}'")

    for doc in docs:
        doc_id = doc["id"]
        metadata = doc.get("metadata", {})
        expected_hash = metadata.get("source_hash")
        raw_content: bytes = doc.get("raw_content", b"")

        if not expected_hash:
            logger.warning(f"[SKIP] doc_id={doc_id}: no source_hash in metadata")
            continue

        actual_hash = compute_sha256(raw_content)

        if actual_hash != expected_hash:
            violation = IntegrityViolation(
                doc_id=doc_id,
                expected_hash=expected_hash,
                actual_hash=actual_hash,
                detected_at=datetime.now(timezone.utc).isoformat(),
                source_url=metadata.get("source_url", "unknown"),
                severity="CRITICAL",
            )
            violations.append(violation)
            logger.error(
                f"[HASH MISMATCH] doc_id={doc_id} "
                f"expected={expected_hash[:12]}... actual={actual_hash[:12]}..."
            )

    # Write structured report
    report = {
        "checked_at": datetime.now(timezone.utc).isoformat(),
        "collection": collection,
        "total_checked": len(docs),
        "violations_found": len(violations),
        "violations": [asdict(v) for v in violations],
    }
    with open(output_report, "w") as f:
        json.dump(report, f, indent=2, ensure_ascii=False)

    logger.info(
        f"Diff check complete: {len(violations)} violation(s) found. "
        f"Report saved to {output_report}"
    )

    if violations:
        # Exit with non-zero code so CI/CD / alerting systems catch it
        sys.exit(1)

    return violations


# --- Stub client for local testing ---
class StubVectorStoreClient:
    def scroll_all(self, collection: str) -> list[dict]:
        return [
            {
                "id": "doc_001",
                "raw_content": b"legitimate content",
                "metadata": {
                    "source_hash": compute_sha256(b"legitimate content"),
                    "source_url": "https://docs.python.org/example",
                },
            },
            {
                "id": "doc_002",
                "raw_content": b"tampered content",  # hash won't match
                "metadata": {
                    "source_hash": compute_sha256(b"original content"),
                    "source_url": "https://owasp.org/example",
                },
            },
        ]


if __name__ == "__main__":
    stub_client = StubVectorStoreClient()
    run_diff_check(stub_client, collection="knowledge_base")
⚠️ 注意：raw_contentの取得方法はVector DBの実装依存（Qdrantならpayloadフィールド、Weaviateならpropertyとして別途保存する設計が必要）。埋め込みベクトルからの逆算はできないため、元コンテンツを別途保存するアーキテクチャを最初から設計に組み込むこと。

ここまでのまとめ：ログは7種類のイベントを最低限記録し、特にindex.hash_mismatchはP1インシデントとして即時通知するルールを設ける。差分チェックスクリプトはCI/CDや定期ジョブに組み込み、違反検知時は非ゼロ終了コードで自動アラートにつなげる。

権限設計とテナント分割の考え方

RBAC ベクトルストアの設計は、「誰がどのコレクションに対してどの操作を実行できるか」を明示的に定義することから始まる。以下の3パターンはシステム規模・マルチテナント要件によって選択する。

設計パターン パターンA: フラットRBAC パターンB: コレクション分離RBAC パターンC: テナント分割+RBAC
概要 全ユーザーが単一コレクションを使用。ロールで操作権限のみ分ける 部署・用途ごとにコレクションを分割。ロールはコレクション単位で付与 テナントごとに独立したインデックスを持ち、さらにコレクション内RBAC
適用規模 小規模・単一テナント（〜5チーム） 中規模・単一組織マルチ部署（5〜20チーム） 大規模・マルチテナントSaaS・外部顧客分離が必要な場合
ロール例 Admin / Reviewer / Writer / Reader Admin / Collection_Owner / Reviewer / Reader（各コレクション） Platform_Admin / Tenant_Admin / Reviewer / Reader（テナントスコープ）
テナント漏洩リスク 高（名前空間なし） 中（コレクション境界で分離） 低（物理/論理分離）
運用コスト 低 中 高（テナント管理・プロビジョニング自動化が必要）
推奨DB Chroma / FAISS Qdrant / Weaviate Pinecone（Namespace）/ Qdrant（コレクション per テナント）
OWASP LLM08対応度 △（基本対応のみ） ○（部署間分離は達成） ◎（クロステナント攻撃シナリオに完全対応）

パターンBとCでは、ロール付与の最小権限原則を徹底する。Writerロールはドキュメントの新規登録のみを許可し、更新・削除はReviewer以上が承認してからAdmin経由で実行する設計が標準だ。ロールの棚卸しは月次で実施する（後述プレイブック参照）。

ここまでのまとめ：テナント分離が必要なシステムではパターンCを選択し、コレクション境界の外側でも認証レイヤーを持つ。パターン選択の根拠（テナント数・漏洩リスク許容度）は設計ドキュメントに残しておくことで監査対応を容易にする。

週次・月次のチェックリスト運用プレイブック

以下のチェックリストをGitリポジトリのIssueテンプレートまたはRunbookとして管理し、担当者が都度チェックを記録する形で運用する。

週次チェックリスト（毎週月曜 AM実施）

  • ☐ 差分チェックスクリプト（diff_checker.py）の最新実行ログを確認し、違反ゼロを確認する
  • ☐ doc.ingest.requested ログから未承認のまま72h超過したドキュメントを抽出し、Reviewerに催促する
  • ☐ index.hash_mismatch イベントの発生件数を確認する（1件でもあれば即時インシデント起票）
  • ☐ クエリログ（query.retrieval）の上位召喚ドキュメントTOP10を確認し、想定外のドキュメントが上位を占めていないか確認する
  • ☐ ホワイトリスト（whitelist.yaml）の変更差分をGitログで確認し、未レビューのPRが存在しないことを確認する
  • ☐ 自動取り込みパイプラインの実行ログを確認し、エラー・リトライが異常頻度で発生していないことを確認する
  • ☐ staging環境のインデックスとproductionの差分を確認し、未反映のドキュメントがないことを確認する

月次チェックリスト（毎月第1営業日実施）

  • ☐ 全ロールの棚卸しを実施し、退職・異動したメンバーの権限を削除する
  • ☐ テナント分離設定（コレクション・名前空間）を実際のクエリでテストし、クロステナント漏洩がないことを確認する
  • ☐ インデックスのスナップショットが直近3世代以上保持されていることを確認する
  • ☐ 社内文書の属性メタデータ（部署ラベル・機密レベル）に不整合がないことをスキャンする
  • ☐ RAG知識ベース 更新フローの承認者（Reviewerロール）が各チームに最低1名以上いることを確認する（バスファクター対策）
  • ☐ OWASPのLLM08最新ガイドラインに変更がないか確認し、対策の陳腐化をチェックする
  • ☐ インシデントログの振り返りを実施し、再発防止策がRunbookに反映済みであることを確認する
  • ☐ 差分チェックスクリプトのカバレッジ（全コレクションを対象にしているか）を確認し、新規追加コレクションが検査対象に含まれていることを確認する

ここまでのまとめ：週次は「異常の即時発見」、月次は「権限・構成の正常性確認」と役割を分けてチェック粒度を設計する。チェックリストはGit管理し、誰がいつ実施したかの証跡を残す。

まとめとロールアサイン

以下のRACI表で、本記事で定義した全タスクの責任分担を示す。R=Responsible（実行）、A=Accountable（最終責任）、C=Consulted（相談）、I=Informed（報告先）。

タスク MLエンジニア バックエンドエンジニア セキュリティ担当 チームリード/マネージャー
ホワイトリスト設計・更新 C R A I
属性フィルタ実装・メタデータ設計 R C C I
インデックス登録承認（Reviewer） A / R C I I
差分チェックスクリプト運用 R R A I
ログ監視・アラート対応 R C A / R I
RBAC設計・ロール付与 C C A / R I
月次ロール棚卸し I I R A
インシデント対応・ロールバック R R A I
スナップショット管理 R C I A
OWASP LLM08定期レビュー C I R / A I
⚠️ 注意：「MLエンジニアがA（最終責任）とR（実行）を兼務するタスク」は、インデックス登録承認のみとする。自己承認を避けるため、自分が登録リクエストしたドキュメントの承認は別のReviewerが担当するルールを明文化すること。

📄 関連記事：