diff --git a/src/fne/network/TrafficNetwork.h b/src/fne/network/TrafficNetwork.h index 57447d25..9bd4bdf4 100644 --- a/src/fne/network/TrafficNetwork.h +++ b/src/fne/network/TrafficNetwork.h @@ -93,6 +93,7 @@ namespace network #define INFLUXDB_ERRSTR_INV_SLOT "invalid slot for talkgroup" #define INFLUXDB_ERRSTR_RID_NOT_PERMITTED "RID not permitted for talkgroup" #define INFLUXDB_ERRSTR_ILLEGAL_RID_ACCESS "illegal/unknown RID attempted access" + #define INFLUXDB_ERRSTR_CALL_NOT_PERMITTED "call not permitted for talkgroup" // --------------------------------------------------------------------------- // Class Prototypes diff --git a/src/fne/network/callhandler/TagAnalogData.cpp b/src/fne/network/callhandler/TagAnalogData.cpp index da3dba29..81580931 100644 --- a/src/fne/network/callhandler/TagAnalogData.cpp +++ b/src/fne/network/callhandler/TagAnalogData.cpp @@ -43,6 +43,7 @@ TagAnalogData::TagAnalogData(TrafficNetwork* network, bool debug) : m_lastParrotSrcId(0U), m_lastParrotDstId(0U), m_status(), + m_rejectedCallStreams(), m_debug(debug) { assert(network != nullptr); @@ -130,6 +131,11 @@ bool TagAnalogData::processFrame(const uint8_t* data, uint32_t len, uint32_t pee else if (!fromUpstream) LogInfoEx(LOG_MASTER, CALL_END_LOG); + // clear any rejected call streams for this TG + m_rejectedCallStreams.lock(false); + m_rejectedCallStreams[dstId].clear(); + m_rejectedCallStreams.unlock(); + if (!tg.config().parrot()) { m_network->m_totalActiveCalls--; if (m_network->m_totalActiveCalls < 0) @@ -217,11 +223,21 @@ bool TagAnalogData::processFrame(const uint8_t* data, uint32_t len, uint32_t pee m_status[dstId].srcId = srcId; m_status[dstId].ssrc = ssrc; m_status.unlock(); + + // because the call stream source has reset, clear any rejected call streams for + // this TG to allow the new source to transmit + m_rejectedCallStreams.lock(false); + m_rejectedCallStreams[dstId].clear(); + m_rejectedCallStreams.unlock(); } else { LogWarning((fromUpstream) ? LOG_PEER : LOG_MASTER, "Analog, Call Collision, peer = %u, ssrc = %u, srcId = %u, dstId = %u, streamId = %u, rxPeer = %u, rxSrcId = %u, rxDstId = %u, rxStreamId = %u, fromUpstream = %u", peerId, ssrc, srcId, dstId, streamId, status.peerId, status.srcId, status.dstId, status.streamId, fromUpstream); + m_rejectedCallStreams.lock(false); + m_rejectedCallStreams[dstId].push_back(streamId); + m_rejectedCallStreams.unlock(); + m_network->m_totalCallCollisions++; return false; @@ -277,6 +293,11 @@ bool TagAnalogData::processFrame(const uint8_t* data, uint32_t len, uint32_t pee m_status[dstId].activeCall = true; m_status.unlock(); + // clear any rejected call streams for this TG + m_rejectedCallStreams.lock(false); + m_rejectedCallStreams[dstId].clear(); + m_rejectedCallStreams.unlock(); + if (!tg.config().parrot()) { m_network->m_totalCallsProcessed++; m_network->m_totalActiveCalls++; @@ -705,6 +726,34 @@ bool TagAnalogData::validate(uint32_t peerId, data::NetData& data, uint32_t stre if (data.getFrameType() == AudioFrameType::TERMINATOR) return true; + // is the call stream rejected? + m_rejectedCallStreams.lock(false); + std::vector rejectedStreams = m_rejectedCallStreams[data.getDstId()]; + if (std::find(rejectedStreams.begin(), rejectedStreams.end(), streamId) != rejectedStreams.end()) { + // report error event to InfluxDB + if (m_network->m_enableInfluxDB) { + influxdb::QueryBuilder() + .meas("call_error_event") + .tag("peerId", std::to_string(peerId)) + .tag("streamId", std::to_string(streamId)) + .tag("srcId", std::to_string(data.getSrcId())) + .tag("dstId", std::to_string(data.getDstId())) + .field("message", std::string(INFLUXDB_ERRSTR_CALL_NOT_PERMITTED)) + .timestamp(std::chrono::duration_cast(std::chrono::system_clock::now().time_since_epoch()).count()) + .requestAsync(m_network->m_influxServer); + } + + if (m_network->m_logDenials) + LogError(LOG_ANALOG, INFLUXDB_ERRSTR_CALL_NOT_PERMITTED ", peer = %u, srcId = %u, dstId = %u", peerId, data.getSrcId(), data.getDstId()); + + m_rejectedCallStreams.unlock(); + + // report In-Call Control to the peer sending traffic + m_network->writePeerICC(peerId, streamId, NET_SUBFUNC::PROTOCOL_SUBFUNC_ANALOG, NET_ICC::REJECT_TRAFFIC, data.getDstId()); + return false; + } + m_rejectedCallStreams.unlock(); + // is this a private call? if (!data.getGroup()) { // is the destination ID a blacklisted ID? diff --git a/src/fne/network/callhandler/TagAnalogData.h b/src/fne/network/callhandler/TagAnalogData.h index 3b3efaae..91f288e7 100644 --- a/src/fne/network/callhandler/TagAnalogData.h +++ b/src/fne/network/callhandler/TagAnalogData.h @@ -4,7 +4,7 @@ * GPLv2 Open Source. Use is subject to license terms. * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER. * - * Copyright (C) 2025 Bryan Biedenkapp, N2PLL + * Copyright (C) 2025-2026 Bryan Biedenkapp, N2PLL * */ /** @@ -203,6 +203,7 @@ namespace network }; typedef std::pair StatusMapPair; concurrent::unordered_map m_status; + concurrent::unordered_map> m_rejectedCallStreams; bool m_debug; diff --git a/src/fne/network/callhandler/TagDMRData.cpp b/src/fne/network/callhandler/TagDMRData.cpp index 4a4046b1..2844717d 100644 --- a/src/fne/network/callhandler/TagDMRData.cpp +++ b/src/fne/network/callhandler/TagDMRData.cpp @@ -48,6 +48,7 @@ TagDMRData::TagDMRData(TrafficNetwork* network, bool debug) : m_lastParrotDstId(0U), m_status(), m_statusPVCall(), + m_rejectedCallStreams(), m_debug(debug) { assert(network != nullptr); @@ -213,6 +214,11 @@ bool TagDMRData::processFrame(const uint8_t* data, uint32_t len, uint32_t peerId LogInfoEx(LOG_MASTER, CALL_END_LOG); } + // clear any rejected call streams for this TG + m_rejectedCallStreams.lock(false); + m_rejectedCallStreams[dstId].clear(); + m_rejectedCallStreams.unlock(); + if (!tg.config().parrot()) { m_network->m_totalActiveCalls--; if (m_network->m_totalActiveCalls < 0) @@ -329,10 +335,20 @@ bool TagDMRData::processFrame(const uint8_t* data, uint32_t len, uint32_t peerId m_status[dstId].srcId = srcId; m_status[dstId].ssrc = ssrc; m_status.unlock(); + + // because the call stream source has reset, clear any rejected call streams for + // this TG to allow the new source to transmit + m_rejectedCallStreams.lock(false); + m_rejectedCallStreams[dstId].clear(); + m_rejectedCallStreams.unlock(); } else { LogWarning((fromUpstream) ? LOG_PEER : LOG_MASTER, "DMR, Call Collision, peer = %u, ssrc = %u, srcId = %u, dstId = %u, slotNo = %u, streamId = %u, rxPeer = %u, rxSrcId = %u, rxDstId = %u, rxSlotNo = %u, rxStreamId = %u, fromUpstream = %u", peerId, ssrc, srcId, dstId, slotNo, streamId, status.peerId, status.srcId, status.dstId, status.slotNo, status.streamId, fromUpstream); + m_rejectedCallStreams.lock(false); + m_rejectedCallStreams[dstId].push_back(streamId); + m_rejectedCallStreams.unlock(); + m_network->m_totalCallCollisions++; return false; @@ -391,6 +407,11 @@ bool TagDMRData::processFrame(const uint8_t* data, uint32_t len, uint32_t peerId m_status[dstId].activeCall = true; m_status.unlock(); + // clear any rejected call streams for this TG + m_rejectedCallStreams.lock(false); + m_rejectedCallStreams[dstId].clear(); + m_rejectedCallStreams.unlock(); + if (!tg.config().parrot()) { m_network->m_totalCallsProcessed++; m_network->m_totalActiveCalls++; @@ -1192,6 +1213,35 @@ bool TagDMRData::validate(uint32_t peerId, data::NetData& data, lc::CSBK* csbk, return true; } + // is the call stream rejected? + m_rejectedCallStreams.lock(false); + std::vector rejectedStreams = m_rejectedCallStreams[data.getDstId()]; + if (std::find(rejectedStreams.begin(), rejectedStreams.end(), streamId) != rejectedStreams.end()) { + // report error event to InfluxDB + if (m_network->m_enableInfluxDB) { + influxdb::QueryBuilder() + .meas("call_error_event") + .tag("peerId", std::to_string(peerId)) + .tag("streamId", std::to_string(streamId)) + .tag("srcId", std::to_string(data.getSrcId())) + .tag("dstId", std::to_string(data.getDstId())) + .field("message", std::string(INFLUXDB_ERRSTR_CALL_NOT_PERMITTED)) + .field("slot", std::to_string(data.getSlotNo())) + .timestamp(std::chrono::duration_cast(std::chrono::system_clock::now().time_since_epoch()).count()) + .requestAsync(m_network->m_influxServer); + } + + if (m_network->m_logDenials) + LogError(LOG_DMR, "DMR Slot %u, " INFLUXDB_ERRSTR_CALL_NOT_PERMITTED ", peer = %u, srcId = %u, dstId = %u", data.getSlotNo(), peerId, data.getSrcId(), data.getDstId()); + + m_rejectedCallStreams.unlock(); + + // report In-Call Control to the peer sending traffic + m_network->writePeerICC(peerId, streamId, NET_SUBFUNC::PROTOCOL_SUBFUNC_DMR, NET_ICC::REJECT_TRAFFIC, data.getDstId(), data.getSlotNo()); + return false; + } + m_rejectedCallStreams.unlock(); + // is this a private call? if (data.getFLCO() == FLCO::PRIVATE) { // is the destination ID a blacklisted ID? diff --git a/src/fne/network/callhandler/TagDMRData.h b/src/fne/network/callhandler/TagDMRData.h index 4890a6cb..1c513133 100644 --- a/src/fne/network/callhandler/TagDMRData.h +++ b/src/fne/network/callhandler/TagDMRData.h @@ -4,7 +4,7 @@ * GPLv2 Open Source. Use is subject to license terms. * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER. * - * Copyright (C) 2023-2024 Bryan Biedenkapp, N2PLL + * Copyright (C) 2023-2026 Bryan Biedenkapp, N2PLL * */ /** @@ -254,6 +254,7 @@ namespace network typedef std::pair StatusMapPair; concurrent::unordered_map m_status; concurrent::unordered_map m_statusPVCall; + concurrent::unordered_map> m_rejectedCallStreams; friend class packetdata::DMRPacketData; packetdata::DMRPacketData* m_packetData; diff --git a/src/fne/network/callhandler/TagNXDNData.cpp b/src/fne/network/callhandler/TagNXDNData.cpp index a11c895f..ed131469 100644 --- a/src/fne/network/callhandler/TagNXDNData.cpp +++ b/src/fne/network/callhandler/TagNXDNData.cpp @@ -51,6 +51,7 @@ TagNXDNData::TagNXDNData(TrafficNetwork* network, bool debug) : m_lastParrotDstId(0U), m_status(), m_statusPVCall(), + m_rejectedCallStreams(), m_debug(debug) { assert(network != nullptr); @@ -252,6 +253,11 @@ bool TagNXDNData::processFrame(const uint8_t* data, uint32_t len, uint32_t peerI LogInfoEx(LOG_MASTER, CALL_END_LOG); } + // clear any rejected call streams for this TG + m_rejectedCallStreams.lock(false); + m_rejectedCallStreams[dstId].clear(); + m_rejectedCallStreams.unlock(); + if (!tg.config().parrot()) { m_network->m_totalActiveCalls--; if (m_network->m_totalActiveCalls < 0) @@ -357,10 +363,20 @@ bool TagNXDNData::processFrame(const uint8_t* data, uint32_t len, uint32_t peerI m_status[dstId].srcId = srcId; m_status[dstId].ssrc = ssrc; m_status.unlock(); + + // because the call stream source has reset, clear any rejected call streams for + // this TG to allow the new source to transmit + m_rejectedCallStreams.lock(false); + m_rejectedCallStreams[dstId].clear(); + m_rejectedCallStreams.unlock(); } else { LogWarning((fromUpstream) ? LOG_PEER : LOG_MASTER, "NXDN, Call Collision, peer = %u, ssrc = %u, srcId = %u, dstId = %u, streamId = %u, rxPeer = %u, rxSrcId = %u, rxDstId = %u, rxStreamId = %u, fromUpstream = %u", peerId, ssrc, srcId, dstId, streamId, status.peerId, status.srcId, status.dstId, status.streamId, fromUpstream); + m_rejectedCallStreams.lock(false); + m_rejectedCallStreams[dstId].push_back(streamId); + m_rejectedCallStreams.unlock(); + m_network->m_totalCallCollisions++; return false; @@ -417,6 +433,11 @@ bool TagNXDNData::processFrame(const uint8_t* data, uint32_t len, uint32_t peerI m_status[dstId].activeCall = true; m_status.unlock(); + // clear any rejected call streams for this TG + m_rejectedCallStreams.lock(false); + m_rejectedCallStreams[dstId].clear(); + m_rejectedCallStreams.unlock(); + if (!tg.config().parrot()) { m_network->m_totalCallsProcessed++; m_network->m_totalActiveCalls++; @@ -1004,6 +1025,34 @@ bool TagNXDNData::validate(uint32_t peerId, lc::RTCH& lc, uint8_t messageType, u if (messageType == MessageType::RTCH_TX_REL || messageType == MessageType::RTCH_TX_REL_EX) return true; + // is the call stream rejected? + m_rejectedCallStreams.lock(false); + std::vector rejectedStreams = m_rejectedCallStreams[lc.getDstId()]; + if (std::find(rejectedStreams.begin(), rejectedStreams.end(), streamId) != rejectedStreams.end()) { + // report error event to InfluxDB + if (m_network->m_enableInfluxDB) { + influxdb::QueryBuilder() + .meas("call_error_event") + .tag("peerId", std::to_string(peerId)) + .tag("streamId", std::to_string(streamId)) + .tag("srcId", std::to_string(lc.getSrcId())) + .tag("dstId", std::to_string(lc.getDstId())) + .field("message", std::string(INFLUXDB_ERRSTR_CALL_NOT_PERMITTED)) + .timestamp(std::chrono::duration_cast(std::chrono::system_clock::now().time_since_epoch()).count()) + .requestAsync(m_network->m_influxServer); + } + + if (m_network->m_logDenials) + LogError(LOG_NXDN, INFLUXDB_ERRSTR_CALL_NOT_PERMITTED ", peer = %u, srcId = %u, dstId = %u", peerId, lc.getSrcId(), lc.getDstId()); + + m_rejectedCallStreams.unlock(); + + // report In-Call Control to the peer sending traffic + m_network->writePeerICC(peerId, streamId, NET_SUBFUNC::PROTOCOL_SUBFUNC_NXDN, NET_ICC::REJECT_TRAFFIC, lc.getDstId()); + return false; + } + m_rejectedCallStreams.unlock(); + // is this a private call? if (!lc.getGroup()) { // is the destination ID a blacklisted ID? diff --git a/src/fne/network/callhandler/TagNXDNData.h b/src/fne/network/callhandler/TagNXDNData.h index 4056901b..cc32301b 100644 --- a/src/fne/network/callhandler/TagNXDNData.h +++ b/src/fne/network/callhandler/TagNXDNData.h @@ -4,7 +4,7 @@ * GPLv2 Open Source. Use is subject to license terms. * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER. * - * Copyright (C) 2023-2024 Bryan Biedenkapp, N2PLL + * Copyright (C) 2023-2026 Bryan Biedenkapp, N2PLL * */ /** @@ -218,6 +218,7 @@ namespace network typedef std::pair StatusMapPair; concurrent::unordered_map m_status; concurrent::unordered_map m_statusPVCall; + concurrent::unordered_map> m_rejectedCallStreams; bool m_debug; diff --git a/src/fne/network/callhandler/TagP25Data.cpp b/src/fne/network/callhandler/TagP25Data.cpp index cc0c319e..a338d595 100644 --- a/src/fne/network/callhandler/TagP25Data.cpp +++ b/src/fne/network/callhandler/TagP25Data.cpp @@ -48,6 +48,7 @@ TagP25Data::TagP25Data(TrafficNetwork* network, bool debug) : m_lastParrotDstId(0U), m_status(), m_statusPVCall(), + m_rejectedCallStreams(), m_packetData(nullptr), m_debug(debug) { @@ -320,6 +321,11 @@ bool TagP25Data::processFrame(const uint8_t* data, uint32_t len, uint32_t peerId LogInfoEx(LOG_MASTER, CALL_END_LOG); } + // clear any rejected call streams for this TG + m_rejectedCallStreams.lock(false); + m_rejectedCallStreams[dstId].clear(); + m_rejectedCallStreams.unlock(); + if (!tg.config().parrot()) { m_network->m_totalActiveCalls--; if (m_network->m_totalActiveCalls < 0) @@ -435,11 +441,21 @@ bool TagP25Data::processFrame(const uint8_t* data, uint32_t len, uint32_t peerId m_status[dstId].srcId = srcId; m_status[dstId].ssrc = ssrc; m_status.unlock(); + + // because the call stream source has reset, clear any rejected call streams for + // this TG to allow the new source to transmit + m_rejectedCallStreams.lock(false); + m_rejectedCallStreams[dstId].clear(); + m_rejectedCallStreams.unlock(); } else { LogWarning((fromUpstream) ? LOG_PEER : LOG_MASTER, "P25, Call Collision, peer = %u, ssrc = %u, sysId = $%03X, netId = $%05X, srcId = %u, dstId = %u, streamId = %u, rxPeer = %u, rxSrcId = %u, rxDstId = %u, rxStreamId = %u, fromUpstream = %u", peerId, ssrc, sysId, netId, srcId, dstId, streamId, status.peerId, status.srcId, status.dstId, status.streamId, fromUpstream); + m_rejectedCallStreams.lock(false); + m_rejectedCallStreams[dstId].push_back(streamId); + m_rejectedCallStreams.unlock(); + m_network->m_totalCallCollisions++; return false; @@ -496,6 +512,11 @@ bool TagP25Data::processFrame(const uint8_t* data, uint32_t len, uint32_t peerId m_status[dstId].activeCall = true; m_status.unlock(); + // clear any rejected call streams for this TG + m_rejectedCallStreams.lock(false); + m_rejectedCallStreams[dstId].clear(); + m_rejectedCallStreams.unlock(); + if (!tg.config().parrot()) { m_network->m_totalCallsProcessed++; m_network->m_totalActiveCalls++; @@ -1645,6 +1666,34 @@ bool TagP25Data::validate(uint32_t peerId, lc::LC& control, DUID::E duid, const return true; } + // is the call stream rejected? + m_rejectedCallStreams.lock(false); + std::vector rejectedStreams = m_rejectedCallStreams[control.getDstId()]; + if (std::find(rejectedStreams.begin(), rejectedStreams.end(), streamId) != rejectedStreams.end()) { + // report error event to InfluxDB + if (m_network->m_enableInfluxDB) { + influxdb::QueryBuilder() + .meas("call_error_event") + .tag("peerId", std::to_string(peerId)) + .tag("streamId", std::to_string(streamId)) + .tag("srcId", std::to_string(control.getSrcId())) + .tag("dstId", std::to_string(control.getDstId())) + .field("message", std::string(INFLUXDB_ERRSTR_CALL_NOT_PERMITTED)) + .timestamp(std::chrono::duration_cast(std::chrono::system_clock::now().time_since_epoch()).count()) + .requestAsync(m_network->m_influxServer); + } + + if (m_network->m_logDenials) + LogError(LOG_P25, INFLUXDB_ERRSTR_CALL_NOT_PERMITTED ", peer = %u, srcId = %u, dstId = %u", peerId, control.getSrcId(), control.getDstId()); + + m_rejectedCallStreams.unlock(); + + // report In-Call Control to the peer sending traffic + m_network->writePeerICC(peerId, streamId, NET_SUBFUNC::PROTOCOL_SUBFUNC_P25, NET_ICC::REJECT_TRAFFIC, control.getDstId()); + return false; + } + m_rejectedCallStreams.unlock(); + // validate private call in-progress bool privateCallInProgress = false; if ((control.getLCO() != LCO::PRIVATE) && !control.getGroup()) { diff --git a/src/fne/network/callhandler/TagP25Data.h b/src/fne/network/callhandler/TagP25Data.h index 64c70dd2..11b4f787 100644 --- a/src/fne/network/callhandler/TagP25Data.h +++ b/src/fne/network/callhandler/TagP25Data.h @@ -4,7 +4,7 @@ * GPLv2 Open Source. Use is subject to license terms. * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER. * - * Copyright (C) 2023-2024 Bryan Biedenkapp, N2PLL + * Copyright (C) 2023-2026 Bryan Biedenkapp, N2PLL * */ /** @@ -267,6 +267,7 @@ namespace network typedef std::pair StatusMapPair; concurrent::unordered_map m_status; concurrent::unordered_map m_statusPVCall; + concurrent::unordered_map> m_rejectedCallStreams; friend class packetdata::P25PacketData; packetdata::P25PacketData* m_packetData;