// SPDX-License-Identifier: GPL-2.0-only /* * Digital Voice Modem - Converged FNE Software * GPLv2 Open Source. Use is subject to license terms. * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER. * * Copyright (C) 2023-2026 Bryan Biedenkapp, N2PLL * */ #include "fne/Defines.h" #include "common/edac/SHA256.h" #include "common/zlib/Compression.h" #include "common/Log.h" #include "common/Utils.h" #include "network/MetadataNetwork.h" #include "common/json/json.h" #include "fne/ActivityLog.h" #include "HostFNE.h" using namespace network; using namespace network::callhandler; using namespace compress; #include // --------------------------------------------------------------------------- // Constants // --------------------------------------------------------------------------- const uint32_t TIMEOUT_MAX_REPL = 5000U; // 5 seconds // --------------------------------------------------------------------------- // Public Class Members // --------------------------------------------------------------------------- /* Initializes a new instance of the MetadataNetwork class. */ MetadataNetwork::MetadataNetwork(HostFNE* host, TrafficNetwork* trafficNetwork, const std::string& address, uint16_t port, uint16_t workerCnt) : BaseNetwork(trafficNetwork->m_peerId, true, trafficNetwork->m_debug, true, true, trafficNetwork->m_allowActivityTransfer, trafficNetwork->m_allowDiagnosticTransfer), m_trafficNetwork(trafficNetwork), m_host(host), m_address(address), m_port(port), m_status(NET_STAT_INVALID), m_peerReplicaActPkt(), m_peerTreeListPkt(), m_threadPool(workerCnt, "meta") { assert(trafficNetwork != nullptr); assert(host != nullptr); assert(!address.empty()); assert(port > 0U); } /* Finalizes a instance of the MetadataNetwork class. */ MetadataNetwork::~MetadataNetwork() = default; /* Sets endpoint preshared encryption key. */ void MetadataNetwork::setPresharedKey(const uint8_t* presharedKey) { m_socket->setPresharedKey(presharedKey); } /* Process a data frames from the network. */ void MetadataNetwork::processNetwork() { if (m_status != NET_STAT_MST_RUNNING) { return; } sockaddr_storage address; uint32_t addrLen; frame::RTPHeader rtpHeader; frame::RTPFNEHeader fneHeader; int length = 0U; // read message UInt8Array buffer = m_frameQueue->read(length, address, addrLen, &rtpHeader, &fneHeader); if (length > 0) { if (m_debug) Utils::dump(1U, "MetadataNetwork::processNetwork(), Network Message", buffer.get(), length); uint32_t peerId = fneHeader.getPeerId(); NetPacketRequest* req = new NetPacketRequest(); req->obj = m_trafficNetwork; req->metadataObj = this; req->peerId = peerId; req->address = address; req->addrLen = addrLen; req->rtpHeader = rtpHeader; req->fneHeader = fneHeader; req->length = length; req->buffer = new uint8_t[length]; ::memcpy(req->buffer, buffer.get(), length); if (!m_threadPool.enqueue(new_pooltask(taskNetworkRx, req))) { LogError(LOG_NET, "Failed to task enqueue network packet request, peerId = %u, %s:%u", peerId, udp::Socket::address(address).c_str(), udp::Socket::port(address)); if (req != nullptr) { if (req->buffer != nullptr) delete[] req->buffer; delete req; } } } } /* Updates the timer by the passed number of milliseconds. */ void MetadataNetwork::clock(uint32_t ms) { if (m_status != NET_STAT_MST_RUNNING) { return; } } /* Opens connection to the network. */ bool MetadataNetwork::open() { if (m_debug) LogInfoEx(LOG_DIAG, "Opening Network"); m_threadPool.start(); m_status = NET_STAT_MST_RUNNING; m_socket = new udp::Socket(m_address, m_port); // reinitialize the frame queue if (m_frameQueue != nullptr) { delete m_frameQueue; m_frameQueue = new FrameQueue(m_socket, m_peerId, false); } bool ret = m_socket->open(); if (!ret) { m_socket->recvBufSize(524288U); // 512K recv buffer m_socket->sendBufSize(524288U); // 512K send buffer m_status = NET_STAT_INVALID; } return ret; } /* Closes connection to the network. */ void MetadataNetwork::close() { if (m_debug) LogInfoEx(LOG_DIAG, "Closing Network"); m_threadPool.stop(); m_threadPool.wait(); m_socket->close(); m_status = NET_STAT_INVALID; } /* Helper to send a metadata message to a peer's metadata port. */ bool MetadataNetwork::writePeerMetadata(FNEPeerConnection* connection, uint32_t ssrc, FrameQueue::OpcodePair opcode, const uint8_t* data, uint32_t length, uint16_t pktSeq, uint32_t streamId) const { if (connection == nullptr) return false; if (m_status != NET_STAT_MST_RUNNING) return false; if (m_frameQueue == nullptr) return false; sockaddr_storage addr; uint32_t addrLen = 0U; uint16_t port = connection->port() + 1U; if (udp::Socket::lookup(connection->address(), port, addr, addrLen) != 0) { LogWarning(LOG_NET, "PEER %u (%s) failed to resolve metadata endpoint %s:%u", connection->id(), connection->identWithQualifier().c_str(), connection->address().c_str(), port); return false; } return m_frameQueue->write(data, length, streamId, connection->id(), ssrc, opcode, pktSeq, addr, addrLen); } // --------------------------------------------------------------------------- // Private Class Members // --------------------------------------------------------------------------- /* Process a data frames from the network. */ void MetadataNetwork::taskNetworkRx(NetPacketRequest* req) { if (req != nullptr) { TrafficNetwork* network = static_cast(req->obj); if (network == nullptr) { if (req != nullptr) { if (req->buffer != nullptr) delete[] req->buffer; delete req; } return; } MetadataNetwork* mdNetwork = static_cast(req->metadataObj); if (mdNetwork == nullptr) { if (req != nullptr) { if (req->buffer != nullptr) delete[] req->buffer; delete req; } return; } if (req == nullptr) return; if (req->length > 0) { uint32_t peerId = req->fneHeader.getPeerId(); uint32_t ssrc = req->rtpHeader.getSSRC(); uint32_t streamId = req->fneHeader.getStreamId(); // process incoming message function opcodes switch (req->fneHeader.getFunction()) { case NET_FUNC::TRANSFER: // Transfer { // resolve peer ID (used for Activity Log and Status Transfer) bool validPeerId = false; uint32_t pktPeerId = 0U; if (peerId > 0 && (network->m_peers.find(peerId) != network->m_peers.end())) { validPeerId = true; pktPeerId = peerId; } else { if (peerId > 0) { // this could be a replica transfer -- in which case, we need to check the SSRC of the packet not the peer ID if (network->m_peers.find(req->rtpHeader.getSSRC()) != network->m_peers.end()) { FNEPeerConnection* connection = network->m_peers[req->rtpHeader.getSSRC()]; if (connection != nullptr) { if (connection->peerClass() == PEER_CONN_CLASS_NEIGHBOR && connection->isReplica()) { validPeerId = true; pktPeerId = req->rtpHeader.getSSRC(); } } } } } // process incoming message subfunction opcodes switch (req->fneHeader.getSubFunction()) { case NET_SUBFUNC::TRANSFER_SUBFUNC_ACTIVITY: // Peer Activity Log Transfer { if (network->m_allowActivityTransfer) { if (pktPeerId > 0 && validPeerId) { FNEPeerConnection* connection = network->m_peers[pktPeerId]; if (connection != nullptr) { std::string ip = udp::Socket::address(req->address); // validate peer (simple validation really) if (connection->connected() && connection->address() == ip) { DECLARE_UINT8_ARRAY(rawPayload, req->length - 11U); ::memcpy(rawPayload, req->buffer + 11U, req->length - 11U); std::string payload(rawPayload, rawPayload + (req->length - 11U)); ::ActivityLog("%.9u (%8s) %s", pktPeerId, connection->identWithQualifier().c_str(), payload.c_str()); // report activity log to InfluxDB if (network->m_enableInfluxDB) { influxdb::QueryBuilder() .meas("activity") .tag("peerId", std::to_string(pktPeerId)) .field("identity", connection->identity()) .field("msg", payload) .timestamp(std::chrono::duration_cast(std::chrono::system_clock::now().time_since_epoch()).count()) .requestAsync(network->m_influxServer); } // repeat traffic to the connected SysView peers if (network->m_peers.size() > 0U) { for (auto peer : network->m_peers) { if (peer.second != nullptr) { if (peer.second->peerClass() == PEER_CONN_CLASS_SYSVIEW) { sockaddr_storage addr = peer.second->socketStorage(); uint32_t addrLen = peer.second->sockStorageLen(); network->m_frameQueue->write(req->buffer, req->length, network->createStreamId(), pktPeerId, network->m_peerId, { NET_FUNC::TRANSFER, NET_SUBFUNC::TRANSFER_SUBFUNC_ACTIVITY }, RTP_END_OF_CALL_SEQ, addr, addrLen); } } else { continue; } } } // attempt to repeat traffic to replica masters if (network->m_host->m_peerNetworks.size() > 0) { for (auto peer : network->m_host->m_peerNetworks) { if (peer.second != nullptr) { if (peer.second->isEnabled() && peer.second->isReplica()) { peer.second->writeMaster({ NET_FUNC::TRANSFER, NET_SUBFUNC::TRANSFER_SUBFUNC_ACTIVITY }, req->buffer, req->length, RTP_END_OF_CALL_SEQ, 0U, true, pktPeerId); } } } } } else { network->writePeerNAK(pktPeerId, network->createStreamId(), TAG_TRANSFER_ACT_LOG, NET_CONN_NAK_FNE_UNAUTHORIZED); } } } } } break; case NET_SUBFUNC::TRANSFER_SUBFUNC_DIAG: // Peer Diagnostic Log Transfer { if (network->m_allowDiagnosticTransfer) { if (peerId > 0 && (network->m_peers.find(peerId) != network->m_peers.end())) { FNEPeerConnection* connection = network->m_peers[peerId]; if (connection != nullptr) { std::string ip = udp::Socket::address(req->address); // validate peer (simple validation really) if (connection->connected() && connection->address() == ip) { DECLARE_UINT8_ARRAY(rawPayload, req->length - 11U); ::memcpy(rawPayload, req->buffer + 11U, req->length - 11U); std::string payload(rawPayload, rawPayload + (req->length - 11U)); bool currState = g_disableTimeDisplay; g_disableTimeDisplay = true; ::Log(9999U, {nullptr, nullptr, 0U, nullptr}, "%.9u (%8s) %s", peerId, connection->identWithQualifier().c_str(), payload.c_str()); g_disableTimeDisplay = currState; // report diagnostic log to InfluxDB if (network->m_enableInfluxDB) { influxdb::QueryBuilder() .meas("diag") .tag("peerId", std::to_string(peerId)) .field("identity", connection->identity()) .field("msg", payload) .timestamp(std::chrono::duration_cast(std::chrono::system_clock::now().time_since_epoch()).count()) .requestAsync(network->m_influxServer); } } else { network->writePeerNAK(peerId, network->createStreamId(), TAG_TRANSFER_DIAG_LOG, NET_CONN_NAK_FNE_UNAUTHORIZED); } } } } } break; case NET_SUBFUNC::TRANSFER_SUBFUNC_STATUS: // Peer Status Transfer { if (pktPeerId > 0 && validPeerId) { FNEPeerConnection* connection = network->m_peers[pktPeerId]; if (connection != nullptr) { std::string ip = udp::Socket::address(req->address); // validate peer (simple validation really) if (connection->connected() && connection->address() == ip) { if (network->m_peers.size() > 0U) { // attempt to repeat status traffic to SysView clients for (auto peer : network->m_peers) { if (peer.second != nullptr) { if (peer.second->peerClass() == PEER_CONN_CLASS_SYSVIEW) { sockaddr_storage addr = peer.second->socketStorage(); uint32_t addrLen = peer.second->sockStorageLen(); if (network->m_debug) { LogDebug(LOG_DIAG, "SysView, srcPeer = %u, dstPeer = %u, peer status message, len = %u", pktPeerId, peer.first, req->length); } network->m_frameQueue->write(req->buffer, req->length, network->createStreamId(), pktPeerId, network->m_peerId, { NET_FUNC::TRANSFER, NET_SUBFUNC::TRANSFER_SUBFUNC_STATUS }, RTP_END_OF_CALL_SEQ, addr, addrLen); } } else { continue; } } // attempt to repeat status traffic to replica masters if (network->m_host->m_peerNetworks.size() > 0) { for (auto peer : network->m_host->m_peerNetworks) { if (peer.second != nullptr) { if (peer.second->isEnabled() && peer.second->isReplica()) { peer.second->writeMaster({ NET_FUNC::TRANSFER, NET_SUBFUNC::TRANSFER_SUBFUNC_STATUS }, req->buffer, req->length, RTP_END_OF_CALL_SEQ, 0U, true, pktPeerId); } } } } } } else { network->writePeerNAK(pktPeerId, network->createStreamId(), TAG_TRANSFER_STATUS, NET_CONN_NAK_FNE_UNAUTHORIZED); } } } } break; case NET_SUBFUNC::TRANSFER_SUBFUNC_PATCH_STATUS: // Console Patch Status Transfer { if (pktPeerId > 0 && validPeerId) { FNEPeerConnection* connection = network->m_peers[pktPeerId]; if (connection != nullptr) { if (!network->patchStatusEnabled()) { network->writePeerNAK(pktPeerId, network->createStreamId(), TAG_TRANSFER_PATCH_STATUS, NET_CONN_NAK_FNE_UNAUTHORIZED); break; } std::string ip = udp::Socket::address(req->address); // Only authenticated console peers may publish or request patch registry state. if (req->length <= 11U) { network->writePeerNAK(pktPeerId, network->createStreamId(), TAG_TRANSFER_PATCH_STATUS, NET_CONN_NAK_ILLEGAL_PACKET); break; } if (connection->connected() && connection->address() == ip && connection->peerClass() == PEER_CONN_CLASS_CONSOLE) { DECLARE_UINT8_ARRAY(rawPayload, req->length - 11U); ::memcpy(rawPayload, req->buffer + 11U, req->length - 11U); std::string payload(rawPayload, rawPayload + (req->length - 11U)); json::value v; std::string err = json::parse(v, payload); if (!err.empty() || !v.is()) { network->writePeerNAK(pktPeerId, network->createStreamId(), TAG_TRANSFER_PATCH_STATUS, NET_CONN_NAK_ILLEGAL_PACKET); break; } json::object reqObj = v.get(); std::string type = "snapshot"; if (reqObj["type"].is()) type = reqObj["type"].get(); if (type == "request") { json::object snapshot = network->patchStatusRegistry().snapshot(); network->writePatchStatusToPeer(pktPeerId, snapshot); break; } // The authenticated peer identity is authoritative; do not allow spoofed peer IDs. reqObj["peerId"].set(pktPeerId); reqObj["originFnePeerId"].set(network->m_peerId); if (!reqObj["peerName"].is() || reqObj["peerName"].get().empty()) reqObj["peerName"].set(connection->identity()); json::object response = json::object(); std::string errorMessage; bool changed = false; if (!network->patchStatusRegistry().publish(reqObj, response, errorMessage, &changed)) { LogWarning(LOG_MASTER, "PEER %u (%s) invalid patch status payload, %s", pktPeerId, connection->identWithQualifier().c_str(), errorMessage.c_str()); network->writePeerNAK(pktPeerId, network->createStreamId(), TAG_TRANSFER_PATCH_STATUS, NET_CONN_NAK_ILLEGAL_PACKET); break; } if (changed) { network->writePatchStatusToConsoles(response); network->replicatePatchStatus(reqObj); } } else { network->writePeerNAK(pktPeerId, network->createStreamId(), TAG_TRANSFER_PATCH_STATUS, NET_CONN_NAK_FNE_UNAUTHORIZED); } } } } break; default: { LogWarning(LOG_MASTER, "PEER %u, unknown/unsupported transfer opcode %u", peerId, req->fneHeader.getSubFunction()); if (network->m_debug) Utils::dump("Unknown/unsupported transfer opcode from the peer", req->buffer, req->length); } break; } } break; case NET_FUNC::ANNOUNCE: // Announce { // process incoming message subfunction opcodes switch (req->fneHeader.getSubFunction()) { case NET_SUBFUNC::ANNC_SUBFUNC_GRP_AFFIL: // Announce Group Affiliation { if (peerId > 0 && (network->m_peers.find(peerId) != network->m_peers.end())) { FNEPeerConnection* connection = network->m_peers[peerId]; if (connection != nullptr) { std::string ip = udp::Socket::address(req->address); std::shared_ptr aff = network->getPeerAffiliations(peerId); if (aff == nullptr) { LogError(LOG_MASTER, "PEER %u (%s) has uninitialized affiliations lookup?", peerId, connection->identWithQualifier().c_str()); network->writePeerNAK(peerId, streamId, TAG_ANNOUNCE, NET_CONN_NAK_INVALID); } // validate peer (simple validation really) if (connection->connected() && connection->address() == ip && aff != nullptr) { uint32_t srcId = GET_UINT24(req->buffer, 0U); // Source Address uint32_t dstId = GET_UINT24(req->buffer, 3U); // Destination Address aff->groupUnaff(srcId); aff->groupAff(srcId, dstId); // attempt to repeat traffic to replica masters if (network->m_host->m_peerNetworks.size() > 0) { for (auto& peer : network->m_host->m_peerNetworks) { if (peer.second != nullptr) { if (peer.second->isEnabled() && peer.second->isReplica()) { peer.second->writeMaster({ NET_FUNC::ANNOUNCE, NET_SUBFUNC::ANNC_SUBFUNC_GRP_AFFIL }, req->buffer, req->length, req->rtpHeader.getSequence(), streamId, true); } } } } } else { network->writePeerNAK(peerId, streamId, TAG_ANNOUNCE, NET_CONN_NAK_FNE_UNAUTHORIZED); } } } } break; case NET_SUBFUNC::ANNC_SUBFUNC_UNIT_REG: // Announce Unit Registration { if (peerId > 0 && (network->m_peers.find(peerId) != network->m_peers.end())) { FNEPeerConnection* connection = network->m_peers[peerId]; if (connection != nullptr) { std::string ip = udp::Socket::address(req->address); std::shared_ptr aff = network->getPeerAffiliations(peerId); if (aff == nullptr) { LogError(LOG_MASTER, "PEER %u (%s) has uninitialized affiliations lookup?", peerId, connection->identWithQualifier().c_str()); network->writePeerNAK(peerId, streamId, TAG_ANNOUNCE, NET_CONN_NAK_INVALID); } // validate peer (simple validation really) if (connection->connected() && connection->address() == ip && aff != nullptr) { uint32_t srcId = GET_UINT24(req->buffer, 0U); // Source Address aff->unitReg(srcId, ssrc); network->m_globalAff->unitReg(srcId, ssrc); // attempt to repeat traffic to replica masters if (network->m_host->m_peerNetworks.size() > 0) { for (auto& peer : network->m_host->m_peerNetworks) { if (peer.second != nullptr) { if (peer.second->isEnabled() && peer.second->isReplica()) { peer.second->writeMaster({ NET_FUNC::ANNOUNCE, NET_SUBFUNC::ANNC_SUBFUNC_UNIT_REG }, req->buffer, req->length, req->rtpHeader.getSequence(), streamId, true, 0U, ssrc); } } } } } else { network->writePeerNAK(peerId, streamId, TAG_ANNOUNCE, NET_CONN_NAK_FNE_UNAUTHORIZED); } } } } break; case NET_SUBFUNC::ANNC_SUBFUNC_UNIT_DEREG: // Announce Unit Deregistration { if (peerId > 0 && (network->m_peers.find(peerId) != network->m_peers.end())) { FNEPeerConnection* connection = network->m_peers[peerId]; if (connection != nullptr) { std::string ip = udp::Socket::address(req->address); std::shared_ptr aff = network->getPeerAffiliations(peerId); if (aff == nullptr) { LogError(LOG_MASTER, "PEER %u (%s) has uninitialized affiliations lookup?", peerId, connection->identWithQualifier().c_str()); network->writePeerNAK(peerId, streamId, TAG_ANNOUNCE, NET_CONN_NAK_INVALID); } // validate peer (simple validation really) if (connection->connected() && connection->address() == ip && aff != nullptr) { uint32_t srcId = GET_UINT24(req->buffer, 0U); // Source Address aff->unitDereg(srcId); network->m_globalAff->unitDereg(srcId); // attempt to repeat traffic to replica masters if (network->m_host->m_peerNetworks.size() > 0) { for (auto& peer : network->m_host->m_peerNetworks) { if (peer.second != nullptr) { if (peer.second->isEnabled() && peer.second->isReplica()) { peer.second->writeMaster({ NET_FUNC::ANNOUNCE, NET_SUBFUNC::ANNC_SUBFUNC_UNIT_DEREG }, req->buffer, req->length, req->rtpHeader.getSequence(), streamId, true); } } } } } else { network->writePeerNAK(peerId, streamId, TAG_ANNOUNCE, NET_CONN_NAK_FNE_UNAUTHORIZED); } } } } break; case NET_SUBFUNC::ANNC_SUBFUNC_GRP_UNAFFIL: // Announce Group Affiliation Removal { if (peerId > 0 && (network->m_peers.find(peerId) != network->m_peers.end())) { FNEPeerConnection* connection = network->m_peers[peerId]; if (connection != nullptr) { std::string ip = udp::Socket::address(req->address); std::shared_ptr aff = network->getPeerAffiliations(peerId); if (aff == nullptr) { LogError(LOG_MASTER, "PEER %u (%s) has uninitialized affiliations lookup?", peerId, connection->identWithQualifier().c_str()); network->writePeerNAK(peerId, streamId, TAG_ANNOUNCE, NET_CONN_NAK_INVALID); } // validate peer (simple validation really) if (connection->connected() && connection->address() == ip && aff != nullptr) { uint32_t srcId = GET_UINT24(req->buffer, 0U); // Source Address aff->groupUnaff(srcId); network->m_globalAff->groupUnaff(srcId); // attempt to repeat traffic to replica masters if (network->m_host->m_peerNetworks.size() > 0) { for (auto& peer : network->m_host->m_peerNetworks) { if (peer.second != nullptr) { if (peer.second->isEnabled() && peer.second->isReplica()) { peer.second->writeMaster({ NET_FUNC::ANNOUNCE, NET_SUBFUNC::ANNC_SUBFUNC_GRP_UNAFFIL }, req->buffer, req->length, req->rtpHeader.getSequence(), streamId, true); } } } } } else { network->writePeerNAK(peerId, streamId, TAG_ANNOUNCE, NET_CONN_NAK_FNE_UNAUTHORIZED); } } } } break; case NET_SUBFUNC::ANNC_SUBFUNC_AFFILS: // Announce Update All Affiliations { if (peerId > 0 && (network->m_peers.find(peerId) != network->m_peers.end())) { FNEPeerConnection* connection = network->m_peers[peerId]; if (connection != nullptr) { std::string ip = udp::Socket::address(req->address); // validate peer (simple validation really) if (connection->connected() && connection->address() == ip) { std::shared_ptr aff = network->getPeerAffiliations(peerId); if (aff == nullptr) { LogError(LOG_MASTER, "PEER %u (%s) has uninitialized affiliations lookup?", peerId, connection->identWithQualifier().c_str()); network->writePeerNAK(peerId, streamId, TAG_ANNOUNCE, NET_CONN_NAK_INVALID); } if (aff != nullptr) { aff->clearGroupAff(0U, true); // update TGID lists uint32_t len = GET_UINT32(req->buffer, 0U); uint32_t offs = 4U; for (uint32_t i = 0; i < len; i++) { uint32_t srcId = GET_UINT24(req->buffer, offs); uint32_t dstId = GET_UINT24(req->buffer, offs + 4U); aff->groupAff(srcId, dstId); network->m_globalAff->groupAff(srcId, dstId); offs += 8U; } LogInfoEx(LOG_MASTER, "PEER %u (%s) announced %u affiliations", peerId, connection->identWithQualifier().c_str(), len); // attempt to repeat traffic to replica masters if (network->m_host->m_peerNetworks.size() > 0) { for (auto& peer : network->m_host->m_peerNetworks) { if (peer.second != nullptr) { if (peer.second->isEnabled() && peer.second->isReplica()) { peer.second->writeMaster({ NET_FUNC::ANNOUNCE, NET_SUBFUNC::ANNC_SUBFUNC_AFFILS }, req->buffer, req->length, req->rtpHeader.getSequence(), streamId, true); } } } } } } else { network->writePeerNAK(peerId, streamId, TAG_ANNOUNCE, NET_CONN_NAK_FNE_UNAUTHORIZED); } } } } break; case NET_SUBFUNC::ANNC_SUBFUNC_UNIT_REGS: // Announce Update All Unit Registrations { if (peerId > 0 && (network->m_peers.find(peerId) != network->m_peers.end())) { FNEPeerConnection* connection = network->m_peers[peerId]; if (connection != nullptr) { std::string ip = udp::Socket::address(req->address); // validate peer (simple validation really) if (connection->connected() && connection->address() == ip) { std::shared_ptr aff = network->getPeerAffiliations(peerId); if (aff == nullptr) { LogError(LOG_MASTER, "PEER %u (%s) has uninitialized affiliations lookup?", peerId, connection->identWithQualifier().c_str()); network->writePeerNAK(peerId, streamId, TAG_ANNOUNCE, NET_CONN_NAK_INVALID); } if (aff != nullptr) { aff->clearUnitReg(); // update unit registration lists uint32_t len = GET_UINT32(req->buffer, 0U); uint32_t offs = 4U; for (uint32_t i = 0; i < len; i++) { uint32_t srcId = GET_UINT24(req->buffer, offs); aff->unitReg(srcId, ssrc); network->m_globalAff->unitReg(srcId, ssrc); offs += 3U; } LogInfoEx(LOG_MASTER, "PEER %u (%s) announced %u unit registrations", peerId, connection->identWithQualifier().c_str(), len); // attempt to repeat traffic to replica masters if (network->m_host->m_peerNetworks.size() > 0) { for (auto& peer : network->m_host->m_peerNetworks) { if (peer.second != nullptr) { if (peer.second->isEnabled() && peer.second->isReplica()) { peer.second->writeMaster({ NET_FUNC::ANNOUNCE, NET_SUBFUNC::ANNC_SUBFUNC_UNIT_REGS }, req->buffer, req->length, req->rtpHeader.getSequence(), streamId, true); } } } } } } else { network->writePeerNAK(peerId, streamId, TAG_ANNOUNCE, NET_CONN_NAK_FNE_UNAUTHORIZED); } } } } break; case NET_SUBFUNC::ANNC_SUBFUNC_SITE_VC: // Announce Site VCs { if (peerId > 0 && (network->m_peers.find(peerId) != network->m_peers.end())) { FNEPeerConnection* connection = network->m_peers[peerId]; if (connection != nullptr) { std::string ip = udp::Socket::address(req->address); // validate peer (simple validation really) if (connection->connected() && connection->address() == ip) { std::vector vcPeers; // update peer association uint32_t len = GET_UINT32(req->buffer, 0U); uint32_t offs = 4U; for (uint32_t i = 0; i < len; i++) { uint32_t vcPeerId = GET_UINT32(req->buffer, offs); if (vcPeerId > 0 && (network->m_peers.find(vcPeerId) != network->m_peers.end())) { FNEPeerConnection* vcConnection = network->m_peers[vcPeerId]; if (vcConnection != nullptr) { vcConnection->ccPeerId(peerId); vcPeers.push_back(vcPeerId); } } offs += 4U; } LogInfoEx(LOG_MASTER, "PEER %u (%s) announced %u VCs", peerId, connection->identWithQualifier().c_str(), len); network->m_ccPeerMap[peerId] = vcPeers; // attempt to repeat traffic to replica masters if (network->m_host->m_peerNetworks.size() > 0) { for (auto& peer : network->m_host->m_peerNetworks) { if (peer.second != nullptr) { if (peer.second->isEnabled() && peer.second->isReplica()) { peer.second->writeMaster({ NET_FUNC::ANNOUNCE, NET_SUBFUNC::ANNC_SUBFUNC_SITE_VC }, req->buffer, req->length, req->rtpHeader.getSequence(), streamId, true); } } } } } else { network->writePeerNAK(peerId, streamId, TAG_ANNOUNCE, NET_CONN_NAK_FNE_UNAUTHORIZED); } } } } break; default: { LogWarning(LOG_MASTER, "PEER %u, unknown/unsupported announcement opcode %u", peerId, req->fneHeader.getSubFunction()); if (network->m_debug) Utils::dump("Unknown/unsupported announcement opcode from the peer", req->buffer, req->length); } break; } } break; case NET_FUNC::KEYS_INVENTORY: // Encryption Key Container Inventory { if (!network->m_host->m_cryptoLookup->isRemoteAccessEnabled()) { LogError(LOG_MASTER, "PEER %u requested enc. key inventory, but remote access is disabled, no response", peerId); break; } lookups::PeerId peerEntry = network->m_peerListLookup->find(peerId); if (peerEntry.peerDefault()) { LogError(LOG_MASTER, "PEER %u requested enc. key inventory but is not allowed, no response", peerId); break; } else { if (!peerEntry.canRequestKeys()) { LogError(LOG_MASTER, "PEER %u requested enc. key inventory but is not allowed, no response", peerId); break; } } // keys inventory operates differently from the rest of the network opcodes...and does not require // an established connection to the master, so we will not validate the peer connection state here if (peerId > 0 && !peerEntry.peerDefault()) { if (req->length < 80) { LogError(LOG_MASTER, "PEER %u requested enc. key inventory, but payload length was invalid (%u bytes), no response", peerId, req->length); break; } // scope intentional { // get the peer password hash from the frame message DECLARE_UINT8_ARRAY(peerHash, 32U); ::memcpy(peerHash, req->buffer + 8U, 32U); uint8_t peerSalt[4U]; ::memset(peerSalt, 0x00U, 4U); ::memcpy(peerSalt, req->buffer + 40U, 4U); std::string passwordForPeer = network->m_password; // check if the peer is in the peer ACL list bool validAcl = true; if (network->m_peerListLookup->getACL()) { if (!network->m_peerListLookup->isPeerAllowed(peerId) && !network->m_peerListLookup->isPeerListEmpty()) { LogWarning(LOG_MASTER, "PEER %u RPTK, failed peer ACL check", peerId); validAcl = false; } else { lookups::PeerId peerEntry = network->m_peerListLookup->find(peerId); if (peerEntry.peerDefault()) { validAcl = false; // default peer IDs are a no-no as they have no data thus fail ACL check } else { passwordForPeer = peerEntry.peerPassword(); if (passwordForPeer.length() == 0) { passwordForPeer = network->m_password; } } } if (network->m_peerListLookup->isPeerListEmpty()) { LogWarning(LOG_MASTER, "Peer List ACL enabled, but we have an empty peer list? Passing all peers."); validAcl = true; } } if (validAcl) { size_t size = passwordForPeer.size(); uint8_t* in = new uint8_t[size + sizeof(uint32_t)]; ::memcpy(in, peerSalt, sizeof(uint32_t)); for (size_t i = 0U; i < size; i++) in[i + sizeof(uint32_t)] = passwordForPeer.at(i); uint8_t out[32U]; edac::SHA256 sha256; sha256.buffer(in, (uint32_t)(size + sizeof(uint32_t)), out); delete[] in; // validate hash bool validHash = false; if (req->length >= 80) { validHash = true; for (uint8_t i = 0; i < 32U; i++) { if (peerHash[i] != out[i]) { validHash = false; break; } } } if (!validHash) { LogError(LOG_MASTER, "PEER %u requested enc. key inventory, but had invalid authentication, no response", peerId); break; } } else { LogError(LOG_MASTER, "PEER %u requested enc. key inventory, but had invalid ACL, no response", peerId); break; } } // scope intentional { // get remote access password hash from the frame message DECLARE_UINT8_ARRAY(remoteAccessHash, 32U); ::memcpy(remoteAccessHash, req->buffer + 44U, 32U); uint8_t remoteSalt[4U]; ::memset(remoteSalt, 0x00U, 4U); ::memcpy(remoteSalt, req->buffer + 76U, 4U); std::string remoteAccessPassword = network->m_host->m_cryptoLookup->getRemotePassword(); size_t size = remoteAccessPassword.size(); uint8_t* in = new uint8_t[size + sizeof(uint32_t)]; ::memcpy(in, remoteSalt, sizeof(uint32_t)); for (size_t i = 0U; i < size; i++) in[i + sizeof(uint32_t)] = remoteAccessPassword.at(i); uint8_t out[32U]; edac::SHA256 sha256; sha256.buffer(in, (uint32_t)(size + sizeof(uint32_t)), out); delete[] in; // validate hash bool validHash = false; if (req->length >= 80) { validHash = true; for (uint8_t i = 0; i < 32U; i++) { if (remoteAccessHash[i] != out[i]) { validHash = false; break; } } } if (!validHash) { LogError(LOG_MASTER, "PEER %u requested enc. key inventory, but had invalid access authentication, no response", peerId); break; } } // scope intentional { // read entire file into buffer std::stringstream b; std::ifstream stream(network->m_host->m_cryptoLookup->filename(), std::ios::in | std::ios::binary); uint32_t len = 0U; UInt8Array bufferUInt8Array = nullptr; uint8_t* buffer = nullptr; if (stream.is_open()) { stream.seekg(0, std::ios::end); len = (uint32_t)stream.tellg(); stream.seekg(0, std::ios::beg); bufferUInt8Array = std::make_unique(len); buffer = bufferUInt8Array.get(); ::memset(buffer, 0x00U, len); uint32_t i = 0U; while (stream.peek() != EOF) { buffer[i] = (uint8_t)stream.get(); i++; } stream.close(); } PacketBuffer pkt(true, "Remote EKC, Key Inventory"); pkt.encode((uint8_t*)buffer, len); LogInfoEx(LOG_REPL, "PEER %u Remote EKC, Key Inventory, blocks %u, streamId = %u", peerId, pkt.fragments.size(), streamId); if (pkt.fragments.size() > 0U) { for (auto frag : pkt.fragments) { // violate most handling rules for responding to packets -- we need to directly respond to the calling peer as // they may not be logged in as a standard peer mdNetwork->m_frameQueue->write(frag.second->data, FRAG_SIZE, streamId, peerId, network->m_peerId, { NET_FUNC::KEYS_INVENTORY, NET_SUBFUNC::NOP }, 0U, req->address, req->addrLen); Thread::sleep(60U); // pace block transmission } } pkt.clear(); } } } break; case NET_FUNC::KEYS_UPDATE: // Encryption Key Container Update { if (!network->m_host->m_cryptoLookup->isRemoteAccessEnabled()) { LogError(LOG_MASTER, "PEER %u requested enc. key update, but remote access is disabled, no response", peerId); break; } lookups::PeerId peerEntry = network->m_peerListLookup->find(peerId); if (peerEntry.peerDefault()) { LogError(LOG_MASTER, "PEER %u requested enc. key update but is not allowed, no response", peerId); break; } else { if (!peerEntry.canRequestKeys()) { LogError(LOG_MASTER, "PEER %u requested enc. key update but is not allowed, no response", peerId); break; } } // keys update operates differently from the rest of the network opcodes...and does not require // an established connection to the master, so we will not validate the peer connection state here. // update is a two-phase flow: (1) auth request frame (80 bytes), then (2) chunked PacketBuffer frames. if (peerId > 0 && !peerEntry.peerDefault()) { if (mdNetwork->m_peerKeyUpdatePkt.find(peerId) == mdNetwork->m_peerKeyUpdatePkt.end()) { if (req->length < 80) { LogError(LOG_MASTER, "PEER %u requested enc. key update, but payload length was invalid (%u bytes), no response", peerId, req->length); break; } // scope intentional { // get the peer password hash from the frame message DECLARE_UINT8_ARRAY(peerHash, 32U); ::memcpy(peerHash, req->buffer + 8U, 32U); uint8_t peerSalt[4U]; ::memset(peerSalt, 0x00U, 4U); ::memcpy(peerSalt, req->buffer + 40U, 4U); std::string passwordForPeer = network->m_password; // check if the peer is in the peer ACL list bool validAcl = true; if (network->m_peerListLookup->getACL()) { if (!network->m_peerListLookup->isPeerAllowed(peerId) && !network->m_peerListLookup->isPeerListEmpty()) { LogWarning(LOG_MASTER, "PEER %u RPTK, failed peer ACL check", peerId); validAcl = false; } else { lookups::PeerId peerEntry = network->m_peerListLookup->find(peerId); if (peerEntry.peerDefault()) { validAcl = false; // default peer IDs are a no-no as they have no data thus fail ACL check } else { passwordForPeer = peerEntry.peerPassword(); if (passwordForPeer.length() == 0) { passwordForPeer = network->m_password; } } } if (network->m_peerListLookup->isPeerListEmpty()) { LogWarning(LOG_MASTER, "Peer List ACL enabled, but we have an empty peer list? Passing all peers."); validAcl = true; } } if (validAcl) { size_t size = passwordForPeer.size(); uint8_t* in = new uint8_t[size + sizeof(uint32_t)]; ::memcpy(in, peerSalt, sizeof(uint32_t)); for (size_t i = 0U; i < size; i++) in[i + sizeof(uint32_t)] = passwordForPeer.at(i); uint8_t out[32U]; edac::SHA256 sha256; sha256.buffer(in, (uint32_t)(size + sizeof(uint32_t)), out); delete[] in; // validate hash bool validHash = false; if (req->length >= 80) { validHash = true; for (uint8_t i = 0; i < 32U; i++) { if (peerHash[i] != out[i]) { validHash = false; break; } } } if (!validHash) { LogError(LOG_MASTER, "PEER %u requested enc. key update, but had invalid authentication, no response", peerId); break; } } else { LogError(LOG_MASTER, "PEER %u requested enc. key update, but had invalid ACL, no response", peerId); break; } } // scope intentional { // get remote access password hash from the frame message DECLARE_UINT8_ARRAY(remoteAccessHash, 32U); ::memcpy(remoteAccessHash, req->buffer + 44U, 32U); uint8_t remoteSalt[4U]; ::memset(remoteSalt, 0x00U, 4U); ::memcpy(remoteSalt, req->buffer + 76U, 4U); std::string remoteAccessPassword = network->m_host->m_cryptoLookup->getRemotePassword(); size_t size = remoteAccessPassword.size(); uint8_t* in = new uint8_t[size + sizeof(uint32_t)]; ::memcpy(in, remoteSalt, sizeof(uint32_t)); for (size_t i = 0U; i < size; i++) in[i + sizeof(uint32_t)] = remoteAccessPassword.at(i); uint8_t out[32U]; edac::SHA256 sha256; sha256.buffer(in, (uint32_t)(size + sizeof(uint32_t)), out); delete[] in; // validate hash bool validHash = false; if (req->length >= 80) { validHash = true; for (uint8_t i = 0; i < 32U; i++) { if (remoteAccessHash[i] != out[i]) { validHash = false; break; } } } if (!validHash) { LogError(LOG_MASTER, "PEER %u requested enc. key update, but had invalid access authentication, no response", peerId); break; } } mdNetwork->m_peerKeyUpdatePkt.insert(peerId, MetadataNetwork::PacketBufferEntry()); MetadataNetwork::PacketBufferEntry& pkt = mdNetwork->m_peerKeyUpdatePkt[peerId]; pkt.buffer = new PacketBuffer(true, "Remote EKC, Key Update"); pkt.streamId = streamId; pkt.locked = false; pkt.timeout = 0U; LogInfoEx(LOG_REPL, "PEER %u Remote EKC, Key Update, authenticated transfer streamId = %u", peerId, streamId); break; } // scope intentional { if (req->length < FRAG_SIZE) { LogWarning(LOG_REPL, "PEER %u Remote EKC, Key Update, ignoring short data phase frame (%u bytes)", peerId, req->length); break; } DECLARE_UINT8_ARRAY(rawPayload, req->length); ::memcpy(rawPayload, req->buffer, req->length); // Utils::dump(1U, "MetadataNetwork::taskNetworkRx(), KEYS_UPDATE, Raw Payload", rawPayload, req->length); MetadataNetwork::PacketBufferEntry& pkt = mdNetwork->m_peerKeyUpdatePkt[peerId]; if (!pkt.locked && pkt.streamId != streamId) { LogError(LOG_REPL, "PEER %u Remote EKC, Key Update, stream ID mismatch, expected %u, got %u", peerId, pkt.streamId, streamId); pkt.buffer->clear(); delete pkt.buffer; pkt.streamId = 0U; mdNetwork->m_peerKeyUpdatePkt.erase(peerId); break; } if (pkt.streamId != streamId) { // otherwise drop the packet break; } if (pkt.locked) { while (pkt.locked && pkt.timeout < TIMEOUT_MAX_REPL) { pkt.timeout++; Thread::sleep(1U); } if (pkt.timeout >= TIMEOUT_MAX_REPL) { LogError(LOG_STP, "PEER %u Remote EKC, Key Update, timeout waiting for packet buffer to unlock", peerId); pkt.buffer->clear(); delete pkt.buffer; pkt.streamId = 0U; mdNetwork->m_peerKeyUpdatePkt.erase(peerId); break; } } pkt.locked = true; pkt.timeout = 0U; uint32_t decompressedLen = 0U; uint8_t* decompressed = nullptr; if (pkt.buffer->decode(rawPayload, &decompressed, &decompressedLen)) { mdNetwork->m_peerKeyUpdatePkt.lock(); std::ostringstream s; s << network->m_cryptoLookup->filename(); std::string filename = s.str(); std::ofstream file(filename, std::ofstream::out); if (file.fail()) { LogError(LOG_PEER, "Cannot open the crypto container file - %s", filename.c_str()); pkt.buffer->clear(); delete pkt.buffer; pkt.streamId = 0U; if (decompressed != nullptr) { delete[] decompressed; } mdNetwork->m_peerKeyUpdatePkt.unlock(); mdNetwork->m_peerKeyUpdatePkt.erase(peerId); break; } for (uint32_t i = 0U; i < decompressedLen; i++) { file << (char)decompressed[i]; } file.close(); network->m_cryptoLookup->stop(true); network->m_cryptoLookup->reload(); pkt.buffer->clear(); delete pkt.buffer; pkt.streamId = 0U; if (decompressed != nullptr) { delete[] decompressed; } mdNetwork->m_peerKeyUpdatePkt.unlock(); mdNetwork->m_peerKeyUpdatePkt.erase(peerId); } else { pkt.locked = false; } } } } break; case NET_FUNC::REPL: if (req->fneHeader.getSubFunction() == NET_SUBFUNC::REPL_ACT_PEER_LIST) { // Peer Replication Active Peer List if (peerId > 0 && (network->m_peers.find(peerId) != network->m_peers.end())) { FNEPeerConnection* connection = network->m_peers[peerId]; if (connection != nullptr) { std::string ip = udp::Socket::address(req->address); // validate peer (simple validation really) if (connection->connected() && connection->address() == ip && connection->peerClass() == PEER_CONN_CLASS_NEIGHBOR && connection->isReplica()) { DECLARE_UINT8_ARRAY(rawPayload, req->length); ::memcpy(rawPayload, req->buffer, req->length); // Utils::dump(1U, "MetadataNetwork::taskNetworkRx(), REPL_ACT_PEER_LIST, Raw Payload", rawPayload, req->length); if (mdNetwork->m_peerReplicaActPkt.find(peerId) == mdNetwork->m_peerReplicaActPkt.end()) { mdNetwork->m_peerReplicaActPkt.insert(peerId, MetadataNetwork::PacketBufferEntry()); MetadataNetwork::PacketBufferEntry& pkt = mdNetwork->m_peerReplicaActPkt[peerId]; pkt.buffer = new PacketBuffer(true, "Peer Replication, Active Peer List"); pkt.streamId = streamId; pkt.locked = false; } else { MetadataNetwork::PacketBufferEntry& pkt = mdNetwork->m_peerReplicaActPkt[peerId]; if (!pkt.locked && pkt.streamId != streamId) { LogError(LOG_REPL, "PEER %u (%s) Peer Replication, Active Peer List, stream ID mismatch, expected %u, got %u", peerId, connection->identWithQualifier().c_str(), pkt.streamId, streamId); pkt.buffer->clear(); pkt.streamId = streamId; } if (pkt.streamId != streamId) { // otherwise drop the packet break; } } MetadataNetwork::PacketBufferEntry& pkt = mdNetwork->m_peerReplicaActPkt[peerId]; if (pkt.locked) { while (pkt.locked && pkt.timeout < TIMEOUT_MAX_REPL) { pkt.timeout++; Thread::sleep(1U); } if (pkt.timeout >= TIMEOUT_MAX_REPL) { LogError(LOG_STP, "PEER %u (%s) Peer Replication, Active Peer List, timeout waiting for packet buffer to unlock", peerId, connection->identWithQualifier().c_str()); pkt.buffer->clear(); pkt.streamId = 0U; mdNetwork->m_peerReplicaActPkt.erase(peerId); break; } } pkt.locked = true; pkt.timeout = 0U; uint32_t decompressedLen = 0U; uint8_t* decompressed = nullptr; if (pkt.buffer->decode(rawPayload, &decompressed, &decompressedLen)) { mdNetwork->m_peerReplicaActPkt.lock(); std::string payload(decompressed + 8U, decompressed + decompressedLen); // parse JSON body json::value v; std::string err = json::parse(v, payload); if (!err.empty()) { LogError(LOG_REPL, "PEER %u (%s) error parsing active peer list, %s", peerId, connection->identWithQualifier().c_str(), err.c_str()); pkt.buffer->clear(); pkt.streamId = 0U; if (decompressed != nullptr) { delete[] decompressed; } mdNetwork->m_peerReplicaActPkt.unlock(); mdNetwork->m_peerReplicaActPkt.erase(peerId); break; } else { // ensure parsed JSON is an array if (!v.is()) { LogError(LOG_REPL, "PEER %u (%s) error parsing active peer list, data was not valid", peerId, connection->identWithQualifier().c_str()); pkt.buffer->clear(); delete pkt.buffer; pkt.streamId = 0U; if (decompressed != nullptr) { delete[] decompressed; } mdNetwork->m_peerReplicaActPkt.unlock(); mdNetwork->m_peerReplicaActPkt.erase(peerId); break; } else { json::array arr = v.get(); LogInfoEx(LOG_REPL, "PEER %u (%s) Peer Replication, Active Peer List, updating %u peer entries", peerId, connection->identWithQualifier().c_str(), arr.size()); network->m_peerReplicaPeers[peerId] = arr; } } pkt.buffer->clear(); delete pkt.buffer; pkt.streamId = 0U; if (decompressed != nullptr) { delete[] decompressed; } mdNetwork->m_peerReplicaActPkt.unlock(); mdNetwork->m_peerReplicaActPkt.erase(peerId); } else { pkt.locked = false; } } else { network->writePeerNAK(peerId, 0U, TAG_PEER_REPLICA, NET_CONN_NAK_FNE_UNAUTHORIZED); } } } } else if (req->fneHeader.getSubFunction() == NET_SUBFUNC::REPL_HA_PARAMS) { // Peer Replication HA Parameters if (peerId > 0 && (network->m_peers.find(peerId) != network->m_peers.end())) { FNEPeerConnection* connection = network->m_peers[peerId]; if (connection != nullptr) { std::string ip = udp::Socket::address(req->address); // validate peer (simple validation really) if (connection->connected() && connection->address() == ip && connection->peerClass() == PEER_CONN_CLASS_NEIGHBOR && connection->isReplica()) { DECLARE_UINT8_ARRAY(rawPayload, req->length); ::memcpy(rawPayload, req->buffer, req->length); std::vector receivedParams; uint32_t len = GET_UINT32(rawPayload, 0U); if (len > 0U) { len /= HA_PARAMS_ENTRY_LEN; } uint8_t offs = 4U; for (uint8_t i = 0U; i < len; i++, offs += HA_PARAMS_ENTRY_LEN) { uint32_t peerId = GET_UINT32(rawPayload, offs); uint32_t ipAddr = GET_UINT32(rawPayload, offs + 4U); uint16_t port = GET_UINT16(rawPayload, offs + 8U); receivedParams.push_back(HAParameters(peerId, ipAddr, port)); } if (receivedParams.size() > 0U) { for (auto rxEntry : receivedParams) { auto it = std::find_if(network->m_peerReplicaHAParams.begin(), network->m_peerReplicaHAParams.end(), [&](HAParameters& x) { if (x.peerId == rxEntry.peerId) return true; return false; }); if (it != network->m_peerReplicaHAParams.end()) { it->masterIP = rxEntry.masterIP; it->masterPort = rxEntry.masterPort; } else { HAParameters param = rxEntry; network->m_peerReplicaHAParams.push_back(param); } if (network->m_debug) { std::string address = __IP_FROM_UINT(rxEntry.masterIP); LogDebugEx(LOG_REPL, "MetadataNetwork::taskNetworkRx()", "PEER %u (%s) Peer Replication, HA Parameters, %s:%u", peerId, connection->identWithQualifier().c_str(), address.c_str(), rxEntry.masterPort); } } if (receivedParams.size() > 0) { LogInfoEx(LOG_REPL, "PEER %u (%s) Peer Replication, HA Parameters, updating %u entries, %u entries", peerId, connection->identWithQualifier().c_str(), receivedParams.size(), network->m_peerReplicaHAParams.size()); // send peer updates to replica peers if (network->m_host->m_peerNetworks.size() > 0) { for (auto peer : network->m_host->m_peerNetworks) { if (peer.second != nullptr) { if (peer.second->isEnabled() && peer.second->isReplica()) { std::vector haParams; network->m_peerReplicaHAParams.lock(false); for (auto entry : network->m_peerReplicaHAParams) { haParams.push_back(entry); } network->m_peerReplicaHAParams.unlock(); peer.second->writeHAParams(haParams); } } } } } } } else { network->writePeerNAK(peerId, 0U, TAG_PEER_REPLICA, NET_CONN_NAK_FNE_UNAUTHORIZED); } } } } else if (req->fneHeader.getSubFunction() == NET_SUBFUNC::REPL_PATCH_STATUS) { // Peer Replication Patch Status if (peerId > 0 && (network->m_peers.find(peerId) != network->m_peers.end())) { FNEPeerConnection* connection = network->m_peers[peerId]; if (connection != nullptr) { std::string ip = udp::Socket::address(req->address); // validate peer (simple validation really) if (connection->connected() && connection->address() == ip && connection->peerClass() == PEER_CONN_CLASS_NEIGHBOR && connection->isReplica()) { DECLARE_UINT8_ARRAY(rawPayload, req->length); ::memcpy(rawPayload, req->buffer, req->length); if (mdNetwork->m_peerPatchStatusPkt.find(peerId) == mdNetwork->m_peerPatchStatusPkt.end()) { mdNetwork->m_peerPatchStatusPkt.insert(peerId, MetadataNetwork::PacketBufferEntry()); MetadataNetwork::PacketBufferEntry& pkt = mdNetwork->m_peerPatchStatusPkt[peerId]; pkt.buffer = new PacketBuffer(true, "Peer Replication, Patch Status"); pkt.streamId = streamId; pkt.locked = false; } else { MetadataNetwork::PacketBufferEntry& pkt = mdNetwork->m_peerPatchStatusPkt[peerId]; if (!pkt.locked && pkt.streamId != streamId) { LogError(LOG_REPL, "PEER %u (%s) Peer Replication, Patch Status, stream ID mismatch, expected %u, got %u", peerId, connection->identWithQualifier().c_str(), pkt.streamId, streamId); pkt.buffer->clear(); pkt.streamId = streamId; } if (pkt.streamId != streamId) { // otherwise drop the packet break; } } MetadataNetwork::PacketBufferEntry& pkt = mdNetwork->m_peerPatchStatusPkt[peerId]; if (pkt.locked) { while (pkt.locked) Thread::sleep(1U); } pkt.locked = true; uint32_t decompressedLen = 0U; uint8_t* decompressed = nullptr; if (pkt.buffer->decode(rawPayload, &decompressed, &decompressedLen)) { mdNetwork->m_peerPatchStatusPkt.lock(); std::string payload(decompressed + 8U, decompressed + decompressedLen); json::value v; std::string err = json::parse(v, payload); if (!err.empty() || !v.is()) { LogError(LOG_REPL, "PEER %u (%s) error parsing patch status replication, %s", peerId, connection->identWithQualifier().c_str(), err.c_str()); pkt.buffer->clear(); pkt.streamId = 0U; if (decompressed != nullptr) { delete[] decompressed; } mdNetwork->m_peerPatchStatusPkt.unlock(); mdNetwork->m_peerPatchStatusPkt.erase(peerId); break; } network->processReplicatedPatchStatus(peerId, v.get()); pkt.buffer->clear(); delete pkt.buffer; pkt.streamId = 0U; if (decompressed != nullptr) { delete[] decompressed; } mdNetwork->m_peerPatchStatusPkt.unlock(); mdNetwork->m_peerPatchStatusPkt.erase(peerId); } else { pkt.locked = false; } } else { network->writePeerNAK(peerId, 0U, TAG_PEER_REPLICA, NET_CONN_NAK_FNE_UNAUTHORIZED); } } } } break; case NET_FUNC::NET_TREE: if (!network->m_enableSpanningTree) break; if (req->fneHeader.getSubFunction() == NET_SUBFUNC::NET_TREE_LIST) { // FNE Network Tree List if (peerId > 0 && (network->m_peers.find(peerId) != network->m_peers.end())) { FNEPeerConnection* connection = network->m_peers[peerId]; if (connection != nullptr) { std::string ip = udp::Socket::address(req->address); // validate peer (simple validation really) if (connection->connected() && connection->address() == ip && connection->peerClass() == PEER_CONN_CLASS_NEIGHBOR) { DECLARE_UINT8_ARRAY(rawPayload, req->length); ::memcpy(rawPayload, req->buffer, req->length); // Utils::dump(1U, "MetadataNetwork::taskNetworkRx(), NET_TREE_LIST, Raw Payload", rawPayload, req->length); if (mdNetwork->m_peerTreeListPkt.find(peerId) == mdNetwork->m_peerTreeListPkt.end()) { mdNetwork->m_peerTreeListPkt.insert(peerId, MetadataNetwork::PacketBufferEntry()); MetadataNetwork::PacketBufferEntry& pkt = mdNetwork->m_peerTreeListPkt[peerId]; pkt.buffer = new PacketBuffer(true, "Network Tree, Tree List"); pkt.streamId = streamId; pkt.locked = false; } else { MetadataNetwork::PacketBufferEntry& pkt = mdNetwork->m_peerTreeListPkt[peerId]; if (!pkt.locked && pkt.streamId != streamId) { LogError(LOG_STP, "PEER %u (%s) Network Tree, Tree List, stream ID mismatch, expected %u, got %u", peerId, connection->identWithQualifier().c_str(), pkt.streamId, streamId); pkt.buffer->clear(); pkt.streamId = streamId; } if (pkt.streamId != streamId) { // otherwise drop the packet break; } } MetadataNetwork::PacketBufferEntry& pkt = mdNetwork->m_peerTreeListPkt[peerId]; if (pkt.locked) { while (pkt.locked && pkt.timeout < TIMEOUT_MAX_REPL) { pkt.timeout++; Thread::sleep(1U); } if (pkt.timeout >= TIMEOUT_MAX_REPL) { LogError(LOG_STP, "PEER %u (%s) Network Tree, Tree List, timeout waiting for packet buffer to unlock", peerId, connection->identWithQualifier().c_str()); pkt.buffer->clear(); pkt.streamId = 0U; mdNetwork->m_peerTreeListPkt.erase(peerId); break; } } pkt.locked = true; pkt.timeout = 0U; uint32_t decompressedLen = 0U; uint8_t* decompressed = nullptr; if (pkt.buffer->decode(rawPayload, &decompressed, &decompressedLen)) { mdNetwork->m_peerTreeListPkt.lock(); std::string payload(decompressed + 8U, decompressed + decompressedLen); // parse JSON body json::value v; std::string err = json::parse(v, payload); if (!err.empty()) { LogError(LOG_STP, "PEER %u (%s) error parsing network tree list, %s", peerId, connection->identWithQualifier().c_str(), err.c_str()); pkt.buffer->clear(); pkt.streamId = 0U; if (decompressed != nullptr) { delete[] decompressed; } mdNetwork->m_peerTreeListPkt.unlock(); mdNetwork->m_peerTreeListPkt.erase(peerId); break; } else { // ensure parsed JSON is an array if (!v.is()) { LogError(LOG_STP, "PEER %u (%s) error parsing network tree list, data was not valid", peerId, connection->identWithQualifier().c_str()); pkt.buffer->clear(); delete pkt.buffer; pkt.streamId = 0U; if (decompressed != nullptr) { delete[] decompressed; } mdNetwork->m_peerTreeListPkt.unlock(); mdNetwork->m_peerTreeListPkt.erase(peerId); break; } else { json::array arr = v.get(); LogInfoEx(LOG_STP, "PEER %u (%s) Network Tree, Tree List, updating %u peer entries", peerId, connection->identWithQualifier().c_str(), arr.size()); std::lock_guard guard(network->m_treeLock); std::vector duplicatePeers; SpanningTree::deserializeTree(arr, network->m_treeRoot, &duplicatePeers); network->logSpanningTree(connection); if (duplicatePeers.size() > 0U) { for (auto dupPeerId : duplicatePeers) { LogWarning(LOG_STP, "PEER %u (%s) Network Tree, Tree Change, disconnecting duplicate peer connection for PEER %u to prevent network loop", peerId, connection->identWithQualifier().c_str(), dupPeerId); network->writeTreeDisconnect(peerId, dupPeerId); } } } } pkt.buffer->clear(); delete pkt.buffer; pkt.streamId = 0U; if (decompressed != nullptr) { delete[] decompressed; } mdNetwork->m_peerTreeListPkt.unlock(); mdNetwork->m_peerTreeListPkt.erase(peerId); } else { pkt.locked = false; } } else { network->writePeerNAK(peerId, 0U, TAG_PEER_REPLICA, NET_CONN_NAK_FNE_UNAUTHORIZED); } } } } break; default: // metadata network ignores unknowns for everything else... break; } } if (req->buffer != nullptr) delete[] req->buffer; delete req; } }