Compare commits
6 Commits
Author | SHA1 | Date |
---|---|---|
Luca Fulchir | 4ddfed358d | |
Luca Fulchir | 9ca4123c37 | |
Luca Fulchir | c3c8238730 | |
Luca Fulchir | a810fc9a9e | |
Luca Fulchir | 9c67210e3e | |
Luca Fulchir | 2fe91d5dd3 |
5
TODO
5
TODO
|
@ -1 +1,6 @@
|
||||||
* Wrapping for everything that wraps (sigh)
|
* Wrapping for everything that wraps (sigh)
|
||||||
|
* track user connection (add u64 from user)
|
||||||
|
* API plit
|
||||||
|
* split API in ThreadLocal, ThreadSafe
|
||||||
|
* split send/recv API in Centralized, Connection
|
||||||
|
* all re wrappers on ThreadLocal-Centralized
|
||||||
|
|
24
flake.lock
24
flake.lock
|
@ -5,11 +5,11 @@
|
||||||
"systems": "systems"
|
"systems": "systems"
|
||||||
},
|
},
|
||||||
"locked": {
|
"locked": {
|
||||||
"lastModified": 1685518550,
|
"lastModified": 1687171271,
|
||||||
"narHash": "sha256-o2d0KcvaXzTrPRIo0kOLV0/QXHhDQ5DTi+OxcjO8xqY=",
|
"narHash": "sha256-BJlq+ozK2B1sJDQXS3tzJM5a+oVZmi1q0FlBK/Xqv7M=",
|
||||||
"owner": "numtide",
|
"owner": "numtide",
|
||||||
"repo": "flake-utils",
|
"repo": "flake-utils",
|
||||||
"rev": "a1720a10a6cfe8234c0e93907ffe81be440f4cef",
|
"rev": "abfb11bd1aec8ced1c9bb9adfe68018230f4fb3c",
|
||||||
"type": "github"
|
"type": "github"
|
||||||
},
|
},
|
||||||
"original": {
|
"original": {
|
||||||
|
@ -38,11 +38,11 @@
|
||||||
},
|
},
|
||||||
"nixpkgs": {
|
"nixpkgs": {
|
||||||
"locked": {
|
"locked": {
|
||||||
"lastModified": 1686921029,
|
"lastModified": 1687555006,
|
||||||
"narHash": "sha256-J1bX9plPCFhTSh6E3TWn9XSxggBh/zDD4xigyaIQBy8=",
|
"narHash": "sha256-GD2Kqb/DXQBRJcHqkM2qFZqbVenyO7Co/80JHRMg2U0=",
|
||||||
"owner": "nixos",
|
"owner": "nixos",
|
||||||
"repo": "nixpkgs",
|
"repo": "nixpkgs",
|
||||||
"rev": "c7ff1b9b95620ce8728c0d7bd501c458e6da9e04",
|
"rev": "33223d479ffde3d05ac16c6dff04ae43cc27e577",
|
||||||
"type": "github"
|
"type": "github"
|
||||||
},
|
},
|
||||||
"original": {
|
"original": {
|
||||||
|
@ -54,11 +54,11 @@
|
||||||
},
|
},
|
||||||
"nixpkgs-unstable": {
|
"nixpkgs-unstable": {
|
||||||
"locked": {
|
"locked": {
|
||||||
"lastModified": 1686960236,
|
"lastModified": 1687502512,
|
||||||
"narHash": "sha256-AYCC9rXNLpUWzD9hm+askOfpliLEC9kwAo7ITJc4HIw=",
|
"narHash": "sha256-dBL/01TayOSZYxtY4cMXuNCBk8UMLoqRZA+94xiFpJA=",
|
||||||
"owner": "nixos",
|
"owner": "nixos",
|
||||||
"repo": "nixpkgs",
|
"repo": "nixpkgs",
|
||||||
"rev": "04af42f3b31dba0ef742d254456dc4c14eedac86",
|
"rev": "3ae20aa58a6c0d1ca95c9b11f59a2d12eebc511f",
|
||||||
"type": "github"
|
"type": "github"
|
||||||
},
|
},
|
||||||
"original": {
|
"original": {
|
||||||
|
@ -98,11 +98,11 @@
|
||||||
"nixpkgs": "nixpkgs_2"
|
"nixpkgs": "nixpkgs_2"
|
||||||
},
|
},
|
||||||
"locked": {
|
"locked": {
|
||||||
"lastModified": 1687055571,
|
"lastModified": 1687660699,
|
||||||
"narHash": "sha256-UvLoO6u5n9TzY80BpM4DaacxvyJl7u9mm9CA72d309g=",
|
"narHash": "sha256-crI/CA/OJc778I5qJhwhhl8/PKKzc0D7vvVxOtjfvSo=",
|
||||||
"owner": "oxalica",
|
"owner": "oxalica",
|
||||||
"repo": "rust-overlay",
|
"repo": "rust-overlay",
|
||||||
"rev": "2de557c780dcb127128ae987fca9d6c2b0d7dc0f",
|
"rev": "b3bd1d49f1ae609c1d68a66bba7a95a9a4256031",
|
||||||
"type": "github"
|
"type": "github"
|
||||||
},
|
},
|
||||||
"original": {
|
"original": {
|
||||||
|
|
|
@ -3,8 +3,9 @@
|
||||||
use crate::{
|
use crate::{
|
||||||
auth::{Domain, ServiceID},
|
auth::{Domain, ServiceID},
|
||||||
connection::{
|
connection::{
|
||||||
|
self,
|
||||||
handshake::{self, Error, Handshake},
|
handshake::{self, Error, Handshake},
|
||||||
Conn, IDRecv, IDSend,
|
Connection, IDRecv, IDSend,
|
||||||
},
|
},
|
||||||
enc::{
|
enc::{
|
||||||
self,
|
self,
|
||||||
|
@ -18,20 +19,27 @@ use crate::{
|
||||||
use ::tokio::sync::oneshot;
|
use ::tokio::sync::oneshot;
|
||||||
|
|
||||||
pub(crate) struct Server {
|
pub(crate) struct Server {
|
||||||
pub id: KeyID,
|
pub(crate) id: KeyID,
|
||||||
pub key: PrivKey,
|
pub(crate) key: PrivKey,
|
||||||
pub domains: Vec<Domain>,
|
pub(crate) domains: Vec<Domain>,
|
||||||
}
|
}
|
||||||
|
|
||||||
pub(crate) type ConnectAnswer = Result<(KeyID, IDSend), crate::Error>;
|
pub(crate) type ConnectAnswer = Result<ConnectOk, crate::Error>;
|
||||||
|
#[derive(Debug)]
|
||||||
|
pub(crate) struct ConnectOk {
|
||||||
|
pub(crate) auth_key_id: KeyID,
|
||||||
|
pub(crate) auth_id_send: IDSend,
|
||||||
|
pub(crate) authsrv_conn: connection::AuthSrvConn,
|
||||||
|
pub(crate) service_conn: Option<connection::ServiceConn>,
|
||||||
|
}
|
||||||
|
|
||||||
pub(crate) struct Client {
|
pub(crate) struct Client {
|
||||||
pub service_id: ServiceID,
|
pub(crate) service_id: ServiceID,
|
||||||
pub service_conn_id: IDRecv,
|
pub(crate) service_conn_id: IDRecv,
|
||||||
pub connection: Conn,
|
pub(crate) connection: Connection,
|
||||||
pub timeout: Option<::tokio::task::JoinHandle<()>>,
|
pub(crate) timeout: Option<::tokio::time::Instant>,
|
||||||
pub answer: oneshot::Sender<ConnectAnswer>,
|
pub(crate) answer: oneshot::Sender<ConnectAnswer>,
|
||||||
pub srv_key_id: KeyID,
|
pub(crate) srv_key_id: KeyID,
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Tracks the keys used by the client and the handshake
|
/// Tracks the keys used by the client and the handshake
|
||||||
|
@ -78,7 +86,7 @@ impl ClientList {
|
||||||
pub_key: PubKey,
|
pub_key: PubKey,
|
||||||
service_id: ServiceID,
|
service_id: ServiceID,
|
||||||
service_conn_id: IDRecv,
|
service_conn_id: IDRecv,
|
||||||
connection: Conn,
|
connection: Connection,
|
||||||
answer: oneshot::Sender<ConnectAnswer>,
|
answer: oneshot::Sender<ConnectAnswer>,
|
||||||
srv_key_id: KeyID,
|
srv_key_id: KeyID,
|
||||||
) -> Result<(KeyID, &mut Client), oneshot::Sender<ConnectAnswer>> {
|
) -> Result<(KeyID, &mut Client), oneshot::Sender<ConnectAnswer>> {
|
||||||
|
@ -128,26 +136,28 @@ impl ClientList {
|
||||||
#[derive(Debug, Clone)]
|
#[derive(Debug, Clone)]
|
||||||
pub(crate) struct AuthNeededInfo {
|
pub(crate) struct AuthNeededInfo {
|
||||||
/// Parsed handshake packet
|
/// Parsed handshake packet
|
||||||
pub handshake: Handshake,
|
pub(crate) handshake: Handshake,
|
||||||
/// hkdf generated from the handshake
|
/// hkdf generated from the handshake
|
||||||
pub hkdf: Hkdf,
|
pub(crate) hkdf: Hkdf,
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Client information needed to fully establish the conenction
|
/// Client information needed to fully establish the conenction
|
||||||
#[derive(Debug)]
|
#[derive(Debug)]
|
||||||
pub(crate) struct ClientConnectInfo {
|
pub(crate) struct ClientConnectInfo {
|
||||||
/// The service ID that we are connecting to
|
/// The service ID that we are connecting to
|
||||||
pub service_id: ServiceID,
|
pub(crate) service_id: ServiceID,
|
||||||
/// The service ID that we are connecting to
|
/// The service ID that we are connecting to
|
||||||
pub service_connection_id: IDRecv,
|
pub(crate) service_connection_id: IDRecv,
|
||||||
/// Parsed handshake packet
|
/// Parsed handshake packet
|
||||||
pub handshake: Handshake,
|
pub(crate) handshake: Handshake,
|
||||||
/// Conn
|
/// Old timeout for the handshake completion
|
||||||
pub connection: Conn,
|
pub(crate) old_timeout: ::tokio::time::Instant,
|
||||||
|
/// Connection
|
||||||
|
pub(crate) connection: Connection,
|
||||||
/// where to wake up the waiting client
|
/// where to wake up the waiting client
|
||||||
pub answer: oneshot::Sender<ConnectAnswer>,
|
pub(crate) answer: oneshot::Sender<ConnectAnswer>,
|
||||||
/// server public key id that we used on the handshake
|
/// server pub(crate)lic key id that we used on the handshake
|
||||||
pub srv_key_id: KeyID,
|
pub(crate) srv_key_id: KeyID,
|
||||||
}
|
}
|
||||||
/// Intermediate actions to be taken while parsing the handshake
|
/// Intermediate actions to be taken while parsing the handshake
|
||||||
#[derive(Debug)]
|
#[derive(Debug)]
|
||||||
|
@ -231,7 +241,7 @@ impl Tracker {
|
||||||
pub_key: PubKey,
|
pub_key: PubKey,
|
||||||
service_id: ServiceID,
|
service_id: ServiceID,
|
||||||
service_conn_id: IDRecv,
|
service_conn_id: IDRecv,
|
||||||
connection: Conn,
|
connection: Connection,
|
||||||
answer: oneshot::Sender<ConnectAnswer>,
|
answer: oneshot::Sender<ConnectAnswer>,
|
||||||
srv_key_id: KeyID,
|
srv_key_id: KeyID,
|
||||||
) -> Result<(KeyID, &mut Client), oneshot::Sender<ConnectAnswer>> {
|
) -> Result<(KeyID, &mut Client), oneshot::Sender<ConnectAnswer>> {
|
||||||
|
@ -366,13 +376,11 @@ impl Tracker {
|
||||||
}
|
}
|
||||||
let hshake =
|
let hshake =
|
||||||
self.hshake_cli.remove(resp.client_key_id).unwrap();
|
self.hshake_cli.remove(resp.client_key_id).unwrap();
|
||||||
if let Some(timeout) = hshake.timeout {
|
|
||||||
timeout.abort();
|
|
||||||
}
|
|
||||||
return Ok(Action::ClientConnect(ClientConnectInfo {
|
return Ok(Action::ClientConnect(ClientConnectInfo {
|
||||||
service_id: hshake.service_id,
|
service_id: hshake.service_id,
|
||||||
service_connection_id: hshake.service_conn_id,
|
service_connection_id: hshake.service_conn_id,
|
||||||
handshake,
|
handshake,
|
||||||
|
old_timeout: hshake.timeout.unwrap(),
|
||||||
connection: hshake.connection,
|
connection: hshake.connection,
|
||||||
answer: hshake.answer,
|
answer: hshake.answer,
|
||||||
srv_key_id: hshake.srv_key_id,
|
srv_key_id: hshake.srv_key_id,
|
||||||
|
|
|
@ -5,21 +5,43 @@ pub mod packet;
|
||||||
pub mod socket;
|
pub mod socket;
|
||||||
pub mod stream;
|
pub mod stream;
|
||||||
|
|
||||||
use ::std::{rc::Rc, vec::Vec};
|
use ::core::num::Wrapping;
|
||||||
|
use ::std::{
|
||||||
|
collections::{BTreeMap, HashMap, VecDeque},
|
||||||
|
vec::Vec,
|
||||||
|
};
|
||||||
|
|
||||||
pub use crate::connection::{handshake::Handshake, packet::Packet};
|
pub use crate::connection::{handshake::Handshake, packet::Packet};
|
||||||
|
|
||||||
use crate::{
|
use crate::{
|
||||||
|
connection::{socket::UdpClient, stream::StreamData},
|
||||||
dnssec,
|
dnssec,
|
||||||
enc::{
|
enc::{
|
||||||
|
self,
|
||||||
asym::PubKey,
|
asym::PubKey,
|
||||||
hkdf::Hkdf,
|
hkdf::Hkdf,
|
||||||
sym::{self, CipherRecv, CipherSend},
|
sym::{self, CipherRecv, CipherSend},
|
||||||
Random,
|
Random,
|
||||||
},
|
},
|
||||||
inner::ThreadTracker,
|
inner::{worker, ThreadTracker},
|
||||||
};
|
};
|
||||||
use ::std::rc;
|
|
||||||
|
/// Connection errors
|
||||||
|
#[derive(::thiserror::Error, Debug, Copy, Clone)]
|
||||||
|
pub enum Error {
|
||||||
|
/// Can't decrypt packet
|
||||||
|
#[error("Decrypt error: {0}")]
|
||||||
|
Decrypt(#[from] crate::enc::Error),
|
||||||
|
/// Error in parsing a packet realated to the connection
|
||||||
|
#[error("Chunk parsing: {0}")]
|
||||||
|
Parse(#[from] stream::Error),
|
||||||
|
/// No such Connection
|
||||||
|
#[error("No suck connection")]
|
||||||
|
NoSuchConnection,
|
||||||
|
/// No such Stream
|
||||||
|
#[error("No suck Stream")]
|
||||||
|
NoSuchStream,
|
||||||
|
}
|
||||||
|
|
||||||
/// Fenrir Connection ID
|
/// Fenrir Connection ID
|
||||||
///
|
///
|
||||||
|
@ -126,25 +148,72 @@ impl ProtocolVersion {
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Connection tracking id. Set by the user
|
||||||
|
#[derive(Debug, Clone, Copy, Eq, PartialEq, Ord, PartialOrd)]
|
||||||
|
pub struct UserTracker(pub ::core::num::NonZeroU64);
|
||||||
|
|
||||||
|
/// Unique tracker of connections
|
||||||
|
#[derive(Debug, Clone, Copy, Eq, PartialEq, Ord, PartialOrd)]
|
||||||
|
pub struct LibTracker(Wrapping<u64>);
|
||||||
|
impl LibTracker {
|
||||||
|
pub(crate) fn new(start: u16) -> Self {
|
||||||
|
Self(Wrapping(start as u64))
|
||||||
|
}
|
||||||
|
pub(crate) fn advance(&mut self, amount: u16) -> Self {
|
||||||
|
let old = self.0;
|
||||||
|
self.0 = self.0 + Wrapping(amount as u64);
|
||||||
|
LibTracker(old)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
/// Collection of connection tracking, but user-given and library generated
|
||||||
|
#[derive(Debug, Copy, Clone)]
|
||||||
|
pub struct ConnTracker {
|
||||||
|
/// Optional tracker set by the user
|
||||||
|
pub user: Option<UserTracker>,
|
||||||
|
/// library generated tracker. Unique and non-repeating
|
||||||
|
pub(crate) lib: LibTracker,
|
||||||
|
}
|
||||||
|
|
||||||
|
impl PartialEq for ConnTracker {
|
||||||
|
fn eq(&self, other: &Self) -> bool {
|
||||||
|
self.lib == other.lib
|
||||||
|
}
|
||||||
|
}
|
||||||
|
impl Eq for ConnTracker {}
|
||||||
|
|
||||||
|
/// Connection to an Authentication Server
|
||||||
|
#[derive(Debug, Copy, Clone)]
|
||||||
|
pub struct AuthSrvConn(pub ConnTracker);
|
||||||
|
/// Connection to a service
|
||||||
|
#[derive(Debug, Copy, Clone)]
|
||||||
|
pub struct ServiceConn(pub ConnTracker);
|
||||||
|
|
||||||
|
/*
|
||||||
|
* TODO: only on Thread{Local,Safe}::Connection oriented flows
|
||||||
/// The connection, as seen from a user of libFenrir
|
/// The connection, as seen from a user of libFenrir
|
||||||
#[derive(Debug)]
|
#[derive(Debug)]
|
||||||
pub struct Connection(rc::Weak<Conn>);
|
pub struct Conn {
|
||||||
|
pub(crate) queue: ::async_channel::Sender<worker::Work>,
|
||||||
/// A single connection and its data
|
pub(crate) tracker: ConnTracker,
|
||||||
#[derive(Debug)]
|
|
||||||
pub(crate) struct Conn {
|
|
||||||
/// Receiving Conn ID
|
|
||||||
pub id_recv: IDRecv,
|
|
||||||
/// Sending Conn ID
|
|
||||||
pub id_send: IDSend,
|
|
||||||
/// The main hkdf used for all secrets in this connection
|
|
||||||
pub hkdf: Hkdf,
|
|
||||||
/// Cipher for decrypting data
|
|
||||||
pub cipher_recv: CipherRecv,
|
|
||||||
/// Cipher for encrypting data
|
|
||||||
pub cipher_send: CipherSend,
|
|
||||||
}
|
}
|
||||||
|
|
||||||
|
impl Conn {
|
||||||
|
/// Queue some data to be sent in this connection
|
||||||
|
// TODO: send_and_wait, that wait for recipient ACK
|
||||||
|
pub async fn send(&mut self, stream: stream::ID, data: Vec<u8>) {
|
||||||
|
use crate::inner::worker::Work;
|
||||||
|
let _ = self
|
||||||
|
.queue
|
||||||
|
.send(Work::UserSend((self.tracker.lib, stream, data)))
|
||||||
|
.await;
|
||||||
|
}
|
||||||
|
/// Get the library tracking id
|
||||||
|
pub fn tracker(&self) -> ConnTracker {
|
||||||
|
self.tracker
|
||||||
|
}
|
||||||
|
}
|
||||||
|
*/
|
||||||
|
|
||||||
/// Role: track the connection direction
|
/// Role: track the connection direction
|
||||||
///
|
///
|
||||||
/// The Role is used to select the correct secrets, and track the direction
|
/// The Role is used to select the correct secrets, and track the direction
|
||||||
|
@ -160,7 +229,48 @@ pub enum Role {
|
||||||
Client,
|
Client,
|
||||||
}
|
}
|
||||||
|
|
||||||
impl Conn {
|
#[derive(Debug)]
|
||||||
|
enum TimerKind {
|
||||||
|
None,
|
||||||
|
SendData(::tokio::time::Instant),
|
||||||
|
Keepalive(::tokio::time::Instant),
|
||||||
|
}
|
||||||
|
|
||||||
|
pub(crate) enum Enqueue {
|
||||||
|
TimerWait,
|
||||||
|
Immediate(::tokio::time::Instant),
|
||||||
|
}
|
||||||
|
|
||||||
|
/// A single connection and its data
|
||||||
|
#[derive(Debug)]
|
||||||
|
pub(crate) struct Connection {
|
||||||
|
/// Receiving Conn ID
|
||||||
|
pub(crate) id_recv: IDRecv,
|
||||||
|
/// Sending Conn ID
|
||||||
|
pub(crate) id_send: IDSend,
|
||||||
|
/// User-managed id to track this connection
|
||||||
|
/// the user can set this to better track this connection
|
||||||
|
pub(crate) user_tracker: Option<UserTracker>,
|
||||||
|
pub(crate) lib_tracker: LibTracker,
|
||||||
|
/// Sending address
|
||||||
|
pub(crate) send_addr: UdpClient,
|
||||||
|
/// The main hkdf used for all secrets in this connection
|
||||||
|
hkdf: Hkdf,
|
||||||
|
/// Cipher for decrypting data
|
||||||
|
pub(crate) cipher_recv: CipherRecv,
|
||||||
|
/// Cipher for encrypting data
|
||||||
|
pub(crate) cipher_send: CipherSend,
|
||||||
|
mtu: usize,
|
||||||
|
next_timer: TimerKind,
|
||||||
|
/// send queue for each Stream
|
||||||
|
send_queue: BTreeMap<stream::ID, stream::SendTracker>,
|
||||||
|
last_stream_sent: stream::ID,
|
||||||
|
/// receive queue for each Stream
|
||||||
|
recv_queue: BTreeMap<stream::ID, stream::Stream>,
|
||||||
|
streams_ready: VecDeque<stream::ID>,
|
||||||
|
}
|
||||||
|
|
||||||
|
impl Connection {
|
||||||
pub(crate) fn new(
|
pub(crate) fn new(
|
||||||
hkdf: Hkdf,
|
hkdf: Hkdf,
|
||||||
cipher: sym::Kind,
|
cipher: sym::Kind,
|
||||||
|
@ -178,19 +288,194 @@ impl Conn {
|
||||||
let cipher_recv = CipherRecv::new(cipher, secret_recv);
|
let cipher_recv = CipherRecv::new(cipher, secret_recv);
|
||||||
let cipher_send = CipherSend::new(cipher, secret_send, rand);
|
let cipher_send = CipherSend::new(cipher, secret_send, rand);
|
||||||
|
|
||||||
|
use ::std::net::{IpAddr, Ipv4Addr, SocketAddr};
|
||||||
Self {
|
Self {
|
||||||
id_recv: IDRecv(ID::Handshake),
|
id_recv: IDRecv(ID::Handshake),
|
||||||
id_send: IDSend(ID::Handshake),
|
id_send: IDSend(ID::Handshake),
|
||||||
|
user_tracker: None,
|
||||||
|
lib_tracker: LibTracker::new(0),
|
||||||
|
// will be overwritten
|
||||||
|
send_addr: UdpClient(SocketAddr::new(
|
||||||
|
IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)),
|
||||||
|
31337,
|
||||||
|
)),
|
||||||
hkdf,
|
hkdf,
|
||||||
cipher_recv,
|
cipher_recv,
|
||||||
cipher_send,
|
cipher_send,
|
||||||
|
mtu: 1200,
|
||||||
|
next_timer: TimerKind::None,
|
||||||
|
send_queue: BTreeMap::new(),
|
||||||
|
last_stream_sent: stream::ID(0),
|
||||||
|
recv_queue: BTreeMap::new(),
|
||||||
|
streams_ready: VecDeque::with_capacity(4),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
pub(crate) fn get_data(&mut self) -> Option<Vec<(stream::ID, Vec<u8>)>> {
|
||||||
|
if self.streams_ready.is_empty() {
|
||||||
|
return None;
|
||||||
|
}
|
||||||
|
let ret_len = self.streams_ready.len();
|
||||||
|
let mut ret = Vec::with_capacity(ret_len);
|
||||||
|
while let Some(stream_id) = self.streams_ready.pop_front() {
|
||||||
|
let stream = match self.recv_queue.get_mut(&stream_id) {
|
||||||
|
Some(stream) => stream,
|
||||||
|
None => continue,
|
||||||
|
};
|
||||||
|
ret.push((stream_id, stream.get()));
|
||||||
|
}
|
||||||
|
Some(ret)
|
||||||
|
}
|
||||||
|
pub(crate) fn recv(
|
||||||
|
&mut self,
|
||||||
|
mut udp: crate::RawUdp,
|
||||||
|
) -> Result<StreamData, Error> {
|
||||||
|
let mut data = &mut udp.data[ID::len()..];
|
||||||
|
let aad = enc::sym::AAD(&[]);
|
||||||
|
self.cipher_recv.decrypt(aad, &mut data)?;
|
||||||
|
let mut bytes_parsed = 0;
|
||||||
|
let mut chunks = Vec::with_capacity(2);
|
||||||
|
loop {
|
||||||
|
let chunk = match stream::Chunk::deserialize(&data[bytes_parsed..])
|
||||||
|
{
|
||||||
|
Ok(chunk) => chunk,
|
||||||
|
Err(e) => {
|
||||||
|
return Err(e.into());
|
||||||
|
}
|
||||||
|
};
|
||||||
|
bytes_parsed = bytes_parsed + chunk.len();
|
||||||
|
chunks.push(chunk);
|
||||||
|
if bytes_parsed == data.len() {
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
let mut data_ready = StreamData::NotReady;
|
||||||
|
for chunk in chunks.into_iter() {
|
||||||
|
let stream_id = chunk.id;
|
||||||
|
let stream = match self.recv_queue.get_mut(&stream_id) {
|
||||||
|
Some(stream) => stream,
|
||||||
|
None => {
|
||||||
|
::tracing::debug!("Ignoring chunk for unknown stream::ID");
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
};
|
||||||
|
match stream.recv(chunk) {
|
||||||
|
Ok(status) => {
|
||||||
|
if !self.streams_ready.contains(&stream_id) {
|
||||||
|
self.streams_ready.push_back(stream_id);
|
||||||
|
}
|
||||||
|
data_ready = data_ready | status;
|
||||||
|
}
|
||||||
|
Err(e) => ::tracing::debug!("stream: {:?}: {:?}", stream_id, e),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
Ok(data_ready)
|
||||||
|
}
|
||||||
|
pub(crate) fn enqueue(
|
||||||
|
&mut self,
|
||||||
|
stream: stream::ID,
|
||||||
|
data: Vec<u8>,
|
||||||
|
) -> Result<Enqueue, Error> {
|
||||||
|
let stream = match self.send_queue.get_mut(&stream) {
|
||||||
|
None => return Err(Error::NoSuchStream),
|
||||||
|
Some(stream) => stream,
|
||||||
|
};
|
||||||
|
stream.enqueue(data);
|
||||||
|
let instant;
|
||||||
|
let ret;
|
||||||
|
self.next_timer = match self.next_timer {
|
||||||
|
TimerKind::None | TimerKind::Keepalive(_) => {
|
||||||
|
instant = ::tokio::time::Instant::now();
|
||||||
|
ret = Enqueue::Immediate(instant);
|
||||||
|
TimerKind::SendData(instant)
|
||||||
|
}
|
||||||
|
TimerKind::SendData(old_timer) => {
|
||||||
|
// There already is some data to be sent
|
||||||
|
// wait for this timer,
|
||||||
|
// or risk going over max transmission rate
|
||||||
|
ret = Enqueue::TimerWait;
|
||||||
|
TimerKind::SendData(old_timer)
|
||||||
|
}
|
||||||
|
};
|
||||||
|
Ok(ret)
|
||||||
|
}
|
||||||
|
pub(crate) fn write_pkt<'a>(
|
||||||
|
&mut self,
|
||||||
|
raw: &'a mut [u8],
|
||||||
|
) -> Result<&'a [u8], enc::Error> {
|
||||||
|
assert!(raw.len() >= self.mtu, "I should have at least 1200 MTU");
|
||||||
|
if self.send_queue.len() == 0 {
|
||||||
|
return Err(enc::Error::NotEnoughData(0));
|
||||||
|
}
|
||||||
|
raw[..ID::len()]
|
||||||
|
.copy_from_slice(&self.id_send.0.as_u64().to_le_bytes());
|
||||||
|
let data_from = ID::len() + self.cipher_send.nonce_len().0;
|
||||||
|
let data_max_to = raw.len() - self.cipher_send.tag_len().0;
|
||||||
|
let mut chunk_from = data_from;
|
||||||
|
let mut available_len = data_max_to - data_from;
|
||||||
|
|
||||||
|
use std::ops::Bound::{Excluded, Included};
|
||||||
|
let last_stream = self.last_stream_sent;
|
||||||
|
|
||||||
|
// Loop over our streams, write them to the packet.
|
||||||
|
// Notes:
|
||||||
|
// * to avoid starvation, just round-robin them all for now
|
||||||
|
// * we can enqueue multiple times the same stream
|
||||||
|
// This is useful especially for Datagram streams
|
||||||
|
'queueloop: {
|
||||||
|
for (id, stream) in self
|
||||||
|
.send_queue
|
||||||
|
.range_mut((Included(last_stream), Included(stream::ID::max())))
|
||||||
|
{
|
||||||
|
if available_len < stream::Chunk::headers_len() + 1 {
|
||||||
|
break 'queueloop;
|
||||||
|
}
|
||||||
|
let bytes =
|
||||||
|
stream.serialize(*id, &mut raw[chunk_from..data_max_to]);
|
||||||
|
if bytes == 0 {
|
||||||
|
break 'queueloop;
|
||||||
|
}
|
||||||
|
available_len = available_len - bytes;
|
||||||
|
chunk_from = chunk_from + bytes;
|
||||||
|
self.last_stream_sent = *id;
|
||||||
|
}
|
||||||
|
if available_len > 0 {
|
||||||
|
for (id, stream) in self.send_queue.range_mut((
|
||||||
|
Included(stream::ID::min()),
|
||||||
|
Excluded(last_stream),
|
||||||
|
)) {
|
||||||
|
if available_len < stream::Chunk::headers_len() + 1 {
|
||||||
|
break 'queueloop;
|
||||||
|
}
|
||||||
|
let bytes = stream
|
||||||
|
.serialize(*id, &mut raw[chunk_from..data_max_to]);
|
||||||
|
if bytes == 0 {
|
||||||
|
break 'queueloop;
|
||||||
|
}
|
||||||
|
available_len = available_len - bytes;
|
||||||
|
chunk_from = chunk_from + bytes;
|
||||||
|
self.last_stream_sent = *id;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if chunk_from == data_from {
|
||||||
|
return Err(enc::Error::NotEnoughData(0));
|
||||||
|
}
|
||||||
|
let data_to = chunk_from + self.cipher_send.tag_len().0;
|
||||||
|
|
||||||
|
// encrypt
|
||||||
|
let aad = sym::AAD(&[]);
|
||||||
|
match self.cipher_send.encrypt(aad, &mut raw[data_from..data_to]) {
|
||||||
|
Ok(_) => Ok(&raw[..data_to]),
|
||||||
|
Err(e) => Err(e),
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
pub(crate) struct ConnList {
|
pub(crate) struct ConnList {
|
||||||
thread_id: ThreadTracker,
|
thread_id: ThreadTracker,
|
||||||
connections: Vec<Option<Rc<Conn>>>,
|
connections: Vec<Option<Connection>>,
|
||||||
|
user_tracker: BTreeMap<LibTracker, usize>,
|
||||||
|
last_tracked: LibTracker,
|
||||||
/// Bitmap to track which connection ids are used or free
|
/// Bitmap to track which connection ids are used or free
|
||||||
ids_used: Vec<::bitmaps::Bitmap<1024>>,
|
ids_used: Vec<::bitmaps::Bitmap<1024>>,
|
||||||
}
|
}
|
||||||
|
@ -206,11 +491,43 @@ impl ConnList {
|
||||||
let mut ret = Self {
|
let mut ret = Self {
|
||||||
thread_id,
|
thread_id,
|
||||||
connections: Vec::with_capacity(INITIAL_CAP),
|
connections: Vec::with_capacity(INITIAL_CAP),
|
||||||
|
user_tracker: BTreeMap::new(),
|
||||||
|
last_tracked: LibTracker(Wrapping(0)),
|
||||||
ids_used: vec![bitmap_id],
|
ids_used: vec![bitmap_id],
|
||||||
};
|
};
|
||||||
ret.connections.resize_with(INITIAL_CAP, || None);
|
ret.connections.resize_with(INITIAL_CAP, || None);
|
||||||
ret
|
ret
|
||||||
}
|
}
|
||||||
|
pub fn get_id_mut(&mut self, id: ID) -> Result<&mut Connection, Error> {
|
||||||
|
let conn_id = match id {
|
||||||
|
ID::ID(conn_id) => conn_id,
|
||||||
|
ID::Handshake => {
|
||||||
|
return Err(Error::NoSuchConnection);
|
||||||
|
}
|
||||||
|
};
|
||||||
|
let id_in_thread: usize =
|
||||||
|
(conn_id.get() / (self.thread_id.total as u64)) as usize;
|
||||||
|
if let Some(conn) = &mut self.connections[id_in_thread] {
|
||||||
|
Ok(conn)
|
||||||
|
} else {
|
||||||
|
return Err(Error::NoSuchConnection);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
pub fn get_mut(
|
||||||
|
&mut self,
|
||||||
|
tracker: LibTracker,
|
||||||
|
) -> Result<&mut Connection, Error> {
|
||||||
|
let idx = if let Some(idx) = self.user_tracker.get(&tracker) {
|
||||||
|
*idx
|
||||||
|
} else {
|
||||||
|
return Err(Error::NoSuchConnection);
|
||||||
|
};
|
||||||
|
if let Some(conn) = &mut self.connections[idx] {
|
||||||
|
Ok(conn)
|
||||||
|
} else {
|
||||||
|
return Err(Error::NoSuchConnection);
|
||||||
|
}
|
||||||
|
}
|
||||||
pub fn len(&self) -> usize {
|
pub fn len(&self) -> usize {
|
||||||
let mut total: usize = 0;
|
let mut total: usize = 0;
|
||||||
for bitmap in self.ids_used.iter() {
|
for bitmap in self.ids_used.iter() {
|
||||||
|
@ -220,7 +537,23 @@ impl ConnList {
|
||||||
}
|
}
|
||||||
/// Only *Reserve* a connection,
|
/// Only *Reserve* a connection,
|
||||||
/// without actually tracking it in self.connections
|
/// without actually tracking it in self.connections
|
||||||
|
pub(crate) fn reserve_and_track<'a>(
|
||||||
|
&'a mut self,
|
||||||
|
mut conn: Connection,
|
||||||
|
) -> (LibTracker, &'a mut Connection) {
|
||||||
|
let (id_conn, id_in_thread) = self.reserve_first_with_idx();
|
||||||
|
conn.id_recv = id_conn;
|
||||||
|
let tracker = self.get_new_tracker(id_in_thread);
|
||||||
|
conn.lib_tracker = tracker;
|
||||||
|
self.connections[id_in_thread] = Some(conn);
|
||||||
|
(tracker, self.connections[id_in_thread].as_mut().unwrap())
|
||||||
|
}
|
||||||
|
/// Only *Reserve* a connection,
|
||||||
|
/// without actually tracking it in self.connections
|
||||||
pub(crate) fn reserve_first(&mut self) -> IDRecv {
|
pub(crate) fn reserve_first(&mut self) -> IDRecv {
|
||||||
|
self.reserve_first_with_idx().0
|
||||||
|
}
|
||||||
|
fn reserve_first_with_idx(&mut self) -> (IDRecv, usize) {
|
||||||
// uhm... bad things are going on here:
|
// uhm... bad things are going on here:
|
||||||
// * id must be initialized, but only because:
|
// * id must be initialized, but only because:
|
||||||
// * rust does not understand that after the `!found` id is always
|
// * rust does not understand that after the `!found` id is always
|
||||||
|
@ -258,10 +591,13 @@ impl ConnList {
|
||||||
let actual_id = ((id_in_thread as u64) * (self.thread_id.total as u64))
|
let actual_id = ((id_in_thread as u64) * (self.thread_id.total as u64))
|
||||||
+ (self.thread_id.id as u64);
|
+ (self.thread_id.id as u64);
|
||||||
let new_id = IDRecv(ID::new_u64(actual_id));
|
let new_id = IDRecv(ID::new_u64(actual_id));
|
||||||
new_id
|
(new_id, id_in_thread)
|
||||||
}
|
}
|
||||||
/// NOTE: does NOT check if the connection has been previously reserved!
|
/// NOTE: does NOT check if the connection has been previously reserved!
|
||||||
pub(crate) fn track(&mut self, conn: Rc<Conn>) -> Result<(), ()> {
|
pub(crate) fn track(
|
||||||
|
&mut self,
|
||||||
|
mut conn: Connection,
|
||||||
|
) -> Result<LibTracker, ()> {
|
||||||
let conn_id = match conn.id_recv {
|
let conn_id = match conn.id_recv {
|
||||||
IDRecv(ID::Handshake) => {
|
IDRecv(ID::Handshake) => {
|
||||||
return Err(());
|
return Err(());
|
||||||
|
@ -270,8 +606,22 @@ impl ConnList {
|
||||||
};
|
};
|
||||||
let id_in_thread: usize =
|
let id_in_thread: usize =
|
||||||
(conn_id.get() / (self.thread_id.total as u64)) as usize;
|
(conn_id.get() / (self.thread_id.total as u64)) as usize;
|
||||||
|
let tracker = self.get_new_tracker(id_in_thread);
|
||||||
|
conn.lib_tracker = tracker;
|
||||||
self.connections[id_in_thread] = Some(conn);
|
self.connections[id_in_thread] = Some(conn);
|
||||||
Ok(())
|
Ok(tracker)
|
||||||
|
}
|
||||||
|
fn get_new_tracker(&mut self, id_in_thread: usize) -> LibTracker {
|
||||||
|
let mut tracker;
|
||||||
|
loop {
|
||||||
|
tracker = self.last_tracked.advance(self.thread_id.total);
|
||||||
|
if self.user_tracker.get(&tracker).is_none() {
|
||||||
|
// like, never gonna happen, it's 64 bit
|
||||||
|
let _ = self.user_tracker.insert(tracker, id_in_thread);
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
tracker
|
||||||
}
|
}
|
||||||
pub(crate) fn remove(&mut self, id: IDRecv) {
|
pub(crate) fn remove(&mut self, id: IDRecv) {
|
||||||
if let IDRecv(ID::ID(raw_id)) = id {
|
if let IDRecv(ID::ID(raw_id)) = id {
|
||||||
|
@ -303,7 +653,6 @@ enum MapEntry {
|
||||||
Present(IDSend),
|
Present(IDSend),
|
||||||
Reserved,
|
Reserved,
|
||||||
}
|
}
|
||||||
use ::std::collections::HashMap;
|
|
||||||
|
|
||||||
/// Link the public key of the authentication server to a connection id
|
/// Link the public key of the authentication server to a connection id
|
||||||
/// so that we can reuse that connection to ask for more authentications
|
/// so that we can reuse that connection to ask for more authentications
|
||||||
|
|
|
@ -1,10 +1,12 @@
|
||||||
//! Errors while parsing streams
|
//! Errors while parsing streams
|
||||||
|
|
||||||
|
|
||||||
/// Crypto errors
|
/// Crypto errors
|
||||||
#[derive(::thiserror::Error, Debug, Copy, Clone)]
|
#[derive(::thiserror::Error, Debug, Copy, Clone)]
|
||||||
pub enum Error {
|
pub enum Error {
|
||||||
/// Error while parsing key material
|
/// Error while parsing key material
|
||||||
#[error("Not enough data for stream chunk: {0}")]
|
#[error("Not enough data for stream chunk: {0}")]
|
||||||
NotEnoughData(usize),
|
NotEnoughData(usize),
|
||||||
|
/// Sequence outside of the window
|
||||||
|
#[error("Sequence out of the sliding window")]
|
||||||
|
OutOfWindow,
|
||||||
}
|
}
|
||||||
|
|
|
@ -19,7 +19,7 @@ pub enum Kind {
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Id of the stream
|
/// Id of the stream
|
||||||
#[derive(Debug, Copy, Clone)]
|
#[derive(Debug, Copy, Clone, PartialEq, Eq, PartialOrd, Ord)]
|
||||||
pub struct ID(pub u16);
|
pub struct ID(pub u16);
|
||||||
|
|
||||||
impl ID {
|
impl ID {
|
||||||
|
@ -27,6 +27,14 @@ impl ID {
|
||||||
pub const fn len() -> usize {
|
pub const fn len() -> usize {
|
||||||
2
|
2
|
||||||
}
|
}
|
||||||
|
/// Minimum possible Stream ID (u16::MIN)
|
||||||
|
pub const fn min() -> Self {
|
||||||
|
Self(u16::MIN)
|
||||||
|
}
|
||||||
|
/// Maximum possible Stream ID (u16::MAX)
|
||||||
|
pub const fn max() -> Self {
|
||||||
|
Self(u16::MAX)
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
/// length of the chunk
|
/// length of the chunk
|
||||||
|
@ -40,6 +48,30 @@ impl ChunkLen {
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
//TODO: make pub?
|
||||||
|
#[derive(Debug, Copy, Clone)]
|
||||||
|
pub(crate) struct SequenceStart(pub(crate) Sequence);
|
||||||
|
impl SequenceStart {
|
||||||
|
pub(crate) fn plus_u32(&self, other: u32) -> Sequence {
|
||||||
|
self.0.plus_u32(other)
|
||||||
|
}
|
||||||
|
pub(crate) fn offset(&self, seq: Sequence) -> usize {
|
||||||
|
if self.0 .0 <= seq.0 {
|
||||||
|
(seq.0 - self.0 .0).0 as usize
|
||||||
|
} else {
|
||||||
|
(seq.0 + (Sequence::max().0 - self.0 .0)).0 as usize
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
// SequenceEnd is INCLUSIVE
|
||||||
|
#[derive(Debug, Copy, Clone)]
|
||||||
|
pub(crate) struct SequenceEnd(pub(crate) Sequence);
|
||||||
|
impl SequenceEnd {
|
||||||
|
pub(crate) fn plus_u32(&self, other: u32) -> Sequence {
|
||||||
|
self.0.plus_u32(other)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
/// Sequence number to rebuild the stream correctly
|
/// Sequence number to rebuild the stream correctly
|
||||||
#[derive(Debug, Copy, Clone)]
|
#[derive(Debug, Copy, Clone)]
|
||||||
pub struct Sequence(pub ::core::num::Wrapping<u32>);
|
pub struct Sequence(pub ::core::num::Wrapping<u32>);
|
||||||
|
@ -48,14 +80,52 @@ impl Sequence {
|
||||||
const SEQ_NOFLAG: u32 = 0x3FFFFFFF;
|
const SEQ_NOFLAG: u32 = 0x3FFFFFFF;
|
||||||
/// return a new sequence number, starting at random
|
/// return a new sequence number, starting at random
|
||||||
pub fn new(rand: &Random) -> Self {
|
pub fn new(rand: &Random) -> Self {
|
||||||
let seq: u32 = 0;
|
let mut raw_seq: [u8; 4] = [0; 4];
|
||||||
rand.fill(&mut seq.to_le_bytes());
|
rand.fill(&mut raw_seq);
|
||||||
|
let seq = u32::from_le_bytes(raw_seq);
|
||||||
Self(::core::num::Wrapping(seq & Self::SEQ_NOFLAG))
|
Self(::core::num::Wrapping(seq & Self::SEQ_NOFLAG))
|
||||||
}
|
}
|
||||||
/// Length of the serialized field
|
/// Length of the serialized field
|
||||||
pub const fn len() -> usize {
|
pub const fn len() -> usize {
|
||||||
4
|
4
|
||||||
}
|
}
|
||||||
|
/// Maximum possible sequence
|
||||||
|
pub const fn max() -> Self {
|
||||||
|
Self(::core::num::Wrapping(Self::SEQ_NOFLAG))
|
||||||
|
}
|
||||||
|
pub(crate) fn is_between(
|
||||||
|
&self,
|
||||||
|
start: SequenceStart,
|
||||||
|
end: SequenceEnd,
|
||||||
|
) -> bool {
|
||||||
|
if start.0 .0 < end.0 .0 {
|
||||||
|
start.0 .0 <= self.0 && self.0 <= end.0 .0
|
||||||
|
} else {
|
||||||
|
start.0 .0 <= self.0 || self.0 <= end.0 .0
|
||||||
|
}
|
||||||
|
}
|
||||||
|
pub(crate) fn remaining_window(&self, end: SequenceEnd) -> u32 {
|
||||||
|
if self.0 <= end.0 .0 {
|
||||||
|
(end.0 .0 .0 - self.0 .0) + 1
|
||||||
|
} else {
|
||||||
|
end.0 .0 .0 + 1 + (Self::max().0 - self.0).0
|
||||||
|
}
|
||||||
|
}
|
||||||
|
pub(crate) fn plus_u32(self, other: u32) -> Self {
|
||||||
|
Self(::core::num::Wrapping(
|
||||||
|
(self.0 .0 + other) & Self::SEQ_NOFLAG,
|
||||||
|
))
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
impl ::core::ops::Add for Sequence {
|
||||||
|
type Output = Self;
|
||||||
|
|
||||||
|
fn add(self, other: Self) -> Self {
|
||||||
|
Self(::core::num::Wrapping(
|
||||||
|
(self.0 + other.0).0 & Self::SEQ_NOFLAG,
|
||||||
|
))
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Chunk of data representing a stream
|
/// Chunk of data representing a stream
|
||||||
|
@ -79,6 +149,10 @@ impl<'a> Chunk<'a> {
|
||||||
const FLAGS_EXCLUDED_BITMASK: u8 = 0x3F;
|
const FLAGS_EXCLUDED_BITMASK: u8 = 0x3F;
|
||||||
const FLAG_START_BITMASK: u8 = 0x80;
|
const FLAG_START_BITMASK: u8 = 0x80;
|
||||||
const FLAG_END_BITMASK: u8 = 0x40;
|
const FLAG_END_BITMASK: u8 = 0x40;
|
||||||
|
/// Return the length of the header of a Chunk
|
||||||
|
pub const fn headers_len() -> usize {
|
||||||
|
ID::len() + ChunkLen::len() + Sequence::len()
|
||||||
|
}
|
||||||
/// Returns the total length of the chunk, including headers
|
/// Returns the total length of the chunk, including headers
|
||||||
pub fn len(&self) -> usize {
|
pub fn len(&self) -> usize {
|
||||||
ID::len() + ChunkLen::len() + Sequence::len() + self.data.len()
|
ID::len() + ChunkLen::len() + Sequence::len() + self.data.len()
|
||||||
|
@ -164,6 +238,26 @@ impl Tracker {
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[derive(Debug, Eq, PartialEq)]
|
||||||
|
pub(crate) enum StreamData {
|
||||||
|
/// not enough data to return somthing to the user
|
||||||
|
NotReady = 0,
|
||||||
|
/// we can return something to the user
|
||||||
|
Ready,
|
||||||
|
}
|
||||||
|
impl ::core::ops::BitOr for StreamData {
|
||||||
|
type Output = Self;
|
||||||
|
|
||||||
|
// Required method
|
||||||
|
fn bitor(self, other: Self) -> Self::Output {
|
||||||
|
if self == StreamData::Ready || other == StreamData::Ready {
|
||||||
|
StreamData::Ready
|
||||||
|
} else {
|
||||||
|
StreamData::NotReady
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
/// Actual stream-tracking structure
|
/// Actual stream-tracking structure
|
||||||
#[derive(Debug, Clone)]
|
#[derive(Debug, Clone)]
|
||||||
pub(crate) struct Stream {
|
pub(crate) struct Stream {
|
||||||
|
@ -180,4 +274,79 @@ impl Stream {
|
||||||
data: Tracker::new(kind, rand),
|
data: Tracker::new(kind, rand),
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
pub(crate) fn recv(&mut self, chunk: Chunk) -> Result<StreamData, Error> {
|
||||||
|
match &mut self.data {
|
||||||
|
Tracker::ROB(tracker) => tracker.recv(chunk),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
pub(crate) fn get(&mut self) -> Vec<u8> {
|
||||||
|
match &mut self.data {
|
||||||
|
Tracker::ROB(tracker) => tracker.get(),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Track what has been sent and what has been ACK'd from a stream
|
||||||
|
#[derive(Debug)]
|
||||||
|
pub(crate) struct SendTracker {
|
||||||
|
queue: Vec<Vec<u8>>,
|
||||||
|
sent: Vec<usize>,
|
||||||
|
ackd: Vec<usize>,
|
||||||
|
chunk_started: bool,
|
||||||
|
is_datagram: bool,
|
||||||
|
next_sequence: Sequence,
|
||||||
|
}
|
||||||
|
impl SendTracker {
|
||||||
|
pub(crate) fn new(rand: &Random) -> Self {
|
||||||
|
Self {
|
||||||
|
queue: Vec::with_capacity(4),
|
||||||
|
sent: Vec::with_capacity(4),
|
||||||
|
ackd: Vec::with_capacity(4),
|
||||||
|
chunk_started: false,
|
||||||
|
is_datagram: false,
|
||||||
|
next_sequence: Sequence::new(rand),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
/// Enqueue user data to be sent
|
||||||
|
pub(crate) fn enqueue(&mut self, data: Vec<u8>) {
|
||||||
|
self.queue.push(data);
|
||||||
|
self.sent.push(0);
|
||||||
|
self.ackd.push(0);
|
||||||
|
}
|
||||||
|
/// Write the user data to the buffer and mark it as sent
|
||||||
|
pub(crate) fn get(&mut self, out: &mut [u8]) -> usize {
|
||||||
|
let data = match self.queue.get(0) {
|
||||||
|
Some(data) => data,
|
||||||
|
None => return 0,
|
||||||
|
};
|
||||||
|
let len = ::std::cmp::min(out.len(), data.len());
|
||||||
|
out[..len].copy_from_slice(&data[self.sent[0]..len]);
|
||||||
|
self.sent[0] = self.sent[0] + len;
|
||||||
|
len
|
||||||
|
}
|
||||||
|
/// Mark the sent data as successfully received from the receiver
|
||||||
|
pub(crate) fn ack(&mut self, size: usize) {
|
||||||
|
todo!()
|
||||||
|
}
|
||||||
|
pub(crate) fn serialize(&mut self, id: ID, raw: &mut [u8]) -> usize {
|
||||||
|
let max_data_len = raw.len() - Chunk::headers_len();
|
||||||
|
let data_len = ::std::cmp::min(max_data_len, self.queue[0].len());
|
||||||
|
let flag_start = !self.chunk_started;
|
||||||
|
let flag_end = self.is_datagram && data_len == self.queue[0].len();
|
||||||
|
let chunk = Chunk {
|
||||||
|
id,
|
||||||
|
flag_start,
|
||||||
|
flag_end,
|
||||||
|
sequence: self.next_sequence,
|
||||||
|
data: &self.queue[0][..data_len],
|
||||||
|
};
|
||||||
|
self.next_sequence = Sequence(
|
||||||
|
self.next_sequence.0 + ::core::num::Wrapping(data_len as u32),
|
||||||
|
);
|
||||||
|
if chunk.flag_end {
|
||||||
|
self.chunk_started = false;
|
||||||
|
}
|
||||||
|
chunk.serialize(raw);
|
||||||
|
data_len
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
|
@ -1,29 +0,0 @@
|
||||||
//! Implementation of the Reliable, Ordered, Bytestream transmission model
|
|
||||||
//! AKA: TCP-like
|
|
||||||
|
|
||||||
use crate::{
|
|
||||||
connection::stream::{Chunk, Error, Sequence},
|
|
||||||
enc::Random,
|
|
||||||
};
|
|
||||||
|
|
||||||
/// Reliable, Ordered, Bytestream stream tracker
|
|
||||||
/// AKA: TCP-like
|
|
||||||
#[derive(Debug, Clone)]
|
|
||||||
pub(crate) struct ReliableOrderedBytestream {
|
|
||||||
window_start: Sequence,
|
|
||||||
window_len: usize,
|
|
||||||
data: Vec<u8>,
|
|
||||||
}
|
|
||||||
|
|
||||||
impl ReliableOrderedBytestream {
|
|
||||||
pub(crate) fn new(rand: &Random) -> Self {
|
|
||||||
Self {
|
|
||||||
window_start: Sequence::new(rand),
|
|
||||||
window_len: 1048576, // 1MB. should be enough for anybody. (lol)
|
|
||||||
data: Vec::new(),
|
|
||||||
}
|
|
||||||
}
|
|
||||||
pub(crate) fn recv(&mut self, chunk: Chunk) -> Result<(), Error> {
|
|
||||||
todo!()
|
|
||||||
}
|
|
||||||
}
|
|
|
@ -0,0 +1,210 @@
|
||||||
|
//! Implementation of the Reliable, Ordered, Bytestream transmission model
|
||||||
|
//! AKA: TCP-like
|
||||||
|
|
||||||
|
use crate::{
|
||||||
|
connection::stream::{
|
||||||
|
Chunk, Error, Sequence, SequenceEnd, SequenceStart, StreamData,
|
||||||
|
},
|
||||||
|
enc::Random,
|
||||||
|
};
|
||||||
|
|
||||||
|
#[cfg(test)]
|
||||||
|
mod tests;
|
||||||
|
|
||||||
|
/// Reliable, Ordered, Bytestream stream tracker
|
||||||
|
/// AKA: TCP-like
|
||||||
|
#[derive(Debug, Clone)]
|
||||||
|
pub(crate) struct ReliableOrderedBytestream {
|
||||||
|
pub(crate) window_start: SequenceStart,
|
||||||
|
window_end: SequenceEnd,
|
||||||
|
pivot: u32,
|
||||||
|
data: Vec<u8>,
|
||||||
|
missing: Vec<(Sequence, Sequence)>,
|
||||||
|
}
|
||||||
|
|
||||||
|
impl ReliableOrderedBytestream {
|
||||||
|
pub(crate) fn new(rand: &Random) -> Self {
|
||||||
|
let window_len = 1048576; // 1MB. should be enough for anybody. (lol)
|
||||||
|
let window_start = SequenceStart(Sequence::new(rand));
|
||||||
|
let window_end = SequenceEnd(window_start.0.plus_u32(window_len - 1));
|
||||||
|
let mut data = Vec::with_capacity(window_len as usize);
|
||||||
|
data.resize(data.capacity(), 0);
|
||||||
|
|
||||||
|
Self {
|
||||||
|
window_start,
|
||||||
|
window_end,
|
||||||
|
pivot: window_len,
|
||||||
|
data,
|
||||||
|
missing: [(window_start.0, window_end.0)].to_vec(),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
pub(crate) fn with_window_size(rand: &Random, size: u32) -> Self {
|
||||||
|
assert!(
|
||||||
|
size < Sequence::max().0 .0,
|
||||||
|
"Max window size is {}",
|
||||||
|
Sequence::max().0 .0
|
||||||
|
);
|
||||||
|
let window_len = size; // 1MB. should be enough for anybody. (lol)
|
||||||
|
let window_start = SequenceStart(Sequence::new(rand));
|
||||||
|
let window_end = SequenceEnd(window_start.0.plus_u32(window_len - 1));
|
||||||
|
let mut data = Vec::with_capacity(window_len as usize);
|
||||||
|
data.resize(data.capacity(), 0);
|
||||||
|
|
||||||
|
Self {
|
||||||
|
window_start,
|
||||||
|
window_end,
|
||||||
|
pivot: window_len,
|
||||||
|
data,
|
||||||
|
missing: [(window_start.0, window_end.0)].to_vec(),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
pub(crate) fn window_size(&self) -> u32 {
|
||||||
|
self.data.len() as u32
|
||||||
|
}
|
||||||
|
pub(crate) fn get(&mut self) -> Vec<u8> {
|
||||||
|
if self.missing.len() == 0 {
|
||||||
|
let (first, second) = self.data.split_at(self.pivot as usize);
|
||||||
|
let mut ret = Vec::with_capacity(self.data.len());
|
||||||
|
ret.extend_from_slice(first);
|
||||||
|
ret.extend_from_slice(second);
|
||||||
|
self.window_start =
|
||||||
|
SequenceStart(self.window_start.plus_u32(ret.len() as u32));
|
||||||
|
self.window_end =
|
||||||
|
SequenceEnd(self.window_end.plus_u32(ret.len() as u32));
|
||||||
|
self.data.clear();
|
||||||
|
return ret;
|
||||||
|
}
|
||||||
|
let data_len = self.window_start.offset(self.missing[0].0);
|
||||||
|
let last_missing_idx = self.missing.len() - 1;
|
||||||
|
let mut last_missing = &mut self.missing[last_missing_idx];
|
||||||
|
last_missing.1 = last_missing.1.plus_u32(data_len as u32);
|
||||||
|
self.window_start =
|
||||||
|
SequenceStart(self.window_start.plus_u32(data_len as u32));
|
||||||
|
self.window_end =
|
||||||
|
SequenceEnd(self.window_end.plus_u32(data_len as u32));
|
||||||
|
|
||||||
|
let mut ret = Vec::with_capacity(data_len);
|
||||||
|
let (first, second) = self.data[..].split_at(self.pivot as usize);
|
||||||
|
let first_len = ::core::cmp::min(data_len, first.len());
|
||||||
|
let second_len = data_len - first_len;
|
||||||
|
|
||||||
|
ret.extend_from_slice(&first[..first_len]);
|
||||||
|
ret.extend_from_slice(&second[..second_len]);
|
||||||
|
|
||||||
|
self.pivot =
|
||||||
|
((self.pivot as usize + data_len) % self.data.len()) as u32;
|
||||||
|
ret
|
||||||
|
}
|
||||||
|
pub(crate) fn recv(&mut self, chunk: Chunk) -> Result<StreamData, Error> {
|
||||||
|
if !chunk
|
||||||
|
.sequence
|
||||||
|
.is_between(self.window_start, self.window_end)
|
||||||
|
{
|
||||||
|
return Err(Error::OutOfWindow);
|
||||||
|
}
|
||||||
|
// make sure we consider only the bytes inside the sliding window
|
||||||
|
let maxlen = ::std::cmp::min(
|
||||||
|
chunk.sequence.remaining_window(self.window_end) as usize,
|
||||||
|
chunk.data.len(),
|
||||||
|
);
|
||||||
|
if maxlen == 0 {
|
||||||
|
// empty window or empty chunk, but we don't care
|
||||||
|
return Err(Error::OutOfWindow);
|
||||||
|
}
|
||||||
|
// translate Sequences to offsets in self.data
|
||||||
|
let data = &chunk.data[..maxlen];
|
||||||
|
let offset = self.window_start.offset(chunk.sequence);
|
||||||
|
let offset_end = offset + chunk.data.len() - 1;
|
||||||
|
|
||||||
|
// Find the chunks we are missing that we can copy,
|
||||||
|
// and fix the missing tracker
|
||||||
|
let mut copy_ranges = Vec::new();
|
||||||
|
let mut to_delete = Vec::new();
|
||||||
|
let mut to_add = Vec::new();
|
||||||
|
// note: the ranges are (INCLUSIVE, INCLUSIVE)
|
||||||
|
for (idx, el) in self.missing.iter_mut().enumerate() {
|
||||||
|
let missing_from = self.window_start.offset(el.0);
|
||||||
|
if missing_from > offset_end {
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
let missing_to = self.window_start.offset(el.1);
|
||||||
|
if missing_to < offset {
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
if missing_from >= offset && missing_from <= offset_end {
|
||||||
|
if missing_to <= offset_end {
|
||||||
|
// [.....chunk.....]
|
||||||
|
// [..missing..]
|
||||||
|
to_delete.push(idx);
|
||||||
|
copy_ranges.push((missing_from, missing_to));
|
||||||
|
} else {
|
||||||
|
// [....chunk....]
|
||||||
|
// [...missing...]
|
||||||
|
copy_ranges.push((missing_from, offset_end));
|
||||||
|
el.0 =
|
||||||
|
el.0.plus_u32(((offset_end - missing_from) + 1) as u32);
|
||||||
|
}
|
||||||
|
} else if missing_from < offset {
|
||||||
|
if missing_to > offset_end {
|
||||||
|
// [..chunk..]
|
||||||
|
// [....missing....]
|
||||||
|
to_add.push((
|
||||||
|
el.0.plus_u32(((offset_end - missing_from) + 1) as u32),
|
||||||
|
el.1,
|
||||||
|
));
|
||||||
|
el.1 = el.0.plus_u32(((offset - missing_from) - 1) as u32);
|
||||||
|
copy_ranges.push((offset, offset_end));
|
||||||
|
} else if offset <= missing_to {
|
||||||
|
// [....chunk....]
|
||||||
|
// [...missing...]
|
||||||
|
copy_ranges.push((offset, (missing_to - 0)));
|
||||||
|
el.1 =
|
||||||
|
el.0.plus_u32(((offset_end - missing_from) - 1) as u32);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
{
|
||||||
|
let mut deleted = 0;
|
||||||
|
for idx in to_delete.into_iter() {
|
||||||
|
self.missing.remove(idx + deleted);
|
||||||
|
deleted = deleted + 1;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
self.missing.append(&mut to_add);
|
||||||
|
self.missing
|
||||||
|
.sort_by(|(from_a, _), (from_b, _)| from_a.0 .0.cmp(&from_b.0 .0));
|
||||||
|
|
||||||
|
// copy only the missing data
|
||||||
|
let (first, second) = self.data[..].split_at_mut(self.pivot as usize);
|
||||||
|
for (from, to) in copy_ranges.into_iter() {
|
||||||
|
let to = to + 1;
|
||||||
|
if from <= first.len() {
|
||||||
|
let first_from = from;
|
||||||
|
let first_to = ::core::cmp::min(first.len(), to);
|
||||||
|
let data_first_from = from - offset;
|
||||||
|
let data_first_to = first_to - offset;
|
||||||
|
first[first_from..first_to]
|
||||||
|
.copy_from_slice(&data[data_first_from..data_first_to]);
|
||||||
|
|
||||||
|
let second_to = to - first_to;
|
||||||
|
let data_second_to = data_first_to + second_to;
|
||||||
|
second[..second_to]
|
||||||
|
.copy_from_slice(&data[data_first_to..data_second_to]);
|
||||||
|
} else {
|
||||||
|
let second_from = from - first.len();
|
||||||
|
let second_to = to - first.len();
|
||||||
|
let data_from = from - offset;
|
||||||
|
let data_to = to - offset;
|
||||||
|
second[second_from..second_to]
|
||||||
|
.copy_from_slice(&data[data_from..data_to]);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if self.missing.len() == 0
|
||||||
|
|| self.window_start.offset(self.missing[0].0) == 0
|
||||||
|
{
|
||||||
|
Ok(StreamData::Ready)
|
||||||
|
} else {
|
||||||
|
Ok(StreamData::NotReady)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
|
@ -0,0 +1,249 @@
|
||||||
|
use crate::{
|
||||||
|
connection::stream::{self, rob::*, Chunk},
|
||||||
|
enc::Random,
|
||||||
|
};
|
||||||
|
|
||||||
|
#[::tracing_test::traced_test]
|
||||||
|
#[test]
|
||||||
|
fn test_stream_rob_sequential() {
|
||||||
|
let rand = Random::new();
|
||||||
|
let mut rob = ReliableOrderedBytestream::with_window_size(&rand, 1048576);
|
||||||
|
|
||||||
|
let mut data = Vec::with_capacity(1024);
|
||||||
|
data.resize(data.capacity(), 0);
|
||||||
|
rand.fill(&mut data[..]);
|
||||||
|
|
||||||
|
let start = rob.window_start.0;
|
||||||
|
|
||||||
|
let chunk = Chunk {
|
||||||
|
id: stream::ID(42),
|
||||||
|
flag_start: true,
|
||||||
|
flag_end: false,
|
||||||
|
sequence: start,
|
||||||
|
data: &data[..512],
|
||||||
|
};
|
||||||
|
let got = rob.get();
|
||||||
|
assert!(&got[..] == &[], "rob: got data?");
|
||||||
|
let _ = rob.recv(chunk);
|
||||||
|
let got = rob.get();
|
||||||
|
assert!(
|
||||||
|
&data[..512] == &got[..],
|
||||||
|
"ROB1: DIFF: {:?} {:?}",
|
||||||
|
&data[..512].len(),
|
||||||
|
&got[..].len()
|
||||||
|
);
|
||||||
|
let chunk = Chunk {
|
||||||
|
id: stream::ID(42),
|
||||||
|
flag_start: false,
|
||||||
|
flag_end: true,
|
||||||
|
sequence: start.plus_u32(512),
|
||||||
|
data: &data[512..],
|
||||||
|
};
|
||||||
|
let _ = rob.recv(chunk);
|
||||||
|
let got = rob.get();
|
||||||
|
assert!(
|
||||||
|
&data[512..] == &got[..],
|
||||||
|
"ROB2: DIFF: {:?} {:?}",
|
||||||
|
&data[512..].len(),
|
||||||
|
&got[..].len()
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
|
#[::tracing_test::traced_test]
|
||||||
|
#[test]
|
||||||
|
fn test_stream_rob_retransmit() {
|
||||||
|
let rand = Random::new();
|
||||||
|
let max_window: usize = 100;
|
||||||
|
let mut rob =
|
||||||
|
ReliableOrderedBytestream::with_window_size(&rand, max_window as u32);
|
||||||
|
|
||||||
|
let mut data = Vec::with_capacity(120);
|
||||||
|
data.resize(data.capacity(), 0);
|
||||||
|
for i in 0..data.len() {
|
||||||
|
data[i] = i as u8;
|
||||||
|
}
|
||||||
|
|
||||||
|
let start = rob.window_start.0;
|
||||||
|
|
||||||
|
let chunk = Chunk {
|
||||||
|
id: stream::ID(42),
|
||||||
|
flag_start: true,
|
||||||
|
flag_end: false,
|
||||||
|
sequence: start,
|
||||||
|
data: &data[..40],
|
||||||
|
};
|
||||||
|
let _ = rob.recv(chunk);
|
||||||
|
let chunk = Chunk {
|
||||||
|
id: stream::ID(42),
|
||||||
|
flag_start: false,
|
||||||
|
flag_end: false,
|
||||||
|
sequence: start.plus_u32(50),
|
||||||
|
data: &data[50..60],
|
||||||
|
};
|
||||||
|
let _ = rob.recv(chunk);
|
||||||
|
let chunk = Chunk {
|
||||||
|
id: stream::ID(42),
|
||||||
|
flag_start: false,
|
||||||
|
flag_end: false,
|
||||||
|
sequence: start.plus_u32(40),
|
||||||
|
data: &data[40..60],
|
||||||
|
};
|
||||||
|
let _ = rob.recv(chunk);
|
||||||
|
let chunk = Chunk {
|
||||||
|
id: stream::ID(42),
|
||||||
|
flag_start: false,
|
||||||
|
flag_end: false,
|
||||||
|
sequence: start.plus_u32(80),
|
||||||
|
data: &data[80..],
|
||||||
|
};
|
||||||
|
let _ = rob.recv(chunk);
|
||||||
|
let chunk = Chunk {
|
||||||
|
id: stream::ID(42),
|
||||||
|
flag_start: false,
|
||||||
|
flag_end: false,
|
||||||
|
sequence: start.plus_u32(50),
|
||||||
|
data: &data[50..90],
|
||||||
|
};
|
||||||
|
let _ = rob.recv(chunk);
|
||||||
|
let chunk = Chunk {
|
||||||
|
id: stream::ID(42),
|
||||||
|
flag_start: false,
|
||||||
|
flag_end: false,
|
||||||
|
sequence: start.plus_u32(max_window as u32),
|
||||||
|
data: &data[max_window..],
|
||||||
|
};
|
||||||
|
let _ = rob.recv(chunk);
|
||||||
|
let chunk = Chunk {
|
||||||
|
id: stream::ID(42),
|
||||||
|
flag_start: false,
|
||||||
|
flag_end: true,
|
||||||
|
sequence: start.plus_u32(90),
|
||||||
|
data: &data[90..max_window],
|
||||||
|
};
|
||||||
|
let _ = rob.recv(chunk);
|
||||||
|
let got = rob.get();
|
||||||
|
assert!(
|
||||||
|
&data[..max_window] == &got[..],
|
||||||
|
"DIFF:\n {:?}\n {:?}",
|
||||||
|
&data[..max_window],
|
||||||
|
&got[..],
|
||||||
|
);
|
||||||
|
}
|
||||||
|
#[::tracing_test::traced_test]
|
||||||
|
#[test]
|
||||||
|
fn test_stream_rob_rolling() {
|
||||||
|
let rand = Random::new();
|
||||||
|
let max_window: usize = 100;
|
||||||
|
let mut rob =
|
||||||
|
ReliableOrderedBytestream::with_window_size(&rand, max_window as u32);
|
||||||
|
|
||||||
|
let mut data = Vec::with_capacity(120);
|
||||||
|
data.resize(data.capacity(), 0);
|
||||||
|
for i in 0..data.len() {
|
||||||
|
data[i] = i as u8;
|
||||||
|
}
|
||||||
|
|
||||||
|
let start = rob.window_start.0;
|
||||||
|
|
||||||
|
let chunk = Chunk {
|
||||||
|
id: stream::ID(42),
|
||||||
|
flag_start: true,
|
||||||
|
flag_end: false,
|
||||||
|
sequence: start,
|
||||||
|
data: &data[..40],
|
||||||
|
};
|
||||||
|
let _ = rob.recv(chunk);
|
||||||
|
let chunk = Chunk {
|
||||||
|
id: stream::ID(42),
|
||||||
|
flag_start: true,
|
||||||
|
flag_end: false,
|
||||||
|
sequence: start.plus_u32(50),
|
||||||
|
data: &data[50..100],
|
||||||
|
};
|
||||||
|
let _ = rob.recv(chunk);
|
||||||
|
let got = rob.get();
|
||||||
|
assert!(
|
||||||
|
&data[..40] == &got[..],
|
||||||
|
"DIFF:\n {:?}\n {:?}",
|
||||||
|
&data[..40],
|
||||||
|
&got[..],
|
||||||
|
);
|
||||||
|
let chunk = Chunk {
|
||||||
|
id: stream::ID(42),
|
||||||
|
flag_start: true,
|
||||||
|
flag_end: false,
|
||||||
|
sequence: start.plus_u32(40),
|
||||||
|
data: &data[40..],
|
||||||
|
};
|
||||||
|
let _ = rob.recv(chunk);
|
||||||
|
let got = rob.get();
|
||||||
|
assert!(
|
||||||
|
&data[40..] == &got[..],
|
||||||
|
"DIFF:\n {:?}\n {:?}",
|
||||||
|
&data[40..],
|
||||||
|
&got[..],
|
||||||
|
);
|
||||||
|
}
|
||||||
|
#[::tracing_test::traced_test]
|
||||||
|
#[test]
|
||||||
|
fn test_stream_rob_rolling_second_case() {
|
||||||
|
let rand = Random::new();
|
||||||
|
let max_window: usize = 100;
|
||||||
|
let mut rob =
|
||||||
|
ReliableOrderedBytestream::with_window_size(&rand, max_window as u32);
|
||||||
|
|
||||||
|
let mut data = Vec::with_capacity(120);
|
||||||
|
data.resize(data.capacity(), 0);
|
||||||
|
for i in 0..data.len() {
|
||||||
|
data[i] = i as u8;
|
||||||
|
}
|
||||||
|
|
||||||
|
let start = rob.window_start.0;
|
||||||
|
|
||||||
|
let chunk = Chunk {
|
||||||
|
id: stream::ID(42),
|
||||||
|
flag_start: true,
|
||||||
|
flag_end: false,
|
||||||
|
sequence: start,
|
||||||
|
data: &data[..40],
|
||||||
|
};
|
||||||
|
let _ = rob.recv(chunk);
|
||||||
|
let chunk = Chunk {
|
||||||
|
id: stream::ID(42),
|
||||||
|
flag_start: true,
|
||||||
|
flag_end: false,
|
||||||
|
sequence: start.plus_u32(50),
|
||||||
|
data: &data[50..100],
|
||||||
|
};
|
||||||
|
let _ = rob.recv(chunk);
|
||||||
|
let got = rob.get();
|
||||||
|
assert!(
|
||||||
|
&data[..40] == &got[..],
|
||||||
|
"DIFF:\n {:?}\n {:?}",
|
||||||
|
&data[..40],
|
||||||
|
&got[..],
|
||||||
|
);
|
||||||
|
let chunk = Chunk {
|
||||||
|
id: stream::ID(42),
|
||||||
|
flag_start: true,
|
||||||
|
flag_end: false,
|
||||||
|
sequence: start.plus_u32(40),
|
||||||
|
data: &data[40..100],
|
||||||
|
};
|
||||||
|
let _ = rob.recv(chunk);
|
||||||
|
let chunk = Chunk {
|
||||||
|
id: stream::ID(42),
|
||||||
|
flag_start: true,
|
||||||
|
flag_end: false,
|
||||||
|
sequence: start.plus_u32(100),
|
||||||
|
data: &data[100..],
|
||||||
|
};
|
||||||
|
let _ = rob.recv(chunk);
|
||||||
|
let got = rob.get();
|
||||||
|
assert!(
|
||||||
|
&data[40..] == &got[..],
|
||||||
|
"DIFF:\n {:?}\n {:?}",
|
||||||
|
&data[40..],
|
||||||
|
&got[..],
|
||||||
|
);
|
||||||
|
}
|
|
@ -241,6 +241,14 @@ impl CipherSend {
|
||||||
pub fn kind(&self) -> Kind {
|
pub fn kind(&self) -> Kind {
|
||||||
self.cipher.kind()
|
self.cipher.kind()
|
||||||
}
|
}
|
||||||
|
/// Get the length of the nonce for this cipher
|
||||||
|
pub fn nonce_len(&self) -> NonceLen {
|
||||||
|
self.cipher.nonce_len()
|
||||||
|
}
|
||||||
|
/// Get the length of the nonce for this cipher
|
||||||
|
pub fn tag_len(&self) -> TagLen {
|
||||||
|
self.cipher.tag_len()
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
/// XChaCha20Poly1305 cipher
|
/// XChaCha20Poly1305 cipher
|
||||||
|
|
110
src/inner/mod.rs
110
src/inner/mod.rs
|
@ -4,6 +4,10 @@
|
||||||
|
|
||||||
pub(crate) mod worker;
|
pub(crate) mod worker;
|
||||||
|
|
||||||
|
use crate::inner::worker::Work;
|
||||||
|
use ::std::{collections::BTreeMap, vec::Vec};
|
||||||
|
use ::tokio::time::Instant;
|
||||||
|
|
||||||
/// Track the total number of threads and our index
|
/// Track the total number of threads and our index
|
||||||
/// 65K cpus should be enough for anybody
|
/// 65K cpus should be enough for anybody
|
||||||
#[derive(Debug, Clone, Copy)]
|
#[derive(Debug, Clone, Copy)]
|
||||||
|
@ -12,3 +16,109 @@ pub(crate) struct ThreadTracker {
|
||||||
/// Note: starts from 1
|
/// Note: starts from 1
|
||||||
pub id: u16,
|
pub id: u16,
|
||||||
}
|
}
|
||||||
|
|
||||||
|
pub(crate) static mut SLEEP_RESOLUTION: ::std::time::Duration =
|
||||||
|
if cfg!(linux) || cfg!(macos) {
|
||||||
|
::std::time::Duration::from_millis(1)
|
||||||
|
} else {
|
||||||
|
// windows
|
||||||
|
::std::time::Duration::from_millis(16)
|
||||||
|
};
|
||||||
|
|
||||||
|
pub(crate) async fn set_minimum_sleep_resolution() {
|
||||||
|
let nanosleep = ::std::time::Duration::from_nanos(1);
|
||||||
|
let mut tests: usize = 3;
|
||||||
|
|
||||||
|
while tests > 0 {
|
||||||
|
let pre_sleep = ::std::time::Instant::now();
|
||||||
|
::tokio::time::sleep(nanosleep).await;
|
||||||
|
let post_sleep = ::std::time::Instant::now();
|
||||||
|
let slept_for = post_sleep - pre_sleep;
|
||||||
|
#[allow(unsafe_code)]
|
||||||
|
unsafe {
|
||||||
|
if slept_for < SLEEP_RESOLUTION {
|
||||||
|
SLEEP_RESOLUTION = slept_for;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
tests = tests - 1;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Sleeping has a higher resolution that we would like for packet pacing.
|
||||||
|
/// So we sleep for however log we need, then chunk up all the work here
|
||||||
|
/// we will end up chunking the work in SLEEP_RESOLUTION, then we will busy wait
|
||||||
|
/// for more precise timing
|
||||||
|
pub(crate) struct Timers {
|
||||||
|
times: BTreeMap<Instant, Work>,
|
||||||
|
}
|
||||||
|
|
||||||
|
impl Timers {
|
||||||
|
pub(crate) fn new() -> Self {
|
||||||
|
Self {
|
||||||
|
times: BTreeMap::new(),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
pub(crate) fn get_next(&self) -> ::tokio::time::Sleep {
|
||||||
|
match self.times.keys().next() {
|
||||||
|
Some(entry) => ::tokio::time::sleep_until((*entry).into()),
|
||||||
|
None => {
|
||||||
|
::tokio::time::sleep(::std::time::Duration::from_secs(3600))
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
pub(crate) fn add(
|
||||||
|
&mut self,
|
||||||
|
duration: ::tokio::time::Duration,
|
||||||
|
work: Work,
|
||||||
|
) -> ::tokio::time::Instant {
|
||||||
|
// the returned time is the key in the map.
|
||||||
|
// Make sure it is unique.
|
||||||
|
//
|
||||||
|
// We can be pretty sure we won't do a lot of stuff
|
||||||
|
// in a single nanosecond, so if we hit a time that is already present
|
||||||
|
// just add a nanosecond and retry
|
||||||
|
let mut time = ::tokio::time::Instant::now() + duration;
|
||||||
|
let mut work = work;
|
||||||
|
loop {
|
||||||
|
if let Some(old_val) = self.times.insert(time, work) {
|
||||||
|
work = self.times.insert(time, old_val).unwrap();
|
||||||
|
time = time + ::std::time::Duration::from_nanos(1);
|
||||||
|
} else {
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
time
|
||||||
|
}
|
||||||
|
pub(crate) fn remove(&mut self, time: ::tokio::time::Instant) {
|
||||||
|
let _ = self.times.remove(&time);
|
||||||
|
}
|
||||||
|
/// Get all the work from now up until now + SLEEP_RESOLUTION
|
||||||
|
pub(crate) fn get_work(&mut self) -> Vec<Work> {
|
||||||
|
let now: ::tokio::time::Instant = ::std::time::Instant::now().into();
|
||||||
|
let mut ret = Vec::with_capacity(4);
|
||||||
|
let mut count_rm = 0;
|
||||||
|
#[allow(unsafe_code)]
|
||||||
|
let next_instant = unsafe { now + SLEEP_RESOLUTION };
|
||||||
|
let mut iter = self.times.iter_mut().peekable();
|
||||||
|
loop {
|
||||||
|
match iter.peek() {
|
||||||
|
None => break,
|
||||||
|
Some(next) => {
|
||||||
|
if *next.0 > next_instant {
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
let mut work = Work::DropHandshake(crate::enc::asym::KeyID(0));
|
||||||
|
let mut entry = iter.next().unwrap();
|
||||||
|
::core::mem::swap(&mut work, &mut entry.1);
|
||||||
|
ret.push(work);
|
||||||
|
count_rm = count_rm + 1;
|
||||||
|
}
|
||||||
|
while count_rm > 0 {
|
||||||
|
self.times.pop_first();
|
||||||
|
count_rm = count_rm - 1;
|
||||||
|
}
|
||||||
|
ret
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
|
@ -11,17 +11,19 @@ use crate::{
|
||||||
},
|
},
|
||||||
packet::{self, Packet},
|
packet::{self, Packet},
|
||||||
socket::{UdpClient, UdpServer},
|
socket::{UdpClient, UdpServer},
|
||||||
Conn, ConnList, IDSend,
|
stream, AuthSrvConn, ConnList, ConnTracker, Connection, IDSend,
|
||||||
|
LibTracker, ServiceConn,
|
||||||
},
|
},
|
||||||
dnssec,
|
dnssec,
|
||||||
enc::{
|
enc::{
|
||||||
|
self,
|
||||||
asym::{self, KeyID, PrivKey, PubKey},
|
asym::{self, KeyID, PrivKey, PubKey},
|
||||||
hkdf::{self, Hkdf},
|
hkdf::{self, Hkdf},
|
||||||
sym, Random, Secret,
|
sym, Random, Secret,
|
||||||
},
|
},
|
||||||
inner::ThreadTracker,
|
inner::ThreadTracker,
|
||||||
};
|
};
|
||||||
use ::std::{sync::Arc, vec::Vec};
|
use ::std::{collections::VecDeque, sync::Arc, vec::Vec};
|
||||||
/// This worker must be cpu-pinned
|
/// This worker must be cpu-pinned
|
||||||
use ::tokio::{
|
use ::tokio::{
|
||||||
net::UdpSocket,
|
net::UdpSocket,
|
||||||
|
@ -44,6 +46,25 @@ pub(crate) struct ConnectInfo {
|
||||||
// TODO: UserID, Token information
|
// TODO: UserID, Token information
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// return to the user the data received from a connection
|
||||||
|
#[derive(Debug, Clone)]
|
||||||
|
pub struct ConnData {
|
||||||
|
/// Connection tracking information
|
||||||
|
pub conn: ConnTracker,
|
||||||
|
/// received data, for each stream
|
||||||
|
pub data: Vec<(stream::ID, Vec<u8>)>,
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Connection event. Mostly used to give the data to the user
|
||||||
|
#[derive(Debug, Clone)]
|
||||||
|
#[non_exhaustive]
|
||||||
|
pub enum Event {
|
||||||
|
/// Work loop has exited. nothing more to do
|
||||||
|
End,
|
||||||
|
/// Data from a connection
|
||||||
|
Data(ConnData),
|
||||||
|
}
|
||||||
|
|
||||||
pub(crate) enum Work {
|
pub(crate) enum Work {
|
||||||
/// ask the thread to report to the main thread the total number of
|
/// ask the thread to report to the main thread the total number of
|
||||||
/// connections present
|
/// connections present
|
||||||
|
@ -51,6 +72,8 @@ pub(crate) enum Work {
|
||||||
Connect(ConnectInfo),
|
Connect(ConnectInfo),
|
||||||
DropHandshake(KeyID),
|
DropHandshake(KeyID),
|
||||||
Recv(RawUdp),
|
Recv(RawUdp),
|
||||||
|
UserSend((LibTracker, stream::ID, Vec<u8>)),
|
||||||
|
SendData((LibTracker, ::tokio::time::Instant)),
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Actual worker implementation.
|
/// Actual worker implementation.
|
||||||
|
@ -64,11 +87,15 @@ pub struct Worker {
|
||||||
token_check: Option<Arc<Mutex<TokenChecker>>>,
|
token_check: Option<Arc<Mutex<TokenChecker>>>,
|
||||||
sockets: Vec<Arc<UdpSocket>>,
|
sockets: Vec<Arc<UdpSocket>>,
|
||||||
queue: ::async_channel::Receiver<Work>,
|
queue: ::async_channel::Receiver<Work>,
|
||||||
|
queue_sender: ::async_channel::Sender<Work>,
|
||||||
queue_timeouts_recv: mpsc::UnboundedReceiver<Work>,
|
queue_timeouts_recv: mpsc::UnboundedReceiver<Work>,
|
||||||
queue_timeouts_send: mpsc::UnboundedSender<Work>,
|
queue_timeouts_send: mpsc::UnboundedSender<Work>,
|
||||||
thread_channels: Vec<::async_channel::Sender<Work>>,
|
thread_channels: Vec<::async_channel::Sender<Work>>,
|
||||||
connections: ConnList,
|
connections: ConnList,
|
||||||
|
// connectsion untracker by the user. (users still needs to get(..) them)
|
||||||
|
untracked_connections: VecDeque<LibTracker>,
|
||||||
handshakes: handshake::Tracker,
|
handshakes: handshake::Tracker,
|
||||||
|
work_timers: super::Timers,
|
||||||
}
|
}
|
||||||
|
|
||||||
#[allow(unsafe_code)]
|
#[allow(unsafe_code)]
|
||||||
|
@ -82,6 +109,7 @@ impl Worker {
|
||||||
token_check: Option<Arc<Mutex<TokenChecker>>>,
|
token_check: Option<Arc<Mutex<TokenChecker>>>,
|
||||||
sockets: Vec<Arc<UdpSocket>>,
|
sockets: Vec<Arc<UdpSocket>>,
|
||||||
queue: ::async_channel::Receiver<Work>,
|
queue: ::async_channel::Receiver<Work>,
|
||||||
|
queue_sender: ::async_channel::Sender<Work>,
|
||||||
) -> ::std::io::Result<Self> {
|
) -> ::std::io::Result<Self> {
|
||||||
let (queue_timeouts_send, queue_timeouts_recv) =
|
let (queue_timeouts_send, queue_timeouts_recv) =
|
||||||
mpsc::unbounded_channel();
|
mpsc::unbounded_channel();
|
||||||
|
@ -118,17 +146,93 @@ impl Worker {
|
||||||
token_check,
|
token_check,
|
||||||
sockets,
|
sockets,
|
||||||
queue,
|
queue,
|
||||||
|
queue_sender,
|
||||||
queue_timeouts_recv,
|
queue_timeouts_recv,
|
||||||
queue_timeouts_send,
|
queue_timeouts_send,
|
||||||
thread_channels: Vec::new(),
|
thread_channels: Vec::new(),
|
||||||
connections: ConnList::new(thread_id),
|
connections: ConnList::new(thread_id),
|
||||||
|
untracked_connections: VecDeque::with_capacity(8),
|
||||||
handshakes,
|
handshakes,
|
||||||
|
work_timers: super::Timers::new(),
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
/// return a handle to the worker that you can use to send data
|
||||||
|
/// The handle will enqueue work in the main worker and is thread-local safe
|
||||||
|
///
|
||||||
|
/// While this does not require `&mut` on the `Worker`, everything
|
||||||
|
/// will be put in the work queue,
|
||||||
|
/// So you might have less immediate results in a few cases
|
||||||
|
pub fn handle(&self) -> Handle {
|
||||||
|
Handle {
|
||||||
|
queue: self.queue_sender.clone(),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
/// change the UserTracker in the connection
|
||||||
|
///
|
||||||
|
/// This is `unsafe` because you will be responsible for manually updating
|
||||||
|
/// any copy of the `ConnTracker` you might have cloned around
|
||||||
|
#[allow(unsafe_code)]
|
||||||
|
pub unsafe fn set_connection_tracker(
|
||||||
|
&mut self,
|
||||||
|
tracker: ConnTracker,
|
||||||
|
new_id: connection::UserTracker,
|
||||||
|
) -> Result<ConnTracker, crate::Error> {
|
||||||
|
let conn = self.connections.get_mut(tracker.lib)?;
|
||||||
|
conn.user_tracker = Some(new_id);
|
||||||
|
Ok(ConnTracker {
|
||||||
|
lib: tracker.lib,
|
||||||
|
user: Some(new_id),
|
||||||
|
})
|
||||||
|
}
|
||||||
|
/// Enqueue data to send
|
||||||
|
pub fn send(
|
||||||
|
&mut self,
|
||||||
|
tracker: LibTracker,
|
||||||
|
stream: stream::ID,
|
||||||
|
data: Vec<u8>,
|
||||||
|
) -> Result<(), crate::Error> {
|
||||||
|
let conn = self.connections.get_mut(tracker)?;
|
||||||
|
conn.enqueue(stream, data)?;
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
|
/// Returns new connections, if any
|
||||||
|
///
|
||||||
|
/// You can provide an optional tracker, different from the library tracker.
|
||||||
|
///
|
||||||
|
/// Differently from the library tracker, you can change this later on,
|
||||||
|
/// but you will be responsible to change it on every `ConnTracker`
|
||||||
|
/// you might have cloned elsewhere
|
||||||
|
pub fn try_get_connection(
|
||||||
|
&mut self,
|
||||||
|
tracker: Option<connection::UserTracker>,
|
||||||
|
) -> Option<ConnTracker> {
|
||||||
|
let ret_tracker = ConnTracker {
|
||||||
|
lib: self.untracked_connections.pop_front()?,
|
||||||
|
user: None,
|
||||||
|
};
|
||||||
|
match tracker {
|
||||||
|
Some(tracker) => {
|
||||||
|
#[allow(unsafe_code)]
|
||||||
|
match unsafe {
|
||||||
|
self.set_connection_tracker(ret_tracker, tracker)
|
||||||
|
} {
|
||||||
|
Ok(tracker) => Some(tracker),
|
||||||
|
Err(_) => {
|
||||||
|
// we had a connection, but it expired before the user
|
||||||
|
// remembered to get it. Just remove it from the queue.
|
||||||
|
None
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
None => Some(ret_tracker),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
/// Continuously loop and process work as needed
|
/// Continuously loop and process work as needed
|
||||||
pub async fn work_loop(&mut self) {
|
pub async fn work_loop(&mut self) -> Result<Event, crate::Error> {
|
||||||
'mainloop: loop {
|
'mainloop: loop {
|
||||||
|
let next_timer = self.work_timers.get_next();
|
||||||
|
::tokio::pin!(next_timer);
|
||||||
let work = ::tokio::select! {
|
let work = ::tokio::select! {
|
||||||
tell_stopped = self.stop_working.recv() => {
|
tell_stopped = self.stop_working.recv() => {
|
||||||
if let Ok(stop_ch) = tell_stopped {
|
if let Ok(stop_ch) = tell_stopped {
|
||||||
|
@ -137,6 +241,13 @@ impl Worker {
|
||||||
}
|
}
|
||||||
break;
|
break;
|
||||||
}
|
}
|
||||||
|
() = &mut next_timer => {
|
||||||
|
let work_list = self.work_timers.get_work();
|
||||||
|
for w in work_list.into_iter() {
|
||||||
|
let _ = self.queue_sender.send(w).await;
|
||||||
|
}
|
||||||
|
continue 'mainloop;
|
||||||
|
}
|
||||||
maybe_timeout = self.queue.recv() => {
|
maybe_timeout = self.queue.recv() => {
|
||||||
match maybe_timeout {
|
match maybe_timeout {
|
||||||
Ok(work) => work,
|
Ok(work) => work,
|
||||||
|
@ -293,12 +404,14 @@ impl Worker {
|
||||||
// are PubKey::Exchange
|
// are PubKey::Exchange
|
||||||
unreachable!()
|
unreachable!()
|
||||||
}
|
}
|
||||||
let mut conn = Conn::new(
|
let mut conn = Connection::new(
|
||||||
hkdf,
|
hkdf,
|
||||||
cipher_selected,
|
cipher_selected,
|
||||||
connection::Role::Client,
|
connection::Role::Client,
|
||||||
&self.rand,
|
&self.rand,
|
||||||
);
|
);
|
||||||
|
let dest = UdpClient(addr.as_sockaddr().unwrap());
|
||||||
|
conn.send_addr = dest;
|
||||||
|
|
||||||
let auth_recv_id = self.connections.reserve_first();
|
let auth_recv_id = self.connections.reserve_first();
|
||||||
let service_conn_id = self.connections.reserve_first();
|
let service_conn_id = self.connections.reserve_first();
|
||||||
|
@ -389,15 +502,13 @@ impl Worker {
|
||||||
// send always from the first socket
|
// send always from the first socket
|
||||||
// FIXME: select based on routing table
|
// FIXME: select based on routing table
|
||||||
let sender = self.sockets[0].local_addr().unwrap();
|
let sender = self.sockets[0].local_addr().unwrap();
|
||||||
let dest = UdpClient(addr.as_sockaddr().unwrap());
|
|
||||||
|
|
||||||
// start the timeout right before sending the packet
|
// start the timeout right before sending the packet
|
||||||
hshake.timeout = Some(::tokio::task::spawn_local(
|
let time_drop = self.work_timers.add(
|
||||||
Self::handshake_timeout(
|
::tokio::time::Duration::from_secs(10),
|
||||||
self.queue_timeouts_send.clone(),
|
Work::DropHandshake(client_key_id),
|
||||||
client_key_id,
|
);
|
||||||
),
|
hshake.timeout = Some(time_drop);
|
||||||
));
|
|
||||||
|
|
||||||
// send packet
|
// send packet
|
||||||
self.send_packet(raw, dest, UdpServer(sender)).await;
|
self.send_packet(raw, dest, UdpServer(sender)).await;
|
||||||
|
@ -413,21 +524,70 @@ impl Worker {
|
||||||
}
|
}
|
||||||
};
|
};
|
||||||
}
|
}
|
||||||
Work::Recv(pkt) => {
|
Work::Recv(pkt) => match self.recv(pkt).await {
|
||||||
self.recv(pkt).await;
|
Ok(event) => return Ok(event),
|
||||||
|
Err(_) => continue 'mainloop,
|
||||||
|
},
|
||||||
|
Work::UserSend((tracker, stream, data)) => {
|
||||||
|
let conn = match self.connections.get_mut(tracker) {
|
||||||
|
Ok(conn) => conn,
|
||||||
|
Err(_) => continue 'mainloop,
|
||||||
|
};
|
||||||
|
use connection::Enqueue;
|
||||||
|
if let Ok(enqueued) = conn.enqueue(stream, data) {
|
||||||
|
match enqueued {
|
||||||
|
Enqueue::Immediate(instant) => {
|
||||||
|
let _ = self
|
||||||
|
.queue_sender
|
||||||
|
.send(Work::SendData((tracker, instant)))
|
||||||
|
.await;
|
||||||
|
}
|
||||||
|
Enqueue::TimerWait => {}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
Work::SendData((tracker, instant)) => {
|
||||||
|
// make sure we don't process events before they are
|
||||||
|
// actually needed.
|
||||||
|
// This is basically busy waiting with extra steps,
|
||||||
|
// but we don't want to spawn lots of timers and
|
||||||
|
// we don't really have a fine-grained sleep that is
|
||||||
|
// multiplatform
|
||||||
|
let now = ::tokio::time::Instant::now();
|
||||||
|
if instant <= now {
|
||||||
|
let _ = self
|
||||||
|
.queue_sender
|
||||||
|
.send(Work::SendData((tracker, instant)))
|
||||||
|
.await;
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
|
||||||
|
let mut raw: Vec<u8> = Vec::with_capacity(1200);
|
||||||
|
raw.resize(raw.capacity(), 0);
|
||||||
|
let conn = match self.connections.get_mut(tracker) {
|
||||||
|
Ok(conn) => conn,
|
||||||
|
Err(_) => continue,
|
||||||
|
};
|
||||||
|
let pkt = match conn.write_pkt(&mut raw) {
|
||||||
|
Ok(pkt) => pkt,
|
||||||
|
Err(enc::Error::NotEnoughData(0)) => continue,
|
||||||
|
Err(e) => {
|
||||||
|
::tracing::error!("Packet generation: {:?}", e);
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
};
|
||||||
|
let dest = conn.send_addr;
|
||||||
|
let src = UdpServer(self.sockets[0].local_addr().unwrap());
|
||||||
|
let len = pkt.len();
|
||||||
|
raw.truncate(len);
|
||||||
|
let _ = self.send_packet(raw, dest, src);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
Ok(Event::End)
|
||||||
async fn handshake_timeout(
|
|
||||||
timeout_queue: mpsc::UnboundedSender<Work>,
|
|
||||||
key_id: KeyID,
|
|
||||||
) {
|
|
||||||
::tokio::time::sleep(::std::time::Duration::from_secs(10)).await;
|
|
||||||
let _ = timeout_queue.send(Work::DropHandshake(key_id));
|
|
||||||
}
|
}
|
||||||
/// Read and do stuff with the raw udp packet
|
/// Read and do stuff with the raw udp packet
|
||||||
async fn recv(&mut self, mut udp: RawUdp) {
|
async fn recv(&mut self, mut udp: RawUdp) -> Result<Event, ()> {
|
||||||
if udp.packet.id.is_handshake() {
|
if udp.packet.id.is_handshake() {
|
||||||
let handshake = match Handshake::deserialize(
|
let handshake = match Handshake::deserialize(
|
||||||
&udp.data[connection::ID::len()..],
|
&udp.data[connection::ID::len()..],
|
||||||
|
@ -435,7 +595,7 @@ impl Worker {
|
||||||
Ok(handshake) => handshake,
|
Ok(handshake) => handshake,
|
||||||
Err(e) => {
|
Err(e) => {
|
||||||
::tracing::debug!("Handshake parsing: {}", e);
|
::tracing::debug!("Handshake parsing: {}", e);
|
||||||
return;
|
return Err(());
|
||||||
}
|
}
|
||||||
};
|
};
|
||||||
let action = match self.handshakes.recv_handshake(
|
let action = match self.handshakes.recv_handshake(
|
||||||
|
@ -445,190 +605,255 @@ impl Worker {
|
||||||
Ok(action) => action,
|
Ok(action) => action,
|
||||||
Err(err) => {
|
Err(err) => {
|
||||||
::tracing::debug!("Handshake recv error {}", err);
|
::tracing::debug!("Handshake recv error {}", err);
|
||||||
return;
|
return Err(());
|
||||||
}
|
}
|
||||||
};
|
};
|
||||||
match action {
|
self.recv_handshake(udp, action).await;
|
||||||
handshake::Action::AuthNeeded(authinfo) => {
|
Err(())
|
||||||
let req;
|
} else {
|
||||||
if let handshake::Data::DirSync(DirSync::Req(r)) =
|
self.recv_packet(udp)
|
||||||
authinfo.handshake.data
|
}
|
||||||
{
|
}
|
||||||
req = r;
|
/// Receive a non-handshake packet
|
||||||
} else {
|
fn recv_packet(&mut self, udp: RawUdp) -> Result<Event, ()> {
|
||||||
::tracing::error!("AuthInfo on non DS::Req");
|
let conn = match self.connections.get_id_mut(udp.packet.id) {
|
||||||
return;
|
Ok(conn) => conn,
|
||||||
}
|
Err(_) => return Err(()),
|
||||||
let req_data = match req.data {
|
};
|
||||||
dirsync::req::State::ClearText(req_data) => req_data,
|
match conn.recv(udp) {
|
||||||
_ => {
|
Ok(stream::StreamData::NotReady) => Err(()),
|
||||||
::tracing::error!("AuthNeeded: expected ClearText");
|
Ok(stream::StreamData::Ready) => Ok(Event::Data(ConnData {
|
||||||
assert!(false, "AuthNeeded: unreachable");
|
conn: ConnTracker {
|
||||||
return;
|
user: conn.user_tracker,
|
||||||
}
|
lib: conn.lib_tracker,
|
||||||
};
|
},
|
||||||
// FIXME: This part can take a while,
|
data: conn.get_data().unwrap(),
|
||||||
// we should just spawn it probably
|
})),
|
||||||
let maybe_auth_check = {
|
Err(e) => {
|
||||||
match &self.token_check {
|
::tracing::trace!("Conn Recv: {:?}", e.to_string());
|
||||||
None => {
|
Err(())
|
||||||
if req_data.auth.user == auth::USERID_ANONYMOUS
|
}
|
||||||
{
|
}
|
||||||
Ok(true)
|
}
|
||||||
} else {
|
/// Receive an handshake packet
|
||||||
Ok(false)
|
async fn recv_handshake(&mut self, udp: RawUdp, action: handshake::Action) {
|
||||||
}
|
match action {
|
||||||
}
|
handshake::Action::AuthNeeded(authinfo) => {
|
||||||
Some(token_check) => {
|
let req;
|
||||||
let tk_check = token_check.lock().await;
|
if let handshake::Data::DirSync(DirSync::Req(r)) =
|
||||||
tk_check(
|
authinfo.handshake.data
|
||||||
req_data.auth.user,
|
{
|
||||||
req_data.auth.token,
|
req = r;
|
||||||
req_data.auth.service_id,
|
} else {
|
||||||
req_data.auth.domain,
|
::tracing::error!("AuthInfo on non DS::Req");
|
||||||
)
|
return;
|
||||||
.await
|
|
||||||
}
|
|
||||||
}
|
|
||||||
};
|
|
||||||
let is_authenticated = match maybe_auth_check {
|
|
||||||
Ok(is_authenticated) => is_authenticated,
|
|
||||||
Err(_) => {
|
|
||||||
::tracing::error!("error in token auth");
|
|
||||||
// TODO: retry?
|
|
||||||
return;
|
|
||||||
}
|
|
||||||
};
|
|
||||||
if !is_authenticated {
|
|
||||||
::tracing::warn!(
|
|
||||||
"Wrong authentication for user {:?}",
|
|
||||||
req_data.auth.user
|
|
||||||
);
|
|
||||||
// TODO: error response
|
|
||||||
return;
|
|
||||||
}
|
|
||||||
// Client has correctly authenticated
|
|
||||||
// TODO: contact the service, get the key and
|
|
||||||
// connection ID
|
|
||||||
let srv_conn_id = connection::ID::new_rand(&self.rand);
|
|
||||||
let srv_secret = Secret::new_rand(&self.rand);
|
|
||||||
let head_len = req.cipher.nonce_len();
|
|
||||||
let tag_len = req.cipher.tag_len();
|
|
||||||
|
|
||||||
let mut auth_conn = Conn::new(
|
|
||||||
authinfo.hkdf,
|
|
||||||
req.cipher,
|
|
||||||
connection::Role::Server,
|
|
||||||
&self.rand,
|
|
||||||
);
|
|
||||||
auth_conn.id_send = IDSend(req_data.id);
|
|
||||||
// track connection
|
|
||||||
let auth_id_recv = self.connections.reserve_first();
|
|
||||||
auth_conn.id_recv = auth_id_recv;
|
|
||||||
|
|
||||||
let resp_data = dirsync::resp::Data {
|
|
||||||
client_nonce: req_data.nonce,
|
|
||||||
id: auth_conn.id_recv.0,
|
|
||||||
service_connection_id: srv_conn_id,
|
|
||||||
service_key: srv_secret,
|
|
||||||
};
|
|
||||||
use crate::enc::sym::AAD;
|
|
||||||
// no aad for now
|
|
||||||
let aad = AAD(&mut []);
|
|
||||||
|
|
||||||
let resp = dirsync::resp::Resp {
|
|
||||||
client_key_id: req_data.client_key_id,
|
|
||||||
data: dirsync::resp::State::ClearText(resp_data),
|
|
||||||
};
|
|
||||||
let encrypt_from =
|
|
||||||
connection::ID::len() + resp.encrypted_offset();
|
|
||||||
let encrypt_until =
|
|
||||||
encrypt_from + resp.encrypted_length(head_len, tag_len);
|
|
||||||
let resp_handshake = Handshake::new(
|
|
||||||
handshake::Data::DirSync(DirSync::Resp(resp)),
|
|
||||||
);
|
|
||||||
let packet = Packet {
|
|
||||||
id: connection::ID::new_handshake(),
|
|
||||||
data: packet::Data::Handshake(resp_handshake),
|
|
||||||
};
|
|
||||||
let tot_len = packet.len(head_len, tag_len);
|
|
||||||
let mut raw_out = Vec::<u8>::with_capacity(tot_len);
|
|
||||||
raw_out.resize(tot_len, 0);
|
|
||||||
packet.serialize(head_len, tag_len, &mut raw_out);
|
|
||||||
|
|
||||||
if let Err(e) = auth_conn
|
|
||||||
.cipher_send
|
|
||||||
.encrypt(aad, &mut raw_out[encrypt_from..encrypt_until])
|
|
||||||
{
|
|
||||||
::tracing::error!("can't encrypt: {:?}", e);
|
|
||||||
return;
|
|
||||||
}
|
|
||||||
self.send_packet(raw_out, udp.src, udp.dst).await;
|
|
||||||
}
|
}
|
||||||
handshake::Action::ClientConnect(cci) => {
|
let req_data = match req.data {
|
||||||
let ds_resp;
|
dirsync::req::State::ClearText(req_data) => req_data,
|
||||||
if let handshake::Data::DirSync(DirSync::Resp(resp)) =
|
_ => {
|
||||||
cci.handshake.data
|
::tracing::error!("AuthNeeded: expected ClearText");
|
||||||
{
|
assert!(false, "AuthNeeded: unreachable");
|
||||||
ds_resp = resp;
|
|
||||||
} else {
|
|
||||||
::tracing::error!("ClientConnect on non DS::Resp");
|
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
// track connection
|
};
|
||||||
let resp_data;
|
// FIXME: This part can take a while,
|
||||||
if let dirsync::resp::State::ClearText(r_data) =
|
// we should just spawn it probably
|
||||||
ds_resp.data
|
let maybe_auth_check = {
|
||||||
{
|
match &self.token_check {
|
||||||
resp_data = r_data;
|
None => {
|
||||||
} else {
|
if req_data.auth.user == auth::USERID_ANONYMOUS {
|
||||||
::tracing::error!(
|
Ok(true)
|
||||||
"ClientConnect on non DS::Resp::ClearText"
|
} else {
|
||||||
);
|
Ok(false)
|
||||||
unreachable!();
|
}
|
||||||
|
}
|
||||||
|
Some(token_check) => {
|
||||||
|
let tk_check = token_check.lock().await;
|
||||||
|
tk_check(
|
||||||
|
req_data.auth.user,
|
||||||
|
req_data.auth.token,
|
||||||
|
req_data.auth.service_id,
|
||||||
|
req_data.auth.domain,
|
||||||
|
)
|
||||||
|
.await
|
||||||
|
}
|
||||||
}
|
}
|
||||||
let auth_srv_conn = IDSend(resp_data.id);
|
};
|
||||||
let mut conn = cci.connection;
|
let is_authenticated = match maybe_auth_check {
|
||||||
conn.id_send = auth_srv_conn;
|
Ok(is_authenticated) => is_authenticated,
|
||||||
let id_recv = conn.id_recv;
|
Err(_) => {
|
||||||
let cipher = conn.cipher_recv.kind();
|
::tracing::error!("error in token auth");
|
||||||
// track the connection to the authentication server
|
// TODO: retry?
|
||||||
if self.connections.track(conn.into()).is_err() {
|
return;
|
||||||
::tracing::error!("Could not track new connection");
|
}
|
||||||
|
};
|
||||||
|
if !is_authenticated {
|
||||||
|
::tracing::warn!(
|
||||||
|
"Wrong authentication for user {:?}",
|
||||||
|
req_data.auth.user
|
||||||
|
);
|
||||||
|
// TODO: error response
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
// Client has correctly authenticated
|
||||||
|
// TODO: contact the service, get the key and
|
||||||
|
// connection ID
|
||||||
|
let srv_conn_id = connection::ID::new_rand(&self.rand);
|
||||||
|
let srv_secret = Secret::new_rand(&self.rand);
|
||||||
|
let head_len = req.cipher.nonce_len();
|
||||||
|
let tag_len = req.cipher.tag_len();
|
||||||
|
|
||||||
|
let mut auth_conn = Connection::new(
|
||||||
|
authinfo.hkdf,
|
||||||
|
req.cipher,
|
||||||
|
connection::Role::Server,
|
||||||
|
&self.rand,
|
||||||
|
);
|
||||||
|
auth_conn.id_send = IDSend(req_data.id);
|
||||||
|
auth_conn.send_addr = udp.src;
|
||||||
|
// track connection
|
||||||
|
let auth_id_recv = self.connections.reserve_first();
|
||||||
|
auth_conn.id_recv = auth_id_recv;
|
||||||
|
let (tracker, auth_conn) =
|
||||||
|
self.connections.reserve_and_track(auth_conn);
|
||||||
|
self.untracked_connections.push_back(tracker);
|
||||||
|
|
||||||
|
let resp_data = dirsync::resp::Data {
|
||||||
|
client_nonce: req_data.nonce,
|
||||||
|
id: auth_conn.id_recv.0,
|
||||||
|
service_connection_id: srv_conn_id,
|
||||||
|
service_key: srv_secret,
|
||||||
|
};
|
||||||
|
use crate::enc::sym::AAD;
|
||||||
|
// no aad for now
|
||||||
|
let aad = AAD(&mut []);
|
||||||
|
|
||||||
|
let resp = dirsync::resp::Resp {
|
||||||
|
client_key_id: req_data.client_key_id,
|
||||||
|
data: dirsync::resp::State::ClearText(resp_data),
|
||||||
|
};
|
||||||
|
let encrypt_from =
|
||||||
|
connection::ID::len() + resp.encrypted_offset();
|
||||||
|
let encrypt_until =
|
||||||
|
encrypt_from + resp.encrypted_length(head_len, tag_len);
|
||||||
|
let resp_handshake = Handshake::new(handshake::Data::DirSync(
|
||||||
|
DirSync::Resp(resp),
|
||||||
|
));
|
||||||
|
let packet = Packet {
|
||||||
|
id: connection::ID::new_handshake(),
|
||||||
|
data: packet::Data::Handshake(resp_handshake),
|
||||||
|
};
|
||||||
|
let tot_len = packet.len(head_len, tag_len);
|
||||||
|
let mut raw_out = Vec::<u8>::with_capacity(tot_len);
|
||||||
|
raw_out.resize(tot_len, 0);
|
||||||
|
packet.serialize(head_len, tag_len, &mut raw_out);
|
||||||
|
|
||||||
|
if let Err(e) = auth_conn
|
||||||
|
.cipher_send
|
||||||
|
.encrypt(aad, &mut raw_out[encrypt_from..encrypt_until])
|
||||||
|
{
|
||||||
|
::tracing::error!("can't encrypt: {:?}", e);
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
self.send_packet(raw_out, udp.src, udp.dst).await;
|
||||||
|
}
|
||||||
|
handshake::Action::ClientConnect(cci) => {
|
||||||
|
self.work_timers.remove(cci.old_timeout);
|
||||||
|
let ds_resp;
|
||||||
|
if let handshake::Data::DirSync(DirSync::Resp(resp)) =
|
||||||
|
cci.handshake.data
|
||||||
|
{
|
||||||
|
ds_resp = resp;
|
||||||
|
} else {
|
||||||
|
::tracing::error!("ClientConnect on non DS::Resp");
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
// track connection
|
||||||
|
let resp_data;
|
||||||
|
if let dirsync::resp::State::ClearText(r_data) = ds_resp.data {
|
||||||
|
resp_data = r_data;
|
||||||
|
} else {
|
||||||
|
::tracing::error!(
|
||||||
|
"ClientConnect on non DS::Resp::ClearText"
|
||||||
|
);
|
||||||
|
unreachable!();
|
||||||
|
}
|
||||||
|
let auth_id_send = IDSend(resp_data.id);
|
||||||
|
let mut conn = cci.connection;
|
||||||
|
conn.id_send = auth_id_send;
|
||||||
|
let id_recv = conn.id_recv;
|
||||||
|
let cipher = conn.cipher_recv.kind();
|
||||||
|
// track the connection to the authentication server
|
||||||
|
let track_auth_conn = match self.connections.track(conn) {
|
||||||
|
Ok(track_auth_conn) => track_auth_conn,
|
||||||
|
Err(_) => {
|
||||||
|
::tracing::error!(
|
||||||
|
"Could not track new auth srv connection"
|
||||||
|
);
|
||||||
self.connections.remove(id_recv);
|
self.connections.remove(id_recv);
|
||||||
|
// FIXME: proper connection closing
|
||||||
let _ = cci.answer.send(Err(
|
let _ = cci.answer.send(Err(
|
||||||
handshake::Error::InternalTracking.into(),
|
handshake::Error::InternalTracking.into(),
|
||||||
));
|
));
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
if cci.service_id != auth::SERVICEID_AUTH {
|
};
|
||||||
// create and track the connection to the service
|
let authsrv_conn = AuthSrvConn(ConnTracker {
|
||||||
// SECURITY: xor with secrets
|
lib: track_auth_conn,
|
||||||
//FIXME: the Secret should be XORed with the client
|
user: None,
|
||||||
// stored secret (if any)
|
});
|
||||||
let hkdf = Hkdf::new(
|
let mut service_conn = None;
|
||||||
hkdf::Kind::Sha3,
|
if cci.service_id != auth::SERVICEID_AUTH {
|
||||||
cci.service_id.as_bytes(),
|
// create and track the connection to the service
|
||||||
resp_data.service_key,
|
// SECURITY: xor with secrets
|
||||||
);
|
//FIXME: the Secret should be XORed with the client
|
||||||
let mut service_connection = Conn::new(
|
// stored secret (if any)
|
||||||
hkdf,
|
let hkdf = Hkdf::new(
|
||||||
cipher,
|
hkdf::Kind::Sha3,
|
||||||
connection::Role::Client,
|
cci.service_id.as_bytes(),
|
||||||
&self.rand,
|
resp_data.service_key,
|
||||||
);
|
);
|
||||||
service_connection.id_recv = cci.service_connection_id;
|
let mut service_connection = Connection::new(
|
||||||
service_connection.id_send =
|
hkdf,
|
||||||
IDSend(resp_data.service_connection_id);
|
cipher,
|
||||||
let _ =
|
connection::Role::Client,
|
||||||
self.connections.track(service_connection.into());
|
&self.rand,
|
||||||
}
|
);
|
||||||
let _ =
|
service_connection.id_recv = cci.service_connection_id;
|
||||||
cci.answer.send(Ok((cci.srv_key_id, auth_srv_conn)));
|
service_connection.id_send =
|
||||||
|
IDSend(resp_data.service_connection_id);
|
||||||
|
let track_serv_conn =
|
||||||
|
match self.connections.track(service_connection) {
|
||||||
|
Ok(track_serv_conn) => track_serv_conn,
|
||||||
|
Err(_) => {
|
||||||
|
::tracing::error!(
|
||||||
|
"Could not track new service connection"
|
||||||
|
);
|
||||||
|
self.connections
|
||||||
|
.remove(cci.service_connection_id);
|
||||||
|
// FIXME: proper connection closing
|
||||||
|
// FIXME: drop auth srv connection if we just
|
||||||
|
// established it
|
||||||
|
let _ = cci.answer.send(Err(
|
||||||
|
handshake::Error::InternalTracking.into(),
|
||||||
|
));
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
};
|
||||||
|
service_conn = Some(ServiceConn(ConnTracker {
|
||||||
|
lib: track_serv_conn,
|
||||||
|
user: None,
|
||||||
|
}));
|
||||||
}
|
}
|
||||||
handshake::Action::Nothing => {}
|
let _ = cci.answer.send(Ok(handshake::tracker::ConnectOk {
|
||||||
};
|
auth_key_id: cci.srv_key_id,
|
||||||
}
|
auth_id_send,
|
||||||
|
authsrv_conn,
|
||||||
|
service_conn,
|
||||||
|
}));
|
||||||
|
}
|
||||||
|
handshake::Action::Nothing => {}
|
||||||
|
};
|
||||||
}
|
}
|
||||||
async fn send_packet(
|
async fn send_packet(
|
||||||
&self,
|
&self,
|
||||||
|
@ -653,3 +878,16 @@ impl Worker {
|
||||||
let _res = src_sock.send_to(&data, client.0).await;
|
let _res = src_sock.send_to(&data, client.0).await;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Handle to send work asyncronously to the worker
|
||||||
|
#[derive(Debug, Clone)]
|
||||||
|
pub struct Handle {
|
||||||
|
queue: ::async_channel::Sender<Work>,
|
||||||
|
}
|
||||||
|
|
||||||
|
impl Handle {
|
||||||
|
// TODO
|
||||||
|
// pub fn send(..)
|
||||||
|
// pub fn set_connection_id(..)
|
||||||
|
// try_get_new_connections()
|
||||||
|
}
|
||||||
|
|
78
src/lib.rs
78
src/lib.rs
|
@ -34,12 +34,12 @@ use crate::{
|
||||||
AuthServerConnections, Packet,
|
AuthServerConnections, Packet,
|
||||||
},
|
},
|
||||||
inner::{
|
inner::{
|
||||||
worker::{ConnectInfo, RawUdp, Work, Worker},
|
worker::{ConnectInfo, Event, RawUdp, Work, Worker},
|
||||||
ThreadTracker,
|
ThreadTracker,
|
||||||
},
|
},
|
||||||
};
|
};
|
||||||
pub use config::Config;
|
pub use config::Config;
|
||||||
pub use connection::Connection;
|
pub use connection::{AuthSrvConn, ServiceConn};
|
||||||
|
|
||||||
/// Main fenrir library errors
|
/// Main fenrir library errors
|
||||||
#[derive(::thiserror::Error, Debug)]
|
#[derive(::thiserror::Error, Debug)]
|
||||||
|
@ -59,15 +59,15 @@ pub enum Error {
|
||||||
/// Handshake errors
|
/// Handshake errors
|
||||||
#[error("Handshake: {0:?}")]
|
#[error("Handshake: {0:?}")]
|
||||||
Handshake(#[from] handshake::Error),
|
Handshake(#[from] handshake::Error),
|
||||||
/// Key error
|
|
||||||
#[error("key: {0:?}")]
|
|
||||||
Key(#[from] crate::enc::Error),
|
|
||||||
/// Resolution problems. wrong or incomplete DNSSEC data
|
/// Resolution problems. wrong or incomplete DNSSEC data
|
||||||
#[error("DNSSEC resolution: {0}")]
|
#[error("DNSSEC resolution: {0}")]
|
||||||
Resolution(String),
|
Resolution(String),
|
||||||
/// Wrapper on encryption errors
|
/// Wrapper on encryption errors
|
||||||
#[error("Encrypt: {0}")]
|
#[error("Crypto: {0}")]
|
||||||
Encrypt(enc::Error),
|
Crypto(#[from] enc::Error),
|
||||||
|
/// Wrapper on connection errors
|
||||||
|
#[error("Connection: {0}")]
|
||||||
|
Connection(#[from] connection::Error),
|
||||||
}
|
}
|
||||||
|
|
||||||
pub(crate) enum StopWorking {
|
pub(crate) enum StopWorking {
|
||||||
|
@ -176,6 +176,7 @@ impl Fenrir {
|
||||||
config: &Config,
|
config: &Config,
|
||||||
tokio_rt: Arc<::tokio::runtime::Runtime>,
|
tokio_rt: Arc<::tokio::runtime::Runtime>,
|
||||||
) -> Result<Self, Error> {
|
) -> Result<Self, Error> {
|
||||||
|
inner::set_minimum_sleep_resolution().await;
|
||||||
let (sender, _) = ::tokio::sync::broadcast::channel(1);
|
let (sender, _) = ::tokio::sync::broadcast::channel(1);
|
||||||
let dnssec = dnssec::Dnssec::new(&config.resolvers)?;
|
let dnssec = dnssec::Dnssec::new(&config.resolvers)?;
|
||||||
// bind sockets early so we can change "port 0" (aka: random)
|
// bind sockets early so we can change "port 0" (aka: random)
|
||||||
|
@ -214,6 +215,7 @@ impl Fenrir {
|
||||||
pub async fn with_workers(
|
pub async fn with_workers(
|
||||||
config: &Config,
|
config: &Config,
|
||||||
) -> Result<(Self, Vec<Worker>), Error> {
|
) -> Result<(Self, Vec<Worker>), Error> {
|
||||||
|
inner::set_minimum_sleep_resolution().await;
|
||||||
let (stop_working, _) = ::tokio::sync::broadcast::channel(1);
|
let (stop_working, _) = ::tokio::sync::broadcast::channel(1);
|
||||||
let dnssec = dnssec::Dnssec::new(&config.resolvers)?;
|
let dnssec = dnssec::Dnssec::new(&config.resolvers)?;
|
||||||
// bind sockets early so we can change "port 0" (aka: random)
|
// bind sockets early so we can change "port 0" (aka: random)
|
||||||
|
@ -382,7 +384,7 @@ impl Fenrir {
|
||||||
&self,
|
&self,
|
||||||
domain: &Domain,
|
domain: &Domain,
|
||||||
service: ServiceID,
|
service: ServiceID,
|
||||||
) -> Result<(), Error> {
|
) -> Result<(AuthSrvConn, Option<ServiceConn>), Error> {
|
||||||
let resolved = self.resolv(domain).await?;
|
let resolved = self.resolv(domain).await?;
|
||||||
self.connect_resolved(resolved, domain, service).await
|
self.connect_resolved(resolved, domain, service).await
|
||||||
}
|
}
|
||||||
|
@ -392,7 +394,7 @@ impl Fenrir {
|
||||||
resolved: dnssec::Record,
|
resolved: dnssec::Record,
|
||||||
domain: &Domain,
|
domain: &Domain,
|
||||||
service: ServiceID,
|
service: ServiceID,
|
||||||
) -> Result<(), Error> {
|
) -> Result<(AuthSrvConn, Option<ServiceConn>), Error> {
|
||||||
loop {
|
loop {
|
||||||
// check if we already have a connection to that auth. srv
|
// check if we already have a connection to that auth. srv
|
||||||
let is_reserved = {
|
let is_reserved = {
|
||||||
|
@ -460,29 +462,28 @@ impl Fenrir {
|
||||||
.await;
|
.await;
|
||||||
|
|
||||||
match recv.await {
|
match recv.await {
|
||||||
Ok(res) => {
|
Ok(res) => match res {
|
||||||
match res {
|
Err(e) => {
|
||||||
Err(e) => {
|
let mut conn_auth_lock = self.conn_auth_srv.lock().await;
|
||||||
let mut conn_auth_lock =
|
conn_auth_lock.remove_reserved(&resolved);
|
||||||
self.conn_auth_srv.lock().await;
|
Err(e)
|
||||||
conn_auth_lock.remove_reserved(&resolved);
|
|
||||||
Err(e)
|
|
||||||
}
|
|
||||||
Ok((key_id, id_send)) => {
|
|
||||||
let key = resolved
|
|
||||||
.public_keys
|
|
||||||
.iter()
|
|
||||||
.find(|k| k.0 == key_id)
|
|
||||||
.unwrap();
|
|
||||||
let mut conn_auth_lock =
|
|
||||||
self.conn_auth_srv.lock().await;
|
|
||||||
conn_auth_lock.add(&key.1, id_send, &resolved);
|
|
||||||
|
|
||||||
//FIXME: user needs to somehow track the connection
|
|
||||||
Ok(())
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
}
|
Ok(connections) => {
|
||||||
|
let key = resolved
|
||||||
|
.public_keys
|
||||||
|
.iter()
|
||||||
|
.find(|k| k.0 == connections.auth_key_id)
|
||||||
|
.unwrap();
|
||||||
|
let mut conn_auth_lock = self.conn_auth_srv.lock().await;
|
||||||
|
conn_auth_lock.add(
|
||||||
|
&key.1,
|
||||||
|
connections.auth_id_send,
|
||||||
|
&resolved,
|
||||||
|
);
|
||||||
|
|
||||||
|
Ok((connections.authsrv_conn, connections.service_conn))
|
||||||
|
}
|
||||||
|
},
|
||||||
Err(e) => {
|
Err(e) => {
|
||||||
// Thread dropped the sender. no more thread?
|
// Thread dropped the sender. no more thread?
|
||||||
let mut conn_auth_lock = self.conn_auth_srv.lock().await;
|
let mut conn_auth_lock = self.conn_auth_srv.lock().await;
|
||||||
|
@ -524,6 +525,7 @@ impl Fenrir {
|
||||||
self.token_check.clone(),
|
self.token_check.clone(),
|
||||||
socks,
|
socks,
|
||||||
work_recv,
|
work_recv,
|
||||||
|
work_send.clone(),
|
||||||
)
|
)
|
||||||
.await?;
|
.await?;
|
||||||
// don't keep around private keys too much
|
// don't keep around private keys too much
|
||||||
|
@ -547,7 +549,6 @@ impl Fenrir {
|
||||||
}
|
}
|
||||||
Ok(worker)
|
Ok(worker)
|
||||||
}
|
}
|
||||||
|
|
||||||
// needs to be called before add_sockets
|
// needs to be called before add_sockets
|
||||||
/// Start one working thread for each physical cpu
|
/// Start one working thread for each physical cpu
|
||||||
/// threads are pinned to each cpu core.
|
/// threads are pinned to each cpu core.
|
||||||
|
@ -589,6 +590,7 @@ impl Fenrir {
|
||||||
let th_tokio_rt = tokio_rt.clone();
|
let th_tokio_rt = tokio_rt.clone();
|
||||||
let th_config = self.cfg.clone();
|
let th_config = self.cfg.clone();
|
||||||
let (work_send, work_recv) = ::async_channel::unbounded::<Work>();
|
let (work_send, work_recv) = ::async_channel::unbounded::<Work>();
|
||||||
|
let th_work_send = work_send.clone();
|
||||||
let th_stop_working = self.stop_working.subscribe();
|
let th_stop_working = self.stop_working.subscribe();
|
||||||
let th_token_check = self.token_check.clone();
|
let th_token_check = self.token_check.clone();
|
||||||
let th_sockets = sockets.clone();
|
let th_sockets = sockets.clone();
|
||||||
|
@ -629,13 +631,23 @@ impl Fenrir {
|
||||||
th_token_check,
|
th_token_check,
|
||||||
th_sockets,
|
th_sockets,
|
||||||
work_recv,
|
work_recv,
|
||||||
|
th_work_send,
|
||||||
)
|
)
|
||||||
.await
|
.await
|
||||||
{
|
{
|
||||||
Ok(worker) => worker,
|
Ok(worker) => worker,
|
||||||
Err(_) => return,
|
Err(_) => return,
|
||||||
};
|
};
|
||||||
worker.work_loop().await
|
loop {
|
||||||
|
match worker.work_loop().await {
|
||||||
|
Ok(_) => continue,
|
||||||
|
Ok(Event::End) => break,
|
||||||
|
Err(e) => {
|
||||||
|
::tracing::error!("Worker: {:?}", e);
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
});
|
});
|
||||||
});
|
});
|
||||||
loop {
|
loop {
|
||||||
|
|
|
@ -88,7 +88,7 @@ async fn test_connection_dirsync() {
|
||||||
.connect_resolved(dnssec_record, &test_domain, auth::SERVICEID_AUTH)
|
.connect_resolved(dnssec_record, &test_domain, auth::SERVICEID_AUTH)
|
||||||
.await
|
.await
|
||||||
{
|
{
|
||||||
Ok(()) => {}
|
Ok((_, _)) => {}
|
||||||
Err(e) => {
|
Err(e) => {
|
||||||
assert!(false, "Err on client connection: {:?}", e);
|
assert!(false, "Err on client connection: {:?}", e);
|
||||||
}
|
}
|
||||||
|
|
Loading…
Reference in New Issue