void/ZeroTierOne
2
0
Fork 0
mirror of https://github.com/zerotier/ZeroTierOne.git synced 2025-06-07 04:53:44 +02:00
Code Issues Projects Releases Packages Wiki Activity Actions

A bunch of reorg and simplification in the network hypervisor.

Browse source
This commit is contained in:
Adam Ierymenko 2022-09-13 10:35:52 -04:00
parent 1f9819e126
commit 32abd20c38
No known key found for this signature in database
GPG key ID: C8877CF2D7A5D7F3
44 changed files with 470 additions and 593 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
Cargo.toml
View file

@ -1,7 +1,7 @@
[workspace]
members = [
"core-crypto",
"crypto",
"network-hypervisor",
"controller",
"system-service",

0
network-hypervisor/src/vl1/careof.rs → attic/careof.rs
View file

3
controller/Cargo.toml
View file

@ -8,5 +8,6 @@ name = "zerotier-controller"
path = "src/main.rs"
[dependencies]
zerotier-core-crypto = { path = "../core-crypto" }
zerotier-crypto = { path = "../crypto" }
zerotier-utils = { path = "../utils" }
zerotier-network-hypervisor = { path = "../network-hypervisor" }

9
core-crypto/README.md
View file

@ -1,9 +0,0 @@
# ZeroTier Core Cryptography Library
------
This is mostly just glue to provide a simple consistent API in front of OpenSSL and some platform-specific crypto APIs.
It's thin and simple enough that we can easily create variants of it in the future for e.g. if we need to support some proprietary FIPS module or something.
It also contains a few utilities and helper functions.

2
core-crypto/Cargo.toml → crypto/Cargo.toml
View file

@ -2,7 +2,7 @@
authors = ["ZeroTier, Inc. <contact@zerotier.com>", "Adam Ierymenko <adam.ierymenko@zerotier.com>"]
edition = "2021"
license = "MPL-2.0"
name = "zerotier-core-crypto"
name = "zerotier-crypto"
version = "0.1.0"
[dependencies]

7
crypto/README.md Normal file
View file

@ -0,0 +1,7 @@
# ZeroTier Cryptography Library
------
Most of this library is just glue to provide a simple safe API around things like OpenSSL or OS-specific crypto APIs.
It also contains ZSSP, the V2 ZeroTier Secure Session Protocol.

14
core-crypto/benches/benchmark_crypto.rs → crypto/benches/benchmark_crypto.rs
View file

@ -1,9 +1,9 @@
use criterion::{criterion_group, criterion_main, Criterion};
use std::time::Duration;
use zerotier_core_crypto::p384::*;
use zerotier_core_crypto::random;
use zerotier_core_crypto::x25519::*;
use zerotier_crypto::p384::*;
use zerotier_crypto::random;
use zerotier_crypto::x25519::*;
pub fn criterion_benchmark(c: &mut Criterion) {
let p384_a = P384KeyPair::generate();
@ -21,8 +21,12 @@ pub fn criterion_benchmark(c: &mut Criterion) {
group.bench_function("ecdhp384", |b| b.iter(|| p384_a.agree(p384_b.public_key()).expect("ecdhp384 failed")));
group.bench_function("ecdhx25519", |b| b.iter(|| x25519_a.agree(&x25519_b_pub)));
group.bench_function("kyber_encapsulate", |b| b.iter(|| pqc_kyber::encapsulate(&kyber_a.public, &mut random::SecureRandom::default()).expect("kyber encapsulate failed")));
group.bench_function("kyber_decapsulate", |b| b.iter(|| pqc_kyber::decapsulate(&kyber_encap.0, &kyber_a.secret).expect("kyber decapsulate failed")));
group.bench_function("kyber_encapsulate", |b| {
b.iter(|| pqc_kyber::encapsulate(&kyber_a.public, &mut random::SecureRandom::default()).expect("kyber encapsulate failed"))
});
group.bench_function("kyber_decapsulate", |b| {
b.iter(|| pqc_kyber::decapsulate(&kyber_encap.0, &kyber_a.secret).expect("kyber decapsulate failed"))
});
group.finish();
}

0
core-crypto/rustfmt.toml → crypto/rustfmt.toml
View file

0
core-crypto/src/aes.rs → crypto/src/aes.rs
View file

0
core-crypto/src/aes_gmac_siv/AES-GMAC-SIV.png → crypto/src/aes_gmac_siv/AES-GMAC-SIV.png
View file

Side by side Swipe Overlay

Before

Width:  |  Height:  |  Size: 150 KiB

After

Width:  |  Height:  |  Size: 150 KiB

0
core-crypto/src/aes_gmac_siv/README.md → crypto/src/aes_gmac_siv/README.md
View file

0
core-crypto/src/aes_gmac_siv/impl_macos.rs → crypto/src/aes_gmac_siv/impl_macos.rs
View file

0
core-crypto/src/aes_gmac_siv/impl_openssl.rs → crypto/src/aes_gmac_siv/impl_openssl.rs
View file

0
core-crypto/src/aes_gmac_siv/mod.rs → crypto/src/aes_gmac_siv/mod.rs
View file

0
core-crypto/src/hash.rs → crypto/src/hash.rs
View file

10
core-crypto/src/kbkdf.rs → crypto/src/kbkdf.rs
View file

@ -1,6 +1,6 @@
// (c) 2020-2022 ZeroTier, Inc. -- currently propritery pending actual release and licensing. See LICENSE.md.
use crate::hash::{hmac_sha384, hmac_sha512};
use crate::hash::*;
use crate::secret::Secret;
/*
@ -12,12 +12,10 @@ use crate::secret::Secret;
* See: https://csrc.nist.gov/publications/detail/sp/800-108/final (page 12)
*/
/// Derive a key using HMAC-SHA384 and a single byte label, ZeroTier variant with "ZT" preface.
pub fn zt_kbkdf_hmac_sha384(key: &[u8], label: u8) -> Secret<48> {
Secret(hmac_sha384(key, &[0, 0, 0, 0, b'Z', b'T', label, 0, 0, 0, 0, 0x01, 0x80]))
}
/// Derive a key using HMAC-SHA512 and a single byte label, ZeroTier variant with "ZT" preface.
pub fn zt_kbkdf_hmac_sha512(key: &[u8], label: u8) -> Secret<64> {
Secret(hmac_sha512(key, &[0, 0, 0, 0, b'Z', b'T', label, 0, 0, 0, 0, 0x02, 0x00]))
}
//pub fn zt_kbkdf_hmac_sha512(key: &[u8], label: u8) -> Secret<64> {
// Secret(hmac_sha512(key, &[0, 0, 0, 0, b'Z', b'T', label, 0, 0, 0, 0, 0x02, 0x00]))
//}

0
core-crypto/src/lib.rs → crypto/src/lib.rs
View file

0
core-crypto/src/p384.rs → crypto/src/p384.rs
View file

0
core-crypto/src/poly1305.rs → crypto/src/poly1305.rs
View file

0
core-crypto/src/random.rs → crypto/src/random.rs
View file

0
core-crypto/src/salsa.rs → crypto/src/salsa.rs
View file

0
core-crypto/src/secret.rs → crypto/src/secret.rs
View file

0
core-crypto/src/x25519.rs → crypto/src/x25519.rs
View file

4
core-crypto/src/zssp.rs → crypto/src/zssp.rs
View file

@ -1614,7 +1614,7 @@ mod tests {
));
let mut ts = 0;
for test_loop in 0..128 {
for test_loop in 0..256 {
for host in [&alice_host, &bob_host] {
let send_to_other = |data: &mut [u8]| {
if std::ptr::eq(host, &alice_host) {
@ -1681,7 +1681,7 @@ mod tests {
);
}
}
for _ in 0..32 {
for _ in 0..4 {
assert!(session
.send(send_to_other, &mut mtu_buffer, &data_buf[..((random::xorshift64_random() as usize) % data_buf.len())])
.is_ok());

2
network-hypervisor/Cargo.toml
View file

@ -10,7 +10,7 @@ default = ["debug_events"]
debug_events = []
[dependencies]
zerotier-core-crypto = { path = "../core-crypto" }
zerotier-crypto = { path = "../crypto" }
zerotier-utils = { path = "../utils" }
async-trait = "^0"
base64 = "^0"

3
network-hypervisor/src/util/buffer.rs
View file

@ -3,8 +3,7 @@
use std::io::{Read, Write};
use std::mem::{size_of, MaybeUninit};
use crate::util::pool::PoolFactory;
use zerotier_utils::pool::PoolFactory;
use zerotier_utils::varint;
/// An I/O buffer with extensions for efficiently reading and writing various objects.

39
network-hypervisor/src/util/canonicalobject.rs
View file

@ -1,39 +0,0 @@
// (c) 2020-2022 ZeroTier, Inc. -- currently propritery pending actual release and licensing. See LICENSE.md.
use std::sync::atomic::{AtomicUsize, Ordering};
use lazy_static::lazy_static;
lazy_static! {
static ref CANONICAL_ID_COUNTER: AtomicUsize = AtomicUsize::new(0);
}
/// An object that implements equality such that each instance is globally unique in a runtime/process.
///
/// This is used to make canonicalized objects like Path and Peer implement eq() accordingly. A unique
/// ID assigned internally from a counter is used instead of the object's location in memory because
/// technically objects can move in Rust. Canonical objects are encased in Arc<> and unlikely to do so,
/// but this is "correct."
#[repr(transparent)]
pub struct CanonicalObject(usize);
impl CanonicalObject {
#[inline(always)]
pub fn new() -> Self {
Self(CANONICAL_ID_COUNTER.fetch_add(1, Ordering::SeqCst))
}
#[inline(always)]
pub fn canonical_instance_id(&self) -> usize {
self.0
}
}
impl PartialEq for CanonicalObject {
#[inline(always)]
fn eq(&self, other: &Self) -> bool {
self.0 == other.0
}
}
impl Eq for CanonicalObject {}

4
network-hypervisor/src/util/gate.rs
View file

@ -1,6 +1,6 @@
// (c) 2020-2022 ZeroTier, Inc. -- currently propritery pending actual release and licensing. See LICENSE.md.
use std::sync::atomic::{AtomicI64, Ordering};
//use std::sync::atomic::{AtomicI64, Ordering};
/// Boolean rate limiter with normal (non-atomic) semantics.
#[repr(transparent)]
@ -30,6 +30,7 @@ impl<const FREQ: i64> IntervalGate<FREQ> {
}
}
/*
/// Boolean rate limiter with atomic semantics.
#[repr(transparent)]
pub struct AtomicIntervalGate<const FREQ: i64>(AtomicI64);
@ -60,3 +61,4 @@ impl<const FREQ: i64> AtomicIntervalGate<FREQ> {
}
}
}
*/

2
network-hypervisor/src/util/mod.rs
View file

@ -1,10 +1,8 @@
// (c) 2020-2022 ZeroTier, Inc. -- currently propritery pending actual release and licensing. See LICENSE.md.
pub mod buffer;
pub(crate) mod canonicalobject;
pub(crate) mod gate;
pub mod marshalable;
pub(crate) mod pool;
/// A value for ticks that indicates that something never happened, and is thus very long before zero ticks.
pub(crate) const NEVER_HAPPENED_TICKS: i64 = -2147483648;

4
network-hypervisor/src/vl1/fragmentedpacket.rs
View file

@ -9,7 +9,7 @@ use crate::vl1::protocol::*;
/// compiler to optimize out any additional state in Option.
pub(crate) struct FragmentedPacket {
pub ts_ticks: i64,
pub frags: [Option<PooledPacketBuffer>; packet_constants::FRAGMENT_COUNT_MAX],
pub frags: [Option<PooledPacketBuffer>; v1::FRAGMENT_COUNT_MAX],
pub have: u8,
pub expecting: u8,
}
@ -17,7 +17,7 @@ pub(crate) struct FragmentedPacket {
impl FragmentedPacket {
pub fn new(ts: i64) -> Self {
// 'have' and 'expecting' must be expanded if this is >8
debug_assert!(packet_constants::FRAGMENT_COUNT_MAX <= 8);
debug_assert!(v1::FRAGMENT_COUNT_MAX <= 8);
Self {
ts_ticks: ts,

80
network-hypervisor/src/vl1/identity.rs
View file

@ -8,11 +8,11 @@ use std::str::FromStr;
use serde::{Deserialize, Deserializer, Serialize, Serializer};
use zerotier_core_crypto::hash::*;
use zerotier_core_crypto::p384::*;
use zerotier_core_crypto::salsa::Salsa;
use zerotier_core_crypto::secret::Secret;
use zerotier_core_crypto::x25519::*;
use zerotier_crypto::hash::*;
use zerotier_crypto::p384::*;
use zerotier_crypto::salsa::Salsa;
use zerotier_crypto::secret::Secret;
use zerotier_crypto::x25519::*;
use zerotier_utils::hex;
@ -218,9 +218,6 @@ impl Identity {
let _ = self_sign_buf.write_all(p384_ecdh.public_key_bytes());
let _ = self_sign_buf.write_all(p384_ecdsa.public_key_bytes());
// Fingerprint includes only the above fields, so calc before appending the ECDSA signature.
self.fingerprint = SHA384::hash(self_sign_buf.as_slice());
// Sign all keys including the x25519 ones with the new P-384 keys.
let ecdsa_self_signature = p384_ecdsa.sign(self_sign_buf.as_slice());
@ -238,6 +235,8 @@ impl Identity {
});
let _ = self.secret.as_mut().unwrap().p384.insert(IdentityP384Secret { ecdh: p384_ecdh, ecdsa: p384_ecdsa });
self.fingerprint = SHA384::hash(self.to_public_bytes().as_bytes());
return Ok(true);
}
return Ok(false);
@ -451,8 +450,8 @@ impl Identity {
/// Convert a byte respresentation into an identity.
///
/// WARNING: this performs basic sanity checking but does NOT perform a full validation of address derivation or self-signatures.
pub fn from_bytes(b: &IdentityBytes) -> Option<Self> {
match b {
pub fn from_bytes(bytes: &IdentityBytes) -> Option<Self> {
let mut id = match bytes {
IdentityBytes::X25519Public(b) => {
let b: &packed::V0 = bytes_as_flat_object(b);
if b.key_type == 0 && b.secret_length == 0 && b.reserved == 0x03 && u16::from_be_bytes(b.ext_len) == 0 {
@ -462,13 +461,7 @@ impl Identity {
ed25519: b.ed25519,
p384: None,
secret: None,
fingerprint: {
let mut sha = SHA384::new();
sha.update(&b.address);
sha.update(&b.x25519);
sha.update(&b.ed25519);
sha.finish()
},
fingerprint: [0; 48],
})
} else {
None
@ -487,13 +480,7 @@ impl Identity {
ed25519: Ed25519KeyPair::from_bytes(&b.ed25519, &b.ed25519_secret)?,
p384: None,
}),
fingerprint: {
let mut sha = SHA384::new();
sha.update(&b.address);
sha.update(&b.x25519);
sha.update(&b.ed25519);
sha.finish()
},
fingerprint: [0; 48],
})
} else {
None
@ -518,16 +505,7 @@ impl Identity {
ed25519_self_signature: b.ed25519_self_signature,
}),
secret: None,
fingerprint: {
let mut sha = SHA384::new();
sha.update(&b.v0.address);
sha.update(&b.v0.x25519);
sha.update(&b.v0.ed25519);
sha.update(&[Self::ALGORITHM_EC_NIST_P384]);
sha.update(&b.ecdh);
sha.update(&b.ecdsa);
sha.finish()
},
fingerprint: [0; 48],
})
} else {
None
@ -559,22 +537,16 @@ impl Identity {
ecdsa: P384KeyPair::from_bytes(&b.ecdsa, &b.ecdsa_secret)?,
}),
}),
fingerprint: {
let mut sha = SHA384::new();
sha.update(&b.v0s.address);
sha.update(&b.v0s.x25519);
sha.update(&b.v0s.ed25519);
sha.update(&[Self::ALGORITHM_EC_NIST_P384]);
sha.update(&b.ecdh);
sha.update(&b.ecdsa);
sha.finish()
},
fingerprint: [0; 48],
})
} else {
None
}
}
}
};
let fingerprint = SHA384::hash(id.as_ref().unwrap().to_public_bytes().as_bytes());
id.as_mut().unwrap().fingerprint = fingerprint;
id
}
/// Read an identity from a reader, inferring its total length from the stream.
@ -723,7 +695,7 @@ impl FromStr for Identity {
sha.update(&keys[2].as_slice()[0..(P384_PUBLIC_KEY_SIZE * 2)]);
}
Ok(Identity {
let mut id = Ok(Identity {
address,
x25519: keys[0].as_slice()[0..32].try_into().unwrap(),
ed25519: keys[0].as_slice()[32..64].try_into().unwrap(),
@ -787,8 +759,13 @@ impl FromStr for Identity {
},
})
},
fingerprint: sha.finish(),
})
fingerprint: [0; 48],
});
let fingerprint = SHA384::hash(id.as_ref().unwrap().to_public_bytes().as_bytes());
id.as_mut().unwrap().fingerprint = fingerprint;
id
}
}
@ -893,6 +870,13 @@ pub enum IdentityBytes {
X25519P384Secret([u8; Identity::BYTE_LENGTH_X25519P384_SECRET]),
}
impl IdentityBytes {
#[inline(always)]
pub fn as_bytes(&self) -> &[u8] {
self.into()
}
}
impl<'a> From<&'a IdentityBytes> for &'a [u8] {
fn from(b: &'a IdentityBytes) -> Self {
match b {

1
network-hypervisor/src/vl1/mod.rs
View file

@ -1,7 +1,6 @@
// (c) 2020-2022 ZeroTier, Inc. -- currently propritery pending actual release and licensing. See LICENSE.md.
mod address;
mod careof;
mod dictionary;
mod endpoint;
mod fragmentedpacket;

43
network-hypervisor/src/vl1/node.rs
View file

@ -14,7 +14,6 @@ use crate::error::InvalidParameterError;
use crate::util::debug_event;
use crate::util::gate::IntervalGate;
use crate::util::marshalable::Marshalable;
use crate::vl1::careof::CareOf;
use crate::vl1::path::{Path, PathServiceResult};
use crate::vl1::peer::Peer;
use crate::vl1::protocol::*;
@ -22,6 +21,7 @@ use crate::vl1::whoisqueue::{QueuedPacket, WhoisQueue};
use crate::vl1::{Address, Endpoint, Identity, RootSet};
use crate::Event;
use zerotier_crypto::random;
use zerotier_utils::hex;
/// Trait implemented by external code to handle events and provide an interface to the system or application.
@ -54,7 +54,7 @@ pub trait SystemInterface: Sync + Send + 'static {
/// Load this node's identity from the data store.
async fn load_node_identity(&self) -> Option<Identity>;
/// Save this node's identity.
/// Save this node's identity to the data store.
async fn save_node_identity(&self, id: &Identity);
/// Called to send a packet over the physical network (virtual -> physical).
@ -133,7 +133,6 @@ struct BackgroundTaskIntervals {
struct RootInfo<SI: SystemInterface> {
sets: HashMap<String, RootSet>,
roots: HashMap<Arc<Peer<SI>>, Vec<Endpoint>>,
care_of: Vec<u8>,
my_root_sets: Option<Vec<u8>>,
sets_modified: bool,
online: bool,
@ -249,7 +248,7 @@ impl<SI: SystemInterface> Node<SI> {
debug_event!(si, "[vl1] loaded identity {}", id.to_string());
Ok(Self {
instance_id: zerotier_core_crypto::random::get_bytes_secure(),
instance_id: random::get_bytes_secure(),
identity: id,
intervals: Mutex::new(BackgroundTaskIntervals::default()),
paths: parking_lot::RwLock::new(HashMap::new()),
@ -257,7 +256,6 @@ impl<SI: SystemInterface> Node<SI> {
roots: RwLock::new(RootInfo {
sets: HashMap::new(),
roots: HashMap::new(),
care_of: Vec::new(),
my_root_sets: None,
sets_modified: false,
online: false,
@ -286,7 +284,7 @@ impl<SI: SystemInterface> Node<SI> {
// The best root is the one that has replied to a HELLO most recently. Since we send HELLOs in unison
// this is a proxy for latency and also causes roots that fail to reply to drop out quickly.
let mut best: Option<&Arc<Peer<SI>>> = None;
let mut best = None;
let mut latest_hello_reply = 0;
for (r, _) in roots.roots.iter() {
let t = r.last_hello_reply_time_ticks.load(Ordering::Relaxed);
@ -470,20 +468,9 @@ impl<SI: SystemInterface> Node<SI> {
new_root_identities.sort_unstable();
if !old_root_identities.eq(&new_root_identities) {
let mut care_of = CareOf::new(si.time_clock());
for id in new_root_identities.iter() {
care_of.add_care_of(id);
}
assert!(care_of.sign(&self.identity));
let care_of = care_of.to_bytes();
{
let mut roots = self.roots.write();
roots.roots = new_roots;
roots.care_of = care_of;
roots.my_root_sets = my_root_sets;
}
si.event(Event::UpdatedRoots(old_root_identities, new_root_identities));
}
}
@ -584,7 +571,7 @@ impl<SI: SystemInterface> Node<SI> {
source_local_interface.to_string()
);
if let Ok(fragment_header) = data.struct_mut_at::<FragmentHeader>(0) {
if let Ok(fragment_header) = data.struct_mut_at::<v1::FragmentHeader>(0) {
if let Some(dest) = Address::from_bytes_fixed(&fragment_header.dest) {
let time_ticks = si.time_ticks();
if dest == self.identity.address {
@ -607,10 +594,10 @@ impl<SI: SystemInterface> Node<SI> {
#[cfg(debug_assertions)]
debug_event!(si, "[vl1] #{:0>16x} packet fully assembled!", fragment_header_id);
if let Ok(packet_header) = frag0.struct_at::<PacketHeader>(0) {
if let Ok(packet_header) = frag0.struct_at::<v1::PacketHeader>(0) {
if let Some(source) = Address::from_bytes(&packet_header.src) {
if let Some(peer) = self.peer(source) {
peer.receive(self, si, ph, time_ticks, &path, &packet_header, frag0, &assembled_packet.frags[1..(assembled_packet.have as usize)])
peer.receive(self, si, ph, time_ticks, &path, packet_header, frag0, &assembled_packet.frags[1..(assembled_packet.have as usize)])
.await;
} else {
self.whois.query(self, si, source, Some(QueuedPacket::Fragmented(assembled_packet)));
@ -621,12 +608,12 @@ impl<SI: SystemInterface> Node<SI> {
}
} else {
#[cfg(debug_assertions)]
if let Ok(packet_header) = data.struct_at::<PacketHeader>(0) {
if let Ok(packet_header) = data.struct_at::<v1::PacketHeader>(0) {
debug_event!(si, "[vl1] #{:0>16x} is unfragmented", u64::from_be_bytes(packet_header.id));
if let Some(source) = Address::from_bytes(&packet_header.src) {
if let Some(peer) = self.peer(source) {
peer.receive(self, si, ph, time_ticks, &path, &packet_header, data.as_ref(), &[]).await;
peer.receive(self, si, ph, time_ticks, &path, packet_header, data.as_ref(), &[]).await;
} else {
self.whois.query(self, si, source, Some(QueuedPacket::Unfragmented(data)));
}
@ -643,19 +630,19 @@ impl<SI: SystemInterface> Node<SI> {
debug_packet_id = u64::from_be_bytes(fragment_header.id);
debug_event!(si, "[vl1] #{:0>16x} forwarding packet fragment to {}", debug_packet_id, dest.to_string());
}
if fragment_header.increment_hops() > FORWARD_MAX_HOPS {
if fragment_header.increment_hops() > v1::FORWARD_MAX_HOPS {
#[cfg(debug_assertions)]
debug_event!(si, "[vl1] #{:0>16x} discarded: max hops exceeded!", debug_packet_id);
return;
}
} else {
if let Ok(packet_header) = data.struct_mut_at::<PacketHeader>(0) {
if let Ok(packet_header) = data.struct_mut_at::<v1::PacketHeader>(0) {
#[cfg(debug_assertions)]
{
debug_packet_id = u64::from_be_bytes(packet_header.id);
debug_event!(si, "[vl1] #{:0>16x} forwarding packet to {}", debug_packet_id, dest.to_string());
}
if packet_header.increment_hops() > FORWARD_MAX_HOPS {
if packet_header.increment_hops() > v1::FORWARD_MAX_HOPS {
#[cfg(debug_assertions)]
debug_event!(si, "[vl1] #{:0>16x} discarded: max hops exceeded!", u64::from_be_bytes(packet_header.id));
return;
@ -681,7 +668,7 @@ impl<SI: SystemInterface> Node<SI> {
}
pub fn is_peer_root(&self, peer: &Peer<SI>) -> bool {
self.roots.read().roots.keys().any(|p| (**p).eq(peer))
self.roots.read().roots.keys().any(|p| p.identity.eq(&peer.identity))
}
pub(crate) fn remote_update_root_set(&self, received_from: &Identity, rs: RootSet) {
@ -727,10 +714,6 @@ impl<SI: SystemInterface> Node<SI> {
self.roots.read().my_root_sets.is_some()
}
pub(crate) fn care_of_bytes(&self) -> Vec<u8> {
self.roots.read().care_of.clone()
}
pub(crate) fn canonical_path(&self, ep: &Endpoint, local_socket: &SI::LocalSocket, local_interface: &SI::LocalInterface, time_ticks: i64) -> Arc<Path<SI>> {
if let Some(path) = self.paths.read().get(&PathKey::Ref(ep, local_socket)) {
return path.clone();

26
network-hypervisor/src/vl1/path.rs
View file

@ -6,12 +6,13 @@ use std::sync::atomic::{AtomicI64, Ordering};
use parking_lot::Mutex;
use crate::util::canonicalobject::CanonicalObject;
use crate::vl1::endpoint::Endpoint;
use crate::vl1::fragmentedpacket::FragmentedPacket;
use crate::vl1::node::*;
use crate::vl1::protocol::*;
use zerotier_crypto::random;
pub(crate) const SERVICE_INTERVAL_MS: i64 = PATH_KEEPALIVE_INTERVAL;
pub(crate) enum PathServiceResult {
@ -25,7 +26,6 @@ pub(crate) enum PathServiceResult {
/// one and only one unique path object. That enables statistics to be tracked
/// for them and uniform application of things like keepalives.
pub struct Path<SI: SystemInterface> {
pub(crate) canonical: CanonicalObject,
pub endpoint: Endpoint,
pub local_socket: SI::LocalSocket,
pub local_interface: SI::LocalInterface,
@ -38,14 +38,13 @@ pub struct Path<SI: SystemInterface> {
impl<SI: SystemInterface> Path<SI> {
pub fn new(endpoint: Endpoint, local_socket: SI::LocalSocket, local_interface: SI::LocalInterface, time_ticks: i64) -> Self {
Self {
canonical: CanonicalObject::new(),
endpoint,
local_socket,
local_interface,
last_send_time_ticks: AtomicI64::new(crate::util::NEVER_HAPPENED_TICKS),
last_receive_time_ticks: AtomicI64::new(crate::util::NEVER_HAPPENED_TICKS),
create_time_ticks: time_ticks,
fragmented_packets: Mutex::new(HashMap::with_capacity_and_hasher(4, PacketIdHasher(zerotier_core_crypto::random::xorshift64_random()))),
fragmented_packets: Mutex::new(HashMap::with_capacity_and_hasher(4, PacketIdHasher(random::xorshift64_random()))),
}
}
@ -57,7 +56,7 @@ impl<SI: SystemInterface> Path<SI> {
// Discard some old waiting packets if the total incoming fragments for a path exceeds a
// sanity limit. This is to prevent memory exhaustion DOS attacks.
let fps = fp.len();
if fps > packet_constants::FRAGMENT_MAX_INBOUND_PACKETS_PER_PATH {
if fps > v1::FRAGMENT_MAX_INBOUND_PACKETS_PER_PATH {
let mut entries: Vec<(i64, u64)> = Vec::new();
entries.reserve(fps);
for f in fp.iter() {
@ -69,7 +68,11 @@ impl<SI: SystemInterface> Path<SI> {
}
}
if fp.entry(packet_id).or_insert_with(|| FragmentedPacket::new(time_ticks)).add_fragment(packet, fragment_no, fragment_expecting_count) {
if fp
.entry(packet_id)
.or_insert_with(|| FragmentedPacket::new(time_ticks))
.add_fragment(packet, fragment_no, fragment_expecting_count)
{
fp.remove(&packet_id)
} else {
None
@ -87,7 +90,7 @@ impl<SI: SystemInterface> Path<SI> {
}
pub(crate) fn service(&self, time_ticks: i64) -> PathServiceResult {
self.fragmented_packets.lock().retain(|_, frag| (time_ticks - frag.ts_ticks) < packet_constants::FRAGMENT_EXPIRATION);
self.fragmented_packets.lock().retain(|_, frag| (time_ticks - frag.ts_ticks) < v1::FRAGMENT_EXPIRATION);
if (time_ticks - self.last_receive_time_ticks.load(Ordering::Relaxed)) < PATH_EXPIRATION_TIME {
if (time_ticks - self.last_send_time_ticks.load(Ordering::Relaxed)) >= PATH_KEEPALIVE_INTERVAL {
self.last_send_time_ticks.store(time_ticks, Ordering::Relaxed);
@ -103,15 +106,6 @@ impl<SI: SystemInterface> Path<SI> {
}
}
impl<SI: SystemInterface> PartialEq for Path<SI> {
#[inline(always)]
fn eq(&self, other: &Self) -> bool {
self.canonical.eq(&other.canonical)
}
}
impl<SI: SystemInterface> Eq for Path<SI> {}
#[repr(transparent)]
struct PacketIdHasher(u64);

304
network-hypervisor/src/vl1/peer.rs
View file

@ -1,42 +1,38 @@
// (c) 2020-2022 ZeroTier, Inc. -- currently propritery pending actual release and licensing. See LICENSE.md.
use std::collections::HashMap;
use std::hash::{Hash, Hasher};
use std::hash::Hash;
use std::sync::atomic::{AtomicI64, AtomicU64, Ordering};
use std::sync::{Arc, Weak};
use parking_lot::{Mutex, RwLock};
use zerotier_core_crypto::random::next_u64_secure;
use zerotier_core_crypto::salsa::Salsa;
use zerotier_core_crypto::secret::Secret;
use zerotier_crypto::poly1305;
use zerotier_crypto::random::next_u64_secure;
use zerotier_crypto::salsa::Salsa;
use zerotier_crypto::secret::Secret;
use crate::util::buffer::BufferReader;
use crate::util::byte_array_range;
use crate::util::canonicalobject::CanonicalObject;
use crate::util::debug_event;
use crate::util::marshalable::Marshalable;
use crate::vl1::address::Address;
use crate::vl1::careof::CareOf;
use crate::vl1::node::*;
use crate::vl1::protocol::*;
use crate::vl1::rootset::RootSet;
use crate::vl1::symmetricsecret::SymmetricSecret;
use crate::vl1::{Dictionary, Endpoint, Identity, Path};
use crate::vl1::{Endpoint, Identity, Path};
use crate::{VERSION_MAJOR, VERSION_MINOR, VERSION_REVISION};
pub(crate) const SERVICE_INTERVAL_MS: i64 = security_constants::EPHEMERAL_SECRET_REKEY_AFTER_TIME / 10;
pub(crate) const SERVICE_INTERVAL_MS: i64 = 10000;
struct PeerPath<SI: SystemInterface> {
path: Weak<Path<SI>>,
canonical_instance_id: usize,
last_receive_time_ticks: i64,
}
struct RemoteNodeInfo {
remote_instance_id: [u8; 16],
reported_local_endpoints: HashMap<Endpoint, i64>,
care_of: Option<CareOf>,
remote_version: u64,
remote_protocol_version: u8,
}
@ -45,8 +41,6 @@ struct RemoteNodeInfo {
///
/// Equality and hashing is implemented in terms of the identity.
pub struct Peer<SI: SystemInterface> {
canonical: CanonicalObject,
// This peer's identity.
pub identity: Identity,
@ -75,27 +69,33 @@ pub struct Peer<SI: SystemInterface> {
}
/// Attempt AEAD packet encryption and MAC validation. Returns message ID on success.
fn try_aead_decrypt(secret: &SymmetricSecret, packet_frag0_payload_bytes: &[u8], packet_header: &PacketHeader, fragments: &[Option<PooledPacketBuffer>], payload: &mut PacketBuffer) -> Option<MessageId> {
fn try_aead_decrypt(
secret: &SymmetricSecret,
packet_frag0_payload_bytes: &[u8],
packet_header: &v1::PacketHeader,
fragments: &[Option<PooledPacketBuffer>],
payload: &mut PacketBuffer,
) -> Option<MessageId> {
let cipher = packet_header.cipher();
match cipher {
security_constants::CIPHER_NOCRYPT_POLY1305 | security_constants::CIPHER_SALSA2012_POLY1305 => {
v1::CIPHER_NOCRYPT_POLY1305 | v1::CIPHER_SALSA2012_POLY1305 => {
let _ = payload.append_bytes(packet_frag0_payload_bytes);
for f in fragments.iter() {
if let Some(f) = f.as_ref() {
if let Ok(f) = f.as_bytes_starting_at(packet_constants::FRAGMENT_HEADER_SIZE) {
if let Ok(f) = f.as_bytes_starting_at(v1::FRAGMENT_HEADER_SIZE) {
let _ = payload.append_bytes(f);
}
}
}
let (mut salsa, poly1305_key) = salsa_poly_create(secret, packet_header, payload.len() + packet_constants::HEADER_SIZE);
let mac = zerotier_core_crypto::poly1305::compute(&poly1305_key, &payload.as_bytes());
let (mut salsa, poly1305_key) = salsa_poly_create(secret, packet_header, payload.len() + v1::HEADER_SIZE);
let mac = poly1305::compute(&poly1305_key, &payload.as_bytes());
if mac[0..8].eq(&packet_header.mac) {
let message_id = u64::from_ne_bytes(packet_header.id);
if cipher == security_constants::CIPHER_SALSA2012_POLY1305 {
if cipher == v1::CIPHER_SALSA2012_POLY1305 {
salsa.crypt_in_place(payload.as_bytes_mut());
Some(message_id)
} else if (payload.u8_at(0).unwrap_or(0) & packet_constants::VERB_MASK) == verbs::VL1_HELLO {
} else if (payload.u8_at(0).unwrap_or(0) & v1::VERB_MASK) == verbs::VL1_HELLO {
Some(message_id)
} else {
// SECURITY: fail if there is no encryption and the message is not HELLO. No other types are allowed
@ -107,7 +107,7 @@ fn try_aead_decrypt(secret: &SymmetricSecret, packet_frag0_payload_bytes: &[u8],
}
}
security_constants::CIPHER_AES_GMAC_SIV => {
v1::CIPHER_AES_GMAC_SIV => {
let mut aes_gmac_siv = secret.aes_gmac_siv.get();
aes_gmac_siv.decrypt_init(&packet_header.aes_gmac_siv_tag());
@ -122,7 +122,7 @@ fn try_aead_decrypt(secret: &SymmetricSecret, packet_frag0_payload_bytes: &[u8],
packet_header.src[2],
packet_header.src[3],
packet_header.src[4],
packet_header.flags_cipher_hops & packet_constants::FLAGS_FIELD_MASK_HIDE_HOPS,
packet_header.flags_cipher_hops & v1::FLAGS_FIELD_MASK_HIDE_HOPS,
]);
if let Ok(b) = payload.append_bytes_get_mut(packet_frag0_payload_bytes.len()) {
@ -130,7 +130,7 @@ fn try_aead_decrypt(secret: &SymmetricSecret, packet_frag0_payload_bytes: &[u8],
}
for f in fragments.iter() {
if let Some(f) = f.as_ref() {
if let Ok(f) = f.as_bytes_starting_at(packet_constants::FRAGMENT_HEADER_SIZE) {
if let Ok(f) = f.as_bytes_starting_at(v1::FRAGMENT_HEADER_SIZE) {
if let Ok(b) = payload.append_bytes_get_mut(f.len()) {
aes_gmac_siv.decrypt(f, b);
}
@ -154,14 +154,14 @@ fn try_aead_decrypt(secret: &SymmetricSecret, packet_frag0_payload_bytes: &[u8],
/// Create initialized instances of Salsa20/12 and Poly1305 for a packet.
/// (Note that this is a legacy cipher suite.)
fn salsa_poly_create(secret: &SymmetricSecret, header: &PacketHeader, packet_size: usize) -> (Salsa<12>, [u8; 32]) {
fn salsa_poly_create(secret: &SymmetricSecret, header: &v1::PacketHeader, packet_size: usize) -> (Salsa<12>, [u8; 32]) {
// Create a per-packet key from the IV, source, destination, and packet size.
let mut key: Secret<32> = secret.key.first_n_clone();
let hb = header.as_bytes();
for i in 0..18 {
key.0[i] ^= hb[i];
}
key.0[18] ^= header.flags_cipher_hops & packet_constants::FLAGS_FIELD_MASK_HIDE_HOPS;
key.0[18] ^= header.flags_cipher_hops & v1::FLAGS_FIELD_MASK_HIDE_HOPS;
key.0[19] ^= packet_size as u8;
key.0[20] ^= packet_size.wrapping_shr(8) as u8;
@ -184,7 +184,6 @@ impl<SI: SystemInterface> Peer<SI> {
pub(crate) fn new(this_node_identity: &Identity, id: Identity, time_ticks: i64) -> Option<Peer<SI>> {
this_node_identity.agree(&id).map(|static_secret| -> Self {
Self {
canonical: CanonicalObject::new(),
identity: id,
identity_symmetric_key: SymmetricSecret::new(static_secret),
paths: Mutex::new(Vec::with_capacity(4)),
@ -199,7 +198,6 @@ impl<SI: SystemInterface> Peer<SI> {
remote_node_info: RwLock::new(RemoteNodeInfo {
remote_instance_id: [0_u8; 16],
reported_local_endpoints: HashMap::new(),
care_of: None,
remote_version: 0,
remote_protocol_version: 0,
}),
@ -266,16 +264,21 @@ impl<SI: SystemInterface> Peer<SI> {
// the same IP address but a different port. This prevents the accumulation of duplicate
// paths to the same peer over different ports.
for pi in paths.iter_mut() {
if pi.canonical_instance_id == new_path.canonical.canonical_instance_id() {
if std::ptr::eq(pi.path.as_ptr(), new_path.as_ref()) {
return;
}
if let Some(p) = pi.path.upgrade() {
match &p.endpoint {
Endpoint::IpUdp(existing_ip) => {
if existing_ip.ip_bytes().eq(new_ip.ip_bytes()) {
debug_event!(si, "[vl1] {} replacing path {} with {} (same IP, different port)", self.identity.address.to_string(), p.endpoint.to_string(), new_path.endpoint.to_string());
debug_event!(
si,
"[vl1] {} replacing path {} with {} (same IP, different port)",
self.identity.address.to_string(),
p.endpoint.to_string(),
new_path.endpoint.to_string()
);
pi.path = Arc::downgrade(new_path);
pi.canonical_instance_id = new_path.canonical.canonical_instance_id();
pi.last_receive_time_ticks = time_ticks;
prioritize_paths(&mut paths);
return;
@ -288,7 +291,7 @@ impl<SI: SystemInterface> Peer<SI> {
}
_ => {
for pi in paths.iter() {
if pi.canonical_instance_id == new_path.canonical.canonical_instance_id() {
if std::ptr::eq(pi.path.as_ptr(), new_path.as_ref()) {
return;
}
}
@ -299,7 +302,6 @@ impl<SI: SystemInterface> Peer<SI> {
debug_event!(si, "[vl1] {} learned new path: {}", self.identity.address.to_string(), new_path.endpoint.to_string());
paths.push(PeerPath::<SI> {
path: Arc::downgrade(new_path),
canonical_instance_id: new_path.canonical.canonical_instance_id(),
last_receive_time_ticks: time_ticks,
});
prioritize_paths(&mut paths);
@ -320,7 +322,15 @@ impl<SI: SystemInterface> Peer<SI> {
///
/// This does not set the fragmentation field in the packet header, MAC, or encrypt the packet. The sender
/// must do that while building the packet. The fragmentation flag must be set if fragmentation will be needed.
async fn internal_send(&self, si: &SI, endpoint: &Endpoint, local_socket: Option<&SI::LocalSocket>, local_interface: Option<&SI::LocalInterface>, max_fragment_size: usize, packet: &PacketBuffer) -> bool {
async fn internal_send(
&self,
si: &SI,
endpoint: &Endpoint,
local_socket: Option<&SI::LocalSocket>,
local_interface: Option<&SI::LocalInterface>,
max_fragment_size: usize,
packet: &PacketBuffer,
) -> bool {
let packet_size = packet.len();
if packet_size > max_fragment_size {
let bytes = packet.as_bytes();
@ -330,18 +340,18 @@ impl<SI: SystemInterface> Peer<SI> {
let mut pos = UDP_DEFAULT_MTU;
let overrun_size = (packet_size - UDP_DEFAULT_MTU) as u32;
let fragment_count = (overrun_size / (UDP_DEFAULT_MTU - packet_constants::FRAGMENT_HEADER_SIZE) as u32) + (((overrun_size % (UDP_DEFAULT_MTU - packet_constants::FRAGMENT_HEADER_SIZE) as u32) != 0) as u32);
debug_assert!(fragment_count <= packet_constants::FRAGMENT_COUNT_MAX as u32);
let fragment_count = (overrun_size / (UDP_DEFAULT_MTU - v1::FRAGMENT_HEADER_SIZE) as u32) + (((overrun_size % (UDP_DEFAULT_MTU - v1::FRAGMENT_HEADER_SIZE) as u32) != 0) as u32);
debug_assert!(fragment_count <= v1::FRAGMENT_COUNT_MAX as u32);
let mut header = FragmentHeader {
let mut header = v1::FragmentHeader {
id: *packet.bytes_fixed_at(0).unwrap(),
dest: *packet.bytes_fixed_at(packet_constants::DESTINATION_INDEX).unwrap(),
fragment_indicator: packet_constants::FRAGMENT_INDICATOR,
dest: *packet.bytes_fixed_at(v1::DESTINATION_INDEX).unwrap(),
fragment_indicator: v1::FRAGMENT_INDICATOR,
total_and_fragment_no: ((fragment_count + 1) << 4) as u8,
reserved_hops: 0,
};
let mut chunk_size = (packet_size - pos).min(UDP_DEFAULT_MTU - packet_constants::HEADER_SIZE);
let mut chunk_size = (packet_size - pos).min(UDP_DEFAULT_MTU - v1::HEADER_SIZE);
loop {
header.total_and_fragment_no += 1;
let next_pos = pos + chunk_size;
@ -350,7 +360,7 @@ impl<SI: SystemInterface> Peer<SI> {
}
pos = next_pos;
if pos < packet_size {
chunk_size = (packet_size - pos).min(UDP_DEFAULT_MTU - packet_constants::HEADER_SIZE);
chunk_size = (packet_size - pos).min(UDP_DEFAULT_MTU - v1::HEADER_SIZE);
} else {
return true;
}
@ -385,20 +395,20 @@ impl<SI: SystemInterface> Peer<SI> {
usize::MAX
};
let flags_cipher_hops = if packet.len() > max_fragment_size {
packet_constants::HEADER_FLAG_FRAGMENTED | security_constants::CIPHER_AES_GMAC_SIV
v1::HEADER_FLAG_FRAGMENTED | v1::CIPHER_AES_GMAC_SIV
} else {
security_constants::CIPHER_AES_GMAC_SIV
v1::CIPHER_AES_GMAC_SIV
};
let mut aes_gmac_siv = self.identity_symmetric_key.aes_gmac_siv.get();
aes_gmac_siv.encrypt_init(&self.next_message_id().to_ne_bytes());
aes_gmac_siv.encrypt_set_aad(&get_packet_aad_bytes(self.identity.address, node.identity.address, flags_cipher_hops));
if let Ok(payload) = packet.as_bytes_starting_at_mut(packet_constants::HEADER_SIZE) {
aes_gmac_siv.encrypt_set_aad(&v1::get_packet_aad_bytes(self.identity.address, node.identity.address, flags_cipher_hops));
if let Ok(payload) = packet.as_bytes_starting_at_mut(v1::HEADER_SIZE) {
aes_gmac_siv.encrypt_first_pass(payload);
aes_gmac_siv.encrypt_first_pass_finish();
aes_gmac_siv.encrypt_second_pass_in_place(payload);
let tag = aes_gmac_siv.encrypt_second_pass_finish();
let header = packet.struct_mut_at::<PacketHeader>(0).unwrap();
let header = packet.struct_mut_at::<v1::PacketHeader>(0).unwrap();
header.id = *byte_array_range::<16, 0, 8>(tag);
header.dest = self.identity.address.to_bytes();
header.src = node.identity.address.to_bytes();
@ -408,7 +418,10 @@ impl<SI: SystemInterface> Peer<SI> {
return false;
}
if self.internal_send(si, &path.endpoint, Some(&path.local_socket), Some(&path.local_interface), max_fragment_size, packet).await {
if self
.internal_send(si, &path.endpoint, Some(&path.local_socket), Some(&path.local_interface), max_fragment_size, packet)
.await
{
self.last_send_time_ticks.store(time_ticks, Ordering::Relaxed);
return true;
}
@ -433,19 +446,6 @@ impl<SI: SystemInterface> Peer<SI> {
return false;
}
pub(crate) fn create_session_metadata(&self, node: &Node<SI>, direct_source_report: Option<&Endpoint>) -> Vec<u8> {
let mut session_metadata = Dictionary::new();
session_metadata.set_bytes(session_metadata::INSTANCE_ID, node.instance_id.to_vec());
session_metadata.set_bytes(session_metadata::CARE_OF, node.care_of_bytes());
if let Some(direct_source) = direct_source_report {
session_metadata.set_bytes(session_metadata::DIRECT_SOURCE, direct_source.to_buffer::<{ Endpoint::MAX_MARSHAL_SIZE }>().unwrap().as_bytes().to_vec());
}
if let Some(my_root_sets) = node.my_root_sets() {
session_metadata.set_bytes(session_metadata::ROOT_SET_UPDATES, my_root_sets);
}
session_metadata.to_bytes()
}
/// Send a HELLO to this peer.
///
/// If explicit_endpoint is not None the packet will be sent directly to this endpoint.
@ -479,12 +479,12 @@ impl<SI: SystemInterface> Peer<SI> {
let message_id = self.next_message_id();
{
let f: &mut (PacketHeader, message_component_structs::HelloFixedHeaderFields) = packet.append_struct_get_mut().unwrap();
let f: &mut (v1::PacketHeader, v1::message_component_structs::HelloFixedHeaderFields) = packet.append_struct_get_mut().unwrap();
f.0.id = message_id.to_ne_bytes();
f.0.dest = self.identity.address.to_bytes();
f.0.src = node.identity.address.to_bytes();
f.0.flags_cipher_hops = security_constants::CIPHER_NOCRYPT_POLY1305;
f.1.verb = verbs::VL1_HELLO | packet_constants::VERB_FLAG_EXTENDED_AUTHENTICATION;
f.0.flags_cipher_hops = v1::CIPHER_NOCRYPT_POLY1305;
f.1.verb = verbs::VL1_HELLO | v1::VERB_FLAG_EXTENDED_AUTHENTICATION;
f.1.version_proto = PROTOCOL_VERSION;
f.1.version_major = VERSION_MAJOR;
f.1.version_minor = VERSION_MINOR;
@ -495,9 +495,9 @@ impl<SI: SystemInterface> Peer<SI> {
debug_assert_eq!(packet.len(), 41);
assert!(packet.append_bytes((&node.identity.to_public_bytes()).into()).is_ok());
let (_, poly1305_key) = salsa_poly_create(&self.identity_symmetric_key, packet.struct_at::<PacketHeader>(0).unwrap(), packet.len());
let mac = zerotier_core_crypto::poly1305::compute(&poly1305_key, packet.as_bytes_starting_at(packet_constants::HEADER_SIZE).unwrap());
packet.as_mut()[packet_constants::MAC_FIELD_INDEX..packet_constants::MAC_FIELD_INDEX + 8].copy_from_slice(&mac[0..8]);
let (_, poly1305_key) = salsa_poly_create(&self.identity_symmetric_key, packet.struct_at::<v1::PacketHeader>(0).unwrap(), packet.len());
let mac = poly1305::compute(&poly1305_key, packet.as_bytes_starting_at(v1::HEADER_SIZE).unwrap());
packet.as_mut()[v1::MAC_FIELD_INDEX..v1::MAC_FIELD_INDEX + 8].copy_from_slice(&mac[0..8]);
self.last_send_time_ticks.store(time_ticks, Ordering::Relaxed);
@ -522,8 +522,18 @@ impl<SI: SystemInterface> Peer<SI> {
/// those fragments after the main packet header and first chunk.
///
/// This returns true if the packet decrypted and passed authentication.
pub(crate) async fn receive<PH: InnerProtocolInterface>(&self, node: &Node<SI>, si: &SI, ph: &PH, time_ticks: i64, source_path: &Arc<Path<SI>>, packet_header: &PacketHeader, frag0: &PacketBuffer, fragments: &[Option<PooledPacketBuffer>]) -> bool {
if let Ok(packet_frag0_payload_bytes) = frag0.as_bytes_starting_at(packet_constants::VERB_INDEX) {
pub(crate) async fn receive<PH: InnerProtocolInterface>(
&self,
node: &Node<SI>,
si: &SI,
ph: &PH,
time_ticks: i64,
source_path: &Arc<Path<SI>>,
packet_header: &v1::PacketHeader,
frag0: &PacketBuffer,
fragments: &[Option<PooledPacketBuffer>],
) -> bool {
if let Ok(packet_frag0_payload_bytes) = frag0.as_bytes_starting_at(v1::VERB_INDEX) {
let mut payload = PacketBuffer::new();
let message_id = if let Some(message_id2) = try_aead_decrypt(&self.identity_symmetric_key, packet_frag0_payload_bytes, packet_header, fragments, &mut payload) {
@ -536,8 +546,8 @@ impl<SI: SystemInterface> Peer<SI> {
};
if let Ok(mut verb) = payload.u8_at(0) {
if (verb & packet_constants::VERB_FLAG_COMPRESSED) != 0 {
let mut decompressed_payload = [0_u8; packet_constants::SIZE_MAX];
if (verb & v1::VERB_FLAG_COMPRESSED) != 0 {
let mut decompressed_payload = [0u8; v1::SIZE_MAX];
decompressed_payload[0] = verb;
if let Ok(dlen) = lz4_flex::block::decompress_into(&payload.as_bytes()[1..], &mut decompressed_payload[1..]) {
payload.set_to(&decompressed_payload[..(dlen + 1)]);
@ -554,15 +564,21 @@ impl<SI: SystemInterface> Peer<SI> {
let mut path_is_known = false;
for p in self.paths.lock().iter_mut() {
if p.canonical_instance_id == source_path.canonical.canonical_instance_id() {
if std::ptr::eq(p.path.as_ptr(), source_path.as_ref()) {
p.last_receive_time_ticks = time_ticks;
path_is_known = true;
break;
}
}
verb &= packet_constants::VERB_MASK; // mask off flags
debug_event!(si, "[vl1] #{:0>16x} decrypted and authenticated, verb: {} ({:0>2x})", u64::from_be_bytes(packet_header.id), verbs::name(verb), verb as u32);
verb &= v1::VERB_MASK; // mask off flags
debug_event!(
si,
"[vl1] #{:0>16x} decrypted and authenticated, verb: {} ({:0>2x})",
u64::from_be_bytes(packet_header.id),
verbs::name(verb),
verb as u32
);
if match verb {
verbs::VL1_NOP => true,
@ -586,27 +602,38 @@ impl<SI: SystemInterface> Peer<SI> {
return false;
}
async fn handle_incoming_hello<PH: InnerProtocolInterface>(&self, si: &SI, ph: &PH, node: &Node<SI>, time_ticks: i64, message_id: MessageId, source_path: &Arc<Path<SI>>, hops: u8, payload: &PacketBuffer) -> bool {
async fn handle_incoming_hello<PH: InnerProtocolInterface>(
&self,
si: &SI,
ph: &PH,
node: &Node<SI>,
time_ticks: i64,
message_id: MessageId,
source_path: &Arc<Path<SI>>,
hops: u8,
payload: &PacketBuffer,
) -> bool {
if !(ph.should_communicate_with(&self.identity) || node.this_node_is_root() || node.is_peer_root(self)) {
debug_event!(si, "[vl1] dropping HELLO from {} due to lack of trust relationship", self.identity.address.to_string());
return true; // packet wasn't invalid, just ignored
}
let mut cursor = 0;
if let Ok(hello_fixed_headers) = payload.read_struct::<message_component_structs::HelloFixedHeaderFields>(&mut cursor) {
if let Ok(hello_fixed_headers) = payload.read_struct::<v1::message_component_structs::HelloFixedHeaderFields>(&mut cursor) {
if let Ok(identity) = Identity::read_bytes(&mut BufferReader::new(payload, &mut cursor)) {
if identity.eq(&self.identity) {
{
let mut remote_node_info = self.remote_node_info.write();
remote_node_info.remote_protocol_version = hello_fixed_headers.version_proto;
remote_node_info.remote_version =
(hello_fixed_headers.version_major as u64).wrapping_shl(48) | (hello_fixed_headers.version_minor as u64).wrapping_shl(32) | (u16::from_be_bytes(hello_fixed_headers.version_revision) as u64).wrapping_shl(16);
remote_node_info.remote_version = (hello_fixed_headers.version_major as u64).wrapping_shl(48)
| (hello_fixed_headers.version_minor as u64).wrapping_shl(32)
| (u16::from_be_bytes(hello_fixed_headers.version_revision) as u64).wrapping_shl(16);
}
let mut packet = PacketBuffer::new();
packet.set_size(packet_constants::HEADER_SIZE);
packet.set_size(v1::HEADER_SIZE);
{
let f: &mut (message_component_structs::OkHeader, message_component_structs::OkHelloFixedHeaderFields) = packet.append_struct_get_mut().unwrap();
let f: &mut (v1::message_component_structs::OkHeader, v1::message_component_structs::OkHelloFixedHeaderFields) = packet.append_struct_get_mut().unwrap();
f.0.verb = verbs::VL1_OK;
f.0.in_re_verb = verbs::VL1_HELLO;
f.0.in_re_message_id = message_id.to_ne_bytes();
@ -616,19 +643,6 @@ impl<SI: SystemInterface> Peer<SI> {
f.1.version_minor = VERSION_MINOR;
f.1.version_revision = VERSION_REVISION.to_be_bytes();
}
if hello_fixed_headers.version_proto >= 20 {
let session_metadata = self.create_session_metadata(
node,
if hops == 0 {
Some(&source_path.endpoint)
} else {
None
},
);
assert!(session_metadata.len() <= 0xffff); // sanity check, should be impossible
assert!(packet.append_u16(session_metadata.len() as u16).is_ok());
assert!(packet.append_bytes(session_metadata.as_slice()).is_ok());
}
return self.send(si, Some(source_path), node, time_ticks, &mut packet).await;
}
@ -639,12 +653,14 @@ impl<SI: SystemInterface> Peer<SI> {
async fn handle_incoming_error<PH: InnerProtocolInterface>(&self, _si: &SI, ph: &PH, _node: &Node<SI>, _time_ticks: i64, source_path: &Arc<Path<SI>>, payload: &PacketBuffer) -> bool {
let mut cursor = 0;
if let Ok(error_header) = payload.read_struct::<message_component_structs::ErrorHeader>(&mut cursor) {
if let Ok(error_header) = payload.read_struct::<v1::message_component_structs::ErrorHeader>(&mut cursor) {
let in_re_message_id: MessageId = u64::from_ne_bytes(error_header.in_re_message_id);
if self.message_id_counter.load(Ordering::Relaxed).wrapping_sub(in_re_message_id) <= PACKET_RESPONSE_COUNTER_DELTA_MAX {
match error_header.in_re_verb {
_ => {
return ph.handle_error(self, &source_path, error_header.in_re_verb, in_re_message_id, error_header.error_code, payload, &mut cursor).await;
return ph
.handle_error(self, &source_path, error_header.in_re_verb, in_re_message_id, error_header.error_code, payload, &mut cursor)
.await;
}
}
}
@ -652,76 +668,36 @@ impl<SI: SystemInterface> Peer<SI> {
return false;
}
async fn handle_incoming_ok<PH: InnerProtocolInterface>(&self, si: &SI, ph: &PH, node: &Node<SI>, time_ticks: i64, source_path: &Arc<Path<SI>>, hops: u8, path_is_known: bool, payload: &PacketBuffer) -> bool {
async fn handle_incoming_ok<PH: InnerProtocolInterface>(
&self,
si: &SI,
ph: &PH,
node: &Node<SI>,
time_ticks: i64,
source_path: &Arc<Path<SI>>,
hops: u8,
path_is_known: bool,
payload: &PacketBuffer,
) -> bool {
let mut cursor = 0;
if let Ok(ok_header) = payload.read_struct::<message_component_structs::OkHeader>(&mut cursor) {
if let Ok(ok_header) = payload.read_struct::<v1::message_component_structs::OkHeader>(&mut cursor) {
let in_re_message_id: MessageId = u64::from_ne_bytes(ok_header.in_re_message_id);
if self.message_id_counter.load(Ordering::Relaxed).wrapping_sub(in_re_message_id) <= PACKET_RESPONSE_COUNTER_DELTA_MAX {
match ok_header.in_re_verb {
verbs::VL1_HELLO => {
if let Ok(ok_hello_fixed_header_fields) = payload.read_struct::<message_component_structs::OkHelloFixedHeaderFields>(&mut cursor) {
if ok_hello_fixed_header_fields.version_proto >= 20 {
// V2 nodes send a whole dictionary for easy extensibiility.
if let Ok(session_metadata_len) = payload.read_u16(&mut cursor) {
if session_metadata_len > 0 {
if let Ok(session_metadata) = payload.read_bytes(session_metadata_len as usize, &mut cursor) {
if let Some(session_metadata) = Dictionary::from_bytes(session_metadata) {
let mut remote_node_info = self.remote_node_info.write();
if hops == 0 {
if let Some(reported_endpoint) = session_metadata.get_bytes(session_metadata::DIRECT_SOURCE) {
if let Some(reported_endpoint) = Endpoint::from_bytes(reported_endpoint) {
#[cfg(debug_assertions)]
let reported_endpoint2 = reported_endpoint.clone();
if remote_node_info.reported_local_endpoints.insert(reported_endpoint, time_ticks).is_none() {
#[cfg(debug_assertions)]
debug_event!(si, "[vl1] {} reported new remote perspective, local endpoint: {}", self.identity.address.to_string(), reported_endpoint2.to_string());
}
}
}
}
if let Some(care_of) = session_metadata.get_bytes(session_metadata::CARE_OF) {
if let Some(care_of) = CareOf::from_bytes(care_of) {
let _ = remote_node_info.care_of.insert(care_of);
}
}
if let Some(instance_id) = session_metadata.get_bytes(session_metadata::INSTANCE_ID) {
remote_node_info.remote_instance_id.fill(0);
let l = instance_id.len().min(remote_node_info.remote_instance_id.len());
(&mut remote_node_info.remote_instance_id[..l]).copy_from_slice(&instance_id[..l]);
}
if let Some(root_set_updates) = session_metadata.get_bytes(session_metadata::ROOT_SET_UPDATES) {
let mut tmp = PacketBuffer::new();
if root_set_updates.len() <= tmp.capacity() {
tmp.set_to(root_set_updates);
let mut cursor = 0;
while cursor < tmp.len() {
if let Ok(rs) = RootSet::unmarshal(&tmp, &mut cursor) {
// This checks the origin node and only allows members of root sets to update them.
node.remote_update_root_set(&self.identity, rs);
} else {
break;
}
}
}
}
}
}
}
}
} else {
// V1 nodes just append the endpoint to which they sent the packet, and we can still use that.
if let Ok(ok_hello_fixed_header_fields) = payload.read_struct::<v1::message_component_structs::OkHelloFixedHeaderFields>(&mut cursor) {
if hops == 0 {
if let Ok(reported_endpoint) = Endpoint::unmarshal(&payload, &mut cursor) {
#[cfg(debug_assertions)]
let reported_endpoint2 = reported_endpoint.clone();
if self.remote_node_info.write().reported_local_endpoints.insert(reported_endpoint, time_ticks).is_none() {
#[cfg(debug_assertions)]
debug_event!(si, "[vl1] {} reported new remote perspective, local endpoint: {}", self.identity.address.to_string(), reported_endpoint2.to_string());
}
debug_event!(
si,
"[vl1] {} reported new remote perspective, local endpoint: {}",
self.identity.address.to_string(),
reported_endpoint2.to_string()
);
}
}
}
@ -760,9 +736,9 @@ impl<SI: SystemInterface> Peer<SI> {
async fn handle_incoming_whois<PH: InnerProtocolInterface>(&self, si: &SI, ph: &PH, node: &Node<SI>, time_ticks: i64, message_id: MessageId, payload: &PacketBuffer) -> bool {
if node.this_node_is_root() || ph.should_communicate_with(&self.identity) {
let mut packet = PacketBuffer::new();
packet.set_size(packet_constants::HEADER_SIZE);
packet.set_size(v1::HEADER_SIZE);
{
let mut f: &mut message_component_structs::OkHeader = packet.append_struct_get_mut().unwrap();
let mut f: &mut v1::message_component_structs::OkHeader = packet.append_struct_get_mut().unwrap();
f.verb = verbs::VL1_OK;
f.in_re_verb = verbs::VL1_WHOIS;
f.in_re_message_id = message_id.to_ne_bytes();
@ -795,9 +771,9 @@ impl<SI: SystemInterface> Peer<SI> {
async fn handle_incoming_echo<PH: InnerProtocolInterface>(&self, si: &SI, ph: &PH, node: &Node<SI>, time_ticks: i64, message_id: MessageId, payload: &PacketBuffer) -> bool {
if ph.should_communicate_with(&self.identity) || node.is_peer_root(self) {
let mut packet = PacketBuffer::new();
packet.set_size(packet_constants::HEADER_SIZE);
packet.set_size(v1::HEADER_SIZE);
{
let mut f: &mut message_component_structs::OkHeader = packet.append_struct_get_mut().unwrap();
let mut f: &mut v1::message_component_structs::OkHeader = packet.append_struct_get_mut().unwrap();
f.verb = verbs::VL1_OK;
f.in_re_verb = verbs::VL1_ECHO;
f.in_re_message_id = message_id.to_ne_bytes();
@ -824,18 +800,18 @@ impl<SI: SystemInterface> Peer<SI> {
}
}
impl<SI: SystemInterface> Hash for Peer<SI> {
#[inline(always)]
fn hash<H: std::hash::Hasher>(&self, state: &mut H) {
state.write_u64(self.identity.address.into());
}
}
impl<SI: SystemInterface> PartialEq for Peer<SI> {
#[inline(always)]
fn eq(&self, other: &Self) -> bool {
self.canonical.eq(&other.canonical)
self.identity.fingerprint.eq(&other.identity.fingerprint)
}
}
impl<SI: SystemInterface> Eq for Peer<SI> {}
impl<SI: SystemInterface> Hash for Peer<SI> {
#[inline(always)]
fn hash<H: Hasher>(&self, state: &mut H) {
self.canonical.canonical_instance_id().hash(state);
}
}

168
network-hypervisor/src/vl1/protocol.rs
View file

@ -52,16 +52,16 @@ pub const PROTOCOL_VERSION: u8 = 20;
pub const PROTOCOL_VERSION_MIN: u8 = 11;
/// Buffer sized for ZeroTier packets.
pub type PacketBuffer = Buffer<{ packet_constants::SIZE_MAX }>;
pub type PacketBuffer = Buffer<{ v1::SIZE_MAX }>;
/// Factory type to supply to a new PacketBufferPool, used in PooledPacketBuffer and PacketBufferPool types.
pub type PacketBufferFactory = crate::util::buffer::PooledBufferFactory<{ crate::vl1::protocol::packet_constants::SIZE_MAX }>;
pub type PacketBufferFactory = crate::util::buffer::PooledBufferFactory<{ crate::vl1::protocol::v1::SIZE_MAX }>;
/// Packet buffer checked out of pool, automatically returns on drop.
pub type PooledPacketBuffer = crate::util::pool::Pooled<PacketBuffer, PacketBufferFactory>;
pub type PooledPacketBuffer = zerotier_utils::pool::Pooled<PacketBuffer, PacketBufferFactory>;
/// Source for instances of PacketBuffer
pub type PacketBufferPool = crate::util::pool::Pool<PacketBuffer, PacketBufferFactory>;
pub type PacketBufferPool = zerotier_utils::pool::Pool<PacketBuffer, PacketBufferFactory>;
/// 64-bit packet (outer) ID.
pub type PacketId = u64;
@ -115,7 +115,9 @@ pub const ADDRESS_RESERVED_PREFIX: u8 = 0xff;
/// Size of an identity fingerprint (SHA384)
pub const IDENTITY_FINGERPRINT_SIZE: usize = 48;
pub mod packet_constants {
pub mod v1 {
use super::*;
/// Size of packet header that lies outside the encryption envelope.
pub const HEADER_SIZE: usize = 27;
@ -192,9 +194,7 @@ pub mod packet_constants {
/// Header (outer) flag indicating that this packet has additional fragments.
pub const HEADER_FLAG_FRAGMENTED: u8 = 0x40;
}
pub mod security_constants {
/// Packet is not encrypted but contains a Poly1305 MAC of the plaintext.
/// Poly1305 is initialized with Salsa20/12 in the same manner as SALSA2012_POLY1305.
pub const CIPHER_NOCRYPT_POLY1305: u8 = 0x00;
@ -218,95 +218,29 @@ pub mod security_constants {
/// KBKDF usage label for the second AES-GMAC-SIV key.
pub const KBKDF_KEY_USAGE_LABEL_AES_GMAC_SIV_K1: u8 = b'1';
/// KBKDF usage label for AES-GCM session keys.
pub const KBKDF_KEY_USAGE_LABEL_AES_GCM_SESSION_KEY: u8 = b'g';
/// KBKDF usage label for AES-GCM session keys.
pub const KBKDF_KEY_USAGE_LABEL_AES_CTR_SESSION_KEY: u8 = b'c';
/// Try to re-key ephemeral keys after this time.
pub const EPHEMERAL_SECRET_REKEY_AFTER_TIME: i64 = 300000; // 5 minutes
/// Maximum number of times to use an ephemeral secret before trying to replace it.
pub const EPHEMERAL_SECRET_REKEY_AFTER_USES: usize = 536870912; // 1/4 the NIST/FIPS security bound of 2^31
/// Ephemeral secret reject after time.
pub const EPHEMERAL_SECRET_REJECT_AFTER_TIME: i64 = EPHEMERAL_SECRET_REKEY_AFTER_TIME * 2;
/// Ephemeral secret reject after uses.
pub const EPHEMERAL_SECRET_REJECT_AFTER_USES: usize = 2147483648; // NIST/FIPS security bound
}
pub mod session_metadata {
/// Random 128-bit ID generated at node startup, allows multiple instances to share an identity and be differentiated.
pub const INSTANCE_ID: &'static str = "i";
/// Endpoint from which HELLO was received, sent with OK(HELLO) if hops is zero.
pub const DIRECT_SOURCE: &'static str = "s";
/// Signed bundle of identity fingerprints through which this node can be reached.
pub const CARE_OF: &'static str = "c";
/// One or more root sets to which THIS node is a member. Included only if this is a root.
pub const ROOT_SET_UPDATES: &'static str = "r";
}
/// Maximum number of packet hops allowed by the protocol.
pub const PROTOCOL_MAX_HOPS: u8 = 7;
/// Maximum number of hops to allow.
pub const FORWARD_MAX_HOPS: u8 = 3;
/// Maximum difference between current message ID and OK/ERROR in-re message ID.
pub const PACKET_RESPONSE_COUNTER_DELTA_MAX: u64 = 4096;
/// Frequency for WHOIS retries
pub const WHOIS_RETRY_INTERVAL: i64 = 1000;
/// Maximum number of WHOIS retries
pub const WHOIS_RETRY_MAX: u16 = 3;
/// Maximum number of packets to queue up behind a WHOIS.
pub const WHOIS_MAX_WAITING_PACKETS: usize = 64;
/// Keepalive interval for paths in milliseconds.
pub const PATH_KEEPALIVE_INTERVAL: i64 = 20000;
/// Path object expiration time in milliseconds since last receive.
pub const PATH_EXPIRATION_TIME: i64 = (PATH_KEEPALIVE_INTERVAL * 2) + 10000;
/// How often to send HELLOs to roots, which is more often than normal peers.
pub const ROOT_HELLO_INTERVAL: i64 = PATH_KEEPALIVE_INTERVAL * 2;
/// How often to send HELLOs to roots when we are offline.
pub const ROOT_HELLO_SPAM_INTERVAL: i64 = 5000;
/// How often to send HELLOs to regular peers.
pub const PEER_HELLO_INTERVAL_MAX: i64 = 300000;
/// Timeout for path association with peers and for peers themselves.
pub const PEER_EXPIRATION_TIME: i64 = (PEER_HELLO_INTERVAL_MAX * 2) + 10000;
/// Proof of work difficulty (threshold) for identity generation.
pub const IDENTITY_POW_THRESHOLD: u8 = 17;
/// Attempt to compress a packet's payload with LZ4
///
/// If this returns true the destination buffer will contain a compressed packet. If false is
/// returned the contents of 'dest' are entirely undefined. This indicates that the data was not
/// compressable or some other error occurred.
pub fn compress_packet<const S: usize>(src: &[u8], dest: &mut Buffer<S>) -> bool {
if src.len() > (packet_constants::VERB_INDEX + 16) {
if src.len() > (VERB_INDEX + 16) {
let compressed_data_size = {
let d = unsafe { dest.entire_buffer_mut() };
d[..packet_constants::VERB_INDEX].copy_from_slice(&src[0..packet_constants::VERB_INDEX]);
d[packet_constants::VERB_INDEX] = src[packet_constants::VERB_INDEX] | packet_constants::VERB_FLAG_COMPRESSED;
lz4_flex::block::compress_into(&src[packet_constants::VERB_INDEX + 1..], &mut d[packet_constants::VERB_INDEX + 1..])
d[..VERB_INDEX].copy_from_slice(&src[..VERB_INDEX]);
d[VERB_INDEX] = src[VERB_INDEX] | VERB_FLAG_COMPRESSED;
lz4_flex::block::compress_into(&src[VERB_INDEX + 1..], &mut d[VERB_INDEX + 1..])
};
if compressed_data_size.is_ok() {
let compressed_data_size = compressed_data_size.unwrap();
if compressed_data_size > 0 && compressed_data_size < (src.len() - packet_constants::VERB_INDEX) {
unsafe { dest.set_size_unchecked(packet_constants::VERB_INDEX + 1 + compressed_data_size) };
if compressed_data_size > 0 && compressed_data_size < (src.len() - VERB_INDEX) {
unsafe { dest.set_size_unchecked(VERB_INDEX + 1 + compressed_data_size) };
return true;
}
}
@ -319,7 +253,7 @@ pub fn compress_packet<const S: usize>(src: &[u8], dest: &mut Buffer<S>) -> bool
/// This will panic if the buffer provided doesn't contain a proper header.
#[inline(always)]
pub fn set_packet_fragment_flag<const S: usize>(pkt: &mut Buffer<S>) {
pkt.as_bytes_mut()[packet_constants::FLAGS_FIELD_INDEX] |= packet_constants::HEADER_FLAG_FRAGMENTED;
pkt.as_bytes_mut()[FLAGS_FIELD_INDEX] |= HEADER_FLAG_FRAGMENTED;
}
/// ZeroTier unencrypted outer packet header
@ -344,30 +278,30 @@ impl PacketHeader {
#[inline(always)]
pub fn cipher(&self) -> u8 {
self.flags_cipher_hops & packet_constants::FLAGS_FIELD_MASK_CIPHER
self.flags_cipher_hops & FLAGS_FIELD_MASK_CIPHER
}
#[inline(always)]
pub fn hops(&self) -> u8 {
self.flags_cipher_hops & packet_constants::FLAGS_FIELD_MASK_HOPS
self.flags_cipher_hops & FLAGS_FIELD_MASK_HOPS
}
#[inline(always)]
pub fn increment_hops(&mut self) -> u8 {
let f = self.flags_cipher_hops;
let h = (f + 1) & packet_constants::FLAGS_FIELD_MASK_HOPS;
self.flags_cipher_hops = (f & packet_constants::FLAGS_FIELD_MASK_HIDE_HOPS) | h;
let h = (f + 1) & FLAGS_FIELD_MASK_HOPS;
self.flags_cipher_hops = (f & FLAGS_FIELD_MASK_HIDE_HOPS) | h;
h
}
#[inline(always)]
pub fn is_fragmented(&self) -> bool {
(self.flags_cipher_hops & packet_constants::HEADER_FLAG_FRAGMENTED) != 0
(self.flags_cipher_hops & HEADER_FLAG_FRAGMENTED) != 0
}
#[inline(always)]
pub fn as_bytes(&self) -> &[u8; packet_constants::HEADER_SIZE] {
unsafe { &*(self as *const Self).cast::<[u8; packet_constants::HEADER_SIZE]>() }
pub fn as_bytes(&self) -> &[u8; HEADER_SIZE] {
unsafe { &*(self as *const Self).cast::<[u8; HEADER_SIZE]>() }
}
#[inline(always)]
@ -384,7 +318,7 @@ pub fn get_packet_aad_bytes(destination: Address, source: Address, flags_cipher_
let mut id = unsafe { MaybeUninit::<[u8; 11]>::uninit().assume_init() };
id[0..5].copy_from_slice(&destination.to_bytes());
id[5..10].copy_from_slice(&source.to_bytes());
id[10] = flags_cipher_hops & packet_constants::FLAGS_FIELD_MASK_HIDE_HOPS;
id[10] = flags_cipher_hops & FLAGS_FIELD_MASK_HIDE_HOPS;
id
}
@ -412,7 +346,7 @@ impl FragmentHeader {
#[inline(always)]
pub fn is_fragment(&self) -> bool {
self.fragment_indicator == packet_constants::FRAGMENT_INDICATOR
self.fragment_indicator == FRAGMENT_INDICATOR
}
#[inline(always)]
@ -427,20 +361,20 @@ impl FragmentHeader {
#[inline(always)]
pub fn hops(&self) -> u8 {
self.reserved_hops & packet_constants::FLAGS_FIELD_MASK_HOPS
self.reserved_hops & FLAGS_FIELD_MASK_HOPS
}
#[inline(always)]
pub fn increment_hops(&mut self) -> u8 {
let f = self.reserved_hops;
let h = (f + 1) & packet_constants::FLAGS_FIELD_MASK_HOPS;
self.reserved_hops = (f & packet_constants::FLAGS_FIELD_MASK_HIDE_HOPS) | h;
let h = (f + 1) & FLAGS_FIELD_MASK_HOPS;
self.reserved_hops = (f & FLAGS_FIELD_MASK_HIDE_HOPS) | h;
h
}
#[inline(always)]
pub fn as_bytes(&self) -> &[u8; packet_constants::FRAGMENT_HEADER_SIZE] {
unsafe { &*(self as *const Self).cast::<[u8; packet_constants::FRAGMENT_HEADER_SIZE]>() }
pub fn as_bytes(&self) -> &[u8; FRAGMENT_HEADER_SIZE] {
unsafe { &*(self as *const Self).cast::<[u8; FRAGMENT_HEADER_SIZE]>() }
}
}
@ -484,6 +418,40 @@ pub(crate) mod message_component_structs {
pub version_revision: [u8; 2], // u16
}
}
}
/// Maximum difference between current message ID and OK/ERROR in-re message ID.
pub const PACKET_RESPONSE_COUNTER_DELTA_MAX: u64 = 4096;
/// Frequency for WHOIS retries
pub const WHOIS_RETRY_INTERVAL: i64 = 1000;
/// Maximum number of WHOIS retries
pub const WHOIS_RETRY_MAX: u16 = 3;
/// Maximum number of packets to queue up behind a WHOIS.
pub const WHOIS_MAX_WAITING_PACKETS: usize = 64;
/// Keepalive interval for paths in milliseconds.
pub const PATH_KEEPALIVE_INTERVAL: i64 = 20000;
/// Path object expiration time in milliseconds since last receive.
pub const PATH_EXPIRATION_TIME: i64 = (PATH_KEEPALIVE_INTERVAL * 2) + 10000;
/// How often to send HELLOs to roots, which is more often than normal peers.
pub const ROOT_HELLO_INTERVAL: i64 = PATH_KEEPALIVE_INTERVAL * 2;
/// How often to send HELLOs to roots when we are offline.
pub const ROOT_HELLO_SPAM_INTERVAL: i64 = 5000;
/// How often to send HELLOs to regular peers.
pub const PEER_HELLO_INTERVAL_MAX: i64 = 300000;
/// Timeout for path association with peers and for peers themselves.
pub const PEER_EXPIRATION_TIME: i64 = (PEER_HELLO_INTERVAL_MAX * 2) + 10000;
/// Proof of work difficulty (threshold) for identity generation.
pub const IDENTITY_POW_THRESHOLD: u8 = 17;
#[cfg(test)]
mod tests {
@ -493,15 +461,15 @@ mod tests {
#[test]
fn representation() {
assert_eq!(size_of::<message_component_structs::OkHeader>(), 10);
assert_eq!(size_of::<message_component_structs::ErrorHeader>(), 11);
assert_eq!(size_of::<PacketHeader>(), packet_constants::HEADER_SIZE);
assert_eq!(size_of::<FragmentHeader>(), packet_constants::FRAGMENT_HEADER_SIZE);
assert_eq!(size_of::<v1::message_component_structs::OkHeader>(), 10);
assert_eq!(size_of::<v1::message_component_structs::ErrorHeader>(), 11);
assert_eq!(size_of::<v1::PacketHeader>(), v1::HEADER_SIZE);
assert_eq!(size_of::<v1::FragmentHeader>(), v1::FRAGMENT_HEADER_SIZE);
let mut foo = [0_u8; 32];
unsafe {
(*foo.as_mut_ptr().cast::<PacketHeader>()).src[0] = 0xff;
assert_eq!((*foo.as_ptr().cast::<FragmentHeader>()).fragment_indicator, 0xff);
(*foo.as_mut_ptr().cast::<v1::PacketHeader>()).src[0] = 0xff;
assert_eq!((*foo.as_ptr().cast::<v1::FragmentHeader>()).fragment_indicator, 0xff);
}
}
}

2
network-hypervisor/src/vl1/rootset.rs
View file

@ -249,7 +249,7 @@ impl RootSet {
}
impl Marshalable for RootSet {
const MAX_MARSHAL_SIZE: usize = crate::vl1::protocol::packet_constants::SIZE_MAX;
const MAX_MARSHAL_SIZE: usize = crate::vl1::protocol::v1::SIZE_MAX;
#[inline(always)]
fn marshal<const BL: usize>(&self, buf: &mut Buffer<BL>) -> std::io::Result<()> {

13
network-hypervisor/src/vl1/symmetricsecret.rs
View file

@ -1,12 +1,13 @@
// (c) 2020-2022 ZeroTier, Inc. -- currently propritery pending actual release and licensing. See LICENSE.md.
use zerotier_core_crypto::aes_gmac_siv::AesGmacSiv;
use zerotier_core_crypto::kbkdf::*;
use zerotier_core_crypto::secret::Secret;
use zerotier_crypto::aes_gmac_siv::AesGmacSiv;
use zerotier_crypto::kbkdf::zt_kbkdf_hmac_sha384;
use zerotier_crypto::secret::Secret;
use crate::util::pool::{Pool, PoolFactory};
use crate::vl1::protocol::*;
use zerotier_utils::pool::{Pool, PoolFactory};
/// A symmetric secret key negotiated between peers.
///
/// This contains the key and several sub-keys and ciphers keyed with sub-keys.
@ -22,8 +23,8 @@ impl SymmetricSecret {
/// Create a new symmetric secret, deriving all sub-keys and such.
pub fn new(key: Secret<64>) -> SymmetricSecret {
let aes_factory = AesGmacSivPoolFactory(
zt_kbkdf_hmac_sha384(&key.0[..48], security_constants::KBKDF_KEY_USAGE_LABEL_AES_GMAC_SIV_K0).first_n_clone(),
zt_kbkdf_hmac_sha384(&key.0[..48], security_constants::KBKDF_KEY_USAGE_LABEL_AES_GMAC_SIV_K1).first_n_clone(),
zt_kbkdf_hmac_sha384(&key.0[..48], v1::KBKDF_KEY_USAGE_LABEL_AES_GMAC_SIV_K0).first_n_clone(),
zt_kbkdf_hmac_sha384(&key.0[..48], v1::KBKDF_KEY_USAGE_LABEL_AES_GMAC_SIV_K1).first_n_clone(),
);
SymmetricSecret { key, aes_gmac_siv: Pool::new(2, aes_factory) }
}

2
system-service/Cargo.toml
View file

@ -11,7 +11,7 @@ path = "src/main.rs"
[dependencies]
zerotier-network-hypervisor = { path = "../network-hypervisor" }
zerotier-core-crypto = { path = "../core-crypto" }
zerotier-crypto = { path = "../crypto" }
async-trait = "^0"
num-traits = "^0"
tokio = { version = "^1", features = ["fs", "io-util", "io-std", "net", "parking_lot", "process", "rt", "rt-multi-thread", "signal", "sync", "time"], default-features = false }

2
system-service/src/datadir.rs
View file

@ -9,7 +9,7 @@ use crate::utils::{read_limit, DEFAULT_FILE_IO_READ_LIMIT};
use parking_lot::{Mutex, RwLock};
use zerotier_core_crypto::random::next_u32_secure;
use zerotier_crypto::random::next_u32_secure;
use zerotier_network_hypervisor::vl1::Identity;
const AUTH_TOKEN_DEFAULT_LENGTH: usize = 48;

17
system-service/src/service.rs
View file

@ -11,7 +11,7 @@ use zerotier_network_hypervisor::vl1::*;
use zerotier_network_hypervisor::vl2::*;
use zerotier_network_hypervisor::*;
use zerotier_core_crypto::random;
use zerotier_crypto::random;
use tokio::time::Duration;
@ -91,7 +91,12 @@ impl ServiceImpl {
}
/// Called in udp_binding_task_main() to service a particular UDP port.
async fn update_udp_bindings_for_port(self: &Arc<Self>, port: u16, interface_prefix_blacklist: &Vec<String>, cidr_blacklist: &Vec<InetAddress>) -> Option<Vec<(LocalInterface, InetAddress, std::io::Error)>> {
async fn update_udp_bindings_for_port(
self: &Arc<Self>,
port: u16,
interface_prefix_blacklist: &Vec<String>,
cidr_blacklist: &Vec<InetAddress>,
) -> Option<Vec<(LocalInterface, InetAddress, std::io::Error)>> {
for ns in {
let mut udp_sockets_by_port = self.udp_sockets_by_port.write().await;
let bp = udp_sockets_by_port.entry(port).or_insert_with(|| BoundUdpPort::new(port));
@ -121,7 +126,8 @@ impl ServiceImpl {
let mut buf = core.get_packet_buffer();
if let Ok((bytes, source)) = socket.recv_from(unsafe { buf.entire_buffer_mut() }).await {
unsafe { buf.set_size_unchecked(bytes) };
core.handle_incoming_physical_packet(&self2, &Endpoint::IpUdp(InetAddress::from(source)), &local_socket, &interface, buf).await;
core.handle_incoming_physical_packet(&self2, &Endpoint::IpUdp(InetAddress::from(source)), &local_socket, &interface, buf)
.await;
} else {
break;
}
@ -137,7 +143,10 @@ impl ServiceImpl {
loop {
let config = self.data.config().await;
if let Some(errors) = self.update_udp_bindings_for_port(config.settings.primary_port, &config.settings.interface_prefix_blacklist, &config.settings.cidr_blacklist).await {
if let Some(errors) = self
.update_udp_bindings_for_port(config.settings.primary_port, &config.settings.interface_prefix_blacklist, &config.settings.cidr_blacklist)
.await
{
for e in errors.iter() {
println!("BIND ERROR: {} {} {}", e.0.to_string(), e.1.to_string(), e.2.to_string());
}

1
utils/Cargo.toml
View file

@ -6,3 +6,4 @@ name = "zerotier-utils"
version = "0.1.0"
[dependencies]
parking_lot = { version = "^0", features = [], default-features = false }

1
utils/src/lib.rs
View file

@ -2,5 +2,6 @@ pub mod arrayvec;
pub mod gatherarray;
pub mod hex;
pub mod memory;
pub mod pool;
pub mod ringbuffermap;
pub mod varint;

2
network-hypervisor/src/util/pool.rs → utils/src/pool.rs
View file

@ -165,7 +165,7 @@ mod tests {
use std::sync::Arc;
use std::time::Duration;
use crate::util::pool::*;
use super::*;
struct TestPoolFactory;