Config apis ACLs

src/v/kafka/server/handlers/alter_configs.cc

@@ -172,6 +172,10 @@ ss::future<response_ptr> alter_configs_handler::handle(
     auto groupped = group_alter_config_resources(
       std::move(request.data.resources));
 
+    auto unauthorized_responsens = authorize_alter_config_resources<
+      alter_configs_resource,
+      alter_configs_resource_response>(ctx, groupped);
+
     std::vector<ss::future<std::vector<alter_configs_resource_response>>>
       futures;
     futures.reserve(2);
@@ -181,6 +185,8 @@ ss::future<response_ptr> alter_configs_handler::handle(
       alter_broker_configuartion(std::move(groupped.broker_changes)));
 
     auto ret = co_await ss::when_all_succeed(futures.begin(), futures.end());
+    // include authorization errors
+    ret.push_back(std::move(unauthorized_responsens));
 
     co_return co_await ctx.respond(
       assemble_alter_config_response<

src/v/kafka/server/handlers/configs/config_utils.h

@@ -20,6 +20,7 @@
 #include "kafka/server/request_context.h"
 #include "kafka/types.h"
 #include "outcome.h"
+#include "security/acl.h"
 
 #include <seastar/core/coroutine.hh>
 #include <seastar/core/sstring.hh>
@@ -69,6 +70,61 @@ T make_error_alter_config_resource_response(
       .resource_type = resource.resource_type,
       .resource_name = resource.resource_name};
 }
+/**
+ * Authorizes groupped alter configuration resources, it returns not authorized
+ * responsens and modifies passed in group_resources<T>
+ */
+template<typename T, typename R>
+std::vector<R> authorize_alter_config_resources(
+  request_context& ctx, groupped_resources<T>& to_authorize) {
+    std::vector<R> not_authorized;
+    /**
+     * Check broker configuration authorization
+     */
+    if (
+      !to_authorize.broker_changes.empty()
+      && !ctx.authorized(
+        security::acl_operation::alter_configs,
+        security::default_cluster_name)) {
+        // not allowed
+        std::transform(
+          to_authorize.broker_changes.begin(),
+          to_authorize.broker_changes.end(),
+          std::back_inserter(not_authorized),
+          [](T& res) {
+              return make_error_alter_config_resource_response<R>(
+                res, error_code::cluster_authorization_failed);
+          });
+        // all broker changes have to be dropped
+        to_authorize.broker_changes.clear();
+    }
+
+    /**
+     * Check topic configuration authorization
+     */
+    auto unauthorized_it = std::partition(
+      to_authorize.topic_changes.begin(),
+      to_authorize.topic_changes.end(),
+      [&ctx](const T& res) {
+          return ctx.authorized(
+            security::acl_operation::alter_configs,
+            model::topic(res.resource_name));
+      });
+
+    std::transform(
+      unauthorized_it,
+      to_authorize.topic_changes.end(),
+      std::back_inserter(not_authorized),
+      [](T& res) {
+          return make_error_alter_config_resource_response<R>(
+            res, error_code::topic_authorization_failed);
+      });
+
+    to_authorize.topic_changes.erase(
+      unauthorized_it, to_authorize.topic_changes.end());
+
+    return not_authorized;
+}
 
 template<typename T, typename R, typename Func>
 ss::future<std::vector<R>> do_alter_topics_configuration(

src/v/kafka/server/handlers/describe_configs.cc

@@ -21,6 +21,7 @@
 #include "model/metadata.h"
 #include "model/namespace.h"
 #include "model/validation.h"
+#include "security/acl.h"
 #include "ssx/sformat.h"
 
 #include <seastar/core/do_with.hh>
@@ -303,6 +304,9 @@ ss::future<response_ptr> describe_configs_handler::handle(
 
     describe_configs_response response;
     response.data.results.reserve(request.data.resources.size());
+    bool cluster_authorized = ctx.authorized(
+      security::acl_operation::describe_configs,
+      security::default_cluster_name);
 
     for (auto& resource : request.data.resources) {
         response.data.results.push_back(describe_configs_result{
@@ -329,6 +333,12 @@ ss::future<response_ptr> describe_configs_handler::handle(
                 result.error_code = error_code::unknown_topic_or_partition;
                 continue;
             }
+
+            if (!ctx.authorized(
+                  security::acl_operation::describe_configs, topic.tp)) {
+                result.error_code = error_code::topic_authorization_failed;
+                continue;
+            }
             /**
              * Redpanda extensions
              */
@@ -407,6 +417,10 @@ ss::future<response_ptr> describe_configs_handler::handle(
         }
 
         case config_resource_type::broker:
+            if (!cluster_authorized) {
+                result.error_code = error_code::cluster_authorization_failed;
+                continue;
+            }
             report_broker_config(result, request.data.include_synonyms);
             break;
 

src/v/kafka/server/handlers/incremental_alter_configs.cc

@@ -218,6 +218,10 @@ ss::future<response_ptr> incremental_alter_configs_handler::handle(
     auto groupped = group_alter_config_resources(
       std::move(request.data.resources));
 
+    auto unauthorized_responsens = authorize_alter_config_resources<
+      incremental_alter_configs_resource,
+      resp_resource_t>(ctx, groupped);
+
     std::vector<ss::future<std::vector<resp_resource_t>>> futures;
     futures.reserve(2);
     futures.push_back(alter_topic_configuration(
@@ -226,6 +230,8 @@ ss::future<response_ptr> incremental_alter_configs_handler::handle(
       alter_broker_configuartion(std::move(groupped.broker_changes)));
 
     auto ret = co_await ss::when_all_succeed(futures.begin(), futures.end());
+    // include authorization errors
+    ret.push_back(std::move(unauthorized_responsens));
 
     co_return co_await ctx.respond(assemble_alter_config_response<
                                    incremental_alter_configs_response,