socket.cpp

/*
 * socket.c: IPTV plugin for the Video Disk Recorder
 *
 * See the README file for copyright information and how to reach the author.
 *
 */

#include <sys/socket.h>
#include <netdb.h>
#include <fcntl.h>
#include <unistd.h>

#include <vdr/device.h>

#include "common.h"
#include "config.h"
#include "log.h"
#include "socket.h"

#define BUFFER_SIZE (1024*1024)

cIptvSocket::cIptvSocket()
    : socketPortM(0),
      socketDescM(-1),
      lastErrorReportM(0),
      packetErrorsM(0),
      sequenceNumberM(-1),
      isActiveM(false) {
    debug1("%s", __PRETTY_FUNCTION__);

    memset(&sockAddrM, 0, sizeof(sockAddrM));
}

cIptvSocket::~cIptvSocket() {
    debug1("%s", __PRETTY_FUNCTION__);

    // Close the socket
    CloseSocket();
}

bool cIptvSocket::OpenSocket(const int portP, const bool isUdpP) {
    debug1("%s (%d, %d)", __PRETTY_FUNCTION__, portP, isUdpP);

    // If socket is there already, and it is bound to a different port, it must
    // be closed first
    if (portP != socketPortM) {
        debug1("%s (%d, %d) Socket tear-down", __PRETTY_FUNCTION__, portP, isUdpP);

        CloseSocket();
    }

    // Bind to the socket if it is not active already
    if (socketDescM < 0) {
        int yes = 1;

        // Create socket
        if (isUdpP) {
            socketDescM = socket(PF_INET, SOCK_DGRAM, 0);
        } else {
            socketDescM = socket(PF_INET, SOCK_STREAM, 0);
        }

        ERROR_IF_RET(socketDescM < 0, "socket()", return false);

        // Make it use non-blocking I/O to avoid stuck read calls
        ERROR_IF_FUNC(fcntl(socketDescM, F_SETFL, O_NONBLOCK), "fcntl(O_NONBLOCK)", CloseSocket(), return false);

        // Allow multiple sockets to use the same PORT number
        ERROR_IF_FUNC(setsockopt(socketDescM, SOL_SOCKET, SO_REUSEADDR, &yes, sizeof(yes)) < 0, "setsockopt(SO_REUSEADDR)", CloseSocket(), return false);
#ifndef __FreeBSD__
        // Allow packet information to be fetched
        ERROR_IF_FUNC(setsockopt(socketDescM, SOL_IP, IP_PKTINFO, &yes, sizeof(yes)) < 0, "setsockopt(IP_PKTINFO)", CloseSocket(), return false);
#endif // __FreeBSD__
        // Bind socket
        memset(&sockAddrM, 0, sizeof(sockAddrM));
        sockAddrM.sin_family = AF_INET;
        sockAddrM.sin_port = htons((uint16_t) (portP & 0xFFFF));
        sockAddrM.sin_addr.s_addr = htonl(INADDR_ANY);

        auto optval = BUFFER_SIZE;
        if ((setsockopt(socketDescM, SOL_SOCKET, SO_RCVBUF, &optval, sizeof(optval))) < 0) {
            debug1("setsockopt(SO_RCVBUF): %s.\n", strerror(errno));
        }

        if (isUdpP) {
            ERROR_IF_FUNC(bind(socketDescM, (struct sockaddr *) &sockAddrM, sizeof(sockAddrM)) < 0, "bind()", CloseSocket(), return false);
        }

        // Update socket port
        socketPortM = portP;
    }

    return true;
}

void cIptvSocket::CloseSocket() {
    debug1("%s", __PRETTY_FUNCTION__);

    // Check if socket exists
    if (socketDescM >= 0) {
        close(socketDescM);
        socketDescM = -1;
        socketPortM = 0;
        memset(&sockAddrM, 0, sizeof(sockAddrM));
    }

    if (packetErrorsM) {
        info("Detected %d RTP packet errors", packetErrorsM);
        packetErrorsM = 0;
        lastErrorReportM = time(nullptr);
    }
}

bool cIptvSocket::CheckAddress(const char *addrP, in_addr_t *inAddrP) {
    if (inAddrP) {
        // First try only the IP address
        *inAddrP = inet_addr(addrP);

        if (*inAddrP==htonl(INADDR_NONE)) {
            debug1("%s (%s, ) Cannot convert to address", __PRETTY_FUNCTION__, addrP);

            // It may be a host name, get the name
            struct hostent *host = gethostbyname(addrP);
            if (!host) {
                char tmp[64];
                error("gethostbyname() failed: %s is not valid address: %s", addrP,
                      strerror_r(h_errno, tmp, sizeof(tmp)));

                return false;
            }

            *inAddrP = inet_addr(*host->h_addr_list);
        }

        return true;
    }

    return false;
}

// UDP socket class
cIptvUdpSocket::cIptvUdpSocket()
    : streamAddrM(htonl(INADDR_ANY)),
      sourceAddrM(htonl(INADDR_ANY)),
      useIGMPv3M(false) {
    debug1("%s", __PRETTY_FUNCTION__);
}

cIptvUdpSocket::~cIptvUdpSocket() {
    debug1("%s", __PRETTY_FUNCTION__);
}

bool cIptvUdpSocket::OpenSocket(const int portP) {
    debug1("%s (%d)", __PRETTY_FUNCTION__, portP);

    streamAddrM = htonl(INADDR_ANY);
    sourceAddrM = htonl(INADDR_ANY);
    useIGMPv3M = false;

    return cIptvSocket::OpenSocket(portP, true);
}

bool cIptvUdpSocket::OpenSocket(const int portP, const char *streamAddrP, const char *sourceAddrP, bool useIGMPv3P) {
    debug1("%s (%d, %s, %s, %d)", __PRETTY_FUNCTION__, portP, streamAddrP, sourceAddrP, useIGMPv3P);

    CheckAddress(streamAddrP, &streamAddrM);
    CheckAddress(sourceAddrP, &sourceAddrM);
    useIGMPv3M = useIGMPv3P;

    return cIptvSocket::OpenSocket(portP, true);
}

void cIptvUdpSocket::CloseSocket() {
    debug1("%s", __PRETTY_FUNCTION__);

    streamAddrM = htonl(INADDR_ANY);
    sourceAddrM = htonl(INADDR_ANY);
    useIGMPv3M = false;
    cIptvSocket::CloseSocket();
}

bool cIptvUdpSocket::JoinMulticast() {
    debug1("%s", __PRETTY_FUNCTION__);

    // Check if socket exists
    if (!isActiveM && (socketDescM >= 0)) {
        // Join a new multicast group
        if (useIGMPv3M) {
            // Source-specific multicast (SSM) is used
            struct group_source_req gsr;
            struct sockaddr_in *grp;
            struct sockaddr_in *src;
            gsr.gsr_interface = 0; // if_nametoindex("any") ?
            grp = (struct sockaddr_in *) &gsr.gsr_group;
            grp->sin_family = AF_INET;
            grp->sin_addr.s_addr = streamAddrM;
            grp->sin_port = 0;
            src = (struct sockaddr_in *) &gsr.gsr_source;
            src->sin_family = AF_INET;
            src->sin_addr.s_addr = sourceAddrM;
            src->sin_port = 0;
            ERROR_IF_RET(setsockopt(socketDescM, SOL_IP, MCAST_JOIN_SOURCE_GROUP, &gsr, sizeof(gsr)) < 0,
                         "setsockopt(MCAST_JOIN_SOURCE_GROUP)",
                         return false);
        } else {
            struct ip_mreq mreq;
            mreq.imr_multiaddr.s_addr = streamAddrM;
            mreq.imr_interface.s_addr = htonl(INADDR_ANY);
            ERROR_IF_RET(setsockopt(socketDescM, SOL_IP, IP_ADD_MEMBERSHIP, &mreq, sizeof(mreq)) < 0,
                         "setsockopt(IP_ADD_MEMBERSHIP)",
                         return false);
        }

        // Update multicasting flag
        isActiveM = true;
    }

    return true;
}

bool cIptvUdpSocket::DropMulticast() {
    debug1("%s", __PRETTY_FUNCTION__);

    // Check if socket exists
    if (isActiveM && (socketDescM >= 0)) {
        // Drop the existing multicast group
        if (useIGMPv3M) {
            // Source-specific multicast (SSM) is used
            struct group_source_req gsr;
            struct sockaddr_in *grp;
            struct sockaddr_in *src;
            gsr.gsr_interface = 0; // if_nametoindex("any") ?
            grp = (struct sockaddr_in *) &gsr.gsr_group;
            grp->sin_family = AF_INET;
            grp->sin_addr.s_addr = streamAddrM;
            grp->sin_port = 0;
            src = (struct sockaddr_in *) &gsr.gsr_source;
            src->sin_family = AF_INET;
            src->sin_addr.s_addr = sourceAddrM;
            src->sin_port = 0;
            ERROR_IF_RET(setsockopt(socketDescM, SOL_IP, MCAST_LEAVE_SOURCE_GROUP, &gsr, sizeof(gsr)) < 0, "setsockopt(MCAST_LEAVE_SOURCE_GROUP)", return false);
        } else {
            struct ip_mreq mreq;
            mreq.imr_multiaddr.s_addr = streamAddrM;
            mreq.imr_interface.s_addr = htonl(INADDR_ANY);
            ERROR_IF_RET(setsockopt(socketDescM, SOL_IP, IP_DROP_MEMBERSHIP, &mreq, sizeof(mreq)) < 0, "setsockopt(IP_DROP_MEMBERSHIP)", return false);
        }

        // Update multicasting flag
        isActiveM = false;
    }

    return true;
}

int cIptvUdpSocket::Read(unsigned char *bufferAddrP, unsigned int bufferLenP) {
    debug16("%s", __PRETTY_FUNCTION__);

    // Error out if socket not initialized
    if (socketDescM <= 0) {
        error("%s Invalid socket", __PRETTY_FUNCTION__);
        return -1;
    }

    int len = 0;
    // Read data from socket in a loop
    do {
        socklen_t addrlen = sizeof(sockAddrM);
        struct msghdr msgh;
        struct iovec iov;
        char cbuf[256];
        len = 0;
        // Initialize iov and msgh structures
        memset(&msgh, 0, sizeof(struct msghdr));
        iov.iov_base = bufferAddrP;
        iov.iov_len = bufferLenP;
        msgh.msg_control = cbuf;
        msgh.msg_controllen = sizeof(cbuf);
        msgh.msg_name = &sockAddrM;
        msgh.msg_namelen = addrlen;
        msgh.msg_iov = &iov;
        msgh.msg_iovlen = 1;
        msgh.msg_flags = 0;

        if (isActiveM && socketDescM && bufferAddrP && (bufferLenP > 0)) {
            len = (int) recvmsg(socketDescM, &msgh, MSG_DONTWAIT);
        } else {
            break;
        }

        if (len > 0) {
#ifndef __FreeBSD__
            // Process auxiliary received data and validate source address
            for (struct cmsghdr *cmsg = CMSG_FIRSTHDR(&msgh); cmsg !=nullptr; cmsg = CMSG_NXTHDR(&msgh, cmsg)) {
                if ((cmsg->cmsg_level == SOL_IP) && (cmsg->cmsg_type == IP_PKTINFO)) {
                    struct in_pktinfo *i = (struct in_pktinfo *) CMSG_DATA(cmsg);
                    if ((i->ipi_addr.s_addr == streamAddrM) || (htonl(INADDR_ANY) == streamAddrM)) {
#endif // __FreeBSD__
                        if (bufferAddrP[0] == TS_SYNC_BYTE) {
                            return len;
                        } else if (len > 3) {
                            // http://tools.ietf.org/html/rfc3550
                            // http://tools.ietf.org/html/rfc2250
                            // Version
                            unsigned int v = (bufferAddrP[0] >> 6) & 0x03;
                            // Extension bit
                            unsigned int x = (bufferAddrP[0] >> 4) & 0x01;
                            // CSCR count
                            unsigned int cc = bufferAddrP[0] & 0x0F;
                            // Payload type: MPEG2 TS = 33
                            //unsigned int pt = bufferAddrP[1] & 0x7F;
                            // Sequence number
                            int seq = ((bufferAddrP[2] & 0xFF) << 8) | (bufferAddrP[3] & 0xFF);
                            if ((((sequenceNumberM + 1)%0xFFFF)==0) && (seq==0xFFFF)) {
                                sequenceNumberM = -1;
                            } else if ((sequenceNumberM >= 0) && (((sequenceNumberM + 1)%0xFFFF)!=seq)) {
                                packetErrorsM++;
                                if (time(nullptr) - lastErrorReportM > eReportIntervalS) {
                                    info("Detected %d RTP packet errors", packetErrorsM);
                                    packetErrorsM = 0;
                                    lastErrorReportM = time(nullptr);
                                }
                                sequenceNumberM = seq;
                            } else {
                                sequenceNumberM = seq;
                            }

                            // Header lenght
                            unsigned int headerlen = (3 + cc)*(unsigned int) sizeof(uint32_t);

                            // Check if extension
                            if (x) {
                                // Extension header length
                                unsigned int ehl = (((bufferAddrP[headerlen + 2] & 0xFF) << 8) | (bufferAddrP[headerlen + 3] & 0xFF));
                                // Update header length
                                headerlen += (ehl + 1)*(unsigned int) sizeof(uint32_t);
                            }

                            // Check that rtp is version 2 and payload contains multiple of TS packet data
                            if ((v == 2) && (((len - headerlen) % TS_SIZE)==0) &&
                                (bufferAddrP[headerlen] == TS_SYNC_BYTE)) {
                                // Set argument point to payload in read buffer
                                memmove(bufferAddrP, &bufferAddrP[headerlen], (len - headerlen));
                                return (len - headerlen);
                            }
                        }
#ifndef __FreeBSD__
                    }
                }
            }
#endif // __FreeBSD__
        }
    } while (len > 0);
    ERROR_IF_RET(len < 0 && errno!=EAGAIN, "recvmsg()", return -1);

    return 0;
}

// TCP socket class
cIptvTcpSocket::cIptvTcpSocket() {
    debug1("%s", __PRETTY_FUNCTION__);
}

cIptvTcpSocket::~cIptvTcpSocket() {
    debug1("%s", __PRETTY_FUNCTION__);
}

bool cIptvTcpSocket::OpenSocket(const int portP, const char *streamAddrP) {
    debug1("%s (%d, %s)", __PRETTY_FUNCTION__, portP, streamAddrP);

    // Socket must be opened before setting the host address
    return (cIptvSocket::OpenSocket(portP, false) && CheckAddress(streamAddrP, &sockAddrM.sin_addr.s_addr));
}

void cIptvTcpSocket::CloseSocket() {
    debug1("%s()", __PRETTY_FUNCTION__);

    cIptvSocket::CloseSocket();
}

bool cIptvTcpSocket::ConnectSocket() {
    debug1("%s", __PRETTY_FUNCTION__);

    if (!isActiveM && (socketDescM >= 0)) {
        int retval = connect(socketDescM, (struct sockaddr *) &sockAddrM, sizeof(sockAddrM));

        // Non-blocking sockets always report in-progress error when connected
        ERROR_IF_RET(retval < 0 && errno!=EINPROGRESS, "connect()", return false);

        // Select with 800ms timeout on the socket completion, check if it is writable
        retval = select_single_desc(socketDescM, 800000, true);
        if (retval < 0) {
            return retval;
        }

        // Select has returned. Get socket errors if there are any
        retval = 0;
        socklen_t len = sizeof(retval);
        getsockopt(socketDescM, SOL_SOCKET, SO_ERROR, &retval, &len);

        // If not any errors, then socket must be ready and connected
        if (retval!=0) {
            char tmp[64];
            error("Connect() failed: %s", strerror_r(retval, tmp, sizeof(tmp)));

            return false;
        }
    }

    return true;
}

int cIptvTcpSocket::Read(unsigned char *bufferAddrP, unsigned int bufferLenP) {
    debug16("%s", __PRETTY_FUNCTION__);

    // Error out if socket not initialized
    if (socketDescM <= 0) {
        error("%s Invalid socket", __PRETTY_FUNCTION__);

        return -1;
    }

    int len = 0;
    socklen_t addrlen = sizeof(sockAddrM);

    // Read data from socket
    if (isActiveM && socketDescM && bufferAddrP && (bufferLenP > 0)) {
        len = (int) recvfrom(socketDescM, bufferAddrP, bufferLenP, MSG_DONTWAIT, (struct sockaddr *) &sockAddrM, &addrlen);
    }

    return len;
}

bool cIptvTcpSocket::ReadChar(char *bufferAddrP, unsigned int timeoutMsP) {
    debug16("%s", __PRETTY_FUNCTION__);

    // Error out if socket not initialized
    if (socketDescM <= 0) {
        error("%s Invalid socket", __PRETTY_FUNCTION__);

        return false;
    }

    socklen_t addrlen = sizeof(sockAddrM);

    // Wait 500ms for data
    int retval = select_single_desc(socketDescM, 1000*timeoutMsP, false);

    // Check if error
    if (retval < 0) {
        return false;
        // Check if data available
    } else if (retval) {
        retval = (int) recvfrom(socketDescM, bufferAddrP, 1, MSG_DONTWAIT, (struct sockaddr *) &sockAddrM, &addrlen);

        if (retval <= 0)
            return false;
    }

    return true;
}

bool cIptvTcpSocket::Write(const char *bufferAddrP, unsigned int bufferLenP) {
    debug16("%s", __PRETTY_FUNCTION__);

    // Error out if socket not initialized
    if (socketDescM <= 0) {
        error("%s Invalid socket", __PRETTY_FUNCTION__);

        return false;
    }

    ERROR_IF_RET(send(socketDescM, bufferAddrP, bufferLenP, 0) < 0, "send()", return false);

    return true;
}

cIptvTcpServerSocket::cIptvTcpServerSocket() : clientSocketDescM(-1) {
    debug1("%s", __PRETTY_FUNCTION__);
}

cIptvTcpServerSocket::~cIptvTcpServerSocket() {
    debug1("%s", __PRETTY_FUNCTION__);
}

int cIptvTcpServerSocket::Read(unsigned char *bufferAddrP, unsigned int bufferLenP) {
    // Error out if socket not initialized
    if (clientSocketDescM <= 0) {
        // error("%s Invalid socket", __PRETTY_FUNCTION__);

        return -1;
    }

    int len = 0;
    socklen_t addrlen = sizeof(sockAddrM);

    // Read data from socket
    if (isActiveM && clientSocketDescM && bufferAddrP && (bufferLenP > 0)) {
        len = (int) recvfrom(clientSocketDescM, bufferAddrP, bufferLenP, MSG_DONTWAIT, (struct sockaddr *) &sockAddrM, &addrlen);
    }

    return len;
}

bool cIptvTcpServerSocket::OpenSocket(int portP, bool isUdpP) {
    if (socketDescM > 0) {
        return true;
    }

    int opt;

    /* create a STREAM (TCP) socket in the INET6 (IPv6) protocol */
    socketDescM = socket(PF_INET6, SOCK_STREAM, 0);
    if (socketDescM < 0) {
        error("%s: could not create socket: %m", __PRETTY_FUNCTION__);
        return false;
    }

    /* let socket accept v6 and v4 (mapped address) connections */
    opt = 0;
    if (setsockopt(socketDescM, IPPROTO_IPV6, IPV6_V6ONLY, &opt, sizeof(opt)) < 0) {
        error("%s: could not turn off IPV6_V6ONLY: %m", __PRETTY_FUNCTION__);
        return false;
    }

    /* allow socket to always reuse the same port */
    opt = 1;
    if (setsockopt(socketDescM, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt)) < 0) {
        error("%s: could not turn on SO_REUSEADDR: %m", __PRETTY_FUNCTION__);
    }

    /* bind socket to any address on port 'mPort' */
    struct sockaddr_in6 server_addr;
    memset(&server_addr, 0, sizeof(server_addr));
    server_addr.sin6_family = AF_INET6;
    server_addr.sin6_addr = in6addr_any;
    server_addr.sin6_port = htons(portP);

    if (bind(socketDescM, (struct sockaddr*) &server_addr,  sizeof(server_addr)) < 0) {
        error("%s: couldnt bind socket to port %d: %m", __PRETTY_FUNCTION__, portP);
        CloseSocket();
        return false;
    }

    /* switch socket to non-blocking */
    opt = fcntl(socketDescM, F_GETFL, 0);
    if (opt < 0) {
        error("%s: couldnt get socket flags: %m", __PRETTY_FUNCTION__);
        CloseSocket();
        return false;
    }
    opt |= O_NONBLOCK;
    if (fcntl(socketDescM, F_SETFL, opt) < 0) {
        error("%s: couldnt set socket flags: %m", __PRETTY_FUNCTION__);
        CloseSocket();
        return false;
    }

    /* start listening on socket */
    if (listen(socketDescM, 5) < 0) {
        error("%s: couldnt start listening: %m", __PRETTY_FUNCTION__);
        CloseSocket();
        return false;
    }

    /* increase read buffer if possible */
    auto optval = BUFFER_SIZE;
    if ((setsockopt(socketDescM, SOL_SOCKET, SO_RCVBUF, &optval, sizeof(optval))) < 0) {
        debug1("setsockopt(SO_RCVBUF): %s.\n", strerror(errno));
    }

    return true;
}

void cIptvTcpServerSocket::CloseSocket() {
    cIptvSocket::CloseSocket();
    if (clientSocketDescM) {
        close(clientSocketDescM);
        clientSocketDescM = 0;
    }
}

bool cIptvTcpServerSocket::Accept() {
    char addrbuf[INET6_ADDRSTRLEN];
    struct sockaddr_in6 client_addr;
    socklen_t len = sizeof(client_addr);

    clientSocketDescM = accept(socketDescM, (struct sockaddr*) &client_addr, &len);

    if (clientSocketDescM < 0) {
        if (errno != EINTR && errno != EAGAIN) {
            error("%s: accept failed: %m", __PRETTY_FUNCTION__);
        }

        return false;
    }

    /* as this is an IPv6 socket, it reports IPv4 addresses as
     * 'IPv4-mapped IPv6 address' with prefix '::ffff:', ie.
     *  '::ffff:127.0.0.1' for '127.0.0.1'
     */
    debug1("connect from %s port %d",
            inet_ntop(AF_INET6, &client_addr.sin6_addr, addrbuf, INET6_ADDRSTRLEN),
            clientSocketDescM);

    return true;
}

bool cIptvTcpServerSocket::ClientConnected() {
    return clientSocketDescM != -1;
}