Агент Apache Kafka

Агент Kafka — авторизатор API Authorizer Kafka, указываемый в authorizer.class.name: брокеры и контроллеры спрашивают его о каждом запросе, и он отвечает по подписанным политикам сервиса данных kafka. Дистрибутив поставляет его одним совместимым с Java 8 JAR-файлом agents/kafka/grantforge-agent-kafka-<version>.jar с перемещёнными внутрь зависимостями.

Интеграционные тесты Testcontainers в репозитории запускают JAR внутри Kafka 3.7.2, 3.9.2, 4.1.2 и 4.3.1 на узлах KRaft, которые одновременно являются брокером и контроллером. Это не сертифицирует все остальные версии. Плагин сервиса описан на странице Apache Kafka.

Как это работает

• Проверяется каждый запрос. Для каждого ресурса, которого касается запрос, Kafka спрашивает об операции ACL; агент отвечает по политикам, где типы ресурсов Kafka служат уровнями ресурсов, а её операции — типами доступа. Отказ доходит до клиента собственной ошибкой Kafka, например TOPIC_AUTHORIZATION_FAILED.
• Идемпотентному продюсеру достаточно топика. Kafka позволяет продюсеру запуститься без idempotent_write на кластере, если он может писать в какой-либо топик, и спрашивает об этом авторизатор. Агент отвечает по самим политикам: разрешённый write на любом топике засчитывается, а запрет — только в политике, которая охватывает все топики.
• Политики видны как ACL. kafka-acls --list и describeAcls Admin API показывают действующие сейчас политики: значение без подстановочных знаков — как литеральный ACL, значение вроде orders* — как ACL с префиксом, а * — как подстановочный знак Kafka; пользователи — как принципалы User:, группы — Group:, роли — Role:, группа public — User:*. То, что ACL выразить не могут, опускается: другие подстановочные знаки, исключённые значения, элементы с условиями и элементы политики, у которой есть исключения из них. Создание и удаление ACL отклоняются: the ACLs of this cluster are GrantForge policies; change them in GrantForge.
• Решения сохраняются. Брокер задаёт одни и те же вопросы при каждой записи и чтении; агент хранит решения текущего снимка, по умолчанию 100 000 штук в течение 10 секунд. Новый снимок начинает с чистого листа, поэтому изменение политики действует, как только агент его получает.
• Повторяющиеся обращения объединяются. Одинаковые разрешённые обращения в пределах окна, по умолчанию 10 секунд, передаются одним событием доступа с их числом и временем последнего; консоль показывает Одинаковых обращений: N. Обращения одинаковы, если у них совпадают пользователь, адрес клиента, ресурс, тип доступа, тип запроса и решающая политика. Отказы передаются по одному и сразу. Что стоит передавать, выбирает Kafka: например, она не просит сообщать о топиках, которые пропускает в списке.
• Суперпользователи не проверяются. Принципалы из super.users могут всё, как и с собственным авторизатором Kafka, и не попадают в аудит.

Развёртывание

1. Добавьте сервис данных kafka и настройте его политики. Откройте Права доступа к данным → Агенты, выберите сервис и выпустите токен агента. Он показывается один раз; сохраните его в файл, доступный для чтения только учётной записи Kafka.

2. Скопируйте JAR в libs каждого брокера и контроллера, например /opt/kafka/libs.

3. Добавьте параметры в server.properties каждого узла, заменив примеры принципалов, адреса, путей и имени экземпляра:

   authorizer.class.name=org.devlive.grantforge.kafka.agent.GrantForgeKafkaAuthorizer
   super.users=User:kafka
   grantforge.server.url=https://grantforge.example.com
   grantforge.token.file=/etc/kafka/grantforge.token
   grantforge.instance=broker-1
   grantforge.cache.dir=/var/lib/kafka/grantforge-cache

   super.users должен называть принципалов, которыми брокеры и контроллеры пользуются между собой, например принципала межброкерного слушателя, иначе политики должны разрешать им то, что они делают, например cluster_action на кластере. Подключения к слушателю PLAINTEXT идут как User:ANONYMOUS; указывайте его, только если ни один клиент не может достучаться до такого слушателя. Каталог кэша должен быть доступен для записи учётной записи Kafka.

4. Перезапускайте брокеры и контроллеры по одному. Слушатели, кроме слушателей контроллеров, открываются, когда у агента появляется первый снимок, но не позже grantforge.start.timeout.ms; до этого, а также всякий раз, когда снимка нет, запросы отклоняются.

5. Сначала перенесите существующие ACL: ACL, сохранённые собственным авторизатором Kafka, перестают действовать, когда агент его заменяет. Заранее выведите их через kafka-acls --list и создайте соответствующие политики.

Агент читает параметры из конфигурации, которую Kafka передаёт своему авторизатору. Узел, который одновременно является брокером и контроллером, создаёт авторизатор для каждой роли; оба используют один агент.

Пользователи, группы и роли

Политики называют принципала, которого Kafka присваивает подключению: имя пользователя SASL, короткое имя принципала Kerberos по sasl.kerberos.principal.to.local.rules или имя сертификата TLS по ssl.principal.mapping.rules. В Kafka нет групп и ролей; в политиках это группы и роли, которые GrantForge назначает пользователю.

Решения и ограничения

• Без пригодного снимка отклоняется всё, кроме запросов суперпользователей. Неиспользуемый сервис оставляет Kafka без проверок, как без авторизатора.
• Токен, привязанный к сервису, тип которого не kafka, отклоняет всё.
• Запрет типа доступа запрещает и то, что он подразумевает: запрет write на топике запрещает describe, поэтому пользователь его не видит. Собственные ACL Kafka здесь ведут себя иначе.
• Вопрос Kafka, может ли пользователь действовать с каким-либо ресурсом типа, не является обращением и не передаётся.
• Решения хранятся grantforge.decision.cache.ttl.ms, поэтому период действия политики начинается или заканчивается с задержкой до этого значения.
• Агент работает в брокерах и контроллерах. REST API Kafka Connect, реестры схем и другие сервисы вокруг Kafka не охвачены.

Связанные руководства

• Apache Kafka
• Сервисы данных, политики и агенты