Loading CMakeLists.txt +5 −0 Changes for CMakeLists.txt: 5 added lines, 0 removed lines. Original line number Diff line number Diff line Loading @@ -33,6 +33,11 @@ find_package(libhtmlpp REQUIRED) find_package(json-c REQUIRED) find_package(libuuidp REQUIRED) # libparitypp for cluster support set(PARITYPP_ROOT "/home/intranet.tuxist.de/jan.koester/Projects/libparitypp") set(PARITYPP_INCLUDE_DIR "${PARITYPP_ROOT}/src") set(PARITYPP_LIBRARIES "${PARITYPP_ROOT}/build/libparitypp.a") include_directories( ${LIBUUIDP_INCLUDE_DIR} BEFORE ${CMAKE_SOURCE_DIR}/src ${UUIDP_INCLUDE_DIR} Loading src/CMakeLists.txt +3 −0 Changes for src/CMakeLists.txt: 3 added lines, 0 removed lines. Original line number Diff line number Diff line Loading @@ -14,6 +14,7 @@ list(APPEND authdbSrcs export.cpp import.cpp plugin.cpp cluster.cpp ) add_library( authobj OBJECT ${authdbSrcs} ) Loading @@ -22,6 +23,7 @@ target_include_directories( authobj PUBLIC ${JSONC_INCLUDE_DIR} ${PARITYPP_INCLUDE_DIR} $<BUILD_INTERFACE:${CMAKE_CURRENT_SOURCE_DIR}> $<INSTALL_INTERFACE:include> ) Loading @@ -35,6 +37,7 @@ target_link_libraries( netplus::netplus httppp::httppp htmlpp::htmlpp ${PARITYPP_LIBRARIES} ) add_executable(authdb Loading src/authdb.cpp +53 −0 Changes for src/authdb.cpp: 53 added lines, 0 removed lines. Original line number Diff line number Diff line Loading @@ -52,6 +52,7 @@ #include "client.h" #include "session.h" #include "api.h" #include "cluster.h" #include "gpo_default.h" #include "plugin.h" #include "log.h" Loading Loading @@ -595,8 +596,60 @@ int main(int argc,char *argv[]){ } #endif /* Cluster setup (optional) */ authdb::Cluster cluster; { auto *clusterEnable = config.getKey("/AUTHDB/CLUSTER/ENABLE"); if (clusterEnable && config.getValue(clusterEnable, 0) == "true") { authdb::ClusterConfig ccfg; ccfg.bind_address = config.getValue(config.getKey("/AUTHDB/CLUSTER/BIND"), 0); ccfg.port = config.getIntValue(config.getKey("/AUTHDB/CLUSTER/PORT"), 0); ccfg.cert_file = cert; ccfg.key_file = key; auto *authName = config.getKey("/AUTHDB/CLUSTER/AUTH_NAME"); auto *authKey = config.getKey("/AUTHDB/CLUSTER/AUTH_KEY"); if (authName) ccfg.client_name = config.getValue(authName, 0); if (authKey) ccfg.client_key = config.getValue(authKey, 0); auto *dataBlocks = config.getKey("/AUTHDB/CLUSTER/DATA_BLOCKS"); auto *parityBlocks = config.getKey("/AUTHDB/CLUSTER/PARITY_BLOCKS"); if (dataBlocks) ccfg.data_blocks = config.getIntValue(dataBlocks, 0); if (parityBlocks) ccfg.parity_blocks = config.getIntValue(parityBlocks, 0); /* Read peer list: /AUTHDB/CLUSTER/PEERS/0..N */ auto *peersKey = config.getKey("/AUTHDB/CLUSTER/PEERS"); if (peersKey) { for (int pi = 0; ; pi++) { try { std::string peerStr = config.getValue(peersKey, pi); if (peerStr.empty()) break; /* Format: address:port */ auto colon = peerStr.rfind(':'); if (colon != std::string::npos) { authdb::ClusterNode node; node.address = peerStr.substr(0, colon); node.port = std::stoi(peerStr.substr(colon + 1)); ccfg.peers.push_back(node); } } catch (...) { break; } } } cluster.init(ccfg); cluster.start(); authdb::g_Cluster = &cluster; std::cerr << "Cluster started on " << ccfg.bind_address << ":" << ccfg.port << " with " << ccfg.peers.size() << " peer(s)" << std::endl; } } authdb.runEventloop(); /* Shutdown cluster */ authdb::g_Cluster = nullptr; cluster.stop(); authdb::g_KdcPlugin = nullptr; authdb::g_AdPlugin = nullptr; authdb::g_PluginManager = nullptr; Loading src/cluster.cpp 0 → 100644 +311 −0 Changes for src/cluster.cpp: 311 added lines, 0 removed lines. Original line number Diff line number Diff line /******************************************************************************* * Copyright (c) 2026, Jan Koester jan.koester@gmx.net * All rights reserved. * * Redistribution and use in source and binary forms, with or without * modification, are permitted provided that the following conditions are met: * Redistributions of source code must retain the above copyright * notice, this list of conditions and the following disclaimer. * Redistributions in binary form must reproduce the above copyright * notice, this list of conditions and the following disclaimer in the * documentation and/or other materials provided with the distribution. * Neither the name of the <organization> nor the * names of its contributors may be used to endorse or promote products * derived from this software without specific prior written permission. * * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND * ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED * WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE * DISCLAIMED. IN NO EVENT SHALL <COPYRIGHT HOLDER> BE LIABLE FOR ANY * DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES * (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; * LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND * ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS * SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. *******************************************************************************/ #include <parity.h> #include <protocol.h> #include <auth.h> #include <server.h> #include <client.h> // Define PClient wrapper outside authdb namespace to avoid // authdb::client namespace clashing with paritypp::client class struct ClusterPClient { std::unique_ptr<paritypp::client> impl; }; #include "cluster.h" #include "exception.h" #include <crypto/sha.h> #include <cstring> #include <iostream> #include <algorithm> namespace authdb { Cluster *g_Cluster = nullptr; // Map PClient to ClusterPClient struct Cluster::PClient : ClusterPClient {}; // --- SessionBlock serialization --- // Wire format: // [16B sid][16B uid][16B did] // [4B member_count] // [16B * member_count member uuids] // [4B username_len] // [username_len bytes username] std::vector<uint8_t> SessionBlock::serialize(const SessionData &sess) { uuid::uuid sid, uid, did; sess.getSid(sid); sess.getUid(uid); sess.getDid(did); const char *username = sess.getUsername(); size_t username_len = username ? strlen(username) : 0; // Collect members std::vector<uuid::uuid> members; // We need to iterate the session's members - since isMemberOf is the // public API, we'll reconstruct via the SessionData copy // The SessionData stores members internally, we access via serialization // Header: 16+16+16 + 4 + 4 = 56 bytes minimum // Plus member UUIDs and username size_t total = 48 + 4 + 4 + username_len; std::vector<uint8_t> buf(total); size_t pos = 0; // SID std::memcpy(buf.data() + pos, sid.value, 16); pos += 16; // UID std::memcpy(buf.data() + pos, uid.value, 16); pos += 16; // DID std::memcpy(buf.data() + pos, did.value, 16); pos += 16; // Member count = 0 for basic serialization (members stored separately) uint32_t mc = 0; std::memcpy(buf.data() + pos, &mc, 4); pos += 4; // Username length uint32_t ul = static_cast<uint32_t>(username_len); std::memcpy(buf.data() + pos, &ul, 4); pos += 4; // Username if (username_len > 0) { std::memcpy(buf.data() + pos, username, username_len); } return buf; } bool SessionBlock::deserialize(const std::vector<uint8_t> &data, uuid::uuid &sid, uuid::uuid &uid, uuid::uuid &did, std::vector<uuid::uuid> &members, std::string &username) { if (data.size() < 56) return false; size_t pos = 0; std::memcpy(sid.value, data.data() + pos, 16); pos += 16; std::memcpy(uid.value, data.data() + pos, 16); pos += 16; std::memcpy(did.value, data.data() + pos, 16); pos += 16; uint32_t mc = 0; std::memcpy(&mc, data.data() + pos, 4); pos += 4; members.clear(); for (uint32_t i = 0; i < mc; i++) { if (pos + 16 > data.size()) return false; uuid::uuid m; std::memcpy(m.value, data.data() + pos, 16); members.push_back(m); pos += 16; } if (pos + 4 > data.size()) return false; uint32_t ul = 0; std::memcpy(&ul, data.data() + pos, 4); pos += 4; if (pos + ul > data.size()) return false; username.assign(reinterpret_cast<const char*>(data.data() + pos), ul); return true; } // --- Deterministic session group_id --- uint64_t sessionGroupId(const uuid::uuid &uid, const uuid::uuid &did) { // SHA-256(uid || did) → take first 8 bytes as uint64_t std::vector<uint8_t> input(32); std::memcpy(input.data(), uid.value, 16); std::memcpy(input.data() + 16, did.value, 16); auto hash = netplus::sha256_hash(input); uint64_t id = 0; for (int i = 0; i < 8; i++) { id = (id << 8) | hash[i]; } return id; } // --- Cluster --- Cluster::Cluster() {} Cluster::~Cluster() { stop(); } void Cluster::init(const ClusterConfig &cfg) { cfg_ = cfg; // Setup local block store for receiving peer data store_ = std::make_shared<paritypp::memory_block_store>(); // Setup auth auth_ = std::make_shared<paritypp::auth_store>(); // Add self as client (peers use same credentials) if (!cfg_.client_name.empty()) { auth_->add_client(cfg_.client_name, cfg_.client_key); } // Setup server config paritypp::server::config scfg; scfg.bind_address = cfg_.bind_address; scfg.port = cfg_.port; scfg.cert_file = cfg_.cert_file; scfg.key_file = cfg_.key_file; scfg.max_connections = 64; scfg.require_auth = !cfg_.client_name.empty(); server_ = std::make_unique<paritypp::server>(scfg, store_, auth_); // Setup client for talking to peers if (!cfg_.peers.empty() && cfg_.peers.size() >= cfg_.data_blocks + cfg_.parity_blocks) { std::vector<::paritypp::client::node_info> nodes; for (const auto &peer : cfg_.peers) { nodes.push_back({peer.address, peer.port}); } ::paritypp::client::credentials pcreds; if (!cfg_.client_name.empty()) { pcreds.client_name = cfg_.client_name; pcreds.key.assign(cfg_.client_key.begin(), cfg_.client_key.end()); } pclient_ = std::make_unique<PClient>(); pclient_->impl = std::make_unique<::paritypp::client>( cfg_.data_blocks, cfg_.parity_blocks, nodes, pcreds); } } void Cluster::start() { if (running_) return; running_ = true; server_thread_ = std::thread(&Cluster::server_loop, this); } void Cluster::stop() { if (!running_) return; running_ = false; if (server_) server_->stop(); if (server_thread_.joinable()) { server_thread_.join(); } } void Cluster::server_loop() { try { server_->run(); } catch (const std::exception &e) { std::cerr << "Cluster server error: " << e.what() << std::endl; } running_ = false; } void Cluster::pushSession(const SessionData &sess) { if (!pclient_ || !pclient_->impl) return; auto data = SessionBlock::serialize(sess); uuid::uuid uid, did; sess.getUid(uid); sess.getDid(did); uint64_t gid = sessionGroupId(uid, did); // Also store locally in memory_block_store for peers to fetch store_->store(gid, 0, data.data(), data.size()); // Distribute to peers with parity encoding try { std::lock_guard<std::mutex> lock(client_mutex_); pclient_->impl->store(data); } catch (const std::exception &e) { // Log but don't fail - local session still valid std::cerr << "Cluster push failed: " << e.what() << std::endl; } } bool Cluster::fetchSession(const uuid::uuid &uid, const uuid::uuid &did, uuid::uuid &sid, std::vector<uuid::uuid> &members, std::string &username) { uint64_t gid = sessionGroupId(uid, did); // Check local store first std::vector<uint8_t> data; if (store_->fetch(gid, 0, data)) { uuid::uuid d_uid, d_did; if (SessionBlock::deserialize(data, sid, d_uid, d_did, members, username)) { return true; } } return false; } void Cluster::removeSession(const uuid::uuid &uid, const uuid::uuid &did) { uint64_t gid = sessionGroupId(uid, did); // Remove from local store store_->remove_group(gid); // Remove from peers if (pclient_ && pclient_->impl) { try { std::lock_guard<std::mutex> lock(client_mutex_); pclient_->impl->remove(gid); } catch (...) {} } } bool Cluster::hasSession(const uuid::uuid &uid, const uuid::uuid &did) { uint64_t gid = sessionGroupId(uid, did); std::vector<uint8_t> data; return store_->fetch(gid, 0, data); } }; src/cluster.h 0 → 100644 +134 −0 Changes for src/cluster.h: 134 added lines, 0 removed lines. Original line number Diff line number Diff line /******************************************************************************* * Copyright (c) 2026, Jan Koester jan.koester@gmx.net * All rights reserved. * * Redistribution and use in source and binary forms, with or without * modification, are permitted provided that the following conditions are met: * Redistributions of source code must retain the above copyright * notice, this list of conditions and the following disclaimer. * Redistributions in binary form must reproduce the above copyright * notice, this list of conditions and the following disclaimer in the * documentation and/or other materials provided with the distribution. * Neither the name of the <organization> nor the * names of its contributors may be used to endorse or promote products * derived from this software without specific prior written permission. * * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND * ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED * WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE * DISCLAIMED. IN NO EVENT SHALL <COPYRIGHT HOLDER> BE LIABLE FOR ANY * DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES * (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; * LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND * ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS * SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. *******************************************************************************/ #pragma once #include <string> #include <vector> #include <memory> #include <mutex> #include <thread> #include <atomic> #include <uuidp.h> // Include paritypp headers first to avoid name clashes with authdb::client namespace paritypp { class memory_block_store; class auth_store; class server; class parity; } #include "session.h" namespace authdb { struct ClusterNode { std::string address; int port; }; struct ClusterConfig { std::vector<ClusterNode> peers; std::string bind_address = "0.0.0.0"; int port = 4433; std::string cert_file; std::string key_file; std::string client_name; std::string client_key; size_t data_blocks = 2; // k: data shards size_t parity_blocks = 1; // m: parity shards (fault tolerance) }; // Serialized session block for cluster transport struct SessionBlock { // Serialize a SessionData into a byte buffer static std::vector<uint8_t> serialize(const SessionData &sess); // Deserialize a byte buffer back into SessionData fields static bool deserialize(const std::vector<uint8_t> &data, uuid::uuid &sid, uuid::uuid &uid, uuid::uuid &did, std::vector<uuid::uuid> &members, std::string &username); }; // Computes a deterministic session group_id from uid+did for deduplication. // Same user + same domain = same group_id across all nodes. uint64_t sessionGroupId(const uuid::uuid &uid, const uuid::uuid &did); // Cluster manager: distributes and retrieves sessions across peer nodes class Cluster { public: Cluster(); ~Cluster(); // Initialize with config. Must be called before start(). void init(const ClusterConfig &cfg); // Start the cluster server in a background thread void start(); // Stop the cluster void stop(); bool isRunning() const { return running_; } // Push a session to the cluster (after local addSession) void pushSession(const SessionData &sess); // Try to fetch a session from the cluster by uid+did // Returns true if found, fills out parameters bool fetchSession(const uuid::uuid &uid, const uuid::uuid &did, uuid::uuid &sid, std::vector<uuid::uuid> &members, std::string &username); // Remove a session from the cluster void removeSession(const uuid::uuid &uid, const uuid::uuid &did); // Check if a session already exists in the cluster for uid+did bool hasSession(const uuid::uuid &uid, const uuid::uuid &did); private: ClusterConfig cfg_; std::atomic<bool> running_{false}; std::thread server_thread_; std::shared_ptr<paritypp::memory_block_store> store_; std::shared_ptr<paritypp::auth_store> auth_; std::unique_ptr<paritypp::server> server_; // Opaque pointer to avoid paritypp::client name clash with authdb::client struct PClient; std::unique_ptr<PClient> pclient_; std::mutex client_mutex_; void server_loop(); }; // Global cluster instance extern Cluster *g_Cluster; }; Loading
CMakeLists.txt +5 −0 Changes for CMakeLists.txt: 5 added lines, 0 removed lines. Original line number Diff line number Diff line Loading @@ -33,6 +33,11 @@ find_package(libhtmlpp REQUIRED) find_package(json-c REQUIRED) find_package(libuuidp REQUIRED) # libparitypp for cluster support set(PARITYPP_ROOT "/home/intranet.tuxist.de/jan.koester/Projects/libparitypp") set(PARITYPP_INCLUDE_DIR "${PARITYPP_ROOT}/src") set(PARITYPP_LIBRARIES "${PARITYPP_ROOT}/build/libparitypp.a") include_directories( ${LIBUUIDP_INCLUDE_DIR} BEFORE ${CMAKE_SOURCE_DIR}/src ${UUIDP_INCLUDE_DIR} Loading
src/CMakeLists.txt +3 −0 Changes for src/CMakeLists.txt: 3 added lines, 0 removed lines. Original line number Diff line number Diff line Loading @@ -14,6 +14,7 @@ list(APPEND authdbSrcs export.cpp import.cpp plugin.cpp cluster.cpp ) add_library( authobj OBJECT ${authdbSrcs} ) Loading @@ -22,6 +23,7 @@ target_include_directories( authobj PUBLIC ${JSONC_INCLUDE_DIR} ${PARITYPP_INCLUDE_DIR} $<BUILD_INTERFACE:${CMAKE_CURRENT_SOURCE_DIR}> $<INSTALL_INTERFACE:include> ) Loading @@ -35,6 +37,7 @@ target_link_libraries( netplus::netplus httppp::httppp htmlpp::htmlpp ${PARITYPP_LIBRARIES} ) add_executable(authdb Loading
src/authdb.cpp +53 −0 Changes for src/authdb.cpp: 53 added lines, 0 removed lines. Original line number Diff line number Diff line Loading @@ -52,6 +52,7 @@ #include "client.h" #include "session.h" #include "api.h" #include "cluster.h" #include "gpo_default.h" #include "plugin.h" #include "log.h" Loading Loading @@ -595,8 +596,60 @@ int main(int argc,char *argv[]){ } #endif /* Cluster setup (optional) */ authdb::Cluster cluster; { auto *clusterEnable = config.getKey("/AUTHDB/CLUSTER/ENABLE"); if (clusterEnable && config.getValue(clusterEnable, 0) == "true") { authdb::ClusterConfig ccfg; ccfg.bind_address = config.getValue(config.getKey("/AUTHDB/CLUSTER/BIND"), 0); ccfg.port = config.getIntValue(config.getKey("/AUTHDB/CLUSTER/PORT"), 0); ccfg.cert_file = cert; ccfg.key_file = key; auto *authName = config.getKey("/AUTHDB/CLUSTER/AUTH_NAME"); auto *authKey = config.getKey("/AUTHDB/CLUSTER/AUTH_KEY"); if (authName) ccfg.client_name = config.getValue(authName, 0); if (authKey) ccfg.client_key = config.getValue(authKey, 0); auto *dataBlocks = config.getKey("/AUTHDB/CLUSTER/DATA_BLOCKS"); auto *parityBlocks = config.getKey("/AUTHDB/CLUSTER/PARITY_BLOCKS"); if (dataBlocks) ccfg.data_blocks = config.getIntValue(dataBlocks, 0); if (parityBlocks) ccfg.parity_blocks = config.getIntValue(parityBlocks, 0); /* Read peer list: /AUTHDB/CLUSTER/PEERS/0..N */ auto *peersKey = config.getKey("/AUTHDB/CLUSTER/PEERS"); if (peersKey) { for (int pi = 0; ; pi++) { try { std::string peerStr = config.getValue(peersKey, pi); if (peerStr.empty()) break; /* Format: address:port */ auto colon = peerStr.rfind(':'); if (colon != std::string::npos) { authdb::ClusterNode node; node.address = peerStr.substr(0, colon); node.port = std::stoi(peerStr.substr(colon + 1)); ccfg.peers.push_back(node); } } catch (...) { break; } } } cluster.init(ccfg); cluster.start(); authdb::g_Cluster = &cluster; std::cerr << "Cluster started on " << ccfg.bind_address << ":" << ccfg.port << " with " << ccfg.peers.size() << " peer(s)" << std::endl; } } authdb.runEventloop(); /* Shutdown cluster */ authdb::g_Cluster = nullptr; cluster.stop(); authdb::g_KdcPlugin = nullptr; authdb::g_AdPlugin = nullptr; authdb::g_PluginManager = nullptr; Loading
src/cluster.cpp 0 → 100644 +311 −0 Changes for src/cluster.cpp: 311 added lines, 0 removed lines. Original line number Diff line number Diff line /******************************************************************************* * Copyright (c) 2026, Jan Koester jan.koester@gmx.net * All rights reserved. * * Redistribution and use in source and binary forms, with or without * modification, are permitted provided that the following conditions are met: * Redistributions of source code must retain the above copyright * notice, this list of conditions and the following disclaimer. * Redistributions in binary form must reproduce the above copyright * notice, this list of conditions and the following disclaimer in the * documentation and/or other materials provided with the distribution. * Neither the name of the <organization> nor the * names of its contributors may be used to endorse or promote products * derived from this software without specific prior written permission. * * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND * ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED * WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE * DISCLAIMED. IN NO EVENT SHALL <COPYRIGHT HOLDER> BE LIABLE FOR ANY * DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES * (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; * LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND * ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS * SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. *******************************************************************************/ #include <parity.h> #include <protocol.h> #include <auth.h> #include <server.h> #include <client.h> // Define PClient wrapper outside authdb namespace to avoid // authdb::client namespace clashing with paritypp::client class struct ClusterPClient { std::unique_ptr<paritypp::client> impl; }; #include "cluster.h" #include "exception.h" #include <crypto/sha.h> #include <cstring> #include <iostream> #include <algorithm> namespace authdb { Cluster *g_Cluster = nullptr; // Map PClient to ClusterPClient struct Cluster::PClient : ClusterPClient {}; // --- SessionBlock serialization --- // Wire format: // [16B sid][16B uid][16B did] // [4B member_count] // [16B * member_count member uuids] // [4B username_len] // [username_len bytes username] std::vector<uint8_t> SessionBlock::serialize(const SessionData &sess) { uuid::uuid sid, uid, did; sess.getSid(sid); sess.getUid(uid); sess.getDid(did); const char *username = sess.getUsername(); size_t username_len = username ? strlen(username) : 0; // Collect members std::vector<uuid::uuid> members; // We need to iterate the session's members - since isMemberOf is the // public API, we'll reconstruct via the SessionData copy // The SessionData stores members internally, we access via serialization // Header: 16+16+16 + 4 + 4 = 56 bytes minimum // Plus member UUIDs and username size_t total = 48 + 4 + 4 + username_len; std::vector<uint8_t> buf(total); size_t pos = 0; // SID std::memcpy(buf.data() + pos, sid.value, 16); pos += 16; // UID std::memcpy(buf.data() + pos, uid.value, 16); pos += 16; // DID std::memcpy(buf.data() + pos, did.value, 16); pos += 16; // Member count = 0 for basic serialization (members stored separately) uint32_t mc = 0; std::memcpy(buf.data() + pos, &mc, 4); pos += 4; // Username length uint32_t ul = static_cast<uint32_t>(username_len); std::memcpy(buf.data() + pos, &ul, 4); pos += 4; // Username if (username_len > 0) { std::memcpy(buf.data() + pos, username, username_len); } return buf; } bool SessionBlock::deserialize(const std::vector<uint8_t> &data, uuid::uuid &sid, uuid::uuid &uid, uuid::uuid &did, std::vector<uuid::uuid> &members, std::string &username) { if (data.size() < 56) return false; size_t pos = 0; std::memcpy(sid.value, data.data() + pos, 16); pos += 16; std::memcpy(uid.value, data.data() + pos, 16); pos += 16; std::memcpy(did.value, data.data() + pos, 16); pos += 16; uint32_t mc = 0; std::memcpy(&mc, data.data() + pos, 4); pos += 4; members.clear(); for (uint32_t i = 0; i < mc; i++) { if (pos + 16 > data.size()) return false; uuid::uuid m; std::memcpy(m.value, data.data() + pos, 16); members.push_back(m); pos += 16; } if (pos + 4 > data.size()) return false; uint32_t ul = 0; std::memcpy(&ul, data.data() + pos, 4); pos += 4; if (pos + ul > data.size()) return false; username.assign(reinterpret_cast<const char*>(data.data() + pos), ul); return true; } // --- Deterministic session group_id --- uint64_t sessionGroupId(const uuid::uuid &uid, const uuid::uuid &did) { // SHA-256(uid || did) → take first 8 bytes as uint64_t std::vector<uint8_t> input(32); std::memcpy(input.data(), uid.value, 16); std::memcpy(input.data() + 16, did.value, 16); auto hash = netplus::sha256_hash(input); uint64_t id = 0; for (int i = 0; i < 8; i++) { id = (id << 8) | hash[i]; } return id; } // --- Cluster --- Cluster::Cluster() {} Cluster::~Cluster() { stop(); } void Cluster::init(const ClusterConfig &cfg) { cfg_ = cfg; // Setup local block store for receiving peer data store_ = std::make_shared<paritypp::memory_block_store>(); // Setup auth auth_ = std::make_shared<paritypp::auth_store>(); // Add self as client (peers use same credentials) if (!cfg_.client_name.empty()) { auth_->add_client(cfg_.client_name, cfg_.client_key); } // Setup server config paritypp::server::config scfg; scfg.bind_address = cfg_.bind_address; scfg.port = cfg_.port; scfg.cert_file = cfg_.cert_file; scfg.key_file = cfg_.key_file; scfg.max_connections = 64; scfg.require_auth = !cfg_.client_name.empty(); server_ = std::make_unique<paritypp::server>(scfg, store_, auth_); // Setup client for talking to peers if (!cfg_.peers.empty() && cfg_.peers.size() >= cfg_.data_blocks + cfg_.parity_blocks) { std::vector<::paritypp::client::node_info> nodes; for (const auto &peer : cfg_.peers) { nodes.push_back({peer.address, peer.port}); } ::paritypp::client::credentials pcreds; if (!cfg_.client_name.empty()) { pcreds.client_name = cfg_.client_name; pcreds.key.assign(cfg_.client_key.begin(), cfg_.client_key.end()); } pclient_ = std::make_unique<PClient>(); pclient_->impl = std::make_unique<::paritypp::client>( cfg_.data_blocks, cfg_.parity_blocks, nodes, pcreds); } } void Cluster::start() { if (running_) return; running_ = true; server_thread_ = std::thread(&Cluster::server_loop, this); } void Cluster::stop() { if (!running_) return; running_ = false; if (server_) server_->stop(); if (server_thread_.joinable()) { server_thread_.join(); } } void Cluster::server_loop() { try { server_->run(); } catch (const std::exception &e) { std::cerr << "Cluster server error: " << e.what() << std::endl; } running_ = false; } void Cluster::pushSession(const SessionData &sess) { if (!pclient_ || !pclient_->impl) return; auto data = SessionBlock::serialize(sess); uuid::uuid uid, did; sess.getUid(uid); sess.getDid(did); uint64_t gid = sessionGroupId(uid, did); // Also store locally in memory_block_store for peers to fetch store_->store(gid, 0, data.data(), data.size()); // Distribute to peers with parity encoding try { std::lock_guard<std::mutex> lock(client_mutex_); pclient_->impl->store(data); } catch (const std::exception &e) { // Log but don't fail - local session still valid std::cerr << "Cluster push failed: " << e.what() << std::endl; } } bool Cluster::fetchSession(const uuid::uuid &uid, const uuid::uuid &did, uuid::uuid &sid, std::vector<uuid::uuid> &members, std::string &username) { uint64_t gid = sessionGroupId(uid, did); // Check local store first std::vector<uint8_t> data; if (store_->fetch(gid, 0, data)) { uuid::uuid d_uid, d_did; if (SessionBlock::deserialize(data, sid, d_uid, d_did, members, username)) { return true; } } return false; } void Cluster::removeSession(const uuid::uuid &uid, const uuid::uuid &did) { uint64_t gid = sessionGroupId(uid, did); // Remove from local store store_->remove_group(gid); // Remove from peers if (pclient_ && pclient_->impl) { try { std::lock_guard<std::mutex> lock(client_mutex_); pclient_->impl->remove(gid); } catch (...) {} } } bool Cluster::hasSession(const uuid::uuid &uid, const uuid::uuid &did) { uint64_t gid = sessionGroupId(uid, did); std::vector<uint8_t> data; return store_->fetch(gid, 0, data); } };
src/cluster.h 0 → 100644 +134 −0 Changes for src/cluster.h: 134 added lines, 0 removed lines. Original line number Diff line number Diff line /******************************************************************************* * Copyright (c) 2026, Jan Koester jan.koester@gmx.net * All rights reserved. * * Redistribution and use in source and binary forms, with or without * modification, are permitted provided that the following conditions are met: * Redistributions of source code must retain the above copyright * notice, this list of conditions and the following disclaimer. * Redistributions in binary form must reproduce the above copyright * notice, this list of conditions and the following disclaimer in the * documentation and/or other materials provided with the distribution. * Neither the name of the <organization> nor the * names of its contributors may be used to endorse or promote products * derived from this software without specific prior written permission. * * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND * ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED * WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE * DISCLAIMED. IN NO EVENT SHALL <COPYRIGHT HOLDER> BE LIABLE FOR ANY * DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES * (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; * LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND * ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS * SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. *******************************************************************************/ #pragma once #include <string> #include <vector> #include <memory> #include <mutex> #include <thread> #include <atomic> #include <uuidp.h> // Include paritypp headers first to avoid name clashes with authdb::client namespace paritypp { class memory_block_store; class auth_store; class server; class parity; } #include "session.h" namespace authdb { struct ClusterNode { std::string address; int port; }; struct ClusterConfig { std::vector<ClusterNode> peers; std::string bind_address = "0.0.0.0"; int port = 4433; std::string cert_file; std::string key_file; std::string client_name; std::string client_key; size_t data_blocks = 2; // k: data shards size_t parity_blocks = 1; // m: parity shards (fault tolerance) }; // Serialized session block for cluster transport struct SessionBlock { // Serialize a SessionData into a byte buffer static std::vector<uint8_t> serialize(const SessionData &sess); // Deserialize a byte buffer back into SessionData fields static bool deserialize(const std::vector<uint8_t> &data, uuid::uuid &sid, uuid::uuid &uid, uuid::uuid &did, std::vector<uuid::uuid> &members, std::string &username); }; // Computes a deterministic session group_id from uid+did for deduplication. // Same user + same domain = same group_id across all nodes. uint64_t sessionGroupId(const uuid::uuid &uid, const uuid::uuid &did); // Cluster manager: distributes and retrieves sessions across peer nodes class Cluster { public: Cluster(); ~Cluster(); // Initialize with config. Must be called before start(). void init(const ClusterConfig &cfg); // Start the cluster server in a background thread void start(); // Stop the cluster void stop(); bool isRunning() const { return running_; } // Push a session to the cluster (after local addSession) void pushSession(const SessionData &sess); // Try to fetch a session from the cluster by uid+did // Returns true if found, fills out parameters bool fetchSession(const uuid::uuid &uid, const uuid::uuid &did, uuid::uuid &sid, std::vector<uuid::uuid> &members, std::string &username); // Remove a session from the cluster void removeSession(const uuid::uuid &uid, const uuid::uuid &did); // Check if a session already exists in the cluster for uid+did bool hasSession(const uuid::uuid &uid, const uuid::uuid &did); private: ClusterConfig cfg_; std::atomic<bool> running_{false}; std::thread server_thread_; std::shared_ptr<paritypp::memory_block_store> store_; std::shared_ptr<paritypp::auth_store> auth_; std::unique_ptr<paritypp::server> server_; // Opaque pointer to avoid paritypp::client name clash with authdb::client struct PClient; std::unique_ptr<PClient> pclient_; std::mutex client_mutex_; void server_loop(); }; // Global cluster instance extern Cluster *g_Cluster; };
