/*Copyright (C) 2006  Adam CharrettThis program is free software; you can redistribute it and/ormodify it under the terms of the GNU General Public Licenseas published by the Free Software Foundation; either version 2of the License, or (at your option) any later version.This program is distributed in the hope that it will be useful,but WITHOUT ANY WARRANTY; without even the implied warranty ofMERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See theGNU General Public License for more details.You should have received a copy of the GNU General Public Licensealong with this program; if not, write to the Free SoftwareFoundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USAbinarycomms.cBinary Communications protocol for control DVBStreamer.*/#include "config.h"#include <limits.h>#include <stdio.h>#include <stdlib.h>#include <string.h>#include <time.h>#include <unistd.h>#include <sys/types.h>#include <sys/poll.h>#include <sys/socket.h>#include <netinet/in.h>#include <arpa/inet.h>#include "ts.h"#include "udpoutput.h"#include "outputs.h"#include "logging.h"#include "cache.h"#include "messages.h"#include "main.h"#include "binarycomms.h"#define MAX_CONNECTIONS 2 /* 1 for monitoring by web and another for control */#define LOG_MALFORMED(_connection, _field) \    printlog(LOG_DEBUG, "%s:%d : Malformed message, not enough data in message to read %s field!\n", \        inet_ntoa((_connection)->clientAddress.sin_addr), (_connection)->clientAddress.sin_port, _field)#define IFAUTHENTICATED(_dofunc, _connection, _message) \    do { \        if ((_connection)->authenticated)\        {\            _dofunc(_connection, _message);\        }\        else \        { \          MessageRERR(_message, RERR_NOTAUTHORISED, "Not authorised!"); \        }\    }while(0)#define MessageRERR(_msg, _errcode, _str) \    do{\        MessageInit(_msg, MSGCODE_RERR);\        MessageEncode(_msg, "bs", _errcode, _str);\    }while(0)#define STRINGIFY(x) #x#define TOSTRING(x) STRINGIFY(x)#define READSTRING(_var)\    do { \        if (MessageReadString(message, &_var))\        {\            LOG_MALFORMED(connection, TOSTRING(_var));\            connection->connected = FALSE;\            return ;\        }\    }while(0)#define READ2STRINGS(_var1, _var2)\    do { \        if (MessageReadString(message, &_var1))\        {\            LOG_MALFORMED(connection, TOSTRING(_var1));\            connection->connected = FALSE;\            return ;\        }\        if (MessageReadString(message, &_var2))\        {\            LOG_MALFORMED(connection, TOSTRING(_var2));\            connection->connected = FALSE;\            free(_var1);\            return ;\        }\    }while(0)#define READUINT8(_var)\    do { \        if (MessageReadUint8(message, &_var))\        {\            LOG_MALFORMED(connection, TOSTRING(_var));\            connection->connected = FALSE;\            return ;\        }\    }while(0)#define READUINT16(_var)\    do { \        if (MessageReadUint16(message, &_var))\        {\            LOG_MALFORMED(connection, TOSTRING(_var));\            connection->connected = FALSE;\            return ;\        }\    }while(0)#define READUINT32(_var)\    do { \        if (MessageReadUint32(message, &_var))\        {\            LOG_MALFORMED(connection, TOSTRING(_var));\            connection->connected = FALSE;\            return ;\        }\    }while(0)typedef struct Connection_t{    int socketfd;    struct sockaddr_in clientAddress;    bool authenticated;    bool active;    bool connected;    pthread_t thread;    Message_t message;}Connection_t;static void HandleConnection(Connection_t *connection);static void ProcessMessage(Connection_t *connection, Message_t *message);static void ProcessInfo(Connection_t *connection, Message_t *message);static void ProcessAuth(Connection_t *connection, Message_t *message);static void ProcessPrimaryServiceSelect(Connection_t *connection, Message_t *message);static void ProcessSecondaryServiceAdd(Connection_t *connection, Message_t *message);static void ProcessSecondaryServiceSet(Connection_t *connection, Message_t *message);static void ProcessSecondaryServiceRemove(Connection_t *connection, Message_t *message);static void ProcessOutputAdd(Connection_t *connection, Message_t *message);static void ProcessOutputRemove(Connection_t *connection, Message_t *message);static void ProcessOutputPIDAdd(Connection_t *connection, Message_t *message);static void ProcessOutputPIDRemove(Connection_t *connection, Message_t *message);static void ProcessPrimaryServiceCurrent(Connection_t *connection, Message_t *message);static void ProcessSecondaryServiceList(Connection_t *connection, Message_t *message);static void ProcessOutputsList(Connection_t *connection, Message_t *message);static void ProcessOutputListPids(Connection_t *connection, Message_t *message);static void ProcessServiceFilterPacketCount(Connection_t *connection, Message_t *message);static void ProcessOutputPacketCount(Connection_t *connection, Message_t *message);static void ProcessTSStats(Connection_t *connection, Message_t *message);static void ProcessFEStatus(Connection_t *connection, Message_t *message);static void ProcessServiceList(Connection_t *connection, Message_t *message, int all);static void ProcessServicePids(Connection_t *connection, Message_t *message);static int activeConnections = 0;static pthread_mutex_t activeConnectionsMutex = PTHREAD_MUTEX_INITIALIZER;static int serverSocket;static Connection_t connections[MAX_CONNECTIONS];static char *infoStreamerName;static char *authUsername;static char *authPassword;static time_t serverStartTime;int BinaryCommsInit(int adapter, char *streamername, char *username, char *password){    struct sockaddr_in serverAddress;    serverSocket = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);    if (serverSocket < 0)    {        printlog(LOG_ERROR, "Failed to create server socket!\n");        return 1;    }    bzero((char *) &serverAddress, sizeof(serverAddress));    serverAddress.sin_family = AF_INET;    serverAddress.sin_addr.s_addr = INADDR_ANY;    serverAddress.sin_port = htons(BINARYCOMMS_PORT + adapter);    if (bind(serverSocket, (struct sockaddr *) &serverAddress, sizeof(serverAddress)) < 0)    {        printlog(LOG_ERROR, "Failed to bind server to port %d", BINARYCOMMS_PORT + adapter);        close(serverSocket);        return 1;    }    listen(serverSocket, 1);    infoStreamerName = strdup(streamername);    authUsername = strdup(username);    authPassword = strdup(password);    time(&serverStartTime);    printlog(LOG_INFO, "Server created %s", ctime(&serverStartTime));    printlog(LOG_DEBUG, "Username    : %s\n", authUsername);    printlog(LOG_DEBUG, "Password    : %s\n", authPassword);    printlog(LOG_DEBUG, "Server Name : %s\n", infoStreamerName);    return 0;}void BinaryCommsDeInit(void){    int i;    close(serverSocket);    for ( i = 0; i < MAX_CONNECTIONS; i ++)    {        if (connections[i].connected)        {            /* Force closure of the sockets should cause the thread to terminate. */            close(connections[i].socketfd);            connections[i].connected = FALSE;        }    }}void BinaryCommsAcceptConnections(void){    struct pollfd pfd[1];    pfd[0].fd = serverSocket;    pfd[0].events = POLLIN;    while (!ExitProgram)    {        if (poll(pfd, 1, 200))        {            if (pfd[0].revents & POLLIN)            {                int clientfd;                struct sockaddr_in clientAddress;                int clientAddressSize;                clientAddressSize = sizeof(clientAddress);                clientfd = accept(serverSocket, (struct sockaddr *) & clientAddress, &clientAddressSize);                pthread_mutex_lock(&activeConnectionsMutex);                if (activeConnections >= MAX_CONNECTIONS)                {                    printlog(LOG_INFO, "Connection attempt from %s:%d rejected as too many open connections!\n",                             inet_ntoa(clientAddress.sin_addr), clientAddress.sin_port);                    close(clientfd);                }                else                {                    int found = -1;                    int i;                    for (i = 0; i < MAX_CONNECTIONS; i ++)                    {                        if (!connections[i].active)                        {                            found = i;                            break;                        }                    }                    if (found >= 0)                    {                        connections[i].active = TRUE;                        connections[i].connected = TRUE;                        connections[i].socketfd = clientfd;                        connections[i].clientAddress = clientAddress;                        activeConnections ++;                        printlog(LOG_INFO, "Connection attempt from %s:%d accepted!\n",                                 inet_ntoa(clientAddress.sin_addr), clientAddress.sin_port);                        pthread_create(&connections[i].thread, NULL, (void*)HandleConnection, (void*)&connections[i]);                    }                    else                    {                        printlog(LOG_INFO, "Connection attempt from %s:%d rejected as no connections structures left!\n",                                 inet_ntoa(clientAddress.sin_addr), clientAddress.sin_port);                        close(clientfd);                    }                }                pthread_mutex_unlock(&activeConnectionsMutex);            }        }    }}static void HandleConnection(Connection_t *connection){    struct pollfd pfd[1];    int socketfd = connection->socketfd;    pfd[0].fd = socketfd;    pfd[0].events = POLLIN;    Message_t *message = &connection->message;    while (!ExitProgram && connection->connected)    {        if (MessageRecv(message, socketfd))        {            /* Socket must be dead exit this connection! */            break;        }        ProcessMessage(connection, message);        if (connection->connected)        {            printlog(LOG_DEBUG, "%s:%d : Response 0x%04x length %d\n",                     inet_ntoa(connection->clientAddress.sin_addr), connection->clientAddress.sin_port,                     MessageGetCode(message), MessageGetLength(message));            if (MessageSend(message, socketfd))            {                /* Socket must be dead exit this connection! */                break;            }        }    }    printlog(LOG_INFO, "%s:%d : Connection closed!\n",             inet_ntoa(connection->clientAddress.sin_addr), connection->clientAddress.sin_port);    /* Close the socket and free our resources */    close(socketfd);    connection->connected = FALSE;    pthread_mutex_lock(&activeConnectionsMutex);    connection->active = FALSE;    activeConnections --;    pthread_mutex_unlock(&activeConnectionsMutex);}static void ProcessMessage(Connection_t *connection, Message_t *message){    printlog(LOG_DEBUG, "%s:%d : Processing message 0x%04x length %d\n",             inet_ntoa(connection->clientAddress.sin_addr), connection->clientAddress.sin_port,             MessageGetCode(message), MessageGetLength(message));    switch (message->code)    {        case MSGCODE_INFO:            ProcessInfo(connection, message);            break;        case MSGCODE_AUTH:            ProcessAuth(connection, message);            break;            /* Control Messages */        case MSGCODE_CSPS:            IFAUTHENTICATED(ProcessPrimaryServiceSelect, connection, message);            break;        case MSGCODE_CSSA:            IFAUTHENTICATED(ProcessSecondaryServiceAdd, connection, message);            break;        case MSGCODE_CSSS:            IFAUTHENTICATED(ProcessSecondaryServiceSet, connection, message);            break;        case MSGCODE_CSSR:            IFAUTHENTICATED(ProcessSecondaryServiceRemove, connection, message);            break;        case MSGCODE_COAO:            IFAUTHENTICATED(ProcessOutputAdd, connection, message);            break;        case MSGCODE_CORO:            IFAUTHENTICATED(ProcessOutputRemove, connection, message);            break;        case MSGCODE_COAP:            IFAUTHENTICATED(ProcessOutputPIDAdd, connection, message);            break;        case MSGCODE_CORP:            IFAUTHENTICATED(ProcessOutputPIDRemove, connection, message);            break;            /* Status Messages */        case MSGCODE_SSPS:            ProcessPrimaryServiceCurrent(connection, message);            break;        case MSGCODE_SSFL:            ProcessSecondaryServiceList(connection, message);            break;        case MSGCODE_SSPC:            ProcessServiceFilterPacketCount(connection, message);            break;        case MSGCODE_SOLO:            ProcessOutputsList(connection, message);            break;        case MSGCODE_SOLP:            ProcessOutputListPids(connection, message);            break;        case MSGCODE_SOPC:            ProcessOutputPacketCount(connection, message);            break;        case MSGCODE_STSS:            ProcessTSStats(connection, message);            break;        case MSGCODE_SFES:            ProcessFEStatus(connection, message);            break;        case MSGCODE_SSLA:            ProcessServiceList(connection, message, 1);            break;        case MSGCODE_SSLM:            ProcessServiceList(connection, message, 0);            break;        case MSGCODE_SSPL:            ProcessServicePids(connection, message);            break;        default:            MessageRERR(message, RERR_GENERIC, "Unknown message type!");            break;    }}static void ProcessInfo(Connection_t *connection, Message_t *message){    uint8_t info;    READUINT8(info);    switch (info)    {        case INFO_NAME:            MessageRERR(message, RERR_OK, infoStreamerName);            break;        case INFO_FETYPE:            MessageRERR(message, RERR_OK, "Not implemented!");            break;        case INFO_AUTHENTICATED:            MessageRERR(message, RERR_OK, connection->authenticated ? "Authenticated" : "Not authenticated");            break;        case INFO_UPSECS:            {                char buffer[11];                time_t now;                time(&now);                sprintf(buffer, "%d", (int)difftime(now, serverStartTime));                MessageRERR(message, RERR_OK, buffer);            }            break;        case INFO_UPTIME:            {                char buffer[50];                time_t now;                int seconds;                int d, h, m, s;                time(&now);                seconds = (int)difftime(now, serverStartTime);                d = seconds / (24 * 60 * 60);                h = (seconds - (d * 24 * 60 * 60)) / (60 * 60);                m = (seconds - ((d * 24 * 60 * 60) + (h * 60 * 60))) / 60;                s = (seconds - ((d * 24 * 60 * 60) + (h * 60 * 60) + (m * 60)));                sprintf(buffer, "%d Days %d Hours %d Minutes %d seconds", d, h, m, s);                MessageRERR(message, RERR_OK, buffer);            }            break;        default:            MessageRERR(message, RERR_GENERIC, "Unknown field");            break;    }}static void ProcessAuth(Connection_t *connection, Message_t *message){    char *msgUsername = NULL;    char *msgPassword = NULL;    READ2STRINGS(msgUsername, msgPassword);    connection->authenticated = (strcmp(msgUsername, authUsername) == 0) &&                                (strcmp(msgPassword, authPassword) == 0);    MessageRERR(message, connection->authenticated ? RERR_OK : RERR_NOTAUTHORISED, NULL);    free(msgUsername);    free(msgPassword);}static void ProcessPrimaryServiceSelect(Connection_t *connection, Message_t *message){    char *serviceName;    READSTRING(serviceName);