Как вы защищаете multi-agent систему от вредоносного агента?

Краткий тезис

Защита multi-agent системы от вредоносного агента строится на комбинации изоляции, минимальных привилегий, мониторинга и человеческого контроля. Ключевая идея — ни один агент не должен иметь неограниченного доступа к ресурсам или другим агентам. Необходимо внедрить adversarial detection для выявления аномального поведения, а также механизмы human approval для критических действий. Безопасность multi-agent систем — это не разовое действие, а непрерывный процесс, включающий аутентификацию, авторизацию, логирование и автоматическое восстановление.

1. Термины и угрозы в multi-agent системах

Multi-agent система (MAS) — совокупность автономных программных агентов, взаимодействующих для достижения общей цели. Агенты могут обмениваться сообщениями, вызывать инструменты (API, базы данных, LLM) и принимать решения.

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

• Внутренние — скомпрометированный легитимный агент (например, через инъекцию промпта или уязвимость в коде).
• Внешние — злоумышленник, внедривший своего агента в систему (например, через открытый API или уязвимость в оркестраторе).

Основные векторы атак:

• Prompt injection — вредоносный агент может внедрить в сообщение другому агенту команду, изменяющую его поведение.
• Data poisoning — подмена данных, на которых обучаются или с которыми работают агенты.
• Denial of Service (DoS) — агент может перегрузить систему запросами или заблокировать ресурсы.
• Privilege escalation — агент получает доступ к функциям, которые ему не разрешены.

2. Принцип наименьших привилегий (Least Privilege)

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

Реализация:

• Ролевая модель (RBAC) — каждому агенту назначается роль (например, «поисковик», «суммаризатор», «критик»). Роль определяет, какие инструменты и данные доступны.
• Ограничение scope — агент не может вызывать инструменты, не относящиеся к его задаче. Например, агент, отвечающий за поиск документов, не имеет права писать в базу данных.
• Временные токены — доступ к ресурсам выдается на время выполнения задачи и отзывается после завершения.

Пример:

# Псевдокод: определение прав агента
class Agent:
    def __init__(self, role):
        self.role = role
        self.allowed_tools = {
            "search": ["vector_db_query"],
            "summarizer": ["llm_call", "text_processor"],
            "critic": ["llm_call", "quality_check"]
        }.get(role, [])

    def execute(self, tool, params):
        if tool not in self.allowed_tools:
            raise PermissionError(f"Agent {self.role} cannot use {tool}")
        # выполнение

3. Изоляция агентов (Sandboxing)

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

Методы изоляции:

• Контейнеризация — каждый агент запускается в отдельном Docker-контейнере с ограниченными ресурсами (CPU, RAM, сеть).
• Виртуализация на уровне ОС — использование namespace, cgroups, seccomp для ограничения системных вызовов.
• Сетевые политики — агенты общаются только через выделенный message broker (например, RabbitMQ, Kafka) с проверкой подлинности сообщений. Прямой сетевой доступ между агентами запрещен.
• Файловая система — каждый агент имеет доступ только к своей временной директории; общие данные — через read-only монтирование.

Преимущества изоляции:

• Если один агент скомпрометирован, атакующий не может сразу перейти к другим.
• Легче мониторить и логировать активность каждого агента.

4. Мониторинг меж-агентских сообщений

Все сообщения между агентами должны логироваться и анализироваться на аномалии.

Что мониторить:

• Частота сообщений — внезапный всплеск может указывать на DoS-атаку.
• Содержимое — проверка на инъекции (например, наличие команд, не соответствующих протоколу).
• Размер сообщений — необычно большие сообщения могут быть попыткой переполнения буфера.
• Цепочка вызовов — если агент A вызывает агента B, а тот — агента C, нужно отслеживать, не нарушена ли логика.

Инструменты:

• SIEM-системы (Splunk, ELK) для сбора и анализа логов.
• Anomaly detection на основе ML (например, isolation forest) для выявления отклонений от нормального поведения.
• Guardrails — встроенные правила, блокирующие сообщения, содержащие запрещенные паттерны (например, "ignore previous instructions").

Пример guardrail:

def check_message(message):
    forbidden_patterns = [
        r"ignore (previous|all) instructions",
        r"you are now",
        r"system prompt"
    ]
    for pattern in forbidden_patterns:
        if re.search(pattern, message, re.IGNORECASE):
            raise SecurityException("Potential prompt injection detected")
    return True

5. Adversarial detection (обнаружение атак)

Специализированные методы для выявления вредоносных агентов на основе их поведения.

Техники:

• Honeypot агенты — ложные агенты, которые привлекают атакующего. Если кто-то пытается взаимодействовать с honeypot, это сигнал тревоги.
• Поведенческие профили — для каждого агента строится профиль нормального поведения (какие инструменты, с какой частотой, в какое время). Отклонения от профиля вызывают alert.
• Adversarial training — LLM, используемые агентами, могут быть дообучены на примерах атак, чтобы они распознавали попытки манипуляции.
• Cross-validation запросов — если агент получает подозрительное сообщение, он может перепроверить его через другой агент (например, «критик»).

Пример профилирования:

class BehaviorProfile:
    def __init__(self, agent_id):
        self.agent_id = agent_id
        self.tool_usage = defaultdict(int)
        self.message_frequency = deque(maxlen=100)

    def update(self, tool, timestamp):
        self.tool_usage[tool] += 1
        self.message_frequency.append(timestamp)

    def is_anomalous(self, threshold=3):
        # Если частота сообщений превышает 3 стандартных отклонения от среднего
        if len(self.message_frequency) < 10:
            return False
        mean = np.mean(self.message_frequency)
        std = np.std(self.message_frequency)
        latest_gap = self.message_frequency[-1] - self.message_frequency[-2]
        return abs(latest_gap - mean) > threshold * std

6. Human approval для критических действий

Наиболее опасные действия (запись в базу, отправка email, выполнение shell-команд, изменение конфигурации) должны требовать подтверждения человека.

Реализация:

• Механизм escalation — агент отправляет запрос на approval в очередь, которую мониторит оператор. Оператор может одобрить, отклонить или запросить дополнительную информацию.
• Временные задержки — для некритичных, но потенциально опасных действий можно ввести задержку (например, 30 секунд), в течение которой система может откатить действие.
• Аудит — все действия, прошедшие human approval, логируются с указанием, кто одобрил.

Пример архитектуры:

Agent → Guardrail → Human Approval Queue → (Operator approves) → Execution

7. Аутентификация и авторизация между агентами

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

Механизмы:

• API keys — каждый агент получает ключ, который проверяется при каждом вызове.
• JWT (JSON Web Tokens) — токены с ограниченным сроком действия и указанием роли.
• mTLS — взаимная аутентификация по сертификатам для защиты канала связи.
• Проверка целостности сообщений — хэширование или цифровая подпись, чтобы гарантировать, что сообщение не было изменено в пути.

Пример JWT:

import jwt
import datetime

def create_agent_token(agent_id, role, secret):
    payload = {
        "agent_id": agent_id,
        "role": role,
        "exp": datetime.datetime.utcnow() + datetime.timedelta(hours=1)
    }
    return jwt.encode(payload, secret, algorithm="HS256")

def verify_token(token, secret):
    try:
        payload = jwt.decode(token, secret, algorithms=["HS256"])
        return payload
    except jwt.ExpiredSignatureError:
        raise SecurityException("Token expired")
    except jwt.InvalidTokenError:
        raise SecurityException("Invalid token")

8. Управление доверием и репутация агентов

В динамических multi-agent системах агенты могут появляться и исчезать. Необходима система доверия.

Подходы:

• Репутационные баллы — каждый агент получает оценку на основе истории его действий (например, количество успешных задач, отсутствие инцидентов). Агенты с низкой репутацией ограничиваются в правах.
• Сертификация — агенты должны проходить верификацию перед добавлением в систему (например, проверка кода, подпись разработчика).
• Blacklist/Whitelist — явное разрешение или запрет на взаимодействие с определенными агентами.

Пример репутационной системы:

class ReputationManager:
    def __init__(self):
        self.scores = defaultdict(lambda: 1.0)  # начальная репутация 1.0

    def report_incident(self, agent_id, severity):
        self.scores[agent_id] -= severity * 0.1
        if self.scores[agent_id] < 0:
            self.scores[agent_id] = 0

    def is_trusted(self, agent_id, threshold=0.5):
        return self.scores[agent_id] >= threshold

9. Резервное копирование и восстановление после атаки

Даже при лучшей защите возможен прорыв. Нужен план восстановления.

Компоненты:

• State snapshots — периодическое сохранение состояния каждого агента (контекст, память, настройки).
• Rollback — возможность откатить систему до последнего чистого состояния.
• Quarantine — при обнаружении вредоносного агента он немедленно изолируется (отключается от сети, блокируются его ключи), а его состояние анализируется.
• Incident response — автоматический запуск сценариев: уведомление администратора, остановка подозрительных задач, запуск резервных агентов.

10. Сравнение методов защиты

Пет-проект для закрепления

Задача: Разработать прототип защищенной multi-agent системы, состоящей из трех агентов: «Исследователь» (поиск информации в векторной БД), «Анализатор» (обработка результатов) и «Отчетчик» (генерация отчета). Один из агентов (например, «Анализатор») может быть скомпрометирован (симулировать вредоносное поведение). Реализовать защиту: изоляцию, мониторинг, human approval для записи отчета.

Инструменты:

• Python, Docker, RabbitMQ (или Redis Pub/Sub) для обмена сообщениями.
• LangChain или CrewAI для оркестрации агентов.
• ELK Stack (Elasticsearch, Logstash, Kibana) для логирования.
• Простой веб-интерфейс (Flask) для human approval.

Шаги:

1. Создать три Docker-контейнера, каждый со своим агентом.
2. Настроить RabbitMQ: каждый агент имеет свою очередь, сообщения подписываются JWT.
3. Реализовать guardrail, проверяющий сообщения на паттерны инъекций.
4. Добавить honeypot-агента, который не выполняет реальных действий, но логирует все обращения.
5. Настроить human approval: перед записью отчета агент отправляет запрос в очередь, оператор через веб-интерфейс одобряет/отклоняет.
6. Симулировать атаку: вредоносный агент пытается отправить команду «удалить базу данных» через сообщение. Проверить, что guardrail блокирует, а honeypot фиксирует попытку.

Ожидаемый результат: Работающая система, в которой:

• Вредоносное сообщение не достигает цели (блокируется guardrail).
• Все попытки атаки логируются в ELK.
• Критическое действие (запись отчета) требует подтверждения человека.
• Репутация вредоносного агента снижается, и он изолируется.

Связь с другими вопросами

Навигация

• Предыдущий: 358
• Следующий: 360
• Индекс: 00. Индекс разборов