Release v0.7.0 (#51)

* Add note to draft a GitHub release too ocne the release is tagged

* consistify newlines

* First pass to create a working hyper+soketto example

* rustfmt

* make trimming a little nicer

* Tidy up some hyper_server example a little

* cargo fmt

* Add comment on how to try hyper_server example

* print local_addr (no real difference but more correct)

* Fix typo

Co-authored-by: Andrew Plaza <[email protected]>

* Wee grammar tweak in comment

* Empty slice is more readable; no need to preserve lifetime

* allow setting custom headers

* Add a feature and update hyper_example to make it easier to use Soketto with http types

* Note about headers, re-export httparse::Header

* rethink 1 (but it turns out this won't work..)

* Make a new Server interface instead so that we have an API we can use

* remove the default_feature

* dedupe code in handshake and undo changes in handshake::server to minimise diff

* Add link to RFC explaining Sec-Websocket-Key

* add basic logger to example so we can get log output for debugging things

* Fix up a couple of comments

* use log since we added a logger, and comment tweaks

* use WebSocketKey alias

* return 28 byte base64 encoding instead of passing buffer in

* Tweak the docs a little

* fix a typo

* Release v0.7.0

Co-authored-by: Andrew Plaza <[email protected]>
Co-authored-by: Tomas Langsetmo <[email protected]>
Co-authored-by: David <[email protected]>

.github/workflows/ci.yml

@@ -2,7 +2,7 @@ name: Rust
 
 on:
   push:
-    # Run jobs when commits are pushed to 
+    # Run jobs when commits are pushed to
     # develop or release-like branches:
     branches:
       - develop
@@ -40,7 +40,7 @@ jobs:
         uses: actions-rs/[email protected]
         with:
           command: check
-          args: --all-targets
+          args: --all-targets --all-features
 
   fmt:
     name: Run rustfmt

CHANGELOG.md

@@ -4,6 +4,11 @@ The format is based on [Keep a Changelog].
 
 [Keep a Changelog]: http://keepachangelog.com/en/1.0.0/
 
+## 0.7.0
+
+- [added] Added the `handshake::http` module and example usage at `examples/hyper_server.rs` to make using Soketto in conjunction with libraries that use the `http` types (like Hyper) simpler [#45](https://github.com/paritytech/soketto/pull/45) [#48](https://github.com/paritytech/soketto/pull/48)
+- [added] Allow setting custom headers on the client to be sent to WebSocket servers when the opening handshake is performed [#47](https://github.com/paritytech/soketto/pull/47)
+
 ## 0.6.0
 
 - [changed] Expose the `Origin` headers from the client handshake on `ClientRequest` [#35](https://github.com/paritytech/soketto/pull/35)

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "soketto"
-version = "0.6.0"
+version = "0.7.0"
 authors = ["Parity Technologies <[email protected]>", "Jason Ozias <[email protected]>"]
 description = "A websocket protocol implementation."
 keywords = ["websocket", "codec", "async", "futures"]
@@ -26,9 +26,17 @@ httparse = { default-features = false, features = ["std"], version = "1.3.4" }
 log = { default-features = false, version = "0.4.8" }
 rand = { default-features = false, features = ["std", "std_rng"], version = "0.8" }
 sha-1 = { default-features = false, version = "0.9" }
+http = { default-features = false, version = "0.2", optional = true }
 
 [dev-dependencies]
 quickcheck = "0.9"
 tokio = { version = "1", features = ["full"] }
 tokio-util = { version = "0.6", features = ["compat"] }
 tokio-stream = { version = "0.1", features = ["net"] }
+hyper = { version = "0.14.10", features = ["full"] }
+env_logger = "0.9.0"
+
+[[example]]
+name = "hyper_server"
+required-features = ["http"]
+

RELEASING.md

@@ -11,15 +11,15 @@ These steps assume that you've checked out the Soketto repository and are in the
 3.  Check that you're happy with the current documentation.
 
     ```
-    cargo doc --open
+    cargo doc --open --all-features
     ```
 
     CI checks for broken internal links at the moment. Optionally you can also confirm that any external links
     are still valid like so:
 
     ```
     cargo install cargo-deadlinks
-    cargo deadlinks --check-http
+    cargo deadlinks --check-http -- --all-features
     ```
 
     If there are minor issues with the documentation, they can be fixed in the release branch.
@@ -65,5 +65,9 @@ These steps assume that you've checked out the Soketto repository and are in the
     git push --tags
     ```
 
+    Once this is pushed, go along to [the releases page on GitHub](https://github.com/paritytech/soketto/releases)
+    and draft a new release which points to the tag you just pushed to `master` above. Copy the changelog comments
+    for the current release into the release description.
+
 10. Merge the `master` branch back to develop so that we keep track of any changes that we made on
     the release branch.

examples/hyper_server.rs

@@ -0,0 +1,128 @@
+// Copyright (c) 2021 Parity Technologies (UK) Ltd.
+//
+// Licensed under the Apache License, Version 2.0
+// <LICENSE-APACHE or http://www.apache.org/licenses/LICENSE-2.0> or the MIT
+// license <LICENSE-MIT or http://opensource.org/licenses/MIT>, at your
+// option. All files in the project carrying such notice may not be copied,
+// modified, or distributed except according to those terms.
+
+// An example of how to use of Soketto alongside Hyper, so that we can handle
+// standard HTTP traffic with Hyper, and WebSocket connections with Soketto, on
+// the same port.
+//
+// To try this, start up the example (`cargo run --example hyper_server`) and then
+// navigate to localhost:3000 and, in the browser JS console, run:
+//
+// ```
+// var socket = new WebSocket("ws://localhost:3000");
+// socket.onmessage = function(msg) { console.log(msg) };
+// socket.send("Hello!");
+// ```
+//
+// You'll see any messages you send echoed back.
+
+use futures::io::{BufReader, BufWriter};
+use hyper::{Body, Request, Response};
+use soketto::{
+	handshake::http::{is_upgrade_request, Server},
+	BoxedError,
+};
+use tokio_util::compat::TokioAsyncReadCompatExt;
+
+/// Start up a hyper server.
+#[tokio::main]
+async fn main() -> Result<(), BoxedError> {
+	env_logger::init();
+
+	let addr = ([127, 0, 0, 1], 3000).into();
+
+	let service =
+		hyper::service::make_service_fn(|_| async { Ok::<_, hyper::Error>(hyper::service::service_fn(handler)) });
+	let server = hyper::Server::bind(&addr).serve(service);
+
+	println!("Listening on http://{} — connect and I'll echo back anything you send!", server.local_addr());
+	server.await?;
+
+	Ok(())
+}
+
+/// Handle incoming HTTP Requests.
+async fn handler(req: Request<Body>) -> Result<hyper::Response<Body>, BoxedError> {
+	if is_upgrade_request(&req) {
+		// Create a new handshake server.
+		let mut server = Server::new();
+
+		// Add any extensions that we want to use.
+		#[cfg(feature = "deflate")]
+		{
+			let deflate = soketto::extension::deflate::Deflate::new(soketto::Mode::Server);
+			server.add_extension(Box::new(deflate));
+		}
+
+		// Attempt the handshake.
+		match server.receive_request(&req) {
+			// The handshake has been successful so far; return the response we're given back
+			// and spawn a task to handle the long-running WebSocket server:
+			Ok(response) => {
+				tokio::spawn(async move {
+					if let Err(e) = websocket_echo_messages(server, req).await {
+						log::error!("Error upgrading to websocket connection: {}", e);
+					}
+				});
+				Ok(response.map(|()| Body::empty()))
+			}
+			// We tried to upgrade and failed early on; tell the client about the failure however we like:
+			Err(e) => {
+				log::error!("Could not upgrade connection: {}", e);
+				Ok(Response::new(Body::from("Something went wrong upgrading!")))
+			}
+		}
+	} else {
+		// The request wasn't an upgrade request; let's treat it as a standard HTTP request:
+		Ok(Response::new(Body::from("Hello HTTP!")))
+	}
+}
+
+/// Echo any messages we get from the client back to them
+async fn websocket_echo_messages(server: Server, req: Request<Body>) -> Result<(), BoxedError> {
+	// The negotiation to upgrade to a WebSocket connection has been successful so far. Next, we get back the underlying
+	// stream using `hyper::upgrade::on`, and hand this to a Soketto server to use to handle the WebSocket communication
+	// on this socket.
+	//
+	// Note: awaiting this won't succeed until the handshake response has been returned to the client, so this must be
+	// spawned on a separate task so as not to block that response being handed back.
+	let stream = hyper::upgrade::on(req).await?;
+	let stream = BufReader::new(BufWriter::new(stream.compat()));
+
+	// Get back a reader and writer that we can use to send and receive websocket messages.
+	let (mut sender, mut receiver) = server.into_builder(stream).finish();
+
+	// Echo any received messages back to the client:
+	let mut message = Vec::new();
+	loop {
+		message.clear();
+		match receiver.receive_data(&mut message).await {
+			Ok(soketto::Data::Binary(n)) => {
+				assert_eq!(n, message.len());
+				sender.send_binary_mut(&mut message).await?;
+				sender.flush().await?
+			}
+			Ok(soketto::Data::Text(n)) => {
+				assert_eq!(n, message.len());
+				if let Ok(txt) = std::str::from_utf8(&message) {
+					sender.send_text(txt).await?;
+					sender.flush().await?
+				} else {
+					break;
+				}
+			}
+			Err(soketto::connection::Error::Closed) => break,
+			Err(e) => {
+				eprintln!("Websocket connection error: {}", e);
+				break;
+			}
+		}
+	}
+
+	Ok(())
+}

src/handshake.rs

@@ -11,10 +11,13 @@
 //! [handshake]: https://tools.ietf.org/html/rfc6455#section-4
 
 pub mod client;
+#[cfg(feature = "http")]
+pub mod http;
 pub mod server;
 
 use crate::extension::{Extension, Param};
 use bytes::BytesMut;
+use sha1::{Digest, Sha1};
 use std::{fmt, io, str};
 
 pub use client::{Client, ServerResponse};
@@ -105,7 +108,15 @@ where
 		bytes.extend_from_slice(b"\r\nSec-WebSocket-Extensions: ")
 	}
 
-	while let Some(e) = iter.next() {
+	append_extension_header_value(iter, bytes)
+}
+
+// Write the extension header value to the given buffer.
+fn append_extension_header_value<'a, I>(mut extensions_iter: std::iter::Peekable<I>, bytes: &mut BytesMut)
+where
+	I: Iterator<Item = &'a Box<dyn Extension + Send>>,
+{
+	while let Some(e) = extensions_iter.next() {
 		bytes.extend_from_slice(e.name().as_bytes());
 		for p in e.params() {
 			bytes.extend_from_slice(b"; ");
@@ -115,12 +126,33 @@ where
 				bytes.extend_from_slice(v.as_bytes())
 			}
 		}
-		if iter.peek().is_some() {
+		if extensions_iter.peek().is_some() {
 			bytes.extend_from_slice(b", ")
 		}
 	}
 }
 
+// This function takes a 16 byte key (base64 encoded, and so 24 bytes of input) that is expected via
+// the `Sec-WebSocket-Key` header during a websocket handshake, and writes the response that's expected
+// to be handed back in the response header `Sec-WebSocket-Accept`.
+//
+// The response is a base64 encoding of a 160bit hash. base64 encoding uses 1 ascii character per 6 bits,
+// and 160 / 6 = 26.66 characters. The output is padded with '=' to the nearest 4 characters, so we need 28
+// bytes in total for all of the characters.
+//
+// See https://datatracker.ietf.org/doc/html/rfc6455#section-1.3 for more information on this.
+fn generate_accept_key<'k>(key_base64: &WebSocketKey) -> [u8; 28] {
+	let mut digest = Sha1::new();
+	digest.update(key_base64);
+	digest.update(KEY);
+	let d = digest.finalize();
+
+	let mut output_buf = [0; 28];
+	let n = base64::encode_config_slice(&d, base64::STANDARD, &mut output_buf);
+	debug_assert_eq!(n, 28, "encoding to base64 should be exactly 28 bytes");
+	output_buf
+}
+
 /// Enumeration of possible handshake errors.
 #[non_exhaustive]
 #[derive(Debug)]

src/handshake/client.rs

@@ -21,6 +21,8 @@ use futures::prelude::*;
 use sha1::{Digest, Sha1};
 use std::{mem, str};
 
+pub use httparse::Header;
+
 const BLOCK_SIZE: usize = 8 * 1024;
 
 /// Websocket client handshake.
@@ -32,8 +34,8 @@ pub struct Client<'a, T> {
 	host: &'a str,
 	/// The HTTP host ressource.
 	resource: &'a str,
-	/// The HTTP origin header.
-	origin: Option<&'a str>,
+	/// The HTTP headers.
+	headers: &'a [Header<'a>],
 	/// A buffer holding the base-64 encoded request nonce.
 	nonce: WebSocketKey,
 	/// The protocols to include in the handshake.
@@ -51,7 +53,7 @@ impl<'a, T: AsyncRead + AsyncWrite + Unpin> Client<'a, T> {
 			socket,
 			host,
 			resource,
-			origin: None,
+			headers: &[],
 			nonce: [0; 24],
 			protocols: Vec::new(),
 			extensions: Vec::new(),
@@ -70,9 +72,11 @@ impl<'a, T: AsyncRead + AsyncWrite + Unpin> Client<'a, T> {
 		mem::take(&mut self.buffer)
 	}
 
-	/// Set the handshake origin header.
-	pub fn set_origin(&mut self, o: &'a str) -> &mut Self {
-		self.origin = Some(o);
+	/// Set connection headers to a slice. These headers are not checked for validity,
+	/// the caller of this method is responsible for verification as well as avoiding
+	/// conflicts with internally set headers.
+	pub fn set_headers(&mut self, h: &'a [Header]) -> &mut Self {
+		self.headers = h;
 		self
 	}
 
@@ -135,10 +139,12 @@ impl<'a, T: AsyncRead + AsyncWrite + Unpin> Client<'a, T> {
 		self.buffer.extend_from_slice(b"\r\nUpgrade: websocket\r\nConnection: Upgrade");
 		self.buffer.extend_from_slice(b"\r\nSec-WebSocket-Key: ");
 		self.buffer.extend_from_slice(&self.nonce);
-		if let Some(o) = &self.origin {
-			self.buffer.extend_from_slice(b"\r\nOrigin: ");
-			self.buffer.extend_from_slice(o.as_bytes())
-		}
+		self.headers.iter().for_each(|h| {
+			self.buffer.extend_from_slice(b"\r\n");
+			self.buffer.extend_from_slice(h.name.as_bytes());
+			self.buffer.extend_from_slice(b": ");
+			self.buffer.extend_from_slice(h.value);
+		});
 		if let Some((last, prefix)) = self.protocols.split_last() {
 			self.buffer.extend_from_slice(b"\r\nSec-WebSocket-Protocol: ");
 			for p in prefix {