diff --git a/Cargo.lock b/Cargo.lock index a040d5900..9379a8bd0 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -377,6 +377,7 @@ dependencies = [ "chrono", "config", "dirs", + "flate2", "futures", "hex", "hmac", diff --git a/Cargo.toml b/Cargo.toml index cbacd9c3d..ff1bcf6cb 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -19,6 +19,7 @@ chrono = { version = "0.4.45", features = ["serde", "clock"] } config = "0.15.23" dirs = "5.0.1" futures = "0.3.32" +flate2 = "1.1.9" hex = "0.4.3" hmac = "0.13.0" indexmap = { version = "2.0", features = ["serde"] } diff --git a/docs/OPENAI_CHAIN_PARITY_REAUDIT.md b/docs/OPENAI_CHAIN_PARITY_REAUDIT.md index bea4a01fa..6b1079acd 100644 --- a/docs/OPENAI_CHAIN_PARITY_REAUDIT.md +++ b/docs/OPENAI_CHAIN_PARITY_REAUDIT.md @@ -26,7 +26,7 @@ Rust 仓库:`/home/zyy/Codes/codex-proxy-rs` 本轮 10 个 OpenAI/Codex 链路点已全部复核并写入账本。结论:当前不能声明与原版 100% 一致;除“代码证据索引与审计边界”外,其余链路均为 `部分对齐`。 最高优先级未对齐项: -- 安全模拟仍缺 TLS 指纹证明,以及 WS permessage-deflate 被上游接受后的压缩 frame 解码证明。 +- 安全模拟仍缺 TLS 指纹证明;WS permessage-deflate offer、协商响应和服务端压缩 frame 解码已补齐测试与实现。 - `/v1/responses` 请求构造仍存在非 `/v1` alias route、`use_websocket:false` 本地扩展等差异;`/v1/responses/review` 强制 review subagent 和 `/v1/responses/compact` 已补齐。 - response 转换和错误恢复仍缺最终错误外壳对齐;tuple schema request conversion/reconvert、streaming premature close 合成、reasoning replay、implicit resume failure restore、`response.failed`/`error` 的 429/402/403 分类、账号 fallback、`previous_response_not_found`/unanswered function call strip-and-retry、5xx same-account retry、CF path-block 404、model unsupported fallback、401 token invalid fallback 已补齐。 - session affinity 的显式续链、implicit resume 主路径、implicit resume failure restore、reasoning replay cache、function call continuation 预检、variant_hash 基础算法、variant identity 和 OpenAI chat `user` conversation identity 已接入。 @@ -116,10 +116,9 @@ Rust 证据: 未完全对齐: - Rust 当前没有测试证明真实传输层 TLS/HTTP 指纹与原版 native transport 行为完全一致。IP 代理/VPN 明确非目标,但 TLS/header 指纹仍属于安全模拟链路,不能在未验证时声明 100%。 -- WebSocket 已按原版 `ws@8.19.0` 发出 `Sec-WebSocket-Extensions: permessage-deflate; client_max_window_bits`,但 Rust 后续 frame 解码仍由 `tungstenite` 负责;若上游真的协商并发送压缩 frame,当前还没有真实上游证据证明不会触发 RSV/压缩解码问题。 缺口/后续动作: -- 若目标是安全模拟 100%,需要继续处理 TLS 指纹和 WS compression negotiation 的真实上游验证;TLS 是否要追到 native transport 级别需要单独决策。 +- 若目标是安全模拟 100%,需要继续处理 TLS 指纹真实上游验证;TLS 是否要追到 native transport 级别需要单独决策。 ## 3. OAuth/device/refresh 链路 @@ -298,6 +297,7 @@ Rust 证据: - 本轮修正后,WS handshake header profile 已独立于 HTTP SSE profile,不再携带 `Content-Type` 与 `Accept: text/event-stream`;`tests/codex_gateway/websocket.rs` 已覆盖该行为。 - 本轮修正后,`src/codex/gateway/transport/websocket/opening.rs` 不再使用 `tokio-tungstenite` 默认 `HeaderMap` opening 序列化,而是显式生成原版 `ws@8.19.0` opening request:`Sec-WebSocket-Extensions` 位于 `Sec-WebSocket-Key` 前,业务 headers 按原版对象插入顺序输出,并恢复 `Authorization`、`ChatGPT-Account-Id`、`User-Agent`、`Accept-Encoding`、`Accept-Language`、`OpenAI-Beta`、`Cookie` 等原版大小写。 - `tests/codex_gateway/websocket.rs::websocket_handshake_should_offer_original_permessage_deflate_extension` 通过 raw TCP 断言 WS opening request 的关键 header 顺序、大小写和 permessage-deflate offer 与原版抓包一致。 +- 本轮修正后,`src/codex/gateway/transport/websocket/deflate.rs` 在服务端接受 `Sec-WebSocket-Extensions: permessage-deflate` 时,会在底层流上解码 server-to-client RSV1 text/binary frame,再交给 `tungstenite` 处理;未协商时保持原始流直通。`tests/codex_gateway/websocket.rs::websocket_should_decode_permessage_deflate_response_frame_when_server_accepts_extension` 先以 `NonZeroReservedBits` 失败,再修复通过,覆盖服务端接受扩展并返回压缩 `response.completed` frame。 - `src/codex/gateway/transport/websocket/mod.rs:230-328` pooled/reused WS stale failure 会重试一次 one-shot;内部 `codex.rate_limits` 会转为 rate-limit headers;terminal event 转为 SSE chunk。 - 本轮修正后,`src/codex/gateway/transport/websocket/mod.rs` 的 one-shot WebSocket handshake 使用 20 秒 open timeout;`connect_websocket_with_timeout_should_fail_when_handshake_stalls` 覆盖握手停滞时返回 timeout。 - `src/codex/gateway/transport/websocket/pool.rs:58-76` 默认 `max_age=55min`、`max_per_account=8`、`maintenance_interval=25s`、`ping_interval=25s`、`ping_timeout=5s`、`liveness_timeout=62.5s`。 @@ -319,6 +319,7 @@ Rust 证据: - WS URL、`response.create` frame 主字段、tool 默认值、service tier、prompt cache key、client metadata、internal rate-limit event、terminal event SSE 包装均已基本对齐。 - WS handshake headers 已去掉原版没有的 `Content-Type` 与 `Accept: text/event-stream`。 - WS opening request 的关键 wire 顺序、大小写和 `Sec-WebSocket-Extensions` offer 已按原版 `ws@8.19.0` 对齐,并由 raw TCP 测试覆盖。 +- WS permessage-deflate 协商后的 server-to-client 压缩 text/binary frame 已支持解码,避免上游接受扩展后因 RSV1 frame 触发 `tungstenite` `NonZeroReservedBits`。 - one-shot WebSocket handshake 已增加原版同等 20 秒 open timeout;普通无历史请求超时后仍可按原版策略降级 HTTP SSE,`previous_response_id` required WS 不降级。 - 连接池核心状态机基本对齐:55 分钟 max age、每账号最多 8 条、严格单 in-flight、busy/cap bypass、stale reuse 一次 fresh retry、idle keepalive、liveness/GC、shutdown、账号状态变更驱逐。 - WebSocket pool 配置面已对齐原版:配置文件显式提供 `ws_pool.enabled/max_age_ms/max_per_account`,运行时按配置启停复用和调整 max age/account cap。 @@ -328,10 +329,9 @@ Rust 证据: 未完全对齐: - pool key 表达不完全一致。原版 key 是 `${entryId}:${rawConversationId}`;Rust key 是 `(base_url, account_id, accountScopedConversationId)`。行为上通过测试证明同 conversation 会复用,但若后续要与原版日志/配置/诊断完全一致,命名与 key 语义仍不同。 - Rust pool maintenance 是集中 sweep ping idle connection;原版是每条 `PersistentWs` 自带 interval,busy 时跳过 ping。默认行为接近,但调度模型不是逐字等价。 -- Rust 已发出原版 permessage-deflate offer,但后续 frame 解码仍由 `tungstenite` 负责;需要真实上游或压缩 frame fixture 证明 OpenAI 不会返回需要解压的 RSV1 frame,或补齐压缩解码能力。 缺口/后续动作: -- 增加 WS compression negotiation 验证:覆盖上游接受 `Sec-WebSocket-Extensions` 后的响应 header 与 frame 行为。 +- 若需要更强证明,可增加真实 OpenAI 上游抓包验证,确认当前 Codex 后端是否实际返回压缩 frame;本地压缩 frame fixture 已覆盖客户端能力。 ## 7. response/SSE/WS frame 转换链路 diff --git a/src/codex/gateway/transport/websocket/deflate.rs b/src/codex/gateway/transport/websocket/deflate.rs new file mode 100644 index 000000000..47ab62187 --- /dev/null +++ b/src/codex/gateway/transport/websocket/deflate.rs @@ -0,0 +1,273 @@ +use std::{ + collections::VecDeque, + io, + pin::Pin, + task::{Context, Poll}, +}; + +use flate2::{Decompress, FlushDecompress, Status}; +use tokio::io::{AsyncRead, AsyncWrite, ReadBuf}; + +const SERVER_FRAME_CHUNK_BYTES: usize = 8192; +const PERMESSAGE_DEFLATE_TAIL: [u8; 4] = [0x00, 0x00, 0xff, 0xff]; +const OPCODE_CONTINUATION: u8 = 0x0; +const OPCODE_TEXT: u8 = 0x1; +const OPCODE_BINARY: u8 = 0x2; +pub(super) struct PerMessageDeflateStream { + inner: S, + enabled: bool, + raw_input: Vec, + decoded_input: VecDeque, + compressed_message: Option, +} + +struct CompressedMessage { + opcode: u8, + payload: Vec, +} + +struct FrameParts { + fin: bool, + rsv1: bool, + opcode: u8, + masked: bool, + payload_offset: usize, + payload_len: usize, + frame_len: usize, +} + +impl PerMessageDeflateStream { + pub(super) fn new(inner: S, enabled: bool, preloaded: Vec) -> Self { + Self { + inner, + enabled, + raw_input: preloaded, + decoded_input: VecDeque::new(), + compressed_message: None, + } + } +} + +impl AsyncRead for PerMessageDeflateStream +where + S: AsyncRead + AsyncWrite + Unpin, +{ + fn poll_read( + self: Pin<&mut Self>, + cx: &mut Context<'_>, + buffer: &mut ReadBuf<'_>, + ) -> Poll> { + let this = self.get_mut(); + if !this.enabled { + return Pin::new(&mut this.inner).poll_read(cx, buffer); + } + + loop { + if copy_decoded_input(&mut this.decoded_input, buffer) { + return Poll::Ready(Ok(())); + } + + if this.rewrite_next_frame()? { + continue; + } + + let mut chunk = [0_u8; SERVER_FRAME_CHUNK_BYTES]; + let mut read_buffer = ReadBuf::new(&mut chunk); + match Pin::new(&mut this.inner).poll_read(cx, &mut read_buffer) { + Poll::Pending => return Poll::Pending, + Poll::Ready(Err(error)) => return Poll::Ready(Err(error)), + Poll::Ready(Ok(())) => { + let filled = read_buffer.filled(); + if filled.is_empty() { + return Poll::Ready(Ok(())); + } + this.raw_input.extend_from_slice(filled); + } + } + } + } +} + +impl AsyncWrite for PerMessageDeflateStream +where + S: AsyncRead + AsyncWrite + Unpin, +{ + fn poll_write( + self: Pin<&mut Self>, + cx: &mut Context<'_>, + data: &[u8], + ) -> Poll> { + Pin::new(&mut self.get_mut().inner).poll_write(cx, data) + } + + fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + Pin::new(&mut self.get_mut().inner).poll_flush(cx) + } + + fn poll_shutdown(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + Pin::new(&mut self.get_mut().inner).poll_shutdown(cx) + } +} + +impl PerMessageDeflateStream { + fn rewrite_next_frame(&mut self) -> io::Result { + let Some(frame) = parse_frame_parts(&self.raw_input)? else { + return Ok(false); + }; + let raw_frame = self.raw_input.drain(..frame.frame_len).collect::>(); + let payload = &raw_frame[frame.payload_offset..frame.payload_offset + frame.payload_len]; + + if frame.masked { + self.decoded_input.extend(raw_frame); + return Ok(true); + } + + if frame.rsv1 && is_data_frame(frame.opcode) { + if frame.fin { + let inflated = inflate_message(payload)?; + self.decoded_input + .extend(encode_server_frame(frame.opcode, false, &inflated)); + } else { + self.compressed_message = Some(CompressedMessage { + opcode: frame.opcode, + payload: payload.to_vec(), + }); + } + return Ok(true); + } + + if frame.opcode == OPCODE_CONTINUATION { + if let Some(message) = self.compressed_message.as_mut() { + message.payload.extend_from_slice(payload); + if frame.fin { + let message = self.compressed_message.take().ok_or_else(|| { + io::Error::new(io::ErrorKind::InvalidData, "missing compressed message") + })?; + let inflated = inflate_message(&message.payload)?; + self.decoded_input.extend(encode_server_frame( + message.opcode, + false, + &inflated, + )); + } + return Ok(true); + } + } + + self.decoded_input.extend(raw_frame); + Ok(true) + } +} + +fn copy_decoded_input(decoded_input: &mut VecDeque, buffer: &mut ReadBuf<'_>) -> bool { + if decoded_input.is_empty() || buffer.remaining() == 0 { + return false; + } + + let copy_len = decoded_input.len().min(buffer.remaining()); + let bytes = decoded_input.drain(..copy_len).collect::>(); + buffer.put_slice(&bytes); + true +} + +fn parse_frame_parts(bytes: &[u8]) -> io::Result> { + if bytes.len() < 2 { + return Ok(None); + } + + let first = bytes[0]; + let second = bytes[1]; + let mut offset = 2; + let payload_len = match second & 0x7f { + len @ 0..=125 => usize::from(len), + 126 => { + if bytes.len() < offset + 2 { + return Ok(None); + } + let len = u16::from_be_bytes([bytes[offset], bytes[offset + 1]]); + offset += 2; + usize::from(len) + } + 127 => { + if bytes.len() < offset + 8 { + return Ok(None); + } + let len = u64::from_be_bytes([ + bytes[offset], + bytes[offset + 1], + bytes[offset + 2], + bytes[offset + 3], + bytes[offset + 4], + bytes[offset + 5], + bytes[offset + 6], + bytes[offset + 7], + ]); + offset += 8; + usize::try_from(len).map_err(|_| { + io::Error::new(io::ErrorKind::InvalidData, "websocket frame is too large") + })? + } + _ => unreachable!("websocket length marker is masked to 7 bits"), + }; + let masked = second & 0x80 != 0; + if masked { + offset += 4; + } + let frame_len = offset + .checked_add(payload_len) + .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidData, "websocket frame overflow"))?; + if bytes.len() < frame_len { + return Ok(None); + } + + Ok(Some(FrameParts { + fin: first & 0x80 != 0, + rsv1: first & 0x40 != 0, + opcode: first & 0x0f, + masked, + payload_offset: offset, + payload_len, + frame_len, + })) +} + +fn inflate_message(payload: &[u8]) -> io::Result> { + let mut compressed = Vec::with_capacity(payload.len() + PERMESSAGE_DEFLATE_TAIL.len()); + compressed.extend_from_slice(payload); + compressed.extend_from_slice(&PERMESSAGE_DEFLATE_TAIL); + + let mut decompressor = Decompress::new(false); + let mut inflated = Vec::with_capacity(payload.len().saturating_mul(4).max(1024)); + match decompressor.decompress_vec(&compressed, &mut inflated, FlushDecompress::Sync) { + Ok(Status::Ok | Status::StreamEnd) => Ok(inflated), + Ok(Status::BufError) if !inflated.is_empty() => Ok(inflated), + Ok(Status::BufError) => Err(io::Error::new( + io::ErrorKind::InvalidData, + "permessage-deflate frame produced no output", + )), + Err(error) => Err(io::Error::new(io::ErrorKind::InvalidData, error)), + } +} + +fn encode_server_frame(opcode: u8, rsv1: bool, payload: &[u8]) -> Vec { + let mut frame = Vec::new(); + let rsv1_bit = if rsv1 { 0x40 } else { 0 }; + frame.push(0x80 | rsv1_bit | opcode); + match payload.len() { + len @ 0..=125 => frame.push(len as u8), + len @ 126..=65_535 => { + frame.push(126); + frame.extend_from_slice(&(len as u16).to_be_bytes()); + } + len => { + frame.push(127); + frame.extend_from_slice(&(len as u64).to_be_bytes()); + } + } + frame.extend_from_slice(payload); + frame +} + +fn is_data_frame(opcode: u8) -> bool { + matches!(opcode, OPCODE_TEXT | OPCODE_BINARY) +} diff --git a/src/codex/gateway/transport/websocket/mod.rs b/src/codex/gateway/transport/websocket/mod.rs index 44d3d1bce..ce9f55e53 100644 --- a/src/codex/gateway/transport/websocket/mod.rs +++ b/src/codex/gateway/transport/websocket/mod.rs @@ -19,6 +19,7 @@ use crate::codex::gateway::transport::{ }; mod codec; +mod deflate; mod opening; mod pool; diff --git a/src/codex/gateway/transport/websocket/opening.rs b/src/codex/gateway/transport/websocket/opening.rs index 8390cf96a..f864e5f40 100644 --- a/src/codex/gateway/transport/websocket/opening.rs +++ b/src/codex/gateway/transport/websocket/opening.rs @@ -21,6 +21,7 @@ use tokio_tungstenite::{ MaybeTlsStream, WebSocketStream, }; +use super::deflate::PerMessageDeflateStream; use super::pool::CodexWsStream; const MAX_OPENING_RESPONSE_BYTES: usize = 64 * 1024; @@ -74,9 +75,11 @@ pub(super) async fn connect_with_original_opening_handshake( return Err(WsError::Http(Box::new(response))); } verify_accept_key(&request, &response)?; + let permessage_deflate = response_accepts_permessage_deflate(&response); + let stream = PerMessageDeflateStream::new(stream, permessage_deflate, buffered); let websocket = - WebSocketStream::from_partially_read(stream, buffered, Role::Client, None).await; + WebSocketStream::from_partially_read(stream, Vec::new(), Role::Client, None).await; Ok((websocket, response)) } @@ -372,3 +375,16 @@ fn verify_accept_key(request: &WsRequest<()>, response: &WsResponse) -> Result<( Err(ProtocolError::SecWebSocketAcceptKeyMismatch.into()) } } + +fn response_accepts_permessage_deflate(response: &WsResponse) -> bool { + response + .headers() + .get_all("sec-websocket-extensions") + .iter() + .filter_map(|value| value.to_str().ok()) + .any(|value| { + value + .split(',') + .any(|extension| extension.trim_start().starts_with("permessage-deflate")) + }) +} diff --git a/src/codex/gateway/transport/websocket/pool.rs b/src/codex/gateway/transport/websocket/pool.rs index 4f6c1f530..208c9809e 100644 --- a/src/codex/gateway/transport/websocket/pool.rs +++ b/src/codex/gateway/transport/websocket/pool.rs @@ -12,6 +12,7 @@ use tokio_tungstenite::{ }; use super::codec::{rate_limit_headers, set_cookie_headers, turn_state}; +use super::deflate::PerMessageDeflateStream; #[derive(Debug, Clone, PartialEq, Eq, Hash)] pub struct CodexWebSocketPoolKey { @@ -356,7 +357,7 @@ struct WebSocketPoolState { shutting_down: bool, } -pub(super) type CodexWsStream = WebSocketStream>; +pub(super) type CodexWsStream = WebSocketStream>>; #[derive(Debug, Clone, PartialEq, Eq)] pub(super) struct CodexWebSocketConnectionMetadata { diff --git a/tests/codex_gateway/websocket.rs b/tests/codex_gateway/websocket.rs index 2d8bba3f0..1e2a2af22 100644 --- a/tests/codex_gateway/websocket.rs +++ b/tests/codex_gateway/websocket.rs @@ -1,5 +1,6 @@ use std::{sync::Arc, time::Duration}; +use flate2::{Compress, Compression, FlushCompress}; use futures::{SinkExt, StreamExt}; use serde_json::{json, Value}; use tokio::{ @@ -11,6 +12,7 @@ use tokio::{ use tokio_tungstenite::{ accept_async, accept_hdr_async, tungstenite::{ + handshake::derive_accept_key, handshake::server::{Request as WsRequest, Response as WsResponse}, Message, }, @@ -492,6 +494,49 @@ async fn websocket_handshake_should_offer_original_permessage_deflate_extension( ); } +#[tokio::test] +async fn websocket_should_decode_permessage_deflate_response_frame_when_server_accepts_extension() { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + let server = tokio::spawn(async move { + let (mut stream, _) = listener.accept().await.unwrap(); + let request = read_http_upgrade_request(&mut stream).await; + let accept_key = websocket_accept_key(&request); + let response = format!( + "HTTP/1.1 101 Switching Protocols\r\n\ + Connection: Upgrade\r\n\ + Upgrade: websocket\r\n\ + Sec-WebSocket-Accept: {accept_key}\r\n\ + Sec-WebSocket-Extensions: permessage-deflate\r\n\ + \r\n" + ); + stream.write_all(response.as_bytes()).await.unwrap(); + let _client_request = read_client_websocket_frame(&mut stream).await; + let payload = websocket_completed_response("resp_ws_deflate", 5, 2); + let compressed_frame = compressed_server_text_frame(&payload); + stream.write_all(&compressed_frame).await.unwrap(); + sleep(Duration::from_millis(50)).await; + }); + + let client = CodexBackendClient::new( + build_reqwest_client(false).unwrap(), + format!("http://{addr}"), + codex_proxy_rs::codex::gateway::fingerprint::model::Fingerprint::default_for_tests(), + ); + let mut request = base_request(); + request.previous_response_id = Some("resp_deflate_prev".to_string()); + + let response = client + .create_response(&request, request_context("req_ws_deflate", None)) + .await + .unwrap(); + + server.await.unwrap(); + assert!(response.body.contains("event: response.completed")); + assert!(response.body.contains("\"id\":\"resp_ws_deflate\"")); + assert_eq!(response.usage.unwrap().input_tokens, 5); +} + #[tokio::test] #[expect( clippy::result_large_err, @@ -1038,6 +1083,72 @@ async fn read_http_upgrade_request(stream: &mut tokio::net::TcpStream) -> String String::from_utf8(request).unwrap() } +async fn read_client_websocket_frame(stream: &mut tokio::net::TcpStream) -> Vec { + let mut header = [0_u8; 2]; + stream.read_exact(&mut header).await.unwrap(); + let mut payload_len = u64::from(header[1] & 0x7f); + if payload_len == 126 { + let mut extended = [0_u8; 2]; + stream.read_exact(&mut extended).await.unwrap(); + payload_len = u64::from(u16::from_be_bytes(extended)); + } else if payload_len == 127 { + let mut extended = [0_u8; 8]; + stream.read_exact(&mut extended).await.unwrap(); + payload_len = u64::from_be_bytes(extended); + } + let mut mask = [0_u8; 4]; + if header[1] & 0x80 != 0 { + stream.read_exact(&mut mask).await.unwrap(); + } + let mut payload = vec![0_u8; payload_len as usize]; + stream.read_exact(&mut payload).await.unwrap(); + if header[1] & 0x80 != 0 { + for (index, byte) in payload.iter_mut().enumerate() { + *byte ^= mask[index % mask.len()]; + } + } + payload +} + +fn websocket_accept_key(raw_request: &str) -> String { + let key = raw_request + .lines() + .find_map(|line| line.strip_prefix("Sec-WebSocket-Key: ")) + .expect("raw websocket request should contain Sec-WebSocket-Key"); + derive_accept_key(key.as_bytes()) +} + +fn compressed_server_text_frame(text: &str) -> Vec { + let mut compressor = Compress::new(Compression::fast(), false); + let mut payload = Vec::with_capacity(text.len() + 32); + compressor + .compress_vec(text.as_bytes(), &mut payload, FlushCompress::Sync) + .unwrap(); + if payload.ends_with(&[0x00, 0x00, 0xff, 0xff]) { + payload.truncate(payload.len() - 4); + } + encode_server_frame(0x1, true, &payload) +} + +fn encode_server_frame(opcode: u8, rsv1: bool, payload: &[u8]) -> Vec { + let mut frame = Vec::new(); + let first = 0x80 | if rsv1 { 0x40 } else { 0 } | opcode; + frame.push(first); + match payload.len() { + len @ 0..=125 => frame.push(len as u8), + len @ 126..=65_535 => { + frame.push(126); + frame.extend_from_slice(&(len as u16).to_be_bytes()); + } + len => { + frame.push(127); + frame.extend_from_slice(&(len as u64).to_be_bytes()); + } + } + frame.extend_from_slice(payload); + frame +} + fn assert_headers_appear_in_order(raw_request: &str, expected_headers: &[&str]) { let mut offset = 0; for expected in expected_headers {