From 9b7174ae200c619cd0753f06146c73f4f5e1a389 Mon Sep 17 00:00:00 2001
From: Alexey Biryukov <alexey@redpanda.com>
Date: Wed, 6 Jul 2022 18:06:31 -0400
Subject: [PATCH] kafka: fixed: find tran coordinator was not ACL verified

The code to handle FindCoordinator request for transaction coordinator type
appeared before the caller is checked for authorization for this operation
against the ACL. Now the chech has been moved before any other handling.
---
 .../kafka/server/handlers/find_coordinator.cc | 30 +++++++++----------
 1 file changed, 15 insertions(+), 15 deletions(-)

diff --git a/src/v/kafka/server/handlers/find_coordinator.cc b/src/v/kafka/server/handlers/find_coordinator.cc
index e10100e028f1..3147c07f1eae 100644
--- a/src/v/kafka/server/handlers/find_coordinator.cc
+++ b/src/v/kafka/server/handlers/find_coordinator.cc
@@ -72,6 +72,21 @@ ss::future<response_ptr> find_coordinator_handler::handle(
     find_coordinator_request request;
     request.decode(ctx.reader(), ctx.header().version);
 
+    if (request.data.key_type == coordinator_type::group) {
+        if (!ctx.authorized(
+              security::acl_operation::describe, group_id(request.data.key))) {
+            return ctx.respond(find_coordinator_response(
+              error_code::group_authorization_failed));
+        }
+    } else if (request.data.key_type == coordinator_type::transaction) {
+        if (!ctx.authorized(
+              security::acl_operation::describe,
+              transactional_id(request.data.key))) {
+            return ctx.respond(find_coordinator_response(
+              error_code::transactional_id_authorization_failed));
+        }
+    }
+
     if (request.data.key_type == coordinator_type::transaction) {
         if (!ctx.are_transactions_enabled()) {
             return ctx.respond(
@@ -98,21 +113,6 @@ ss::future<response_ptr> find_coordinator_handler::handle(
           find_coordinator_response(error_code::unsupported_version));
     }
 
-    if (request.data.key_type == coordinator_type::group) {
-        if (!ctx.authorized(
-              security::acl_operation::describe, group_id(request.data.key))) {
-            return ctx.respond(find_coordinator_response(
-              error_code::group_authorization_failed));
-        }
-    } else if (request.data.key_type == coordinator_type::transaction) {
-        if (!ctx.authorized(
-              security::acl_operation::describe,
-              transactional_id(request.data.key))) {
-            return ctx.respond(find_coordinator_response(
-              error_code::transactional_id_authorization_failed));
-        }
-    }
-
     return ss::do_with(
       std::move(ctx),
       [request = std::move(request)](request_context& ctx) mutable {