Skip to content

Do not enable TopicAccessValidator if Kafka has unsupported authorizedOperations #2888

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
Sergio Peña (spena) merged 7 commits into from
May 29, 2019
Merged

Do not enable TopicAccessValidator if Kafka has unsupported authorizedOperations #2888

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
c9a44e6
If authorizedOperations is NULL, then allow access to the Topic
spena May 28, 2019
2195332
Return Dummy validator if Cluster authorizedOperations is Null
spena May 28, 2019
318e70e
Merge branch '5.3.x' into kafka_2.2_authorized
spena May 29, 2019
2a60a5b
Switch message from info to warn
spena May 29, 2019
1ef65b1
Use dummy validator for tests
spena May 29, 2019
5ab8735
Merge branch '5.3.x' into kafka_2.2_authorized
spena May 29, 2019
1ac3033
Remove unecessary comment from tests
spena May 29, 2019
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 5 additions & 1 deletion ksql-engine/src/main/java/io/confluent/ksql/engine/AuthorizationTopicAccessValidator.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,6 +33,8 @@
/**
* Checks if a {@link ServiceContext} has access to the source and target topics of transient
* and persistent query statements.
* </p>
* This validator only works on Kakfa 2.3 or later.
*/
public class AuthorizationTopicAccessValidator implements TopicAccessValidator {
@Override
Expand Down Expand Up @@ -121,7 +123,9 @@ private void checkAccess(
final Set<AclOperation> authorizedOperations = serviceContext.getTopicClient()
.describeTopic(topicName).authorizedOperations();

if (!authorizedOperations.contains(operation)) {
// Kakfa 2.2 or lower do not support authorizedOperations(). In case of running on a
// unsupported broker version, then the authorizeOperation will be null.
if (authorizedOperations != null && !authorizedOperations.contains(operation)) {
// This error message is similar to what Kafka throws when it cannot access the topic
// due to an authorization error. I used this message to keep a consistent message.
throw new KsqlException(String.format(
Expand Down
20 changes: 15 additions & 5 deletions ksql-engine/src/main/java/io/confluent/ksql/engine/TopicAccessValidatorFactory.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -37,9 +37,17 @@ public static TopicAccessValidator create(
final ServiceContext serviceContext,
final MetaStore metaStore
) {
if (isKafkaAuthorizerEnabled(serviceContext.getAdminClient())) {
LOG.info("KSQL topic authorization checks enabled.");
return new AuthorizationTopicAccessValidator();
final AdminClient adminClient = serviceContext.getAdminClient();

if (isKafkaAuthorizerEnabled(adminClient)) {
if (KafkaClusterUtil.isAuthorizedOperationsSupported(adminClient)) {
LOG.info("KSQL topic authorization checks enabled.");
return new AuthorizationTopicAccessValidator();
}

LOG.info("The Kafka broker has an authorization service enabled, but the Kafka "

@rodesai Rohan (rodesai) May 28, 2019 •
edited
Loading

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think we should log this at warning level. The resulting behavior is likely not what the user intended. Might be good to get Michael Drogalis (@MichaelDrogalis) take on what the behavior should be - maybe it would be better to just hard-fail and require the user to either upgrade Kafka or not configure the validator.

@spena Sergio Peña (spena) May 29, 2019

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The topic access validator is used mostly to provide better permission error messages when the KSQL user won't be able to execute a persistent query on KSQL. I don't think a warning message is a correct level for that case (I am not against about it, though). However, failing KSQL because is not using the correct version of Kafka might be out of scope of our current release. Do we have a compatibility list of Kafka versions that KSQL should support?

@MichaelDrogalis Michael Drogalis (MichaelDrogalis) May 29, 2019

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Do we have a compatibility list of Kafka versions that KSQL should support?
Not that I'm aware of, no.

I agree with Sergio Peña (@spena) that it's probably a bit late to introduce a code-path to hard-fail here -- though switching the log level up to warn does feel like the right play to me. Sergio Peña (@spena) Can you expand a bit about why you think info is right here? Maybe I'm missing some nuance.

@spena Sergio Peña (spena) May 29, 2019

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I feel it's not a warn because if permission checks are not available, KSQL will still fail further in the code without providing good error messages. However, it is harmless to switch the level to warn now that I am thinking about it. I will do the switch and update this PR.

+ "version does not support authorizedOperations(). "
+ "KSQL topic authorization checks will not be enabled.");
}

// Dummy validator if a Kafka authorizer is not enabled
Expand All @@ -62,8 +70,10 @@ private static boolean isKafkaAuthorizerEnabled(final AdminClient adminClient) {
if (e.getCause() instanceof ClusterAuthorizationException) {
return true;
}
}

return false;
// Throw the unknown exception to avoid leaving the Server unsecured if a different
// error was thrown
throw e;
}
}
}
15 changes: 15 additions & 0 deletions ksql-engine/src/main/java/io/confluent/ksql/services/KafkaClusterUtil.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,8 +21,11 @@
import java.util.Collection;
import java.util.Collections;
import java.util.Map;

import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.Config;
import org.apache.kafka.clients.admin.DescribeClusterOptions;
import org.apache.kafka.clients.admin.DescribeClusterResult;
import org.apache.kafka.common.Node;
import org.apache.kafka.common.config.ConfigResource;
import org.slf4j.Logger;
Expand All @@ -35,6 +38,18 @@ private KafkaClusterUtil() {

}

public static boolean isAuthorizedOperationsSupported(final AdminClient adminClient) {
try {
final DescribeClusterResult authorizedOperations = adminClient.describeCluster(
new DescribeClusterOptions().includeAuthorizedOperations(true)
);

return authorizedOperations.authorizedOperations().get() != null;
} catch (Exception e) {

@agavra Almog Gavra (agavra) May 29, 2019

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

do we need to check for ClusterAuthorizationException like we do above?

@spena Sergio Peña (spena) May 29, 2019

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I don't think so. By the time this method is called, the TopicAccessValidatorFactory already checked if the user is authorized to access the cluster.

throw new KsqlServerException("Could not get Kafka authorized operations!", e);
}
}

public static Config getConfig(final AdminClient adminClient) {
try {
final Collection<Node> brokers = adminClient.describeCluster().nodes().get();
Expand Down
15 changes: 15 additions & 0 deletions ...-engine/src/test/java/io/confluent/ksql/engine/AuthorizationTopicAccessValidatorTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -97,6 +97,21 @@ private Statement givenStatement(final String sql) {
return ksqlEngine.prepare(ksqlEngine.parse(sql).get(0)).getStatement();
}

@Test
public void shouldAllowAnyOperationIfPermissionsAreNull() {
// This test case verifies permissions will not work if the Kafka broker returns NULL
Comment thread
spena marked this conversation as resolved.
Outdated

// Given:
givenTopicPermissions(TOPIC_1, null);
final Statement statement = givenStatement("SELECT * FROM " + STREAM_TOPIC_1 + ";");

// When:
accessValidator.validate(serviceContext, metaStore, statement);

// Then:
// Above command should not throw any exception
}

@Test
public void shouldSingleSelectWithReadPermissionsAllowed() {
// Given:
Expand Down
35 changes: 30 additions & 5 deletions ksql-engine/src/test/java/io/confluent/ksql/engine/TopicAccessValidatorFactoryTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,10 +15,12 @@

package io.confluent.ksql.engine;

import static org.easymock.EasyMock.anyObject;
import static org.easymock.EasyMock.expect;
import static org.easymock.EasyMock.mock;
import static org.easymock.EasyMock.replay;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.Matchers.any;
import static org.hamcrest.Matchers.instanceOf;
import static org.hamcrest.Matchers.is;
import static org.hamcrest.Matchers.not;
Expand All @@ -29,13 +31,17 @@
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Set;

import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.Config;
import org.apache.kafka.clients.admin.ConfigEntry;
import org.apache.kafka.clients.admin.DescribeClusterOptions;
import org.apache.kafka.clients.admin.DescribeClusterResult;
import org.apache.kafka.clients.admin.DescribeConfigsResult;
import org.apache.kafka.common.KafkaFuture;
import org.apache.kafka.common.Node;
import org.apache.kafka.common.acl.AclOperation;
import org.apache.kafka.common.config.ConfigResource;
import org.easymock.EasyMock;
import org.easymock.EasyMockRunner;
Expand Down Expand Up @@ -66,7 +72,7 @@ public void setUp() {
@Test
public void shouldReturnAuthorizationValidator() {
// Given:
givenKafkaAuthorizer("an-authorizer-class");
givenKafkaAuthorizer("an-authorizer-class", Collections.emptySet());

// When:
final TopicAccessValidator validator = TopicAccessValidatorFactory.create(serviceContext, null);
Expand All @@ -78,7 +84,19 @@ public void shouldReturnAuthorizationValidator() {
@Test
public void shouldReturnDummyValidator() {
// Given:
givenKafkaAuthorizer("");
givenKafkaAuthorizer("", Collections.emptySet());

// When:
final TopicAccessValidator validator = TopicAccessValidatorFactory.create(serviceContext, null);

// Then
assertThat(validator, not(instanceOf(AuthorizationTopicAccessValidator.class)));
}

@Test
public void shouldReturnDummyValidatorIfAuthorizedOperationsReturnNull() {
// Given:
givenKafkaAuthorizer("an-authorizer-class", null);

// When:
final TopicAccessValidator validator = TopicAccessValidatorFactory.create(serviceContext, null);
Expand All @@ -87,8 +105,13 @@ public void shouldReturnDummyValidator() {
assertThat(validator, not(instanceOf(AuthorizationTopicAccessValidator.class)));
}

private void givenKafkaAuthorizer(final String className) {
expect(adminClient.describeCluster()).andReturn(describeClusterResult());
private void givenKafkaAuthorizer(
final String className,
final Set<AclOperation> authOperations
) {
expect(adminClient.describeCluster()).andReturn(describeClusterResult(authOperations));
expect(adminClient.describeCluster(anyObject()))
.andReturn(describeClusterResult(authOperations));
expect(adminClient.describeConfigs(describeBrokerRequest()))
.andReturn(describeBrokerResult(Collections.singletonList(
new ConfigEntry(KAFKA_AUTHORIZER_CLASS_NAME, className)
Expand All @@ -97,10 +120,12 @@ private void givenKafkaAuthorizer(final String className) {
replay(adminClient);
}

private DescribeClusterResult describeClusterResult() {
private DescribeClusterResult describeClusterResult(final Set<AclOperation> authOperations) {
final Collection<Node> nodes = Collections.singletonList(node);
final DescribeClusterResult describeClusterResult = EasyMock.mock(DescribeClusterResult.class);
expect(describeClusterResult.nodes()).andReturn(KafkaFuture.completedFuture(nodes));
expect(describeClusterResult.authorizedOperations())
.andReturn(KafkaFuture.completedFuture(authOperations));
replay(describeClusterResult);
return describeClusterResult;
}
Expand Down