sr/context: implement /contexts/{context}/... prefixed handlers

src/v/pandaproxy/schema_registry/BUILD

@@ -82,6 +82,7 @@ redpanda_cc_library(
         "avro.h",
         "compatibility.h",
         "configuration.h",
+        "context_router.h",
         "error.h",
         "errors.h",
         "exceptions.h",
@@ -251,6 +252,7 @@ redpanda_cc_library(
         ":types",
         "//src/v/config:startup_config",
         "//src/v/kafka/client",
+        "//src/v/utils:variant",
     ],
 )
 

src/v/pandaproxy/schema_registry/auth.h

@@ -15,6 +15,7 @@
 #include "pandaproxy/schema_registry/types.h"
 #include "pandaproxy/server.h"
 #include "security/acl.h"
+#include "utils/variant.h"
 
 #include <variant>
 
@@ -42,10 +43,19 @@ class auth {
     // AuthZ is required to be performed in the handler as the resource is
     // unknown
     using deferred = named_type<std::monostate, class deferred_tag>;
+    // AuthZ will be performed against the context-qualified subject extracted
+    // from both the {context} and {subject} path parameters
+    using context_prefix_subject
+      = named_type<std::monostate, class context_prefix_subject_tag>;
 
     using op = security::acl_operation;
+    /// Authorization-time resource type.
     using resource
       = std::variant<none, deferred, global, context_subject, cluster>;
+    /// Route-registration-time resource type — includes
+    /// `context_prefix_subject`, which is resolved to `context_subject` before
+    /// authorization.
+    using route_resource = extend_variant_t<resource, context_prefix_subject>;
 
     using regular_function_handler = ss::noncopyable_function<
       ss::future<server::reply_t>(server::request_t, server::reply_t)>;
@@ -55,14 +65,14 @@ class auth {
     using function_handler
       = std::variant<regular_function_handler, deferred_function_handler>;
 
-    auth(level lvl, std::optional<op> op, resource res)
+    auth(level lvl, std::optional<op> op, route_resource res)
       : _lvl{lvl}
       , _op{op}
       , _res{std::move(res)} {}
 
     level get_level() const { return _lvl; }
     std::optional<op> get_op() const { return _op; }
-    const resource& get_resource() const { return _res; }
+    const route_resource& get_resource() const { return _res; }
     bool is_deferred() const {
         return std::holds_alternative<auth::deferred>(get_resource());
     }
@@ -77,7 +87,7 @@ class auth {
 private:
     level _lvl;
     std::optional<op> _op;
-    resource _res;
+    route_resource _res;
 };
 
 } // namespace pandaproxy::schema_registry

src/v/pandaproxy/schema_registry/authorization.cc

@@ -13,6 +13,7 @@
 #include "container/chunked_hash_map.h"
 #include "pandaproxy/api/api-doc/schema_registry.json.hh"
 #include "pandaproxy/parsing/httpd.h"
+#include "pandaproxy/schema_registry/context_router.h"
 #include "pandaproxy/schema_registry/service.h"
 #include "pandaproxy/schema_registry/sharded_store.h"
 #include "pandaproxy/schema_registry/types.h"
@@ -54,15 +55,22 @@ namespace {
 
 auth::resource
 extract_resource_from_request(const server::request_t& rq, const auth& auth) {
-    auto resource = auth.get_resource();
-    ss::visit(
-      resource,
-      [&rq](context_subject& ctx_sub) {
-          ctx_sub = context_subject::from_string(
+    return ss::visit(
+      auth.get_resource(),
+      [&rq](const context_subject&) -> auth::resource {
+          return context_subject::from_string(
             parse::request_param<ss::sstring>(*rq.req, "subject"));
       },
-      [](const auto&) {});
-    return resource;
+      [&rq](const auth::context_prefix_subject&) -> auth::resource {
+          auto ctx = normalize_context(
+            parse::request_param<ss::sstring>(*rq.req, "context"));
+          auto sub = parse::request_param<ss::sstring>(*rq.req, "subject");
+          if (!starts_with_context(sub)) {
+              sub = fmt::format(":{}:{}", ctx, sub);
+          }
+          return context_subject::from_string(sub);
+      },
+      [](const auto& res) -> auth::resource { return res; });
 }
 
 void throw_unauthorized() {

src/v/pandaproxy/schema_registry/context_router.h

@@ -0,0 +1,98 @@
+// Copyright 2026 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
+
+#pragma once
+
+#include "base/seastarx.h"
+#include "pandaproxy/schema_registry/errors.h"
+
+#include <seastar/core/sstring.hh>
+#include <seastar/http/request.hh>
+
+#include <fmt/format.h>
+
+#include <string_view>
+
+namespace pandaproxy::schema_registry {
+
+/// \brief Normalize a context name from a URL path parameter.
+inline ss::sstring normalize_context(std::string_view ctx) {
+    if (ctx.starts_with(':')) {
+        ctx.remove_prefix(1);
+    }
+
+    if (ctx.ends_with(':')) {
+        ctx.remove_suffix(1);
+    }
+
+    if (ctx.find(':') != std::string_view::npos) {
+        throw as_exception(context_invalid(ctx));
+    }
+
+    if (!ctx.starts_with('.')) {
+        return {fmt::format(".{}", ctx)};
+    }
+    return ss::sstring(ctx);
+}
+
+/// \brief Check if a string already has a context prefix.
+inline bool starts_with_context(std::string_view s) {
+    return s.starts_with(":.") || s.starts_with(":*:");
+}
+
+/// \brief Scope the "subject" path parameter by prepending the context.
+///
+/// ctx is expected to be in the form ".name" (with leading dot). The
+/// resulting subject is ":.ctx:subject".
+inline void
+scope_subject_param(ss::http::request& req, const ss::sstring& ctx) {
+    auto sub = req.get_path_param("subject");
+    if (!starts_with_context(sub)) {
+        auto nctx = normalize_context(ctx);
+        req.param.set(
+          ss::sstring("subject"),
+          ss::sstring(fmt::format("/:{0}:{1}", nctx, sub)));
+    }
+}
+
+/// \brief Inject or prepend context into the "subject" query parameter.
+inline void
+scope_subject_query(ss::http::request& req, const ss::sstring& ctx) {
+    auto nctx = normalize_context(ctx);
+    auto existing = req.get_query_param("subject");
+    if (existing.empty()) {
+        req.set_query_param("subject", fmt::format(":{0}:", nctx));
+    } else if (!starts_with_context(existing)) {
+        req.set_query_param("subject", fmt::format(":{0}:{1}", nctx, existing));
+    }
+}
+
+/// \brief Inject or prepend context into the "subjectPrefix" query parameter.
+inline void
+scope_subject_prefix_query(ss::http::request& req, const ss::sstring& ctx) {
+    auto nctx = normalize_context(ctx);
+    auto existing = req.get_query_param("subjectPrefix");
+    if (existing.empty()) {
+        req.set_query_param("subjectPrefix", fmt::format(":{0}:", nctx));
+    } else if (!starts_with_context(existing)) {
+        req.set_query_param(
+          "subjectPrefix", fmt::format(":{0}:{1}", nctx, existing));
+    }
+}
+
+/// \brief Inject the context as a context-only qualified subject path
+/// parameter.
+inline void
+inject_context_as_subject(ss::http::request& req, const ss::sstring& ctx) {
+    auto nctx = normalize_context(ctx);
+    req.param.set(
+      ss::sstring("subject"), ss::sstring(fmt::format("/:{0}:", nctx)));
+}
+
+} // namespace pandaproxy::schema_registry

src/v/pandaproxy/schema_registry/error.cc

@@ -96,6 +96,8 @@ struct error_category final : std::error_category {
             return "The specified context is not empty";
         case error_code::subject_invalid:
             return "The specified subject is not valid";
+        case error_code::context_invalid:
+            return "The specified context name is not valid";
         }
         return "(unrecognized error)";
     }
@@ -164,6 +166,8 @@ struct error_category final : std::error_category {
             return reply_error_code::context_not_empty; // 42211
         case error_code::subject_invalid:
             return reply_error_code::subject_invalid; // 42208
+        case error_code::context_invalid:
+            return reply_error_code::bad_request; // 400
         }
         return {};
     }

src/v/pandaproxy/schema_registry/error.h

@@ -47,6 +47,7 @@ enum class error_code {
     writes_disabled,
     context_not_empty,
     subject_invalid,
+    context_invalid,
 };
 
 std::error_code make_error_code(error_code);

src/v/pandaproxy/schema_registry/errors.h

@@ -261,6 +261,12 @@ inline error_info context_not_empty(const context& ctx) {
       fmt::format("The specified context '{}' is not empty.", ctx())};
 }
 
+inline error_info context_invalid(std::string_view ctx) {
+    return error_info{
+      error_code::context_invalid,
+      fmt::format("The specified context '{}' is not valid.", ctx)};
+}
+
 inline bool failed_subject_schema_lookup(std::error_code ec) {
     return ec == error_code::subject_not_found
            || ec == error_code::subject_version_not_found;

src/v/pandaproxy/schema_registry/handlers.cc

@@ -20,6 +20,7 @@
 #include "pandaproxy/logger.h"
 #include "pandaproxy/parsing/httpd.h"
 #include "pandaproxy/schema_registry/authorization.h"
+#include "pandaproxy/schema_registry/context_router.h"
 #include "pandaproxy/schema_registry/error.h"
 #include "pandaproxy/schema_registry/errors.h"
 #include "pandaproxy/schema_registry/exceptions.h"
@@ -1514,7 +1515,7 @@ delete_context(server::request_t rq, server::reply_t rp) {
     parse_accept_header(rq, rp);
 
     auto ctx_str = parse::request_param<ss::sstring>(*rq.req, "context");
-    auto ctx = context{ctx_str};
+    auto ctx = context{normalize_context(ctx_str)};
 
     if (ctx == default_context) {
         throw as_exception(