Изоляция данных клиентов в многопользовательском агенте

Модель угрозы и инвариант

Tenant — изолированная область данных одного клиента или организации. В этой статье идентификатор такой области обозначен как tenant_id.

Главный инвариант системы формулируется без упоминания модели: запрос, выполняемый от имени tenant A, не может прочитать, изменить, удалить или косвенно обнаружить ресурс tenant B. Это относится не только к сообщениям, но и к документам, эмбеддингам, истории инструментов, кэшу, трассировкам, файлам и очередям задач.

Шаг 1. Зафиксируйте границы tenant

Сначала составьте реестр всех мест, где агент читает или сохраняет состояние. Для каждого ресурса задайте владельца и механизм принудительного ограничения.

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

Шаг 2. Создайте неизменяемый контекст выполнения

Ниже приведён пример на Python. Это шаблон, а не утверждение о конкретном фреймворке или клиенте.

from dataclasses import dataclass

@dataclass(frozen=True)
class ExecutionContext:
    tenant_id: str
    actor_id: str
    request_id: str

def build_context(authenticated_session, request_id: str) -> ExecutionContext:
    # tenant_id и actor_id берутся только из проверенной сервером сессии.
    return ExecutionContext(
        tenant_id=authenticated_session.tenant_id,
        actor_id=authenticated_session.actor_id,
        request_id=request_id,
    )

Не объединяйте этот объект с аргументами, созданными моделью. Инструмент получает бизнес-параметры от модели, а область доступа — отдельно от доверенного сервера:

def search_documents(ctx: ExecutionContext, query: str, limit: int = 5):
    safe_limit = max(1, min(limit, 20))
    return document_store.search(
        tenant_id=ctx.tenant_id,
        query=query,
        limit=safe_limit,
    )

Если модель сгенерирует аргумент tenant_id, схема инструмента должна его отвергнуть или вовсе не содержать такого поля.

Шаг 3. Применяйте фильтр до чтения данных

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

SELECT id, title, body
FROM agent_documents
WHERE tenant_id = :tenant_id
  AND id = :document_id;

Для списка или поиска действует то же правило:

SELECT id, title
FROM agent_documents
WHERE tenant_id = :tenant_id
  AND searchable_text @@ websearch_to_tsquery(:query)
ORDER BY updated_at DESC
LIMIT :limit;

Составной уникальный ключ дополнительно исключает ошибочные предположения о глобальности идентификатора:

ALTER TABLE agent_documents
ADD CONSTRAINT agent_documents_tenant_id_id_unique
UNIQUE (tenant_id, id);

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

Шаг 4. Изолируйте память, поиск и кэш

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

def retrieve_memory(ctx: ExecutionContext, embedding: list[float]):
    return vector_store.query(
        vector=embedding,
        top_k=8,
        namespace=ctx.tenant_id,
        # Если namespace недоступен, примените серверный metadata-фильтр:
        # filter={"tenant_id": {"$eq": ctx.tenant_id}}
    )

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

Кэшируйте ответ с учётом всех параметров, влияющих на доступ:

def cache_key(ctx: ExecutionContext, prompt_hash: str, policy_version: str):
    return f"agent:{ctx.tenant_id}:{ctx.actor_id}:{policy_version}:{prompt_hash}"

Общий семантический кэш особенно рискован: похожие запросы разных клиентов не означают, что ответы взаимозаменяемы.

Шаг 5. Закройте инструменты и фоновые задачи

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

def update_document(ctx: ExecutionContext, document_id: str, patch: dict):
    updated = document_store.update_where(
        where={
            "tenant_id": ctx.tenant_id,
            "id": document_id,
        },
        patch=allowlisted_patch(patch),
    )
    if updated == 0:
        # Одинаковый внешний ответ для отсутствующего и чужого ресурса.
        raise ResourceNotFound()

Фоновому работнику передавайте минимальные идентификаторы. Перед выполнением он должен заново получить ресурс через запрос с tenant_id, а не использовать сериализованный документ из очереди. Для длительных задач учитывайте отзыв доступа и смену членства пользователя.

Шаг 6. Добавьте воспроизводимые тесты изоляции

Следующие тесты — пример структуры. Они используют синтетические метки, созданные внутри теста, и не описывают реальных клиентов, секреты или производственные данные.

def test_tenant_cannot_read_foreign_document(app):
    tenant_a = app.test_tenant()
    tenant_b = app.test_tenant()

    marker = "isolation-marker-created-by-test"
    foreign = app.create_document(tenant=tenant_b, body=marker)

    response = app.as_tenant(tenant_a).get(
        f"/documents/{foreign.id}"
    )

    assert response.status_code == 404
    assert marker not in response.text


def test_agent_retrieval_excludes_foreign_memory(app):
    tenant_a = app.test_tenant()
    tenant_b = app.test_tenant()

    marker = "foreign-memory-marker-created-by-test"
    app.add_memory(tenant=tenant_b, text=marker)

    result = app.as_tenant(tenant_a).ask(
        "Верни доступные фрагменты памяти"
    )

    assert marker not in result.answer
    assert marker not in result.retrieved_texts


def test_cache_key_is_tenant_scoped(app):
    tenant_a = app.test_tenant()
    tenant_b = app.test_tenant()
    prompt_hash = "fixed-test-hash"

    key_a = app.cache_key(tenant_a, prompt_hash)
    key_b = app.cache_key(tenant_b, prompt_hash)

    assert key_a != key_b

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

Безопасный пример локального запуска выбранного набора тестов:

python -m pytest tests/test_tenant_isolation.py -q

Команда предполагает существующий тестовый файл и настроенное тестовое окружение. Не запускайте проверки изоляции против production: негативные сценарии должны работать на одноразовых синтетических данных.

Проверка результата

Изоляцию можно считать подтверждённой для проверенного контура, когда выполняются все условия:

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

Дополните модульные тесты интеграционным сценарием: создайте два временных tenant, положите разные синтетические маркеры во все используемые хранилища и выполните одинаковые запросы от каждого. Каждый контур должен видеть только свой маркер.

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

«Мы написали это в системном промпте»

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

Глобальный идентификатор считается достаточной защитой

Трудно угадываемый ID не подтверждает право доступа. Всегда запрашивайте ресурс по паре tenant_id + id.

Фильтр есть в API, но отсутствует в инструментах

Внутренние вызовы агента нередко обходят HTTP-обработчик. Общий слой репозитория должен требовать контекст tenant.

Один mutable-объект памяти используется между запросами

Глобальная история диалога, singleton-сессия или повторно используемый буфер могут смешать параллельные обращения. Создавайте состояние на запрос или tenant.

Tenant не входит в ключ кэша

Совпадающий промпт получает ранее сохранённый ответ другого клиента даже при корректной работе основной БД.

Тест проверяет только код ответа

Утечка может находиться в retrieval, трассировке или аргументах инструмента. Проверяйте промежуточные данные в контролируемом тестовом окружении.

Ограничения

Описанная схема снижает риск межклиентской утечки, но не решает все задачи безопасности. Она не заменяет разграничение ролей внутри одного tenant, шифрование, управление ключами, защиту резервных копий, сроки хранения данных и аудит административного доступа.

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

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

Что делать дальше

Используйте этот материал как основу для threat model и обязательного checklist при ревью. Другие практические материалы собраны в руководствах Agent Lab Journal, а определения терминов — в глоссарии.