Настроить AdmissionController для проверки прав агента перед вызовом tool

ТЕХНИЧЕСКОЕ ЗАДАНИЕ: Настроить AdmissionController для проверки прав агента перед вызовом tool

1. Цель задачи

Реализовать механизм авторизации вызовов tool'ов агентом на основе политик доступа. Вы научитесь проектировать простую RBAC (role‑based control|access control) модель для LLM‑агента, разрабатывать AdmissionControllermiddleware, проверяющий права перед выполнением tool’а, и интегрировать его в конвейер вызова. Ключевой результат при попытке агента вызвать tool, на который у него нет прав, вызов блокируется и возвращается понятная ошибка, а в логах фиксируется инцидент.

2. Исходные данные

Что нужноОткуда взять
LLM‑агент с набором tool’овНапишите сами (минимум 3 tool: read_file, write_file, execute_command)
Роли агентов (admin, user, guest) и привязка прав к ролямОпределите в YAML‑файле политик
AdmissionControllerРазработать в рамках задачи
Среда выполнения (Python 3.10+, виртуальное окружение)Локально

Если нет реального агента — симулируем:

  1. Создайте простой скрипт, который эмулирует агента: читает JSON‑запрос с полями agent_role и tool_name.
  2. Напишите YAML‑файл policies.yaml с правилами доступа:
    roles:
      admin:
        tools: ["read_file", "write_file", "execute_command"]
      user:
        tools: ["read_file"]
      guest:
        tools: []
    
  3. Убедитесь, что агент передаёт свою роль в контексте вызова.

3. Технологический стек

КомпонентИнструментыНазначение
Язык программированияPython 3.10+Реализация AdmissionController и тестов
Фреймворк агентаLangChain (опционально)Удобная обёртка для tool’ов
AdmissionControllerFastAPI (как отдельный микросервис)Принимает запрос, проверяет права, возвращает решение
Формат политикYAML (PyYAML)Хранение RBAC‑правил
ТестированиеpytestПроверка блокировки и разрешения
ЛогированиеPython logging + JSON‑форматФиксация отказов
КонтейнеризацияDocker (опционально)Изоляция микросервиса

4. Этапы выполнения

Этап 1: Проектирование политик и ролей (30 минут)

Действия

  1. Определите роли агентов (минимум 3) и набор tool’ов для каждой роли.
  2. Создайте файл policies.yaml:
    roles:
      admin:
        tools: ["read_file", "write_file", "execute_command"]
      user:
        tools: ["read_file"]
      guest:
        tools: []
    default_policy: deny  # если роль не определена
    
  3. Напишите функцию load_policies(path: str) -> dict, которая загружает и парсит YAML.
  4. Добавьте валидацию: вызовет исключение, если роль или tool отсутствуют в политиках.

Ожидаемый результат этапа Файл policies.yaml и функции загрузки/валидации.

Этап 2: Разработка AdmissionController (1.5 часа)

Действия

  1. Реализуйте класс AdmissionController:
    class AdmissionController:
        def __init__(self, policies_path: str):
            self.policies = self._load_policies(policies_path)
    
        def _load_policies(self, path: str) -> dict:
            with open(path) as f:
                return yaml.safe_load(f)
    
        def check(self, agent_role: str, tool_name: str) -> bool:
            role_config = self.policies["roles"].get(agent_role)
            if role_config is None:
                return False  # deny by default
            allowed_tools = role_config.get("tools", [])
            return tool_name in allowed_tools
    
  2. Добавьте логирование: если check вернул False, записать в лог событие {"event": "admission_denied", "role": agent_role, "tool": tool_name, "timestamp": ...}.
  3. Оформите AdmissionController как микросервис на FastAPI (один endpoint POST /check):
    @app.post("/check")
    async def check_access(request: AccessRequest):
        allowed = controller.check(request.agent_role, request.tool_name)
        return {"allowed": allowed}
    
    Pydantic-модель AccessRequest: agent_role: str, tool_name: str, request_id: str.
  4. Напишите Dockerfile для сервиса (опционально).

Ожидаемый результат этапа Работающий микросервис AdmissionController с endpoint /check.

Этап 3: Интеграция с агентом (1 час)

Действия

  1. Создайте функцию agent_call_tool(agent_role: str, tool_name: str, **kwargs), которая:
    • отправляет запрос к AdmissionController (HTTP POST).
    • если ответ allowed == False — поднимает исключение PermissionDeniedError или возвращает специальное сообщение.
    • если разрешено — вызывает настоящий tool.
  2. Реализуйте три tool’а для демонстрации:
    • read_file(path) — эмулирует чтение (просто возвращает содержимое).
    • write_file(path, content) — эмулирует запись.
    • execute_command(cmd) — эмулирует выполнение (только логирует, ничего не делает на самом деле).
  3. Напишите скрипт demo.py, который создаёт агента с ролью user и пытается вызвать execute_command. Вы должны получить блокировку.

Ожидаемый результат этапа Агент интегрирован с AdmissionController; при отсутствии прав вызов блокируется.

Этап 4: Тестирование (45 минут)

Действия

  1. Напишите pytest тесты:
    def test_admin_allowed():
        assert controller.check("admin", "execute_command") == True
    
    def test_user_denied():
        assert controller.check("user", "execute_command") == False
    
    def test_unknown_role():
        assert controller.check("hacker", "read_file") == False
    
    def test_unknown_tool():
        assert controller.check("admin", "unknown_tool") == False
    
    def test_integration_via_http():
        response = client.post("/check", json={"agent_role": "guest", "tool_name": "read_file"})
        assert response.json()["allowed"] == False
    
  2. Проверьте, что сервис корректно обрабатывает edge‑cases: пустая роль, None, отсутствие политики.
  3. Запустите тесты и добейтесь 100% прохождения.

Ожидаемый результат этапа Все тесты проходят, coverage > 90%.

Этап 5: Мониторинг и логирование (30 минут)

Действия

  1. Настройте логирование в формате JSON (можно использовать python-json-logger).
  2. Напишите обработчик, который при отказе в доступе дополнительно отправляет метрику (например, через Prometheus client, но можно просто увеличивать счётчик в памяти).
  3. Добавьте endpoint /metrics для prometheus, экспортирующий admission_denied_total.

Ожидаемый результат этапа Логи в JSON, метрики доступны на /metrics.

5. Критерии приемки (Definition of Done)

  • AdmissionController запускается как отдельный сервис (на локальной машине или в Docker).
  • При запросе tool’а, разрешённого для роли агента, вызов происходит успешно.
  • При запросе запрещённого tool’а вызов возвращает HTTP 403 или эквивалентную ошибку с JSON {"error": "permission_denied", ...}.
  • Логирование отказов содержит роли, tool и request ID.
  • Политики хранятся в YAML и могут быть изменены без перезапуска сервиса (если реализована горячая перезагрузка по HTTP‑сигналу или watch файла — бонус).
  • Тесты покрывают минимум 5 сценариев (разрешение, запрет, неизвестная роль, неизвестный tool, интеграция через HTTP).
  • Код содержит обработку ошибок (несуществующий файл политик, некорректный JSON запрос).
  • Документирована процедура запуска (README с одним абзацем).
  • Опционально: Docker Compose для сервиса.

6. Ожидаемый результат

Основной артефакт репозиторий со следующей структурой:

admission-controller/
├── policies.yaml
├── controller.py (класс AdmissionController)
├── main.py (FastAPI-сервер)
├── agent.py (эмуляция агента с интеграцией)
├── demo.py (скрипт для проверки)
├── tests/
│   ├── test_controller.py
│   └── test_integration.py
├── requirements.txt
├── Dockerfile
└── README.md

Содержание полный код AdmissionController с поддержкой YAML-политик, FastAPI-сервером, тестами и демо-скриптом. Дополнительно: лог-файл с записью нескольких отказов, метрики (если реализованы).

7. Возможные сложности и их решение

СложностьРешение
AdmissionController становится узким местом (latency при каждом вызове)Добавить кеширование решений (с TTL, например, 5 секунд) или использовать асинхронную проверку.
Политики меняются, сервис должен подхватывать изменения без restartРеализовать механизм релоада: watch файла политик (watchdog) или endpoint /reload.
Агент не передаёт свою рольВ реальной системе роль агента берётся из JWT-токена или session контекста. В задаче — просто передавать в запросе.
Сложность отладки при большом количестве tool’овЛогировать все проверки (разрешения и отказы) с request ID.
Различия в форматах ошибок между agent и controllerОпределить единый контракт (Pydantic models) и тестировать его.

8. Бюджет времени (оценка)

ЭтапВремя
Этап 1: Проектирование политик30 мин
Этап 2: Разработка AdmissionController1.5 ч
Этап 3: Интеграция с агентом1 ч
Этап 4: Тестирование45 мин
Этап 5: Мониторинг и логирование30 мин
Итого4 ч 15 мин

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

9. Связанные вопросы из базы знаний

ВопросТема
10Политики доступа в LLM-системах
15Авторизация вызовов tool (call)
42Архитектурный паттерн AdmissionController
103RBAC в AI-агентах
201Логирование отказов в микросервисах
305Тестирование безопасности tool’ов
410Производительность middleware при авторизации
555Интеграция FastAPI с YAML-конфигами
612Unit-тестирование решений контроля доступа
789Механизмы горячей перезагрузки конфигурации

10. Чек-лист самопроверки

  • Я реализовал класс AdmissionController, который корректно применяет политики.
  • Я настроил FastAPI-сервис и протестировал его через curl или requests.
  • Я написал минимум 5 pytest-тестов, и все они проходят.
  • Я проверил сценарий, когда агент с ролью guest пытается вызвать execute_command — получает отказ.
  • Я убедился, что логирование отказов записывается с role, tool и request ID.
  • Я задокументировал запуск в README (одна команда docker run или uvicorn main:app).