feat: perform topic permission checks for KSQL service principal

[spena]

Description

KSQL performs topic permission checks for the CLI user requesting to execute a command. This returns correct authorization error messages if the CLI user does not have access to the statement resources.

However, some commands require that the KSQL server executes them in the background. This PR to perform those permissions checks for the KSQL server as well.

The following output messages are displayed:

1. User does not have READ permissions on a topic

ksql> create stream aug15(id int) with (kafka_topic='aug15', value_format='json');
Authorization denied to Read on topic(s): [aug15]

2. KSQL does not have READ permissions on a topic

ksql> create stream aug15(id int) with (kafka_topic='aug15', value_format='json');
The KSQL server cannot execute the given command.
Caused by: Authorization denied to Describe on topic(s): [aug15]

3. Same User and KSQL authorizations denies but with CSAS

ksql> create stream aug15_stream as select * from aug15;
Authorization denied to Create on topic(s): [AUG15_STREAM]

ksql> create stream aug15_stream as select * from aug15;
The KSQL server cannot execute the given command.
Caused by: Authorization denied to Describe on topic(s): [AUG15_STREAM]

4. Fix authorization error message when executing INSERT INTO VALUES

ksql> insert into aug15(id) values (2);
Failed to insert values into stream/table: AUG15
Caused by: Authorization denied to Write on topic(s): [aug15]

Testing done

• Added unit tests
• Verified manually (See above output messages)

Reviewer checklist

• Ensure docs are updated if necessary. (eg. if a user visible feature is being added or changed).
• Ensure relevant issues are linked (description should include text like "Fixes #")

[vcrfxia]

Thanks @spena -- haven't gotten a chance to look at the actual code changes yet but I wonder if we can clarify the error messages to more directly call out whether it's the KSQL server user or actual user that's missing permissions. For example, if I ran a command and just saw the error

The KSQL server cannot execute the given command.
Caused by: Authorization denied to Describe on topic(s): [AUG15_STREAM]

or

Authorization denied to Create on topic(s): [AUG15_STREAM]

without having read your PR description I wouldn't know that one of these specifically means that the KSQL server user doesn't have the necessary permissions to execute the statement, whereas the other means that I (the user of KSQL) don't have the necessary permissions. Could we update the message to be more explicit? For example, Authorization for User:bob to Describe on topic(s): [AUG15_STREAM] or Authorization denied for KSQL server user (User:ksql) to Describe on topics(s): [AUG15_STREAM] or something similar?

[spena]

@vcrfxia Yes, I'd like to do that. Do you think I can do a follow-up PR for that?

I started thinking about how I could get the username of both the Client and the Server, and this is how:

1. The client is obtained on the REST endpoint and pass it up to the KafkaTopicClientImpl
2. The server is obtained from the Kafka Config and pass it up to the kafkaTopicClientImpl

So 2 different places where to get them. Plus, I need to differentiate between users, either is a Client user or a Server user so the KafkaTopicClientImpl can display the right message.

This is the code that I am trying to include the username: https://github.com/confluentinc/ksql/blob/master/ksql-engine/src/main/java/io/confluent/ksql/services/KafkaTopicClientImpl.java#L120

I think I can create a KsqlSecurityContext where I can pass the username and the usertype. For instance, KsqlSecurityContext.of("sergio", UserType.CLIENT_USER), so the KsqlAuthorizationException knows the type of user and can display if the error is from the KSQL server or not. This change will impact unit tests from the ServiceContext, KafkaTopicClient, etc. Better another PR?

[vcrfxia]

Thanks @spena , I'm fine with improving the error messages in a follow-up PR as you've suggested. Thanks for looking into that!

Left a question inline about the code changes in this PR.

[vcrfxia]

Why are we moving this check from the RequestValidator to the executor? It feels like RequestValidator is the right place for it, since it's validating user permissions.

[spena]

I forgot to add a comment in the PR to explain this. But the reason is that performing the permission check for KSQL in a CSAS fails because KSQL does not have DESCRIBE permissions on the sink topic.

The RequestValidator creates the CSAS sink topic just before checking the permissions. This sink topic is created by the Client user. The creation, though, is sandboxed. This means is just a fake topic creation that gives WRITE/READ permissions to the Client user. But, it does not for KSQL, so KSQL fails with authorization denied. Even giving permissions in Kafka for KSQL fails in that part of the code.

The best place I thought was to find the non-fake injector that creates the topic in Kafka. I found it here, in the DistributingExecutor, just before persisting the command. So, now it made sense to me that the permissions checks for the Client and KSQL users are performed before persisting the command and after the User has created the sink topics. The checks should be done on non-sandboxed environments.

[vcrfxia]

Is it always the case that if the (client) user creates a sink topic via CSAS, then the KSQL user will have permissions for the topic as well? If so, we can update the sandbox to not only give READ/WRITE permissions to the client user for the sink topic, but also to the KSQL user (within the sandbox).

Or if the KSQL server user does not always automatically get permissions for a sink topic when it is created by the client user, what determines when the KSQL server user does or doesn't get permissions?

[spena]

It's the 2nd case the KSQL server user does not always automatically get permissions for a sink topic when it is created by the client user.

In 5.3 and next 5.4 releases, it is up to the client user (or an Admin) to also grant permissions to the KSQL server user to access the sink topic. KSQL does not have the ability to grant permissions to itself yet. These permissions checks for the KSQL server will give more feedback to the client user about why the command execution failed.

[vcrfxia]

In 5.3 and next 5.4 releases, it is up to the client user (or an Admin) to also grant permissions to the KSQL server user to access the sink topic.

If this is the case, then can't we check whether the KSQL server user has permissions to access the sink topic in the RequestValidator? If the topic does not currently exist, then the KSQL server user will not have permissions to access it once it is created. If the topic does currently exist, then the SandboxedKafkaTopicClient can delegate the verification to the actual SandboxedKafkaTopicClient.

[apurvam]

What default permissions does the SandboxedAdminClient grant? Just read and write on created topics? My understanding was that this sandbox should have no real effect, ie. all topic creates and deletes going through it are ignored. So why should the default permissions granted by the client matter?

[vcrfxia]

My understanding was that this sandbox should have no real effect, ie. all topic creates and deletes going through it are ignored. So why should the default permissions granted by the client matter?

The sandboxed clients are used by the request validator to validates requests, i.e., if a statement executes successfully in the sandbox, then the statement passes validation and goes on to be executed. If we have a perfect sandbox, then all requests that succeed in the sandbox will succeed outside the sandbox and vice versa. But if the sandbox is imperfect, e.g., if the sandbox grants read and write permissions on created topics when this does not happen in reality, then some (CSAS) statements will succeed in the sandbox but fail out the sandbox, which is bad since those statements would pass validation but fail partway through execution.

Ideally, we have a perfect sandbox and continue to validate statements by executing them in the sandbox. However, the tl;dr of my conversation with @spena (which continued offline as well) is that creating this perfect sandbox is difficult as there's no easy interface to check whether a user has create permissions for a topic. @spena is following up to see how others perform this check, since it feels it should be a relatively common pattern.

[spena]

Seems we don't have an option to check if a user has CREATE permissions.

We could use describeAcls if we use Kafka ACLs only, but if Kafka uses an external authorization service, like RBAC, then it won't return the permissions.

I synced with other teams, and they depend on Kafka denying or allowing the createTopic operation, which is what KSQL is currently doing it. And Kafka does not have plans to integrate the describeAcls with external authorization libraries.

Our only option is to keep this PR as it is. If Kafka allows the user to create the sink topic, but the permission checks fail with KSQL not having WRITE permissions, then the sink topic will be created even if the CSAS command fail.

[vcrfxia]

Seems we don't have an option to check if a user has CREATE permissions.

Then we'll have to do it in the executor by simply trying to create the topic, as you said. @agavra made the suggestion that we have the KafkaTopicClientImpl log when it creates topics, so users have a way to confirm whether and when topics were created by KSQL, to help mitigate potential confusion around KSQL creating topics on commands that subsequently fail (e.g., because either the user or KSQL server user does not have WRITE permissions on the newly created topic).

Regarding the check for whether the user and KSQL server user have WRITE permissions on the sink topic, is it possible to check that piece within the validator, or is it not possible because the topic needs to exist first?

[spena]

It's not possible to do it in the validator.

Kafka has two ways to check for permissions:

1. With describeAcls. This requires the user to have Describe permissions on the Cluster. This does not return ACLs handled by an external authorization provider, like RBAC.
2. With describeTopic. This returns ACLs and RBAC permissions merged into the returned TopicDescription. However, this requires the topic to exists, so we cannot check for CREATE.

So far, we can only keep the check in the executor, at least for 5.4. If Kafka decides to support RBAC in describeAcls, for instance.

[spena]

@vcrfxia I decided to put back the topic validator on the RequestValidator now that I understand the reason to do it there. Like, validate all statements have the right ACLs in Kafka before executing them.

I added the KSQL permission check, and now the output looks like this in case the KSQL service principal does not have permissions:

The KSQL service principal is not authorized to execute the command: Authorization denied to Read on topic(s): [t1]

There are 2 things left:

1. Check Write permissions on the sink topic when is created in the DistributingExecutor.
2. Check if the user has Create permissions on the SandboxedKafkaTopicClient

I wanted to add the Write permissions checks on the Sink topic in the DistributingExecutor in this patch, but there is duplicated code I would add (like the RequestValidator checkTopicPermissions). I have better plans to improve this code later when including the SR permission checks, so I will fix this in a follow-up PR. Note that this Write permission issue is already in 5.3 so thi PR is not leaving a regression.

And second, I will add the Create permissions check in another PR too. I'd like to merge this PR which just adds a permission check for KSQL, and fix the bugs in others PR.

[spena]

I had to append the e.getMessage() here because KSQL does not print the root cause message when a KsqlStatementException is detected. I didn't want to fix that because I don't know the reason of not including the whole stack. Better fix it later.

[spena]

This PR is replaced by #3261