GRIN/nym
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

add noise lib

Browse Source
This commit is contained in:
Simon Wicky
2025-05-20 16:03:47 +02:00
committed by Georgio Nicolas
parent 0de4aea77b
commit 764639971c
9 changed files with 368 additions and 244 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.lock
Generated
+6 -8
View File
@@ -6777,28 +6777,26 @@ dependencies = [
name = "nym-noise"
version = "0.1.0"
dependencies = [
"anyhow",
"arc-swap",
"bytes",
"futures",
"nym-crypto 0.4.0",
"log",
"nym-crypto",
"nym-noise-keys",
"pin-project",
"rand_chacha 0.3.1",
"sha2 0.10.9",
"serde",
"sha2 0.10.8",
"snow",
"strum 0.26.3",
"thiserror 2.0.12",
"tokio",
"tokio-util",
"tracing",
]
[[package]]
name = "nym-noise-keys"
version = "0.1.0"
dependencies = [
"nym-crypto 0.4.0",
"nym-crypto",
"schemars",
"serde",
"utoipa",
@@ -10097,7 +10095,7 @@ dependencies = [
"curve25519-dalek",
"rand_core 0.6.4",
"rustc_version 0.4.1",
"sha2 0.10.9",
"sha2 0.10.8",
"subtle 2.6.1",
]
common/nymnoise/Cargo.toml
+2 -12
View File
@@ -9,11 +9,11 @@ license.workspace = true
arc-swap = { workspace = true }
bytes = { workspace = true }
futures = { workspace = true }
tracing = { workspace = true }
log = { workspace = true }
pin-project = { workspace = true }
serde = { workspace = true, features = ["derive"] }
sha2 = { workspace = true }
snow = { workspace = true }
strum = { workspace = true, features = ["derive"] }
thiserror = { workspace = true }
tokio = { workspace = true, features = ["net", "io-util", "time"] }
tokio-util = { workspace = true, features = ["codec"] }
@@ -21,13 +21,3 @@ tokio-util = { workspace = true, features = ["codec"] }
# internal
nym-crypto = { path = "../crypto" }
nym-noise-keys = { path = "keys" }
[dev-dependencies]
anyhow = { workspace = true }
tokio = { workspace = true, features = ["full"] }
rand_chacha = { workspace = true }
nym-crypto = { path = "../crypto", features = ["rand"] }
[lints]
workspace = true
common/nymnoise/keys/Cargo.toml
+1 -4
View File
@@ -11,7 +11,4 @@ serde = { workspace = true, features = ["derive"] }
utoipa = { workspace = true }
# internal
nym-crypto = { path = "../../crypto", features = ["asymmetric", "serde"] }
[lints]
workspace = true
nym-crypto = { path = "../../crypto" }
common/nymnoise/keys/src/lib.rs
+6 -9
View File
@@ -5,28 +5,25 @@ use nym_crypto::asymmetric::x25519;
use nym_crypto::asymmetric::x25519::serde_helpers::bs58_x25519_pubkey;
use serde::{Deserialize, Serialize};
#[derive(Copy, Clone, Debug, Serialize, Deserialize, PartialEq)]
#[derive(Copy, Clone, Debug, Serialize, Deserialize)]
#[serde(from = "u8", into = "u8")]
pub enum NoiseVersion {
V1,
Unknown(u8), //Implies a newer version we don't know
V1 = 1,
Unknown,
}
impl From<u8> for NoiseVersion {
fn from(value: u8) -> Self {
match value {
1 => NoiseVersion::V1,
other => NoiseVersion::Unknown(other),
_ => NoiseVersion::Unknown,
}
}
}
impl From<NoiseVersion> for u8 {
fn from(version: NoiseVersion) -> Self {
match version {
NoiseVersion::V1 => 1,
NoiseVersion::Unknown(other) => other,
}
version as u8
}
}
@@ -34,7 +31,7 @@ impl From<NoiseVersion> for u8 {
pub struct VersionedNoiseKey {
#[schemars(with = "u8")]
#[schema(value_type = u8)]
pub supported_version: NoiseVersion,
pub version: NoiseVersion,
#[schemars(with = "String")]
#[serde(with = "bs58_x25519_pubkey")]
common/nymnoise/src/config.rs
+22 -61
View File
@@ -1,39 +1,31 @@
// Copyright 2025 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: Apache-2.0
// SPDX-License-Identifier: GPL-3.0-only
use std::{
collections::HashMap,
net::{IpAddr, SocketAddr},
sync::Arc,
time::Duration,
};
use arc_swap::ArcSwap;
use nym_crypto::asymmetric::x25519;
use nym_noise_keys::{NoiseVersion, VersionedNoiseKey};
use snow::params::NoiseParams;
use strum::{EnumIter, FromRepr};
#[derive(Default, Debug, Clone, Copy, EnumIter, FromRepr, Eq, PartialEq)]
#[repr(u8)]
#[non_exhaustive]
#[derive(Default, Debug, Clone, Copy)]
pub enum NoisePattern {
#[default]
XKpsk3 = 1,
IKpsk2 = 2,
XKpsk3,
IKpsk2,
}
impl NoisePattern {
pub(crate) const fn as_str(&self) -> &'static str {
pub(crate) fn as_str(&self) -> &'static str {
match self {
Self::XKpsk3 => "Noise_XKpsk3_25519_AESGCM_SHA256",
Self::IKpsk2 => "Noise_IKpsk2_25519_ChaChaPoly_BLAKE2s", //Wireguard handshake (not exactly though)
}
}
// SAFETY: we have tests to ensure that hardcoded pattern are correct
#[allow(clippy::unwrap_used)]
pub(crate) fn psk_position(&self) -> u8 {
//automatic parsing, works for correct pattern, more convenient
match self.as_str().find("psk") {
@@ -41,16 +33,11 @@ impl NoisePattern {
let psk_index = n + 3;
let psk_char = self.as_str().chars().nth(psk_index).unwrap();
psk_char.to_string().parse().unwrap()
//if this fails, it means hardcoded pattern are wrong
}
None => 0,
}
}
// SAFETY : we have tests to ensure that hardcoded pattern are correct
#[allow(clippy::unwrap_used)]
pub(crate) fn as_noise_params(&self) -> NoiseParams {
self.as_str().parse().unwrap()
}
}
#[derive(Debug, Default)]
@@ -81,7 +68,7 @@ impl NoiseNetworkView {
pub fn swap_view(&self, new: HashMap<SocketAddr, VersionedNoiseKey>) {
let noise_support = new
.iter()
.map(|(s_addr, key)| (s_addr.ip(), key.supported_version))
.map(|(s_addr, key)| (s_addr.ip(), key.version))
.collect::<HashMap<_, _>>();
self.keys.inner.store(Arc::new(new));
self.support.inner.store(Arc::new(noise_support));
@@ -94,22 +81,16 @@ pub struct NoiseConfig {
pub(crate) local_key: Arc<x25519::KeyPair>,
pub(crate) pattern: NoisePattern,
pub(crate) timeout: Duration,
pub(crate) unsafe_disabled: bool, // allows for nodes to not attempt to do a noise handshake, VERY UNSAFE, FOR DEBUG PURPOSE ONLY
}
impl NoiseConfig {
pub fn new(
noise_key: Arc<x25519::KeyPair>,
network: NoiseNetworkView,
timeout: Duration,
) -> Self {
pub fn new(noise_key: Arc<x25519::KeyPair>, network: NoiseNetworkView) -> Self {
NoiseConfig {
network,
local_key: noise_key,
pattern: Default::default(),
timeout,
unsafe_disabled: false,
}
}
@@ -133,39 +114,19 @@ impl NoiseConfig {
// Only for phased update
//SW This can lead to some troubles if two nodes shares the same IP and one support Noise but not the other. This in only for the progressive update though and there is no workaround
pub(crate) fn get_noise_support(&self, ip_addr: IpAddr) -> Option<NoiseVersion> {
let plain_ip_support = self.network.support.inner.load().get(&ip_addr).copied();
// SW default bind address being [::]:1789, it can happen that a responder sees the ipv6-mapped address of the initiator, this check for that
let canonical_ip = &ip_addr.to_canonical();
let canonical_ip_support = self.network.support.inner.load().get(canonical_ip).copied();
plain_ip_support.or(canonical_ip_support)
}
}
#[cfg(test)]
mod tests {
use snow::params::NoiseParams;
use super::NoisePattern;
use std::str::FromStr;
use strum::IntoEnumIterator;
// The goal of these is to make sure every NoisePatterns are correct and unwrap can be used on them
#[test]
fn noise_patterns_are_valid() {
for pattern in NoisePattern::iter() {
assert!(NoiseParams::from_str(pattern.as_str()).is_ok())
}
}
#[test]
fn noise_patterns_psk_position_is_valid() {
for pattern in NoisePattern::iter() {
match pattern {
NoisePattern::XKpsk3 => assert_eq!(pattern.psk_position(), 3),
NoisePattern::IKpsk2 => assert_eq!(pattern.psk_position(), 2),
}
}
self.network
.support
.inner
.load()
.get(&ip_addr)
.copied()
.or_else(|| {
self.network
.support
.inner
.load()
.get(&ip_addr.to_canonical()) // SW default bind address being [::]:1789, it can happen that a responder sees the ipv6-mapped address of the initiator, this check for that
.copied()
})
}
}
common/nymnoise/src/connection.rs
+24 -19
View File
@@ -1,24 +1,32 @@
// Copyright 2025 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: Apache-2.0
// Copyright 2024 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: GPL-3.0-only
use std::io;
use pin_project::pin_project;
use tokio::io::{AsyncRead, AsyncWrite};
use tokio::{
io::{AsyncRead, AsyncWrite},
net::TcpStream,
};
use crate::stream::NoiseStream;
//SW once plain TCP support is dropped, this whole enum can be dropped, and we can only propagate NoiseStream
#[pin_project(project = ConnectionProj)]
pub enum Connection<C> {
Raw(#[pin] C),
Noise(#[pin] Box<NoiseStream<C>>),
pub enum Connection {
Tcp(#[pin] TcpStream),
Noise(#[pin] NoiseStream),
}
impl<C> AsyncRead for Connection<C>
where
C: AsyncRead + AsyncWrite + Unpin,
{
impl Connection {
pub fn peer_addr(&self) -> Result<std::net::SocketAddr, io::Error> {
match self {
Self::Noise(stream) => stream.peer_addr(),
Self::Tcp(stream) => stream.peer_addr(),
}
}
}
impl AsyncRead for Connection {
fn poll_read(
self: std::pin::Pin<&mut Self>,
cx: &mut std::task::Context<'_>,
@@ -26,15 +34,12 @@ where
) -> std::task::Poll<io::Result<()>> {
match self.project() {
ConnectionProj::Noise(stream) => stream.poll_read(cx, buf),
ConnectionProj::Raw(stream) => stream.poll_read(cx, buf),
ConnectionProj::Tcp(stream) => stream.poll_read(cx, buf),
}
}
}
impl<C> AsyncWrite for Connection<C>
where
C: AsyncWrite + AsyncRead + Unpin,
{
impl AsyncWrite for Connection {
fn poll_write(
self: std::pin::Pin<&mut Self>,
cx: &mut std::task::Context<'_>,
@@ -42,7 +47,7 @@ where
) -> std::task::Poll<Result<usize, io::Error>> {
match self.project() {
ConnectionProj::Noise(stream) => stream.poll_write(cx, buf),
ConnectionProj::Raw(stream) => stream.poll_write(cx, buf),
ConnectionProj::Tcp(stream) => stream.poll_write(cx, buf),
}
}
fn poll_flush(
@@ -51,7 +56,7 @@ where
) -> std::task::Poll<Result<(), io::Error>> {
match self.project() {
ConnectionProj::Noise(stream) => stream.poll_flush(cx),
ConnectionProj::Raw(stream) => stream.poll_flush(cx),
ConnectionProj::Tcp(stream) => stream.poll_flush(cx),
}
}
@@ -61,7 +66,7 @@ where
) -> std::task::Poll<Result<(), io::Error>> {
match self.project() {
ConnectionProj::Noise(stream) => stream.poll_shutdown(cx),
ConnectionProj::Raw(stream) => stream.poll_shutdown(cx),
ConnectionProj::Tcp(stream) => stream.poll_shutdown(cx),
}
}
}
common/nymnoise/src/error.rs
+6 -61
View File
@@ -1,7 +1,6 @@
// Copyright 2025 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: Apache-2.0
// Copyright 2023 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: GPL-3.0-only
use nym_noise_keys::NoiseVersion;
use snow::Error;
use std::io;
use thiserror::Error;
@@ -11,10 +10,10 @@ pub enum NoiseError {
#[error("encountered a Noise decryption error")]
DecryptionError,
#[error("encountered a Noise Protocol error: {0}")]
#[error("encountered a Noise Protocol error - {0}")]
ProtocolError(Error),
#[error("encountered an IO error: {0}")]
#[error("encountered an IO error - {0}")]
IoError(#[from] io::Error),
#[error("Incorrect state")]
@@ -23,62 +22,8 @@ pub enum NoiseError {
#[error("Handshake did not complete")]
HandshakeError,
#[error("unknown noise version (encoded value: {encoded})")]
UnknownVersion { encoded: u8 },
#[error("unknown noise pattern (encoded value: {encoded})")]
UnknownPattern { encoded: u8 },
#[error("unknown noise message type (encoded value: {encoded})")]
UnknownMessageType { encoded: u8 },
#[error("failed to generate psk for requested version {noise_version}")]
PskGenerationFailure { noise_version: u8 },
#[error("noise initiator attempted to use version v{noise_version} of the protocol - we don't know how to handle it")]
UnknownVersionHandshake { noise_version: u8 },
#[error("noise initiator attempted to use an unexpected noise pattern. we're configured for {configured} while it requested {received}")]
UnexpectedNoisePattern {
configured: &'static str,
received: &'static str,
},
#[error("handshake version has unexpectedly changed. initial was {initial:?} and received {received:?}")]
UnexpectedHandshakeVersion {
initial: NoiseVersion,
received: NoiseVersion,
},
#[error("data packet version has unexpectedly changed. initial was {initial:?} and received {received:?}")]
UnexpectedDataVersion {
initial: NoiseVersion,
received: NoiseVersion,
},
#[error("received a non-handshake message during noise handshake")]
NonHandshakeMessageReceived,
#[error("received a non-data message post noise handshake")]
NonDataMessageReceived,
#[error("handshake message exceeded maximum size (got {size} bytes)")]
HandshakeTooBig { size: usize },
#[error("noise message exceeded maximum size (got {size} bytes)")]
DataTooBig { size: usize },
#[error("Handshake timeout")]
HandshakeTimeout(#[from] tokio::time::error::Elapsed),
}
impl NoiseError {
pub(crate) fn naive_to_io_error(self) -> std::io::Error {
match self {
NoiseError::IoError(err) => err,
other => std::io::Error::other(other),
}
}
#[error("Unknown noise version")]
UnknownVersion,
}
impl From<Error> for NoiseError {
common/nymnoise/src/lib.rs
+110 -70
View File
@@ -1,63 +1,59 @@
// Copyright 2025 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: Apache-2.0
// Copyright 2023 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: GPL-3.0-only
use crate::config::{NoiseConfig, NoisePattern};
use crate::connection::Connection;
use crate::error::NoiseError;
use crate::stream::NoiseStream;
use log::*;
use nym_crypto::asymmetric::x25519;
use nym_noise_keys::NoiseVersion;
use snow::error::Prerequisite;
use snow::Error;
use sha2::{Digest, Sha256};
use snow::{error::Prerequisite, Builder, Error};
use tokio::net::TcpStream;
use tracing::{error, warn};
pub mod config;
pub mod connection;
pub mod error;
pub mod stream;
use crate::config::NoiseConfig;
use crate::connection::Connection;
use crate::error::NoiseError;
use crate::stream::NoiseStreamBuilder;
const NOISE_PSK_PREFIX: &[u8] = b"NYMTECH_NOISE_dQw4w9WgXcQ";
pub const LATEST_NOISE_VERSION: NoiseVersion = NoiseVersion::V1;
pub const NOISE_VERSION: NoiseVersion = NoiseVersion::V1;
// TODO: this should be behind some trait because presumably, depending on the version,
// other arguments would be needed
mod psk_gen {
use crate::error::NoiseError;
use crate::stream::Psk;
use crate::NOISE_PSK_PREFIX;
use nym_crypto::asymmetric::x25519;
use nym_noise_keys::NoiseVersion;
use sha2::{Digest, Sha256};
async fn upgrade_noise_initiator_v1(
conn: TcpStream,
pattern: NoisePattern,
local_private_key: &x25519::PrivateKey,
remote_pub_key: &x25519::PublicKey,
) -> Result<Connection, NoiseError> {
trace!("Perform Noise Handshake, initiator side");
pub(crate) fn generate_psk(
responder_pub_key: x25519::PublicKey,
version: NoiseVersion,
) -> Result<Psk, NoiseError> {
match version {
NoiseVersion::V1 => Ok(generate_psk_v1(responder_pub_key)),
NoiseVersion::Unknown(noise_version) => {
Err(NoiseError::PskGenerationFailure { noise_version })
}
}
}
let secret = [
NOISE_PSK_PREFIX.to_vec(),
remote_pub_key.to_bytes().to_vec(),
]
.concat();
let secret_hash = Sha256::digest(secret);
fn generate_psk_v1(responder_pub_key: x25519::PublicKey) -> [u8; 32] {
let mut hasher = Sha256::new();
hasher.update(NOISE_PSK_PREFIX);
hasher.update(responder_pub_key.to_bytes());
hasher.finalize().into()
}
let handshake = Builder::new(pattern.as_str().parse()?)
.local_private_key(&local_private_key.to_bytes())
.remote_public_key(&remote_pub_key.to_bytes())
.psk(pattern.psk_position(), &secret_hash)
.build_initiator()?;
let noise_stream = NoiseStream::new(conn, handshake);
Ok(Connection::Noise(noise_stream.perform_handshake().await?))
}
pub async fn upgrade_noise_initiator(
conn: TcpStream,
config: &NoiseConfig,
) -> Result<Connection<TcpStream>, NoiseError> {
) -> Result<Connection, NoiseError> {
if config.unsafe_disabled {
warn!("Noise is disabled in the config. Not attempting any handshake");
return Ok(Connection::Raw(conn));
return Ok(Connection::Tcp(conn));
}
//Get init material
@@ -66,34 +62,67 @@ pub async fn upgrade_noise_initiator(
Error::Prereq(Prerequisite::RemotePublicKey)
})?;
let Some(key) = config.get_noise_key(&responder_addr) else {
warn!("{responder_addr} can't speak Noise yet, falling back to TCP");
return Ok(Connection::Raw(conn));
};
let handshake_version = match key.supported_version {
NoiseVersion::V1 => NoiseVersion::V1,
// We're talking to a more recent node, but we can't adapt. Let's try to do our best and if it fails, it fails.
// If that node sees we're older, it will try to adapt too.
NoiseVersion::Unknown(version) => {
warn!("{responder_addr} is announcing an v{version} version of Noise that we don't know how to parse, we will attempt to downgrade to our current highest supported version");
LATEST_NOISE_VERSION
match config.get_noise_key(&responder_addr) {
Some(key) => match key.version {
NoiseVersion::V1 => {
upgrade_noise_initiator_v1(
conn,
config.pattern,
config.local_key.private_key(),
&key.x25519_pubkey,
)
.await
}
NoiseVersion::Unknown => {
error!(
"{:?} is announcing an unknown version of Noise",
responder_addr
);
Err(NoiseError::UnknownVersion)
}
},
None => {
warn!(
"{:?} can't speak Noise yet, falling back to TCP",
responder_addr
);
Ok(Connection::Tcp(conn))
}
};
NoiseStreamBuilder::new(conn)
.perform_initiator_handshake(config, handshake_version, key.x25519_pubkey)
.await
.map(|stream| Connection::Noise(Box::new(stream)))
}
}
async fn upgrade_noise_responder_v1(
conn: TcpStream,
pattern: NoisePattern,
local_public_key: &x25519::PublicKey,
local_private_key: &x25519::PrivateKey,
) -> Result<Connection, NoiseError> {
trace!("Perform Noise Handshake, responder side");
let secret = [
NOISE_PSK_PREFIX.to_vec(),
local_public_key.to_bytes().to_vec(),
]
.concat();
let secret_hash = Sha256::digest(secret);
let handshake = Builder::new(pattern.as_str().parse()?)
.local_private_key(&local_private_key.to_bytes())
.psk(pattern.psk_position(), &secret_hash)
.build_responder()?;
let noise_stream = NoiseStream::new(conn, handshake);
Ok(Connection::Noise(noise_stream.perform_handshake().await?))
}
pub async fn upgrade_noise_responder(
conn: TcpStream,
config: &NoiseConfig,
) -> Result<Connection<TcpStream>, NoiseError> {
) -> Result<Connection, NoiseError> {
if config.unsafe_disabled {
warn!("Noise is disabled in the config. Not attempting any handshake");
return Ok(Connection::Raw(conn));
return Ok(Connection::Tcp(conn));
}
//Get init material
@@ -105,14 +134,25 @@ pub async fn upgrade_noise_responder(
}
};
// if responder doesn't announce noise support, we fallback to tcp
if config.get_noise_support(initiator_addr.ip()).is_none() {
warn!("{initiator_addr} can't speak Noise yet, falling back to TCP",);
return Ok(Connection::Raw(conn));
};
NoiseStreamBuilder::new(conn)
.perform_responder_handshake(config)
.await
.map(|stream| Connection::Noise(Box::new(stream)))
// Port is random and we just need the support info
match config.get_noise_support(initiator_addr.ip()) {
None => {
warn!(
"{:?} can't speak Noise yet, falling back to TCP",
initiator_addr
);
Ok(Connection::Tcp(conn))
}
//responder's info on version is shaky, so initiator has to adapt. This behavior can change in the future
Some(_) => {
//Existing node supporting Noise
upgrade_noise_responder_v1(
conn,
config.pattern,
config.local_key.public_key(),
config.local_key.private_key(),
)
.await
}
}
}
common/nymnoise/src/stream.rs
+191
View File
@@ -0,0 +1,191 @@
// Copyright 2023 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: GPL-3.0-only
use crate::error::NoiseError;
use bytes::BytesMut;
use futures::{Sink, SinkExt, Stream, StreamExt};
use pin_project::pin_project;
use snow::{HandshakeState, TransportState};
use std::cmp::min;
use std::collections::VecDeque;
use std::io;
use std::pin::Pin;
use std::task::Poll;
use tokio::{
io::{AsyncRead, AsyncWrite, ReadBuf},
net::TcpStream,
};
use tokio_util::codec::{Framed, LengthDelimitedCodec};
const MAXMSGLEN: usize = 65535;
const TAGLEN: usize = 16;
/// Wrapper around a TcpStream
#[pin_project]
pub struct NoiseStream {
#[pin]
inner_stream: Framed<TcpStream, LengthDelimitedCodec>,
handshake: Option<HandshakeState>,
noise: Option<TransportState>,
dec_buffer: VecDeque<u8>,
}
impl NoiseStream {
pub(crate) fn new(inner_stream: TcpStream, handshake: HandshakeState) -> NoiseStream {
NoiseStream {
inner_stream: LengthDelimitedCodec::builder()
.length_field_type::<u16>()
.new_framed(inner_stream),
handshake: Some(handshake),
noise: None,
dec_buffer: VecDeque::with_capacity(MAXMSGLEN),
}
}
pub(crate) async fn perform_handshake(mut self) -> Result<Self, NoiseError> {
//Check if we are in the correct state
let Some(mut handshake) = self.handshake else {
return Err(NoiseError::IncorrectStateError);
};
self.handshake = None;
while !handshake.is_handshake_finished() {
if handshake.is_my_turn() {
self.send_handshake_msg(&mut handshake).await?;
} else {
self.recv_handshake_msg(&mut handshake).await?;
}
}
self.noise = Some(handshake.into_transport_mode()?);
Ok(self)
}
async fn send_handshake_msg(
&mut self,
handshake: &mut HandshakeState,
) -> Result<(), NoiseError> {
let mut buf = BytesMut::zeroed(MAXMSGLEN + TAGLEN);
let len = handshake.write_message(&[], &mut buf)?;
buf.truncate(len);
self.inner_stream.send(buf.into()).await?;
Ok(())
}
async fn recv_handshake_msg(
&mut self,
handshake: &mut HandshakeState,
) -> Result<(), NoiseError> {
match self.inner_stream.next().await {
Some(Ok(msg)) => {
let mut buf = vec![0u8; MAXMSGLEN];
handshake.read_message(&msg, &mut buf)?;
Ok(())
}
Some(Err(err)) => Err(NoiseError::IoError(err)),
None => Err(NoiseError::HandshakeError),
}
}
pub fn peer_addr(&self) -> Result<std::net::SocketAddr, io::Error> {
self.inner_stream.get_ref().peer_addr()
}
}
impl AsyncRead for NoiseStream {
fn poll_read(
self: Pin<&mut Self>,
cx: &mut std::task::Context<'_>,
buf: &mut ReadBuf<'_>,
) -> Poll<std::io::Result<()>> {
let projected_self = self.project();
match projected_self.inner_stream.poll_next(cx) {
Poll::Pending => {
//no new data, waking is already scheduled.
//Nothing new to decrypt, only check if we can return something from dec_storage, happens after
}
Poll::Ready(Some(Ok(noise_msg))) => {
//We have a new moise msg
let mut dec_msg = vec![0u8; MAXMSGLEN];
let len = match projected_self.noise {
Some(transport_state) => {
match transport_state.read_message(&noise_msg, &mut dec_msg) {
Ok(len) => len,
Err(_) => return Poll::Ready(Err(io::ErrorKind::InvalidInput.into())),
}
}
None => return Poll::Ready(Err(io::ErrorKind::Other.into())),
};
projected_self.dec_buffer.extend(&dec_msg[..len]);
}
Poll::Ready(Some(Err(err))) => return Poll::Ready(Err(err)),
//Stream is done, return Ok with nothing in buf
Poll::Ready(None) => return Poll::Ready(Ok(())),
}
//check and return what we can
let read_len = min(buf.remaining(), projected_self.dec_buffer.len());
if read_len > 0 {
buf.put_slice(
&projected_self
.dec_buffer
.drain(..read_len)
.collect::<Vec<u8>>(),
);
return Poll::Ready(Ok(()));
}
//If we end up here, it must mean the previous poll_next was pending as well, otherwise something was returned. Hence waking is already scheduled
Poll::Pending
}
}
impl AsyncWrite for NoiseStream {
fn poll_write(
self: Pin<&mut Self>,
cx: &mut std::task::Context<'_>,
buf: &[u8],
) -> Poll<Result<usize, std::io::Error>> {
let mut projected_self = self.project();
match projected_self.inner_stream.as_mut().poll_ready(cx) {
Poll::Pending => Poll::Pending,
Poll::Ready(Err(err)) => Poll::Ready(Err(err)),
Poll::Ready(Ok(())) => {
let mut noise_buf = BytesMut::zeroed(MAXMSGLEN + TAGLEN);
let Ok(len) = (match projected_self.noise {
Some(transport_state) => transport_state.write_message(buf, &mut noise_buf),
None => return Poll::Ready(Err(io::ErrorKind::Other.into())),
}) else {
return Poll::Ready(Err(io::ErrorKind::InvalidInput.into()));
};
noise_buf.truncate(len);
match projected_self.inner_stream.start_send(noise_buf.into()) {
Ok(()) => Poll::Ready(Ok(buf.len())),
Err(e) => Poll::Ready(Err(e)),
}
}
}
}
fn poll_flush(
self: Pin<&mut Self>,
cx: &mut std::task::Context<'_>,
) -> Poll<Result<(), std::io::Error>> {
self.project().inner_stream.poll_flush(cx)
}
fn poll_shutdown(
self: Pin<&mut Self>,
cx: &mut std::task::Context<'_>,
) -> Poll<Result<(), std::io::Error>> {
self.project().inner_stream.poll_close(cx)
}
}