inter-server communication

5 files changed, 358 insertions, 68 deletions

diff --git a/bin/tbroadd b/bin/tbroadd
index 11454d0..cef1eef 100755
--- a/bin/tbroadd
+++ b/bin/tbroadd
diff --git a/bin/tbroaddv b/bin/tbroaddv
index 23a13cf..829a8de 100755
--- a/bin/tbroaddv
+++ b/bin/tbroaddv
diff --git a/bin/ubroadd b/bin/ubroadd
index 178b082..d4da5d4 100755
--- a/bin/ubroadd
+++ b/bin/ubroadd
diff --git a/bin/ubroaddv b/bin/ubroaddv
index c8a946e..edceaeb 100755
--- a/bin/ubroaddv
+++ b/bin/ubroaddv
diff --git a/broadd.c b/broadd.c
index b72c362..baffd87 100644
--- a/broadd.c
+++ b/broadd.c
@@ -5,6 +5,7 @@
 ---------- BUFS           -> buffer size, values >MTU are not recommended on udp
 ---------- RATE_LIMIT_S   -> rate limit expiry window
 ---------- CLIENT_TIMEOUT -> how long to wait before evicting a udp client
+---------- WILDCARD       -> whether or not to enable wildcard. set to 0 or 1.
 ------ difference:
 ---------- udp mode does not send m_ok or p_ok
 ---------- send `N\n` to initiate connection in udp mode
@@ -15,7 +16,7 @@
 ----------
 ---------- connect
 -------------- if in udp mode, server sends `T<timeout_sec>\n` to tell you when timeout will occur.
--------------- server sends `I<id>`
+-------------- server sends `IT<id>` or `IU<id>`, depending on whether it's tcp or udp
 -------------- this is your id
 -------------- if server is full, it instead sends "full"
 ---------- send
@@ -27,10 +28,10 @@
 ------------------ only clients with `password==first_8_bytes` receive your msg
 ------------------ message shorter than 8 returns "Err"
 ------------------ default password is "00000000" ASCII
--------------- `S<BUFS_max-byte buffer>`
------------------- broadcast using default password "00000000"
 -------------- `N<whatever>`
 ------------------ does nothing. use this for initiating your UDP connection.
+-------------- `W`
+------------------ returns `W0` or `W1`, depending on whether the server has wildcard enabled or not.
 ---------- receive
 ---------------- when you send P for password
 -------------------- server sends back "p_ok"
@@ -39,9 +40,17 @@
 ---------------- when server is shutting down
 -------------------- server sends "bye"
 -------------------- close the socket when you get this msg
----------------- when admin sends you msg
+---------------- when an msg is broadcasted
+-------------------- M<id>:{T<timestamp>}<bytes>
+---------------- when admin sends you a msg
 -------------------- server sends `A<BUFS-1_max-byte buffer>`
 -------------------- this message is sent to all, regardless of password
+------ admin panel:
+-------- type the commands into server stdin to use
+-------- spaces are omitted before/after args
+-------- `E` -> closes the server
+-------- `C <target-ip> <target-port>` -> connects to another server, relaying all its messages to the current server's clients
+-------- `B <msg>` -> broadcasts message to all currently connected clients
 */
 
 // ======== change these ========
@@ -58,6 +67,9 @@
 
     // client timeout for udp mode in seconds
     #define CLIENT_TIMEOUT 30
+
+    // max age for timestamped message
+    #define DROP_OLD 5
 #endif
 // ==============================
 
@@ -76,6 +88,7 @@
 #include <stdlib.h>
 #include <string.h>
 #include <sys/socket.h>
+#include <sys/time.h>
 #include <sys/uio.h>
 #include <time.h>
 #include <unistd.h>
@@ -99,11 +112,65 @@
     #define onlytcp(...) __VA_ARGS__
 #endif
 
+// glob consts using env vars
 int env_max_clients  = MAX_CLIENTS;
 int env_bufs         = BUFS;
 int env_rate_limit_s = RATE_LIMIT_S;
+int env_wildcard     = 1;
 onlyudp(int env_client_timeout = CLIENT_TIMEOUT);
 
+#define DEDUP_SIZE 65536
+#define DEDUP_MASK (DEDUP_SIZE - 1)
+
+#define DEDUP_WINDOW 5
+
+typedef struct {
+    uint32_t hash;
+    time_t   seen_at;
+} dedup_entry_t;
+
+dedup_entry_t dedup_table[ DEDUP_SIZE ];
+
+uint32_t hash_data(char* data, int len) {
+    uint32_t hash = 2166136261u;
+    for (int i = 0; i < len; i++) {
+        hash ^= (uint8_t) data[ i ];
+        hash *= 16777619;
+    }
+    return hash;
+}
+
+int is_duplicate(char* payload, int len) {
+    if (len <= 0) return 0;
+
+    uint32_t h   = hash_data(payload, len);
+    uint32_t idx = h & DEDUP_MASK;
+    time_t   now = time(NULL);
+
+    if (dedup_table[ idx ].hash == h) {
+        if (now - dedup_table[ idx ].seen_at < DEDUP_WINDOW) { return 1; }
+    }
+
+    dedup_table[ idx ].hash    = h;
+    dedup_table[ idx ].seen_at = now;
+
+    return 0;
+}
+
+#define MAX_RELAYS 64
+
+struct relay {
+    int                     fd;
+    int                     is_udp;
+    time_t                  last_hb; // udp: last heartbeat sent
+    int                     remote_timeout;
+    struct sockaddr_storage addr;    // udp
+    socklen_t               addrlen; // udp
+};
+
+struct relay relays[ MAX_RELAYS ];
+int          relay_count = 0;
+
 // stdin nonblocking
 #define NB_EOF   -1
 #define NB_AGAIN -2
@@ -177,7 +244,8 @@ fold int nb_getstring(char str[], size_t max_size) {
 #define likely(x)   __builtin_expect(!!(x), 1)
 #define unlikely(x) __builtin_expect(!!(x), 0)
 
-uint64_t default_pass;
+uint64_t default_pass;  // default password for when a client joins.
+uint64_t wildcard_pass; // a client with this password will receive all messages with any password.
 
 onlytcp(struct client {
     int      fd;       // descriptor
@@ -198,7 +266,7 @@ onlyudp(struct client {
 struct client* clients = NULL;
 int            active  = 0;
 int            cap     = 0;
-uint64_t       nid     = 1; // next id to assign
+uint64_t       nid     = 2; // next id to assign
 
 // accept new client
 onlytcp(void add_client(int fd) {
@@ -219,7 +287,7 @@ onlytcp(void add_client(int fd) {
     clients[ active ].last_rl  = time(NULL);
     clients[ active ].rl_bytes = 0;
     char buf[ 64 ];
-    int  len = sprintf(buf, "I%" PRIu64 "\n", clients[ active ].id);
+    int  len = sprintf(buf, "IT%" PRIu64 "\n", clients[ active ].id);
     write(fd, buf, len);
     active++;
     log("clients:%d\n", active);
@@ -236,44 +304,212 @@ void del_client(int i) {
 int sock;
 
 // broadcast to all clients whose password matches
-void broadcast_msg(int si, char* payload, int msg_len, int use_default) {
-    char pay_buf[ 64 ];
-    int  pay_len = sprintf(pay_buf, "M%" PRIu64 ":", clients[ si ].id);
-    int  sent    = 0;
+void broadcast_msg(uint64_t sender_id, char* payload, int msg_len) {
+    if (is_duplicate(payload, msg_len)) return;
 
     uint64_t p;
-    char*    body;
-    int      body_len;
 
-    if (use_default) {
-        p        = default_pass;
-        body     = payload;
-        body_len = msg_len;
-    } else {
-        if (msg_len < 8) return;
-        memcpy(&p, payload, 8);
-        body     = payload + 8;
-        body_len = msg_len - 8;
+    // separate
+    if (msg_len < 8) return;
+    memcpy(&p, payload, 8);
+    char* raw_body     = payload + 8;
+    int   raw_body_len = msg_len - 8;
+
+    // sign with timestamp
+    char   temp_body_buf[ BUFS + 32 ];
+    char*  final_body;
+    int    final_body_len;
+    time_t now            = time(NULL);
+    int    is_timestamped = 0;
+
+    // dont dupe
+    if (raw_body_len >= 13 && strncmp(raw_body, "{T", 2) == 0 && raw_body[ 12 ] == '}') {
+        // get time
+        char t_str[ 11 ];
+        memcpy(t_str, raw_body + 2, 10);
+        t_str[ 10 ]   = '\0';
+        long msg_time = atol(t_str);
+
+        // check
+        if (now - msg_time > DROP_OLD) {
+            // old
+            return;
+        }
+
+        // valid
+        is_timestamped = 1;
+        final_body     = raw_body;
+        final_body_len = raw_body_len;
     }
 
-    struct iovec iov[ 2 ] = { { pay_buf, (size_t) pay_len }, { body, (size_t) body_len } };
+    if (!is_timestamped) {
+        // add stamp
+        if (raw_body_len + 13 > BUFS) {
+            // too big, drop
+            return;
+        }
+
+        // write
+        sprintf(temp_body_buf, "{T%010ld}", now);
+        // append body
+        memcpy(temp_body_buf + 13, raw_body, raw_body_len);
+
+        final_body     = temp_body_buf;
+        final_body_len = raw_body_len + 13;
+    }
 
+    int sent = 0;
+
+    // bake normal
+    char normal_buf[ 64 + BUFS ];
+    int  normal_len = sprintf(normal_buf, "M%" PRIu64 ":", sender_id);
+    memcpy(normal_buf + normal_len, final_body, final_body_len);
+    normal_len += final_body_len;
+
+    // bake wildcard
+    char wild_buf[ 1 + 8 + BUFS ];
+    wild_buf[ 0 ] = 'M';
+    memcpy(wild_buf + 1, &p, 8);
+    memcpy(wild_buf + 9, final_body, final_body_len);
+    int wild_len = 1 + 8 + final_body_len;
+
+    // send
     for (int j = 0; j < active; j++) {
-        if (j != si && p == clients[ j ].pass) {
-            onlytcp(writev(clients[ j ].fd, iov, 2));
-            onlyudp({
-                char out[ 64 + env_bufs ];
-                memcpy(out, pay_buf, pay_len);
-                memcpy(out + pay_len, body, body_len);
-                sendto(sock, out, pay_len + body_len, 0, (struct sockaddr*) &clients[ j ].addr, clients[ j ].addrlen);
-            });
-            sent++;
-        }
+        if (clients[ j ].id == sender_id) continue;
+        int is_wildcard = env_wildcard && clients[ j ].pass == wildcard_pass;
+        if (p != clients[ j ].pass && !is_wildcard) continue;
+
+        char* send_buf = is_wildcard ? wild_buf : normal_buf;
+        int   send_len = is_wildcard ? wild_len : normal_len;
+
+        onlytcp(write(clients[ j ].fd, send_buf, send_len));
+        onlyudp(sendto(sock, send_buf, send_len, 0, (struct sockaddr*) &clients[ j ].addr, clients[ j ].addrlen));
+        sent++;
     }
+
     if (sent > 0) log("sentto:%d\n", sent);
 }
 
-char *admin_msg;
+void connect_relay(const char* ip, const char* port) {
+    if (relay_count >= MAX_RELAYS) {
+        printf("relays full\n");
+        return;
+    }
+
+    // try tcp
+    struct addrinfo hint = {}, *result;
+    hint.ai_family       = AF_UNSPEC;
+    hint.ai_socktype     = SOCK_STREAM;
+    if (getaddrinfo(ip, port, &hint, &result) != 0) {
+        perror("relay: getaddrinfo failed\n");
+        return;
+    }
+
+    int  is_udp = 0;
+    int  fd     = socket(result->ai_family, result->ai_socktype, result->ai_protocol);
+    char gbuf[ 64 ];
+
+    if (fd >= 0 && connect(fd, result->ai_addr, result->ai_addrlen) == 0) {
+        // read greeting
+        int n = read(fd, gbuf, sizeof(gbuf) - 1);
+        if (n > 0 && gbuf[ 0 ] == 'I' && gbuf[ 1 ] == 'T') {
+            // confirmed tcp
+            char pbuf[ 9 ] = "P";
+            memcpy(pbuf + 1, &wildcard_pass, 8);
+            write(fd, pbuf, 9);
+            read(fd, gbuf, sizeof(gbuf)); // p_ok
+        } else {
+            // not a broadd tcp server
+            close(fd);
+            fd = -1;
+        }
+    } else {
+        if (fd >= 0) {
+            close(fd);
+            fd = -1;
+        }
+    }
+    freeaddrinfo(result);
+
+    // udp fallback
+    if (fd < 0) {
+        hint.ai_socktype = SOCK_DGRAM;
+        if (getaddrinfo(ip, port, &hint, &result) != 0) {
+            printf("relay: getaddrinfo failed\n");
+            return;
+        }
+        fd = socket(result->ai_family, result->ai_socktype, result->ai_protocol);
+        if (fd < 0) {
+            freeaddrinfo(result);
+            printf("relay: socket failed\n");
+            return;
+        }
+
+        // prevent hanging
+        struct timeval tv;
+        tv.tv_sec  = 2;
+        tv.tv_usec = 0;
+        setsockopt(fd, SOL_SOCKET, SO_RCVTIMEO, (const char*) &tv, sizeof tv);
+
+        sendto(fd, "N\n", 2, 0, result->ai_addr, result->ai_addrlen);
+
+        if (connect(fd, result->ai_addr, result->ai_addrlen) < 0) {
+            close(fd);
+            freeaddrinfo(result);
+            printf("relay: udp connect failed\n");
+            return;
+        }
+
+        struct sockaddr_storage src;
+        socklen_t               srclen = sizeof(src);
+
+        int rn = read(fd, gbuf, sizeof(gbuf) - 1);
+        if (rn <= 0) {
+            printf("relay: udp timeout\n");
+            close(fd);
+            freeaddrinfo(result);
+            return;
+        }
+        gbuf[ rn ]         = '\0';
+        int remote_timeout = (gbuf[ 0 ] == 'T') ? atoi(gbuf + 1) : 30;
+
+        rn = read(fd, gbuf, sizeof(gbuf) - 1);
+        if (rn <= 0 || gbuf[ 0 ] != 'I' || gbuf[ 1 ] != 'U') {
+            close(fd);
+            freeaddrinfo(result);
+            printf("relay: invalid handshake\n");
+            return;
+        }
+
+        char pbuf[ 9 ] = "P";
+        memcpy(pbuf + 1, &wildcard_pass, 8);
+        sendto(fd, pbuf, 9, 0, result->ai_addr, result->ai_addrlen);
+        is_udp = 1;
+
+        struct relay* r   = &relays[ relay_count++ ];
+        r->fd             = fd;
+        r->is_udp         = 1;
+        r->last_hb        = time(NULL);
+        r->remote_timeout = remote_timeout;
+        memcpy(&r->addr, result->ai_addr, result->ai_addrlen);
+        r->addrlen = result->ai_addrlen;
+        freeaddrinfo(result);
+
+        tv.tv_sec = 0;
+        setsockopt(fd, SOL_SOCKET, SO_RCVTIMEO, (const char*) &tv, sizeof tv);
+    } else {
+        struct relay* r   = &relays[ relay_count++ ];
+        r->fd             = fd;
+        r->is_udp         = 0;
+        r->last_hb        = time(NULL);
+        r->remote_timeout = 0;
+    }
+
+    fcntl(fd, F_SETFL, fcntl(fd, F_GETFL, 0) | O_NONBLOCK);
+    printf("relay: connected to %s:%s (%s)\n", ip, port, is_udp ? "udp" : "tcp");
+}
+
+char* admin_msg;
 
 int main(int argc, char** argv) {
     if (argc < 3) {
@@ -281,8 +517,6 @@ int main(int argc, char** argv) {
         exit(1);
     }
 
-    admin_msg = malloc(env_bufs);
-
     onlytcp(printf("hi from tcp;" verb(" verbose mode") "\n"));
     onlyudp(printf("hi from udp;" verb(" verbose mode") "\n"));
 
@@ -313,13 +547,20 @@ int main(int argc, char** argv) {
             else
                 printf("CLIENT_TIMEOUT: ignored (must be >0)\n");
         });
+        if ((e = getenv("WILDCARD"))) {
+            int v        = atoi(e);
+            env_wildcard = (v != 0);
+        }
     }
 
+    admin_msg = malloc(env_bufs);
+
     set_stdin_nonblocking();
 
     admin_msg[ 0 ] = 'A';
 
     memcpy(&default_pass, "00000000", 8);
+    memcpy(&wildcard_pass, "00000001", 8);
 
     struct addrinfo hint = {}, *result;
     hint.ai_family       = AF_UNSPEC;
@@ -345,7 +586,7 @@ int main(int argc, char** argv) {
     fcntl(sock, F_SETFL, fcntl(sock, F_GETFL, 0) | O_NONBLOCK);
 
     // fds[0] = server socket, fds[1] = stdin, fds[2..] = clients
-    struct pollfd fds[ env_max_clients + 2 ];
+    struct pollfd fds[ env_max_clients + 2 + MAX_RELAYS ];
     fds[ 1 ].fd     = STDIN_FILENO;
     fds[ 1 ].events = POLLIN;
 
@@ -363,7 +604,23 @@ int main(int argc, char** argv) {
             n++;
         });
 
-        poll(fds, n, onlyudp(env_client_timeout * 1000) onlytcp(-1));
+        for (int i = 0; i < relay_count; i++) {
+            fds[ n ].fd     = relays[ i ].fd;
+            fds[ n ].events = POLLIN;
+            n++;
+        }
+
+        int relay_base = n - relay_count;
+
+        int poll_timeout = onlyudp(env_client_timeout * 1000) onlytcp(-1);
+        // find min poll timeout
+        for (int i = 0; i < relay_count; i++) {
+            if (relays[ i ].is_udp && relays[ i ].remote_timeout > 0) {
+                int t = relays[ i ].remote_timeout * 500;
+                if (poll_timeout < 0 || t < poll_timeout) poll_timeout = t;
+            }
+        }
+        poll(fds, n, poll_timeout);
 
         time_t now = time(NULL);
 
@@ -377,12 +634,31 @@ int main(int argc, char** argv) {
 
         // check admin stdin
         if (fds[ 1 ].revents & POLLIN) {
-            resval = nb_getstring(admin_msg + 1, env_bufs - 1);
-            if (resval == NB_EOF) {
+            char admin_buf[ BUFS ];
+            resval = nb_getstring(admin_buf, sizeof(admin_buf));
+            if (resval == NB_EOF || (resval > 0 && admin_buf[ 0 ] == 'E')) {
                 printf("dying\n");
                 run_shutdown = 1;
             } else if (resval > 0) {
-                got_msg = 1;
+                if (admin_buf[ 0 ] == 'B') {
+                    // skip 'B' and any spaces
+                    char* msg = admin_buf + 1;
+                    while (*msg == ' ') msg++;
+                    int mlen = resval - (int) (msg - admin_buf);
+                    if (mlen > 0) {
+                        admin_msg[ 0 ] = 'A';
+                        memcpy(admin_msg + 1, msg, mlen);
+                        got_msg = 1;
+                        resval  = mlen;
+                    }
+                } else if (admin_buf[ 0 ] == 'C') {
+                    char ip[ 64 ], port[ 16 ];
+                    if (sscanf(admin_buf + 1, " %63s %15s", ip, port) == 2) connect_relay(ip, port);
+                    else
+                        printf("usage: C <ip> <port>\n");
+                } else {
+                    printf("unknown admin command\n");
+                }
             }
         }
 
@@ -411,7 +687,7 @@ int main(int argc, char** argv) {
             int                     j = -1);
 
         // process existing clients
-        onlytcp(for (int i = n - 1; i >= 2; i--)) //
+        onlytcp(for (int i = relay_base - 1; i >= 2; i--)) //
             onlyudp(
                 while (fds[ 0 ].revents & POLLIN && //
                        (udp_srclen = sizeof(udp_src))
@@ -447,15 +723,11 @@ int main(int argc, char** argv) {
                     clients[ j ].rl_bytes = 0;
                     active++;
                     // send timeout amount
-                    sendto(
-                        sock,
-                        "T" stringify(env_client_timeout) "\n",
-                        strlen(stringify(env_client_timeout)) + 2,
-                        0,
-                        (struct sockaddr*) &udp_src,
-                        udp_srclen);
+                    char tbuf[ 32 ];
+                    int  tlen = sprintf(tbuf, "T%d\n", env_client_timeout);
+                    sendto(sock, tbuf, tlen, 0, (struct sockaddr*) &udp_src, udp_srclen);
                     char ibuf[ 32 ];
-                    int  ilen = sprintf(ibuf, "I%" PRIu64 "\n", clients[ j ].id);
+                    int  ilen = sprintf(ibuf, "IU%" PRIu64 "\n", clients[ j ].id);
                     sendto(sock, ibuf, ilen, 0, (struct sockaddr*) &udp_src, udp_srclen);
 
                     log("clients:%d\n", active);
@@ -488,28 +760,15 @@ int main(int argc, char** argv) {
                         onlyudp(sendto(sock, "Rate\n", 5, 0, (struct sockaddr*) &udp_src, udp_srclen));
                     } else {
                         clients[ j ].rl_bytes += plen;
-                        broadcast_msg(j, buf + 1, plen, 0);
+                        broadcast_msg(clients[ j ].id, buf + 1, plen);
                         onlytcp(write(clients[ j ].fd, "m_ok\n", 5));
                     }
-
-                } else if (buf[ 0 ] == 'S' && rd >= 2) {
-                    // S: broadcast with default password
-                    if (likely(now - clients[ j ].last_rl >= env_rate_limit_s)) {
-                        clients[ j ].last_rl  = now;
-                        clients[ j ].rl_bytes = 0;
-                    }
-                    int plen = rd - 1;
-                    if (unlikely(clients[ j ].rl_bytes + plen > env_bufs)) {
-                        onlytcp(write(clients[ j ].fd, "Rate\n", 5));
-                        onlyudp(sendto(sock, "Rate\n", 5, 0, (struct sockaddr*) &udp_src, udp_srclen));
-                    } else {
-                        clients[ j ].rl_bytes += plen;
-                        broadcast_msg(j, buf + 1, plen, 1);
-                        onlytcp(write(clients[ j ].fd, "m_ok\n", 6));
-                    }
-
                 } else if (buf[ 0 ] == 'N') {
                     // no-op
+                } else if (buf[ 0 ] == 'W') {
+                    char str_to_send[] = { 'W', env_wildcard + '0', '\n' };
+                    onlytcp(write(clients[ j ].fd, str_to_send, 3));
+                    onlyudp(sendto(sock, str_to_send, 3, 0, (struct sockaddr*) &udp_src, udp_srclen));
                 } else {
                     onlytcp(write(clients[ j ].fd, "Err\n", 4));
                     onlyudp(sendto(sock, "Err\n", 4, 0, (struct sockaddr*) &udp_src, udp_srclen));
@@ -520,6 +779,37 @@ int main(int argc, char** argv) {
             onlytcp(if (unlikely(got_msg)) write(clients[ j ].fd, admin_msg, resval + 1));
         }
 
+        // process relay fds
+        for (int i = relay_count - 1; i >= 0; i--) {
+            struct relay*  r  = &relays[ i ];
+            struct pollfd* pf = &fds[ relay_base + i ];
+
+            // udp heartbeat
+            if (r->is_udp && r->remote_timeout > 0 && now - r->last_hb >= r->remote_timeout / 2) {
+                write(r->fd, "N\n", 2);
+                r->last_hb = now;
+            }
+
+            if (!(pf->revents & POLLIN)) continue;
+
+            char rbuf[ BUFS ];
+            int  rrd;
+
+            rrd = read(r->fd, rbuf, sizeof(rbuf));
+            if (rrd <= 0) {
+                printf("relay: disconnected\n");
+                close(r->fd);
+                if (i < relay_count - 1) relays[ i ] = relays[ relay_count - 1 ];
+                relay_count--;
+                continue;
+            }
+            rbuf[ rrd ] = 0;
+
+            // broadcast relay msg with id1
+            printf("got input: `%s`\n", rbuf);
+            if (rbuf[ 0 ] == 'M' && rrd >= 9) { broadcast_msg(1, rbuf + 1, rrd - 1); }
+        }
+
         onlyudp(if (unlikely(got_msg)) {
             for (int i = 0; i < active; i++)
                 sendto(sock, admin_msg, resval + 1, 0, (struct sockaddr*) &clients[ i ].addr, clients[ i ].addrlen);