From cc3342aa724914c012cf88d8333cac87228fdc59 Mon Sep 17 00:00:00 2001 From: vshakitskiy Date: Tue, 4 Aug 2026 14:47:03 +0300 Subject: [PATCH] implement websockets --- CHANGELOG.md | 22 +- dev/autobahn.gleam | 52 +++ gleam.toml | 2 +- manifest.toml | 4 +- src/ewe.gleam | 294 +++++++++++++++++ src/ewe/internal/connection.gleam | 18 +- src/ewe/internal/file.gleam | 3 +- src/ewe/internal/http1.gleam | 16 + src/ewe/internal/http1/connection.gleam | 25 ++ src/ewe/internal/http1/encoder.gleam | 96 ++++-- src/ewe/internal/http1/parser.gleam | 48 ++- src/ewe/internal/http1/websocket.gleam | 376 ++++++++++++++++++++++ src/ewe/internal/websocket.gleam | 18 ++ src/ewe/internal/websocket_ffi.erl | 7 + test/ewe/internal/http1/parser_test.gleam | 37 ++- 15 files changed, 965 insertions(+), 53 deletions(-) create mode 100644 dev/autobahn.gleam create mode 100644 src/ewe/internal/http1/websocket.gleam create mode 100644 src/ewe/internal/websocket.gleam create mode 100644 src/ewe/internal/websocket_ffi.erl diff --git a/CHANGELOG.md b/CHANGELOG.md index df975cf..3cfad8c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -29,17 +29,23 @@ - In replacement of `chunked_body`/`send_chunk`/`chunked_continue`/`chunked_stop` there is now `stream_response`/`send_chunk`/`finish_chunk`/`finish_response` and no init/loop callback for response streaming anymore. -- Response streaming and server-sent events no longer spawn a process per - response, they run in the connection process itself. -- `send_chunk` and `send_event` no longer return a `Result`. Writing to a client - that has gone now ends the handler where it stands rather than letting it - carry on producing a body with nowhere to go. A stream that ends this way is - not an error and is not reported as one while a handler that crashes for its - own reasons still does. `on_close` runs either way. +- Response streaming, server-sent events and websockets no longer spawn a + process per response, they run in the connection process itself. +- `send_chunk`, `send_event`, `send_text_frame` and `send_binary_frame` no + longer return a `Result`. Writing to a client that has gone now ends the + handler where it stands rather than letting it carry on producing a body with + nowhere to go. A stream that ends this way is not an error and is not reported + as one while a handler that crashes for its own reasons still does. `on_close` + runs either way. - Rename `SSEConnection`, `SSEEvent` and `SSENext` to `SseConnection`, `SseEvent` and `SseNext`. -- Add `comment`, for the server-sent events comment that keeps an idle stream +- Add `comment` for the server-sent events comment that keeps an idle stream from being closed by an intermediary. +- Rename the `WebsocketMessage` variants to `TextFrame`, `BinaryFrame` and + `UserMessage`. +- In replacement of the `CloseCode` variants that each carried their own data + there is now `CloseReason`, either `NoCloseReason` or a code and a description + which `send_close_frame` takes. ## v4.0.1 - 04.06.2026 diff --git a/dev/autobahn.gleam b/dev/autobahn.gleam new file mode 100644 index 0000000..9411b75 --- /dev/null +++ b/dev/autobahn.gleam @@ -0,0 +1,52 @@ +//// Echo server for the Autobahn test suite. Run `make autobahn_test`. + +import ewe +import gleam/erlang/process +import gleam/http/request +import gleam/http/response +import logging + +pub fn main() -> Nil { + logging.configure() + logging.set_level(logging.Warning) + + let listener_name = process.new_name("autobahn_listener") + let connection_factory_name = process.new_name("autobahn_factory") + + let assert Ok(_started) = + ewe.new(listener_name:, connection_factory_name:, handler: handle_request) + |> ewe.bind("0.0.0.0") + |> ewe.listening(on: 8080) + |> ewe.start + + process.sleep_forever() +} + +fn handle_request( + request: request.Request(ewe.Connection), +) -> response.Response(ewe.Body) { + ewe.upgrade_websocket( + request:, + on_init: fn(_conn, messages) { #(Nil, messages) }, + handler: echo_message, + on_close: fn(_conn, _state) { Nil }, + ) +} + +fn echo_message( + conn: ewe.WebsocketConnection, + state: Nil, + message: ewe.WebsocketMessage(Nil), +) -> ewe.WebsocketNext(Nil, Nil) { + case message { + ewe.TextFrame(text) -> { + ewe.send_text_frame(conn, text) + ewe.websocket_continue(state) + } + ewe.BinaryFrame(data) -> { + ewe.send_binary_frame(conn, data) + ewe.websocket_continue(state) + } + ewe.UserMessage(_message) -> ewe.websocket_continue(state) + } +} diff --git a/gleam.toml b/gleam.toml index 3b99791..4f21813 100644 --- a/gleam.toml +++ b/gleam.toml @@ -15,7 +15,7 @@ gleam_otp = ">= 1.1.0 and < 2.0.0" gleam_http = ">= 4.3.0 and < 5.0.0" logging = ">= 1.3.0 and < 2.0.0" gleam_erlang = ">= 1.3.0 and < 2.0.0" -websocks = ">= 3.0.0 and < 4.0.0" +websocks = ">= 4.0.1 and < 5.0.0" [dev-dependencies] gleeunit = ">= 1.0.0 and < 2.0.0" diff --git a/manifest.toml b/manifest.toml index 7e5d050..0a8964c 100644 --- a/manifest.toml +++ b/manifest.toml @@ -16,7 +16,7 @@ packages = [ { name = "gleeunit", version = "1.11.0", build_tools = ["gleam"], requirements = ["gleam_stdlib"], otp_app = "gleeunit", source = "hex", outer_checksum = "EC31ABA74256AEA531EDF8169931D775BBB384FED0A8A1BDC4DD9354E3E21826" }, { name = "glisten", version = "9.0.1", build_tools = ["gleam"], requirements = ["gleam_erlang", "gleam_otp", "gleam_stdlib", "logging"], source = "git", repo = "https://github.com/vshakitskiy/glisten.git", commit = "63e9a39f7dc35526c2f62128f44f871dab245d15" }, { name = "logging", version = "1.5.0", build_tools = ["gleam"], requirements = ["gleam_stdlib"], otp_app = "logging", source = "hex", outer_checksum = "BC5F18CE5DD9686100229FE5409BDC3DD5C46D5A7DF2F804AD2D8F0DD6C5060E" }, - { name = "websocks", version = "3.0.1", build_tools = ["gleam"], requirements = ["gleam_crypto", "gleam_erlang", "gleam_stdlib"], otp_app = "websocks", source = "hex", outer_checksum = "C70340E5B6C3390383ADA17029DCA6F8903863A7AD8CD8E1520EDCC4FE70D6FD" }, + { name = "websocks", version = "4.0.1", build_tools = ["gleam"], requirements = ["gleam_crypto", "gleam_erlang", "gleam_stdlib"], otp_app = "websocks", source = "hex", outer_checksum = "89B0C31A032CBE28D4C5FB5CC25A8A690669FEBA501C63F99A4E8F5748750B2A" }, ] [requirements] @@ -28,4 +28,4 @@ gleam_stdlib = { version = ">= 1.0.0 and < 2.0.0" } gleeunit = { version = ">= 1.0.0 and < 2.0.0" } glisten = { git = "https://github.com/vshakitskiy/glisten.git", ref = "63e9a39" } logging = { version = ">= 1.3.0 and < 2.0.0" } -websocks = { version = ">= 3.0.0 and < 4.0.0" } +websocks = { version = ">= 4.0.1 and < 5.0.0" } diff --git a/src/ewe.gleam b/src/ewe.gleam index 37c9ff6..cececb2 100644 --- a/src/ewe.gleam +++ b/src/ewe.gleam @@ -5,7 +5,9 @@ import ewe/internal/http1/body as http1_body import ewe/internal/http1/connection as http1 import ewe/internal/http1/encoder import ewe/internal/http1/sse as http1_sse +import ewe/internal/http1/websocket as http1_websocket import ewe/internal/sse +import ewe/internal/websocket import gleam/bytes_tree import gleam/erlang/process import gleam/http @@ -20,12 +22,15 @@ import gleam/otp/factory_supervisor as factory import gleam/otp/static_supervisor as supervisor import gleam/otp/supervision import gleam/result +import gleam/string import glisten import glisten/internal/handler import glisten/internal/listener import glisten/socket import glisten/socket/options import glisten/transport +import logging +import websocks pub type Connection = connection.Connection @@ -37,6 +42,7 @@ pub type Body { File(connection.File) Streaming(connection.Streaming) Sse(connection.Sse) + Websocket(connection.Websocket) } pub type IpAddress { @@ -393,6 +399,7 @@ fn to_internal_body(body: Body) -> connection.Body { File(file) -> connection.File(file) Streaming(streaming) -> connection.Streaming(streaming) Sse(sse) -> connection.Sse(sse) + Websocket(websocket) -> connection.Websocket(websocket) } } @@ -723,3 +730,290 @@ pub fn sse( response.set_body(response, Sse(connection.SseMetadata(stream))) } + +/// A handle for sending frames on an open WebSocket. +pub type WebsocketConnection = + connection.WebsocketConnection + +/// What the client sent or what the rest of your program sent to the subject +/// given to `on_init`. Ping and pong frames are answered by the server and do +/// not reach the handler. +pub type WebsocketMessage(user_message) { + TextFrame(text: String) + BinaryFrame(data: BitArray) + UserMessage(message: user_message) +} + +fn from_internal_websocket_message( + message: websocket.Message(user_message), +) -> WebsocketMessage(user_message) { + case message { + websocket.TextFrame(text) -> TextFrame(text) + websocket.BinaryFrame(data) -> BinaryFrame(data) + websocket.UserMessage(message) -> UserMessage(message) + } +} + +/// What a WebSocket does after the handler has dealt with a message. Build it +/// with `websocket_continue`, `websocket_stop` or `websocket_stop_abnormal`. +pub opaque type WebsocketNext(user_state, user_message) { + WebsocketContinue(user_state, Option(process.Selector(user_message))) + WebsocketStop + WebsocketStopAbnormal(reason: String) +} + +/// Carries on handling further messages with `user_state` and the selector the +/// connection already has. +pub fn websocket_continue( + user_state: user_state, +) -> WebsocketNext(user_state, user_message) { + WebsocketContinue(user_state, None) +} + +/// Carries on listening on `selector` from here on instead of the one the +/// connection was started with. +pub fn websocket_continue_with_selector( + user_state: user_state, + selector: process.Selector(user_message), +) -> WebsocketNext(user_state, user_message) { + WebsocketContinue(user_state, Some(selector)) +} + +/// Ends the WebSocket. +pub fn websocket_stop() -> WebsocketNext(user_state, user_message) { + WebsocketStop +} + +/// Ends the WebSocket reporting `reason` as the cause. +pub fn websocket_stop_abnormal( + reason: String, +) -> WebsocketNext(user_state, user_message) { + WebsocketStopAbnormal(reason) +} + +/// Why a WebSocket is being closed, sent to the client in the close frame. +pub type CloseReason { + /// Close without saying why. + NoCloseReason + /// Close with a status code and a description, which may be empty. + CloseReason(code: CloseCode, reason: String) +} + +/// The status code a close frame carries. The codes that exist only to be +/// reported locally such as 1005 and 1006 are absent. Sending one is a +/// protocol violation. +pub type CloseCode { + /// The connection did what it was for and is closing normally (1000). + NormalClosure + /// The endpoint is going away, from a server shutdown or a client navigating + /// away (1001). + GoingAway + /// The other end broke the protocol (1002). + ProtocolError + /// Data arrived that this endpoint cannot accept (1003). + UnsupportedData + /// A message did not match the type it declared, such as a text frame that + /// is not UTF-8 (1007). + InvalidPayloadData + /// The other end broke your rules, when no more specific code applies (1008). + PolicyViolation + /// A message was larger than this endpoint will handle (1009). + MessageTooBig + /// An extension the client required was not negotiated (1010). + MandatoryExtension + /// Something went wrong on this side (1011). + InternalError + /// The server is restarting, and clients may reconnect shortly (1012). + ServiceRestart + /// The server is overloaded and the client should retry later (1013). + TryAgainLater + /// An upstream server answered badly (1014). + BadGateway + /// An application specific code, which must be between 3000 and 4999. + ApplicationCode(code: Int) +} + +fn to_internal_close_reason(reason: CloseReason) -> websocks.CloseReason { + case reason { + NoCloseReason -> websocks.NoCloseReason + CloseReason(code:, reason:) -> + websocks.CloseReason(to_internal_close_code(code), reason) + } +} + +fn to_internal_close_code(code: CloseCode) -> websocks.CloseCode { + case code { + NormalClosure -> websocks.NormalClosure + GoingAway -> websocks.GoingAway + ProtocolError -> websocks.ProtocolError + UnsupportedData -> websocks.UnsupportedData + InvalidPayloadData -> websocks.InvalidPayloadData + PolicyViolation -> websocks.PolicyViolation + MessageTooBig -> websocks.MessageTooBig + MandatoryExtension -> websocks.MandatoryExtension + InternalError -> websocks.InternalError + ServiceRestart -> websocks.ServiceRestart + TryAgainLater -> websocks.TryAgainLater + BadGateway -> websocks.BadGateway + ApplicationCode(code:) -> websocks.ApplicationCode(code:) + } +} + +/// Sends a text frame. If the client has gone the WebSocket ends here. +pub fn send_text_frame(conn: WebsocketConnection, text: String) -> Nil { + case conn { + connection.Http1Websocket(conn) -> http1_websocket.send_text(conn, text) + connection.Http2Websocket -> todo as "HTTP/2 is not implemented yet!" + } +} + +/// Sends a binary frame. If the client has gone the WebSocket ends here. +pub fn send_binary_frame(conn: WebsocketConnection, data: BitArray) -> Nil { + case conn { + connection.Http1Websocket(conn) -> http1_websocket.send_binary(conn, data) + connection.Http2Websocket -> todo as "HTTP/2 is not implemented yet!" + } +} + +/// Starts the closing handshake and ends the WebSocket. Return the value this +/// gives back from your handler, no frame can be sent after it. +pub fn send_close_frame( + conn: WebsocketConnection, + reason: CloseReason, +) -> WebsocketNext(user_state, user_message) { + case conn { + connection.Http1Websocket(conn) -> + http1_websocket.send_close(conn, to_internal_close_reason(reason)) + connection.Http2Websocket -> todo as "HTTP/2 is not implemented yet!" + } + + WebsocketStop +} + +/// Turns the response into a WebSocket which runs until the handler stops it +/// or the client goes away. The connection stops being HTTP once the handshake +/// is written so it never carries another request. +/// +/// A request that is not a valid handshake is answered with a 400 and the +/// handler is never run. +/// +/// `on_init` is called once, with an empty selector to add whatever the rest of +/// your program sends this connection to and returns the starting state along +/// with that selector. +/// `handler` is called for each frame from the client and each message the +/// selector picks up. +/// `on_close` is called once however the WebSocket ended. +pub fn upgrade_websocket( + request request: request.Request(Connection), + on_init on_init: fn(WebsocketConnection, process.Selector(user_message)) -> + #(user_state, process.Selector(user_message)), + handler handler: fn( + WebsocketConnection, + user_state, + WebsocketMessage(user_message), + ) -> WebsocketNext(user_state, user_message), + on_close on_close: fn(WebsocketConnection, user_state) -> Nil, +) -> response.Response(Body) { + let step = fn(conn, state, message) { + case handler(conn, state, from_internal_websocket_message(message)) { + WebsocketContinue(state, messages) -> websocket.Proceed(state, messages) + WebsocketStop -> websocket.Halt(connection.Stopped) + WebsocketStopAbnormal(reason) -> + websocket.Halt(connection.StoppedAbnormal(reason)) + } + } + + let socket = fn(conn) { + case conn { + connection.Http1Websocket(conn) -> + http1_websocket.run(conn, on_init, step, on_close) + connection.Http2Websocket -> todo as "HTTP/2 is not implemented yet!" + } + } + + case request.body { + connection.Http1(conn) -> + case http1_websocket.handshake(request.method, conn) { + Ok(http1_websocket.Handshake(accept:, compression:)) -> { + let context = websocks.create_context(compression, websocks.Server) + + response.Response( + status: 101, + headers: handshake_headers(accept, compression), + body: Websocket(connection.WebsocketMetadata( + context:, + handler: socket, + )), + ) + } + Error(error) -> { + logging.log( + logging.Debug, + "Rejected a WebSocket handshake: " + <> http1_websocket.handshake_error_to_string(error), + ) + + response.set_body(response.new(400), Empty) + } + } + connection.Http2 -> todo as "HTTP/2 is not implemented yet!" + } +} + +fn handshake_headers( + accept: String, + compression: option.Option(websocks.CompressionExtensions), +) -> List(#(String, String)) { + let headers = [ + #("connection", "upgrade"), + #("upgrade", "websocket"), + #("sec-websocket-accept", accept), + ] + + case compression { + Some(extensions) -> [ + #("sec-websocket-extensions", compression_header(extensions)), + ..headers + ] + None -> headers + } +} + +fn compression_header(extensions: websocks.CompressionExtensions) -> String { + let websocks.CompressionExtensions( + client_no_context_takeover:, + client_max_window_bits:, + server_no_context_takeover:, + server_max_window_bits:, + ) = extensions + + ["permessage-deflate"] + |> append_flag(client_no_context_takeover, "client_no_context_takeover") + |> append_flag(server_no_context_takeover, "server_no_context_takeover") + |> append_window_bits(client_max_window_bits, "client_max_window_bits") + |> append_window_bits(server_max_window_bits, "server_max_window_bits") + |> list.reverse + |> string.join("; ") +} + +fn append_flag( + parameters: List(String), + enabled: Bool, + name: String, +) -> List(String) { + case enabled { + True -> [name, ..parameters] + False -> parameters + } +} + +fn append_window_bits( + parameters: List(String), + bits: Option(Int), + name: String, +) -> List(String) { + case bits { + Some(bits) -> [name <> "=" <> int.to_string(bits), ..parameters] + None -> parameters + } +} diff --git a/src/ewe/internal/connection.gleam b/src/ewe/internal/connection.gleam index 4806b3a..a1c2c47 100644 --- a/src/ewe/internal/connection.gleam +++ b/src/ewe/internal/connection.gleam @@ -4,6 +4,7 @@ import gleam/erlang/process import gleam/option import glisten import glisten/internal/handler +import websocks pub type Connection { Http1(http1.Connection) @@ -17,18 +18,28 @@ pub type Body { File(File) Streaming(Streaming) Sse(Sse) + Websocket(Websocket) } pub type Sse { SseMetadata(handler: fn(SseConnection) -> Outcome) } +/// The context is built during the handshake where the negotiated extensions +/// are known and handed to whichever protocol goes on to run the socket. +pub type Websocket { + WebsocketMetadata( + context: websocks.Context, + handler: fn(WebsocketConnection) -> Outcome, + ) +} + pub type Streaming { StreamingMetadata(handler: fn(ResponseWriter) -> Nil) } /// A raw descriptor belongs to the process that opened it, so whether the -/// handler can carry one depends on the protocol: an HTTP/1 handler runs in the +/// handler can carry one depends on the protocol. An HTTP/1 handler runs in the /// process that writes the socket, an HTTP/2 stream handler does not. pub type File { /// Already open, and closed by whoever writes or drops the response. @@ -49,6 +60,11 @@ pub type SseConnection { Http2Sse } +pub type WebsocketConnection { + Http1Websocket(http1.WebsocketConnection) + Http2Websocket +} + pub type Outcome { Stopped StoppedAbnormal(reason: String) diff --git a/src/ewe/internal/file.gleam b/src/ewe/internal/file.gleam index 1679159..ee5599c 100644 --- a/src/ewe/internal/file.gleam +++ b/src/ewe/internal/file.gleam @@ -79,7 +79,8 @@ pub fn release_body(body: connection.Body) -> Nil { | connection.Text(..) | connection.Empty | connection.Streaming(..) - | connection.Sse(..) -> Nil + | connection.Sse(..) + | connection.Websocket(..) -> Nil } } diff --git a/src/ewe/internal/http1.gleam b/src/ewe/internal/http1.gleam index dcd393e..430051c 100644 --- a/src/ewe/internal/http1.gleam +++ b/src/ewe/internal/http1.gleam @@ -59,6 +59,7 @@ pub fn handle_message( read: 0, chunk_remaining: 0, config: state.config, + upgrade: metadata.upgrade, ) let response = @@ -229,6 +230,21 @@ fn send_response( } } } + // Once the handshake is written the connection has stopped being HTTP, so + // it never goes back to the request loop however the socket ends. + encoder.RemainderWebsocket(context:, handler: websocket_handler) -> { + use Nil <- result.try(transport.send(transport, socket, head)) + + let outcome = + http1.WebsocketConnection(transport:, socket:, context:) + |> connection.Http1Websocket + |> websocket_handler + + case outcome { + connection.Stopped -> Ok(SentClose) + connection.StoppedAbnormal(reason) -> Ok(SentAbnormal(reason)) + } + } encoder.RemainderSse(handler: sse_handler, framing:) -> { use Nil <- result.try(transport.send(transport, socket, head)) diff --git a/src/ewe/internal/http1/connection.gleam b/src/ewe/internal/http1/connection.gleam index a84318a..cc71aa4 100644 --- a/src/ewe/internal/http1/connection.gleam +++ b/src/ewe/internal/http1/connection.gleam @@ -1,6 +1,8 @@ import gleam/erlang/process +import gleam/option import glisten/socket import glisten/transport +import websocks /// The limits and timeouts an HTTP/1 connection is held to. pub type Config { @@ -39,6 +41,7 @@ pub type Connection { read: Int, chunk_remaining: Int, config: Config, + upgrade: option.Option(Upgrade), ) } @@ -49,6 +52,18 @@ pub type Framing { NoBody } +/// What a request asked to become instead of HTTP/1, gathered as the headers +/// go past rather than looked up again afterwards. Whether the fields amount to +/// a handshake is for whoever answers it to decide. +pub type Upgrade { + WebsocketUpgrade( + key: option.Option(String), + version: option.Option(String), + extensions: option.Option(String), + ) + OtherUpgrade(name: String) +} + /// Handlers run inside the connection process, so they report what they did to /// the request body and the response stream by messaging it. pub type Signal { @@ -105,3 +120,13 @@ pub type SseConnection { framing: StreamFraming, ) } + +/// The connection has stopped being HTTP by this point so there is nothing to +/// keep alive and nothing to report back about reuse. +pub type WebsocketConnection { + WebsocketConnection( + transport: transport.Transport, + socket: socket.Socket, + context: websocks.Context, + ) +} diff --git a/src/ewe/internal/http1/encoder.gleam b/src/ewe/internal/http1/encoder.gleam index 3e66293..eefbb83 100644 --- a/src/ewe/internal/http1/encoder.gleam +++ b/src/ewe/internal/http1/encoder.gleam @@ -15,6 +15,7 @@ import gleam/result import gleam/string import glisten/socket import glisten/transport +import websocks pub type EncodeError { UnsafeHeader(name: String) @@ -37,6 +38,10 @@ pub type Remainder { handler: fn(connection.SseConnection) -> connection.Outcome, framing: http1.StreamFraming, ) + RemainderWebsocket( + context: websocks.Context, + handler: fn(connection.WebsocketConnection) -> connection.Outcome, + ) } pub type Encoded { @@ -60,9 +65,12 @@ pub fn encode_response( let keep_alive = http1.and_keep_alive(keep_alive, state.keep_alive) let status = response.status - let encoded = case is_bodyless(status) { - True -> bodyless(state, status, keep_alive, response.body) - False -> encode_body(state, status, keep_alive, version, response.body) + let encoded = case response.body, is_bodyless(status) { + connection.Websocket(connection.WebsocketMetadata(context:, handler:)), + _bodyless + -> switching_protocols(state, status, context, handler) + body, True -> bodyless(state, status, keep_alive, body) + body, False -> encode_body(state, status, keep_alive, version, body) } // A HEAD response keeps the framing headers it would have had minus the body. @@ -88,6 +96,27 @@ fn bodyless( Encoded(build_head(state, status, keep_alive, <<>>), keep_alive, NoRemainder) } +/// The handshake's own `connection` and `upgrade` headers are the whole point +/// of the response so no framing header is written and the head is closed off +/// without one. +fn switching_protocols( + state: EncodeState, + status: Int, + context: websocks.Context, + handler: fn(connection.WebsocketConnection) -> connection.Outcome, +) -> Encoded { + let head = + append_date(state.tree) + |> bytes_tree.append(<<"\r\n":utf8>>) + |> bytes_tree.prepend(status_line(status)) + + Encoded( + head, + http1.CloseAfterResponse, + RemainderWebsocket(context:, handler:), + ) +} + fn encode_body( state: EncodeState, status: Int, @@ -119,6 +148,7 @@ fn encode_body( encode_stream(state, status, keep_alive, version, handler) connection.Sse(connection.SseMetadata(handler)) -> encode_sse(state, status, keep_alive, version, handler) + connection.Websocket(..) -> bodyless(state, status, keep_alive, body) } } @@ -129,7 +159,8 @@ fn drop_body(encoded: Encoded) -> Encoded { NoRemainder | RemainderInline(..) | RemainderStream(..) - | RemainderSse(..) -> Nil + | RemainderSse(..) + | RemainderWebsocket(..) -> Nil } Encoded(..encoded, remainder: NoRemainder) @@ -318,11 +349,13 @@ fn finish(writer: ResponseWriter) -> Nil { type Reserved { Framing FramingAndSse + Handshake } fn reserved(body: connection.Body) -> Reserved { case body { connection.Sse(..) -> FramingAndSse + connection.Websocket(..) -> Handshake connection.Bytes(..) | connection.Text(..) | connection.Empty @@ -339,10 +372,12 @@ fn encode_headers( use state, #(name, value) <- list.try_fold(headers, initial) // TODO: just trust the handler? - case name { - "content-length" | "transfer-encoding" | "date" -> Ok(state) - "content-type" | "cache-control" if reserved == FramingAndSse -> Ok(state) - "connection" -> + case name, reserved { + "date", _reserved -> Ok(state) + _name, Handshake -> append_header(state, name, value) + "content-length", _reserved | "transfer-encoding", _reserved -> Ok(state) + "content-type", FramingAndSse | "cache-control", FramingAndSse -> Ok(state) + "connection", _reserved -> case parser.find_unsafe_header_byte(value) { Error(Nil) -> { let lowered = value |> bit_array.from_string |> parser.lowercase_ascii @@ -354,29 +389,32 @@ fn encode_headers( } Ok(_position) -> Error(UnsafeHeader(name)) } - _other -> - case - parser.find_unsafe_header_byte(name), - parser.find_unsafe_header_byte(value) - { - Error(Nil), Error(Nil) -> { - let tree = - bytes_tree.append_string(state.tree, name) - |> bytes_tree.append(<<": ":utf8>>) - |> bytes_tree.append_string(value) - |> bytes_tree.append(<<"\r\n":utf8>>) - - Ok(EncodeState(..state, tree:)) - } - _other, _other -> Error(UnsafeHeader(name)) - } + _name, _reserved -> append_header(state, name, value) + } +} + +fn append_header( + state: EncodeState, + name: String, + value: String, +) -> Result(EncodeState, EncodeError) { + case + parser.find_unsafe_header_byte(name), + parser.find_unsafe_header_byte(value) + { + Error(Nil), Error(Nil) -> { + let tree = + bytes_tree.append_string(state.tree, name) + |> bytes_tree.append(<<": ":utf8, value:utf8, "\r\n":utf8>>) + + Ok(EncodeState(..state, tree:)) + } + _name, _value -> Error(UnsafeHeader(name)) } } fn append_date(tree: bytes_tree.BytesTree) -> bytes_tree.BytesTree { - bytes_tree.append_string(tree, "date: ") - |> bytes_tree.append(clock.get()) - |> bytes_tree.append(<<"\r\n":utf8>>) + bytes_tree.append(tree, <<"date: ":utf8, clock.get():bits, "\r\n":utf8>>) } fn append_connection( @@ -388,9 +426,7 @@ fn append_connection( http1.CloseAfterResponse -> <<"close":utf8>> } - bytes_tree.append(tree, <<"connection: ":utf8>>) - |> bytes_tree.append(value) - |> bytes_tree.append(<<"\r\n":utf8>>) + bytes_tree.append(tree, <<"connection: ":utf8, value:bits, "\r\n":utf8>>) } fn status_line(status: Int) -> BitArray { diff --git a/src/ewe/internal/http1/parser.gleam b/src/ewe/internal/http1/parser.gleam index 1fa8404..36ca208 100644 --- a/src/ewe/internal/http1/parser.gleam +++ b/src/ewe/internal/http1/parser.gleam @@ -25,7 +25,7 @@ pub type Metadata { Metadata( framing: http1.Framing, keep_alive: http1.KeepAlive, - upgrade: option.Option(String), + upgrade: option.Option(http1.Upgrade), ) } @@ -415,6 +415,9 @@ pub type HeaderState { connection: ConnectionIntent, connection_upgrade: Bool, upgrade: option.Option(String), + websocket_key: option.Option(String), + websocket_version: option.Option(String), + websocket_extensions: option.Option(String), host: option.Option(#(String, option.Option(Int))), ) } @@ -426,6 +429,9 @@ pub fn initial_header_state() -> HeaderState { connection: NothingRequested, connection_upgrade: False, upgrade: option.None, + websocket_key: option.None, + websocket_version: option.None, + websocket_extensions: option.None, host: option.None, ) } @@ -459,12 +465,22 @@ fn complete_metadata( NothingRequested, Http11 -> http1.KeepAlive NothingRequested, Http10 -> http1.CloseAfterResponse } - let upgrade = case state.connection_upgrade { - True -> state.upgrade - False -> option.None + Metadata(framing:, keep_alive:, upgrade: resolve_upgrade(state)) +} + +/// An `Upgrade` header only asks for anything if the `Connection` header named +/// it, so one without the other is nothing. +fn resolve_upgrade(state: HeaderState) -> option.Option(http1.Upgrade) { + case state.connection_upgrade, state.upgrade { + True, option.Some("websocket") -> + option.Some(http1.WebsocketUpgrade( + key: state.websocket_key, + version: state.websocket_version, + extensions: state.websocket_extensions, + )) + True, option.Some(name) -> option.Some(http1.OtherUpgrade(name)) + True, option.None | False, _upgrade -> option.None } - - Metadata(framing:, keep_alive:, upgrade:) } pub fn parse_headers( @@ -561,6 +577,26 @@ fn classify( let lowered = value |> lowercase_ascii |> unsafe_to_string StepDone(HeaderState(..state, upgrade: option.Some(lowered))) } + // The key is base64 and is echoed back as sent, so unlike the rest it is + // kept with its case. + "sec-websocket-key" -> + StepDone( + HeaderState( + ..state, + websocket_key: option.Some(unsafe_to_string(value)), + ), + ) + "sec-websocket-version" -> + StepDone( + HeaderState( + ..state, + websocket_version: option.Some(unsafe_to_string(value)), + ), + ) + "sec-websocket-extensions" -> { + let lowered = value |> lowercase_ascii |> unsafe_to_string + StepDone(HeaderState(..state, websocket_extensions: option.Some(lowered))) + } "host" -> case state.host { option.Some(_host) -> ParseError(DuplicateHost) diff --git a/src/ewe/internal/http1/websocket.gleam b/src/ewe/internal/http1/websocket.gleam new file mode 100644 index 0000000..d10d8ea --- /dev/null +++ b/src/ewe/internal/http1/websocket.gleam @@ -0,0 +1,376 @@ +import ewe/internal/connection +import ewe/internal/http1/connection as http1 +import ewe/internal/stream +import ewe/internal/websocket +import gleam/bytes_tree +import gleam/dynamic +import gleam/erlang/atom +import gleam/erlang/process +import gleam/http +import gleam/option +import gleam/result +import glisten/socket +import glisten/socket/options +import glisten/transport +import websocks + +pub type HandshakeError { + MethodNotGet + NotAnUpgrade + NotWebsocket + UnsupportedVersion + MissingKey +} + +pub fn handshake_error_to_string(error: HandshakeError) -> String { + case error { + MethodNotGet -> "the handshake must be a GET" + NotAnUpgrade -> "the connection header does not request an upgrade" + NotWebsocket -> "the upgrade header does not name websocket" + UnsupportedVersion -> "only sec-websocket-version 13 is supported" + MissingKey -> "missing sec-websocket-key header" + } +} + +/// What the handshake settled on. The key the client checks the reply against +/// and the compression it asked for. +pub type Handshake { + Handshake( + accept: String, + compression: option.Option(websocks.CompressionExtensions), + ) +} + +/// Everything this needs was picked up while the headers were parsed so the +/// request is not walked again here. +pub fn handshake( + method: http.Method, + conn: http1.Connection, +) -> Result(Handshake, HandshakeError) { + use Nil <- result.try(case method { + http.Get -> Ok(Nil) + _method -> Error(MethodNotGet) + }) + + case conn.upgrade { + option.None -> Error(NotAnUpgrade) + option.Some(http1.OtherUpgrade(..)) -> Error(NotWebsocket) + option.Some(http1.WebsocketUpgrade(key:, version:, extensions:)) -> { + use Nil <- result.try(case version { + option.Some("13") -> Ok(Nil) + option.Some(_other) | option.None -> Error(UnsupportedVersion) + }) + + use key <- result.map(option.to_result(key, MissingKey)) + + let compression = case extensions { + option.Some(header) -> + case websocks.has_deflate(header) { + True -> option.Some(websocks.get_compression_extensions(header)) + False -> option.None + } + option.None -> option.None + } + + Handshake(accept: websocks.compute_accept(key), compression:) + } + } +} + +/// How many socket messages are delivered before the loop rearms. +const active_count = 100 + +pub fn run( + conn: http1.WebsocketConnection, + on_init: fn(connection.WebsocketConnection, process.Selector(user_message)) -> + #(user_state, process.Selector(user_message)), + step: fn( + connection.WebsocketConnection, + user_state, + websocket.Message(user_message), + ) -> websocket.Step(user_state, user_message), + on_close: fn(connection.WebsocketConnection, user_state) -> Nil, +) -> connection.Outcome { + let #(state, messages) = + on_init(connection.Http1Websocket(conn), process.new_selector()) + + case activate(conn) { + Ok(Nil) -> + loop(conn, merge_socket_selector(messages), state, step, on_close) + Error(reason) -> + socket.reason_to_string(reason) + |> connection.StoppedAbnormal + |> ended(conn, state, on_close, _) + } +} + +/// Every way a socket ends frees the compression resources the context holds +/// and runs the handler's `on_close`. +fn ended( + conn: http1.WebsocketConnection, + state: user_state, + on_close: fn(connection.WebsocketConnection, user_state) -> Nil, + outcome: connection.Outcome, +) -> connection.Outcome { + websocks.close_context(conn.context) + on_close(connection.Http1Websocket(conn), state) + outcome +} + +fn loop( + conn: http1.WebsocketConnection, + selector: process.Selector(Received(user_message)), + state: user_state, + step: fn( + connection.WebsocketConnection, + user_state, + websocket.Message(user_message), + ) -> websocket.Step(user_state, user_message), + on_close: fn(connection.WebsocketConnection, user_state) -> Nil, +) -> connection.Outcome { + case process.selector_receive_forever(selector) { + Closed -> ended(conn, state, on_close, connection.Stopped) + Failed(reason) -> + ended(conn, state, on_close, connection.StoppedAbnormal(reason)) + Exhausted -> + case activate(conn) { + Ok(Nil) -> loop(conn, selector, state, step, on_close) + Error(reason) -> + socket.reason_to_string(reason) + |> connection.StoppedAbnormal + |> ended(conn, state, on_close, _) + } + Packet(data) -> + websocks.push_data(conn.context, data) + |> with_context(conn, _) + |> drain(selector, state, step, on_close) + Received(message) -> + websocket.UserMessage(message) + |> deliver(conn, selector, state, step, on_close, _) + } +} + +/// One read can carry several frames so the buffer is drained before the loop +/// waits on the socket again. +fn drain( + conn: http1.WebsocketConnection, + selector: process.Selector(Received(user_message)), + state: user_state, + step: fn( + connection.WebsocketConnection, + user_state, + websocket.Message(user_message), + ) -> websocket.Step(user_state, user_message), + on_close: fn(connection.WebsocketConnection, user_state) -> Nil, +) -> connection.Outcome { + case websocks.next_frame(conn.context) { + Error(_violation) -> + close(conn, websocks.CloseReason(websocks.ProtocolError, "")) + |> resolve(conn, state, on_close, _) + Ok(websocks.MoreData(context:)) -> + with_context(conn, context) + |> loop(selector, state, step, on_close) + Ok(websocks.Decoded(frame:, context:)) -> { + let conn = with_context(conn, context) + + case frame { + // Answered here rather than handed on since a peer's keepalive is not + // the handler's business. + websocks.Control(websocks.Ping(payload)) -> + case + write(conn, websocks.encode_pong_frame(payload:, masking: none)) + { + Ok(Nil) -> drain(conn, selector, state, step, on_close) + Error(reason) -> + socket.reason_to_string(reason) + |> connection.StoppedAbnormal + |> ended(conn, state, on_close, _) + } + websocks.Control(websocks.Pong(_payload)) -> + drain(conn, selector, state, step, on_close) + // The peer started the closing handshake so it is echoed back and the + // socket is done. + websocks.Control(websocks.Close(reason)) -> + close(conn, reason) |> resolve(conn, state, on_close, _) + websocks.Text(payload) -> + websocket.TextFrame(unsafe_to_string(payload)) + |> deliver(conn, selector, state, step, on_close, _) + websocks.Binary(payload) -> + websocket.BinaryFrame(payload) + |> deliver(conn, selector, state, step, on_close, _) + // Fragments are reassembled by the decoder so one never surfaces. + websocks.Continuation(_payload) -> + drain(conn, selector, state, step, on_close) + } + } + } +} + +fn deliver( + conn: http1.WebsocketConnection, + selector: process.Selector(Received(user_message)), + state: user_state, + step: fn( + connection.WebsocketConnection, + user_state, + websocket.Message(user_message), + ) -> websocket.Step(user_state, user_message), + on_close: fn(connection.WebsocketConnection, user_state) -> Nil, + message: websocket.Message(user_message), +) -> connection.Outcome { + let handle = connection.Http1Websocket(conn) + + case stream.rescue_dead(fn() { step(handle, state, message) }) { + Error(_reason) -> ended(conn, state, on_close, connection.Stopped) + Ok(websocket.Proceed(user_state: state, messages:)) -> { + let selector = case messages { + option.Some(messages) -> merge_socket_selector(messages) + option.None -> selector + } + + drain(conn, selector, state, step, on_close) + } + Ok(websocket.Halt(outcome)) -> ended(conn, state, on_close, outcome) + } +} + +/// A close the server sends ends the socket either way, so only whether the +/// frame reached the peer decides how it is reported. +fn resolve( + conn: http1.WebsocketConnection, + state: user_state, + on_close: fn(connection.WebsocketConnection, user_state) -> Nil, + sent: Result(Nil, socket.SocketReason), +) -> connection.Outcome { + case sent { + Ok(Nil) -> ended(conn, state, on_close, connection.Stopped) + Error(reason) -> + socket.reason_to_string(reason) + |> connection.StoppedAbnormal + |> ended(conn, state, on_close, _) + } +} + +pub fn send_text(conn: http1.WebsocketConnection, text: String) -> Nil { + websocks.encode_text_frame( + payload: bit_array_from_string(text), + context: conn.context, + masking: none, + ) + |> write_or_die(conn, _) +} + +pub fn send_binary(conn: http1.WebsocketConnection, data: BitArray) -> Nil { + websocks.encode_binary_frame( + payload: data, + context: conn.context, + masking: none, + ) + |> write_or_die(conn, _) +} + +pub fn send_close( + conn: http1.WebsocketConnection, + reason: websocks.CloseReason, +) -> Nil { + websocks.encode_close_frame(reason:, masking: none) + |> write_or_die(conn, _) +} + +const none = option.None + +fn with_context( + conn: http1.WebsocketConnection, + context: websocks.Context, +) -> http1.WebsocketConnection { + http1.WebsocketConnection(..conn, context:) +} + +fn close( + conn: http1.WebsocketConnection, + reason: websocks.CloseReason, +) -> Result(Nil, socket.SocketReason) { + write(conn, websocks.encode_close_frame(reason:, masking: none)) +} + +fn write( + conn: http1.WebsocketConnection, + frame: BitArray, +) -> Result(Nil, socket.SocketReason) { + bytes_tree.from_bit_array(frame) + |> transport.send(conn.transport, conn.socket, _) +} + +fn write_or_die(conn: http1.WebsocketConnection, frame: BitArray) -> Nil { + case write(conn, frame) { + Ok(Nil) -> Nil + Error(reason) -> stream.dead(reason) + } +} + +/// glisten rearms the socket only once its loop callback returns and a socket +/// does not return until it is over, so the frames have to be asked for here. +fn activate( + conn: http1.WebsocketConnection, +) -> Result(Nil, socket.SocketReason) { + transport.set_opts(conn.transport, conn.socket, [ + options.ActiveMode(options.Count(active_count)), + ]) +} + +type Received(user_message) { + Received(user_message) + Packet(BitArray) + Closed + Failed(reason: String) + Exhausted +} + +fn merge_socket_selector( + messages: process.Selector(user_message), +) -> process.Selector(Received(user_message)) { + process.map_selector(messages, Received) + |> process.merge_selector(socket_selector()) +} + +fn socket_selector() -> process.Selector(Received(user_message)) { + process.new_selector() + |> process.select_record(atom.create("tcp"), 2, packet) + |> process.select_record(atom.create("ssl"), 2, packet) + |> process.select_record(atom.create("tcp_closed"), 1, closed) + |> process.select_record(atom.create("ssl_closed"), 1, closed) + |> process.select_record(atom.create("tcp_error"), 2, failed) + |> process.select_record(atom.create("ssl_error"), 2, failed) + |> process.select_record(atom.create("tcp_passive"), 1, exhausted) + |> process.select_record(atom.create("ssl_passive"), 1, exhausted) +} + +fn packet(record: dynamic.Dynamic) -> Received(user_message) { + Packet(socket_payload(record)) +} + +fn closed(_record: dynamic.Dynamic) -> Received(user_message) { + Closed +} + +fn failed(record: dynamic.Dynamic) -> Received(user_message) { + socket_error_reason(record) + |> socket.reason_to_string + |> Failed +} + +fn exhausted(_record: dynamic.Dynamic) -> Received(user_message) { + Exhausted +} + +@external(erlang, "http1_ffi", "socket_error_reason") +fn socket_error_reason(record: dynamic.Dynamic) -> socket.SocketReason + +@external(erlang, "websocket_ffi", "socket_payload") +fn socket_payload(record: dynamic.Dynamic) -> BitArray + +@external(erlang, "ewe_ffi", "identity") +fn unsafe_to_string(payload: BitArray) -> String + +@external(erlang, "ewe_ffi", "identity") +fn bit_array_from_string(text: String) -> BitArray diff --git a/src/ewe/internal/websocket.gleam b/src/ewe/internal/websocket.gleam new file mode 100644 index 0000000..6aaf149 --- /dev/null +++ b/src/ewe/internal/websocket.gleam @@ -0,0 +1,18 @@ +import ewe/internal/connection +import gleam/erlang/process +import gleam/option + +/// What a protocol's socket loop does once the handler has seen a message. +pub type Step(user_state, user_message) { + Proceed( + user_state: user_state, + messages: option.Option(process.Selector(user_message)), + ) + Halt(connection.Outcome) +} + +pub type Message(user_message) { + TextFrame(text: String) + BinaryFrame(data: BitArray) + UserMessage(message: user_message) +} diff --git a/src/ewe/internal/websocket_ffi.erl b/src/ewe/internal/websocket_ffi.erl new file mode 100644 index 0000000..84f2eea --- /dev/null +++ b/src/ewe/internal/websocket_ffi.erl @@ -0,0 +1,7 @@ +-module(websocket_ffi). + +-export([socket_payload/1]). + +%% Payload carried by a `{tcp, Socket, Data}` message. +socket_payload({_Tag, _Socket, Data}) -> + Data. diff --git a/test/ewe/internal/http1/parser_test.gleam b/test/ewe/internal/http1/parser_test.gleam index 362fada..7e68dd2 100644 --- a/test/ewe/internal/http1/parser_test.gleam +++ b/test/ewe/internal/http1/parser_test.gleam @@ -365,7 +365,8 @@ pub fn websocket_upgrade_requested_test() { let assert Ok(parser.Complete(_head, metadata, _remaining)) = parse(buffer) - assert metadata.upgrade == Some("websocket") + assert metadata.upgrade + == Some(http1.WebsocketUpgrade(key: None, version: None, extensions: None)) } pub fn upgrade_token_case_insensitive_test() { @@ -375,7 +376,8 @@ pub fn upgrade_token_case_insensitive_test() { let assert Ok(parser.Complete(_head, metadata, _remaining)) = parse(buffer) - assert metadata.upgrade == Some("websocket") + assert metadata.upgrade + == Some(http1.WebsocketUpgrade(key: None, version: None, extensions: None)) } pub fn upgrade_among_multiple_connection_tokens_test() { @@ -385,7 +387,7 @@ pub fn upgrade_among_multiple_connection_tokens_test() { let assert Ok(parser.Complete(_head, metadata, _remaining)) = parse(buffer) - assert metadata.upgrade == Some("h2c") + assert metadata.upgrade == Some(http1.OtherUpgrade("h2c")) } pub fn upgrade_header_without_connection_token_ignored_test() { @@ -406,7 +408,8 @@ pub fn upgrade_header_before_connection_header_test() { let assert Ok(parser.Complete(_head, metadata, _remaining)) = parse(buffer) - assert metadata.upgrade == Some("websocket") + assert metadata.upgrade + == Some(http1.WebsocketUpgrade(key: None, version: None, extensions: None)) as "order of Upgrade vs. Connection headers shouldn't matter" } @@ -416,3 +419,29 @@ pub fn no_upgrade_requested_test() { assert metadata.upgrade == None } + +pub fn websocket_handshake_fields_collected_test() { + let buffer = << + "GET /ws HTTP/1.1\r\nHost: example.com\r\nConnection: Upgrade\r\nUpgrade: websocket\r\nSec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==\r\nSec-WebSocket-Version: 13\r\nSec-WebSocket-Extensions: Permessage-Deflate; client_max_window_bits\r\n\r\n":utf8, + >> + + let assert Ok(parser.Complete(_head, metadata, _remaining)) = parse(buffer) + + assert metadata.upgrade + == Some(http1.WebsocketUpgrade( + // Base64 is case sensitive, so the key alone keeps the case it arrived in. + key: Some("dGhlIHNhbXBsZSBub25jZQ=="), + version: Some("13"), + extensions: Some("permessage-deflate; client_max_window_bits"), + )) +} + +pub fn websocket_headers_without_an_upgrade_are_ignored_test() { + let buffer = << + "GET /ws HTTP/1.1\r\nHost: example.com\r\nSec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==\r\nSec-WebSocket-Version: 13\r\n\r\n":utf8, + >> + + let assert Ok(parser.Complete(_head, metadata, _remaining)) = parse(buffer) + + assert metadata.upgrade == None +} -- 2.51.2