diff --git a/src/endpoint.rs b/src/endpoint.rs index f6adec6ef..17d864c42 100644 --- a/src/endpoint.rs +++ b/src/endpoint.rs @@ -1,5 +1,5 @@ use std::collections::{HashMap, VecDeque, BTreeMap}; -use std::{io, cmp}; +use std::{io, cmp, fmt}; use std::net::SocketAddrV6; use std::sync::Arc; use std::time::Duration; @@ -233,7 +233,7 @@ impl Endpoint { Header::VersionNegotiate { id }.encode(&mut buf); buf.put_u32::(0x0a1a2a3a); // reserved version buf.put_u32::(VERSION); // supported version - self.transmit(remote, buf.into()); + self.io.push_back(Io::Transmit { destination: remote, packet: buf.into() }); return; } Err(_) => { @@ -261,13 +261,13 @@ impl Endpoint { // if self.config.listen.is_none() { - trace!(self.log, "dropping packet from unknown connection"); + debug!(self.log, "dropping packet from unrecognized connection"; "header" => ?packet.header); return; } if let Header::Long { ty, id, number } = packet.header { // MAY buffer non-initial packets a little for better 0RTT behavior if ty == packet::INITIAL && data.len() >= MIN_INITIAL_SIZE { - self.handle_initial(remote, id, number, packet.payload); + self.handle_initial(now, remote, id, number, packet.payload); return; } } @@ -290,11 +290,11 @@ impl Endpoint { self.rng.fill_bytes(&mut buf[start..start+padding]); } buf.extend(&reset_token_for(&self.state.reset_key, id)); - self.transmit(remote, buf.into()); + self.io.push_back(Io::Transmit { destination: remote, packet: buf.into() }); } } - pub fn connect(&mut self, local: SocketAddrV6, remote: SocketAddrV6) -> Result { + pub fn connect(&mut self, now: u64, local: SocketAddrV6, remote: SocketAddrV6) -> Result { let mut tls = Ssl::new(&self.tls)?; let id = self.rng.gen(); tls.set_ex_data(*CONNECTION_INFO_INDEX, ConnectionInfo { id, remote }); @@ -304,48 +304,25 @@ impl Endpoint { Err(e) => return Err(e.into()), }; let conn = self.add_connection(id); - - let mut buf = Vec::::new(); - buf.reserve_exact(MIN_INITIAL_SIZE); - let packet = self.connections[conn.0].get_tx_number() as u32; - encode_long_header(&mut buf, packet::INITIAL, id, packet); - let client_hello = tls.get_mut().take_outgoing().to_vec(); - frame::stream(&mut buf, StreamId(0), None, true, false, &client_hello); + self.connections[conn.0].client = true; + let packet = self.transmit_handshake(now, conn, remote, (&tls.get_mut().take_outgoing()[..]).into()); self.connections[conn.0].state = Some(State::Handshake(state::Handshake { tls, clienthello_packet: Some(packet), })); - if buf.len() < MIN_INITIAL_SIZE { - buf.resize(MIN_INITIAL_SIZE, frame::Type::PADDING.into()); - } - self.transmit(remote, buf.into()); Ok(conn) } - fn transmit(&mut self, destination: SocketAddrV6, packet: Box<[u8]>) { - self.io.push_back(Io::Transmit { destination, packet }); - } - fn add_connection(&mut self, id: ConnectionId) -> ConnectionHandle { let i = self.connections.insert(Connection::new(id, self.initial_packet_number.sample(&mut self.rng).into(), &self.config)); self.connection_ids.insert(id, ConnectionHandle(i)); ConnectionHandle(i) } - fn handle_initial(&mut self, remote: SocketAddrV6, id: ConnectionId, packet_number: u32, payload: Bytes) { + fn handle_initial(&mut self, now: u64, remote: SocketAddrV6, id: ConnectionId, packet_number: u32, payload: Bytes) { let mut stream = MemoryStream::new(); - // TODO: Proper reassembly - for frame in frame::Iter::new(payload) { - match frame { - Frame::Padding => {} - Frame::Stream(frame::Stream { id, data, .. }) => { - if id != StreamId(0) { return; } // Invalid packet - stream.extend_incoming(&data[..]); - } - _ => { return; } // Invalid packet - } - } - + if !parse_initial(&mut stream, payload) { return; } // TODO: Send close? + let offset = stream.incoming_len() as u64; let mut tls = Ssl::new(&self.tls).unwrap(); // TODO: is this reliable? tls.set_ex_data(*CONNECTION_INFO_INDEX, ConnectionInfo { id, remote }); let mut tls = SslStreamBuilder::new(tls, stream); @@ -354,16 +331,16 @@ impl Endpoint { match tls.accept() { Ok(_) => unreachable!(), Err(HandshakeError::WouldBlock(mut tls)) => { - trace!(self.log, "got fresh cookie"); - let handshake = tls.get_mut().take_outgoing().to_owned(); + trace!(self.log, "performing handshake"; "connection" => %id); if let Some(params) = tls.ssl().ex_data(*TRANSPORT_PARAMS_INDEX).cloned() { let params = params.expect("transport parameter errors should have aborted the handshake"); let conn = self.add_connection(id); + self.connections[conn.0].stream0_data = frame::StreamAssembler::with_offset(offset); + self.transmit_handshake(now, conn, remote, (&tls.get_mut().take_outgoing()[..]).into()); self.connections[conn.0].state = Some(State::Handshake(state::Handshake { tls, clienthello_packet: None, })); - self.events.push_back(Event::Connected(conn)); } else { debug!(self.log, "ClientHello missing transport params extension"); // TODO: Respond with CONNECTION_CLOSE @@ -376,93 +353,168 @@ impl Endpoint { } } Err(None) => { - trace!(self.log, "sending HelloRetryRequest"); + trace!(self.log, "sending HelloRetryRequest"; "connection" => %id); let data = tls.get_mut().take_outgoing(); let mut buf = Vec::::new(); buf.reserve_exact(17 + data.len()); encode_long_header(&mut buf, packet::RETRY, id, packet_number); - frame::stream(&mut buf, StreamId(0), None, true, false, &data); - self.transmit(remote, buf.into()); + frame::Stream { + id: StreamId(0), + offset: 0, + fin: false, + data: data, + }.encode(false, &mut buf); + self.io.push_back(Io::Transmit { destination: remote, packet: buf.into() }); } Err(Some(e)) => { - debug!(self.log, "stateless handshake failed"; "reason" => %e); + debug!(self.log, "stateless handshake failed"; "connection" => %id, "reason" => %e); // TODO: Respond with CONNECTION_CLOSE unimplemented!() } } } - fn handle_connected(&mut self, now: u64, conn: ConnectionHandle, remote: SocketAddrV6, packet: Packet) { - trace!(self.log, "connection got packet"; "id" => conn.0, "len" => packet.payload.len()); - let was_closed = self.connections[conn.0].state.as_ref().unwrap().is_closed(); - self.connections[conn.0].state = Some(match self.connections[conn.0].state.take().unwrap() { + fn handle_connected_inner(&mut self, now: u64, conn: ConnectionHandle, remote: SocketAddrV6, packet: Packet, state: State) -> Option { + Some(match state { State::Handshake(mut state) => { match packet.header { - Header::Long { ty: packet::RETRY, number, .. } => { - match state.clienthello_packet { - Some(x) if number < x => { - // Retry corresponds to an outdated Initial; ignore - State::Handshake(state) - } - None => { - // Received Retry as a server + Header::Long { ty: packet::RETRY, number, id: conn_id, .. } => { + if state.clienthello_packet.is_none() { + // Received Retry as a server + debug!(self.log, "received retry from client"; "connection" => %conn_id); + self.events.push_back(Event::ConnectionLost { connection: conn, reason: ConnectionError::HandshakeFailed }); + State::HandshakeFailed(state::HandshakeFailed { + reason: TransportError::PROTOCOL_VIOLATION, + alert: None, + }) + } else if state.clienthello_packet.unwrap() != number { + // Retry corresponds to an outdated Initial; must be a duplicate, so ignore it + State::Handshake(state) + } else if self.connections[conn.0].stream0.rx_offset != 0 { + // Received current Retry after Handshake + debug!(self.log, "received seemingly-valid retry following handshake packets"); + self.events.push_back(Event::ConnectionLost { connection: conn, reason: ConnectionError::HandshakeFailed }); + State::HandshakeFailed(state::HandshakeFailed { + reason: TransportError::PROTOCOL_VIOLATION, + alert: None, + }) + } else if !parse_initial(state.tls.get_mut(), packet.payload.clone()) { + debug!(self.log, "invalid retry payload"); + self.events.push_back(Event::ConnectionLost { connection: conn, reason: ConnectionError::HandshakeFailed }); + State::HandshakeFailed(state::HandshakeFailed { + reason: TransportError::PROTOCOL_VIOLATION, + alert: None, + }) + } else { match state.tls.handshake() { + Err(HandshakeError::WouldBlock(mut tls)) => { + trace!(self.log, "resending ClientHello"); + let id = self.connections[conn.0].id; + // Discard transport state + self.connections[conn.0] = Connection::new( + id, self.initial_packet_number.sample(&mut self.rng).into(), &self.config + ); + self.connections[conn.0].client = true; + // Send updated ClientHello + let packet = self.transmit_handshake(now, conn, remote, (&tls.get_mut().take_outgoing()[..]).into()); + State::Handshake(state::Handshake { tls, clienthello_packet: Some(packet) }) + }, + Ok(_) => { + debug!(self.log, "unexpectedly completed handshake in RETRY packet"); self.events.push_back(Event::ConnectionLost { connection: conn, reason: ConnectionError::HandshakeFailed }); State::HandshakeFailed(state::HandshakeFailed { reason: TransportError::PROTOCOL_VIOLATION, alert: None, }) } - Some(_) => { - // Send updated ClientHello - parse_handshake(state.tls.get_mut(), packet.payload.clone()); - match state.tls.handshake() { - Err(HandshakeError::WouldBlock(mut tls)) => { - trace!(self.log, "resending ClientHello"); - // TODO: Reliable transmit - let mut buf = Vec::::new(); - buf.reserve_exact(MIN_INITIAL_SIZE); - let packet = self.connections[conn.0].get_tx_number() as u32; - encode_long_header(&mut buf, packet::INITIAL, self.connections[conn.0].id, packet); - frame::stream(&mut buf, StreamId(0), None, true, false, &tls.get_mut().take_outgoing()); - if buf.len() < MIN_INITIAL_SIZE { - buf.resize(MIN_INITIAL_SIZE, frame::Type::PADDING.into()); - } - self.transmit(remote, buf.into()); - State::Handshake(state::Handshake { tls, clienthello_packet: Some(packet) }) - }, - Ok(_) => { - debug!(self.log, "unexpectedly completed handshake in RETRY packet"); + Err(HandshakeError::Failure(mut tls)) => { + debug!(self.log, "handshake failed"; "reason" => %tls.error()); + self.events.push_back(Event::ConnectionLost { connection: conn, reason: ConnectionError::HandshakeFailed }); + State::HandshakeFailed(state::HandshakeFailed { + reason: TransportError::TLS_HANDSHAKE_FAILED, + alert: Some(tls.get_mut().take_outgoing().to_owned().into()), + }) + } + Err(HandshakeError::SetupFailure(e)) => { + error!(self.log, "handshake setup failed"; "reason" => %e); + self.events.push_back(Event::ConnectionLost { connection: conn, reason: ConnectionError::InternalError }); + State::HandshakeFailed(state::HandshakeFailed { + reason: TransportError::INTERNAL_ERROR, + alert: None, + }) + } + } + } + } + Header::Long { ty: packet::HANDSHAKE, id, .. } => { + // Complete handshake (and ultimately send Finished) + for frame in frame::Iter::new(packet.payload) { + match frame { + Frame::Padding => {} + Frame::Stream(frame::Stream { id, offset, data, .. }) => { + if id != StreamId(0) { + debug!(self.log, "non-stream-0 frame in handshake"); self.events.push_back(Event::ConnectionLost { connection: conn, reason: ConnectionError::HandshakeFailed }); - State::HandshakeFailed(state::HandshakeFailed { - reason: TransportError::TLS_HANDSHAKE_FAILED, + return Some(State::HandshakeFailed(state::HandshakeFailed { + reason: TransportError::PROTOCOL_VIOLATION, alert: None, - }) - } - Err(HandshakeError::Failure(mut tls)) => { - debug!(self.log, "handshake failed"; "reason" => %tls.error()); - self.events.push_back(Event::ConnectionLost { connection: conn, reason: ConnectionError::HandshakeFailed }); - State::HandshakeFailed(state::HandshakeFailed { - reason: TransportError::TLS_HANDSHAKE_FAILED, - alert: Some(tls.get_mut().take_outgoing().to_owned().into()), - }) - } - Err(HandshakeError::SetupFailure(e)) => { - error!(self.log, "handshake setup failed"; "reason" => %e); - self.events.push_back(Event::ConnectionLost { connection: conn, reason: ConnectionError::InternalError }); - State::HandshakeFailed(state::HandshakeFailed { - reason: TransportError::INTERNAL_ERROR, - alert: None, - }) + })); } + self.connections[conn.0].stream0_data.insert(offset, data); + } + Frame::Ack(ack) => { + let time = self.connections[conn.0].on_ack_received(&self.config, now, true, ack); + self.io.push_back(Io::TimerStart { + connection: conn, + timer: Timer::LossDetection, + time, + }); + } + _ => { + debug!(self.log, "invalid frame type in handshake"); + self.events.push_back(Event::ConnectionLost { connection: conn, reason: ConnectionError::HandshakeFailed }); + return Some(State::HandshakeFailed(state::HandshakeFailed { + reason: TransportError::PROTOCOL_VIOLATION, + alert: None, + })); } } } + while let Some(segment) = self.connections[conn.0].stream0_data.next() { + self.connections[conn.0].stream0.rx_offset += segment.len() as u64; + state.tls.get_mut().extend_incoming(&segment); + } + match state.tls.handshake() { + Ok(mut tls) => { + trace!(self.log, "established"; "connection" => %id); + self.transmit_handshake(now, conn, remote, (&tls.get_mut().take_outgoing()[..]).into()); + self.events.push_back(Event::Connected(conn)); + State::Established(state::Established { tls }) + } + Err(HandshakeError::WouldBlock(mut tls)) => { + trace!(self.log, "handshake ongoing"; "connection" => %id); + self.transmit_handshake(now, conn, remote, (&tls.get_mut().take_outgoing()[..]).into()); + State::Handshake(state::Handshake { tls, clienthello_packet: state.clienthello_packet }) + } + Err(HandshakeError::Failure(mut tls)) => { + debug!(self.log, "handshake failed"; "connection" => %id, "reason" => %tls.error()); + self.events.push_back(Event::ConnectionLost { connection: conn, reason: ConnectionError::HandshakeFailed }); + State::HandshakeFailed(state::HandshakeFailed { + reason: TransportError::TLS_HANDSHAKE_FAILED, + alert: Some(tls.get_mut().take_outgoing().to_owned().into()), + }) + } + Err(HandshakeError::SetupFailure(e)) => { + error!(self.log, "handshake failed"; "connection" => %id, "reason" => %e); + self.events.push_back(Event::ConnectionLost { connection: conn, reason: ConnectionError::InternalError }); + State::HandshakeFailed(state::HandshakeFailed { + reason: TransportError::INTERNAL_ERROR, + alert: None, + }) + } + } } - Header::Long { ty: packet::HANDSHAKE, .. } => { - // Complete handshake (and ultimately send Finished) - unimplemented!() - } - Header::Long { .. } => { + Header::Long { ty, .. } => { + debug!(self.log, "unexpected packet type"; "type" => ty); self.events.push_back(Event::ConnectionLost { connection: conn, reason: ConnectionError::HandshakeFailed }); State::HandshakeFailed(state::HandshakeFailed { reason: TransportError::PROTOCOL_VIOLATION, @@ -473,7 +525,7 @@ impl Endpoint { // TODO: MUST ignore if supported version is listed self.forget(conn); self.events.push_back(Event::ConnectionLost { connection: conn, reason: ConnectionError::VersionMismatch }); - return; // Connection forgotten, no further processing is useful + return None; // Connection forgotten, no further processing is useful } // TODO: SHOULD buffer these. Header::Short { .. } => { @@ -482,15 +534,30 @@ impl Endpoint { } } State::Established(state) => { - unimplemented!(); + // TODO State::Established(state) } State::HandshakeFailed(state) => { - // TODO: Switch to close if draining + // TODO: Switch to draining if packet has a close unimplemented!() } State::Draining => State::Draining, - }); + }) + } + + fn handle_connected(&mut self, now: u64, conn: ConnectionHandle, remote: SocketAddrV6, packet: Packet) { + trace!(self.log, "connection got packet"; "id" => conn.0, "len" => packet.payload.len()); + // TODO: ACK + let was_closed = self.connections[conn.0].state.as_ref().unwrap().is_closed(); + { + let initial_state = self.connections[conn.0].state.take().unwrap(); + if let Some(state) = self.handle_connected_inner(now, conn, remote, packet, initial_state) { + self.connections[conn.0].state = Some(state); + } else { + // Connection was abandoned + return; + } + } if !was_closed && self.connections[conn.0].state.as_ref().unwrap().is_closed() { self.io.push_back(Io::TimerStart { @@ -551,9 +618,54 @@ impl Endpoint { self.connection_ids.remove(&self.connections[conn.0].id); self.connections.remove(conn.0); } + + fn transmit_handshake(&mut self, now: u64, conn: ConnectionHandle, destination: SocketAddrV6, messages: Bytes) -> u32 { + // TODO: Fragmentation + debug_assert!(!messages.is_empty()); + let mut buf = Vec::::new(); + let packet_number = self.connections[conn.0].get_tx_number() as u32; + let tx_offset = { + let x = &mut self.connections[conn.0].stream0.tx_offset; + let initial = *x; + *x += messages.len() as u64; + initial + }; + let ty = if self.connections[conn.0].client && tx_offset == 0 { packet::INITIAL } else { packet::HANDSHAKE }; + encode_long_header(&mut buf, ty, self.connections[conn.0].id, packet_number); + let frame = frame::Stream { + id: StreamId(0), + offset: tx_offset, + fin: false, + data: messages, + }; + frame.encode(true, &mut buf); + if ty == packet::INITIAL && buf.len() < MIN_INITIAL_SIZE { + buf.resize(MIN_INITIAL_SIZE, frame::Type::PADDING.into()); + } + + let bytes = buf.len() as u16; + self.io.push_back(Io::Transmit { destination, packet: buf.into() }); + self.on_packet_sent(now, conn, true, packet_number as u64, SentPacket { + time: now, + bytes: Some(bytes), + stream: vec![frame], + ..SentPacket::default() + }); + packet_number + } + + fn on_packet_sent(&mut self, now: u64, conn: ConnectionHandle, in_handshake: bool, packet_number: u64, packet: SentPacket) { + if let Some(time) = self.connections[conn.0].on_packet_sent(&self.config, now, in_handshake, packet_number, packet) { + self.io.push_back(Io::TimerStart { + connection: conn, + timer: Timer::LossDetection, + time + }); + } + } } -fn encode_long_header(buf: &mut Vec, ty: u8, id: ConnectionId, packet: u32) { +fn encode_long_header(buf: &mut W, ty: u8, id: ConnectionId, packet: u32) { buf.put_u8(0b10000000 | ty); buf.put_u64::(id.0); buf.put_u32::(VERSION); @@ -567,16 +679,26 @@ impl Rand for ConnectionId { fn rand(rng: &mut R) -> Self { ConnectionId(rng.gen()) } } +impl fmt::Display for ConnectionId { + fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result { + write!(f, "{:016x}", self.0) + } +} + + struct Connection { id: ConnectionId, state: Option, tx_packet_number: u64, + stream0: Stream, + stream0_data: frame::StreamAssembler, streams: HashMap, /// Packets we haven't yet acknowledged pending_acks: Vec, /// Acks we've sent which haven't been acked in turn unconfirmed_acks: HashMap>, - /// Present iff we're the client + client: bool, + /// Present iff we're the client and the handshake is complete reset_token: Option<[u8; 16]>, // @@ -649,19 +771,34 @@ struct SentPacket { stream: Vec, } +impl Default for SentPacket { + fn default() -> Self { Self { + time: 0, + bytes: None, + max_stream_data: false, + max_data: false, + max_stream_id: false, + ack: false, + new_connection_id: None, + stream: Vec::new(), + }} +} + impl SentPacket { fn ack_only(&self) -> bool { self.bytes.is_none() } } impl Connection { fn new(id: ConnectionId, tx_packet_number: u64, config: &Config) -> Self { - let mut streams = HashMap::with_capacity(1); - streams.insert(StreamId(0), Stream::new()); Self { - id, tx_packet_number, streams, + id, tx_packet_number, + stream0: Stream::new(), + stream0_data: frame::StreamAssembler::new(), + streams: HashMap::new(), state: None, pending_acks: Vec::new(), unconfirmed_acks: HashMap::new(), + client: false, reset_token: None, handshake_count: 0, @@ -782,13 +919,13 @@ impl Connection { self.loss_time = 0; let mut lost_packets = Vec::::new(); let delay_until_lost; - let factor = cmp::max(self.latest_rtt, self.smoothed_rtt); + let rtt = cmp::max(self.latest_rtt, self.smoothed_rtt); if config.using_time_loss_detection { // factor * (1 + fraction) - delay_until_lost = factor + (factor * config.time_reordering_fraction as u64) >> 16; + delay_until_lost = rtt + (rtt * config.time_reordering_fraction as u64) >> 16; } else if largest_acked == self.largest_sent_packet { // Early retransmit alarm. - delay_until_lost = (5 * factor) / 4; + delay_until_lost = (5 * rtt) / 4; } else { delay_until_lost = u64::max_value(); } @@ -847,9 +984,7 @@ impl Connection { config.min_tlp_timeout); } else { // RTO alarm - alarm_duration = self.smoothed_rtt + 4 * self.rttvar + self.max_ack_delay; - alarm_duration = cmp::max(alarm_duration, config.min_rto_timeout); - alarm_duration = alarm_duration * 2u64.pow(self.rto_count); + alarm_duration = self.rto(config); } self.time_of_last_sent_packet + alarm_duration } @@ -862,10 +997,15 @@ impl Connection { } struct Stream { + tx_offset: u64, + rx_offset: u64, } impl Stream { - fn new() -> Self { Self {} } + fn new() -> Self { Self { + tx_offset: 0, + rx_offset: 0, + }} } #[derive(Debug, Copy, Clone)] @@ -1143,15 +1283,21 @@ mod packet { } /// Forward data from an Initial or Retry packet to a stream for a TLS context -fn parse_handshake(stream: &mut MemoryStream, payload: Bytes) { +fn parse_initial(stream: &mut MemoryStream, payload: Bytes) -> bool { + let mut staging = frame::StreamAssembler::new(); for frame in frame::Iter::new(payload) { match frame { - Frame::Stream(frame::Stream { id: StreamId(0), data, .. }) => { - stream.extend_incoming(&data[..]); + Frame::Padding => {} + Frame::Stream(frame::Stream { id, offset, data, .. }) => { + if id != StreamId(0) { return false; } // Invalid packet + staging.insert(offset, data); } - _ => {} + _ => { return false; } // Invalid packet } } + while let Some(data) = staging.next() { stream.extend_incoming(&data); } + if !staging.is_empty() { return false; } // Invalid packet (incomplete stream) + true } #[derive(Debug, Clone)] diff --git a/src/frame.rs b/src/frame.rs index 303b7bbb0..d96ad18de 100644 --- a/src/frame.rs +++ b/src/frame.rs @@ -1,4 +1,5 @@ use std::{mem, fmt, io}; +use std::collections::BTreeMap; use bytes::{Bytes, BufMut}; @@ -77,11 +78,27 @@ pub struct Ack { } #[derive(Debug, Clone)] -pub struct Stream { +pub struct Stream { pub id: StreamId, pub offset: u64, pub fin: bool, - pub data: Bytes, + pub data: T, +} + +impl Stream + where T: AsRef<[u8]> +{ + pub fn encode(&self, length: bool, out: &mut W) { + let mut ty = 0x10; + if self.offset != 0 { ty |= 0x04; } + if length { ty |= 0x02; } + if self.fin { ty |= 0x01; } + out.put_u8(ty); + varint::write(self.id.0, out).unwrap(); + if self.offset != 0 { varint::write(self.offset, out).unwrap(); } + if length { varint::write(self.data.as_ref().len() as u64, out).unwrap(); } + out.put_slice(self.data.as_ref()); + } } pub struct Iter(Bytes); @@ -156,18 +173,6 @@ impl Iterator for Iter { } } -pub fn stream(out: &mut Vec, id: StreamId, offset: Option, length: bool, fin: bool, data: &[u8]) { - let mut ty = 0x10; - if offset.is_some() { ty |= 0x04; } - if length { ty |= 0x02; } - if fin { ty |= 0x01; } - out.put_u8(ty); - varint::write(id.0, out).unwrap(); - if let Some(o) = offset { varint::write(o, out).unwrap(); } - if length { varint::write(data.len() as u64, out).unwrap(); } - out.extend_from_slice(data); -} - #[derive(Debug, Clone)] pub struct AckIter { next: u64, @@ -235,3 +240,70 @@ impl StreamId { } impl From for StreamId { fn from(x: u64) -> Self { StreamId(x) } } + +#[derive(Debug)] +pub struct StreamAssembler { + offset: u64, + /// (offset, data) + segments: BTreeMap, +} + +impl StreamAssembler { + pub fn new() -> Self { Self::with_offset(0) } + pub fn with_offset(x: u64) -> Self { Self { offset: x, segments: BTreeMap::new() } } + pub fn is_empty(&self) -> bool { self.segments.is_empty() } + + pub fn next(&mut self) -> Option { + if let Some(data) = self.segments.remove(&self.offset) { + self.offset += data.len() as u64; + Some(data) + } else { None } + } + + pub fn insert(&mut self, mut offset: u64, mut data: Bytes) { + let end_offset = if let Some((&prev_off, prev_data)) = self.segments.range(..offset).rev().next() { + prev_off + prev_data.len() as u64 + } else { + self.offset + }; + if let Some(relative) = end_offset.checked_sub(offset) { + if relative >= data.len() as u64 { return; } + offset += relative; + data.advance(relative as usize); + } + if let Some((&next_off, next_data)) = self.segments.range(offset..).next() { + if offset == next_off { return; } + data.truncate((next_off - offset) as usize); + } + self.segments.insert(offset, data); + } +} + +#[cfg(test)] +mod test { + use super::*; + + #[test] + fn assemble_ordered() { + let mut x = StreamAssembler::new(); + assert_matches!(x.next(), None); + x.insert(0, (&b"123"[..]).into()); + assert_matches!(x.next(), Some(ref y) if &y[..] == b"123"); + x.insert(3, (&b"456"[..]).into()); + assert_matches!(x.next(), Some(ref y) if &y[..] == b"456"); + x.insert(6, (&b"789"[..]).into()); + x.insert(9, (&b"10"[..]).into()); + assert_matches!(x.next(), Some(ref y) if &y[..] == b"789"); + assert_matches!(x.next(), Some(ref y) if &y[..] == b"10"); + } + + #[test] + fn assemble_unordered() { + let mut x = StreamAssembler::new(); + x.insert(3, (&b"456"[..]).into()); + assert_matches!(x.next(), None); + x.insert(0, (&b"123"[..]).into()); + assert_matches!(x.next(), Some(ref y) if &y[..] == b"123"); + assert_matches!(x.next(), Some(ref y) if &y[..] == b"456"); + } +} diff --git a/src/lib.rs b/src/lib.rs index 6026a3a8a..3e91768d4 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -16,6 +16,10 @@ extern crate serde_derive; extern crate lazy_static; #[macro_use] extern crate slog; +#[cfg(test)] +#[macro_use] +extern crate assert_matches; + mod varint; mod memory_stream; diff --git a/src/memory_stream.rs b/src/memory_stream.rs index 068619d14..11d4f81bd 100644 --- a/src/memory_stream.rs +++ b/src/memory_stream.rs @@ -18,6 +18,8 @@ impl MemoryStream { } pub fn take_outgoing(&mut self) -> Outgoing { Outgoing(&mut self.outgoing) } + + pub fn incoming_len(&self) -> usize { self.incoming.get_ref().len() - self.incoming.position() as usize } } impl Read for MemoryStream { @@ -54,3 +56,7 @@ impl<'a> Deref for Outgoing<'a> { type Target = [u8]; fn deref(&self) -> &[u8] { &self.0 } } + +impl<'a> AsRef<[u8]> for Outgoing<'a> { + fn as_ref(&self) -> &[u8] { &self.0 } +} diff --git a/src/transport_parameters.rs b/src/transport_parameters.rs index a58aaef18..f46b8187f 100644 --- a/src/transport_parameters.rs +++ b/src/transport_parameters.rs @@ -193,7 +193,7 @@ mod test { use bytes::IntoBuf; #[test] - fn identity() { + fn encode_default() { let mut buf = Vec::new(); let params = TransportParameters::default(); params.write(&mut buf); diff --git a/tests/tests.rs b/tests/tests.rs index d860d9f78..9b9fdc32c 100644 --- a/tests/tests.rs +++ b/tests/tests.rs @@ -12,6 +12,7 @@ use std::net::SocketAddrV6; use openssl::pkey::{PKey}; use openssl::rsa::Rsa; use openssl::x509::X509; +use openssl::asn1::Asn1Time; use slog::{Logger, Drain}; use quicr::*; @@ -36,6 +37,8 @@ impl Pair { let key = PKey::from_rsa(Rsa::generate(2048).unwrap()).unwrap(); let mut cert = X509::builder().unwrap(); cert.set_pubkey(&key).unwrap(); + cert.set_not_before(&Asn1Time::days_from_now(0).unwrap()).unwrap(); + cert.set_not_after(&Asn1Time::days_from_now(u32::max_value()).unwrap()).unwrap(); cert.sign(&key, openssl::hash::MessageDigest::sha256()).unwrap(); let cert = cert.build(); let server = Endpoint::new( @@ -83,7 +86,7 @@ impl Pair { fn connect() { let log = logger(); let mut pair = Pair::new(log); - if let Err(e) = pair.client.connect(pair.client_addr, pair.server_addr) { + if let Err(e) = pair.client.connect(0, pair.client_addr, pair.server_addr) { panic!("{}", e); } pair.drive();