gsadmin/noq
1
0
Fork 0
mirror of https://github.com/n0-computer/noq.git synced 2026-09-24 20:25:00 +00:00
Code Issues Packages Projects Releases Wiki Activity

Support customized Cid generation

Browse Source
This commit is contained in:
Wenjie Li
2020-09-17 12:14:56 -07:00
committed by Dirkjan Ochtman
parent 81d326a0f6
commit b6f89afd88
7 changed files with 158 additions and 97 deletions
Download Patch File Download Diff File Expand all files Collapse all files
quinn-proto/src/config.rs
+24 -18
View File
@@ -8,6 +8,7 @@ use crate::crypto::types::{Certificate, CertificateChain, PrivateKey};
use crate::{
congestion,
crypto::{self, ClientConfig as _, HmacKey as _, ServerConfig as _},
shared::{ConnectionIdGenerator, RandomConnectionIdGenerator},
VarInt, MAX_CID_SIZE,
};
@@ -297,9 +298,12 @@ pub struct EndpointConfig<S>
where
S: crypto::Session,
{
pub(crate) local_cid_len: usize,
pub(crate) reset_key: Arc<S::HmacKey>,
pub(crate) max_udp_payload_size: u64,
/// cid generator factory
/// Create a cid generator for local cid in Endpoint struct
pub(crate) connection_id_generator_factory:
Arc<dyn Fn() -> Box<dyn ConnectionIdGenerator> + Send + Sync>,
}
impl<S> EndpointConfig<S>
@@ -308,25 +312,27 @@ where
{
/// Create a default config with a particular `reset_key`
pub fn new(reset_key: S::HmacKey) -> Self {
Self {
local_cid_len: 8,
reset_key: Arc::new(reset_key),
max_udp_payload_size: MAX_UDP_PAYLOAD_SIZE,
}
let cid_factory: fn() -> Box<dyn ConnectionIdGenerator> =
|| Box::new(RandomConnectionIdGenerator::default());
EndpointConfig::new_with_cid_generator(reset_key, cid_factory)
}
/// Length of connection IDs for the endpoint.
/// new_with_cid_generator is designed for backward compatibility
///
/// This must be no greater than 20. If zero, incoming packets are mapped to connections only by
/// their source address. Otherwise, the connection ID field is used alone, allowing for source
/// address to change and for multiple connections from a single address. When local_cid_len >
/// 0, at most 3/4 * 2^(local_cid_len * 8) simultaneous connections can be supported.
pub fn local_cid_len(&mut self, value: usize) -> Result<&mut Self, ConfigError> {
if value > MAX_CID_SIZE {
return Err(ConfigError::OutOfBounds);
/// EndpointConfig can still call fn new() to use a random cid generator
/// new_with_cid_generator() can accept any customized cid generator that
/// implements ConnectionIdGenerator trait
pub fn new_with_cid_generator<
F: Fn() -> Box<dyn ConnectionIdGenerator> + Send + Sync + 'static,
>(
reset_key: S::HmacKey,
factory: F,
) -> Self {
Self {
reset_key: Arc::new(reset_key),
max_udp_payload_size: MAX_UDP_PAYLOAD_SIZE,
connection_id_generator_factory: Arc::new(factory),
}
self.local_cid_len = value;
Ok(self)
}
/// Private key used to send authenticated connection resets to peers who were
@@ -359,9 +365,9 @@ where
impl<S: crypto::Session> fmt::Debug for EndpointConfig<S> {
fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
fmt.debug_struct("EndpointConfig")
.field("local_cid_len", &self.local_cid_len)
.field("reset_key", &"[ elided ]")
.field("max_udp_payload_size", &self.max_udp_payload_size)
.field("cid_generator_factory", &"[ elided ]")
.finish()
}
}
@@ -380,9 +386,9 @@ impl<S: crypto::Session> Default for EndpointConfig<S> {
impl<S: crypto::Session> Clone for EndpointConfig<S> {
fn clone(&self) -> Self {
Self {
local_cid_len: self.local_cid_len,
reset_key: self.reset_key.clone(),
max_udp_payload_size: self.max_udp_payload_size,
connection_id_generator_factory: self.connection_id_generator_factory.clone(),
}
}
}
quinn-proto/src/connection/mod.rs
+8 -7
View File
@@ -16,7 +16,7 @@ use tracing::{debug, error, trace, trace_span, warn};
use crate::{
cid_queue::CidQueue,
coding::BufMutExt,
config::{EndpointConfig, ServerConfig, TransportConfig},
config::{ServerConfig, TransportConfig},
crypto::{self, HeaderKey, KeyPair, Keys, PacketKey},
frame,
frame::{Close, Datagram, FrameStruct},
@@ -60,7 +60,6 @@ pub struct Connection<S>
where
S: crypto::Session,
{
endpoint_config: Arc<EndpointConfig<S>>,
server_config: Option<Arc<ServerConfig<S>>>,
config: Arc<TransportConfig>,
rng: StdRng,
@@ -78,6 +77,8 @@ where
/// Exactly one prior to `self.rem_cids.offset` except during processing of certain
/// NEW_CONNECTION_ID frames.
rem_cid_seq: u64,
/// cid length used to decode short packet
local_cid_len: usize,
path: PathData,
prev_path: Option<PathData>,
state: State,
@@ -165,7 +166,6 @@ where
S: crypto::Session,
{
pub(crate) fn new(
endpoint_config: Arc<EndpointConfig<S>>,
server_config: Option<Arc<ServerConfig<S>>>,
config: Arc<TransportConfig>,
init_cid: ConnectionId,
@@ -174,6 +174,7 @@ where
remote: SocketAddr,
crypto: S,
now: Instant,
local_cid_len: usize,
) -> Self {
let side = if server_config.is_some() {
Side::Server
@@ -194,13 +195,13 @@ where
.as_ref()
.map_or(false, |c| c.use_stateless_retry);
let mut this = Self {
endpoint_config,
server_config,
crypto,
handshake_cid: loc_cid,
rem_cid,
rem_handshake_cid: rem_cid,
rem_cid_seq: 0,
local_cid_len,
path: PathData::new(
remote,
config.initial_rtt,
@@ -1652,7 +1653,7 @@ where
self.total_recvd = self.total_recvd.wrapping_add(data.len() as u64);
let mut remaining = Some(data);
while let Some(data) = remaining {
match PartialDecode::new(data, self.endpoint_config.local_cid_len) {
match PartialDecode::new(data, self.local_cid_len) {
Ok((partial_decode, rest)) => {
remaining = rest;
self.handle_decode(now, remote, ecn, partial_decode);
@@ -2279,7 +2280,7 @@ where
self.streams.received_stop_sending(id, error_code);
}
Frame::RetireConnectionId { sequence } => {
if self.endpoint_config.local_cid_len == 0 {
if self.local_cid_len == 0 {
return Err(TransportError::PROTOCOL_VIOLATION(
"RETIRE_CONNECTION_ID when CIDs aren't in use",
));
@@ -2486,7 +2487,7 @@ where
/// Issue an initial set of connection IDs to the peer
fn issue_cids(&mut self) {
if self.endpoint_config.local_cid_len == 0 {
if self.local_cid_len == 0 {
return;
}
quinn-proto/src/endpoint.rs
+66 -53
View File
@@ -25,8 +25,8 @@ use crate::{
frame,
packet::{Header, Packet, PacketDecodeError, PacketNumber, PartialDecode},
shared::{
ConnectionEvent, ConnectionEventInner, ConnectionId, EcnCodepoint, EndpointEvent,
EndpointEventInner, IssuedCid,
ConnectionEvent, ConnectionEventInner, ConnectionId, ConnectionIdGenerator, EcnCodepoint,
EndpointEvent, EndpointEventInner, IssuedCid, RandomConnectionIdGenerator,
},
transport_parameters::TransportParameters,
ResetToken, RetryToken, Side, Transmit, TransportError, MAX_CID_SIZE, MIN_INITIAL_SIZE,
@@ -54,6 +54,7 @@ where
/// recipient, if any.
connection_reset_tokens: ResetTokenTable,
connections: Slab<ConnectionMeta>,
local_cid_generator: Box<dyn ConnectionIdGenerator>,
config: Arc<EndpointConfig<S>>,
server_config: Option<Arc<ServerConfig<S>>>,
incoming_handshakes: usize,
@@ -82,6 +83,7 @@ where
connection_remotes: HashMap::new(),
connection_reset_tokens: ResetTokenTable::default(),
connections: Slab::new(),
local_cid_generator: (config.connection_id_generator_factory.as_ref())(),
incoming_handshakes: 0,
reject_new_connections: false,
config,
@@ -152,45 +154,46 @@ where
data: BytesMut,
) -> Option<(ConnectionHandle, DatagramEvent<S>)> {
let datagram_len = data.len();
let (first_decode, remaining) = match PartialDecode::new(data, self.config.local_cid_len) {
Ok(x) => x,
Err(PacketDecodeError::UnsupportedVersion {
source,
destination,
version,
}) => {
if !self.is_server() {
debug!("dropping packet with unsupported version");
let (first_decode, remaining) =
match PartialDecode::new(data, self.local_cid_generator.cid_len()) {
Ok(x) => x,
Err(PacketDecodeError::UnsupportedVersion {
source,
destination,
version,
}) => {
if !self.is_server() {
debug!("dropping packet with unsupported version");
return None;
}
trace!("sending version negotiation");
// Negotiate versions
let mut buf = Vec::<u8>::new();
Header::VersionNegotiate {
random: self.rng.gen::<u8>() | 0x40,
src_cid: destination,
dst_cid: source,
}
.encode(&mut buf);
// Grease with a reserved version
if version != 0x0a1a_2a3a {
buf.write::<u32>(0x0a1a_2a3a);
} else {
buf.write::<u32>(0x0a1a_2a4a);
}
buf.write(VERSION); // supported version
self.transmits.push_back(Transmit {
destination: remote,
ecn: None,
contents: buf.into(),
});
return None;
}
trace!("sending version negotiation");
// Negotiate versions
let mut buf = Vec::<u8>::new();
Header::VersionNegotiate {
random: self.rng.gen::<u8>() | 0x40,
src_cid: destination,
dst_cid: source,
Err(e) => {
trace!("malformed header: {}", e);
return None;
}
.encode(&mut buf);
// Grease with a reserved version
if version != 0x0a1a_2a3a {
buf.write::<u32>(0x0a1a_2a3a);
} else {
buf.write::<u32>(0x0a1a_2a4a);
}
buf.write(VERSION); // supported version
self.transmits.push_back(Transmit {
destination: remote,
ecn: None,
contents: buf.into(),
});
return None;
}
Err(e) => {
trace!("malformed header: {}", e);
return None;
}
};
};
//
// Handle packet on existing connection, if any
@@ -198,7 +201,9 @@ where
let dst_cid = first_decode.dst_cid();
let known_ch = {
let ch = if self.config.local_cid_len > 0 {
let ch = if self.local_cid_generator.cid_len() > 0
&& self.local_cid_generator.validate_cid(&dst_cid)
{
self.connection_ids.get(&dst_cid)
} else {
None
@@ -211,7 +216,7 @@ where
}
})
.or_else(|| {
if self.config.local_cid_len == 0 {
if self.local_cid_generator.cid_len() == 0 {
self.connection_remotes.get(&remote)
} else {
None
@@ -341,7 +346,7 @@ where
if self.is_full() {
return Err(ConnectError::TooManyConnections);
}
let remote_id = ConnectionId::random(&mut self.rng, MAX_CID_SIZE);
let remote_id = RandomConnectionIdGenerator::new(MAX_CID_SIZE).generate_cid();
trace!(initial_dcid = %remote_id);
let (ch, conn) = self.add_connection(
remote_id,
@@ -376,11 +381,11 @@ where
fn new_cid(&mut self) -> ConnectionId {
loop {
let cid = ConnectionId::random(&mut self.rng, self.config.local_cid_len);
let cid = self.local_cid_generator.generate_cid();
if !self.connection_ids.contains_key(&cid) {
break cid;
}
assert!(self.config.local_cid_len > 0);
assert!(self.local_cid_generator.cid_len() > 0);
}
}
@@ -398,8 +403,13 @@ where
config,
server_name,
} => {
let params =
TransportParameters::new::<S>(&config.transport, &self.config, loc_cid, None);
let params = TransportParameters::new::<S>(
&config.transport,
&self.config,
&self.local_cid_generator,
loc_cid,
None,
);
(
None,
config.crypto.start_session(&server_name, &params)?,
@@ -414,6 +424,7 @@ where
let params = TransportParameters::new(
&config.transport,
&self.config,
&self.local_cid_generator,
loc_cid,
Some(config),
);
@@ -432,7 +443,6 @@ where
};
let conn = Connection::new(
Arc::clone(&self.config),
server_config,
transport_config,
init_cid,
@@ -441,6 +451,7 @@ where
remote,
tls,
now,
self.local_cid_generator.cid_len(),
);
let id = self.connections.insert(ConnectionMeta {
init_cid,
@@ -451,7 +462,7 @@ where
});
let ch = ConnectionHandle(id);
if self.config.local_cid_len > 0 {
if self.local_cid_generator.cid_len() > 0 {
self.connection_ids.insert(loc_cid, ch);
} else {
self.connection_remotes.insert(remote, ch);
@@ -518,7 +529,8 @@ where
}
if dst_cid.len() < 8
&& (!server_config.use_stateless_retry || dst_cid.len() != self.config.local_cid_len)
&& (!server_config.use_stateless_retry
|| dst_cid.len() != self.local_cid_generator.cid_len())
{
debug!(
"rejecting connection due to invalid DCID length {}",
@@ -692,10 +704,11 @@ where
/// We leave some space unused so that `new_cid` can be relied upon to finish quickly. We don't
/// bother to check when CID longer than 4 bytes are used because 2^40 connections is a lot.
fn is_full(&self) -> bool {
self.config.local_cid_len <= 4
&& self.config.local_cid_len != 0
&& (2usize.pow(self.config.local_cid_len as u32 * 8) - self.connection_ids.len())
< 2usize.pow(self.config.local_cid_len as u32 * 8 - 2)
self.local_cid_generator.cid_len() <= 4
&& self.local_cid_generator.cid_len() != 0
&& (2usize.pow(self.local_cid_generator.cid_len() as u32 * 8)
- self.connection_ids.len())
< 2usize.pow(self.local_cid_generator.cid_len() as u32 * 8 - 2)
}
}
@@ -793,7 +806,7 @@ pub enum ConnectError {
EndpointStopping,
/// The number of active connections on the local endpoint is at the limit
///
/// Try a larger `EndpointConfig::local_cid_len`.
/// Try a larger cid length.
#[error(display = "too many connections")]
TooManyConnections,
/// The domain name supplied was malformed
quinn-proto/src/shared.rs
+47 -13
View File
@@ -1,7 +1,7 @@
use std::{fmt, net::SocketAddr, time::Instant};
use bytes::{Buf, BufMut, BytesMut};
use rand::Rng;
use rand::RngCore;
use crate::{coding::BufExt, packet::PartialDecode, ResetToken, MAX_CID_SIZE};
@@ -76,18 +76,6 @@ impl ConnectionId {
res
}
pub(crate) fn random<R: Rng>(rng: &mut R, len: usize) -> Self {
debug_assert!(len <= MAX_CID_SIZE);
let mut res = Self {
len: len as u8,
bytes: [0; MAX_CID_SIZE],
};
let mut rng_bytes = [0; MAX_CID_SIZE];
rng.fill_bytes(&mut rng_bytes);
res.bytes[..len].clone_from_slice(&rng_bytes[..len]);
res
}
/// Decode from long header format
pub(crate) fn decode_long(buf: &mut impl Buf) -> Option<Self> {
let len = buf.get::<u8>().ok()? as usize;
@@ -134,6 +122,52 @@ impl fmt::Display for ConnectionId {
}
}
/// Generates connection IDs for incoming connections
pub trait ConnectionIdGenerator: Send {
/// Generates a connection ID for a new connection
fn generate_cid(&mut self) -> ConnectionId;
/// Performs any validation it needs (e.g. HMAC, etc)
fn validate_cid(&mut self, cid: &ConnectionId) -> bool;
/// Returns the length of a connection id for cononections created by this generator
fn cid_len(&self) -> usize;
}
#[derive(Debug, Clone, Copy)]
pub struct RandomConnectionIdGenerator {
cid_len: usize,
}
impl Default for RandomConnectionIdGenerator {
fn default() -> Self {
Self { cid_len: 8 }
}
}
impl RandomConnectionIdGenerator {
pub fn new(cid_len: usize) -> Self {
debug_assert!(cid_len <= MAX_CID_SIZE);
Self { cid_len }
}
}
impl ConnectionIdGenerator for RandomConnectionIdGenerator {
fn generate_cid(&mut self) -> ConnectionId {
let mut res = ConnectionId {
len: self.cid_len as u8,
bytes: [0; MAX_CID_SIZE],
};
rand::thread_rng().fill_bytes(&mut res.bytes[..self.cid_len]);
res
}
/// Cid is an array of random bytes. We only verify the length
fn validate_cid(&mut self, cid: &ConnectionId) -> bool {
cid.len as usize == self.cid_len
}
/// Provide the length of dst_cid in short header packet
fn cid_len(&self) -> usize {
self.cid_len
}
}
/// Explicit congestion notification codepoint
#[repr(u8)]
#[derive(Debug, Copy, Clone, Eq, PartialEq)]
quinn-proto/src/tests/mod.rs
+7 -2
View File
@@ -15,6 +15,7 @@ use tracing::info;
use super::*;
use crate::crypto::Session as _;
use crate::shared::{ConnectionIdGenerator, RandomConnectionIdGenerator};
mod util;
use util::*;
@@ -49,9 +50,11 @@ fn version_negotiate_server() {
fn version_negotiate_client() {
let _guard = subscribe();
let server_addr = "[::2]:7890".parse().unwrap();
let cid_generator_factory: fn() -> Box<dyn ConnectionIdGenerator> =
|| Box::new(RandomConnectionIdGenerator::new(0));
let mut client = Endpoint::new(
Arc::new(EndpointConfig {
local_cid_len: 0,
connection_id_generator_factory: Arc::new(cid_generator_factory),
..Default::default()
}),
None,
@@ -1119,9 +1122,11 @@ fn implicit_open() {
#[test]
fn zero_length_cid() {
let _guard = subscribe();
let cid_generator_factory: fn() -> Box<dyn ConnectionIdGenerator> =
|| Box::new(RandomConnectionIdGenerator::new(0));
let mut pair = Pair::new(
Arc::new(EndpointConfig {
local_cid_len: 0,
connection_id_generator_factory: Arc::new(cid_generator_factory),
..EndpointConfig::default()
}),
server_config(),
quinn-proto/src/token.rs
+3 -2
View File
@@ -140,6 +140,7 @@ mod test {
#[test]
fn token_sanity() {
use super::*;
use crate::shared::{ConnectionIdGenerator, RandomConnectionIdGenerator};
use crate::MAX_CID_SIZE;
use rand::RngCore;
use ring::hmac;
@@ -152,9 +153,9 @@ mod test {
rand::thread_rng().fill_bytes(&mut key);
let key = <hmac::Key as HmacKey>::new(&key).unwrap();
let addr = SocketAddr::new(Ipv6Addr::LOCALHOST.into(), 4433);
let retry_src_cid = ConnectionId::random(&mut rand::thread_rng(), MAX_CID_SIZE);
let retry_src_cid = RandomConnectionIdGenerator::new(MAX_CID_SIZE).generate_cid();
let token = RetryToken {
orig_dst_cid: ConnectionId::random(&mut rand::thread_rng(), MAX_CID_SIZE),
orig_dst_cid: RandomConnectionIdGenerator::new(MAX_CID_SIZE).generate_cid(),
issued: UNIX_EPOCH + Duration::new(42, 0), // Fractional seconds would be lost
};
let encoded = token.encode(&key, &addr, &retry_src_cid);
quinn-proto/src/transport_parameters.rs
+3 -2
View File
@@ -19,7 +19,7 @@ use crate::{
coding::{BufExt, BufMutExt, UnexpectedEnd},
config::{EndpointConfig, ServerConfig, TransportConfig},
crypto,
shared::ConnectionId,
shared::{ConnectionId, ConnectionIdGenerator},
ResetToken, Side, TransportError, VarInt, MAX_CID_SIZE, RESET_TOKEN_SIZE,
};
@@ -116,6 +116,7 @@ impl TransportParameters {
pub(crate) fn new<S>(
config: &TransportConfig,
endpoint_config: &EndpointConfig<S>,
cid_gen: &Box<dyn ConnectionIdGenerator>,
initial_src_cid: ConnectionId,
server_config: Option<&ServerConfig<S>>,
) -> Self
@@ -138,7 +139,7 @@ impl TransportParameters {
}),
max_ack_delay: 0,
disable_active_migration: server_config.map_or(false, |c| !c.migration),
active_connection_id_limit: if endpoint_config.local_cid_len == 0 {
active_connection_id_limit: if cid_gen.cid_len() == 0 {
2 // i.e. default, i.e. unsent
} else {
// + 1 to account for the currently used CID, which isn't kept in the queue