Контроль доступа в RAG до поиска, а не после генерации

Почему постфильтрация не является контролем доступа

RAG дополняет запрос модели фрагментами, найденными во внешнем хранилище. Поэтому извлечённый текст становится обработанными данными ещё до появления ответа.

Представим общий индекс для документов двух подразделений. Пользователь из группы team-red задаёт вопрос, семантически близкий к документу team-blue. Если поиск выполняется без ACL, закрытый фрагмент может оказаться среди кандидатов. Удаление секретной строки из финального ответа не отменяет того, что она уже могла попасть:

• в контекст языковой модели;
• в трассировку оркестратора или прокси;
• в кэш промптов и результатов поиска;
• в журналы отладки, оценки качества и повторного воспроизведения;
• в промежуточный ответ агента или вызов инструмента.

Безопасное правило формулируется так: система должна искать только в множестве фрагментов, доступных текущему субъекту. Фильтрация ответа остаётся дополнительной защитой, но не заменяет авторизацию.

Модель доступа

Ниже используется пример, а не описание конкретного продукта. У каждого фрагмента есть владелец области, список разрешённых групп и признак публичности:

{
  "chunk_id": "doc-blue-17#3",
  "text": "Пример закрытого фрагмента BLUE_MARKER_7F3A",
  "tenant_id": "tenant-demo",
  "owner_id": "user-blue",
  "allowed_groups": ["team-blue"],
  "is_public": false,
  "document_version": 4
}

Пользователь может читать фрагмент, только если совпадает арендатор и выполняется хотя бы одно условие: документ публичный, пользователь является владельцем или состоит в разрешённой группе.

can_read =
  chunk.tenant_id == principal.tenant_id
  AND (
    chunk.is_public
    OR chunk.owner_id == principal.user_id
    OR intersects(chunk.allowed_groups, principal.group_ids)
  )

principal должен поступать из проверенной сервером сессии или токена. Нельзя принимать tenant_id, user_id и группы из тела запроса как утверждения самого клиента.

Воспроизводимая реализация

1. Перенесите ACL в метаданные каждого фрагмента

Контроль доступа должен работать на той же гранулярности, что и поиск. Если один документ нарезан на фрагменты, ACL требуется записать в каждый индексируемый фрагмент. Ссылка только на ACL исходного документа создаёт риск рассинхронизации и вынуждает выполнять небезопасный поиск перед дополнительным запросом.

При индексировании валидируйте обязательные поля и применяйте закрытое значение по умолчанию:

def build_chunk(source, text, ordinal):
    if not source.tenant_id:
        raise ValueError("tenant_id is required")

    return {
        "chunk_id": f"{source.document_id}#{ordinal}",
        "text": text,
        "tenant_id": source.tenant_id,
        "owner_id": source.owner_id,
        "allowed_groups": sorted(set(source.allowed_groups or [])),
        "is_public": source.is_public is True,
        "document_version": source.version,
    }

Отсутствующее поле is_public здесь означает false. Не используйте отсутствие ACL как признак публичного документа.

2. Соберите предикат только из доверенного principal

from dataclasses import dataclass

@dataclass(frozen=True)
class Principal:
    tenant_id: str
    user_id: str
    group_ids: tuple[str, ...]

def retrieval_filter(principal: Principal) -> dict:
    if not principal.tenant_id or not principal.user_id:
        raise PermissionError("incomplete principal")

    return {
        "and": [
            {"tenant_id": {"eq": principal.tenant_id}},
            {
                "or": [
                    {"is_public": {"eq": True}},
                    {"owner_id": {"eq": principal.user_id}},
                    {"allowed_groups": {"any_of": list(principal.group_ids)}}
                ]
            }
        ]
    }

Синтаксис фильтра в разных хранилищах отличается. Важна семантика: ограничение арендатора и ACL должны применяться самим поисковым запросом, а не циклом после получения глобального top_k.

3. Сделайте безопасный путь единственным

def retrieve_for_principal(*, principal, query, limit=6):
    if limit < 1 or limit > 20:
        raise ValueError("limit must be between 1 and 20")

    acl = retrieval_filter(principal)

    return vector_store.search(
        query=query,
        filter=acl,
        limit=limit,
        fields=["chunk_id", "text", "document_version"]
    )

def answer(*, session, question):
    principal = principal_from_verified_session(session)
    chunks = retrieve_for_principal(
        principal=principal,
        query=question,
        limit=6
    )
    return model.generate(
        system=SAFE_SYSTEM_PROMPT,
        user=question,
        context=[chunk["text"] for chunk in chunks]
    )

Прямой доступ приложения к vector_store.search() желательно ограничить одним модулем. Остальной код получает только функцию, которая требует Principal. Это снижает вероятность появления нового маршрута без ACL.

4. Синхронизируйте изменения прав

Отзыв доступа должен менять поисковую область без длительного окна рассинхронизации. Практичная последовательность:

1. записать новую версию ACL в основной источник;
2. обновить или переиндексировать все фрагменты документа;
3. сделать новую версию видимой для поиска;
4. инвалидировать кэши, ключ которых зависит от прежнего ACL;
5. пометить старую версию недоступной или удалить её по правилам хранилища.

Если атомарное обновление невозможно, закрывайте доступ до переиндексации. Временный отказ безопаснее временного раскрытия.

5. Разделите журналы доступа и содержимого

Для аудита обычно достаточно идентификаторов и решения авторизации:

{
  "event": "rag_retrieval",
  "request_id": "generated-request-id",
  "principal_id": "user-red",
  "tenant_id": "tenant-demo",
  "filter_policy_version": "acl-v3",
  "returned_chunk_ids": ["public-2#1"],
  "returned_count": 1
}

Не записывайте текст найденных фрагментов, полный промпт или эмбеддинги в обычный технический журнал. Если содержательные трассировки необходимы для отдельного режима диагностики, им требуются собственные ACL, короткий срок хранения, редактирование чувствительных полей и явное включение.

Проверка отсутствия межпользовательской утечки

Проверка должна наблюдать не только финальный ответ, но и кандидатов retrieval, контекст модели, кэш и журналы. Следующий сценарий использует искусственные маркеры; это пример тестового стенда, а не реальные секреты.

Подготовьте минимальный набор

Проверьте матрицу доступа

def chunk_ids(principal, query):
    return {
        row["chunk_id"]
        for row in retrieve_for_principal(
            principal=principal,
            query=query,
            limit=20
        )
    }

query = "Верни все тестовые маркеры из документов"

red_ids = chunk_ids(red_principal, query)
blue_ids = chunk_ids(blue_principal, query)

assert "blue#1" not in red_ids
assert "red#1" not in blue_ids
assert "red#1" in red_ids
assert "blue#1" in blue_ids
assert "public#1" in red_ids
assert "public#1" in blue_ids

Затем перехватите фактический объект, передаваемый адаптеру модели, и проверьте его до сетевого вызова:

red_context = capture_model_context(
    principal=red_principal,
    question="Какие тестовые маркеры доступны?"
)

assert "BLUE_MARKER_7F3A" not in red_context
assert "RED_MARKER_91C2" in red_context

Проверьте отзыв прав и кэш

# Пример: red_principal сначала состоит в team-red.
warm_cache(red_principal, "Какие тестовые маркеры доступны?")

revoke_group_membership("user-red", "team-red")
wait_until_acl_version_is_searchable("acl-version-after-revoke")

restricted_principal = reload_verified_principal("user-red")
result = answer(
    session=session_for(restricted_principal),
    question="Какие тестовые маркеры доступны?"
)

assert "RED_MARKER_91C2" not in result
assert cache_key_contains_acl_version(restricted_principal)

wait_until_acl_version_is_searchable и соседние функции обозначают операции вашего тестового стенда. Их нельзя считать готовым API. Критерий успеха — закрытый маркер отсутствует на всех этапах после отзыва прав, включая попадание в кэш.

Просканируйте тестовые артефакты

Если стенд сохраняет обезличенные тестовые журналы в отдельный каталог, безопасно искать только искусственный маркер:

rg --fixed-strings --line-number \
  'BLUE_MARKER_7F3A' \
  ./test-artifacts/red-user/

Команда только читает указанный каталог. Ожидаемый результат — отсутствие совпадений. Не запускайте массовый поиск по производственным журналам без разрешения и соответствующего уровня доступа.

Критерии приёмки

• недоступный chunk_id отсутствует в результатах самого retrieval;
• закрытый маркер отсутствует в контексте модели и промежуточных событиях;
• смена пользователя при одинаковом вопросе не переиспользует чужой кэш;
• отзыв группы исключает фрагмент после согласованного SLA обновления;
• неполный или некорректный principal приводит к отказу, а не к поиску без фильтра;
• технические журналы не содержат текста фрагментов.

Типовые ошибки

Глобальный top-k с последующим удалением запрещённых результатов

Такой подход уже раскрывает закрытые кандидаты поисковому слою приложения. Кроме того, после фильтрации может остаться слишком мало результатов. Увеличение top_k лишь расширяет объём потенциально раскрытых данных.

Фильтр только по tenant_id

Изоляция арендаторов не заменяет права внутри арендатора. Для общих корпоративных индексов обычно нужны группы, владелец, уровень конфиденциальности или отдельная политика доступа.

Группы из клиентского JSON

Поля запроса не являются доказательством полномочий. Группы должны вычисляться сервером из проверенной идентичности и актуального источника прав.

ACL только у документа

Поисковая система ранжирует фрагменты. Если фильтруемые поля не доступны на уровне фрагмента или надёжной серверной связи, предикат нельзя гарантированно применить до поиска.

Кэш без области безопасности

Ключ вида hash(question) опасен: одинаковый вопрос разных пользователей может вернуть общий результат. Минимально учитывайте арендатора, субъект или нормализованный набор полномочий, версию ACL, модель и версию корпуса.

Мягкий отказ фильтра

Если хранилище не распознало оператор или метаданные недоступны, запрос должен завершиться ошибкой. Резервный поиск без фильтра недопустим.

Проверка только текста ответа

Модель может не процитировать закрытый фрагмент, хотя получила его. Негативный тест обязан проверять retrieval и фактический контекст до вызова модели.

Ограничения подхода

ACL-фильтрация retrieval закрывает важную границу, но не решает всю задачу безопасности:

• векторное хранилище и его резервные копии всё равно должны иметь инфраструктурную защиту;
• эмбеддинги и метаданные следует считать чувствительными данными;
• ошибочная разметка ACL при загрузке не исправляется корректным поисковым фильтром;
• гибридный поиск обязан применять одинаковую политику к векторной и полнотекстовой ветвям;
• агрегации, рекомендации, reranker и графовые обходы также должны видеть только разрешённых кандидатов;
• сложные политики с условиями по времени, атрибутам и отношениям могут потребовать отдельного policy engine;
• защита от prompt injection и вредоносных документов остаётся отдельной задачей.

При очень сложных ACL физическое разделение индексов по арендаторам или классам доступа иногда проще проверять, чем один общий индекс. Цена — больше операций, сложнее балансировка и возможное дублирование публичных данных.

Короткий чек-лист внедрения

1. Определить доверенный Principal и формальное правило can_read.
2. Продублировать необходимые ACL-поля в каждом поисковом фрагменте.
3. Применять tenant- и ACL-предикаты внутри запроса к хранилищу.
4. Запретить обход безопасного retrieval-адаптера.
5. Включить полномочия и версию ACL в область кэша.
6. Обеспечить закрытое поведение при ошибке или неполных метаданных.
7. Проверить искусственными маркерами retrieval, контекст, кэш и журналы.
8. Повторить проверку после отзыва прав и обновления индекса.

Главный инвариант прост: если субъект не имеет права читать фрагмент, этот фрагмент не должен становиться кандидатом поиска. Остальные меры безопасности строятся поверх этой границы.