From 47ebfd537197d2e638683574c8fb7b66b913cdd4 Mon Sep 17 00:00:00 2001 From: Michael Davidsaver Date: Sun, 16 May 2021 08:11:54 -0700 Subject: [PATCH] Accept CMD_ORIGIN_TAG Implement local mcast hack --- src/client.cpp | 40 +++---- src/pvaproto.h | 18 +++- src/udp_collector.cpp | 243 +++++++++++++++++++++++++++++++----------- 3 files changed, 213 insertions(+), 88 deletions(-) diff --git a/src/client.cpp b/src/client.cpp index 0486067..117c8a0 100644 --- a/src/client.cpp +++ b/src/client.cpp @@ -769,43 +769,27 @@ bool ContextImpl::onSearch() } return false; // wait for more I/O - } else if(nrx<8) { - // maybe a zero (body) length packet? - // maybe an OS error? + } - log_info_printf(io, "UDP ignore runt%s\n", ""); - return true; + FixedBuf M(true, searchMsg.data(), nrx); + Header head{}; + from_wire(M, head); // overwrites M.be - } else if(searchMsg[0]!=0xca || searchMsg[1]==0 || (searchMsg[2]&(pva_flags::Control|pva_flags::SegMask))) { - // minimum header size is 8 bytes - // ID byte must by 0xCA (because PVA has some paternal envy) - // ignore incompatible version 0 + if(!M.good() || (head.flags&(pva_flags::Control|pva_flags::SegMask))) { // UDP packets can't contain control messages, or use segmentation - log_info_printf(io, "UDP ignore header%u %02x%02x%02x%02x\n", - unsigned(nrx), searchMsg[0], searchMsg[1], searchMsg[2], searchMsg[3]); + log_hex_printf(io, Level::Debug, &searchMsg[0], nrx, "Ignore UDP message from %s\n", src.tostring().c_str()); return true; } log_hex_printf(io, Level::Debug, &searchMsg[0], nrx, "UDP search Rx %d from %s\n", nrx, src.tostring().c_str()); - bool be = searchMsg[2]&pva_flags::MSB; - - FixedBuf M(be, searchMsg.data(), nrx); - - const uint8_t cmd = M[3]; - M.skip(4, __FILE__, __LINE__); - - uint32_t len=0; - from_wire(M, len); - - if(len > M.size() && M.good()) { - log_info_printf(io, "UDP ignore header%u %02x%02x%02x%02x\n", - unsigned(M.size()), M[0], M[1], M[2], M[3]); + if(head.len > M.size() && M.good()) { + log_info_printf(io, "UDP ignore header truncated%s", "\n"); return true; } - if(cmd==CMD_SEARCH_RESPONSE) { + if(head.cmd==CMD_SEARCH_RESPONSE) { procSearchReply(*this, src, M, false); } else { @@ -978,8 +962,10 @@ void ContextImpl::tickSearch() unsigned(ntx), unsigned(consumed)); } else { - log_debug_printf(io, "Search to %s %s\n", pair.first.tostring().c_str(), - pair.second ? "ucast" : "bcast"); + log_hex_printf(io, Level::Debug, (char*)searchMsg.data(), consumed, + "Search to %s %s\n", + pair.first.tostring().c_str(), + pair.second ? "ucast" : "bcast"); } } *pflags |= 0x80; // TCP search is always "unicast" diff --git a/src/pvaproto.h b/src/pvaproto.h index 6768555..db0afa7 100644 --- a/src/pvaproto.h +++ b/src/pvaproto.h @@ -47,7 +47,7 @@ protected: constexpr Buffer(bool be, uint8_t* buf, size_t n) :pos(buf), limit(buf+n), be(be) {} virtual ~Buffer() {} public: - const bool be; + bool be; // all sub-classes define // bool refill(size_t more) @@ -613,6 +613,22 @@ void to_wire(Buf& buf, const Header& H) void to_evbuf(evbuffer *buf, const Header& H, bool be); +template +void from_wire(Buf& buf, Header& H) +{ + if(!buf.ensure(8) || buf[0]!=0xca || buf[1]==0) { + buf.fault(__FILE__, __LINE__); + + } else { + H.cmd = buf[3]; + H.flags = buf[2]; + // Set/change buffer endianness + buf.be = H.flags&pva_flags::MSB; + buf.skip(4u, __FILE__, __LINE__); + from_wire(buf, H.len); + } +} + }} // namespace pvxs::impl #endif // PVAPROTO_H diff --git a/src/udp_collector.cpp b/src/udp_collector.cpp index 9551672..c0e65c6 100644 --- a/src/udp_collector.cpp +++ b/src/udp_collector.cpp @@ -36,6 +36,7 @@ struct UDPCollector : public UDPManager::Search, { UDPManager::Pvt* const manager; SockAddr bind_addr; + SockAddr mcast_addr; std::string name; evsocket sock; evevent rx; @@ -47,6 +48,11 @@ struct UDPCollector : public UDPManager::Search, std::set listeners; + // loopback collector for our port. + std::shared_ptr loSibling; + // for the loSibling, list of non-lo siblings for forwards + std::set otherSiblings; + UDPCollector(UDPManager::Pvt* manager, const SockAddr& bind_addr); ~UDPCollector(); @@ -54,6 +60,8 @@ struct UDPCollector : public UDPManager::Search, { SockAddr dest; + buf.resize(0x10001); + // For Search messages, we use PV name strings in-place by adding nils. // Ensure one extra byte at the end of the buffer for a nil after the last PV name recvfromx rx{sock.sock, (char*)&buf[0], buf.size()-1, &src, &dest}; @@ -74,48 +82,38 @@ struct UDPCollector : public UDPManager::Search, } return false; // wait for more I/O - } else if(nrx<8) { - // maybe a zero (body) length packet? - // maybe an OS error? - - log_info_printf(logio, "UDP ignore runt on %s\n", name.c_str()); - return true; - - } else if(buf[0]!=0xca || buf[1]==0 || (buf[2]&(pva_flags::Control|pva_flags::SegMask))) { - // minimum header size is 8 bytes - // ID byte must by 0xCA (because PVA has some paternal envy) - // ignore incompatible version 0 - // UDP packets can't contain control messages, or use segmentation - - log_info_printf(logio, "UDP ignore header%u %02x%02x%02x%02x on %s\n", - unsigned(nrx), buf[0], buf[1], buf[2], buf[3], - name.c_str()); - return true; } - log_hex_printf(logio, Level::Debug, &buf[0], nrx, "UDP Rx %d, %s -> %s\n", - nrx, src.tostring().c_str(), dest.tostring().c_str()); + log_hex_printf(logio, Level::Debug, &buf[0], nrx, "UDP %d Rx %d, %s -> %s (%s)\n", + sock.sock, nrx, src.tostring().c_str(), dest.tostring().c_str(), bind_addr.tostring().c_str()); + + process_one(dest, &buf[0], nrx, false); + return true; + } + + void process_one(const SockAddr& dest, const uint8_t* buf, int nrx, bool originated) + { + FixedBuf M(true, const_cast(buf), nrx); + Header head{}; + from_wire(M, head); // overwrites M.be + + if(!M.good() || (head.flags&(pva_flags::Control|pva_flags::SegMask))) { + // UDP packets can't contain control messages, or use segmentation + + log_hex_printf(logio, Level::Debug, &buf[0], nrx, "Ignore UDP message from %s\n", src.tostring().c_str()); + return; + } names.clear(); - bool be = buf[2]&pva_flags::MSB; - - FixedBuf M(be, buf.data(), nrx); - - uint8_t cmd = M[3]; - - M.skip(4, __FILE__, __LINE__); - uint32_t len=0; - from_wire(M, len); - - if(len > M.size() && M.good()) { + if(head.len > M.size() && M.good()) { log_info_printf(logio, "UDP ignore header%u %02x%02x%02x%02x on %s\n", unsigned(M.size()), M[0], M[1], M[2], M[3], name.c_str()); - return true; + return; } - switch(cmd) { + switch(head.cmd) { case CMD_SEARCH: { uint8_t flags = 0; @@ -123,17 +121,37 @@ struct UDPCollector : public UDPManager::Search, uint16_t port = 0; from_wire(M, searchID); + auto save_flags = M.save(); from_wire(M, flags); mustReply = flags&pva_search_flags::MustReply; + M.skip(3, __FILE__, __LINE__); // unused/reserved + auto save_replyAddr = M.save(); from_wire(M, replyAddr); from_wire(M, port); if(replyAddr.isAny()) { replyAddr = src; + if(originated) { + log_err_printf(logio, "CMD_ORIGIN_TAG search with reply to sender never works%s", "\n"); + return; + } } replyAddr.setPort(port); + if(M.good() && !originated && (flags&pva_search_flags::Unicast) && loSibling && dest.family()!=AF_UNSPEC) { + // clear unicast flag in forwarded message + *save_flags &= ~pva_search_flags::Unicast; + // recipient of forwarded message must use, and trust, replyAddr in body :( + { + FixedBuf R(M.be, save_replyAddr, 16u); + to_wire(R, replyAddr); + assert(R.good()); + } + loSibling->forwardM(dest, &buf[0], nrx); + return; + } + // so far, only "tcp" transport has ever been seen. // however, we will consider and ignore any others which might appear bool foundtcp = false; @@ -171,6 +189,9 @@ struct UDPCollector : public UDPManager::Search, M.skip(chlen.size, __FILE__, __LINE__); } + // used by our reply() + src = replyAddr; + if(M.good()) { // ensure nil for final PV name *M.save() = '\0'; @@ -209,11 +230,41 @@ struct UDPCollector : public UDPManager::Search, } } } - } break; } - return true; + case CMD_ORIGIN_TAG: { + SockAddr origin; // aka. original destination + from_wire(M, origin); + M.skip(head.len-16u, __FILE__, __LINE__); + + // only allow one CMD_ORIGIN_TAG message per packet + // only accept when sent to the mcast address from the loopback address + // since we only join the mcast group on loopback this will hopefully + // frustrate attempts to inject CMD_ORIGIN_TAG externally. + if(M.good() && !originated && dest.compare(mcast_addr,false)==0 && src.isLO()) { + origin.setPort(bind_addr.port()); + + originated = true; + log_debug_printf(logio, "Accept as originated from %s\n", origin.tostring().c_str()); + for(auto sib : otherSiblings) { + sib->process_one(origin, M.save(), M.size(), true); + } + return; + } + log_debug_printf(logio, "Ignore originated from %s %c%c%c%c\n", + origin.tostring().c_str(), + M.good() ? 'T' : 'F', + !originated ? 'T' : 'F', + dest.compare(mcast_addr,false)==0 ? 'T' : 'F', + src.isLO() ? 'T' : 'F'); + + break; + } + + default: + break; // ignore unknown + } } void handle(short ev) { @@ -234,6 +285,8 @@ struct UDPCollector : public UDPManager::Search, } } + void forwardM(const SockAddr& origin, const uint8_t* buf, size_t len); + // Search interface public: virtual bool reply(const void *msg, size_t msglen) const override; @@ -255,14 +308,35 @@ struct UDPManager::Pvt { // we should only be destroyed after that last collector has removed itself assert(collectors.empty()); } + + std::shared_ptr collect(const SockAddr& dest) + { + std::shared_ptr collector; + + if(dest.port()!=0) { + auto it = collectors.find(dest); + if(it!=collectors.end()) { + try { + collector = it->second->shared_from_this(); + }catch(std::bad_weak_ptr&){ + // nothing to do + } + } + } + + if(!collector) { + collector.reset(new UDPCollector(this, dest)); + } + return collector; + } }; -UDPCollector::UDPCollector(UDPManager::Pvt *manager, const SockAddr& bind_addr) +UDPCollector::UDPCollector(UDPManager::Pvt *manager, const SockAddr& requested_bind_addr) :manager(manager) - ,bind_addr(bind_addr) - ,sock(bind_addr.family(), SOCK_DGRAM, 0) + ,bind_addr(requested_bind_addr) + ,mcast_addr(AF_INET, "224.0.0.128") + ,sock(requested_bind_addr.family(), SOCK_DGRAM, 0) ,rx(event_new(manager->loop.base, sock.sock, EV_READ|EV_PERSIST, &handle_static, this)) - ,buf(0x10001) ,beaconMsg(src) { manager->loop.assertInLoop(); @@ -270,22 +344,61 @@ UDPCollector::UDPCollector(UDPManager::Pvt *manager, const SockAddr& bind_addr) epicsSocketEnableAddressUseForDatagramFanout(sock.sock); enable_SO_RXQ_OVFL(sock.sock); enable_IP_PKTINFO(sock.sock); - sock.bind(this->bind_addr); - name = "UDP "+this->bind_addr.tostring(); + sock.bind(bind_addr); + name = "UDP "+bind_addr.tostring(); - log_info_printf(logsetup, "Bound to %s\n", name.c_str()); + auto lo_addr(SockAddr::loopback(bind_addr.family(), bind_addr.port())); + assert(lo_addr.port()!=0u); + mcast_addr.setPort(bind_addr.port()); + + /* Bind this address to receive multicast over loopback, and also to send them. + * Notes: + * - Linux, it would be possible to bind to the mcast address in order to receive only + * packets so destined. It would also be possible to send mcasts from a socket + * so bound. + * - OSX, it would be possible to bind to the mcast address, but not to send mcasts. + * - Winsock, it is not possible to bind to the mcast address. + * + * So we take the least common denominator across all platforms, which is to bind to the wildcard. + * This socket may then receive unicasts which need to be forwarded. + * Also broadcasts, which will be ignored (if no listeners). + */ + auto mcast_bind_addr(SockAddr::any(AF_INET, bind_addr.port())); + + if(bind_addr!=mcast_bind_addr) { + loSibling = manager->collect(mcast_bind_addr); + assert(this!=loSibling.get()); + + log_info_printf(logsetup, "Bound %d to %s with sibling %s\n", sock.sock, name.c_str(), loSibling->name.c_str()); + + } else { + // join group to receive + sock.mcast_join(mcast_addr, lo_addr); + // setup for re-transmit + sock.mcast_ttl(1); // make default explicit + sock.mcast_loop(true); + sock.mcast_iface(lo_addr); + + log_info_printf(logsetup, "Bound %d to %s as lo\n", sock.sock, name.c_str()); + } if(event_add(rx.get(), nullptr)) throw std::runtime_error("Unable to create collector Rx event"); - manager->collectors[this->bind_addr] = this; + if(loSibling) + loSibling->otherSiblings.insert(this); + + manager->collectors[bind_addr] = this; } UDPCollector::~UDPCollector() { manager->loop.assertInLoop(); - manager->collectors.erase(this->bind_addr); + if(loSibling) + loSibling->otherSiblings.erase(this); + + manager->collectors.erase(bind_addr); // we should only be destroyed after that last listener has removed itself assert(listeners.empty()); @@ -390,21 +503,8 @@ UDPListener::UDPListener(const std::shared_ptr &manager, SockAd { manager->loop.assertInLoop(); - if(dest.port()!=0) { - auto it = manager->collectors.find(dest); - if(it!=manager->collectors.end()) { - try { - collector = it->second->shared_from_this(); - }catch(std::bad_weak_ptr&){ - // nothing to do - } - } - } - - if(!collector) { - collector.reset(new UDPCollector(manager.get(), dest)); - dest = collector->bind_addr; - } + collector = manager->collect(dest); + dest = collector->bind_addr; } UDPListener::~UDPListener() @@ -433,18 +533,41 @@ void UDPListener::start(bool s) }); } +void UDPCollector::forwardM(const SockAddr& origin, const uint8_t* obuf, size_t olen) +{ + log_debug_printf(logio, "Forward as originated for %s\n", + origin.tostring().c_str()); + + buf.resize(8+16+olen); + { + FixedBuf M(true, &buf[0], buf.size()); + + to_wire(M, Header{CMD_ORIGIN_TAG, 0, 16u}); + to_wire(M, origin); + assert(M.good()); + memcpy(M.save(), obuf, olen); + } + + src = mcast_addr; + reply(&buf[0], buf.size()); +} + bool UDPCollector::reply(const void *msg, size_t msglen) const { manager->loop.assertInLoop(); + log_hex_printf(logio, Level::Debug, msg, msglen, "Send %s -> %s\n", + bind_addr.tostring().c_str(), src.tostring().c_str()); + int ntx = sendto(sock.sock, (char*)msg, msglen, 0, &src->sa, src.size()); if(ntx<0) { int err = evutil_socket_geterror(sock.sock); if(err==SOCK_EWOULDBLOCK || err==EAGAIN || err==SOCK_EINTR) { // nothing to do here } else { - log_warn_printf(logio, "UDP TX Error on %s : %s\n", name.c_str(), - evutil_socket_error_to_string(err)); + log_warn_printf(logio, "UDP TX Error on %s -> %s : (%d) %s\n", + name.c_str(), src.tostring().c_str(), + err, evutil_socket_error_to_string(err)); } return false; // wait for more I/O }