From 6d8ec570b65c5cd63f5cc3b9b132ecb7f1ae1a41 Mon Sep 17 00:00:00 2001
From: Michael Davidsaver <mdavidsaver@gmail.com>
Date: Fri, 22 Oct 2021 10:51:20 -0500
Subject: [PATCH] Add pvxmshim

---
 tools/Makefile  |   3 +
 tools/mshim.cpp | 302 ++++++++++++++++++++++++++++++++++++++++++++++++
 2 files changed, 305 insertions(+)
 create mode 100644 tools/mshim.cpp

diff --git a/tools/Makefile b/tools/Makefile
index 2bb5a03..4e3eef4 100644
--- a/tools/Makefile
+++ b/tools/Makefile
@@ -35,6 +35,9 @@ pvxcall_SRCS += call.cpp
 PROD += pvxlist
 pvxlist_SRCS += list.cpp
 
+PROD += pvxmshim
+pvxmshim_SRCS += mshim.cpp
+
 #===========================
 
 include $(TOP)/configure/RULES
diff --git a/tools/mshim.cpp b/tools/mshim.cpp
new file mode 100644
index 0000000..7d7ab68
--- /dev/null
+++ b/tools/mshim.cpp
@@ -0,0 +1,302 @@
+/**
+ * Copyright - See the COPYRIGHT that is included with this distribution.
+ * pvxs is distributed subject to a Software License Agreement found
+ * in file LICENSE that is included with this distribution.
+ */
+
+#include <map>
+#include <vector>
+#include <iostream>
+#include <string>
+#include <exception>
+
+#include <epicsVersion.h>
+#include <epicsGetopt.h>
+
+#include <pvxs/log.h>
+#include <pvxs/server.h>
+#include "utilpvt.h"
+#include "evhelper.h"
+#include "udp_collector.h"
+
+using namespace pvxs;
+
+DEFINE_LOGGER(applog, "mshim");
+
+namespace {
+
+void usage(const char* argv0)
+{
+    std::cerr<<
+                "Usage: "<<argv0<<" [-L <ip>[@iface]]... [-F]\n"
+                "\n"
+                "  -L <ip>                 Interface address to listen on.\n"
+                "  -L <ip>[@iface]         Join multicast group, optionally via a certain interface\n"
+                "                          to override the default selected by the OS.\n"
+                "  -F <ip>                 Forward received packets to destination unicast/broadcast address.\n"
+                "  -F <ip>[,ttl#][@iface]  Forward received packets to destination multicast group.\n"
+                "                          Optionally override OS default TTL and outbound interface selected\n"
+                "                          by the OS.\n"
+                "  -p <port#>              Default port number.  (overrides $EPICS_PVA_BROADCAST_PORT)\n"
+                "  -h                      Show this message.\n"
+                "  -V                      Show versions.\n"
+                "\n"
+                "  Compatibility shim for IPv4 multicast by non-aware PVA clients/servers.\n"
+                "\n"
+                "  Examples:\n"
+                "\n"
+                "    1. Forwarding searches from local clients to a multicast group via the default interface.\n"
+                "    2. Forwarding beacons from multicast group via the default interface to local clients.\n"
+                "\n"
+                "    "<<argv0<<" -L 127.0.0.1:15076 -F 224.1.1.1,255 &  # 1\n"
+                "    "<<argv0<<" -L 224.1.1.1,255 -F 127.0.0.1:15076 &  # 2\n"
+    <<std::endl;
+}
+
+SockEndpoint parseEP(const char* optarg, const server::Config& conf)
+{
+    SockEndpoint ep;
+    try {
+        ep = SockEndpoint(optarg, conf.udp_port);
+
+    }catch(std::exception& e){
+        std::cerr<<"Error: Invalid group spec. '"<<escape(optarg)<<"' : "<<e.what()<<std::endl;
+        exit(1);
+    }
+    if(ep.addr.family()!=AF_INET) {
+        std::cerr<<"Only IPv4 addresses are supported"<<std::endl;
+        exit(1);
+
+    } else if(ep.addr.port()==0) {
+        std::cerr<<"Non-zero port number required"<<std::endl;
+        exit(1);
+    }
+    return ep;
+}
+
+struct App {
+    const SockAttach attach;
+    IfaceMap& ifmap;
+    const evsocket sockTx{AF_INET, SOCK_DGRAM, 0};
+    std::vector<SockEndpoint> destinations;
+
+    // effectively local to UDPManager worker
+    std::vector<uint8_t> scratch;
+
+    App()
+        :ifmap(IfaceMap::instance())
+        ,scratch(0x10000)
+    {
+        auto bind_addr(SockAddr::any(sockTx.af));
+        sockTx.bind(bind_addr);
+        sockTx.mcast_loop(true);
+    }
+
+    void onSearch(const UDPManager::Search& msg)
+    {
+        FixedBuf buf(true, scratch);
+        auto save_header = buf.save();
+        buf._skip(8);
+        to_wire(buf, msg.searchID);
+        auto save_flags = buf.save();
+        to_wire(buf, {
+                    uint8_t(msg.mustReply ? pva_search_flags::MustReply : 0u),
+                    0,0,0
+                });
+        assert(!msg.server.isAny() && msg.server.family()==AF_INET); // UDPManager has already handled this case
+        to_wire(buf, msg.server);
+        to_wire(buf, uint16_t(msg.server.port()));
+
+        size_t nproto = msg.otherproto.size();
+        if(msg.protoTCP)
+            nproto++;
+
+        to_wire(buf, Size{nproto});
+        if(msg.protoTCP)
+            to_wire(buf, "tcp");
+        for(auto& prot : msg.otherproto) {
+            to_wire(buf, prot);
+        }
+
+        to_wire(buf, uint16_t(msg.names.size()));
+
+        for(auto& name : msg.names) {
+            to_wire(buf, name.id);
+            to_wire(buf, name.name);
+        }
+
+        if(!buf.good()) {
+            log_warn_printf(applog, "Unable to construct CMD_SEARCH to forward. %s:%d\n",
+                            buf.file(), buf.line());
+            return;
+        }
+
+        auto bufsize = buf.save()-save_header;
+        {
+            FixedBuf buf(true, save_header, 8u);
+            to_wire(buf, Header{CMD_SEARCH, 0, uint32_t(bufsize-8u)});
+        }
+
+        for(auto& dest : destinations) {
+            if(dest.addr.isMCast() || ifmap.is_broadcast(dest.addr)) {
+                *save_flags &= ~pva_search_flags::Unicast;
+
+            } else {
+                *save_flags |= pva_search_flags::Unicast;
+            }
+
+            sockTx.mcast_prep_sendto(dest);
+
+            auto ret = sendto(sockTx.sock, (char*)scratch.data(), bufsize, 0,
+                              &dest.addr->sa, dest.addr.size());
+
+            if(ret < 0) {
+                int err = evutil_socket_geterror(sockTx.sock);
+                if(err==SOCK_EWOULDBLOCK || err==EAGAIN || err==SOCK_EINTR) {
+                    break; // too bad, better luck next time
+
+                } else {
+                    log_warn_printf(applog, "Unable to send search to %s, skip.  (%d) %s\n",
+                                    std::string(SB()<<dest).c_str(), err, evutil_socket_error_to_string(err));
+                }
+
+            } else if(ret!=bufsize) {
+                log_warn_printf(applog, "Sent truncated search to %s?\n",
+                                std::string(SB()<<dest).c_str());
+
+            } else {
+                log_debug_printf(applog, "Forwarded search to %s -> %s -> %s?\n",
+                                 msg.src.tostring().c_str(),
+                                 msg.server.tostring().c_str(),
+                                 std::string(SB()<<dest).c_str());
+            }
+        }
+    }
+
+    void onBeacon(const UDPManager::Beacon& msg)
+    {
+        FixedBuf buf(true, scratch);
+        auto save_header = buf.save();
+        buf._skip(8);
+        _to_wire<12>(buf, msg.guid.data(), false, __FILE__, __LINE__);
+        to_wire(buf, uint32_t(0u)); // skip flags, seq, and change count.  unused
+        assert(!msg.server.isAny() && msg.server.family()==AF_INET); // UDPManager has already handled this case
+        to_wire(buf, msg.server);
+        to_wire(buf, uint16_t(msg.server.port()));
+        to_wire(buf, msg.proto);
+        // "NULL" serverStatus
+        to_wire(buf, uint8_t(0xff));
+
+        if(!buf.good()) {
+            log_warn_printf(applog, "Unable to construct CMD_SEARCH to forward. %s:%d\n",
+                            buf.file(), buf.line());
+            return;
+        }
+
+        auto bufsize = buf.save()-save_header;
+        {
+            FixedBuf buf(true, save_header, 8u);
+            to_wire(buf, Header{CMD_SEARCH, 0, uint32_t(bufsize-8u)});
+        }
+
+        for(auto& dest : destinations) {
+            sockTx.mcast_prep_sendto(dest);
+
+            auto ret = sendto(sockTx.sock, (char*)scratch.data(), bufsize, 0,
+                              &dest.addr->sa, dest.addr.size());
+
+            if(ret < 0) {
+                int err = evutil_socket_geterror(sockTx.sock);
+                if(err==SOCK_EWOULDBLOCK || err==EAGAIN || err==SOCK_EINTR) {
+                    break; // too bad, better luck next time
+
+                } else {
+                    log_warn_printf(applog, "Unable to send beacon to %s, skip.  (%d) %s\n",
+                                    std::string(SB()<<dest).c_str(), err, evutil_socket_error_to_string(err));
+                }
+
+            } else if(ret!=bufsize) {
+                log_warn_printf(applog, "Sent truncated beacon to %s?\n",
+                                std::string(SB()<<dest).c_str());
+
+            } else {
+                log_debug_printf(applog, "Forwarded beacon to %s -> %s -> %s?\n",
+                                 msg.src.tostring().c_str(),
+                                 msg.server.tostring().c_str(),
+                                 std::string(SB()<<dest).c_str());
+            }
+        }
+    }
+};
+
+} // namespace
+
+int main(int argc, char *argv[])
+{
+    try {
+        SockAttach attach;
+        logger_config_env();
+        App app;
+
+        auto conf(server::Config::fromEnv());
+        auto manager(UDPManager::instance());
+        std::vector<std::unique_ptr<UDPListener>> listeners;
+
+        auto onS = [&app](const UDPManager::Search& msg) {app.onSearch(msg);};
+        auto onB = [&app](const UDPManager::Beacon& msg) {app.onBeacon(msg);};
+
+        {
+            int opt;
+            while ((opt = getopt(argc, argv, "L:F:phV")) != -1) {
+                switch(opt) {
+                case 'L':
+                {
+                    SockEndpoint ep(parseEP(optarg, conf));
+                    listeners.push_back(manager.onSearch(ep, onS));
+                    listeners.push_back(manager.onBeacon(ep, onB));
+                    break;
+                }
+                case 'F':
+                    app.destinations.push_back(parseEP(optarg, conf));
+                    break;
+                case 'p':
+                    conf.udp_port = parseTo<uint64_t>(optarg);
+                    break;
+                case 'h':
+                    usage(argv[0]);
+                    return 0;
+                case 'V':
+                    std::cout<<version_str()<<"\n";
+                    std::cout<<EPICS_VERSION_STRING<<"\n";
+                    std::cout<<"libevent "<<event_get_version()<<std::endl;
+                    return 0;
+                default:
+                    usage(argv[0]);
+                    std::cerr<<"\nError: Unknown argument: "<<char(opt)<<std::endl;
+                    return 1;
+                }
+            }
+        }
+
+        if(argc!=optind) {
+            usage(argv[0]);
+            std::cerr<<"Error: Unexpected arguments."<<std::endl;
+            return 1;
+        }
+
+        for(auto& listener : listeners) {
+            listener->start();
+        }
+
+        epicsEvent done;
+        SigInt H([&done]() {
+            done.signal();
+        });
+        done.wait();
+
+        return 0;
+    }catch(std::exception& e){
+        std::cerr<<"Error: "<<e.what()<<"\n";
+        return 1;
+    }
+}