WebSocket permessage-deflate compression (RFC 7692) (#853)

Project.toml

@@ -14,13 +14,15 @@ Reseau = "802f3686-a58f-41ce-bb0c-3c43c75bba36"
 SHA = "ea8e919c-243c-51af-8825-aaa63cd721ce"
 URIs = "5c2747f8-b7ea-4ff2-ba2e-563bfd36b1d4"
 UUIDs = "cf7118a7-6976-5b1a-9a39-7adc72f591a4"
+Zlib_jll = "83775a58-1f1d-513f-b197-d71354ab007a"
 
 [compat]
 CodecZlib = "0.7"
 EnumX = "1"
 PrecompileTools = "1.2.1"
 Reseau = "1.3"
 URIs = "1.6.1"
+Zlib_jll = "1.2.12"
 julia = "1.10"
 
 [extras]

docs/src/guides/protocols.md

@@ -58,6 +58,34 @@ public docstrings.
 - `read_idle_timeout`
 - `write_idle_timeout`
 
+### Message compression (permessage-deflate)
+
+HTTP.jl supports the WebSocket permessage-deflate extension ([RFC 7692](https://www.rfc-editor.org/rfc/rfc7692)),
+which DEFLATE-compresses each message. It is **opt-in on both ends** via
+`compress = true` and is negotiated during the handshake — if either side
+declines, the connection transparently falls back to uncompressed frames.
+
+```julia
+# server advertises permessage-deflate; clients may negotiate it
+server = HTTP.WebSockets.listen!("127.0.0.1", 0; listenany = true, compress = true) do ws
+    for msg in ws
+        HTTP.WebSockets.send(ws, msg)
+    end
+end
+
+# client offers compression
+HTTP.WebSockets.open("ws://" * HTTP.WebSockets.server_addr(server); compress = true) do ws
+    HTTP.WebSockets.send(ws, repeat("compress me ", 1000))  # sent compressed
+    HTTP.WebSockets.receive(ws)
+end
+```
+
+`compress` is also accepted by `HTTP.WebSockets.upgrade` for servers that mix
+HTTP and WebSocket routes. Compression is most beneficial for larger, repetitive
+text/JSON payloads; tiny or already-compressed (binary/media) messages gain
+little. Decompressed message size is bounded by `maxframesize`, guarding against
+decompression bombs.
+
 ## HTTP/2 Support
 
 `HTTP.jl` supports HTTP/2 through the normal client and server APIs.

src/http_websocket_codec.jl

@@ -201,6 +201,13 @@ mutable struct WsDecoder
     expecting_continuation::Bool
     fragment_opcode::UInt8
     text_fragment_payload::Vector{UInt8}
+    # permessage-deflate (RFC 7692): `pmce_enabled` is set when the extension was
+    # negotiated (so RSV1 is allowed); `message_compressed` carries the current
+    # message's RSV1 across its fragments. When a message is compressed the
+    # decoder defers UTF-8 validation — the bytes are compressed until the upper
+    # layer inflates the reassembled message.
+    pmce_enabled::Bool
+    message_compressed::Bool
     # Reused result vector for ws_decoder_process! — contents are valid only
     # until the next process! call on this decoder.
     frames_scratch::Vector{WsFrame{Vector{UInt8}}}
@@ -223,6 +230,8 @@ function ws_decoder_new()::WsDecoder
         false,
         UInt8(0),
         UInt8[],
+        false,
+        false,
         WsFrame{Vector{UInt8}}[],
     )
 end
@@ -266,9 +275,13 @@ function ws_decoder_process!(f::Union{Nothing,Function}, dec::WsDecoder, data::A
             pos += 1
             dec.fin = (b & 0x80) != 0
             dec.rsv = ((b & 0x40) != 0, (b & 0x20) != 0, (b & 0x10) != 0)
-            (dec.rsv[1] || dec.rsv[2] || dec.rsv[3]) && _ws_throw_protocol_error("unexpected websocket RSV bits without negotiated extensions")
             dec.opcode = b & 0x0f
             dec.opcode in (0x00, 0x01, 0x02, 0x08, 0x09, 0x0a) || _ws_throw_protocol_error("invalid websocket opcode")
+            # RSV2/RSV3 are never valid (no negotiated extension uses them). RSV1
+            # is the permessage-deflate per-message-compressed bit: valid only
+            # when negotiated, and only on the first frame of a data message.
+            (dec.rsv[2] || dec.rsv[3]) && _ws_throw_protocol_error("unexpected websocket RSV bits without negotiated extensions")
+            dec.rsv[1] && !(dec.pmce_enabled && dec.opcode in (0x01, 0x02)) && _ws_throw_protocol_error("unexpected websocket RSV1 bit")
             is_control = ws_is_control_frame(dec.opcode)
             is_control && !dec.fin && _ws_throw_protocol_error("control frames must not be fragmented")
             if !is_control
@@ -278,6 +291,8 @@ function ws_decoder_process!(f::Union{Nothing,Function}, dec::WsDecoder, data::A
                     dec.expecting_continuation && _ws_throw_protocol_error("unexpected new data frame while fragmented message is open")
                     empty!(dec.text_fragment_payload)
                     dec.fragment_opcode = dec.fin ? UInt8(0) : dec.opcode
+                    # RSV1 on the first data frame marks the whole message compressed.
+                    dec.message_compressed = dec.rsv[1]
                 end
                 dec.expecting_continuation = !dec.fin
             end
@@ -366,16 +381,19 @@ function ws_decoder_process!(f::Union{Nothing,Function}, dec::WsDecoder, data::A
             if dec.opcode == UInt8(WsOpcode.CLOSE)
                 ws_validate_close_payload(dec.payload_buf)
             elseif dec.opcode == UInt8(WsOpcode.TEXT)
-                if dec.fin
+                # Compressed text is validated upstream, after decompression.
+                if dec.message_compressed
+                    # nothing to validate here
+                elseif dec.fin
                     isvalid(String, dec.payload_buf) || _ws_throw_invalid_payload("invalid UTF-8 text frame payload")
                 else
                     append!(dec.text_fragment_payload, dec.payload_buf)
                 end
             elseif dec.opcode == UInt8(WsOpcode.CONTINUATION)
                 if dec.fragment_opcode == UInt8(WsOpcode.TEXT)
-                    append!(dec.text_fragment_payload, dec.payload_buf)
+                    dec.message_compressed || append!(dec.text_fragment_payload, dec.payload_buf)
                     if dec.fin
-                        if !isvalid(String, dec.text_fragment_payload)
+                        if !dec.message_compressed && !isvalid(String, dec.text_fragment_payload)
                             dec.expecting_continuation = false
                             dec.fragment_opcode = UInt8(0)
                             empty!(dec.text_fragment_payload)
@@ -470,13 +488,17 @@ mutable struct WSConn
     outgoing_spare::Vector{UInt8}
     max_incoming_payload_length::UInt64
     incoming_message_payload_total::UInt64
+    # permessage-deflate state, or nothing when the extension is not negotiated.
+    pmce::Union{Nothing,PMCEContext}
 end
 
 function WSConn(;
     is_client::Bool=true,
     max_incoming_payload_length::UInt64=UInt64(0),
+    pmce::Union{Nothing,PMCEContext}=nothing,
 )::WSConn
     decoder = ws_decoder_new()
+    decoder.pmce_enabled = pmce !== nothing
     return WSConn(
         is_client,
         true,
@@ -488,6 +510,7 @@ function WSConn(;
         UInt8[],
         max_incoming_payload_length,
         UInt64(0),
+        pmce,
     )
 end
 
@@ -547,7 +570,7 @@ function _ws_append_frame!(out::Vector{UInt8}, frame::WsFrame)::Nothing
     return nothing
 end
 
-function ws_send_frame!(ws::WSConn, opcode::UInt8, payload::AbstractVector{UInt8}; fin::Bool=true)::Nothing
+function ws_send_frame!(ws::WSConn, opcode::UInt8, payload::AbstractVector{UInt8}; fin::Bool=true, rsv1::Bool=false)::Nothing
     ws.is_open || throw(ProtocolError("websocket connection is closed"))
     if ws_is_control_frame(opcode)
         !fin && throw(ArgumentError("control frames must not be fragmented"))
@@ -565,6 +588,7 @@ function ws_send_frame!(ws::WSConn, opcode::UInt8, payload::AbstractVector{UInt8
         fin=fin,
         masked=ws.is_client,
         masking_key=masking_key,
+        rsv=(rsv1, false, false),
     )
     @lock ws.out_lock begin
         _ws_append_frame!(ws.outgoing, frame)