Starters for ten

Since I was enjoying the C examples so much, initially I wanted to write the whole thing in C, but I found the APIs presented by the handful of C libraries around WebSockets either unfamiliar or complicated. Instead I opted for Rust to do the actual “plugging” together of ØMQ sockets with WebSocket.

There are (naturally) ØMQ bindings for Rust and there is approximately one WebSocket library for the language (the latter makes choosing a lib significantly easier) which has a nice readable event loop. Since, in order for users-with-browsers to be able to access it, the WebSocket server must sit on the boundary of our hypothetical ØMQ network. That means it presents an obvious attack vector and Rust's promises around memory safety make it an attratice choice for internet-exposed interfaces. We will use Nginx to “add” a TLS layer to incoming requests and handle half of the HTTPS handshake – I’m assuming Nginx’s implementation is reasonably secure.

Deep breaths

Let’s take a step back and sketch out what we are trying to do; we have a browser that will send a wss:// request to an Nginx server that will upgrade, shake hands with and proxy_pass this request to a WebSocket server. Once the WS connection is established, the WS server will hook up a ØMQ socket and be responsible for proxying messages that appear on the socket through to the browser, via the connected WebSocket. There follows a diagram of our proposed infra.

Reading, as is tradition, from left to right, let's examine the config of each component.

Target and proxy

On the part of the browser we don’t need to do anything unusual. Let’s imagine some client app is already running there, all we need to do is open a secure WebSocket connection with wss:// and remember to pass the string argument rust-websocket to specifiy the protocol we shall use (to please our Rusty WS server). We write

var ws = new WebSocket('wss://url', 'rust-websocket');

Ref: src/client.html

Nginx’s job is to accept HTTPS pre-flight requests, upgrade them and proxy them on to the WebSocket server. The prescient parts of the config are

server {
    listen <port> ssl;  # only accept connections over TLS
    index index.html;
    location /<path> {
        proxy_http_version 1.1;
        proxy_set_header Upgrade $http_upgrade;
        proxy_set_header Connection "Upgrade";
        proxy_set_header Sec-WebSocket-Key $http_sec_websocket_key;
        proxy_pass http://<hostname>;
    }
}

actually

MITM

Now for the centrepiece of our infra, the actual bridge between ØMQ and WebSocket. Because I envisioned the browser client as a passive “listener”, I have only implemented this bridge in the server to client direction (it would be easy to implemented the other “side” of the bridge). The source code is wrapped in a Cargo package (and a Dockerfile for libzmq at system level) to manage dependencies.

I based by work on the server example given by rust-websocket project, but replaced echo loop with a polling loop to recieve messages on a ØMQ SUB socket and re-publish them on the connected WebSocket.

diff --git a/src/zmq-wss/src/main.rs b/src/zmq-wss/src/example.rs
index ed50182..0cb8065 100644
--- a/src/zmq-wss/src/main.rs
+++ b/src/zmq-wss/src/example.rs
@@ -1,14 +1,13 @@
 extern crate websocket;
+extern crate zmq;

 use std::thread;
 use websocket::sync::Server;
 use websocket::OwnedMessage;

 fn main() {
+  let ws_server = Server::bind("0.0.0.0:80").unwrap();
-  let server = Server::bind("127.0.0.1:2794").unwrap();

+  for request in ws_server.filter_map(Result::ok) {
-  for request in server.filter_map(Result::ok) {
     // Spawn a new thread for each connection.
     thread::spawn(move || {
       if !request.protocols().contains(&"rust-websocket".to_string()) {
@@ -16,29 +15,34 @@ fn main() {
         return;
       }

+      let client = request.use_protocol("rust-websocket").accept().unwrap();
-      let mut client = request.use_protocol("rust-websocket").accept().unwrap();
-
       let ip = client.peer_addr().unwrap();
-
       println!("Connection from {}", ip);

+      let (_, mut sender) = client.split().unwrap();
-      let message = OwnedMessage::Text("Hello".to_string());
-      client.send_message(&message).unwrap();
-
-      let (mut receiver, mut sender) = client.split().unwrap();

+      let zmq_ctx = zmq::Context::new();
+      let subscriber = zmq_ctx.socket(zmq::SUB).unwrap();
+      assert!(subscriber.connect("tcp://zmq-pub:5556").is_ok());
+      assert!(subscriber.set_subscribe(b"").is_ok());
-      for message in receiver.incoming_messages() {
-        let message = message.unwrap();

+      let mut zmq_msg = zmq::Message::new().unwrap();
+      loop {
+        if subscriber.recv(&mut zmq_msg, zmq::DONTWAIT).is_err() {
+          continue;
-        match message {
-          OwnedMessage::Close(_) => {
-            let message = OwnedMessage::Close(None);
-            sender.send_message(&message).unwrap();
-            println!("Client {} disconnected", ip);
-            return;
-          }
-          OwnedMessage::Ping(ping) => {
-            let message = OwnedMessage::Pong(ping);
-            sender.send_message(&message).unwrap();
-          }
-          _ => sender.send_message(&message).unwrap(),
         }
+        let ws_message = OwnedMessage::Text(
+          zmq_msg.as_str().unwrap().to_string()
+        );
+        sender.send_message(&ws_message).unwrap();
       }
     });
   }
 }

Let's look at the patch more closely ...

let zmq_ctx = zmq::Context::new();
let subscriber = zmq_ctx.socket(zmq::SUB).unwrap();
assert!(subscriber.connect("tcp://zmq-pub:5556").is_ok());

is the standard ØMQ set up,

assert!(subscriber.set_subscribe(b"").is_ok());

subscribes to all messages (the b"" means “no filter”),

let mut zmq_msg = zmq::Message::new().unwrap();

creates a message container for reuse inside the loop,

loop {
  if subscriber.recv(&mut zmq_msg, zmq::DONTWAIT).is_err() {
    continue;
  }
  // ...

nonblocking message receive (we could do other work this loop), populates the message container or returns an error,

  // ...
  let ws_message = OwnedMessage::Text(
    zmq_msg.as_str().unwrap().to_string()
  );
  sender.send_message(&ws_message).unwrap();
}

we treat all messages as strings, but WebSocket supports binary messages as well, so the same message format possibilities that exist for ØMQ are also available here, send the ØMQ message to the browser over WebSocket as a text frame.

And that’s it. All source (mostly a Docker Compose file) is available on GitLab, there follows a diagram of what we just built.