Add RBAC message discarding filters

magnusg-kdab · magnusg-kdab · commit 73a842cc4d27 · 2026-07-13T15:40:11.000+02:00
Signed-off-by: Magnus Groß &lt;magnus.gross+github@kdab.com&gt;
diff --git a/src/client/include/Client.hpp b/src/client/include/Client.hpp
@@ -244,7 +244,7 @@ class SubscriptionClient : public MDClientBase {
     }
 
     void set(const URI<STRICT> & /*uri*/, std::string_view, const std::span<const std::byte> & /*request*/) override {
-        throw std::logic_error("get not implemented");
+        throw std::logic_error("set not implemented");
     }
 
     void subscribe(const URI<STRICT> &uri, std::string_view /*reqId*/) override {
@@ -399,18 +399,20 @@ class MDClientCtx : public ClientBase {
                 _requests.erase(0); // todo: lookup correct subscription
             }
         }
-        sendCmd(cmd.topic, cmd.command, req_id, cmd.data);
+        sendCmd(cmd.topic, cmd.command, req_id, cmd.data, cmd.rbac);
     }
 
 private:
-    void sendCmd(const URI<STRICT> &uri, mdp::Command commandType, std::size_t req_id, IoBuffer data = {}) const {
+    void sendCmd(const URI<STRICT> &uri, mdp::Command commandType, std::size_t req_id, IoBuffer data = {}, IoBuffer rbac = {}) const {
         const bool        isSet = commandType == mdp::Command::Set;
         zmq::MessageFrame cmdType{ std::string{ static_cast<char>(commandType) } };
         cmdType.send(_control_socket_send, ZMQ_SNDMORE).assertSuccess();
         zmq::MessageFrame reqId{ std::to_string(req_id) };
         reqId.send(_control_socket_send, ZMQ_SNDMORE).assertSuccess();
         zmq::MessageFrame endpoint{ std::string(uri.str()) };
         endpoint.send(_control_socket_send, isSet ? ZMQ_SNDMORE : 0).assertSuccess();
+        zmq::MessageFrame rbacframe{ std::move(rbac) };
+        rbacframe.send(_control_socket_send, 0).assertSuccess();
         if (isSet) {
             zmq::MessageFrame dataframe{ std::move(data) };
             dataframe.send(_control_socket_send, 0).assertSuccess();
@@ -421,13 +423,19 @@ class MDClientCtx : public ClientBase {
         zmq::MessageFrame cmd;
         zmq::MessageFrame reqId;
         zmq::MessageFrame endpoint;
+        zmq::MessageFrame rbac;
         while (cmd.receive(_control_socket_recv, ZMQ_DONTWAIT).isValid()) {
             if (!reqId.receive(_control_socket_recv, ZMQ_DONTWAIT).isValid()) {
                 throw std::logic_error("invalid request received: failure receiving message");
             }
             if (!endpoint.receive(_control_socket_recv, ZMQ_DONTWAIT).isValid()) {
                 throw std::logic_error("invalid request received: invalid message contents");
             }
+            if (!rbac.receive(_control_socket_recv, ZMQ_DONTWAIT).isValid()) {
+                throw std::logic_error("invalid request received: invalid rbac data");
+            }
+            auto        rbacdata = as_bytes(std::span(rbac.data().data(), rbac.data().size()));
+            IoBuffer    rbacData{ reinterpret_cast<const char *>(rbacdata.data()), rbacdata.size() };
             URI<STRICT> uri{ std::string(endpoint.data()) };
             auto       &client = getClient(uri);
             if (cmd.data().size() != 1) {
diff --git a/src/majordomo/include/majordomo/Broker.hpp b/src/majordomo/include/majordomo/Broker.hpp
@@ -280,6 +280,7 @@ class Broker {
     bool                                                    _connectedToDns    = false;
 
     const IoBuffer                                          _rbac              = IoBuffer("RBAC=ADMIN,abcdef12345");
+    std::function<void(BrokerMessage &)>                    _rbacHandler;
 
     std::atomic<bool>                                       _shutdownRequested = false;
 
@@ -422,6 +423,10 @@ class Broker {
         _subscriptionMatcher.addFilter<Filter>(key);
     }
 
+    void setRbacHandler(decltype(_rbacHandler) handler) {
+        _rbacHandler = std::move(handler);
+    }
+
     /**
      * Bind broker to endpoint, can call this multiple times. We use a single
      * socket for both clients and workers.
@@ -855,6 +860,10 @@ class Broker {
             auto clientMessage = std::move(client.requests.front());
             client.requests.pop_front();
 
+            if (_rbacHandler) {
+                _rbacHandler(clientMessage);
+            }
+
             if (auto service = bestMatchingService(clientMessage.serviceName)) {
                 if (service->internalHandler) {
                     auto reply = service->internalHandler(std::move(clientMessage));
diff --git a/src/majordomo/include/majordomo/Worker.hpp b/src/majordomo/include/majordomo/Worker.hpp
@@ -98,6 +98,7 @@ class BasicWorker {
     std::shared_mutex                                        _notificationHandlersLock;
     const std::string                                        _notifyAddress;
     const IoBuffer                                           _defaultRbacToken = IoBuffer("RBAC=NONE,");
+    std::function<bool(const IoBuffer &rbac)>                _rbacPredicate;
 
 public:
     static constexpr std::string_view name = serviceName.data();
@@ -184,6 +185,10 @@ class BasicWorker {
         disconnect();
     }
 
+    void setRbacPredicate(decltype(_rbacPredicate) rbacPredicate) {
+        _rbacPredicate = std::move(rbacPredicate);
+    }
+
 protected:
     void setHandler(std::function<void(RequestContext &)> handler) {
         _handler = std::move(handler);
@@ -352,6 +357,12 @@ class BasicWorker {
         const auto clientRole = parse_rbac::role(request.rbac.asString());
         const auto permission = _permissionsByRole.at(clientRole, _defaultPermission);
 
+        if (_rbacPredicate && !_rbacPredicate(request.rbac)) {
+            auto errorReply  = replyFromRequest(request);
+            errorReply.error = "request rejected by RBAC handler";
+            return errorReply;
+        }
+
         if (request.command == mdp::Command::Get && !(permission == Permission::RW || permission == Permission::RO)) {
             auto errorReply  = replyFromRequest(request);
             errorReply.error = std::format("GET access denied to role '{}'", clientRole);
diff --git a/src/majordomo/test/rbac_tests.cpp b/src/majordomo/test/rbac_tests.cpp
@@ -1,3 +1,5 @@
+#include <Client.hpp>
+#include <concepts/majordomo/helpers.hpp>
 #include <majordomo/Rbac.hpp>
 
 #include <catch2/catch.hpp>
@@ -30,3 +32,142 @@ TEST_CASE("RBAC parser tests", "[rbac][parsing]") {
     STATIC_REQUIRE2(parse_rbac::roleAndHash("RBAC=ADMIN,hash") == std::pair("ADMIN"sv, "hash"sv));
     STATIC_REQUIRE2(parse_rbac::hash("RBAC=ADMIN,hash,invalidHash") == "hash,invalidHash"); // TODO in java, this throws, should we throw/return "", too?
 }
+
+namespace opencmw {
+
+struct RbacContext {
+    int dummy;
+};
+
+struct RbacInput {
+    int dummy;
+};
+
+struct RbacOutput {
+    int i;
+};
+
+} // namespace opencmw
+
+ENABLE_REFLECTION_FOR(opencmw::RbacContext, dummy)
+ENABLE_REFLECTION_FOR(opencmw::RbacInput, dummy)
+ENABLE_REFLECTION_FOR(opencmw::RbacOutput, i)
+
+namespace {
+using RbacWorkerType = majordomo::Worker<"/rbac", opencmw::RbacContext, opencmw::RbacInput, opencmw::RbacOutput, majordomo::description<"Rbac test">>;
+};
+
+class RbacWorker : public RbacWorkerType {
+public:
+    template<typename BrokerType>
+    explicit RbacWorker(const BrokerType &broker) : RbacWorkerType(broker, {}) { init(); };
+
+private:
+    void init() {
+        RbacWorkerType::setCallback([this](const majordomo::RequestContext &rawCtx, const opencmw::RbacContext &context, const opencmw::RbacInput &in, opencmw::RbacContext &replyContext, opencmw::RbacOutput &out) {
+            out.i = 42;
+        });
+    };
+};
+
+TEST_CASE("RBAC basic predicate tests", "[rbac][predicate]") {
+    opencmw::majordomo::Broker<> broker{ "/Broker", {} };
+    RbacWorker                   worker{ broker };
+
+    worker.setRbacPredicate([](auto &) { return false; });
+
+    REQUIRE(broker.bind(opencmw::URI<>("mds://127.0.0.1:12345")));
+    REQUIRE(broker.bind(opencmw::URI<>("mdp://127.0.0.1:12346")));
+
+    RunInThread brokerThread(broker);
+    RunInThread workerThread(worker);
+
+    REQUIRE(waitUntilWorkerServiceAvailable(broker.context, worker));
+
+    std::vector<std::unique_ptr<opencmw::client::ClientBase>> clients;
+    clients.emplace_back(std::make_unique<opencmw::client::MDClientCtx>(broker.context, 20ms, ""));
+    opencmw::client::ClientContext client{ std::move(clients) };
+
+    opencmw::IoBuffer              inBuf;
+    opencmw::RbacInput             in;
+    opencmw::serialise<opencmw::YaS>(inBuf, in);
+
+    std::condition_variable cv;
+    std::mutex              m;
+    bool                    done = false;
+
+    client.set(opencmw::URI("mdp://127.0.0.1:12346/rbac"), [&](const mdp::Message &reply) {
+        REQUIRE(reply.error.size());
+
+        worker.setRbacPredicate([](auto& rb){
+                return true;});
+
+        opencmw::IoBuffer iBuf;
+        opencmw::serialise<opencmw::YaS>(iBuf, in);
+        client.set(opencmw::URI("mdp://127.0.0.1:12346/rbac"), [&](const mdp::Message &reply) {
+                auto outBuf = reply.data;
+                REQUIRE(reply.error.empty());
+                REQUIRE(outBuf.size());
+                opencmw::RbacOutput out;
+                opencmw::deserialise<opencmw::YaS, opencmw::ProtocolCheck::IGNORE>(outBuf, out);
+                REQUIRE(out.i == 42);
+
+                std::lock_guard lk(m);
+                done = true;
+                cv.notify_one();
+        }, std::move(iBuf)); }, std::move(inBuf));
+
+    std::unique_lock lk(m);
+    cv.wait(lk, [&] { return done; });
+}
+
+TEST_CASE("RBAC predicate tests", "[rbac][predicate]") {
+    opencmw::majordomo::Broker<> broker{ "/Broker", {} };
+    RbacWorker                   worker{ broker };
+
+    broker.setRbacHandler([](auto &msg) { msg.rbac.put("RBAC=USER,hash"); });
+    worker.setRbacPredicate([](auto &buf) { return buf.asString().contains("ADMIN"); });
+
+    REQUIRE(broker.bind(opencmw::URI<>("mds://127.0.0.1:12345")));
+    REQUIRE(broker.bind(opencmw::URI<>("mdp://127.0.0.1:12346")));
+
+    RunInThread brokerThread(broker);
+    RunInThread workerThread(worker);
+
+    REQUIRE(waitUntilWorkerServiceAvailable(broker.context, worker));
+
+    std::vector<std::unique_ptr<opencmw::client::ClientBase>> clients;
+    clients.emplace_back(std::make_unique<opencmw::client::MDClientCtx>(broker.context, 20ms, ""));
+    opencmw::client::ClientContext client{ std::move(clients) };
+
+    opencmw::IoBuffer              inBuf;
+    opencmw::RbacInput             in;
+    opencmw::serialise<opencmw::YaS>(inBuf, in);
+
+    std::condition_variable cv;
+    std::mutex              m;
+    bool                    done = false;
+
+    client.set(opencmw::URI("mdp://127.0.0.1:12346/rbac"), [&](const mdp::Message &reply) {
+        REQUIRE(reply.error.size());
+
+        broker.setRbacHandler([](auto& msg){msg.rbac.put("RBAC=ADMIN,hash");});
+
+        opencmw::IoBuffer iBuf;
+        opencmw::serialise<opencmw::YaS>(iBuf, in);
+        client.set(opencmw::URI("mdp://127.0.0.1:12346/rbac"), [&](const mdp::Message &reply) {
+                auto outBuf = reply.data;
+                REQUIRE_MESSAGE(reply.error.empty(), reply.error);
+                REQUIRE(outBuf.size());
+                opencmw::RbacOutput out;
+                opencmw::deserialise<opencmw::YaS, opencmw::ProtocolCheck::IGNORE>(outBuf, out);
+                REQUIRE(out.i == 42);
+
+                std::lock_guard lk(m);
+                done = true;
+                cv.notify_one();
+        }, std::move(iBuf)); }, std::move(inBuf));
+
+    std::unique_lock lk(m);
+    cv.wait(lk, [&] { return done; });
+}
diff --git a/src/services/include/services/KeyStore.hpp b/src/services/include/services/KeyStore.hpp
@@ -56,6 +56,14 @@ class KeystoreWorker : public KeystoreWorkerType {
         _keys[keyHash(roleKey.key)] = roleKey;
     }
 
+    void addRoles(const std::string &publicKey, const std::string &roles) {
+        const auto hash = keyHash(publicKey);
+        if (_keys.contains(hash)) {
+            auto &key = _keys.at(hash);
+            key.role  = roles;
+        }
+    }
+
 private:
     void init() {
         KeystoreWorkerType::setCallback([this](const majordomo::RequestContext &rawCtx, const KeystoreContext &context, const KeystoreInput &in, KeystoreContext &replyContext, KeystoreOutput &out) {
diff --git a/src/services/include/services/OAuthClient.hpp b/src/services/include/services/OAuthClient.hpp
@@ -324,6 +324,9 @@ class OAuthWorker : public OAuthWorkerType {
 
                 // get RBAC roles
                 out.roles = _client.getAssignedRoles(out.accessToken);
+                if (!in.publicKey.empty()) {
+                    _keystore.addRoles(in.publicKey, out.roles);
+                }
             }
         });
         _keystoreThread = std::thread([this] { _keystore.run(); });
diff --git a/src/services/test/OAuthClient_tests.cpp b/src/services/test/OAuthClient_tests.cpp
@@ -6,6 +6,7 @@
 
 #include <Client.hpp>
 #include <concepts/majordomo/helpers.hpp>
+#include <Debug.hpp>
 #include <majordomo/Rbac.hpp>
 #include <services/OAuthClient.hpp>
 
@@ -128,6 +129,7 @@ TEST_CASE("Worker test", "[OAuth]") {
     client.set(opencmw::URI("mdp://127.0.0.1:12346/oauth"), [&](const mdp::Message &reply) {
         opencmw::OAuthOutput out;
         auto outBuf = reply.data;
+        REQUIRE_MESSAGE(reply.error.empty(), reply.error);
         opencmw::deserialise<opencmw::YaS, opencmw::ProtocolCheck::IGNORE>(outBuf, out);
 
         REQUIRE(out.authorizationUri.size());
@@ -140,6 +142,7 @@ TEST_CASE("Worker test", "[OAuth]") {
         client.set(opencmw::URI("mdp://127.0.0.1:12346/oauth"), [&](const mdp::Message& rep) {
                 opencmw::OAuthOutput tokenOut;
                 auto tokenOutBuf = rep.data;
+                REQUIRE_MESSAGE(rep.error.empty(), rep.error);
                 opencmw::deserialise<opencmw::YaS, opencmw::ProtocolCheck::IGNORE>(tokenOutBuf, tokenOut);
 
                 // we must have got an access token now
@@ -154,6 +157,7 @@ TEST_CASE("Worker test", "[OAuth]") {
                 client.set(opencmw::URI("mdp://127.0.0.1:12346/keystore"), [&](const mdp::Message &keyResp) {
                         opencmw::KeystoreOutput keyOut;
                         auto keyOutBuf = keyResp.data;
+                        REQUIRE_MESSAGE(keyResp.error.empty(), keyResp.error);
                         opencmw::deserialise<opencmw::YaS, opencmw::ProtocolCheck::IGNORE>(keyOutBuf, keyOut);
 
                         // the hashes should be the same

Original file line number	Diff line number	Diff line change
`@@ -324,6 +324,9 @@ class OAuthWorker : public OAuthWorkerType {`
`324`	`324`
`325`	`325`	`// get RBAC roles`
`326`	`326`	`out.roles = _client.getAssignedRoles(out.accessToken);`
	`327`	`+ if (!in.publicKey.empty()) {`
	`328`	`+ _keystore.addRoles(in.publicKey, out.roles);`
	`329`	`+ }`
`327`	`330`	`}`
`328`	`331`	`});`
`329`	`332`	`_keystoreThread = std::thread([this] { _keystore.run(); });`