ignore non valid utf8 messages published on websocket connections

include/ngx_http_push_stream_module_utils.h

@@ -309,4 +309,6 @@ ngx_http_push_stream_requested_channel_t *ngx_http_push_stream_parse_channels_id
 
 ngx_int_t                   ngx_http_push_stream_create_shmtx(ngx_shmtx_t *mtx, ngx_shmtx_sh_t *addr, u_char *name);
 
+ngx_flag_t                  ngx_http_push_stream_is_utf8(u_char *p, size_t n);
+
 #endif /* NGX_HTTP_PUSH_STREAM_MODULE_UTILS_H_ */

misc/spec/subscriber/websocket_spec.rb

@@ -635,4 +635,43 @@ describe "Subscriber WebSocket" do
       end
     end
   end
+
+  it "should accept non latin characters" do
+    channel = 'ch_test_publish_non_latin'
+
+    nginx_run_server(config) do |conf|
+      EventMachine.run do
+        ws = WebSocket::EventMachine::Client.connect(:uri => "ws://#{nginx_host}:#{nginx_port}/ws/#{channel}")
+        ws.onmessage do |text, type|
+          expect(text).to eq("\xD8\xA3\xD9\x8E\xD8\xA8\xD9\x92\xD8\xAC\xD9\x8E\xD8\xAF\xD9\x90\xD9\x8A\xD9\x8E\xD9\x91\xD8\xA9 \xD8\xB9\xD9\x8E")
+          EventMachine.stop
+        end
+
+        EM.add_timer(1) do
+          ws.send "\xD8\xA3\xD9\x8E\xD8\xA8\xD9\x92\xD8\xAC\xD9\x8E\xD8\xAF\xD9\x90\xD9\x8A\xD9\x8E\xD9\x91\xD8\xA9 \xD8\xB9\xD9\x8E"
+        end
+      end
+    end
+  end
+
+  it "should reject an invalid utf8 sequence" do
+    channel = 'ch_test_publish_invalid_utf8'
+
+    nginx_run_server(config) do |conf|
+      EventMachine.run do
+        ws = WebSocket::EventMachine::Client.connect(:uri => "ws://#{nginx_host}:#{nginx_port}/ws/#{channel}")
+        ws.onmessage do |text, type|
+          fail("Should not have received the '#{text.force_encoding('UTF-8')}'")
+        end
+
+        ws.onclose do
+          EventMachine.stop
+        end
+
+        EM.add_timer(1) do
+          ws.send "\xA3\xD9\x8E\xD8\xA8\xD9\x92\xD8\xAC\xD9\x8E\xD8\xAF\xD9\x90\xD9\x8A\xD9\x8E\xD9\x91\xD8\xA9 \xD8\xB9\xD9\x8E"
+        end
+      end
+    end
+  end
 end

src/ngx_http_push_stream_module_utils.c

@@ -2311,3 +2311,30 @@ ngx_http_push_stream_create_shmtx(ngx_shmtx_t *mtx, ngx_shmtx_sh_t *addr, u_char
 
     return NGX_OK;
 }
+
+
+ngx_flag_t
+ngx_http_push_stream_is_utf8(u_char *p, size_t n)
+{
+    u_char  c, *last;
+    size_t  len;
+
+    last = p + n;
+
+    for (len = 0; p < last; len++) {
+
+        c = *p;
+
+        if (c < 0x80) {
+            p++;
+            continue;
+        }
+
+        if (ngx_utf8_decode(&p, n) > 0x10ffff) {
+            /* invalid UTF-8 */
+            return 0;
+        }
+    }
+
+    return 1;
+}

src/ngx_http_push_stream_module_websocket.c

@@ -287,6 +287,10 @@ ngx_http_push_stream_websocket_reading(ngx_http_request_t *r)
                             }
                         }
 
+                        if (!ngx_http_push_stream_is_utf8(ctx->frame->payload, ctx->frame->payload_len)) {
+                            goto finalize;
+                        }
+
                         for (q = ngx_queue_head(&ctx->subscriber->subscriptions); q != ngx_queue_sentinel(&ctx->subscriber->subscriptions); q = ngx_queue_next(q)) {
                             ngx_http_push_stream_subscription_t *subscription = ngx_queue_data(q, ngx_http_push_stream_subscription_t, queue);
                             if (subscription->channel->for_events) {