use crate::noise::{error::NoiseHandshakeError, stream::NoiseStream};
use aptos_config::{
config::{Peer, PeerRole, PeerSet},
network_id::NetworkContext,
};
use aptos_crypto::{noise, x25519};
use aptos_infallible::{duration_since_epoch, RwLock};
use aptos_logger::trace;
use aptos_types::PeerId;
use futures::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt};
use netcore::transport::ConnectionOrigin;
use short_hex_str::{AsShortHexStr, ShortHexStr};
use std::{collections::HashMap, convert::TryFrom as _, fmt::Debug, sync::Arc};
#[derive(Default)]
pub struct AntiReplayTimestamps(HashMap<x25519::PublicKey, u64>);
impl AntiReplayTimestamps {
pub const TIMESTAMP_SIZE: usize = 8;
pub fn now() -> [u8; Self::TIMESTAMP_SIZE] {
let now: u64 = duration_since_epoch().as_millis() as u64;
now.to_le_bytes()
}
pub fn is_replay(&self, pubkey: x25519::PublicKey, timestamp: u64) -> bool {
if let Some(last_timestamp) = self.0.get(&pubkey) {
×tamp <= last_timestamp
} else {
false
}
}
pub fn store_timestamp(&mut self, pubkey: x25519::PublicKey, timestamp: u64) {
self.0
.entry(pubkey)
.and_modify(|last_timestamp| *last_timestamp = timestamp)
.or_insert(timestamp);
}
}
pub enum HandshakeAuthMode {
Mutual {
anti_replay_timestamps: RwLock<AntiReplayTimestamps>,
trusted_peers: Arc<RwLock<PeerSet>>,
},
MaybeMutual(Arc<RwLock<PeerSet>>),
}
impl HandshakeAuthMode {
pub fn mutual(trusted_peers: Arc<RwLock<PeerSet>>) -> Self {
HandshakeAuthMode::Mutual {
anti_replay_timestamps: RwLock::new(AntiReplayTimestamps::default()),
trusted_peers,
}
}
pub fn maybe_mutual(trusted_peers: Arc<RwLock<PeerSet>>) -> Self {
HandshakeAuthMode::MaybeMutual(trusted_peers)
}
pub fn server_only() -> Self {
HandshakeAuthMode::maybe_mutual(Arc::new(RwLock::new(HashMap::default())))
}
fn anti_replay_timestamps(&self) -> Option<&RwLock<AntiReplayTimestamps>> {
match &self {
HandshakeAuthMode::Mutual {
anti_replay_timestamps,
..
} => Some(anti_replay_timestamps),
HandshakeAuthMode::MaybeMutual(_) => None,
}
}
}
pub struct NoiseUpgrader {
pub network_context: NetworkContext,
noise_config: noise::NoiseConfig,
auth_mode: HandshakeAuthMode,
}
impl NoiseUpgrader {
pub fn new(
network_context: NetworkContext,
key: x25519::PrivateKey,
auth_mode: HandshakeAuthMode,
) -> Self {
Self {
network_context,
noise_config: noise::NoiseConfig::new(key),
auth_mode,
}
}
#[allow(dead_code)]
pub async fn upgrade_with_noise<TSocket>(
&self,
socket: TSocket,
origin: ConnectionOrigin,
remote_public_key: Option<x25519::PublicKey>,
) -> Result<(x25519::PublicKey, NoiseStream<TSocket>), NoiseHandshakeError>
where
TSocket: AsyncRead + AsyncWrite + Debug + Unpin,
{
let socket = match origin {
ConnectionOrigin::Outbound => {
let remote_public_key = match remote_public_key {
Some(key) => key,
None if cfg!(any(test, feature = "fuzzing")) => unreachable!(),
None => return Err(NoiseHandshakeError::MissingServerPublicKey),
};
self.upgrade_outbound(socket, remote_public_key, AntiReplayTimestamps::now)
.await?
}
ConnectionOrigin::Inbound => {
let (socket, _peer_id, _) = self.upgrade_inbound(socket).await?;
socket
}
};
let remote_public_key = socket.get_remote_static();
Ok((remote_public_key, socket))
}
const PROLOGUE_SIZE: usize = PeerId::LENGTH + x25519::PUBLIC_KEY_SIZE;
const CLIENT_MESSAGE_SIZE: usize =
Self::PROLOGUE_SIZE + noise::handshake_init_msg_len(AntiReplayTimestamps::TIMESTAMP_SIZE);
const SERVER_MESSAGE_SIZE: usize = noise::handshake_resp_msg_len(0);
pub async fn upgrade_outbound<TSocket, F>(
&self,
mut socket: TSocket,
remote_public_key: x25519::PublicKey,
time_provider: F,
) -> Result<NoiseStream<TSocket>, NoiseHandshakeError>
where
TSocket: AsyncRead + AsyncWrite + Debug + Unpin,
F: Fn() -> [u8; AntiReplayTimestamps::TIMESTAMP_SIZE],
{
let mut client_message = [0; Self::CLIENT_MESSAGE_SIZE];
client_message[..PeerId::LENGTH].copy_from_slice(self.network_context.peer_id().as_ref());
client_message[PeerId::LENGTH..Self::PROLOGUE_SIZE]
.copy_from_slice(remote_public_key.as_slice());
let (prologue_msg, client_noise_msg) = client_message.split_at_mut(Self::PROLOGUE_SIZE);
let payload = time_provider();
let mut rng = rand::rngs::OsRng;
let initiator_state = self
.noise_config
.initiate_connection(
&mut rng,
prologue_msg,
remote_public_key,
Some(&payload),
client_noise_msg,
)
.map_err(NoiseHandshakeError::BuildClientHandshakeMessageFailed)?;
trace!(
"{} noise client: handshake write: remote_public_key: {}",
self.network_context,
remote_public_key,
);
socket
.write_all(&client_message)
.await
.map_err(NoiseHandshakeError::ClientWriteFailed)?;
socket
.flush()
.await
.map_err(NoiseHandshakeError::ClientFlushFailed)?;
trace!(
"{} noise client: handshake read: remote_public_key: {}",
self.network_context,
remote_public_key,
);
let mut server_response = [0u8; Self::SERVER_MESSAGE_SIZE];
socket
.read_exact(&mut server_response)
.await
.map_err(NoiseHandshakeError::ClientReadFailed)?;
trace!(
"{} noise client: handshake finalize: remote_public_key: {}",
self.network_context,
remote_public_key,
);
let (_, session) = self
.noise_config
.finalize_connection(initiator_state, &server_response)
.map_err(NoiseHandshakeError::ClientFinalizeFailed)?;
Ok(NoiseStream::new(socket, session))
}
pub async fn upgrade_inbound<TSocket>(
&self,
mut socket: TSocket,
) -> Result<(NoiseStream<TSocket>, PeerId, PeerRole), NoiseHandshakeError>
where
TSocket: AsyncRead + AsyncWrite + Debug + Unpin,
{
let mut client_message = [0; Self::CLIENT_MESSAGE_SIZE];
trace!("{} noise server: handshake read", self.network_context);
socket
.read_exact(&mut client_message)
.await
.map_err(NoiseHandshakeError::ServerReadFailed)?;
let (remote_peer_id, self_expected_public_key) =
client_message[..Self::PROLOGUE_SIZE].split_at(PeerId::LENGTH);
let remote_peer_id = PeerId::try_from(remote_peer_id)
.map_err(|_| NoiseHandshakeError::InvalidClientPeerId(hex::encode(remote_peer_id)))?;
let remote_peer_short = remote_peer_id.short_str();
if remote_peer_id == self.network_context.peer_id() {
return Err(NoiseHandshakeError::SelfDialDetected);
}
if self_expected_public_key != self.noise_config.public_key().as_slice() {
return Err(NoiseHandshakeError::ClientExpectingDifferentPubkey(
remote_peer_short,
hex::encode(self_expected_public_key),
));
}
let (prologue, client_init_message) = client_message.split_at(Self::PROLOGUE_SIZE);
let (remote_public_key, handshake_state, payload) = self
.noise_config
.parse_client_init_message(prologue, client_init_message)
.map_err(|err| NoiseHandshakeError::ServerParseClient(remote_peer_short, err))?;
let peer_role = match &self.auth_mode {
HandshakeAuthMode::Mutual { trusted_peers, .. } => {
match trusted_peers.read().get(&remote_peer_id) {
Some(peer) => {
Self::authenticate_inbound(remote_peer_short, peer, &remote_public_key)
}
None => Err(NoiseHandshakeError::UnauthenticatedClient(
remote_peer_short,
remote_peer_id,
)),
}
}
HandshakeAuthMode::MaybeMutual(trusted_peers) => {
match trusted_peers.read().get(&remote_peer_id) {
Some(peer) => {
Self::authenticate_inbound(remote_peer_short, peer, &remote_public_key)
}
None => {
let derived_remote_peer_id =
aptos_types::account_address::from_identity_public_key(
remote_public_key,
);
if derived_remote_peer_id != remote_peer_id {
Err(NoiseHandshakeError::ClientPeerIdMismatch(
remote_peer_short,
remote_peer_id,
derived_remote_peer_id,
))
} else {
Ok(PeerRole::Unknown)
}
}
}
}
}?;
if let Some(anti_replay_timestamps) = self.auth_mode.anti_replay_timestamps() {
if payload.len() != AntiReplayTimestamps::TIMESTAMP_SIZE {
return Err(NoiseHandshakeError::MissingAntiReplayTimestamp(
remote_peer_short,
));
}
let mut client_timestamp = [0u8; AntiReplayTimestamps::TIMESTAMP_SIZE];
client_timestamp.copy_from_slice(&payload);
let client_timestamp = u64::from_le_bytes(client_timestamp);
let mut anti_replay_timestamps = anti_replay_timestamps.write();
if anti_replay_timestamps.is_replay(remote_public_key, client_timestamp) {
return Err(NoiseHandshakeError::ServerReplayDetected(
remote_peer_short,
client_timestamp,
));
}
anti_replay_timestamps.store_timestamp(remote_public_key, client_timestamp);
}
let mut rng = rand::rngs::OsRng;
let mut server_response = [0u8; Self::SERVER_MESSAGE_SIZE];
let session = self
.noise_config
.respond_to_client(&mut rng, handshake_state, None, &mut server_response)
.map_err(|err| {
NoiseHandshakeError::BuildServerHandshakeMessageFailed(remote_peer_short, err)
})?;
trace!(
"{} noise server: handshake write: remote_peer_id: {}",
self.network_context,
remote_peer_short,
);
socket
.write_all(&server_response)
.await
.map_err(|err| NoiseHandshakeError::ServerWriteFailed(remote_peer_short, err))?;
trace!(
"{} noise server: handshake finalize: remote_peer_id: {}",
self.network_context,
remote_peer_short,
);
Ok((NoiseStream::new(socket, session), remote_peer_id, peer_role))
}
fn authenticate_inbound(
remote_peer_short: ShortHexStr,
peer: &Peer,
remote_public_key: &x25519::PublicKey,
) -> Result<PeerRole, NoiseHandshakeError> {
if !peer.keys.contains(remote_public_key) {
return Err(NoiseHandshakeError::UnauthenticatedClientPubkey(
remote_peer_short,
hex::encode(remote_public_key.as_slice()),
));
}
Ok(peer.role)
}
}
#[cfg(test)]
mod test {
use super::*;
use crate::testutils::fake_socket::ReadWriteTestSocket;
use aptos_config::config::{Peer, PeerRole};
use aptos_crypto::{test_utils::TEST_SEED, traits::Uniform as _};
use futures::{executor::block_on, future::join};
use memsocket::MemorySocket;
use rand::SeedableRng as _;
const TEST_SEED_2: [u8; 32] = [42; 32];
fn build_peers(
is_mutual_auth: bool,
) -> (
(NoiseUpgrader, x25519::PublicKey),
(NoiseUpgrader, x25519::PublicKey),
) {
let mut rng = ::rand::rngs::StdRng::from_seed(TEST_SEED);
let client_private_key = x25519::PrivateKey::generate(&mut rng);
let client_public_key = client_private_key.public_key();
let server_private_key = x25519::PrivateKey::generate(&mut rng);
let server_public_key = server_private_key.public_key();
let (client_auth, server_auth, client_peer_id, server_peer_id) = if is_mutual_auth {
let client_peer_id = PeerId::random();
let client_pubkey_set = [client_public_key].iter().copied().collect();
let server_peer_id = PeerId::random();
let server_pubkey_set = [server_public_key].iter().copied().collect();
let trusted_peers = Arc::new(RwLock::new(
vec![
(
client_peer_id,
Peer::new(Vec::new(), client_pubkey_set, PeerRole::Validator),
),
(
server_peer_id,
Peer::new(Vec::new(), server_pubkey_set, PeerRole::Validator),
),
]
.into_iter()
.collect(),
));
let client_auth = HandshakeAuthMode::mutual(trusted_peers.clone());
let server_auth = HandshakeAuthMode::mutual(trusted_peers);
(client_auth, server_auth, client_peer_id, server_peer_id)
} else {
let client_peer_id =
aptos_types::account_address::from_identity_public_key(client_public_key);
let server_peer_id =
aptos_types::account_address::from_identity_public_key(server_public_key);
(
HandshakeAuthMode::server_only(),
HandshakeAuthMode::server_only(),
client_peer_id,
server_peer_id,
)
};
let client = NoiseUpgrader::new(
NetworkContext::mock_with_peer_id(client_peer_id),
client_private_key,
client_auth,
);
let server = NoiseUpgrader::new(
NetworkContext::mock_with_peer_id(server_peer_id),
server_private_key,
server_auth,
);
((client, client_public_key), (server, server_public_key))
}
fn perform_handshake(
client: &NoiseUpgrader,
server: &NoiseUpgrader,
server_public_key: x25519::PublicKey,
) -> (
Result<NoiseStream<MemorySocket>, NoiseHandshakeError>,
Result<(NoiseStream<MemorySocket>, PeerId, PeerRole), NoiseHandshakeError>,
) {
let (dialer_socket, listener_socket) = MemorySocket::new_pair();
block_on(join(
client.upgrade_outbound(dialer_socket, server_public_key, AntiReplayTimestamps::now),
server.upgrade_inbound(listener_socket),
))
}
fn bad_timestamp(value: u64) -> impl Fn() -> [u8; AntiReplayTimestamps::TIMESTAMP_SIZE] {
move || value.to_le_bytes()
}
#[test]
fn test_timestamp_replay() {
let ((client, _), (server, server_public_key)) =
build_peers(true );
let (dialer_socket, listener_socket) = MemorySocket::new_pair();
let (client_session, server_session) = block_on(join(
client.upgrade_outbound(dialer_socket, server_public_key, bad_timestamp(1)),
server.upgrade_inbound(listener_socket),
));
client_session.unwrap();
server_session.unwrap();
let (dialer_socket, listener_socket) = MemorySocket::new_pair();
let (client_session, server_session) = block_on(join(
client.upgrade_outbound(dialer_socket, server_public_key, bad_timestamp(0)),
server.upgrade_inbound(listener_socket),
));
client_session.unwrap_err();
server_session.unwrap_err();
let (dialer_socket, listener_socket) = MemorySocket::new_pair();
let (client_session, server_session) = block_on(join(
client.upgrade_outbound(dialer_socket, server_public_key, bad_timestamp(1)),
server.upgrade_inbound(listener_socket),
));
client_session.unwrap_err();
server_session.unwrap_err();
let (dialer_socket, listener_socket) = MemorySocket::new_pair();
let (client_session, server_session) = block_on(join(
client.upgrade_outbound(dialer_socket, server_public_key, bad_timestamp(2)),
server.upgrade_inbound(listener_socket),
));
client_session.unwrap();
server_session.unwrap();
}
fn test_handshake_success(is_mutual_auth: bool) {
let ((client, client_public_key), (server, server_public_key)) =
build_peers(is_mutual_auth);
let (client_res, server_res) = perform_handshake(&client, &server, server_public_key);
let client_stream = client_res.unwrap();
let (server_stream, _, _) = server_res.unwrap();
assert_eq!(client_stream.get_remote_static(), server_public_key);
assert_eq!(server_stream.get_remote_static(), client_public_key);
}
#[test]
fn test_handshake_success_server_only_auth() {
test_handshake_success(false );
}
#[test]
fn test_handshake_success_mutual_auth() {
test_handshake_success(true );
}
fn test_handshake_self_fails(is_mutual_auth: bool) {
let (_, (server, server_public_key)) = build_peers(is_mutual_auth);
let (client_res, server_res) = perform_handshake(&server, &server, server_public_key);
client_res.unwrap_err();
server_res.unwrap_err();
}
#[test]
fn test_handshake_self_fails_server_only_auth() {
test_handshake_self_fails(false );
}
#[test]
fn test_handshake_self_fails_mutual_auth() {
test_handshake_self_fails(true );
}
#[test]
fn test_handshake_unauthed_keypair_fails_mutual_auth() {
let mut rng = ::rand::rngs::StdRng::from_seed(TEST_SEED_2);
let client_private_key = x25519::PrivateKey::generate(&mut rng);
let ((mut client, _), (server, server_public_key)) =
build_peers(true );
client.noise_config = noise::NoiseConfig::new(client_private_key);
let (client_res, server_res) = perform_handshake(&client, &server, server_public_key);
client_res.unwrap_err();
server_res.unwrap_err();
}
#[test]
fn test_handshake_unauthed_peerid_fails_mutual_auth() {
let mut rng = ::rand::rngs::StdRng::from_seed(TEST_SEED_2);
let client_private_key = x25519::PrivateKey::generate(&mut rng);
let client_peer_id = PeerId::random();
let client = NoiseUpgrader::new(
NetworkContext::mock_with_peer_id(client_peer_id),
client_private_key,
HandshakeAuthMode::mutual(Arc::new(RwLock::new(HashMap::new()))),
);
let (_, (server, server_public_key)) = build_peers(true );
let (client_res, server_res) = perform_handshake(&client, &server, server_public_key);
client_res.unwrap_err();
server_res.unwrap_err();
}
#[test]
fn test_handshake_client_peerid_mismatch_fails_server_only_auth() {
::aptos_logger::Logger::init_for_testing();
let ((mut client, _), (server, server_public_key)) =
build_peers(false );
client.network_context = NetworkContext::mock_with_peer_id(PeerId::random());
let (client_res, server_res) = perform_handshake(&client, &server, server_public_key);
client_res.unwrap_err();
server_res.unwrap_err();
}
#[test]
fn test_handshake_fragmented_reads() {
let (mut dialer_socket, mut listener_socket) = ReadWriteTestSocket::new_pair();
dialer_socket.set_fragmented_read();
listener_socket.set_fragmented_read();
let ((client, _client_public_key), (server, server_public_key)) = build_peers(false);
let (client_session, server_session) = block_on(join(
client.upgrade_outbound(dialer_socket, server_public_key, AntiReplayTimestamps::now),
server.upgrade_inbound(listener_socket),
));
client_session.unwrap();
server_session.unwrap();
}
}