Avoir de l'écoute pour les messages dans l'implémentation websocket C/C++ maison

Question

Donc je suis en train de mettre en place un websocket en C/C++, et après beaucoup de désordre j'ai eu la poignée de main pour travailler (c'était une erreur d'espace ..) . De toute façon, maintenant je suis perdu sur la façon d'écouter les messages après l'établissement de la poignée de main, j'apprends au fur et à mesure, donc c'est un peu le bordel, mais je répondrai à toutes les questions sur le code, bien sûr. Voilà ce que j'ai en ce moment:

#include <sys/socket.h> 
#include <netinet/in.h> 
#include <arpa/inet.h> 
#include <stdio.h> 
#include <stdlib.h> 
#include <unistd.h> 
#include <errno.h> 
#include <string.h> 
#include <sys/types.h> 
#include <time.h> 
//cpp                          
#include <string> 
#include <iostream> 
#include <openssl/sha.h> 
#include "include/base64.h" 

int main(int argc, char *argv[]) 
{ 
    int listenfd = 0, connfd = 0; 
    struct sockaddr_in serv_addr; 

    time_t ticks; 

    listenfd = socket(AF_INET, SOCK_STREAM, 0); 
    memset(&serv_addr, '0', sizeof(serv_addr)); 

    serv_addr.sin_family = AF_INET; 
    serv_addr.sin_addr.s_addr = htonl(INADDR_ANY); 
    serv_addr.sin_port = htons(8080); 

    bind(listenfd, (struct sockaddr*)&serv_addr, sizeof(serv_addr)); 
    listen(listenfd, 8080); 
    bool ws = false; 
    while(1) 
    { 
     int res = 0; 
     if(ws){ 
      while(true){ 
       char buffer_ws[1400]; 
       res = recv(listenfd, buffer_ws, 1400, 0); 
       if(res > 0) 
        std::cout << "Data!" << std::endl; 
       sleep(1); 
      } 
     } 
     connfd = accept(listenfd, (struct sockaddr*)NULL, NULL); 

     std::cout << "connfd = " << connfd << std::endl; 

     char buffer[1400]; 

     read(connfd, buffer, 1400); 

     std::string buf(buffer); 
     std::cout << buf << std::endl; 

     std::string reply; 

     ws = strstr(buffer, "Upgrade: websocket"); 
     if(ws){ // if websocket handshake. This works              
      std::cout << "<websocket>" << std::endl; 
      std::string key = buf.substr(buf.find("Sec-WebSocket-Key") + 19, 
           buf.substr(buf.find("Sec-WebSocket-Key")).find("\n") - 20); 

      std::cout << "key = " << key << std::endl; 
      key.append("258EAFA5-E914-47DA-95CA-C5AB0DC85B11"); 
      std::cout << "key = " << key << " length = " << key.length() << std::endl; 
      unsigned char const* hash = SHA1(reinterpret_cast<const unsigned char*>(key.c_str()), key.lengt\ 
h(), nullptr); 
      std::string b64 = base64_encode(hash, 20); 
      std::cout << "b64 = " << b64 << std::endl; 

      reply = 
       "HTTP/1.1 101 Switching Protocols\r\n" 
       "Upgrade: websocket\r\n" 
       "Connection: Upgrade\r\n" 
       "Sec-WebSocket-Accept: " + b64 + "\r\n\r\n"; 
     }else{ 
      std::cout << "<other>" << std::endl; 
      reply = 
       "HTTP/1.1 200 OK\n" 
       "\n<script>var ws = new WebSocket('ws://192.168.10.117:8080');\n" 
       "ws.addEventListener('open',function(event){" 
       "\n\tconsole.log('open!!');\n\tws.send('yo bro!');\n});\n" 
       "var sendmessage = function(){console.log('click');ws.send('test123');};</script>" 
       "<input type=\"text\"><button onclick=\"sendmessage()\">send</button>"; 
     } 
     send(connfd, reply.c_str(), reply.size(), 0); 
     close(connfd); 
     sleep(1); 
    } 
} 

En ce moment, je ne reçoivent rien lorsque je tente d'envoyer du websocket client.

Answer 1

@Jontahan,

Je pense que vous êtes hors d'un excellent départ et des tonnes de plaisir avec ce projet - mais ce n'est pas une petite entreprise et il y a beaucoup de choses à apprendre et à fixer comme vous allez.

Il y a quelques problèmes de conception dans votre code:

1. Bloc vous de progresser à l'étape suivante (l'écoute des messages tout en acceptant de nouvelles connexions); et
2. Introduire des vulnérabilités de sécurité significatives pouvant entraîner un déni de service (DoS).

Par exemple, après le blocage sur accept (au cours de laquelle, vous ne pouvez pas lire les messages des clients entrants), aller de l'avant et bloc sur read, en attendant la requête HTTP (encore une fois, votre code est en attente d'un réseau événement alors qu'il pourrait avoir de meilleures choses à faire).

Cela bloque le comportement de votre code d'effectuer des tâches, puisque votre code est en attente pour IO (accept/read) alors que d'autres événements pourraient se produire en même temps (c.-à-ce que si un message websocket arrive alors que votre code est occupé en attente sur accept pour les 10 dernières minutes?). En outre, ce comportement introduit une vulnérabilité de sécurité liée aux clients lents (c'est-à-dire, les clients avec une mauvaise réception ou des attaquants malveillants). Par exemple, que se passe-t-il si la requête HTTP prend une minute entière pour arriver? Et s'il arrive défragmenté? Et s'il arrivait une ligne à la fois? et s'il arrivait une lettre à la fois?

...

Votre code est un bon début pour tester la poignée de main, mais la conception de base doit être réexaminé avant le code peut être utilisé comme serveur réel.

Une solution - souvent employée pour les petits serveurs - est d'utiliser un design fil par connexion, donc après accept est donné naissance à un nouveau fil et il est chargé de gérer la connexion (il bloque sur read sans perturber les autres threads) . Cependant, cette solution est sous-optimale et présente un certain nombre de risques de sécurité mineurs (chaque conception présente des risques de sécurité, il s'agit de choisir le risque minimal possible).

Une autre solution, telle que mise en œuvre par certains des meilleurs (nginx/node.js) consiste à utiliser une conception «événementielle» avec un seul fil. Ceci est parfois appelé "réacteur".

Ce design est de loin l'un des meilleurs, mais il souffre d'autres problèmes et nécessite beaucoup d'attention dans le code, donc les tâches/tâches lentes ne paralysent pas tout le serveur.

Cette conception est souvent mis en œuvre à l'aide kqueue, epoll ou (faute d'un meilleur choix) poll/select

En psedo-code, il pourrait ressembler à ceci:

void defer_task(void (*func((void *)), void * arg) { 
    // place task in queue 
} 

void run_tasks() { 
    while (queue->not_empty()) { 
    task = grab_oldest_task(); 
    task.func(task.arg); 
    } 
} 

void task_listen(void * data) { 
    // open listening socket 
} 

void task_poll(void * _) { 
    // poll existing clients and listening sockets 
    // probably using `kqueue`/`epoll`/`poll`/`select` 
    defer_task(task_ondata, (void*)fd); 
    // finish with rescheduling the poll task 
    defer_task(task_poll, NULL); 
} 

void task_ondata(void * data) { 
    int fd = (intptr_t)data; 
    // handle `accept`/`read` making sure all sockets are non-blocking 
} 

int main(void) { 
    defer_task(task_listen, NULL); 
    defer_task(task_poll, NULL); 
    run_tasks(); 
} 

Voici un exemple Pour un système de planification des tâches, j'ai écrit ... il n'a pas été testé en production, mais il pourrait clarifier ce que je voulais dire sur le defer_task:

/* ***************************************************************************** 
API declarations - should be moved to a separate .h file. 
***************************************************************************** */ 

/** Defer an execution of a function for later. Returns -1 on error.*/ 
int defer(void (*func)(void *), void *arg); 

/** Performs all deferred functions until the queue had been depleated. */ 
void defer_perform(void); 

/** returns true if there are deferred functions waiting for execution. */ 
int defer_has_queue(void); 

/* ***************************************************************************** 
Compile time settings 
***************************************************************************** */ 

#ifndef DEFER_QUEUE_BUFFER 
#define DEFER_QUEUE_BUFFER 1024 
#endif 


/* ***************************************************************************** 
spinlock/sync for tasks 
***************************************************************************** */ 
#if defined(__unix__) || defined(__APPLE__) || defined(__linux__) 
#define _GNU_SOURCE 
#include <time.h> 
#endif /* _GNU_SOURCE */ 
#include <stdlib.h> 

/* manage the way threads "wait" for the lock to release */ 
#if defined(__unix__) || defined(__APPLE__) || defined(__linux__) 
/* nanosleep seems to be the most effective and efficient reschedule */ 
#define defer_nanosleep(length)            \ 
    {                   \ 
    static const struct timespec tm = {.tv_nsec = length};      \ 
    nanosleep(&tm, NULL);              \ 
    } 
#define reschedule_thread() defer_nanosleep(1) 
#define throttle_thread() defer_nanosleep(8388608UL) 

#else /* no effective rescheduling, just spin... */ 
#define reschedule_thread() 
#define throttle_thread() 
#endif 

/** locks use a single byte */ 
typedef volatile unsigned char spn_lock_i; 
/** The initail value of an unlocked spinlock. */ 
#define SPN_LOCK_INIT 0 

/* Select the correct compiler builtin method. */ 
#if defined(__has_builtin) 
#if __has_builtin(__sync_swap) 
#define SPN_LOCK_BUILTIN(...) __sync_swap(__VA_ARGS__) 
#elif __has_builtin(__sync_fetch_and_or) 
#define SPN_LOCK_BUILTIN(...) __sync_fetch_and_or(__VA_ARGS__) 
#else 
#error Required builtin "__sync_swap" or "__sync_fetch_and_or" missing from compiler. 
#endif /* defined(__has_builtin) */ 
#elif __GNUC__ > 3 
#define SPN_LOCK_BUILTIN(...) __sync_fetch_and_or(__VA_ARGS__) 
#else 
#error Required builtin "__sync_swap" or "__sync_fetch_and_or" not found. 
#endif 

/** returns 1 and 0 if the lock was successfully aquired (TRUE == FAIL). */ 
static inline int spn_trylock(spn_lock_i *lock) { 
    return SPN_LOCK_BUILTIN(lock, 1); 
} 

/** Releases a lock. */ 
static inline __attribute__((unused)) void spn_unlock(spn_lock_i *lock) { 
    __asm__ volatile("" ::: "memory"); 
    *lock = 0; 
} 
/** returns a lock's state (non 0 == Busy). */ 
static inline __attribute__((unused)) int spn_is_locked(spn_lock_i *lock) { 
    __asm__ volatile("" ::: "memory"); 
    return *lock; 
} 
/** Busy waits for the lock. */ 
static inline __attribute__((unused)) void spn_lock(spn_lock_i *lock) { 
    while (spn_trylock(lock)) { 
    reschedule_thread(); 
    } 
} 

/* ***************************************************************************** 
Data Structures 
***************************************************************************** */ 

typedef struct { 
    void (*func)(void *); 
    void *arg; 
} task_s; 

typedef struct task_node_s { 
    task_s task; 
    struct task_node_s *next; 
} task_node_s; 

static task_node_s tasks_buffer[DEFER_QUEUE_BUFFER]; 

static struct { 
    task_node_s *first; 
    task_node_s **last; 
    task_node_s *pool; 
    spn_lock_i lock; 
    unsigned char initialized; 
} deferred = {.first = NULL, 
       .last = &deferred.first, 
       .pool = NULL, 
       .lock = 0, 
       .initialized = 0}; 

/* ***************************************************************************** 
API 
***************************************************************************** */ 

/** Defer an execution of a function for later. */ 
int defer(void (*func)(void *), void *arg) { 
    if (!func) 
    return -1; 
    task_node_s *task; 
    spn_lock(&deferred.lock); 
    if (deferred.pool) { 
    task = deferred.pool; 
    deferred.pool = deferred.pool->next; 
    } else if (deferred.initialized) { 
    task = malloc(sizeof(task_node_s)); 
    if (!task) 
     goto error; 
    } else { /* lazy initialization of task buffer */ 
    deferred.initialized = 1; 
    task = tasks_buffer; 
    deferred.pool = tasks_buffer + 1; 
    for (size_t i = 2; i < DEFER_QUEUE_BUFFER; i++) { 
     tasks_buffer[i - 1].next = tasks_buffer + i; 
    } 
    } 
    *deferred.last = task; 
    deferred.last = &task->next; 
    task->task.func = func; 
    task->task.arg = arg; 
    task->next = NULL; 
    spn_unlock(&deferred.lock); 
    return 0; 
error: 
    spn_unlock(&deferred.lock); 
    return -1; 
} 

/** Performs all deferred functions until the queue had been depleted. */ 
void defer_perform(void) { 
    task_node_s *tmp; 
    task_s task; 
restart: 
    spn_lock(&deferred.lock); 
    tmp = deferred.first; 
    if (tmp) { 
    deferred.first = tmp->next; 
    if (!deferred.first) 
     deferred.last = &deferred.first; 
    task = tmp->task; 
    if (tmp <= tasks_buffer + (DEFER_QUEUE_BUFFER - 1) && tmp >= tasks_buffer) { 
     tmp->next = deferred.pool; 
     deferred.pool = tmp; 
    } else { 
     free(tmp); 
    } 
    spn_unlock(&deferred.lock); 
    task.func(task.arg); 
    goto restart; 
    } else 
    spn_unlock(&deferred.lock); 
} 

/** returns true if there are deferred functions waiting for execution. */ 
int defer_has_queue(void) { return deferred.first != NULL; }