Add 'VPN' capability for playing with friends feature

src/emsocket/Link.h

@@ -32,6 +32,9 @@ public:
     using Receiver = std::function<void(const void *, size_t)>;
 
     virtual ~Link() { }
+
+    virtual void connect(const SocketAddr &addr) = 0;
+    virtual void sendto(const void *data, size_t len, const SocketAddr &addr) = 0;
     virtual void send(const void *data, size_t len) = 0;
 };
 

src/emsocket/ProxyLink.cpp

@@ -45,32 +45,51 @@ extern "C" {
     EMSCRIPTEN_KEEPALIVE
     void proxylink_onclose(void *thisPtr);
     EMSCRIPTEN_KEEPALIVE
-    void proxylink_onmessage(void *thisPtr, const void *buf, size_t n);
+    void proxylink_onmessage(void *thisPtr, const void *buf, size_t n, const char *ip, uint16_t port);
 }
 
-EM_JS(int, proxylink_new, (const char* ip, uint16_t port, bool udp, void *thisPtr), {
+EM_JS(int, proxylink_new, (uint16_t bind_port, bool udp, void *thisPtr), {
     if (!self.hasOwnProperty('w_proxylink_onopen')) {
         self.w_proxylink_onopen = Module.cwrap('proxylink_onopen', null, ['number']);
         self.w_proxylink_onerror = Module.cwrap('proxylink_onerror', null, ['number']);
         self.w_proxylink_onclose = Module.cwrap('proxylink_onclose', null, ['number']);
-        self.w_proxylink_onmessage = Module.cwrap('proxylink_onmessage', null, ['number', 'number', 'number']);
+        self.w_proxylink_onmessage = Module.cwrap('proxylink_onmessage', null, ['number', 'number', 'number', 'number', 'number']);
     }
 
-    const link = new ProxyLink(UTF8ToString(ip), port, udp);
+    const link = new ProxyLink(bind_port, udp);
     link.onopen = () => { w_proxylink_onopen(thisPtr); };
     link.onerror = () => { w_proxylink_onerror(thisPtr); };
     link.onclose = () => { w_proxylink_onclose(thisPtr); };
-    link.onmessage = (data) => {
+    link.onmessage = (data, ip, port) => {
         var len = data.byteLength;
-        // TODO: Get rid of this allocation
+        // TODO: Get rid of these allocations
         var buf = _malloc(len);
         HEAPU8.set(new Uint8Array(data), buf);
-        w_proxylink_onmessage(thisPtr, buf, len);
+        var ip_length = lengthBytesUTF8(ip) + 1;
+        var ip_buf = _malloc(ip_length);
+        stringToUTF8(ip, ip_buf, ip_length);
+        w_proxylink_onmessage(thisPtr, buf, len, ip_buf, port);
         _free(buf);
+        _free(ip_buf);
     };
     return link.index;
 });
 
+EM_JS(void, proxylink_connect, (int index, const char* ip, uint16_t port), {
+    const link = ProxyLink.get(index);
+    if (link) {
+        link.connect(UTF8ToString(ip), port);
+    }
+});
+
+EM_JS(void, proxylink_sendto, (int index, const void *data, int len, const char *dest_ip, uint16_t dest_port), {
+    const link = ProxyLink.get(index);
+    if (link) {
+        link.sendto(HEAPU8.subarray(data, data + len), UTF8ToString(dest_ip), dest_port);
+    }
+});
+
+
 EM_JS(void, proxylink_send, (int index, const void *data, int len), {
     const link = ProxyLink.get(index);
     if (link) {
@@ -96,14 +115,14 @@ public:
     ProxyLink(const ProxyLink &) = delete;
     ProxyLink& operator=(const ProxyLink &) = delete;
 
-    ProxyLink(VirtualSocket *vs_, const SocketAddr &addr_, bool udp_)
+    ProxyLink(VirtualSocket *vs_, uint16_t bind_port_, bool udp_)
         : vs(vs_),
-          addr(addr_),
+          bind_port(bind_port_),
           udp(udp_),
           wsIndex(-1)
     {
         emsocket_run_on_io_thread(true, [this]() {
-            wsIndex = proxylink_new(addr.getIP().c_str(), addr.getPort(), udp, this);
+            wsIndex = proxylink_new(bind_port, udp, this);
         });
         assert(wsIndex > 0);
     }
@@ -114,9 +133,29 @@ public:
         });
     }
 
+    // Called from external thread
+    virtual void connect(const SocketAddr &addr) {
+        // Move to I/O thread.
+        int wsIndex_ = wsIndex;
+        emsocket_run_on_io_thread(false, [wsIndex_, addr]() {
+            proxylink_connect(wsIndex_, addr.getIP().c_str(), addr.getPort());
+        });
+    }
+
+    // Called from external thread
+    virtual void sendto(const void *data, size_t len, const SocketAddr &addr) {
+        // Move to I/O thread.
+        int wsIndex_ = wsIndex;
+        Packet *pkt = new Packet(SocketAddr(), data, len);
+        emsocket_run_on_io_thread(false, [wsIndex_, pkt, addr]() {
+            proxylink_sendto(wsIndex_, &pkt->data[0], pkt->data.size(), addr.getIP().c_str(), addr.getPort());
+            delete pkt;
+        });
+    }
+
     // Called from external thread
     virtual void send(const void *data, size_t len) {
-        // Called from an external thread. Move it to the I/O thread.
+        // Move to I/O thread.
         int wsIndex_ = wsIndex;
         Packet *pkt = new Packet(SocketAddr(), data, len);
         emsocket_run_on_io_thread(false, [wsIndex_, pkt]() {
@@ -143,7 +182,8 @@ public:
     }
 
     // Called from I/O thread
-    void onmessage(const void *buf, int n) {
+    void onmessage(const void *buf, int n, const char *ip, uint16_t port) {
+        SocketAddr addr(ip, port);
         vs->linkReceived(addr, buf, n);
     }
 
@@ -154,13 +194,13 @@ public:
     }
 private:
     VirtualSocket *vs;
-    SocketAddr addr;
+    uint16_t bind_port;
     bool udp;
     int wsIndex;
 };
 
-Link* make_proxy_link(VirtualSocket* vs, const SocketAddr &addr, bool udp) {
+Link* make_proxy_link(VirtualSocket* vs, uint16_t bindport, bool udp) {
-    return new ProxyLink(vs, addr, udp);
+    return new ProxyLink(vs, bindport, udp);
 }
 
 } // namespace
@@ -183,6 +223,6 @@ void proxylink_onclose(void *thisPtr) {
 }
 
 EMSCRIPTEN_KEEPALIVE
-void proxylink_onmessage(void *thisPtr, const void *buf, size_t n) {
+void proxylink_onmessage(void *thisPtr, const void *buf, size_t n, const char *ip, uint16_t port) {
-    return reinterpret_cast<ProxyLink*>(thisPtr)->onmessage(buf, n);
+    return reinterpret_cast<ProxyLink*>(thisPtr)->onmessage(buf, n, ip, port);
 }

src/emsocket/ProxyLink.h

@@ -30,6 +30,6 @@ namespace emsocket {
 
 class VirtualSocket;
 
-Link* make_proxy_link(VirtualSocket* vs, const SocketAddr &addr, bool udp);
+Link* make_proxy_link(VirtualSocket* vs, uint16_t bindport, bool udp);
 
 } // namespace

src/emsocket/VirtualSocket.cpp

@@ -147,6 +147,7 @@ bool VirtualSocket::bind(const SocketAddr& addr) {
         bindAddr = addr;
         bindAddr.setPort(port);
     }
+    link = make_proxy_link(this, port, is_udp);
     return true;
 }
 
@@ -185,14 +186,13 @@ bool VirtualSocket::startConnect(const SocketAddr &dest) {
         return false;
     } else {
         assert(!is_udp);
-        assert(!link);
         assert(!is_connected);
         assert(!is_shutdown);
         if (!isBound()) {
             bind(SocketAddr()); // bind to random port
         }
         remoteAddr = dest;
-        link = make_proxy_link(this, dest, is_udp);
+        link->connect(dest);
         return true;
     }
     std::cerr << "emsocket no proxy set" << std::endl;
@@ -249,8 +249,8 @@ ssize_t VirtualSocket::recvfrom(void *buf, size_t n, SocketAddr *from) {
 void VirtualSocket::sendto(const void *buf, size_t n, const SocketAddr& to) {
     assert(is_udp);
     assert(isBound());
-    SocketAddr sourceAddr("127.0.0.1", bindAddr.getPort());
     if (to.isLocalHost()) {
+        SocketAddr sourceAddr("127.0.0.1", bindAddr.getPort());
         bool sent = false;
         {
             VSLOCK();
@@ -266,16 +266,7 @@ void VirtualSocket::sendto(const void *buf, size_t n, const SocketAddr& to) {
         }
         return;
     }
-    if (link) {
+    link->sendto(buf, n, to);
-        if (remoteAddr != to) {
-            std::cerr << "emsocket: Reuse of socket for multiple destinations not supported" << std::endl;
-            return;
-        }
-    } else {
-        remoteAddr = to;
-        link = make_proxy_link(this, to, is_udp);
-    }
-    link->send(buf, n);
 }
 
 } // namespace

src/emsocket/emsocketctl.cpp

@@ -46,6 +46,8 @@ using namespace emsocket;
 
 static void *io_thread_main(void *);
 
+EMSCRIPTEN_KEEPALIVE
+extern "C"
 void emsocket_init(void) {
     if (didInit) return;
     didInit = true;
@@ -62,6 +64,8 @@ EM_JS(void, _set_proxy, (const char *url), {
     setProxy(UTF8ToString(url));
 });
 
+EMSCRIPTEN_KEEPALIVE
+extern "C"
 void emsocket_set_proxy(const char *url) {
     char *urlcopy = strdup(url);
     emsocket_run_on_io_thread(false, [urlcopy]() {
@@ -70,6 +74,19 @@ void emsocket_set_proxy(const char *url) {
     });
 }
 
+EM_JS(void, _set_vpn, (const char *vpn), {
+    setVPN(UTF8ToString(vpn));
+});
+
+EMSCRIPTEN_KEEPALIVE
+extern "C"
+void emsocket_set_vpn(const char *vpn) {
+    char *vpncopy = strdup(vpn);
+    emsocket_run_on_io_thread(false, [vpncopy]() {
+        _set_vpn(vpncopy);
+        free(vpncopy);
+    });
+}
 
 static void io_thread_reenter(void) {
     std::vector<std::function<void()> > callbacks;

src/emsocket/emsocketctl.h

@@ -28,7 +28,8 @@ extern "C" {
 #endif
 
 void emsocket_init(void);
-void emsocket_set_proxy(const char *proxyUrl);
+void emsocket_set_proxy(const char *url);
+void emsocket_set_vpn(const char *vpn);
 
 #ifdef __cplusplus
 }

src/emsocket/proxy.js

@@ -1,26 +1,114 @@
+// This code runs on a WebWorker wrapped inside a function body.
+// To export a global symbol, assigned it through 'self'.
 self.proxyUrl = "";
 function setProxy(url) {
     self.proxyUrl = url;
 }
 self.setProxy = setProxy;
 
-self.textEncoder = new TextEncoder();
+self.vpnCode = "";
-self.textDecoder = new TextDecoder();
+function setVPN(code) {
+  self.vpnCode = code;
+}
+self.setVPN = setVPN;
+
+function inet_ntop(n) {
+    const a = (n >> 24) & 0xFF;
+    const b = (n >> 16) & 0xFF;
+    const c = (n >>  8) & 0xFF;
+    const d = (n >>  0) & 0xFF;
+    return `${a}.${b}.${c}.${d}`;
+}
+self.inet_ntop = inet_ntop;
+
+function inet_pton(ip) {
+    const ret = new ArrayBuffer(4);
+    const v = new DataView(ret);
+    var [a, b, c, d] = ip.split('.');
+    v.setUint8(0, parseInt(a));
+    v.setUint8(1, parseInt(b));
+    v.setUint8(2, parseInt(c));
+    v.setUint8(3, parseInt(d));
+    return ret;
+}
+self.inet_pton = inet_pton;
+
+self.EP_MAGIC = 0x778B4CF3;
+
+function unencapsulate(data) {
+    // Data is encapsulated with a 12 byte header.
+    // Magic      - 4 bytes EP_MAGIC
+    // Dest IP    - 4 bytes 0xAABBCCDD for AA.BB.CC.DD
+    // Dest Port  - 2 bytes
+    // Packet Len - 2 bytes
+    if (!(data instanceof ArrayBuffer)) {
+        throw new Error("Received text over encapsulated channel");
+    }
+    if (data.byteLength < 12) {
+        throw new Error("Encapsulated header not present (short message)");
+    }
+    const view = new DataView(data);
+    const magic = view.getUint32(0);
+    if (magic != EP_MAGIC) {
+        throw new Error("Encapsulated packet header corrupted");
+    }
+    const src_ip = inet_ntop(view.getUint32(4));
+    const src_port = view.getUint16(8);
+    const pktlen = view.getUint16(10);
+    if (data.byteLength != 12 + pktlen) {
+        throw new Error("Invalid encapsulated packet length");
+    }
+    return [src_ip, src_port, data.slice(12)];
+}
+self.unencapsulate = unencapsulate;
+
+function encapsulate(dest_ip, dest_port, data) {
+    const edata = new ArrayBuffer(12 + data.byteLength);
+    const view = new DataView(edata);
+    view.setUint32(0, EP_MAGIC);
+    (new Uint8Array(edata, 4, 4)).set(new Uint8Array(inet_pton(dest_ip)));
+    view.setUint16(8, dest_port);
+    view.setUint16(10, data.byteLength);
+    (new Uint8Array(edata, 12)).set(data);
+    return edata;
+}
+self.encapsulate = encapsulate;
 
 class ProxyLink {
-    constructor(ip, port, udp) {
+    constructor(bind_port, udp) {
-        this.ip = ip;
+        this.bind_port = bind_port;
-        this.port = port;
         this.udp = udp;
         this.onopen = null;
         this.onerror = null;
         this.onclose = null;
         this.onmessage = null;
-        this.sentProxyRequest = false;
+        this.expectHandshake = null;
         this.userEnabled = false;
         this.userBuffer = [];
         this.index = ProxyLink.links.length;
         ProxyLink.links.push(this);
+        this.ws = null;
+        this.activated = false;
+        this.dead = false;
+        this.encapsulated = false;
+        this.receive_info = null;
+        if (this.udp && vpnCode) {
+            this.encapsulated = true;
+            this._activate();
+        }
+    }
+
+    connect(ip, port) {
+        if (this.udp)
+            throw new Error('ProxyLink: connect() called on udp socket');
+        this.connect_info = [ip, port];
+        this._activate();
+    }
+
+    _activate() {
+        if (this.activated)
+            throw new Error('ProxyLink activated twice');
+        this.activated = true;
         const ws = new WebSocket(proxyUrl);
         this.ws = ws;
         ws.binaryType = "arraybuffer";
@@ -31,10 +119,17 @@ class ProxyLink {
     }
 
     handleOpen() {
+        var req;
         // Send proxy request
-        const req = `PROXY IPV4 ${this.udp ? "UDP" : "TCP"} ${this.ip} ${this.port}`;
+        if (this.encapsulated) {
-        this.ws.send(textEncoder.encode(req));
+            req = `VPN ${vpnCode} BIND IPV4 UDP ${this.bind_port}`;
-        this.sentProxyRequest = true;
+            this.expectHandshake = 'BIND OK';
+        } else {
+            const [ip, port] = this.connect_info;
+            req = `PROXY IPV4 ${this.udp ? "UDP" : "TCP"} ${ip} ${port}`;
+            this.expectHandshake = 'PROXY OK';
+	}
+        this.ws.send(req);
     }
 
     handleError() {
@@ -48,40 +143,75 @@ class ProxyLink {
     }
 
     handleMessage(e) {
+        try {
+            this._handleMessage(e);
+        } catch (err) {
+            console.log("ProxyLink javascript exception");
+            console.log(err);
+            this._close();
+        }
+    }
+
+    _handleMessage(e) {
+        const data = e.data;
         if (!this.userEnabled) {
             // Waiting for proxy auth message
-            if (!this.sentProxyRequest) {
+            if (!this.expectHandshake) {
-                console.log("Got invalid message before proxy request");
+                throw new Error("Invalid message before proxy request");
-                this._close();
-                return;
             }
-            const ok = textDecoder.decode(e.data);
+            if (data != this.expectHandshake) {
-            if (ok != "PROXY OK") {
+                throw new Error("Invalid handshake response");
-                console.log("Got invalid proxy message");
-                this._close();
-                return;
             }
-            //console.log("Got proxy OK");
             this._enable();
             this.onopen();
             return;
         }
-        //console.log("Relaying message");
+        if (this.encapsulated) {
-        this.onmessage(e.data);
+            const [src_ip, src_port, rdata] = unencapsulate(data);
+            this.onmessage(rdata, src_ip, src_port);
+        } else {
+            const [src_ip, src_port] = this.connect_info;
+            this.onmessage(data, src_ip, src_port);
+        }
+    }
+
+    sendto(data, ip, port) {
+        if (this.dead) return;
+        if (!this.activated) {
+            this.connect_info = [ip, port];
+            this._activate();
+        }
+        if (this.encapsulated) {
+            const edata = encapsulate(ip, port, data);
+            this._send(edata);
+        } else {
+            if (this.connect_info[0] !== ip || this.connect_info[1] !== port) {
+                throw new Error('ProxyLink: Address mismatch on non-encapsulated link');
+            }
+            this._send(data);
+        }
     }
 
     send(data) {
-        //console.log("Got send");
+        if (this.dead) return;
-        const ws = this.ws;
+        if (!this.activated) {
-        if (!ws) return;
+            throw new Error('ProxyLink: send before connect');
-        // If this copy isn't done, send fails with:
+        }
-        //   "The provided ArrayBufferView value must not be shared."
+        if (this.encapsulated) {
-        data = new Uint8Array(data);
+            throw new Error('ProxyLink: Encapsulated send not supported');
+        }
+        this._send(data);
+    }
+
+    _send(data) {
+        if (typeof data !== 'string') {
+            // If this copy isn't done, send fails with:
+            //   "The provided ArrayBufferView value must not be shared."
+            data = new Uint8Array(data);
+        }
         if (this.userEnabled) {
-            //console.log("Sending direct");
+            this.ws.send(data);
-            ws.send(data);
         } else {
-            //console.log("Sending to userBuffer");
             this.userBuffer.push(data);
         }
     }
@@ -89,14 +219,15 @@ class ProxyLink {
     _enable() {
         this.userEnabled = true;
         for (const data of this.userBuffer) {
-            //console.log("Flushing from buffer");
             this.ws.send(data);
         }
         this.userBuffer = null;
     }
 
-    // Call this internally. It will dispatch callbacks.
+    // Call this internally to dispatch the onclose callback but leave
+    // this link on the list.
     _close() {
+        this.dead = true;
         const ws = this.ws;
         if (ws) {
             ws.onopen = ws.onerror = ws.onclose = ws.onmessage = null;
@@ -109,7 +240,7 @@ class ProxyLink {
     }
 
     // This should only be called externally, because it does not
-    // invoke callbacks, and removes the index from the list.
+    // invoke onclose(), and immediately removes this link from the list.
     close() {
         this.onopen = null;
         this.onerror = null;