KAFKA-13646: Implement KIP-801: KRaft authorizer #11649

cmccabe · 2022-01-05T19:15:04Z

Currently, when using KRaft mode, users still have to have an Apache ZooKeeper instance if they want to use AclAuthorizer. We should have a built-in Authorizer for KRaft mode that does not depend on ZooKeeper. This PR introduces such an authorizer, called StandardAuthorizer. See KIP-801 for a full description of the new Authorizer design.

Authorizer.java: add aclCount API as described in KIP-801. StandardAuthorizer is currently the only authorizer that implements it, but eventually we may implement it for AclAuthorizer and others as well.

ControllerApis.scala: fix a bug where createPartitions was authorized using CREATE on the topic resource rather than ALTER on the topic resource as it should have been.

QuorumTestHarness: rename the controller endpoint to CONTROLLER for consistency (the brokers already called it that). This is relevant in AuthorizerIntegrationTest where we are examining endpoint names. Also add the controllerServers call.

TestUtils.scala: adapt the ACL functions to be usable from KRaft, by ensuring that they use the Authorizer from the current active controller.

BrokerMetadataPublisher.scala: add broker-side ACL application logic.

Controller.java: add ACL APIs. Also add a findAllTopicIds API in order to make junit tests that use KafkaServerTestHarness#getTopicNames and KafkaServerTestHarness#getTopicIds work smoothly.

AuthorizerIntegrationTest.scala: convert over testAuthorizationWithTopicExisting (more to come soon)

QuorumController.java: add logic for replaying ACL-based records. This means storing them in the new AclControlManager object, and integrating them into controller snapshots. It also means applying the changes in the Authorizer, if one is configured. In renounce, when reverting to a snapshot, also set newBytesSinceLastSnapshot to 0.

There are a few follow-up items not in this PR. One is actually creating the new acl count metric. Another is implementing the "start after we reach the high water mark" authorizer logic. Also, this doesn't implement the new busy error code. Finally, we need to implement logIfDenied.

metadata/src/main/java/org/apache/kafka/metadata/authorizer/ClusterMetadataAuthorizer.java

metadata/src/main/java/org/apache/kafka/controller/AclControlManager.java

hachikuji · 2022-02-07T18:29:25Z

core/src/main/scala/kafka/server/ControllerApis.scala

@@ -689,7 +689,7 @@ class ControllerApis(val requestChannel: RequestChannel,
  def handleCreatePartitions(request: RequestChannel.Request): Unit = {
    val future = createPartitions(request.body[CreatePartitionsRequest].data,
      authHelper.authorize(request.context, CREATE, CLUSTER, CLUSTER_NAME),
-      names => authHelper.filterByAuthorized(request.context, CREATE, TOPIC, names)(n => n))
+      names => authHelper.filterByAuthorized(request.context, ALTER, TOPIC, names)(n => n))


I was going to suggest pulling this fix out so that we could backport more easily, but I guess it doesn't matter since kraft never had an authorizer before.

We probably should pull out the CREATE -> ALTER fix since in theory you could be using KRaft + AclAuthorizer. If you want I can do that now (but it won't have a unit test, hope that's OK).

(It is tested by AuthorizerIntegrationTest but I wouldn't want to backport that to 3.{0,1})

core/src/main/scala/kafka/server/ControllerServer.scala

hachikuji · 2022-02-07T18:52:30Z

core/src/test/scala/unit/kafka/integration/KafkaServerTestHarness.scala

-    getController().kafkaController.controllerContext.topicNames.toMap
+    if (isKRaftTest()) {
+      val result = new util.HashMap[Uuid, String]()
+      controllerServer.controller.findAllTopicIds(Long.MaxValue).get().entrySet().forEach {


nit: I wonder if we may as well add a findAllTopicNames (or something like that) to go along with findAllTopicIds.

Hmm, I don't see why that would need to be in the controller API.

In general I wish this test harness function didn't exist (maybe we should mark it as @deprecated).

hachikuji · 2022-02-07T19:01:26Z

metadata/src/main/java/org/apache/kafka/controller/Controller.java

@@ -90,6 +91,16 @@
    CompletableFuture<Map<String, ResultOrError<Uuid>>> findTopicIds(long deadlineNs,
                                                                     Collection<String> topicNames);

+    /**
+     * Find the ids for all topic names. Note that this function should only used for


nit: only be used

hachikuji · 2022-02-07T19:11:37Z

core/src/test/scala/integration/kafka/api/AuthorizerIntegrationTest.scala

+      properties.put(KafkaConfig.AuthorizerClassNameProp, classOf[StandardAuthorizer].getName)
+      properties.put(StandardAuthorizer.SUPER_USERS_CONFIG, BrokerPrincipal.toString())
+    } else {
+      properties.put(KafkaConfig.AuthorizerClassNameProp, classOf[AclAuthorizer].getName)


Why is it not necessary to identify the broker as a super-user for AclAuthorizer?

There is a chicken / egg problem with the broker connecting to the controller quorum in the absence of ACLs. Basically the bootstrapping problem.

Making the broker principal a super user doesn't really change the test logic since the test in ZK mode just gives the broker all the permissions it needs in the setUp function, then moves on (it doesn't test cases where the broker lacks permissions)

hachikuji · 2022-02-07T21:47:15Z

metadata/src/main/java/org/apache/kafka/metadata/authorizer/StandardAuthorizer.java

+    static Set<String> getConfiguredSuperUsers(Map<String, ?> configs) {
+        Object configValue = configs.get(SUPER_USERS_CONFIG);
+        if (configValue == null) return Collections.emptySet();
+        String stringValue = configValue.toString().trim();


The logic in AclAuthorizer is a little different. We do the split first and then trim:

superUsers = configs.get(AclAuthorizer.SuperUsersProp).collect { case str: String if str.nonEmpty => str.split(";").map(s => SecurityUtils.parseKafkaPrincipal(s.trim)).toSet }.getOrElse(Set.empty[KafkaPrincipal])

OK, we can do it that way here, too. I suppose we should also call SecurityUtils.parseKafkaPrincipal just to filter out bad values.

hachikuji · 2022-02-07T22:10:43Z

metadata/src/main/java/org/apache/kafka/metadata/authorizer/StandardAuthorizerData.java

+    /**
+     * Contains all of the current ACLs sorted by (resource type, pattern type, resource name).
+     */
+    private final ConcurrentSkipListSet<StandardAcl> aclsByResource;


For a follow-up, but we will probably need to add something like AclAuthorizerBenchmark.

https://issues.apache.org/jira/browse/KAFKA-13650

metadata/src/main/java/org/apache/kafka/metadata/authorizer/StandardAuthorizerData.java

hachikuji · 2022-02-07T22:21:59Z

metadata/src/main/java/org/apache/kafka/metadata/authorizer/StandardAuthorizerData.java

+            StandardAcl acl = iterator.next();
+            if (!aclIsInCorrectSection(action, acl)) break;
+            AuthorizationResult result = findResult(action, requestContext, acl);
+            if (ALLOWED.equals(result)) {


nit: ALLOWED == result

Add logIfDenied = true to a few places where we are doing an initial check against a CLUSTER resource. Convert over the other tests in AuthorizerIntegrationTest. Add a special case for the result of deleteTopics since in KRaft mode we honor DESCRIBE permissions there, and in ZK mode we do not. In testPatternSubscriptionMatchingInternalTopic, retry subscribing. In testCreatePermissionNeededToReadFromNonExistentTopic, don't depend on ZK to check topic existence. In a few spots, check that an ACL gets propagated to all authorizers rather than just waiting for it to make it to a single authorizer. TestUtils: extend timeout of waitAndVerifyAcls from 15 seconds to 45 just to avoid rare flakes. Rename some variables to avoid shadowing. StandardAuthorizerData: add trace logging for addAcl / removeAcl. Re-arrange StandardAcl sort order so that we always step forward in the map. Fix a bug in the ACL scanning that could have led to ACLs being ignored. StandardAuthorizerTest: add two better authorization tests.

core/src/main/scala/kafka/server/KafkaApis.scala

cmccabe · 2022-02-08T21:24:32Z

Failed tests were:

ConnectionQuotasTest#testNonDefaultConnectionCountLimitAndRateLimit
DynamicBrokerReconfigurationTest#testThreadPoolResize

These tests are not related to this change (they don't deal with the new authorizer) and running them locally, they pass.

hachikuji · 2022-02-08T21:10:54Z

metadata/src/main/java/org/apache/kafka/metadata/authorizer/StandardAuthorizerData.java

+                      AuthorizationResultBuilder builder) {
+        NavigableSet<StandardAcl> headSet = aclsByResource.tailSet(exemplar, true);
+        String resourceName = action.resourcePattern().name();
+        for (Iterator<StandardAcl> iterator = headSet.iterator();


nit: I don't think we need the iterator

I really prefer to be explicit here, since we are iterating over a set that is potentially changing. I know that Java will only evaluate the y in for (x: y) {} once but I think there are some languages that don't, so no point being obscure.

metadata/src/main/java/org/apache/kafka/metadata/authorizer/ClusterMetadataAuthorizer.java

metadata/src/main/java/org/apache/kafka/metadata/authorizer/StandardAuthorizerData.java

hachikuji · 2022-02-08T22:44:28Z

metadata/src/main/java/org/apache/kafka/metadata/authorizer/StandardAuthorizerData.java

+                      StandardAcl exemplar,
+                      AuthorizableRequestContext requestContext,
+                      AuthorizationResultBuilder builder) {
+        NavigableSet<StandardAcl> headSet = aclsByResource.tailSet(exemplar, true);


Should we rename this to tailSet?

oops. hehe. fixed.

hachikuji · 2022-02-08T22:46:51Z

metadata/src/main/java/org/apache/kafka/metadata/authorizer/StandardAuthorizerData.java

+    private final AuthorizationResult defaultResult;
+
+    /**
+     * Contains all of the current ACLs sorted by (resource type, pattern type, resource name).


nit: need to update this since resource name is before pattern type now

hachikuji · 2022-02-08T22:54:00Z

metadata/src/test/java/org/apache/kafka/metadata/authorizer/StandardAclTest.java

+            AclPermissionType.DENY));
+    }
+
+    private static final int signum(int input) {


nit: final is redundant for private static method

hachikuji · 2022-02-08T23:00:22Z

metadata/src/main/java/org/apache/kafka/metadata/authorizer/StandardAuthorizerData.java

+        StandardAcl exemplar = new StandardAcl(
+            action.resourcePattern().resourceType(),
+            action.resourcePattern().name(),
+            PatternType.UNKNOWN,


Maybe worth mentioning somewhere in here that UNKNOWN pattern type is ordered before others since the initial search depends on that?

Good point. I will add a comment.

hachikuji · 2022-02-08T23:20:13Z

metadata/src/main/java/org/apache/kafka/controller/AclControlManager.java

+        List<ApiMessageAndVersion> records = new ArrayList<>(acls.size());
+        for (AclBinding acl : acls) {
+            try {
+                validateNewAcl(acl);


These validations seem to make sense. Currently they are all implemented by AclAuthorizer, so I wanted to ask whether it makes sense to pull them up. Basically are there any cases where a custom authorizer might be depending on any of the cases this validation is excluding?

I think it would make sense to pull them up eventually in a follow-up PR.

cmccabe · 2022-02-09T08:11:24Z

Test failures are not related

hachikuji

LGTM. There are a few minor improvements I noticed while reviewing again, but I will submit a separate patch for them.

…#11745) In #11649, we fixed one permission inconsistency between kraft and zk authorization for the `CreatePartitions` request. Previously kraft was requiring `CREATE` permission on the `Topic` resource when it should have required `ALTER`. A second inconsistency is that kraft was also allowing `CREATE` on the `Cluster` resource, which is not supported in zk clusters and was not documented in KIP-195: https://cwiki.apache.org/confluence/display/KAFKA/KIP-195%3A+AdminClient.createPartitions. This patch fixes this inconsistency and adds additional test coverage for both cases. Reviewers: José Armando García Sancio <jsancio@gmail.com>

cmccabe changed the title ~~KRaft authorizer~~ KRaft authorizer [WIP] Jan 6, 2022

YeonCheolGit reviewed Jan 7, 2022

View reviewed changes

metadata/src/main/java/org/apache/kafka/metadata/authorizer/ClusterMetadataAuthorizer.java Show resolved Hide resolved

YeonCheolGit reviewed Jan 7, 2022

View reviewed changes

metadata/src/main/java/org/apache/kafka/controller/AclControlManager.java Show resolved Hide resolved

cmccabe force-pushed the kraft-authorizer branch from 3888e89 to 9ee906f Compare February 2, 2022 01:00

cmccabe changed the title ~~KRaft authorizer [WIP]~~ KAFKA-13646: Implement KIP-801: KRaft authorizer Feb 4, 2022

cmccabe added the kraft label Feb 4, 2022

cmccabe force-pushed the kraft-authorizer branch from 9ee906f to 5785ab1 Compare February 4, 2022 21:22

cmccabe force-pushed the kraft-authorizer branch from 5785ab1 to 60085ea Compare February 4, 2022 21:22

KAFKA-13646: Implement KIP-801: KRaft authorizer apache#11649

a62b135

cmccabe force-pushed the kraft-authorizer branch from 60085ea to a62b135 Compare February 4, 2022 21:35

hachikuji reviewed Feb 7, 2022

View reviewed changes

core/src/main/scala/kafka/server/KafkaApis.scala Outdated Show resolved Hide resolved

cmccabe added 2 commits February 7, 2022 16:42

Address review comments (and fix handleEnvelope function name)

0f36d1c

fix checkstyle

7b32232

hachikuji reviewed Feb 8, 2022

View reviewed changes

cmccabe added 2 commits February 8, 2022 15:59

Merge branch 'trunk' into kraft-authorizer

47643a6

address review comments

b1a7526

hachikuji approved these changes Feb 9, 2022

View reviewed changes

hachikuji merged commit d35283f into apache:trunk Feb 9, 2022

hachikuji mentioned this pull request Feb 9, 2022

KAFKA-13661; Consistent permissions for CreatePartitions API #11745

Merged

3 tasks

gengweifeng mentioned this pull request Dec 29, 2022

oauth-keycloak-authorizer doesn't support kraft cluster strimzi/strimzi-kafka-oauth#178

Open

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

KAFKA-13646: Implement KIP-801: KRaft authorizer #11649

KAFKA-13646: Implement KIP-801: KRaft authorizer #11649

cmccabe commented Jan 5, 2022 •

edited

hachikuji Feb 7, 2022

cmccabe Feb 7, 2022

hachikuji Feb 7, 2022

cmccabe Feb 7, 2022 •

edited

hachikuji Feb 7, 2022

hachikuji Feb 7, 2022

cmccabe Feb 7, 2022 •

edited

hachikuji Feb 7, 2022

cmccabe Feb 8, 2022

hachikuji Feb 7, 2022

cmccabe Feb 8, 2022

hachikuji Feb 7, 2022

cmccabe commented Feb 8, 2022

hachikuji Feb 8, 2022

cmccabe Feb 9, 2022

hachikuji Feb 8, 2022

cmccabe Feb 9, 2022

hachikuji Feb 8, 2022

hachikuji Feb 8, 2022

hachikuji Feb 8, 2022

cmccabe Feb 9, 2022

hachikuji Feb 8, 2022

cmccabe Feb 9, 2022

cmccabe commented Feb 9, 2022

hachikuji left a comment

KAFKA-13646: Implement KIP-801: KRaft authorizer #11649

KAFKA-13646: Implement KIP-801: KRaft authorizer #11649

Conversation

cmccabe commented Jan 5, 2022 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

cmccabe Feb 7, 2022 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

cmccabe Feb 7, 2022 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

cmccabe commented Feb 8, 2022

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

cmccabe commented Feb 9, 2022

hachikuji left a comment

Choose a reason for hiding this comment

cmccabe commented Jan 5, 2022 •

edited

cmccabe Feb 7, 2022 •

edited

cmccabe Feb 7, 2022 •

edited