nameserver.hpp

Go to the documentation of this file.

 
#ifndef JOIN_FABRIC_NAMESERVER_HPP
#define JOIN_FABRIC_NAMESERVER_HPP
 
// libjoin.
#include <join/datagram_socket.hpp>
#include <join/mdns_protocol.hpp>
#include <join/dns_protocol.hpp>
#include <join/dns_message.hpp>
#include <join/condition.hpp>
#include <join/reactor.hpp>
 
namespace join
{
    template <typename Protocol>
    class BasicDatagramNameServer : public EventHandler
    {
    public:
        using Socket = typename Protocol::Socket;
        using Endpoint = typename Protocol::Endpoint;
 
        explicit BasicDatagramNameServer (Reactor& reactor = ReactorThread::reactor ())
        : _reactor (reactor)
        , _buffer (std::make_unique<char[]> (Protocol::maxMsgSize))
        {
        }
 
        BasicDatagramNameServer (const BasicDatagramNameServer& other) = delete;
 
        BasicDatagramNameServer& operator= (const BasicDatagramNameServer& other) = delete;
 
        BasicDatagramNameServer (BasicDatagramNameServer&& other) = delete;
 
        BasicDatagramNameServer& operator= (BasicDatagramNameServer&& other) = delete;
 
        virtual ~BasicDatagramNameServer () noexcept = default;
 
        virtual int bind (const Endpoint& endpoint) noexcept
        {
            if (_socket.bind (endpoint) == -1)
            {
                return -1;  // LCOV_EXCL_LINE
            }
 
            _reactor.addHandler (_socket.handle (), this);
 
            return 0;
        }
 
        virtual void close () noexcept
        {
            _reactor.delHandler (_socket.handle ());
            _socket.close ();
        }
 
        int reply (const DnsPacket& query, const std::vector<ResourceRecord>& answers = {},
                   const std::vector<ResourceRecord>& authorities = {},
                   const std::vector<ResourceRecord>& additionals = {}, uint16_t rcode = 0)
        {
            DnsPacket packet{};
            packet.id = query.id;
            packet.flags = (uint16_t (1) << 15) | (query.flags & 0x7800) | (uint16_t (1) << 10) | (rcode & 0x000F);
            packet.dest = query.src;
            packet.port = query.port;
 
            packet.questions = query.questions;
            packet.answers = answers;
            packet.authorities = authorities;
            packet.additionals = additionals;
 
            return send (packet);
        }
 
        virtual void onQuery (const DnsPacket& packet) = 0;
 
    protected:
        virtual void onReadable ([[maybe_unused]] int fd) override
        {
            Endpoint from;
            ssize_t size = _socket.readFrom (_buffer.get (), Protocol::maxMsgSize, &from);
            if (size >= int (_headerSize))
            {
                std::stringstream data;
                data.rdbuf ()->pubsetbuf (_buffer.get (), size);
 
                DnsPacket packet;
                _message.deserialize (packet, data);
                packet.src = from.ip ();
                packet.dest = _socket.localEndpoint ().ip ();
                packet.port = from.port ();
 
                if ((packet.flags & 0x8000) == 0)
                {
                    this->onQuery (packet);
                }
            }
        }
 
        int send (DnsPacket& packet)
        {
            std::stringstream data;
            if (_message.serialize (packet, data) == -1)
            {
                // LCOV_EXCL_START
                lastError = make_error_code (Errc::InvalidParam);
                return -1;
                // LCOV_EXCL_STOP
            }
 
            std::string buffer = data.str ();
            if (buffer.size () > Protocol::maxMsgSize)
            {
                // LCOV_EXCL_START
                lastError = make_error_code (Errc::MessageTooLong);
                return -1;
                // LCOV_EXCL_STOP
            }
 
            if (_socket.writeTo (buffer.data (), buffer.size (), {packet.dest, packet.port}) == -1)
            {
                return -1;  // LCOV_EXCL_LINE
            }
 
            return 0;
        };
 
        static constexpr size_t _headerSize = 12;
 
        DnsMessage _message;
 
        Socket _socket;
 
        Reactor& _reactor;
 
        std::unique_ptr<char[]> _buffer;
    };
 
    template <typename Protocol>
    class BasicDatagramPeer : public BasicDatagramNameServer<Protocol>
    {
    public:
        using Socket = typename BasicDatagramNameServer<Protocol>::Socket;
        using Endpoint = typename BasicDatagramNameServer<Protocol>::Endpoint;
 
        using DnsNotify = std::function<void (const DnsPacket&)>;
 
        DnsNotify onSuccess;
 
        DnsNotify onFailure;
 
        explicit BasicDatagramPeer (unsigned int ifindex, Reactor& reactor = ReactorThread::reactor ())
        : BasicDatagramNameServer<Protocol> (reactor)
#ifdef DEBUG
        , onSuccess (defaultOnSuccess)
        , onFailure (defaultOnFailure)
#else
        , onSuccess (nullptr)
        , onFailure (nullptr)
#endif
        , _ifindex (ifindex)
        {
        }
 
        explicit BasicDatagramPeer (const std::string& interface, Reactor& reactor = ReactorThread::reactor ())
        : BasicDatagramPeer<Protocol> (if_nametoindex (interface.c_str ()), reactor)
        {
        }
 
        BasicDatagramPeer (const BasicDatagramPeer& other) = delete;
 
        BasicDatagramPeer& operator= (const BasicDatagramPeer& other) = delete;
 
        BasicDatagramPeer (BasicDatagramPeer&& other) = delete;
 
        BasicDatagramPeer& operator= (BasicDatagramPeer&& other) = delete;
 
        virtual ~BasicDatagramPeer () noexcept = default;
 
        using BasicDatagramNameServer<Protocol>::bind;
 
        int bind (int family) noexcept
        {
            IpAddress maddress = Protocol::multicastAddress (family);
            Endpoint endpoint{IpAddress (family), Protocol::defaultPort};
 
            if ((this->_socket.opened () == false) && (this->_socket.open (endpoint.protocol ()) == -1))
            {
                return -1;  // LCOV_EXCL_LINE
            }
 
            if (this->_socket.setOption (Socket::ReusePort, 1) == -1)
            {
                // LCOV_EXCL_START
                this->close ();
                return -1;
                // LCOV_EXCL_STOP
            }
 
            if (this->_socket.bind (endpoint) == -1)
            {
                // LCOV_EXCL_START
                this->close ();
                return -1;
                // LCOV_EXCL_STOP
            }
 
            if (endpoint.protocol ().family () == AF_INET6)
            {
                ipv6_mreq mreq{};
                ::memcpy (&mreq.ipv6mr_multiaddr, maddress.addr (), maddress.length ());
                mreq.ipv6mr_interface = _ifindex;
                if (::setsockopt (this->_socket.handle (), IPPROTO_IPV6, IPV6_ADD_MEMBERSHIP, &mreq, sizeof (mreq)) ==
                    -1)
                {
                    // LCOV_EXCL_START
                    lastError = std::error_code (errno, std::generic_category ());
                    this->close ();
                    return -1;
                    // LCOV_EXCL_STOP
                }
                if (::setsockopt (this->_socket.handle (), IPPROTO_IPV6, IPV6_MULTICAST_IF, &_ifindex,
                                  sizeof (_ifindex)) == -1)
                {
                    // LCOV_EXCL_START
                    lastError = std::error_code (errno, std::generic_category ());
                    this->close ();
                    return -1;
                    // LCOV_EXCL_STOP
                }
            }
            else
            {
                // LCOV_EXCL_START: IPv4 multicast not supported by github action containers.
                ip_mreqn mreq{};
                ::memcpy (&mreq.imr_multiaddr, maddress.addr (), maddress.length ());
                mreq.imr_ifindex = static_cast<int> (_ifindex);
                if (::setsockopt (this->_socket.handle (), IPPROTO_IP, IP_ADD_MEMBERSHIP, &mreq, sizeof (mreq)) == -1)
                {
                    lastError = std::error_code (errno, std::generic_category ());
                    this->close ();
                    return -1;
                }
                if (::setsockopt (this->_socket.handle (), IPPROTO_IP, IP_MULTICAST_IF, &mreq, sizeof (mreq)) == -1)
                {
                    lastError = std::error_code (errno, std::generic_category ());
                    this->close ();
                    return -1;
                }
                // LCOV_EXCL_STOP
            }
 
#ifndef DEBUG
            if (this->_socket.setOption (Socket::MulticastLoop, 0) == -1)
            {
                // LCOV_EXCL_START
                this->close ();
                return -1;
                // LCOV_EXCL_STOP
            }
#endif
 
            this->_reactor.addHandler (this->_socket.handle (), this);
 
            return 0;
        }
 
        int probe (const std::vector<ResourceRecord>& records)
        {
            if (records.empty ())
            {
                lastError = make_error_code (Errc::InvalidParam);
                return -1;
            }
 
            DnsPacket packet{};
            packet.id = 0;
            packet.flags = 0;
            IpAddress mcast = Protocol::multicastAddress (this->_socket.family ());
            packet.dest = IpAddress (mcast.addr (), mcast.length (), _ifindex);
            packet.port = Protocol::defaultPort;
 
            for (auto const& record : records)
            {
                QuestionRecord question;
                question.host = record.host;
                question.type = DnsMessage::RecordType::ANY;
                question.dnsclass = DnsMessage::RecordClass::IN | 0x8000;
                packet.questions.push_back (question);
 
                packet.authorities.push_back (record);
            }
 
            return this->send (packet);
        }
 
        int announce (const std::vector<ResourceRecord>& records)
        {
            if (records.empty ())
            {
                lastError = make_error_code (Errc::InvalidParam);
                return -1;
            }
 
            DnsPacket packet{};
            packet.id = 0;
            packet.flags = (uint16_t (1) << 15) | (uint16_t (1) << 10);
            IpAddress mcast = Protocol::multicastAddress (this->_socket.family ());
            packet.dest = IpAddress (mcast.addr (), mcast.length (), _ifindex);
            packet.port = Protocol::defaultPort;
 
            for (auto const& record : records)
            {
                packet.answers.push_back (record);
            }
 
            return this->send (packet);
        }
 
        int goodbye (const std::vector<ResourceRecord>& records)
        {
            if (records.empty ())
            {
                lastError = make_error_code (Errc::InvalidParam);
                return -1;
            }
 
            DnsPacket packet{};
            packet.id = 0;
            packet.flags = (uint16_t (1) << 15) | (uint16_t (1) << 10);
            IpAddress mcast = Protocol::multicastAddress (this->_socket.family ());
            packet.dest = IpAddress (mcast.addr (), mcast.length (), _ifindex);
            packet.port = Protocol::defaultPort;
 
            for (auto const& record : records)
            {
                ResourceRecord goodbye = record;
                goodbye.ttl = 0;
                packet.answers.push_back (goodbye);
            }
 
            return this->send (packet);
        }
 
        int browse (const std::string& serviceType)
        {
            if (serviceType.empty ())
            {
                lastError = make_error_code (Errc::InvalidParam);
                return -1;
            }
 
            DnsPacket packet{};
            packet.id = 0;
            packet.flags = 0;
            IpAddress mcast = Protocol::multicastAddress (this->_socket.family ());
            packet.dest = IpAddress (mcast.addr (), mcast.length (), _ifindex);
            packet.port = Protocol::defaultPort;
 
            QuestionRecord question;
            question.host = serviceType;
            question.type = DnsMessage::RecordType::PTR;
            question.dnsclass = DnsMessage::RecordClass::IN;
            packet.questions.push_back (question);
 
            return this->send (packet);
        }
 
        IpAddressList resolveAllAddress (const std::string& host, int family,
                                         std::chrono::milliseconds timeout = std::chrono::seconds (5))
        {
            if (host.empty ())
            {
                return {};
            }
 
            DnsPacket packet{};
            packet.id = join::randomize<uint16_t> ();
            packet.flags = 1 << 8;
 
            QuestionRecord question;
            question.host = host;
            question.type = (family == AF_INET6) ? DnsMessage::RecordType::AAAA : DnsMessage::RecordType::A;
            question.dnsclass = DnsMessage::RecordClass::IN;
            packet.questions.push_back (question);
 
            if (query (packet, timeout) == -1)
            {
                return {};
            }
 
            IpAddressList addresses;
 
            for (auto const& answer : packet.answers)
            {
                if (!answer.addr.isWildcard () && (answer.type == question.type))
                {
                    addresses.push_back (answer.addr);
                }
            }
 
            return addresses;
        }
 
        IpAddressList resolveAllAddress (const std::string& host,
                                         std::chrono::milliseconds timeout = std::chrono::seconds (5))
        {
            IpAddressList addresses;
 
            for (auto const& family : {AF_INET, AF_INET6})
            {
                IpAddressList tmp = resolveAllAddress (host, family, timeout);
                addresses.insert (addresses.end (), tmp.begin (), tmp.end ());
            }
 
            return addresses;
        }
 
        IpAddress resolveAddress (const std::string& host, int family,
                                  std::chrono::milliseconds timeout = std::chrono::seconds (5))
        {
            for (auto const& address : resolveAllAddress (host, family, timeout))
            {
                return address;
            }
 
            return IpAddress (family);
        }
 
        IpAddress resolveAddress (const std::string& host, std::chrono::milliseconds timeout = std::chrono::seconds (5))
        {
            for (auto const& address : resolveAllAddress (host, timeout))
            {
                return address;
            }
 
            return {};
        }
 
        AliasList resolveAllName (const IpAddress& address,
                                  std::chrono::milliseconds timeout = std::chrono::seconds (5))
        {
            if (address.isWildcard ())
            {
                return {};
            }
 
            DnsPacket packet{};
            packet.id = join::randomize<uint16_t> ();
            packet.flags = 1 << 8;
 
            QuestionRecord question;
            question.host = address.toArpa ();
            question.type = DnsMessage::RecordType::PTR;
            question.dnsclass = DnsMessage::RecordClass::IN;
            packet.questions.push_back (question);
 
            if (query (packet, timeout) == -1)
            {
                return {};
            }
 
            AliasList aliases;
 
            for (auto const& answer : packet.answers)
            {
                if (!answer.name.empty () && (answer.type == DnsMessage::RecordType::PTR))
                {
                    aliases.insert (answer.name);
                }
            }
 
            return aliases;
        }
 
        std::string resolveName (const IpAddress& address, std::chrono::milliseconds timeout = std::chrono::seconds (5))
        {
            for (auto const& alias : resolveAllName (address, timeout))
            {
                return alias;
            }
 
            return {};
        }
 
        virtual void onAnnouncement (const DnsPacket& packet) = 0;
 
    protected:
        void onReadable ([[maybe_unused]] int fd) override final
        {
            Endpoint from;
            ssize_t size = this->_socket.readFrom (this->_buffer.get (), Protocol::maxMsgSize, &from);
            if (size >= int (this->_headerSize))
            {
                std::stringstream data;
                data.rdbuf ()->pubsetbuf (this->_buffer.get (), size);
 
                DnsPacket packet;
                this->_message.deserialize (packet, data);
                IpAddress mcast = Protocol::multicastAddress (this->_socket.family ());
                packet.src = from.ip ();
                packet.dest = IpAddress (mcast.addr (), mcast.length (), _ifindex);
                packet.port = from.port ();
 
                if ((packet.flags & 0x8000) == 0)
                {
                    bool unicast = false;
                    for (auto const& q : packet.questions)
                    {
                        if (q.dnsclass & 0x8000)
                        {
                            unicast = true;
                            break;
                        }
                    }
 
                    if (!unicast)
                    {
                        packet.src = IpAddress (mcast.addr (), mcast.length (), _ifindex);
                    }
 
                    this->onQuery (packet);
                    return;
                }
 
                {
                    ScopedLock<Mutex> lock (_syncMutex);
 
                    auto it = _pending.find (packet.id);
                    if (it != _pending.end ())
                    {
                        it->second->packet = packet;
                        it->second->ec = DnsMessage::decodeError (packet.flags & 0x000F);
                        it->second->cond.signal ();
                        return;
                    }
                }
 
                onAnnouncement (packet);
            }
        }
 
#ifdef DEBUG
        /*
         * @brief default callback called when a lookup sequence succeed.
         * @param packet DNS packet.
         */
        static void defaultOnSuccess (const DnsPacket& packet)
        {
            std::cout << std::endl;
            std::cout << "PEER: " << packet.dest << "#" << packet.port << std::endl;
 
            std::cout << std::endl;
            std::cout << ";; QUESTION SECTION: " << std::endl;
            for (auto const& question : packet.questions)
            {
                std::cout << question.host;
                std::cout << "  " << DnsMessage::typeName (question.type);
                std::cout << "  " << DnsMessage::className (question.dnsclass);
                std::cout << std::endl;
            }
 
            std::cout << std::endl;
            std::cout << ";; ANSWER SECTION: " << std::endl;
            for (auto const& answer : packet.answers)
            {
                std::cout << answer.host;
                std::cout << "  " << DnsMessage::typeName (answer.type);
                std::cout << "  " << DnsMessage::className (answer.dnsclass);
                std::cout << "  " << answer.ttl;
                if (answer.type == DnsMessage::RecordType::A)
                {
                    std::cout << "  " << answer.addr;
                }
                else if (answer.type == DnsMessage::RecordType::PTR)
                {
                    std::cout << "  " << answer.name;
                }
                else if (answer.type == DnsMessage::RecordType::AAAA)
                {
                    std::cout << "  " << answer.addr;
                }
                std::cout << std::endl;
            }
        }
 
        /*
         * @brief default callback called when a lookup sequence failed.
         * @param packet DNS packet.
         */
        static void defaultOnFailure (const DnsPacket& packet)
        {
            std::cout << std::endl;
            std::cout << "PEER: " << packet.dest << "#" << packet.port << std::endl;
 
            std::cout << std::endl;
            std::cout << ";; QUESTION SECTION: " << std::endl;
            for (auto const& question : packet.questions)
            {
                std::cout << question.host;
                std::cout << "  " << DnsMessage::typeName (question.type);
                std::cout << "  " << DnsMessage::className (question.dnsclass);
                std::cout << std::endl;
            }
 
            std::cout << std::endl;
            std::cout << lastError.message () << std::endl;
        }
#endif
 
        void notify (const DnsNotify& func, const DnsPacket& packet) const noexcept
        {
            if (func)
            {
                func (packet);
            }
        }
 
        int query (DnsPacket& packet, std::chrono::milliseconds timeout)
        {
            IpAddress mcast = Protocol::multicastAddress (this->_socket.family ());
            packet.dest = IpAddress (mcast.addr (), mcast.length (), _ifindex);
            packet.port = Protocol::defaultPort;
 
            std::stringstream data;
            if (this->_message.serialize (packet, data) == -1)
            {
                // LCOV_EXCL_START
                lastError = make_error_code (Errc::InvalidParam);
                return -1;
                // LCOV_EXCL_STOP
            }
 
            std::string buffer = data.str ();
            if (buffer.size () > Protocol::maxMsgSize)
            {
                // LCOV_EXCL_START
                lastError = make_error_code (Errc::MessageTooLong);
                return -1;
                // LCOV_EXCL_STOP
            }
 
            ScopedLock<Mutex> lock (_syncMutex);
 
            auto inserted = _pending.emplace (packet.id, std::make_unique<PendingRequest> ());
            if (!inserted.second)
            {
                // LCOV_EXCL_START
                lastError = make_error_code (Errc::OperationFailed);
                notify (onFailure, packet);
                return -1;
                // LCOV_EXCL_STOP
            }
 
            if (this->_socket.writeTo (buffer.data (), buffer.size (), {packet.dest, packet.port}) == -1)
            {
                // LCOV_EXCL_START
                _pending.erase (inserted.first);
                notify (onFailure, packet);
                return -1;
                // LCOV_EXCL_STOP
            }
 
            if (!inserted.first->second->cond.timedWait (lock, timeout))
            {
                // LCOV_EXCL_START
                _pending.erase (inserted.first);
                lastError = make_error_code (Errc::TimedOut);
                notify (onFailure, packet);
                return -1;
                // LCOV_EXCL_STOP
            }
 
            auto pendingReq = std::move (inserted.first->second);
            _pending.erase (inserted.first);
 
            if (pendingReq->ec)
            {
                // LCOV_EXCL_START
                lastError = pendingReq->ec;
                notify (onFailure, packet);
                return -1;
                // LCOV_EXCL_STOP
            }
 
            packet = std::move (pendingReq->packet);
            notify (onSuccess, packet);
 
            return 0;
        }
 
        unsigned int _ifindex;
 
        struct PendingRequest
        {
            Condition cond;     
            DnsPacket packet;   
            std::error_code ec; 
        };
 
        std::unordered_map<uint16_t, std::unique_ptr<PendingRequest>> _pending;
 
        Mutex _syncMutex;
    };
}
 
#endif