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 service principal is not authorized to execute the command
Caused by: Authorization denied to Describe on topic(s): [aug15]

3. Same User and KSQL authorizations denies on CSAS

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

ksql> create stream aug15_stream as select * from aug15;
The KSQL service principal is not authorized to execute the command
Caused by: Authorization denied to Write on topic(s): [AUG15_STREAM]

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 #")

[spena]

Update:

I decided to back to the initial proposal where the permissions checks are done on the DistributingExecutor instead of the RequestValidator. The problem with having the checks for KSQL in the RequetValidator is that when CREATE is used on a non-existent topic, then KSQL creates the topic for the user, and CREATE AS SELECT also creates the sink topic but in a Sandboxed environment which cannot be used to check permissions for the KSQL service principal.

I needed to check after the topic has been created to validate the User and KSQL have read/write permissions on those topics. For the KsqlResource, only 3 statements are checked: INSERT INTO, CREATE, CREATE AS SELECT. Two of them needs to be re-checked in the DistributingExecutor, so it did not make sense to have an extra check in the RequestValidator too.

[agavra]

I don't think DistributingExecutor is the right place for this logic (it has nothing to do with distributing the command). Furthermore, I think that separating it from the site of topic creation might cause us issues down the line (what if I refactor the TopicCreateInjector to be applied somewhere other than where it is now?). That being said - I think that this check should be done either inside the TopicCreateInjector or, if possible, even couple it with the KafkaTopicClient to make sure that we don't ever create a topic without later checking for permissions.

Lastly - is there any work being done by Kafka team to allow us to check whether or not we would have write/read permissions on a topic? e.g. is there a call that allows us to create (verify only) and also return the allowed permissions?

[spena]

@agavra Kafka has an API to validate if a topic can be created. KSQL already supports that: 0ea157b

However, the problem with this PR is not checking if a User can create a topic. The problem is that the User and KSQL must have Write permissions on the newly created sink topic (or Read permissions when running CREATE on a non-existent source topic). Doing it on the RequestValidator was impossible because it uses a Sandboxed environment. Kafka also has a way to check what permissions a User has on a topic, but if the topic does not exist (because it is sandboxed) then I cannot validate the right permission there.

Doing it in the KafkaTopicClient has the same challenges. The createTopic should not validate if a User has Write permissions on the topic after it is created. It should only create (which it throws an exception if the User is not authorized to create`).

The only way I found is to do those Write permissions checks after the Injector created the topics. If for some reason the TopicCreateInjector changes, then the permissions checks will fail, so the person doing that refactor will need to figure out how to fix it, right?

[agavra]

Sorry if I was unclear about the Kafka extensions - I meant if Kafka is working on extending the CreateTopicOptions to include a permissions check (R/W not just Create).

The createTopic should not validate if a User has Write permissions on the topic after it is created.

Why not? Do we ever want to create a topic we can't read and write from? I feel like the API could very well be createTopic(name, partitions, replicas, Set<Permissions> requiredPermissions). That way we can make it "look like" we never created the topic if we can't create it with those permissions. (And if Kafka ever supports a "create with permissions" API, we could switch over to use that and never create it in first place). cc @vcrfxia earlier had a suggestion similar to that.

If for some reason the TopicCreateInjector changes, then the permissions checks will fail, so the person doing that refactor will need to figure out how to fix it, right?

Yes - but that makes the code harder to iterate on and more coupled, which slows us down in the long term.

[spena]

Got it. I like your idea of validating R/W permissions with Kafka during the creating request. I'm not aware of plans to extend the API to include that, though. There is one request I will make to validate a DELETE operation (like the CREATE) too. I will request your proposal as well. However, I don't think they will add it by 5.4 where I am planning to improve these error messages.

For 5.4, I could probably add the same behavior on the KafkaTopicClient. Create the topic, then validate if it has the permissions; If not, then delete the topic again. I don't like this too much, though, because the delete could fail for any reasons (no permissions to delete, a network issue, or any other denied operation). So I don't know if we would want to extend our API if we don't have an atomic operation in Kafka for doing so. Thoughts?

Anyway, that fixes the problem when checking if a newly created topic has Write permissions. But I still need to check that the KSQL service principal has the same Write permissions which can only be done after the User has created the topic.

[agavra]

I think that makes sense - I still don't love having it in the Distributing executor, but I guess we don't have a good workaround :/