k/s: Add support for describe_registry_acls tagged field

michael-redpanda · michael-redpanda · commit 817bac5f45d0 · 2025-08-20T18:41:15.000-04:00
If this new tagged field is true, then return Schema Registry resources
in the DescribeAcls response.  Will also accept SR resource types if the
new tagged field is true.

Signed-off-by: Michael Boquard &lt;michael@redpanda.com&gt;
diff --git a/src/v/kafka/server/handlers/describe_acls.cc b/src/v/kafka/server/handlers/describe_acls.cc
@@ -28,7 +28,8 @@ namespace kafka {
 static void fill_response(
   request_context& ctx,
   security::acl_binding_filter& filter,
-  describe_acls_response_data& response) {
+  describe_acls_response_data& response,
+  bool describing_registry_resource) {
     /*
      * collapse common acls by pattern
      */
@@ -47,10 +48,14 @@ static void fill_response(
         describe_acls_resource resource;
 
         // resource pattern
-        resource.type = details::to_kafka_resource_type(entry.first.resource());
+        resource.type
+          = describing_registry_resource
+              ? details::to_kafka_registry_resource_type(entry.first.resource())
+              : details::to_kafka_resource_type(entry.first.resource());
         resource.name = entry.first.name();
         resource.pattern_type = details::to_kafka_pattern_type(
           entry.first.pattern());
+        resource.registry_resource = describing_registry_resource;
 
         // acl entries
         for (auto& acl : entry.second) {
@@ -113,7 +118,7 @@ ss::future<response_ptr> describe_acls_handler::handle(
 
     try {
         auto filter = details::to_acl_binding_filter(request.data);
-        fill_response(ctx, filter, data);
+        fill_response(ctx, filter, data, request.data.describe_registry_acls);
     } catch (const details::acl_conversion_error& e) {
         vlog(klog.debug, "Error describing ACLs: {}", e.what());
         data.error_code = error_code::invalid_request;
diff --git a/src/v/kafka/server/handlers/details/security.h b/src/v/kafka/server/handlers/details/security.h
@@ -18,6 +18,9 @@
 #include "security/scram_credential.h"
 namespace kafka::details {
 
+constexpr auto sr_subject_wire_value = 100;
+constexpr auto sr_registry_wire_value = 101;
+
 using acl_conversion_error = security::acl_conversion_error;
 
 inline security::acl_principal to_acl_principal(const ss::sstring& principal) {
@@ -52,6 +55,18 @@ inline security::resource_type to_resource_type(int8_t type) {
     }
 }
 
+inline security::resource_type to_registry_resource_type(int8_t type) {
+    switch (type) {
+    case sr_subject_wire_value:
+        return security::resource_type::sr_subject;
+    case sr_registry_wire_value:
+        return security::resource_type::sr_registry;
+    default:
+        throw acl_conversion_error(
+          fmt::format("Invalid registry resource type: {}", type));
+    }
+}
+
 inline security::pattern_type to_pattern_type(int8_t type) {
     switch (type) {
     case 3:
@@ -156,7 +171,9 @@ to_resource_pattern_filter(const describe_acls_request_data& request) {
         // wildcard
         break;
     default:
-        resource_type = to_resource_type(request.resource_type);
+        resource_type = request.describe_registry_acls
+                          ? to_registry_resource_type(request.resource_type)
+                          : to_resource_type(request.resource_type);
     }
 
     std::optional<security::resource_pattern_filter::pattern_filter_type>
@@ -174,7 +191,12 @@ to_resource_pattern_filter(const describe_acls_request_data& request) {
     }
 
     return security::resource_pattern_filter(
-      resource_type, request.resource_name_filter, pattern_filter);
+      resource_type,
+      request.resource_name_filter,
+      pattern_filter,
+      request.describe_registry_acls
+        ? security::resource_pattern_filter::resource_subsystem::schema_registry
+        : security::resource_pattern_filter::resource_subsystem::kafka);
 }
 
 /*
@@ -238,7 +260,17 @@ inline int8_t to_kafka_resource_type(security::resource_type type) {
         vassert(
           false, "Schema Registry resources are not supported in kafka ACLs");
     }
-    __builtin_unreachable();
+}
+
+inline int8_t to_kafka_registry_resource_type(security::resource_type type) {
+    switch (type) {
+    case security::resource_type::sr_subject:
+        return sr_subject_wire_value;
+    case security::resource_type::sr_registry:
+        return sr_registry_wire_value;
+    default:
+        vassert(false, "Request only for Schema Registry resources");
+    }
 }
 
 inline int8_t to_kafka_pattern_type(security::pattern_type type) {
diff --git a/src/v/kafka/server/tests/BUILD b/src/v/kafka/server/tests/BUILD
@@ -934,3 +934,25 @@ redpanda_cc_btest(
         "@seastar//:testing",
     ],
 )
+
+redpanda_cc_btest(
+    name = "describe_acls_test",
+    timeout = "short",
+    srcs = [
+        "describe_acls_test.cc",
+    ],
+    env = {"RP_FIXTURE_ENV": "1"},
+    tags = ["exclusive"],
+    deps = [
+        "//src/v/cluster",
+        "//src/v/kafka/protocol",
+        "//src/v/kafka/protocol:describe_acls",
+        "//src/v/kafka/server",
+        "//src/v/model",
+        "//src/v/redpanda/tests:fixture",
+        "//src/v/security",
+        "//src/v/test_utils:seastar_boost",
+        "@seastar",
+        "@seastar//:testing",
+    ],
+)
diff --git a/src/v/kafka/server/tests/describe_acls_test.cc b/src/v/kafka/server/tests/describe_acls_test.cc
@@ -0,0 +1,164 @@
+// Copyright 2025 Redpanda Data, Inc.
+//
+// Use of this software is governed by the Business Source License
+// included in the file licenses/BSL.md
+//
+// As of the Change Date specified in that file, in accordance with
+// the Business Source License, use of this software will be governed
+// by the Apache License, Version 2.0
+
+#include "cluster/security_frontend.h"
+#include "kafka/protocol/describe_acls.h"
+#include "redpanda/tests/fixture.h"
+#include "security/acl.h"
+
+using namespace std::chrono_literals;
+
+class describe_acls_fixture : public redpanda_thread_fixture {
+protected:
+    void create_acl(security::acl_binding binding) {
+        app.controller->get_security_frontend()
+          .local()
+          .create_acls({std::move(binding)}, 5s)
+          .get();
+    }
+};
+
+FIXTURE_TEST(describe_acls_any_kafka, describe_acls_fixture) {
+    // This test verifies that when requesting any ACL and
+    // "describe_registry_acls" is false, only receive back Kafka resource ACLs
+    wait_for_controller_leadership().get();
+
+    create_acl(security::acl_binding{
+      security::resource_pattern{
+        security::resource_type::topic,
+        "test-topic",
+        security::pattern_type::literal},
+      security::acl_entry{
+        security::acl_principal{
+          security::principal_type::user, "User:test-user"},
+        security::acl_host::wildcard_host(),
+        security::acl_operation::read,
+        security::acl_permission::allow}});
+
+    create_acl(security::acl_binding{
+      security::resource_pattern{
+        security::resource_type::sr_subject,
+        "test-topic-value",
+        security::pattern_type::literal},
+      security::acl_entry{
+        security::acl_principal{
+          security::principal_type::user, "User:test-user"},
+        security::acl_host::wildcard_host(),
+        security::acl_operation::read,
+        security::acl_permission::allow}});
+
+    auto client = make_kafka_client().get();
+    auto deferred_close = ss::defer([&client] { client.stop().get(); });
+    client.connect().get();
+
+    kafka::describe_acls_request req;
+    req.data.describe_registry_acls = false;
+    req.data.resource_type = 1;
+    req.data.resource_pattern_type = 1;
+    req.data.operation = 1;
+    req.data.permission_type = 1;
+
+    auto resp = client.dispatch(std::move(req), kafka::api_version(2)).get();
+    BOOST_REQUIRE(!resp.data.errored());
+    BOOST_REQUIRE_EQUAL(resp.data.error_code, kafka::error_code::none);
+    BOOST_REQUIRE_EQUAL(resp.data.resources.size(), 1);
+    BOOST_REQUIRE_EQUAL(resp.data.resources[0].type, 2);
+    BOOST_REQUIRE_EQUAL(resp.data.resources[0].name, "test-topic");
+    BOOST_REQUIRE(!resp.data.resources[0].registry_resource);
+}
+
+FIXTURE_TEST(describe_acls_any_sr, describe_acls_fixture) {
+    // This test verifies that when requesting any ACL and
+    // "describe_registry_acls" is true, only receive back Schema Registry
+    // resource ACLs
+    wait_for_controller_leadership().get();
+
+    create_acl(security::acl_binding{
+      security::resource_pattern{
+        security::resource_type::topic,
+        "test-topic",
+        security::pattern_type::literal},
+      security::acl_entry{
+        security::acl_principal{
+          security::principal_type::user, "User:test-user"},
+        security::acl_host::wildcard_host(),
+        security::acl_operation::read,
+        security::acl_permission::allow}});
+
+    create_acl(security::acl_binding{
+      security::resource_pattern{
+        security::resource_type::sr_subject,
+        "test-topic-value",
+        security::pattern_type::literal},
+      security::acl_entry{
+        security::acl_principal{
+          security::principal_type::user, "User:test-user"},
+        security::acl_host::wildcard_host(),
+        security::acl_operation::read,
+        security::acl_permission::allow}});
+
+    auto client = make_kafka_client().get();
+    auto deferred_close = ss::defer([&client] { client.stop().get(); });
+    client.connect().get();
+
+    kafka::describe_acls_request req;
+    req.data.describe_registry_acls = true;
+    req.data.resource_type = 1;
+    req.data.resource_pattern_type = 1;
+    req.data.operation = 1;
+    req.data.permission_type = 1;
+
+    auto resp = client.dispatch(std::move(req), kafka::api_version(2)).get();
+    BOOST_REQUIRE(!resp.data.errored());
+    BOOST_REQUIRE_EQUAL(resp.data.error_code, kafka::error_code::none);
+    BOOST_REQUIRE_EQUAL(resp.data.resources.size(), 1);
+    BOOST_REQUIRE_EQUAL(resp.data.resources[0].type, 100); // 100 is SR_SUBJECT
+    BOOST_REQUIRE_EQUAL(resp.data.resources[0].name, "test-topic-value");
+    BOOST_REQUIRE(resp.data.resources[0].registry_resource);
+}
+
+FIXTURE_TEST(describe_acls_invalid_for_kafka, describe_acls_fixture) {
+    auto client = make_kafka_client().get();
+    auto deferred_close = ss::defer([&client] { client.stop().get(); });
+    client.connect().get();
+
+    kafka::describe_acls_request req;
+    req.data.describe_registry_acls = false;
+    // Request SR_SUBJECT resource type, which is invalid if
+    // describe_registry_acls is false
+    req.data.resource_type = 100;
+    req.data.resource_pattern_type = 1;
+    req.data.operation = 1;
+    req.data.permission_type = 1;
+
+    auto resp = client.dispatch(std::move(req), kafka::api_version(2)).get();
+    BOOST_REQUIRE(resp.data.errored());
+    BOOST_REQUIRE_EQUAL(
+      resp.data.error_code, kafka::error_code::invalid_request);
+}
+
+FIXTURE_TEST(describe_acls_invalid_for_sr, describe_acls_fixture) {
+    auto client = make_kafka_client().get();
+    auto deferred_close = ss::defer([&client] { client.stop().get(); });
+    client.connect().get();
+
+    kafka::describe_acls_request req;
+    req.data.describe_registry_acls = true;
+    // Request TOPIC resource type, which is invalid if describe_registry_acls
+    // is true
+    req.data.resource_type = 2;
+    req.data.resource_pattern_type = 1;
+    req.data.operation = 1;
+    req.data.permission_type = 1;
+
+    auto resp = client.dispatch(std::move(req), kafka::api_version(2)).get();
+    BOOST_REQUIRE(resp.data.errored());
+    BOOST_REQUIRE_EQUAL(
+      resp.data.error_code, kafka::error_code::invalid_request);
+}
