Implement HTTP/2 framing, multiplexed streams, and flow control

+191 -20

crates/net/src/client.rs

··· 1 - //! High-level HTTP/1.1 client with connection pooling. 1 + //! High-level HTTP client with connection pooling and HTTP/2 support. 2 2 //! 3 - //! Brings together TCP, TLS 1.3, DNS, URL parsing, and HTTP message 4 - //! parsing into a single `HttpClient` that can fetch HTTP and HTTPS URLs. 3 + //! Brings together TCP, TLS 1.3, DNS, URL parsing, HTTP/1.1, and HTTP/2 4 + //! into a single `HttpClient` that can fetch HTTP and HTTPS URLs. 5 + //! When connecting over TLS, ALPN negotiation automatically selects HTTP/2 6 + //! if the server supports it, falling back to HTTP/1.1 otherwise. 5 7 6 8 use std::collections::HashMap; 7 9 use std::fmt; ··· 12 14 13 15 use crate::cookie::{CookieJar, RequestContext}; 14 16 use crate::http::{self, Headers, HttpResponse, Method}; 17 + use crate::http2::connection::Http2Connection; 18 + use crate::http2::frame::Http2Error; 15 19 use crate::tcp::{self, TcpConnection}; 16 20 use crate::tls::handshake::{self, HandshakeError, TlsStream}; 17 21 ··· 45 49 Http(http::HttpError), 46 50 /// Too many redirects. 47 51 TooManyRedirects, 52 + /// HTTP/2 protocol error. 53 + Http2(Http2Error), 48 54 /// Connection was closed unexpectedly. 49 55 ConnectionClosed, 50 56 /// I/O error. ··· 60 66 Self::Tls(e) => write!(f, "TLS error: {e}"), 61 67 Self::Http(e) => write!(f, "HTTP error: {e}"), 62 68 Self::TooManyRedirects => write!(f, "too many redirects"), 69 + Self::Http2(e) => write!(f, "HTTP/2 error: {e}"), 63 70 Self::ConnectionClosed => write!(f, "connection closed"), 64 71 Self::Io(e) => write!(f, "I/O error: {e}"), 65 72 } ··· 87 94 impl From<io::Error> for ClientError { 88 95 fn from(e: io::Error) -> Self { 89 96 Self::Io(e) 97 + } 98 + } 99 + 100 + impl From<Http2Error> for ClientError { 101 + fn from(e: Http2Error) -> Self { 102 + Self::Http2(e) 90 103 } 91 104 } 92 105 ··· 201 214 // HttpClient 202 215 // --------------------------------------------------------------------------- 203 216 204 - /// High-level HTTP/1.1 client with connection pooling, redirect following, and cookie jar. 217 + /// Key for HTTP/2 connection pooling (one multiplexed connection per origin). 218 + #[derive(Hash, Eq, PartialEq, Clone, Debug)] 219 + struct H2ConnectionKey { 220 + host: String, 221 + port: u16, 222 + } 223 + 224 + /// High-level HTTP client with connection pooling, HTTP/2 support, redirect following, and cookie jar. 205 225 pub struct HttpClient { 206 226 pool: ConnectionPool, 227 + h2_connections: HashMap<H2ConnectionKey, Http2Connection<TlsStream<TcpConnection>>>, 207 228 max_redirects: u32, 208 229 connect_timeout: Duration, 209 230 read_timeout: Duration, ··· 215 236 pub fn new() -> Self { 216 237 Self { 217 238 pool: ConnectionPool::new(DEFAULT_MAX_IDLE_TIME, DEFAULT_MAX_PER_HOST), 239 + h2_connections: HashMap::new(), 218 240 max_redirects: DEFAULT_MAX_REDIRECTS, 219 241 connect_timeout: DEFAULT_CONNECT_TIMEOUT, 220 242 read_timeout: DEFAULT_READ_TIMEOUT, ··· 338 360 } 339 361 } 340 362 363 + // Try HTTP/2 for TLS connections 364 + if is_tls { 365 + let h2_key = H2ConnectionKey { 366 + host: host.clone(), 367 + port, 368 + }; 369 + 370 + // Check if we have an existing HTTP/2 connection for this origin 371 + if self.h2_connections.contains_key(&h2_key) { 372 + let response = self.execute_h2_request( 373 + &h2_key, 374 + method, 375 + &path, 376 + &host, 377 + &merged_headers, 378 + body, 379 + url, 380 + ); 381 + 382 + match response { 383 + Ok(resp) => return Ok(resp), 384 + Err(ClientError::Http2(_)) => { 385 + // HTTP/2 connection failed — remove it and fall through to new connection 386 + self.h2_connections.remove(&h2_key); 387 + } 388 + Err(e) => return Err(e), 389 + } 390 + } 391 + 392 + // Try to establish a new connection with ALPN 393 + let tcp = TcpConnection::connect_timeout(&host, port, self.connect_timeout)?; 394 + let (tls, alpn) = handshake::connect_with_alpn(tcp, &host, &["h2", "http/1.1"])?; 395 + 396 + if alpn.as_deref() == Some("h2") { 397 + // HTTP/2 negotiated — create HTTP/2 connection 398 + let h2_conn = Http2Connection::new(tls)?; 399 + self.h2_connections.insert(h2_key.clone(), h2_conn); 400 + 401 + return self.execute_h2_request( 402 + &h2_key, 403 + method, 404 + &path, 405 + &host, 406 + &merged_headers, 407 + body, 408 + url, 409 + ); 410 + } 411 + 412 + // HTTP/1.1 over TLS — use normal path 413 + let conn = Connection::Tls(tls); 414 + return self.execute_h1_request(conn, method, &path, &host, &merged_headers, body, url); 415 + } 416 + 417 + // Plain HTTP (no TLS) — always HTTP/1.1 341 418 let key = ConnectionKey { 342 419 host: host.clone(), 343 420 port, 344 - is_tls, 421 + is_tls: false, 345 422 }; 346 423 347 - // Try to reuse a pooled connection, fall back to new connection 348 - let mut conn = match self.pool.take(&key) { 424 + let conn = match self.pool.take(&key) { 349 425 Some(conn) => conn, 350 - None => self.connect(&host, port, is_tls)?, 426 + None => { 427 + let tcp = TcpConnection::connect_timeout(&host, port, self.connect_timeout)?; 428 + Connection::Plain(tcp) 429 + } 351 430 }; 352 431 432 + self.execute_h1_request(conn, method, &path, &host, &merged_headers, body, url) 433 + } 434 + 435 + /// Execute an HTTP/1.1 request over the given connection. 436 + #[allow(clippy::too_many_arguments)] 437 + fn execute_h1_request( 438 + &mut self, 439 + mut conn: Connection, 440 + method: Method, 441 + path: &str, 442 + host: &str, 443 + headers: &Headers, 444 + body: Option<&[u8]>, 445 + url: &Url, 446 + ) -> Result<HttpResponse> { 353 447 conn.set_read_timeout(Some(self.read_timeout))?; 354 448 355 - // Serialize and send request 356 - let request_bytes = http::serialize_request(method, &path, &host, &merged_headers, body); 449 + let request_bytes = http::serialize_request(method, path, host, headers, body); 357 450 conn.write_all(&request_bytes)?; 358 451 conn.flush()?; 359 452 360 - // Read and parse response 361 453 let response = read_response(&mut conn)?; 362 454 363 - // Store Set-Cookie headers from the response. 455 + // Store Set-Cookie headers 364 456 let set_cookies: Vec<String> = response 365 457 .headers 366 458 .get_all("Set-Cookie") ··· 373 465 374 466 // Return connection to pool if keep-alive 375 467 if !response.connection_close() { 468 + // Determine the connection key for pooling 469 + let is_tls = matches!(conn, Connection::Tls(_)); 470 + let port = url 471 + .port_or_default() 472 + .unwrap_or(if is_tls { 443 } else { 80 }); 473 + let key = ConnectionKey { 474 + host: host.to_string(), 475 + port, 476 + is_tls, 477 + }; 376 478 self.pool.put(key, conn); 377 479 } 378 480 379 481 Ok(response) 380 482 } 381 483 382 - /// Establish a new connection (plain TCP or TLS). 383 - fn connect(&self, host: &str, port: u16, is_tls: bool) -> Result<Connection> { 384 - let tcp = TcpConnection::connect_timeout(host, port, self.connect_timeout)?; 484 + /// Execute a request over an existing HTTP/2 connection. 485 + #[allow(clippy::too_many_arguments)] 486 + fn execute_h2_request( 487 + &mut self, 488 + h2_key: &H2ConnectionKey, 489 + method: Method, 490 + path: &str, 491 + authority: &str, 492 + headers: &Headers, 493 + body: Option<&[u8]>, 494 + url: &Url, 495 + ) -> Result<HttpResponse> { 496 + let extra_headers: Vec<(String, String)> = headers 497 + .iter() 498 + .filter(|(name, _)| { 499 + // Skip pseudo-headers and host (authority is used instead) 500 + let lower = name.to_ascii_lowercase(); 501 + lower != "host" && lower != "connection" && lower != "transfer-encoding" 502 + }) 503 + .map(|(name, value)| (name.to_ascii_lowercase(), value.to_string())) 504 + .collect(); 385 505 386 - if is_tls { 387 - let tls = handshake::connect(tcp, host)?; 388 - Ok(Connection::Tls(tls)) 389 - } else { 390 - Ok(Connection::Plain(tcp)) 506 + let h2_conn = self.h2_connections.get_mut(h2_key).unwrap(); 507 + 508 + let stream_id = 509 + h2_conn.send_request(method.as_str(), path, authority, &extra_headers, body)?; 510 + 511 + let (resp_headers, resp_body, status_code) = h2_conn.read_response(stream_id)?; 512 + 513 + // Convert HTTP/2 response to HttpResponse 514 + let mut response_headers = Headers::new(); 515 + for (name, value) in &resp_headers { 516 + let name_str = String::from_utf8_lossy(name); 517 + let value_str = String::from_utf8_lossy(value); 518 + if !name_str.starts_with(':') { 519 + response_headers.add(&name_str, &value_str); 520 + } 521 + } 522 + 523 + // Store Set-Cookie headers 524 + let set_cookies: Vec<String> = response_headers 525 + .get_all("Set-Cookie") 526 + .into_iter() 527 + .map(|s| s.to_string()) 528 + .collect(); 529 + for header in &set_cookies { 530 + self.cookie_jar.store_from_header(header, url); 391 531 } 532 + 533 + Ok(HttpResponse { 534 + version: "HTTP/2".to_string(), 535 + status_code, 536 + reason: reason_phrase(status_code).to_string(), 537 + headers: response_headers, 538 + body: resp_body, 539 + }) 392 540 } 393 541 } 394 542 ··· 551 699 } 552 700 553 701 BodyStrategy::ReadUntilClose 702 + } 703 + 704 + /// Standard HTTP reason phrase for a status code. 705 + fn reason_phrase(status: u16) -> &'static str { 706 + match status { 707 + 200 => "OK", 708 + 201 => "Created", 709 + 204 => "No Content", 710 + 301 => "Moved Permanently", 711 + 302 => "Found", 712 + 304 => "Not Modified", 713 + 307 => "Temporary Redirect", 714 + 308 => "Permanent Redirect", 715 + 400 => "Bad Request", 716 + 401 => "Unauthorized", 717 + 403 => "Forbidden", 718 + 404 => "Not Found", 719 + 405 => "Method Not Allowed", 720 + 500 => "Internal Server Error", 721 + 502 => "Bad Gateway", 722 + 503 => "Service Unavailable", 723 + _ => "", 724 + } 554 725 } 555 726 556 727 /// Check if chunked body data contains the terminating `0\r\n\r\n`.

+898

crates/net/src/http2/connection.rs

··· 1 + //! HTTP/2 connection management (RFC 7540). 2 + //! 3 + //! Manages the HTTP/2 connection lifecycle: preface exchange, SETTINGS 4 + //! negotiation, stream multiplexing, flow control, and request/response. 5 + 6 + use std::collections::HashMap; 7 + use std::io::{Read, Write}; 8 + 9 + use super::frame::{ 10 + self, data_frame, goaway_frame, headers_frame, ping_frame, settings_ack_frame, settings_frame, 11 + window_update_frame, ErrorCode, Frame, Http2Error, Result, CONNECTION_PREFACE, 12 + DEFAULT_MAX_FRAME_SIZE, FLAG_ACK, FLAG_END_HEADERS, FLAG_END_STREAM, FRAME_CONTINUATION, 13 + FRAME_DATA, FRAME_GOAWAY, FRAME_HEADERS, FRAME_PING, FRAME_RST_STREAM, FRAME_SETTINGS, 14 + FRAME_WINDOW_UPDATE, SETTINGS_ENABLE_PUSH, SETTINGS_HEADER_TABLE_SIZE, 15 + SETTINGS_INITIAL_WINDOW_SIZE, SETTINGS_MAX_CONCURRENT_STREAMS, SETTINGS_MAX_FRAME_SIZE, 16 + SETTINGS_MAX_HEADER_LIST_SIZE, 17 + }; 18 + use super::hpack::{Decoder, Encoder, HeaderField}; 19 + use super::stream::{Stream, StreamState, DEFAULT_INITIAL_WINDOW_SIZE}; 20 + 21 + /// HTTP/2 response: (headers as (name, value) pairs, body bytes, status code). 22 + pub type H2Response = (Vec<(Vec<u8>, Vec<u8>)>, Vec<u8>, u16); 23 + 24 + // --------------------------------------------------------------------------- 25 + // Connection-level settings 26 + // --------------------------------------------------------------------------- 27 + 28 + /// Peer's settings (received from the server). 29 + struct PeerSettings { 30 + header_table_size: u32, 31 + enable_push: bool, 32 + max_concurrent_streams: u32, 33 + initial_window_size: u32, 34 + max_frame_size: u32, 35 + max_header_list_size: u32, 36 + } 37 + 38 + impl Default for PeerSettings { 39 + fn default() -> Self { 40 + Self { 41 + header_table_size: 4096, 42 + enable_push: true, 43 + max_concurrent_streams: u32::MAX, 44 + initial_window_size: DEFAULT_INITIAL_WINDOW_SIZE, 45 + max_frame_size: DEFAULT_MAX_FRAME_SIZE, 46 + max_header_list_size: u32::MAX, 47 + } 48 + } 49 + } 50 + 51 + // --------------------------------------------------------------------------- 52 + // HTTP/2 Connection 53 + // --------------------------------------------------------------------------- 54 + 55 + /// An HTTP/2 connection over a TLS stream. 56 + /// 57 + /// Manages stream multiplexing, flow control, HPACK encoding/decoding, 58 + /// and the SETTINGS handshake. 59 + pub struct Http2Connection<S> { 60 + /// Underlying transport (TLS stream). 61 + stream: S, 62 + /// Active streams indexed by stream ID. 63 + streams: HashMap<u32, Stream>, 64 + /// Next client-initiated stream ID (odd, starting at 1). 65 + next_stream_id: u32, 66 + /// Connection-level send window. 67 + conn_send_window: i64, 68 + /// Connection-level receive window. 69 + conn_recv_window: i64, 70 + /// Peer's (server's) settings. 71 + peer_settings: PeerSettings, 72 + /// HPACK encoder for request headers. 73 + hpack_encoder: Encoder, 74 + /// HPACK decoder for response headers. 75 + hpack_decoder: Decoder, 76 + /// Pending encoder table size update (from peer SETTINGS). 77 + pending_table_size_update: Option<usize>, 78 + /// Whether we have received the initial SETTINGS from the peer. 79 + settings_received: bool, 80 + /// Whether a GOAWAY has been received. 81 + goaway_received: bool, 82 + /// Last stream ID from GOAWAY. 83 + goaway_last_stream_id: u32, 84 + } 85 + 86 + impl<S: Read + Write> Http2Connection<S> { 87 + /// Perform the HTTP/2 connection handshake. 88 + /// 89 + /// Sends the connection preface and initial SETTINGS, then reads the 90 + /// server's SETTINGS and acknowledges it. 91 + pub fn new(mut stream: S) -> Result<Self> { 92 + // Send connection preface 93 + stream 94 + .write_all(CONNECTION_PREFACE) 95 + .map_err(Http2Error::Io)?; 96 + 97 + // Send our SETTINGS 98 + let our_settings = [ 99 + (SETTINGS_MAX_CONCURRENT_STREAMS, 100), 100 + (SETTINGS_INITIAL_WINDOW_SIZE, DEFAULT_INITIAL_WINDOW_SIZE), 101 + (SETTINGS_ENABLE_PUSH, 0), // Disable server push 102 + ]; 103 + let settings = settings_frame(&our_settings, false); 104 + settings.write_to(&mut stream)?; 105 + 106 + let mut conn = Self { 107 + stream, 108 + streams: HashMap::new(), 109 + next_stream_id: 1, 110 + conn_send_window: DEFAULT_INITIAL_WINDOW_SIZE as i64, 111 + conn_recv_window: DEFAULT_INITIAL_WINDOW_SIZE as i64, 112 + peer_settings: PeerSettings::default(), 113 + hpack_encoder: Encoder::new(4096), 114 + hpack_decoder: Decoder::new(4096), 115 + pending_table_size_update: None, 116 + settings_received: false, 117 + goaway_received: false, 118 + goaway_last_stream_id: 0, 119 + }; 120 + 121 + // Read frames until we get the server's SETTINGS 122 + conn.read_until_settings()?; 123 + 124 + // Send a larger connection-level window update to allow more data 125 + // Default is 65535; bump to ~1MB for better throughput 126 + let window_bump = 1_048_576 - DEFAULT_INITIAL_WINDOW_SIZE; 127 + if window_bump > 0 { 128 + let wu = window_update_frame(0, window_bump); 129 + wu.write_to(&mut conn.stream)?; 130 + conn.conn_recv_window += window_bump as i64; 131 + } 132 + 133 + Ok(conn) 134 + } 135 + 136 + /// Send an HTTP request and return the stream ID. 137 + /// 138 + /// The request headers are HPACK-encoded and sent as a HEADERS frame. 139 + /// If a body is provided, it is sent as DATA frame(s). 140 + pub fn send_request( 141 + &mut self, 142 + method: &str, 143 + path: &str, 144 + authority: &str, 145 + extra_headers: &[(String, String)], 146 + body: Option<&[u8]>, 147 + ) -> Result<u32> { 148 + if self.goaway_received { 149 + return Err(Http2Error::Protocol("connection received GOAWAY".into())); 150 + } 151 + 152 + let stream_id = self.next_stream_id; 153 + self.next_stream_id += 2; 154 + 155 + // Build pseudo-headers + regular headers 156 + let mut headers = vec![ 157 + HeaderField::new(b":method", method.as_bytes()), 158 + HeaderField::new(b":scheme", b"https"), 159 + HeaderField::new(b":path", path.as_bytes()), 160 + HeaderField::new(b":authority", authority.as_bytes()), 161 + ]; 162 + for (name, value) in extra_headers { 163 + headers.push(HeaderField::new(name.as_bytes(), value.as_bytes())); 164 + } 165 + 166 + // Encode headers with HPACK 167 + let mut header_block = Vec::new(); 168 + if let Some(new_size) = self.pending_table_size_update.take() { 169 + self.hpack_encoder 170 + .encode_table_size_update(&mut header_block, new_size); 171 + } 172 + header_block.extend_from_slice(&self.hpack_encoder.encode(&headers)); 173 + 174 + let end_stream = body.is_none(); 175 + 176 + // Create stream and transition to open 177 + let mut stream = Stream::new( 178 + stream_id, 179 + self.peer_settings.initial_window_size, 180 + DEFAULT_INITIAL_WINDOW_SIZE, 181 + ); 182 + stream.send_headers()?; 183 + if end_stream { 184 + stream.send_end_stream()?; 185 + } 186 + self.streams.insert(stream_id, stream); 187 + 188 + // Send HEADERS frame 189 + let hdr_frame = headers_frame(stream_id, header_block, end_stream); 190 + hdr_frame.write_to(&mut self.stream)?; 191 + 192 + // Send body as DATA frames if present 193 + if let Some(data) = body { 194 + self.send_data(stream_id, data)?; 195 + } 196 + 197 + Ok(stream_id) 198 + } 199 + 200 + /// Read the response for a given stream. 201 + /// 202 + /// Reads frames until the stream is complete (END_STREAM received on 203 + /// both headers and data). Returns the response headers and body. 204 + pub fn read_response(&mut self, stream_id: u32) -> Result<H2Response> { 205 + // Read frames until this stream is done 206 + loop { 207 + let stream = self 208 + .streams 209 + .get(&stream_id) 210 + .ok_or_else(|| Http2Error::Protocol(format!("stream {stream_id} not found")))?; 211 + 212 + if stream.is_closed() || stream.state == StreamState::HalfClosedRemote { 213 + break; 214 + } 215 + 216 + self.read_and_process_frame()?; 217 + } 218 + 219 + let stream = self 220 + .streams 221 + .get(&stream_id) 222 + .ok_or_else(|| Http2Error::Protocol(format!("stream {stream_id} not found")))?; 223 + 224 + let status = stream.status_code.unwrap_or(0); 225 + Ok((stream.response_headers.clone(), stream.body.clone(), status)) 226 + } 227 + 228 + /// Close the connection gracefully by sending GOAWAY. 229 + pub fn close(&mut self) -> Result<()> { 230 + let last_id = if self.next_stream_id > 1 { 231 + self.next_stream_id - 2 232 + } else { 233 + 0 234 + }; 235 + let frame = goaway_frame(last_id, ErrorCode::NoError); 236 + frame.write_to(&mut self.stream)?; 237 + Ok(()) 238 + } 239 + 240 + // ----------------------------------------------------------------------- 241 + // Internal: frame sending 242 + // ----------------------------------------------------------------------- 243 + 244 + /// Send body data as DATA frame(s) with flow control. 245 + fn send_data(&mut self, stream_id: u32, data: &[u8]) -> Result<()> { 246 + let mut offset = 0; 247 + while offset < data.len() { 248 + let remaining = data.len() - offset; 249 + // Respect both connection and stream flow control windows 250 + let stream = self 251 + .streams 252 + .get(&stream_id) 253 + .ok_or_else(|| Http2Error::Protocol("stream not found".into()))?; 254 + 255 + let max_by_stream = stream.send_window.max(0) as usize; 256 + let max_by_conn = self.conn_send_window.max(0) as usize; 257 + let max_by_frame = self.peer_settings.max_frame_size as usize; 258 + let chunk_size = remaining 259 + .min(max_by_stream) 260 + .min(max_by_conn) 261 + .min(max_by_frame); 262 + 263 + if chunk_size == 0 { 264 + // Flow control window exhausted — read frames to get WINDOW_UPDATE 265 + self.read_and_process_frame()?; 266 + continue; 267 + } 268 + 269 + let is_last = offset + chunk_size >= data.len(); 270 + let chunk = &data[offset..offset + chunk_size]; 271 + let frame = data_frame(stream_id, chunk.to_vec(), is_last); 272 + frame.write_to(&mut self.stream)?; 273 + 274 + // Update flow control 275 + self.conn_send_window -= chunk_size as i64; 276 + let stream = self.streams.get_mut(&stream_id).unwrap(); 277 + stream.consume_send_window(chunk_size as u32)?; 278 + if is_last { 279 + stream.send_end_stream()?; 280 + } 281 + 282 + offset += chunk_size; 283 + } 284 + Ok(()) 285 + } 286 + 287 + // ----------------------------------------------------------------------- 288 + // Internal: frame reading 289 + // ----------------------------------------------------------------------- 290 + 291 + /// Read frames until we get the initial SETTINGS from the server. 292 + /// Also consumes the SETTINGS ACK for our settings if present. 293 + fn read_until_settings(&mut self) -> Result<()> { 294 + let mut got_settings = false; 295 + let mut got_settings_ack = false; 296 + 297 + for _ in 0..32 { 298 + if got_settings && got_settings_ack { 299 + return Ok(()); 300 + } 301 + let frame = Frame::read_from(&mut self.stream, self.peer_settings.max_frame_size)?; 302 + 303 + if frame.header.frame_type == FRAME_SETTINGS { 304 + if frame.header.has_flag(FLAG_ACK) { 305 + got_settings_ack = true; 306 + // Process it (no-op for ACK, but keeps the flow clean) 307 + self.process_frame(frame)?; 308 + } else { 309 + got_settings = true; 310 + self.process_frame(frame)?; 311 + } 312 + } else { 313 + // Process non-SETTINGS frames normally 314 + self.process_frame(frame)?; 315 + } 316 + } 317 + 318 + if !self.settings_received { 319 + return Err(Http2Error::Protocol( 320 + "did not receive SETTINGS from server".into(), 321 + )); 322 + } 323 + Ok(()) 324 + } 325 + 326 + /// Read one frame from the transport and process it. 327 + fn read_and_process_frame(&mut self) -> Result<()> { 328 + let frame = Frame::read_from(&mut self.stream, self.peer_settings.max_frame_size)?; 329 + self.process_frame(frame) 330 + } 331 + 332 + /// Process a received frame. 333 + fn process_frame(&mut self, frame: Frame) -> Result<()> { 334 + match frame.header.frame_type { 335 + FRAME_SETTINGS => self.handle_settings(&frame), 336 + FRAME_HEADERS => self.handle_headers(&frame), 337 + FRAME_CONTINUATION => self.handle_continuation(&frame), 338 + FRAME_DATA => self.handle_data(&frame), 339 + FRAME_WINDOW_UPDATE => self.handle_window_update(&frame), 340 + FRAME_RST_STREAM => self.handle_rst_stream(&frame), 341 + FRAME_GOAWAY => self.handle_goaway(&frame), 342 + FRAME_PING => self.handle_ping(&frame), 343 + _ => { 344 + // Unknown frame types MUST be ignored (RFC 7540 §4.1) 345 + Ok(()) 346 + } 347 + } 348 + } 349 + 350 + fn handle_settings(&mut self, frame: &Frame) -> Result<()> { 351 + if frame.header.stream_id != 0 { 352 + return Err(Http2Error::Protocol("SETTINGS on non-zero stream".into())); 353 + } 354 + 355 + if frame.header.has_flag(FLAG_ACK) { 356 + // ACK for our SETTINGS — nothing more to do 357 + return Ok(()); 358 + } 359 + 360 + let settings = frame::parse_settings(&frame.payload)?; 361 + for s in &settings { 362 + match s.id { 363 + SETTINGS_HEADER_TABLE_SIZE => { 364 + self.peer_settings.header_table_size = s.value; 365 + self.pending_table_size_update = Some(s.value as usize); 366 + } 367 + SETTINGS_ENABLE_PUSH => { 368 + self.peer_settings.enable_push = s.value != 0; 369 + } 370 + SETTINGS_MAX_CONCURRENT_STREAMS => { 371 + self.peer_settings.max_concurrent_streams = s.value; 372 + } 373 + SETTINGS_INITIAL_WINDOW_SIZE => { 374 + if s.value > 0x7FFF_FFFF { 375 + return Err(Http2Error::FlowControl); 376 + } 377 + // Adjust existing streams' send windows 378 + let delta = s.value as i64 - self.peer_settings.initial_window_size as i64; 379 + for stream in self.streams.values_mut() { 380 + stream.send_window += delta; 381 + } 382 + self.peer_settings.initial_window_size = s.value; 383 + } 384 + SETTINGS_MAX_FRAME_SIZE => { 385 + if s.value < DEFAULT_MAX_FRAME_SIZE || s.value > frame::MAX_FRAME_SIZE_LIMIT { 386 + return Err(Http2Error::Protocol( 387 + "invalid SETTINGS_MAX_FRAME_SIZE".into(), 388 + )); 389 + } 390 + self.peer_settings.max_frame_size = s.value; 391 + } 392 + SETTINGS_MAX_HEADER_LIST_SIZE => { 393 + self.peer_settings.max_header_list_size = s.value; 394 + } 395 + _ => {} // Unknown settings MUST be ignored (RFC 7540 §6.5.2) 396 + } 397 + } 398 + 399 + self.settings_received = true; 400 + 401 + // Send SETTINGS ACK 402 + let ack = settings_ack_frame(); 403 + ack.write_to(&mut self.stream)?; 404 + 405 + Ok(()) 406 + } 407 + 408 + fn handle_headers(&mut self, frame: &Frame) -> Result<()> { 409 + let stream_id = frame.header.stream_id; 410 + if stream_id == 0 { 411 + return Err(Http2Error::Protocol("HEADERS on stream 0".into())); 412 + } 413 + 414 + let stream = match self.streams.get_mut(&stream_id) { 415 + Some(s) => s, 416 + None => { 417 + // Server-initiated stream (even ID) or unknown — ignore 418 + return Ok(()); 419 + } 420 + }; 421 + 422 + stream.header_block.extend_from_slice(&frame.payload); 423 + 424 + if frame.header.has_flag(FLAG_END_HEADERS) { 425 + self.decode_headers(stream_id)?; 426 + } 427 + 428 + if frame.header.has_flag(FLAG_END_STREAM) { 429 + let stream = self.streams.get_mut(&stream_id).unwrap(); 430 + stream.recv_end_stream()?; 431 + } 432 + 433 + Ok(()) 434 + } 435 + 436 + fn handle_continuation(&mut self, frame: &Frame) -> Result<()> { 437 + let stream_id = frame.header.stream_id; 438 + let stream = match self.streams.get_mut(&stream_id) { 439 + Some(s) => s, 440 + None => return Ok(()), 441 + }; 442 + 443 + stream.header_block.extend_from_slice(&frame.payload); 444 + 445 + if frame.header.has_flag(FLAG_END_HEADERS) { 446 + self.decode_headers(stream_id)?; 447 + } 448 + 449 + Ok(()) 450 + } 451 + 452 + /// Decode accumulated header block for a stream. 453 + fn decode_headers(&mut self, stream_id: u32) -> Result<()> { 454 + let stream = self.streams.get_mut(&stream_id).unwrap(); 455 + let header_block = std::mem::take(&mut stream.header_block); 456 + 457 + let headers = self.hpack_decoder.decode(&header_block)?; 458 + 459 + for hf in &headers { 460 + if hf.name == b":status" { 461 + if let Ok(s) = std::str::from_utf8(&hf.value) { 462 + stream.status_code = s.parse().ok(); 463 + } 464 + } 465 + } 466 + 467 + stream.response_headers = headers.into_iter().map(|hf| (hf.name, hf.value)).collect(); 468 + 469 + Ok(()) 470 + } 471 + 472 + fn handle_data(&mut self, frame: &Frame) -> Result<()> { 473 + let stream_id = frame.header.stream_id; 474 + if stream_id == 0 { 475 + return Err(Http2Error::Protocol("DATA on stream 0".into())); 476 + } 477 + 478 + let data_len = frame.payload.len() as u32; 479 + 480 + // Connection-level flow control 481 + self.conn_recv_window -= data_len as i64; 482 + if self.conn_recv_window < 0 { 483 + return Err(Http2Error::FlowControl); 484 + } 485 + 486 + let stream = match self.streams.get_mut(&stream_id) { 487 + Some(s) => s, 488 + None => { 489 + // Stream may have been reset or doesn't exist — send connection-level 490 + // WINDOW_UPDATE to reclaim the window, then ignore. 491 + let wu = window_update_frame(0, data_len); 492 + wu.write_to(&mut self.stream)?; 493 + self.conn_recv_window += data_len as i64; 494 + return Ok(()); 495 + } 496 + }; 497 + 498 + stream.consume_recv_window(data_len)?; 499 + stream.body.extend_from_slice(&frame.payload); 500 + 501 + if frame.header.has_flag(FLAG_END_STREAM) { 502 + stream.recv_end_stream()?; 503 + } 504 + 505 + // Send WINDOW_UPDATE for stream and connection to keep data flowing 506 + if data_len > 0 { 507 + let stream_wu = window_update_frame(stream_id, data_len); 508 + stream_wu.write_to(&mut self.stream)?; 509 + let stream = self.streams.get_mut(&stream_id).unwrap(); 510 + stream.increase_recv_window(data_len); 511 + 512 + let conn_wu = window_update_frame(0, data_len); 513 + conn_wu.write_to(&mut self.stream)?; 514 + self.conn_recv_window += data_len as i64; 515 + } 516 + 517 + Ok(()) 518 + } 519 + 520 + fn handle_window_update(&mut self, frame: &Frame) -> Result<()> { 521 + if frame.payload.len() != 4 { 522 + return Err(Http2Error::Protocol( 523 + "WINDOW_UPDATE payload must be 4 bytes".into(), 524 + )); 525 + } 526 + let increment = u32::from_be_bytes([ 527 + frame.payload[0], 528 + frame.payload[1], 529 + frame.payload[2], 530 + frame.payload[3], 531 + ]) & 0x7FFF_FFFF; 532 + 533 + if increment == 0 { 534 + return Err(Http2Error::Protocol( 535 + "WINDOW_UPDATE increment must be non-zero".into(), 536 + )); 537 + } 538 + 539 + if frame.header.stream_id == 0 { 540 + self.conn_send_window += increment as i64; 541 + if self.conn_send_window > 0x7FFF_FFFF { 542 + return Err(Http2Error::FlowControl); 543 + } 544 + } else if let Some(stream) = self.streams.get_mut(&frame.header.stream_id) { 545 + stream.increase_send_window(increment)?; 546 + } 547 + 548 + Ok(()) 549 + } 550 + 551 + fn handle_rst_stream(&mut self, frame: &Frame) -> Result<()> { 552 + if frame.payload.len() != 4 { 553 + return Err(Http2Error::Protocol( 554 + "RST_STREAM payload must be 4 bytes".into(), 555 + )); 556 + } 557 + let error_code = u32::from_be_bytes([ 558 + frame.payload[0], 559 + frame.payload[1], 560 + frame.payload[2], 561 + frame.payload[3], 562 + ]); 563 + let error_code = ErrorCode::from_u32(error_code); 564 + 565 + if let Some(stream) = self.streams.get_mut(&frame.header.stream_id) { 566 + // Don't propagate the error — just mark the stream as closed. 567 + // The caller will see the error when reading the response. 568 + stream.state = StreamState::Closed; 569 + // Store the error for later retrieval, but don't fail the whole connection. 570 + let _ = stream.recv_rst_stream(error_code); 571 + } 572 + 573 + Ok(()) 574 + } 575 + 576 + fn handle_goaway(&mut self, frame: &Frame) -> Result<()> { 577 + if frame.payload.len() < 8 { 578 + return Err(Http2Error::Protocol("GOAWAY payload too short".into())); 579 + } 580 + let last_stream_id = u32::from_be_bytes([ 581 + frame.payload[0], 582 + frame.payload[1], 583 + frame.payload[2], 584 + frame.payload[3], 585 + ]) & 0x7FFF_FFFF; 586 + let error_code = u32::from_be_bytes([ 587 + frame.payload[4], 588 + frame.payload[5], 589 + frame.payload[6], 590 + frame.payload[7], 591 + ]); 592 + let error_code = ErrorCode::from_u32(error_code); 593 + 594 + self.goaway_received = true; 595 + self.goaway_last_stream_id = last_stream_id; 596 + 597 + // Close streams with ID > last_stream_id 598 + for (id, stream) in &mut self.streams { 599 + if *id > last_stream_id { 600 + stream.state = StreamState::Closed; 601 + } 602 + } 603 + 604 + if error_code != ErrorCode::NoError { 605 + let debug_data = if frame.payload.len() > 8 { 606 + String::from_utf8_lossy(&frame.payload[8..]).to_string() 607 + } else { 608 + String::new() 609 + }; 610 + return Err(Http2Error::GoAway(error_code, debug_data)); 611 + } 612 + 613 + Ok(()) 614 + } 615 + 616 + fn handle_ping(&mut self, frame: &Frame) -> Result<()> { 617 + if frame.header.stream_id != 0 { 618 + return Err(Http2Error::Protocol("PING on non-zero stream".into())); 619 + } 620 + if frame.payload.len() != 8 { 621 + return Err(Http2Error::Protocol("PING payload must be 8 bytes".into())); 622 + } 623 + 624 + if frame.header.has_flag(FLAG_ACK) { 625 + // PING ACK — we don't track outgoing pings, so just ignore 626 + return Ok(()); 627 + } 628 + 629 + // Respond with PING ACK containing same payload 630 + let mut data = [0u8; 8]; 631 + data.copy_from_slice(&frame.payload); 632 + let ack = ping_frame(data, true); 633 + ack.write_to(&mut self.stream)?; 634 + 635 + Ok(()) 636 + } 637 + } 638 + 639 + // --------------------------------------------------------------------------- 640 + // Tests 641 + // --------------------------------------------------------------------------- 642 + 643 + #[cfg(test)] 644 + mod tests { 645 + use super::super::frame::FrameHeader; 646 + use super::*; 647 + 648 + /// A mock transport that records writes and replays reads. 649 + struct MockTransport { 650 + /// Data to be read by the connection. 651 + read_buf: Vec<u8>, 652 + read_pos: usize, 653 + /// Data written by the connection. 654 + write_buf: Vec<u8>, 655 + } 656 + 657 + impl MockTransport { 658 + fn new(read_data: Vec<u8>) -> Self { 659 + Self { 660 + read_buf: read_data, 661 + read_pos: 0, 662 + write_buf: Vec::new(), 663 + } 664 + } 665 + 666 + fn written(&self) -> &[u8] { 667 + &self.write_buf 668 + } 669 + } 670 + 671 + impl Read for MockTransport { 672 + fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> { 673 + if self.read_pos >= self.read_buf.len() { 674 + return Err(io::Error::new(io::ErrorKind::UnexpectedEof, "no more data")); 675 + } 676 + let available = &self.read_buf[self.read_pos..]; 677 + let to_copy = available.len().min(buf.len()); 678 + buf[..to_copy].copy_from_slice(&available[..to_copy]); 679 + self.read_pos += to_copy; 680 + Ok(to_copy) 681 + } 682 + } 683 + 684 + impl Write for MockTransport { 685 + fn write(&mut self, buf: &[u8]) -> io::Result<usize> { 686 + self.write_buf.extend_from_slice(buf); 687 + Ok(buf.len()) 688 + } 689 + 690 + fn flush(&mut self) -> io::Result<()> { 691 + Ok(()) 692 + } 693 + } 694 + 695 + use std::io; 696 + 697 + /// Build server preface: SETTINGS frame + SETTINGS ACK 698 + fn build_server_preface(settings: &[(u16, u32)]) -> Vec<u8> { 699 + let mut buf = Vec::new(); 700 + let sf = settings_frame(settings, false); 701 + sf.write_to(&mut buf).unwrap(); 702 + // Also write a SETTINGS ACK for our SETTINGS 703 + let ack = settings_ack_frame(); 704 + ack.write_to(&mut buf).unwrap(); 705 + buf 706 + } 707 + 708 + #[test] 709 + fn connection_handshake() { 710 + let server_data = build_server_preface(&[ 711 + (SETTINGS_MAX_CONCURRENT_STREAMS, 128), 712 + (SETTINGS_INITIAL_WINDOW_SIZE, 65535), 713 + ]); 714 + 715 + let transport = MockTransport::new(server_data); 716 + let conn = Http2Connection::new(transport).unwrap(); 717 + 718 + assert!(conn.settings_received); 719 + assert_eq!(conn.peer_settings.max_concurrent_streams, 128); 720 + assert_eq!(conn.next_stream_id, 1); 721 + } 722 + 723 + #[test] 724 + fn connection_sends_preface() { 725 + let server_data = build_server_preface(&[]); 726 + let transport = MockTransport::new(server_data); 727 + let conn = Http2Connection::new(transport).unwrap(); 728 + 729 + // Verify connection preface was sent 730 + let written = &conn.stream.write_buf; 731 + assert!(written.starts_with(CONNECTION_PREFACE)); 732 + } 733 + 734 + #[test] 735 + fn send_request_allocates_stream_id() { 736 + let server_data = build_server_preface(&[]); 737 + let transport = MockTransport::new(server_data); 738 + let mut conn = Http2Connection::new(transport).unwrap(); 739 + 740 + // We can't fully test send_request without a response, but we can 741 + // verify stream ID allocation by checking next_stream_id advances. 742 + // Append response data for a simple 200 OK 743 + let mut response_data = Vec::new(); 744 + // HEADERS frame with a simple :status 200 745 + let mut encoder = Encoder::new(4096); 746 + let headers = vec![HeaderField::new(b":status", b"200")]; 747 + let block = encoder.encode(&headers); 748 + let hf = headers_frame(1, block, true); 749 + hf.write_to(&mut response_data).unwrap(); 750 + 751 + // Add response data to the read buffer 752 + conn.stream.read_buf.extend_from_slice(&response_data); 753 + 754 + let stream_id = conn 755 + .send_request("GET", "/", "example.com", &[], None) 756 + .unwrap(); 757 + assert_eq!(stream_id, 1); 758 + assert_eq!(conn.next_stream_id, 3); 759 + 760 + // Read the response 761 + let (_headers, body, status) = conn.read_response(1).unwrap(); 762 + assert_eq!(status, 200); 763 + assert!(body.is_empty()); 764 + } 765 + 766 + #[test] 767 + fn handles_ping() { 768 + let mut server_data = build_server_preface(&[]); 769 + 770 + // Server sends a PING 771 + let ping = ping_frame([1, 2, 3, 4, 5, 6, 7, 8], false); 772 + ping.write_to(&mut server_data).unwrap(); 773 + 774 + // Also send a SETTINGS ACK so we don't hang 775 + // (our WINDOW_UPDATE is already handled) 776 + 777 + let transport = MockTransport::new(server_data); 778 + let mut conn = Http2Connection::new(transport).unwrap(); 779 + 780 + // Process the PING frame 781 + conn.read_and_process_frame().unwrap(); 782 + 783 + // Verify PING ACK was sent (it will be in the write buffer) 784 + // The write buffer contains: preface + settings + window_update + settings_ack + ping_ack 785 + let written = &conn.stream.write_buf; 786 + // Find the PING ACK frame at the end 787 + let last_frame_start = written.len() - 9 - 8; // header + 8 bytes payload 788 + let header = FrameHeader::decode( 789 + &written[last_frame_start..last_frame_start + 9] 790 + .try_into() 791 + .unwrap(), 792 + ); 793 + assert_eq!(header.frame_type, FRAME_PING); 794 + assert!(header.has_flag(FLAG_ACK)); 795 + assert_eq!(&written[last_frame_start + 9..], &[1, 2, 3, 4, 5, 6, 7, 8]); 796 + } 797 + 798 + #[test] 799 + fn handles_goaway_no_error() { 800 + let mut server_data = build_server_preface(&[]); 801 + 802 + // Server sends GOAWAY with NO_ERROR 803 + let ga = goaway_frame(0, ErrorCode::NoError); 804 + ga.write_to(&mut server_data).unwrap(); 805 + 806 + let transport = MockTransport::new(server_data); 807 + let mut conn = Http2Connection::new(transport).unwrap(); 808 + 809 + conn.read_and_process_frame().unwrap(); 810 + assert!(conn.goaway_received); 811 + } 812 + 813 + #[test] 814 + fn handles_goaway_with_error() { 815 + let mut server_data = build_server_preface(&[]); 816 + 817 + let ga = goaway_frame(0, ErrorCode::ProtocolError); 818 + ga.write_to(&mut server_data).unwrap(); 819 + 820 + let transport = MockTransport::new(server_data); 821 + let mut conn = Http2Connection::new(transport).unwrap(); 822 + 823 + let result = conn.read_and_process_frame(); 824 + assert!(matches!( 825 + result, 826 + Err(Http2Error::GoAway(ErrorCode::ProtocolError, _)) 827 + )); 828 + } 829 + 830 + #[test] 831 + fn settings_updates_peer_config() { 832 + let server_data = build_server_preface(&[ 833 + (SETTINGS_MAX_FRAME_SIZE, 32768), 834 + (SETTINGS_HEADER_TABLE_SIZE, 8192), 835 + ]); 836 + 837 + let transport = MockTransport::new(server_data); 838 + let conn = Http2Connection::new(transport).unwrap(); 839 + 840 + assert_eq!(conn.peer_settings.max_frame_size, 32768); 841 + assert_eq!(conn.peer_settings.header_table_size, 8192); 842 + } 843 + 844 + #[test] 845 + fn window_update_connection_level() { 846 + let mut server_data = build_server_preface(&[]); 847 + 848 + // Server sends connection-level WINDOW_UPDATE 849 + let wu = window_update_frame(0, 1000); 850 + wu.write_to(&mut server_data).unwrap(); 851 + 852 + let transport = MockTransport::new(server_data); 853 + let mut conn = Http2Connection::new(transport).unwrap(); 854 + 855 + let initial_window = conn.conn_send_window; 856 + conn.read_and_process_frame().unwrap(); 857 + assert_eq!(conn.conn_send_window, initial_window + 1000); 858 + } 859 + 860 + #[test] 861 + fn full_request_response() { 862 + let mut server_data = build_server_preface(&[]); 863 + 864 + // Build a response: HEADERS with :status 200, then DATA with body 865 + let mut encoder = Encoder::new(4096); 866 + let resp_headers = vec![ 867 + HeaderField::new(b":status", b"200"), 868 + HeaderField::new(b"content-type", b"text/plain"), 869 + ]; 870 + let block = encoder.encode(&resp_headers); 871 + let hf = headers_frame(1, block, false); 872 + hf.write_to(&mut server_data).unwrap(); 873 + 874 + // DATA frame with body 875 + let body = b"Hello, HTTP/2!"; 876 + let df = data_frame(1, body.to_vec(), true); 877 + df.write_to(&mut server_data).unwrap(); 878 + 879 + let transport = MockTransport::new(server_data); 880 + let mut conn = Http2Connection::new(transport).unwrap(); 881 + 882 + let stream_id = conn 883 + .send_request("GET", "/hello", "example.com", &[], None) 884 + .unwrap(); 885 + assert_eq!(stream_id, 1); 886 + 887 + let (headers, resp_body, status) = conn.read_response(1).unwrap(); 888 + assert_eq!(status, 200); 889 + assert_eq!(resp_body, b"Hello, HTTP/2!"); 890 + 891 + // Check headers include content-type 892 + let ct = headers 893 + .iter() 894 + .find(|(n, _)| n == b"content-type") 895 + .map(|(_, v)| v.as_slice()); 896 + assert_eq!(ct, Some(b"text/plain".as_slice())); 897 + } 898 + }

+766

crates/net/src/http2/frame.rs

··· 1 + //! HTTP/2 binary framing layer (RFC 7540 §4). 2 + //! 3 + //! Implements the 9-byte frame header and all standard frame types: 4 + //! DATA, HEADERS, PRIORITY, RST_STREAM, SETTINGS, PUSH_PROMISE, 5 + //! PING, GOAWAY, WINDOW_UPDATE, CONTINUATION. 6 + 7 + use std::fmt; 8 + use std::io::{self, Read, Write}; 9 + 10 + // --------------------------------------------------------------------------- 11 + // Constants 12 + // --------------------------------------------------------------------------- 13 + 14 + /// Size of the HTTP/2 frame header (9 bytes). 15 + pub const FRAME_HEADER_SIZE: usize = 9; 16 + 17 + /// Default maximum frame payload size (RFC 7540 §4.2). 18 + pub const DEFAULT_MAX_FRAME_SIZE: u32 = 16384; 19 + 20 + /// Maximum allowed value for SETTINGS_MAX_FRAME_SIZE (RFC 7540 §4.2). 21 + pub const MAX_FRAME_SIZE_LIMIT: u32 = 16_777_215; 22 + 23 + /// HTTP/2 connection preface magic bytes. 24 + pub const CONNECTION_PREFACE: &[u8] = b"PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n"; 25 + 26 + // --------------------------------------------------------------------------- 27 + // Frame type codes (RFC 7540 §6) 28 + // --------------------------------------------------------------------------- 29 + 30 + pub const FRAME_DATA: u8 = 0x0; 31 + pub const FRAME_HEADERS: u8 = 0x1; 32 + pub const FRAME_PRIORITY: u8 = 0x2; 33 + pub const FRAME_RST_STREAM: u8 = 0x3; 34 + pub const FRAME_SETTINGS: u8 = 0x4; 35 + pub const FRAME_PUSH_PROMISE: u8 = 0x5; 36 + pub const FRAME_PING: u8 = 0x6; 37 + pub const FRAME_GOAWAY: u8 = 0x7; 38 + pub const FRAME_WINDOW_UPDATE: u8 = 0x8; 39 + pub const FRAME_CONTINUATION: u8 = 0x9; 40 + 41 + // --------------------------------------------------------------------------- 42 + // Frame flags 43 + // --------------------------------------------------------------------------- 44 + 45 + pub const FLAG_END_STREAM: u8 = 0x1; 46 + pub const FLAG_END_HEADERS: u8 = 0x4; 47 + pub const FLAG_PADDED: u8 = 0x8; 48 + pub const FLAG_PRIORITY: u8 = 0x20; 49 + pub const FLAG_ACK: u8 = 0x1; 50 + 51 + // --------------------------------------------------------------------------- 52 + // Settings identifiers (RFC 7540 §6.5.2) 53 + // --------------------------------------------------------------------------- 54 + 55 + pub const SETTINGS_HEADER_TABLE_SIZE: u16 = 0x1; 56 + pub const SETTINGS_ENABLE_PUSH: u16 = 0x2; 57 + pub const SETTINGS_MAX_CONCURRENT_STREAMS: u16 = 0x3; 58 + pub const SETTINGS_INITIAL_WINDOW_SIZE: u16 = 0x4; 59 + pub const SETTINGS_MAX_FRAME_SIZE: u16 = 0x5; 60 + pub const SETTINGS_MAX_HEADER_LIST_SIZE: u16 = 0x6; 61 + 62 + // --------------------------------------------------------------------------- 63 + // Error codes (RFC 7540 §7) 64 + // --------------------------------------------------------------------------- 65 + 66 + #[derive(Debug, Clone, Copy, PartialEq, Eq)] 67 + pub enum ErrorCode { 68 + NoError, 69 + ProtocolError, 70 + InternalError, 71 + FlowControlError, 72 + SettingsTimeout, 73 + StreamClosed, 74 + FrameSizeError, 75 + RefusedStream, 76 + Cancel, 77 + CompressionError, 78 + ConnectError, 79 + EnhanceYourCalm, 80 + InadequateSecurity, 81 + Http11Required, 82 + } 83 + 84 + impl ErrorCode { 85 + pub fn from_u32(code: u32) -> Self { 86 + match code { 87 + 0x0 => Self::NoError, 88 + 0x1 => Self::ProtocolError, 89 + 0x2 => Self::InternalError, 90 + 0x3 => Self::FlowControlError, 91 + 0x4 => Self::SettingsTimeout, 92 + 0x5 => Self::StreamClosed, 93 + 0x6 => Self::FrameSizeError, 94 + 0x7 => Self::RefusedStream, 95 + 0x8 => Self::Cancel, 96 + 0x9 => Self::CompressionError, 97 + 0xa => Self::ConnectError, 98 + 0xb => Self::EnhanceYourCalm, 99 + 0xc => Self::InadequateSecurity, 100 + 0xd => Self::Http11Required, 101 + _ => Self::InternalError, 102 + } 103 + } 104 + 105 + pub fn as_u32(self) -> u32 { 106 + match self { 107 + Self::NoError => 0x0, 108 + Self::ProtocolError => 0x1, 109 + Self::InternalError => 0x2, 110 + Self::FlowControlError => 0x3, 111 + Self::SettingsTimeout => 0x4, 112 + Self::StreamClosed => 0x5, 113 + Self::FrameSizeError => 0x6, 114 + Self::RefusedStream => 0x7, 115 + Self::Cancel => 0x8, 116 + Self::CompressionError => 0x9, 117 + Self::ConnectError => 0xa, 118 + Self::EnhanceYourCalm => 0xb, 119 + Self::InadequateSecurity => 0xc, 120 + Self::Http11Required => 0xd, 121 + } 122 + } 123 + } 124 + 125 + impl fmt::Display for ErrorCode { 126 + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { 127 + match self { 128 + Self::NoError => write!(f, "NO_ERROR"), 129 + Self::ProtocolError => write!(f, "PROTOCOL_ERROR"), 130 + Self::InternalError => write!(f, "INTERNAL_ERROR"), 131 + Self::FlowControlError => write!(f, "FLOW_CONTROL_ERROR"), 132 + Self::SettingsTimeout => write!(f, "SETTINGS_TIMEOUT"), 133 + Self::StreamClosed => write!(f, "STREAM_CLOSED"), 134 + Self::FrameSizeError => write!(f, "FRAME_SIZE_ERROR"), 135 + Self::RefusedStream => write!(f, "REFUSED_STREAM"), 136 + Self::Cancel => write!(f, "CANCEL"), 137 + Self::CompressionError => write!(f, "COMPRESSION_ERROR"), 138 + Self::ConnectError => write!(f, "CONNECT_ERROR"), 139 + Self::EnhanceYourCalm => write!(f, "ENHANCE_YOUR_CALM"), 140 + Self::InadequateSecurity => write!(f, "INADEQUATE_SECURITY"), 141 + Self::Http11Required => write!(f, "HTTP_1_1_REQUIRED"), 142 + } 143 + } 144 + } 145 + 146 + // --------------------------------------------------------------------------- 147 + // Error type 148 + // --------------------------------------------------------------------------- 149 + 150 + #[derive(Debug)] 151 + pub enum Http2Error { 152 + /// I/O error during frame read/write. 153 + Io(io::Error), 154 + /// Frame exceeds maximum allowed size. 155 + FrameTooLarge(u32), 156 + /// Protocol violation. 157 + Protocol(String), 158 + /// Stream was reset by remote. 159 + StreamReset(ErrorCode), 160 + /// Connection-level GOAWAY received. 161 + GoAway(ErrorCode, String), 162 + /// Flow control window exceeded. 163 + FlowControl, 164 + /// HPACK header compression/decompression error. 165 + Compression(super::hpack::HpackError), 166 + /// Connection closed. 167 + ConnectionClosed, 168 + } 169 + 170 + impl fmt::Display for Http2Error { 171 + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { 172 + match self { 173 + Self::Io(e) => write!(f, "I/O error: {e}"), 174 + Self::FrameTooLarge(sz) => write!(f, "frame too large: {sz} bytes"), 175 + Self::Protocol(msg) => write!(f, "protocol error: {msg}"), 176 + Self::StreamReset(code) => write!(f, "stream reset: {code}"), 177 + Self::GoAway(code, msg) => write!(f, "GOAWAY: {code} {msg}"), 178 + Self::FlowControl => write!(f, "flow control error"), 179 + Self::Compression(e) => write!(f, "HPACK error: {e:?}"), 180 + Self::ConnectionClosed => write!(f, "connection closed"), 181 + } 182 + } 183 + } 184 + 185 + impl From<io::Error> for Http2Error { 186 + fn from(err: io::Error) -> Self { 187 + Self::Io(err) 188 + } 189 + } 190 + 191 + impl From<super::hpack::HpackError> for Http2Error { 192 + fn from(err: super::hpack::HpackError) -> Self { 193 + Self::Compression(err) 194 + } 195 + } 196 + 197 + pub type Result<T> = std::result::Result<T, Http2Error>; 198 + 199 + // --------------------------------------------------------------------------- 200 + // Frame header 201 + // --------------------------------------------------------------------------- 202 + 203 + /// Parsed HTTP/2 frame header (9 bytes on the wire). 204 + #[derive(Debug, Clone, PartialEq, Eq)] 205 + pub struct FrameHeader { 206 + /// Payload length (24-bit, max 16,777,215). 207 + pub length: u32, 208 + /// Frame type (DATA=0, HEADERS=1, etc.). 209 + pub frame_type: u8, 210 + /// Frame flags. 211 + pub flags: u8, 212 + /// Stream identifier (31-bit, high bit reserved). 213 + pub stream_id: u32, 214 + } 215 + 216 + impl FrameHeader { 217 + /// Encode the header into 9 bytes. 218 + pub fn encode(&self) -> [u8; 9] { 219 + let mut buf = [0u8; 9]; 220 + buf[0] = (self.length >> 16) as u8; 221 + buf[1] = (self.length >> 8) as u8; 222 + buf[2] = self.length as u8; 223 + buf[3] = self.frame_type; 224 + buf[4] = self.flags; 225 + let sid = self.stream_id & 0x7FFF_FFFF; // mask reserved bit 226 + buf[5] = (sid >> 24) as u8; 227 + buf[6] = (sid >> 16) as u8; 228 + buf[7] = (sid >> 8) as u8; 229 + buf[8] = sid as u8; 230 + buf 231 + } 232 + 233 + /// Decode a frame header from 9 bytes. 234 + pub fn decode(buf: &[u8; 9]) -> Self { 235 + let length = (buf[0] as u32) << 16 | (buf[1] as u32) << 8 | buf[2] as u32; 236 + let frame_type = buf[3]; 237 + let flags = buf[4]; 238 + let stream_id = 239 + (buf[5] as u32) << 24 | (buf[6] as u32) << 16 | (buf[7] as u32) << 8 | buf[8] as u32; 240 + let stream_id = stream_id & 0x7FFF_FFFF; // mask reserved bit 241 + Self { 242 + length, 243 + frame_type, 244 + flags, 245 + stream_id, 246 + } 247 + } 248 + 249 + pub fn has_flag(&self, flag: u8) -> bool { 250 + self.flags & flag != 0 251 + } 252 + } 253 + 254 + // --------------------------------------------------------------------------- 255 + // Frame (header + payload) 256 + // --------------------------------------------------------------------------- 257 + 258 + /// A complete HTTP/2 frame. 259 + #[derive(Debug, Clone)] 260 + pub struct Frame { 261 + pub header: FrameHeader, 262 + pub payload: Vec<u8>, 263 + } 264 + 265 + impl Frame { 266 + pub fn new(frame_type: u8, flags: u8, stream_id: u32, payload: Vec<u8>) -> Self { 267 + Self { 268 + header: FrameHeader { 269 + length: payload.len() as u32, 270 + frame_type, 271 + flags, 272 + stream_id, 273 + }, 274 + payload, 275 + } 276 + } 277 + 278 + /// Read a frame from a reader. 279 + pub fn read_from<R: Read>(reader: &mut R, max_frame_size: u32) -> Result<Self> { 280 + let mut header_buf = [0u8; 9]; 281 + read_exact(reader, &mut header_buf)?; 282 + let header = FrameHeader::decode(&header_buf); 283 + 284 + if header.length > max_frame_size { 285 + return Err(Http2Error::FrameTooLarge(header.length)); 286 + } 287 + 288 + let mut payload = vec![0u8; header.length as usize]; 289 + if !payload.is_empty() { 290 + read_exact(reader, &mut payload)?; 291 + } 292 + 293 + Ok(Self { header, payload }) 294 + } 295 + 296 + /// Write a frame to a writer. 297 + pub fn write_to<W: Write>(&self, writer: &mut W) -> Result<()> { 298 + let header_bytes = self.header.encode(); 299 + writer.write_all(&header_bytes)?; 300 + if !self.payload.is_empty() { 301 + writer.write_all(&self.payload)?; 302 + } 303 + Ok(()) 304 + } 305 + } 306 + 307 + /// Read exactly `buf.len()` bytes, returning ConnectionClosed on EOF. 308 + fn read_exact<R: Read>(reader: &mut R, buf: &mut [u8]) -> Result<()> { 309 + let mut offset = 0; 310 + while offset < buf.len() { 311 + match reader.read(&mut buf[offset..]) { 312 + Ok(0) => return Err(Http2Error::ConnectionClosed), 313 + Ok(n) => offset += n, 314 + Err(e) if e.kind() == io::ErrorKind::Interrupted => continue, 315 + Err(e) => return Err(Http2Error::Io(e)), 316 + } 317 + } 318 + Ok(()) 319 + } 320 + 321 + // --------------------------------------------------------------------------- 322 + // Frame constructors 323 + // --------------------------------------------------------------------------- 324 + 325 + /// Build a SETTINGS frame. 326 + pub fn settings_frame(settings: &[(u16, u32)], ack: bool) -> Frame { 327 + let flags = if ack { FLAG_ACK } else { 0 }; 328 + let mut payload = Vec::with_capacity(settings.len() * 6); 329 + if !ack { 330 + for &(id, value) in settings { 331 + payload.extend_from_slice(&id.to_be_bytes()); 332 + payload.extend_from_slice(&value.to_be_bytes()); 333 + } 334 + } 335 + Frame::new(FRAME_SETTINGS, flags, 0, payload) 336 + } 337 + 338 + /// Build a WINDOW_UPDATE frame. 339 + pub fn window_update_frame(stream_id: u32, increment: u32) -> Frame { 340 + let payload = (increment & 0x7FFF_FFFF).to_be_bytes().to_vec(); 341 + Frame::new(FRAME_WINDOW_UPDATE, 0, stream_id, payload) 342 + } 343 + 344 + /// Build a HEADERS frame with encoded header block. 345 + pub fn headers_frame(stream_id: u32, header_block: Vec<u8>, end_stream: bool) -> Frame { 346 + let mut flags = FLAG_END_HEADERS; 347 + if end_stream { 348 + flags |= FLAG_END_STREAM; 349 + } 350 + Frame::new(FRAME_HEADERS, flags, stream_id, header_block) 351 + } 352 + 353 + /// Build a DATA frame. 354 + pub fn data_frame(stream_id: u32, data: Vec<u8>, end_stream: bool) -> Frame { 355 + let flags = if end_stream { FLAG_END_STREAM } else { 0 }; 356 + Frame::new(FRAME_DATA, flags, stream_id, data) 357 + } 358 + 359 + /// Build a RST_STREAM frame. 360 + pub fn rst_stream_frame(stream_id: u32, error_code: ErrorCode) -> Frame { 361 + let payload = error_code.as_u32().to_be_bytes().to_vec(); 362 + Frame::new(FRAME_RST_STREAM, 0, stream_id, payload) 363 + } 364 + 365 + /// Build a GOAWAY frame. 366 + pub fn goaway_frame(last_stream_id: u32, error_code: ErrorCode) -> Frame { 367 + let mut payload = Vec::with_capacity(8); 368 + payload.extend_from_slice(&(last_stream_id & 0x7FFF_FFFF).to_be_bytes()); 369 + payload.extend_from_slice(&error_code.as_u32().to_be_bytes()); 370 + Frame::new(FRAME_GOAWAY, 0, 0, payload) 371 + } 372 + 373 + /// Build a PING frame. 374 + pub fn ping_frame(data: [u8; 8], ack: bool) -> Frame { 375 + let flags = if ack { FLAG_ACK } else { 0 }; 376 + Frame::new(FRAME_PING, flags, 0, data.to_vec()) 377 + } 378 + 379 + /// Build a SETTINGS ACK frame. 380 + pub fn settings_ack_frame() -> Frame { 381 + settings_frame(&[], true) 382 + } 383 + 384 + // --------------------------------------------------------------------------- 385 + // Settings parsing 386 + // --------------------------------------------------------------------------- 387 + 388 + /// A single SETTINGS parameter (id, value). 389 + #[derive(Debug, Clone, Copy, PartialEq, Eq)] 390 + pub struct Setting { 391 + pub id: u16, 392 + pub value: u32, 393 + } 394 + 395 + /// Parse SETTINGS frame payload into a list of settings. 396 + pub fn parse_settings(payload: &[u8]) -> Result<Vec<Setting>> { 397 + if !payload.len().is_multiple_of(6) { 398 + return Err(Http2Error::Protocol( 399 + "SETTINGS payload length not multiple of 6".to_string(), 400 + )); 401 + } 402 + let mut settings = Vec::with_capacity(payload.len() / 6); 403 + let mut offset = 0; 404 + while offset + 6 <= payload.len() { 405 + let id = u16::from_be_bytes([payload[offset], payload[offset + 1]]); 406 + let value = u32::from_be_bytes([ 407 + payload[offset + 2], 408 + payload[offset + 3], 409 + payload[offset + 4], 410 + payload[offset + 5], 411 + ]); 412 + settings.push(Setting { id, value }); 413 + offset += 6; 414 + } 415 + Ok(settings) 416 + } 417 + 418 + // --------------------------------------------------------------------------- 419 + // Tests 420 + // --------------------------------------------------------------------------- 421 + 422 + #[cfg(test)] 423 + mod tests { 424 + use super::*; 425 + 426 + // -- FrameHeader encode/decode -- 427 + 428 + #[test] 429 + fn frame_header_roundtrip() { 430 + let header = FrameHeader { 431 + length: 1024, 432 + frame_type: FRAME_DATA, 433 + flags: FLAG_END_STREAM, 434 + stream_id: 1, 435 + }; 436 + let encoded = header.encode(); 437 + let decoded = FrameHeader::decode(&encoded); 438 + assert_eq!(header, decoded); 439 + } 440 + 441 + #[test] 442 + fn frame_header_max_length() { 443 + let header = FrameHeader { 444 + length: MAX_FRAME_SIZE_LIMIT, 445 + frame_type: FRAME_DATA, 446 + flags: 0, 447 + stream_id: 0, 448 + }; 449 + let encoded = header.encode(); 450 + let decoded = FrameHeader::decode(&encoded); 451 + assert_eq!(decoded.length, MAX_FRAME_SIZE_LIMIT); 452 + } 453 + 454 + #[test] 455 + fn frame_header_masks_reserved_bit() { 456 + let mut buf = [0u8; 9]; 457 + // Set the reserved bit (high bit of stream ID) 458 + buf[5] = 0x80; 459 + buf[6] = 0x00; 460 + buf[7] = 0x00; 461 + buf[8] = 0x01; 462 + let header = FrameHeader::decode(&buf); 463 + assert_eq!(header.stream_id, 1); 464 + } 465 + 466 + #[test] 467 + fn frame_header_zero_stream() { 468 + let header = FrameHeader { 469 + length: 0, 470 + frame_type: FRAME_SETTINGS, 471 + flags: 0, 472 + stream_id: 0, 473 + }; 474 + let encoded = header.encode(); 475 + let decoded = FrameHeader::decode(&encoded); 476 + assert_eq!(decoded.stream_id, 0); 477 + } 478 + 479 + #[test] 480 + fn frame_header_has_flag() { 481 + let header = FrameHeader { 482 + length: 0, 483 + frame_type: FRAME_HEADERS, 484 + flags: FLAG_END_STREAM | FLAG_END_HEADERS, 485 + stream_id: 1, 486 + }; 487 + assert!(header.has_flag(FLAG_END_STREAM)); 488 + assert!(header.has_flag(FLAG_END_HEADERS)); 489 + assert!(!header.has_flag(FLAG_PADDED)); 490 + } 491 + 492 + // -- Frame read/write roundtrip -- 493 + 494 + #[test] 495 + fn frame_write_read_roundtrip() { 496 + let frame = Frame::new(FRAME_DATA, FLAG_END_STREAM, 3, vec![1, 2, 3, 4]); 497 + let mut buf = Vec::new(); 498 + frame.write_to(&mut buf).unwrap(); 499 + 500 + let mut cursor = &buf[..]; 501 + let read_frame = Frame::read_from(&mut cursor, DEFAULT_MAX_FRAME_SIZE).unwrap(); 502 + 503 + assert_eq!(read_frame.header, frame.header); 504 + assert_eq!(read_frame.payload, frame.payload); 505 + } 506 + 507 + #[test] 508 + fn frame_empty_payload() { 509 + let frame = Frame::new(FRAME_SETTINGS, FLAG_ACK, 0, vec![]); 510 + let mut buf = Vec::new(); 511 + frame.write_to(&mut buf).unwrap(); 512 + assert_eq!(buf.len(), FRAME_HEADER_SIZE); 513 + 514 + let mut cursor = &buf[..]; 515 + let read_frame = Frame::read_from(&mut cursor, DEFAULT_MAX_FRAME_SIZE).unwrap(); 516 + assert!(read_frame.payload.is_empty()); 517 + } 518 + 519 + #[test] 520 + fn frame_too_large_rejected() { 521 + let header = FrameHeader { 522 + length: DEFAULT_MAX_FRAME_SIZE + 1, 523 + frame_type: FRAME_DATA, 524 + flags: 0, 525 + stream_id: 1, 526 + }; 527 + let mut buf = Vec::new(); 528 + buf.extend_from_slice(&header.encode()); 529 + // Don't need actual payload - error happens before reading it 530 + buf.extend_from_slice(&vec![0u8; (DEFAULT_MAX_FRAME_SIZE + 1) as usize]); 531 + 532 + let mut cursor = &buf[..]; 533 + let result = Frame::read_from(&mut cursor, DEFAULT_MAX_FRAME_SIZE); 534 + assert!(matches!(result, Err(Http2Error::FrameTooLarge(_)))); 535 + } 536 + 537 + // -- Settings frame -- 538 + 539 + #[test] 540 + fn settings_frame_encoding() { 541 + let settings = vec![ 542 + (SETTINGS_MAX_CONCURRENT_STREAMS, 100), 543 + (SETTINGS_INITIAL_WINDOW_SIZE, 65535), 544 + ]; 545 + let frame = settings_frame(&settings, false); 546 + assert_eq!(frame.header.frame_type, FRAME_SETTINGS); 547 + assert_eq!(frame.header.stream_id, 0); 548 + assert_eq!(frame.header.flags, 0); 549 + assert_eq!(frame.payload.len(), 12); // 2 settings * 6 bytes 550 + } 551 + 552 + #[test] 553 + fn settings_ack_empty() { 554 + let frame = settings_ack_frame(); 555 + assert_eq!(frame.header.frame_type, FRAME_SETTINGS); 556 + assert!(frame.header.has_flag(FLAG_ACK)); 557 + assert!(frame.payload.is_empty()); 558 + } 559 + 560 + #[test] 561 + fn parse_settings_roundtrip() { 562 + let settings = vec![ 563 + (SETTINGS_HEADER_TABLE_SIZE, 4096), 564 + (SETTINGS_MAX_CONCURRENT_STREAMS, 128), 565 + (SETTINGS_INITIAL_WINDOW_SIZE, 1048576), 566 + (SETTINGS_MAX_FRAME_SIZE, 32768), 567 + ]; 568 + let frame = settings_frame(&settings, false); 569 + let parsed = parse_settings(&frame.payload).unwrap(); 570 + assert_eq!(parsed.len(), 4); 571 + assert_eq!( 572 + parsed[0], 573 + Setting { 574 + id: SETTINGS_HEADER_TABLE_SIZE, 575 + value: 4096 576 + } 577 + ); 578 + assert_eq!( 579 + parsed[1], 580 + Setting { 581 + id: SETTINGS_MAX_CONCURRENT_STREAMS, 582 + value: 128 583 + } 584 + ); 585 + assert_eq!( 586 + parsed[2], 587 + Setting { 588 + id: SETTINGS_INITIAL_WINDOW_SIZE, 589 + value: 1048576 590 + } 591 + ); 592 + assert_eq!( 593 + parsed[3], 594 + Setting { 595 + id: SETTINGS_MAX_FRAME_SIZE, 596 + value: 32768 597 + } 598 + ); 599 + } 600 + 601 + #[test] 602 + fn parse_settings_invalid_length() { 603 + let result = parse_settings(&[0, 1, 2, 3, 4]); 604 + assert!(result.is_err()); 605 + } 606 + 607 + // -- WINDOW_UPDATE frame -- 608 + 609 + #[test] 610 + fn window_update_encoding() { 611 + let frame = window_update_frame(1, 32768); 612 + assert_eq!(frame.header.frame_type, FRAME_WINDOW_UPDATE); 613 + assert_eq!(frame.header.stream_id, 1); 614 + assert_eq!(frame.payload.len(), 4); 615 + let increment = u32::from_be_bytes([ 616 + frame.payload[0], 617 + frame.payload[1], 618 + frame.payload[2], 619 + frame.payload[3], 620 + ]); 621 + assert_eq!(increment, 32768); 622 + } 623 + 624 + #[test] 625 + fn window_update_masks_reserved_bit() { 626 + let frame = window_update_frame(0, 0x8000_0001); 627 + let increment = u32::from_be_bytes([ 628 + frame.payload[0], 629 + frame.payload[1], 630 + frame.payload[2], 631 + frame.payload[3], 632 + ]); 633 + // High bit should be masked off 634 + assert_eq!(increment, 1); 635 + } 636 + 637 + // -- HEADERS frame -- 638 + 639 + #[test] 640 + fn headers_frame_with_end_stream() { 641 + let block = vec![0x82, 0x86]; // HPACK-encoded 642 + let frame = headers_frame(1, block.clone(), true); 643 + assert_eq!(frame.header.frame_type, FRAME_HEADERS); 644 + assert!(frame.header.has_flag(FLAG_END_HEADERS)); 645 + assert!(frame.header.has_flag(FLAG_END_STREAM)); 646 + assert_eq!(frame.payload, block); 647 + } 648 + 649 + #[test] 650 + fn headers_frame_without_end_stream() { 651 + let frame = headers_frame(3, vec![0x82], false); 652 + assert!(frame.header.has_flag(FLAG_END_HEADERS)); 653 + assert!(!frame.header.has_flag(FLAG_END_STREAM)); 654 + } 655 + 656 + // -- DATA frame -- 657 + 658 + #[test] 659 + fn data_frame_end_stream() { 660 + let frame = data_frame(1, vec![1, 2, 3], true); 661 + assert_eq!(frame.header.frame_type, FRAME_DATA); 662 + assert!(frame.header.has_flag(FLAG_END_STREAM)); 663 + assert_eq!(frame.payload, vec![1, 2, 3]); 664 + } 665 + 666 + #[test] 667 + fn data_frame_no_end_stream() { 668 + let frame = data_frame(1, vec![1, 2, 3], false); 669 + assert!(!frame.header.has_flag(FLAG_END_STREAM)); 670 + } 671 + 672 + // -- RST_STREAM frame -- 673 + 674 + #[test] 675 + fn rst_stream_encoding() { 676 + let frame = rst_stream_frame(5, ErrorCode::Cancel); 677 + assert_eq!(frame.header.frame_type, FRAME_RST_STREAM); 678 + assert_eq!(frame.header.stream_id, 5); 679 + let code = u32::from_be_bytes([ 680 + frame.payload[0], 681 + frame.payload[1], 682 + frame.payload[2], 683 + frame.payload[3], 684 + ]); 685 + assert_eq!(code, ErrorCode::Cancel.as_u32()); 686 + } 687 + 688 + // -- GOAWAY frame -- 689 + 690 + #[test] 691 + fn goaway_encoding() { 692 + let frame = goaway_frame(7, ErrorCode::NoError); 693 + assert_eq!(frame.header.frame_type, FRAME_GOAWAY); 694 + assert_eq!(frame.header.stream_id, 0); 695 + let last_id = u32::from_be_bytes([ 696 + frame.payload[0], 697 + frame.payload[1], 698 + frame.payload[2], 699 + frame.payload[3], 700 + ]); 701 + let code = u32::from_be_bytes([ 702 + frame.payload[4], 703 + frame.payload[5], 704 + frame.payload[6], 705 + frame.payload[7], 706 + ]); 707 + assert_eq!(last_id, 7); 708 + assert_eq!(code, 0); 709 + } 710 + 711 + // -- PING frame -- 712 + 713 + #[test] 714 + fn ping_frame_encoding() { 715 + let data = [1, 2, 3, 4, 5, 6, 7, 8]; 716 + let frame = ping_frame(data, false); 717 + assert_eq!(frame.header.frame_type, FRAME_PING); 718 + assert!(!frame.header.has_flag(FLAG_ACK)); 719 + assert_eq!(frame.payload, data); 720 + } 721 + 722 + #[test] 723 + fn ping_ack_frame() { 724 + let data = [0u8; 8]; 725 + let frame = ping_frame(data, true); 726 + assert!(frame.header.has_flag(FLAG_ACK)); 727 + } 728 + 729 + // -- ErrorCode -- 730 + 731 + #[test] 732 + fn error_code_roundtrip() { 733 + let codes = [ 734 + ErrorCode::NoError, 735 + ErrorCode::ProtocolError, 736 + ErrorCode::InternalError, 737 + ErrorCode::FlowControlError, 738 + ErrorCode::SettingsTimeout, 739 + ErrorCode::StreamClosed, 740 + ErrorCode::FrameSizeError, 741 + ErrorCode::RefusedStream, 742 + ErrorCode::Cancel, 743 + ErrorCode::CompressionError, 744 + ErrorCode::ConnectError, 745 + ErrorCode::EnhanceYourCalm, 746 + ErrorCode::InadequateSecurity, 747 + ErrorCode::Http11Required, 748 + ]; 749 + for code in codes { 750 + assert_eq!(ErrorCode::from_u32(code.as_u32()), code); 751 + } 752 + } 753 + 754 + #[test] 755 + fn error_code_unknown_maps_to_internal() { 756 + assert_eq!(ErrorCode::from_u32(0xFF), ErrorCode::InternalError); 757 + } 758 + 759 + // -- Connection preface -- 760 + 761 + #[test] 762 + fn connection_preface_correct() { 763 + assert_eq!(CONNECTION_PREFACE, b"PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n"); 764 + assert_eq!(CONNECTION_PREFACE.len(), 24); 765 + } 766 + }

+7 -1

crates/net/src/http2/mod.rs

··· 1 - //! HTTP/2 protocol implementation. 1 + //! HTTP/2 protocol implementation (RFC 7540/9113). 2 + //! 3 + //! Provides binary framing, stream multiplexing, flow control, 4 + //! HPACK header compression, and an HTTP/2 connection abstraction. 2 5 6 + pub mod connection; 7 + pub mod frame; 3 8 pub mod hpack; 9 + pub mod stream;

+347

crates/net/src/http2/stream.rs

··· 1 + //! HTTP/2 stream state machine (RFC 7540 §5.1). 2 + //! 3 + //! Tracks per-stream state transitions: idle → open → half-closed → closed. 4 + 5 + use super::frame::{ErrorCode, Http2Error, Result}; 6 + 7 + // --------------------------------------------------------------------------- 8 + // Default flow control window 9 + // --------------------------------------------------------------------------- 10 + 11 + /// Default initial window size per RFC 7540 §6.9.2. 12 + pub const DEFAULT_INITIAL_WINDOW_SIZE: u32 = 65535; 13 + 14 + // --------------------------------------------------------------------------- 15 + // Stream state (RFC 7540 §5.1) 16 + // --------------------------------------------------------------------------- 17 + 18 + /// State of an HTTP/2 stream. 19 + #[derive(Debug, Clone, Copy, PartialEq, Eq)] 20 + pub enum StreamState { 21 + /// Stream ID has been reserved but no frames sent/received. 22 + Idle, 23 + /// HEADERS sent/received; both sides can send frames. 24 + Open, 25 + /// Local side has sent END_STREAM; can still receive. 26 + HalfClosedLocal, 27 + /// Remote side has sent END_STREAM; can still send. 28 + HalfClosedRemote, 29 + /// Both sides have sent END_STREAM or RST_STREAM received. 30 + Closed, 31 + } 32 + 33 + // --------------------------------------------------------------------------- 34 + // Stream 35 + // --------------------------------------------------------------------------- 36 + 37 + /// An HTTP/2 stream with state, flow control, and buffered data. 38 + pub struct Stream { 39 + /// Stream identifier (odd = client-initiated). 40 + pub id: u32, 41 + /// Current state. 42 + pub state: StreamState, 43 + /// Send flow control window (how many bytes we can still send). 44 + pub send_window: i64, 45 + /// Receive flow control window (how many bytes the peer can still send). 46 + pub recv_window: i64, 47 + /// Accumulated response header block fragments. 48 + pub header_block: Vec<u8>, 49 + /// Decoded response headers (populated after END_HEADERS). 50 + pub response_headers: Vec<(Vec<u8>, Vec<u8>)>, 51 + /// Accumulated response body data. 52 + pub body: Vec<u8>, 53 + /// HTTP status code from :status pseudo-header. 54 + pub status_code: Option<u16>, 55 + } 56 + 57 + impl Stream { 58 + /// Create a new client-initiated stream. 59 + pub fn new(id: u32, initial_send_window: u32, initial_recv_window: u32) -> Self { 60 + Self { 61 + id, 62 + state: StreamState::Idle, 63 + send_window: initial_send_window as i64, 64 + recv_window: initial_recv_window as i64, 65 + header_block: Vec::new(), 66 + response_headers: Vec::new(), 67 + body: Vec::new(), 68 + status_code: None, 69 + } 70 + } 71 + 72 + /// Transition to Open state (when HEADERS is sent). 73 + pub fn send_headers(&mut self) -> Result<()> { 74 + match self.state { 75 + StreamState::Idle => { 76 + self.state = StreamState::Open; 77 + Ok(()) 78 + } 79 + _ => Err(Http2Error::Protocol(format!( 80 + "cannot send HEADERS in state {:?}", 81 + self.state 82 + ))), 83 + } 84 + } 85 + 86 + /// Handle sending END_STREAM. 87 + pub fn send_end_stream(&mut self) -> Result<()> { 88 + match self.state { 89 + StreamState::Open => { 90 + self.state = StreamState::HalfClosedLocal; 91 + Ok(()) 92 + } 93 + StreamState::HalfClosedRemote => { 94 + self.state = StreamState::Closed; 95 + Ok(()) 96 + } 97 + _ => Err(Http2Error::Protocol(format!( 98 + "cannot send END_STREAM in state {:?}", 99 + self.state 100 + ))), 101 + } 102 + } 103 + 104 + /// Handle receiving END_STREAM from remote. 105 + pub fn recv_end_stream(&mut self) -> Result<()> { 106 + match self.state { 107 + StreamState::Open => { 108 + self.state = StreamState::HalfClosedRemote; 109 + Ok(()) 110 + } 111 + StreamState::HalfClosedLocal => { 112 + self.state = StreamState::Closed; 113 + Ok(()) 114 + } 115 + _ => Err(Http2Error::Protocol(format!( 116 + "received END_STREAM in state {:?}", 117 + self.state 118 + ))), 119 + } 120 + } 121 + 122 + /// Handle receiving RST_STREAM. 123 + pub fn recv_rst_stream(&mut self, error_code: ErrorCode) -> Result<()> { 124 + self.state = StreamState::Closed; 125 + if error_code != ErrorCode::NoError { 126 + return Err(Http2Error::StreamReset(error_code)); 127 + } 128 + Ok(()) 129 + } 130 + 131 + /// Consume bytes from the send window. Returns error if window is exhausted. 132 + pub fn consume_send_window(&mut self, bytes: u32) -> Result<()> { 133 + self.send_window -= bytes as i64; 134 + if self.send_window < 0 { 135 + return Err(Http2Error::FlowControl); 136 + } 137 + Ok(()) 138 + } 139 + 140 + /// Consume bytes from the receive window. 141 + pub fn consume_recv_window(&mut self, bytes: u32) -> Result<()> { 142 + self.recv_window -= bytes as i64; 143 + if self.recv_window < 0 { 144 + return Err(Http2Error::FlowControl); 145 + } 146 + Ok(()) 147 + } 148 + 149 + /// Increase the send window (on WINDOW_UPDATE from peer). 150 + pub fn increase_send_window(&mut self, increment: u32) -> Result<()> { 151 + self.send_window += increment as i64; 152 + // RFC 7540 §6.9.1: window size must not exceed 2^31 - 1 153 + if self.send_window > 0x7FFF_FFFF { 154 + return Err(Http2Error::FlowControl); 155 + } 156 + Ok(()) 157 + } 158 + 159 + /// Increase the receive window (when we send WINDOW_UPDATE). 160 + pub fn increase_recv_window(&mut self, increment: u32) { 161 + self.recv_window += increment as i64; 162 + } 163 + 164 + /// Check if the stream can receive data. 165 + pub fn can_recv(&self) -> bool { 166 + matches!(self.state, StreamState::Open | StreamState::HalfClosedLocal) 167 + } 168 + 169 + /// Check if the stream is done (fully closed). 170 + pub fn is_closed(&self) -> bool { 171 + self.state == StreamState::Closed 172 + } 173 + } 174 + 175 + // --------------------------------------------------------------------------- 176 + // Tests 177 + // --------------------------------------------------------------------------- 178 + 179 + #[cfg(test)] 180 + mod tests { 181 + use super::*; 182 + 183 + #[test] 184 + fn new_stream_is_idle() { 185 + let s = Stream::new(1, 65535, 65535); 186 + assert_eq!(s.state, StreamState::Idle); 187 + assert_eq!(s.id, 1); 188 + } 189 + 190 + #[test] 191 + fn idle_to_open_on_send_headers() { 192 + let mut s = Stream::new(1, 65535, 65535); 193 + s.send_headers().unwrap(); 194 + assert_eq!(s.state, StreamState::Open); 195 + } 196 + 197 + #[test] 198 + fn open_to_half_closed_local_on_send_end_stream() { 199 + let mut s = Stream::new(1, 65535, 65535); 200 + s.send_headers().unwrap(); 201 + s.send_end_stream().unwrap(); 202 + assert_eq!(s.state, StreamState::HalfClosedLocal); 203 + } 204 + 205 + #[test] 206 + fn open_to_half_closed_remote_on_recv_end_stream() { 207 + let mut s = Stream::new(1, 65535, 65535); 208 + s.send_headers().unwrap(); 209 + s.recv_end_stream().unwrap(); 210 + assert_eq!(s.state, StreamState::HalfClosedRemote); 211 + } 212 + 213 + #[test] 214 + fn half_closed_local_to_closed_on_recv_end_stream() { 215 + let mut s = Stream::new(1, 65535, 65535); 216 + s.send_headers().unwrap(); 217 + s.send_end_stream().unwrap(); 218 + s.recv_end_stream().unwrap(); 219 + assert_eq!(s.state, StreamState::Closed); 220 + } 221 + 222 + #[test] 223 + fn half_closed_remote_to_closed_on_send_end_stream() { 224 + let mut s = Stream::new(1, 65535, 65535); 225 + s.send_headers().unwrap(); 226 + s.recv_end_stream().unwrap(); 227 + s.send_end_stream().unwrap(); 228 + assert_eq!(s.state, StreamState::Closed); 229 + } 230 + 231 + #[test] 232 + fn rst_stream_closes() { 233 + let mut s = Stream::new(1, 65535, 65535); 234 + s.send_headers().unwrap(); 235 + s.recv_rst_stream(ErrorCode::NoError).unwrap(); 236 + assert_eq!(s.state, StreamState::Closed); 237 + } 238 + 239 + #[test] 240 + fn rst_stream_with_error() { 241 + let mut s = Stream::new(1, 65535, 65535); 242 + s.send_headers().unwrap(); 243 + let result = s.recv_rst_stream(ErrorCode::Cancel); 244 + assert!(matches!( 245 + result, 246 + Err(Http2Error::StreamReset(ErrorCode::Cancel)) 247 + )); 248 + assert_eq!(s.state, StreamState::Closed); 249 + } 250 + 251 + #[test] 252 + fn send_headers_in_non_idle_fails() { 253 + let mut s = Stream::new(1, 65535, 65535); 254 + s.send_headers().unwrap(); 255 + assert!(s.send_headers().is_err()); 256 + } 257 + 258 + #[test] 259 + fn send_end_stream_in_idle_fails() { 260 + let mut s = Stream::new(1, 65535, 65535); 261 + assert!(s.send_end_stream().is_err()); 262 + } 263 + 264 + #[test] 265 + fn recv_end_stream_in_idle_fails() { 266 + let mut s = Stream::new(1, 65535, 65535); 267 + assert!(s.recv_end_stream().is_err()); 268 + } 269 + 270 + // -- Flow control -- 271 + 272 + #[test] 273 + fn consume_send_window() { 274 + let mut s = Stream::new(1, 65535, 65535); 275 + s.consume_send_window(1000).unwrap(); 276 + assert_eq!(s.send_window, 64535); 277 + } 278 + 279 + #[test] 280 + fn consume_send_window_overflow() { 281 + let mut s = Stream::new(1, 100, 65535); 282 + assert!(s.consume_send_window(101).is_err()); 283 + } 284 + 285 + #[test] 286 + fn consume_recv_window() { 287 + let mut s = Stream::new(1, 65535, 65535); 288 + s.consume_recv_window(5000).unwrap(); 289 + assert_eq!(s.recv_window, 60535); 290 + } 291 + 292 + #[test] 293 + fn increase_send_window() { 294 + let mut s = Stream::new(1, 65535, 65535); 295 + s.consume_send_window(1000).unwrap(); 296 + s.increase_send_window(500).unwrap(); 297 + assert_eq!(s.send_window, 65035); 298 + } 299 + 300 + #[test] 301 + fn increase_send_window_overflow() { 302 + let mut s = Stream::new(1, 0x7FFF_FFFF, 65535); 303 + assert!(s.increase_send_window(1).is_err()); 304 + } 305 + 306 + #[test] 307 + fn increase_recv_window() { 308 + let mut s = Stream::new(1, 65535, 65535); 309 + s.consume_recv_window(1000).unwrap(); 310 + s.increase_recv_window(1000); 311 + assert_eq!(s.recv_window, 65535); 312 + } 313 + 314 + #[test] 315 + fn can_recv_in_open_state() { 316 + let mut s = Stream::new(1, 65535, 65535); 317 + s.send_headers().unwrap(); 318 + assert!(s.can_recv()); 319 + } 320 + 321 + #[test] 322 + fn can_recv_in_half_closed_local() { 323 + let mut s = Stream::new(1, 65535, 65535); 324 + s.send_headers().unwrap(); 325 + s.send_end_stream().unwrap(); 326 + assert!(s.can_recv()); 327 + } 328 + 329 + #[test] 330 + fn cannot_recv_in_half_closed_remote() { 331 + let mut s = Stream::new(1, 65535, 65535); 332 + s.send_headers().unwrap(); 333 + s.recv_end_stream().unwrap(); 334 + assert!(!s.can_recv()); 335 + } 336 + 337 + #[test] 338 + fn is_closed() { 339 + let mut s = Stream::new(1, 65535, 65535); 340 + assert!(!s.is_closed()); 341 + s.send_headers().unwrap(); 342 + assert!(!s.is_closed()); 343 + s.send_end_stream().unwrap(); 344 + s.recv_end_stream().unwrap(); 345 + assert!(s.is_closed()); 346 + } 347 + }

+170 -27

crates/net/src/tls/handshake.rs

··· 41 41 const EXT_SERVER_NAME: u16 = 0; 42 42 const EXT_SUPPORTED_GROUPS: u16 = 10; 43 43 const EXT_SIGNATURE_ALGORITHMS: u16 = 13; 44 + const EXT_ALPN: u16 = 16; 44 45 const EXT_SUPPORTED_VERSIONS: u16 = 43; 45 46 const EXT_KEY_SHARE: u16 = 51; 46 47 ··· 210 211 /// Build a ClientHello handshake message. 211 212 /// 212 213 /// Returns (handshake_message, x25519_private_key). 213 - fn build_client_hello(server_name: &str) -> (Vec<u8>, [u8; 32]) { 214 + fn build_client_hello(server_name: &str, alpn_protocols: &[&str]) -> (Vec<u8>, [u8; 32]) { 214 215 // Generate X25519 ephemeral keypair 215 216 let mut private_key = [0u8; 32]; 216 217 random_bytes(&mut private_key); ··· 248 249 push_u8(&mut body, 0); // null 249 250 250 251 // Extensions 251 - let extensions = build_extensions(server_name, &public_key); 252 + let extensions = build_extensions(server_name, &public_key, alpn_protocols); 252 253 push_u16(&mut body, extensions.len() as u16); 253 254 push_bytes(&mut body, &extensions); 254 255 ··· 261 262 (msg, private_key) 262 263 } 263 264 264 - fn build_extensions(server_name: &str, x25519_public: &[u8; 32]) -> Vec<u8> { 265 + fn build_extensions( 266 + server_name: &str, 267 + x25519_public: &[u8; 32], 268 + alpn_protocols: &[&str], 269 + ) -> Vec<u8> { 265 270 let mut exts = Vec::with_capacity(256); 266 271 267 272 // SNI extension (server_name) ··· 324 329 for alg in sig_algs { 325 330 push_u16(&mut exts, alg); 326 331 } 332 + } 333 + 334 + // ALPN extension (RFC 7301) 335 + if !alpn_protocols.is_empty() { 336 + let mut protocol_list = Vec::new(); 337 + for proto in alpn_protocols { 338 + let bytes = proto.as_bytes(); 339 + protocol_list.push(bytes.len() as u8); 340 + protocol_list.extend_from_slice(bytes); 341 + } 342 + push_u16(&mut exts, EXT_ALPN); 343 + push_u16(&mut exts, (2 + protocol_list.len()) as u16); // extension data length 344 + push_u16(&mut exts, protocol_list.len() as u16); // protocol list length 345 + push_bytes(&mut exts, &protocol_list); 327 346 } 328 347 329 348 exts ··· 416 435 // Encrypted handshake message parsing 417 436 // --------------------------------------------------------------------------- 418 437 419 - fn parse_encrypted_extensions(data: &[u8]) -> Result<()> { 438 + fn parse_encrypted_extensions(data: &[u8]) -> Result<Option<String>> { 420 439 let mut offset = 0; 421 - let _extensions_len = read_u16(data, &mut offset)?; 422 - // We don't require any specific encrypted extensions for now. 423 - // Just validate the format is parseable. 424 - Ok(()) 440 + let extensions_len = read_u16(data, &mut offset)? as usize; 441 + let extensions_end = offset + extensions_len; 442 + let mut alpn_protocol = None; 443 + 444 + while offset < extensions_end { 445 + let ext_type = read_u16(data, &mut offset)?; 446 + let ext_len = read_u16(data, &mut offset)? as usize; 447 + let ext_data = read_bytes(data, &mut offset, ext_len)?; 448 + 449 + if ext_type == EXT_ALPN { 450 + // Parse ALPN response: protocol_list_len(2) + protocol_len(1) + protocol 451 + let mut eoff = 0; 452 + let _list_len = read_u16(ext_data, &mut eoff)?; 453 + let proto_len = read_u8(ext_data, &mut eoff)? as usize; 454 + let proto = read_bytes(ext_data, &mut eoff, proto_len)?; 455 + alpn_protocol = Some( 456 + std::str::from_utf8(proto) 457 + .map_err(|_| HandshakeError::Malformed("invalid ALPN protocol UTF-8"))? 458 + .to_string(), 459 + ); 460 + } 461 + } 462 + 463 + Ok(alpn_protocol) 425 464 } 426 465 427 466 /// Parse a Certificate handshake message (RFC 8446 §4.4.2). ··· 652 691 } 653 692 } 654 693 694 + impl<S: Read + Write> io::Read for TlsStream<S> { 695 + fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> { 696 + TlsStream::read(self, buf).map_err(|e| io::Error::other(e.to_string())) 697 + } 698 + } 699 + 700 + impl<S: Read + Write> io::Write for TlsStream<S> { 701 + fn write(&mut self, buf: &[u8]) -> io::Result<usize> { 702 + TlsStream::write(self, buf).map_err(|e| io::Error::other(e.to_string())) 703 + } 704 + 705 + fn flush(&mut self) -> io::Result<()> { 706 + // TLS writes are flushed per record 707 + Ok(()) 708 + } 709 + } 710 + 655 711 // --------------------------------------------------------------------------- 656 712 // Handshake state machine 657 713 // --------------------------------------------------------------------------- ··· 660 716 /// 661 717 /// Returns a `TlsStream` ready for application data. 662 718 pub fn connect<S: Read + Write>(stream: S, server_name: &str) -> Result<TlsStream<S>> { 719 + let (tls_stream, _alpn) = connect_with_alpn(stream, server_name, &[])?; 720 + Ok(tls_stream) 721 + } 722 + 723 + /// Perform a TLS 1.3 handshake with ALPN negotiation. 724 + /// 725 + /// Returns a `TlsStream` and the negotiated ALPN protocol (if any). 726 + pub fn connect_with_alpn<S: Read + Write>( 727 + stream: S, 728 + server_name: &str, 729 + alpn_protocols: &[&str], 730 + ) -> Result<(TlsStream<S>, Option<String>)> { 663 731 let mut record_layer = RecordLayer::new(stream); 664 732 665 733 // Step 1: Build and send ClientHello 666 - let (client_hello_msg, x25519_private) = build_client_hello(server_name); 734 + let (client_hello_msg, x25519_private) = build_client_hello(server_name, alpn_protocols); 667 735 668 736 let ch_record = TlsRecord::new(ContentType::Handshake, client_hello_msg.clone()); 669 737 record_layer.write_record(&ch_record)?; ··· 702 770 if ee_type != HANDSHAKE_ENCRYPTED_EXTENSIONS { 703 771 return Err(HandshakeError::UnexpectedMessage(ee_type)); 704 772 } 705 - parse_encrypted_extensions(&ee_body)?; 773 + let alpn_protocol = parse_encrypted_extensions(&ee_body)?; 706 774 transcript.update(&ee_full); 707 775 708 776 // Step 6: Read Certificate ··· 806 874 server_app_iv, 807 875 )); 808 876 809 - Ok(TlsStream { 810 - record_layer, 811 - read_buffer: Vec::new(), 812 - read_offset: 0, 813 - }) 877 + Ok(( 878 + TlsStream { 879 + record_layer, 880 + read_buffer: Vec::new(), 881 + read_offset: 0, 882 + }, 883 + alpn_protocol, 884 + )) 814 885 } 815 886 816 887 // --------------------------------------------------------------------------- ··· 982 1053 983 1054 #[test] 984 1055 fn client_hello_has_correct_type() { 985 - let (msg, _) = build_client_hello("example.com"); 1056 + let (msg, _) = build_client_hello("example.com", &[]); 986 1057 assert_eq!(msg[0], HANDSHAKE_CLIENT_HELLO); 987 1058 } 988 1059 989 1060 #[test] 990 1061 fn client_hello_has_valid_length() { 991 - let (msg, _) = build_client_hello("example.com"); 1062 + let (msg, _) = build_client_hello("example.com", &[]); 992 1063 let body_len = (msg[1] as usize) << 16 | (msg[2] as usize) << 8 | msg[3] as usize; 993 1064 assert_eq!(msg.len(), 4 + body_len); 994 1065 } 995 1066 996 1067 #[test] 997 1068 fn client_hello_starts_with_legacy_version() { 998 - let (msg, _) = build_client_hello("example.com"); 1069 + let (msg, _) = build_client_hello("example.com", &[]); 999 1070 assert_eq!(msg[4], 0x03); 1000 1071 assert_eq!(msg[5], 0x03); 1001 1072 } 1002 1073 1003 1074 #[test] 1004 1075 fn client_hello_has_32_byte_random() { 1005 - let (msg1, _) = build_client_hello("example.com"); 1006 - let (msg2, _) = build_client_hello("example.com"); 1076 + let (msg1, _) = build_client_hello("example.com", &[]); 1077 + let (msg2, _) = build_client_hello("example.com", &[]); 1007 1078 // Random bytes start at offset 6 (after type(1) + length(3) + version(2)) 1008 1079 let random1 = &msg1[6..38]; 1009 1080 let random2 = &msg2[6..38]; ··· 1015 1086 1016 1087 #[test] 1017 1088 fn client_hello_has_session_id() { 1018 - let (msg, _) = build_client_hello("example.com"); 1089 + let (msg, _) = build_client_hello("example.com", &[]); 1019 1090 // session_id_len at offset 38 1020 1091 assert_eq!(msg[38], 32); // 32-byte session ID 1021 1092 } 1022 1093 1023 1094 #[test] 1024 1095 fn client_hello_has_cipher_suites() { 1025 - let (msg, _) = build_client_hello("example.com"); 1096 + let (msg, _) = build_client_hello("example.com", &[]); 1026 1097 // After version(2) + random(32) + session_id_len(1) + session_id(32) 1027 1098 let cs_offset = 4 + 2 + 32 + 1 + 32; 1028 1099 let cs_len = u16::from_be_bytes([msg[cs_offset], msg[cs_offset + 1]]); ··· 1031 1102 1032 1103 #[test] 1033 1104 fn client_hello_returns_private_key() { 1034 - let (_, private_key) = build_client_hello("example.com"); 1105 + let (_, private_key) = build_client_hello("example.com", &[]); 1035 1106 assert_eq!(private_key.len(), 32); 1036 1107 } 1037 1108 ··· 1287 1358 #[test] 1288 1359 fn extensions_contain_sni() { 1289 1360 let key = [0u8; 32]; 1290 - let exts = build_extensions("example.com", &key); 1361 + let exts = build_extensions("example.com", &key, &[]); 1291 1362 // First extension should be SNI (type 0x0000) 1292 1363 assert_eq!(exts[0], 0x00); 1293 1364 assert_eq!(exts[1], 0x00); ··· 1296 1367 #[test] 1297 1368 fn extensions_contain_supported_versions() { 1298 1369 let key = [0u8; 32]; 1299 - let exts = build_extensions("example.com", &key); 1370 + let exts = build_extensions("example.com", &key, &[]); 1300 1371 // Search for supported_versions extension type (0x002b = 43) 1301 1372 let mut found = false; 1302 1373 let mut i = 0; ··· 1315 1386 #[test] 1316 1387 fn extensions_contain_key_share() { 1317 1388 let key = [0x42u8; 32]; 1318 - let exts = build_extensions("example.com", &key); 1389 + let exts = build_extensions("example.com", &key, &[]); 1319 1390 let mut found = false; 1320 1391 let mut i = 0; 1321 1392 while i + 4 <= exts.len() { ··· 1337 1408 #[test] 1338 1409 fn extensions_contain_sig_algorithms() { 1339 1410 let key = [0u8; 32]; 1340 - let exts = build_extensions("example.com", &key); 1411 + let exts = build_extensions("example.com", &key, &[]); 1341 1412 let mut found = false; 1342 1413 let mut i = 0; 1343 1414 while i + 4 <= exts.len() { ··· 1381 1452 let result = parse_server_hello(&body).unwrap(); 1382 1453 assert_eq!(result.cipher_suite, CipherSuite::Aes128Gcm); 1383 1454 assert_eq!(result.server_x25519_public, key); 1455 + } 1456 + 1457 + // -- ALPN extension -- 1458 + 1459 + #[test] 1460 + fn extensions_contain_alpn_when_requested() { 1461 + let key = [0u8; 32]; 1462 + let exts = build_extensions("example.com", &key, &["h2", "http/1.1"]); 1463 + let mut found = false; 1464 + let mut i = 0; 1465 + while i + 4 <= exts.len() { 1466 + let ext_type = u16::from_be_bytes([exts[i], exts[i + 1]]); 1467 + let ext_len = u16::from_be_bytes([exts[i + 2], exts[i + 3]]) as usize; 1468 + if ext_type == EXT_ALPN { 1469 + found = true; 1470 + // Verify protocol list structure 1471 + let ext_data = &exts[i + 4..i + 4 + ext_len]; 1472 + let list_len = u16::from_be_bytes([ext_data[0], ext_data[1]]) as usize; 1473 + assert_eq!(list_len, ext_data.len() - 2); 1474 + // First protocol: "h2" (length 2) 1475 + assert_eq!(ext_data[2], 2); 1476 + assert_eq!(&ext_data[3..5], b"h2"); 1477 + // Second protocol: "http/1.1" (length 8) 1478 + assert_eq!(ext_data[5], 8); 1479 + assert_eq!(&ext_data[6..14], b"http/1.1"); 1480 + break; 1481 + } 1482 + i += 4 + ext_len; 1483 + } 1484 + assert!(found, "ALPN extension not found"); 1485 + } 1486 + 1487 + #[test] 1488 + fn extensions_no_alpn_when_empty() { 1489 + let key = [0u8; 32]; 1490 + let exts = build_extensions("example.com", &key, &[]); 1491 + let mut i = 0; 1492 + while i + 4 <= exts.len() { 1493 + let ext_type = u16::from_be_bytes([exts[i], exts[i + 1]]); 1494 + let ext_len = u16::from_be_bytes([exts[i + 2], exts[i + 3]]) as usize; 1495 + assert_ne!(ext_type, EXT_ALPN, "ALPN extension should not be present"); 1496 + i += 4 + ext_len; 1497 + } 1498 + } 1499 + 1500 + #[test] 1501 + fn parse_encrypted_extensions_with_alpn() { 1502 + let mut data = Vec::new(); 1503 + // Build extension data with ALPN response 1504 + let mut ext_body = Vec::new(); 1505 + // ALPN extension 1506 + push_u16(&mut ext_body, EXT_ALPN); 1507 + let proto = b"h2"; 1508 + let alpn_data_len = 2 + 1 + proto.len(); 1509 + push_u16(&mut ext_body, alpn_data_len as u16); 1510 + push_u16(&mut ext_body, (1 + proto.len()) as u16); // protocol list length 1511 + push_u8(&mut ext_body, proto.len() as u8); 1512 + push_bytes(&mut ext_body, proto); 1513 + 1514 + push_u16(&mut data, ext_body.len() as u16); 1515 + push_bytes(&mut data, &ext_body); 1516 + 1517 + let result = parse_encrypted_extensions(&data).unwrap(); 1518 + assert_eq!(result, Some("h2".to_string())); 1519 + } 1520 + 1521 + #[test] 1522 + fn parse_encrypted_extensions_no_alpn() { 1523 + let mut data = Vec::new(); 1524 + push_u16(&mut data, 0); // empty extensions 1525 + let result = parse_encrypted_extensions(&data).unwrap(); 1526 + assert_eq!(result, None); 1384 1527 } 1385 1528 1386 1529 // -- current_datetime sanity check --

Configure Feed

Configure Feed