keynesis_network/
net.rs

1/*!
2Wrapper/helpers of the ASMTP protocol on top of TCP
3
4While it still possible to use the low level [`Handle`] for the implementation
5of the protocol. The `net` module provides the necessary toolbox for an efficient
6and simple to use network implementation
7*/
8
9use crate::SessionId;
10use crate::{
11    accept,
12    handle::{Handle, HandleReadHalf, HandleWriteHalf},
13};
14use anyhow::{bail, Context as _, Result};
15use bytes::Bytes;
16use futures::prelude::*;
17use keynesis_core::key::PublicKey;
18use keynesis_core::key::{
19    ed25519::{self},
20    Dh,
21};
22use rand_core::{CryptoRng, RngCore};
23use std::fmt::Debug;
24use std::{
25    fmt::{self, Display},
26    net::SocketAddr,
27    pin::Pin,
28    task::{Context, Poll},
29};
30use tokio::net::{
31    lookup_host,
32    tcp::{OwnedReadHalf, OwnedWriteHalf},
33    TcpListener, TcpStream, ToSocketAddrs,
34};
35
36/// object that will listen to inbound connections and handle incoming connections
37/// accordingly.
38///
39/// The process of handling the data received from the peers (handling the handshake)
40/// is done asynchronously so we can start a new thread to process the new peer and
41/// accept new connections straight away.
42///
43pub struct Listener {
44    listener: TcpListener,
45}
46
47/// A bidirectional, encrypted and authenticated connection with a peer
48///
49/// the connection can be conveniently split into its halves ([`ConnectionWriter`]
50/// and [`ConnectionReader`]) for more convenient handling of that the protocol
51/// may be available to write data and receive data at different time.
52///
53pub struct Connection<PK>
54where
55    PK: PublicKey,
56{
57    writer: ConnectionWriter<PK>,
58    reader: ConnectionReader<PK>,
59}
60
61/// writer halve of the authenticated encrypted connection with the peer
62pub struct ConnectionWriter<PK>
63where
64    PK: PublicKey,
65{
66    writer: HandleWriteHalf<OwnedWriteHalf, PK>,
67    peer_addr: SocketAddr,
68}
69
70/// reader halve of the authenticated encrypted connection with the peer
71pub struct ConnectionReader<PK>
72where
73    PK: PublicKey,
74{
75    reader: HandleReadHalf<OwnedReadHalf, PK>,
76    peer_addr: SocketAddr,
77}
78
79/// object to accept incoming connection
80///
81/// this is split from the [`Listener`]'s [`accept`](Listener::accept) function
82/// so that we can process the handshake in a separate thread/task and not block
83/// other incoming connections.
84///
85pub struct Accepting<RNG, K = ed25519::SecretKey>
86where
87    K: Dh,
88{
89    handle: accept::Accepting<OwnedReadHalf, OwnedWriteHalf, RNG, K>,
90    peer_addr: SocketAddr,
91}
92
93impl Listener {
94    /// create a new listener object
95    ///
96    /// will listen for incoming connection at the given [`ToSocketAddrs`] address.
97    ///
98    pub async fn new<A>(addr: A) -> Result<Self>
99    where
100        A: ToSocketAddrs + Display,
101    {
102        let listener = TcpListener::bind(&addr)
103            .await
104            .with_context(|| format!("Cannot listen to {}", addr))?;
105
106        Ok(Self { listener })
107    }
108
109    /// start accepting a new incoming connection
110    ///
111    /// this function _blocks_ until a new inbound connection happens. This function
112    /// does not perform any handshake verification of any sort. This will is better
113    /// to run it in an different task so you can start accepting new inbound
114    /// connections
115    pub async fn accept<RNG, K>(&self, rng: RNG) -> Result<Accepting<RNG, K>>
116    where
117        RNG: CryptoRng + RngCore,
118        K: Dh,
119    {
120        let (stream, peer_addr) = self
121            .listener
122            .accept()
123            .await
124            .context("Cannot accept new peer from the listener")?;
125
126        let (reader, writer) = stream.into_split();
127
128        let handle = Handle::<_, _, K::Public>::accept::<K, _>(rng, reader, writer);
129
130        Ok(Accepting { handle, peer_addr })
131    }
132}
133
134impl<RNG, K> Accepting<RNG, K>
135where
136    RNG: CryptoRng + RngCore,
137    K: Dh,
138{
139    /// the inbound new connection's remote address
140    ///
141    /// beware that this may not reflect the peer's actual address as they may be
142    /// behind routers. Also this is not the same as the address that may be
143    /// advertised by the peer in the gossip as they might use a different port
144    /// number or a different routes for inbounds or outbound connections
145    /// depending on their IT configuration.
146    ///
147    /// This is however available so that server implementation may decides
148    /// to blacklist inbound connections coming from certain area or known
149    /// IP addresses that are known to be not welcomed.
150    ///
151    pub fn remote_address(&self) -> SocketAddr {
152        self.peer_addr
153    }
154
155    /// perform the handshake check with the inbound peer
156    ///
157    /// except to receive the first message of the [Noise **IK**] handshake.
158    /// The [`Dh`] implemented by the remote peer must be the same as the
159    /// one implemented here.
160    ///
161    /// [Noise **IK**]: https://noiseexplorer.com/patterns/IK/
162    #[tracing::instrument(skip(k, check_id), level = "debug")]
163    pub async fn handshake<F>(self, k: &K, check_id: F) -> Result<Connection<K::Public>>
164    where
165        F: Fn(&K::Public) -> bool,
166        K::Public: Debug + Display,
167    {
168        let Self { handle, peer_addr } = self;
169
170        tracing::debug!("processing remote's handshake");
171
172        let handle = handle
173            .accept(k, check_id)
174            .await
175            .with_context(|| format!("Failed to handshake with {}", peer_addr))?;
176
177        tracing::debug!(
178            session_id = %handle.session_id(),
179            id = %handle.remote_public_identity(),
180            "handshake succeed",
181        );
182
183        let (reader, writer) = handle.split();
184        let reader = ConnectionReader { reader, peer_addr };
185        let writer = ConnectionWriter { writer, peer_addr };
186
187        Ok(Connection { reader, writer })
188    }
189}
190
191impl<PK> ConnectionReader<PK>
192where
193    PK: PublicKey,
194{
195    /// retrieve the public identity of the peer
196    ///
197    pub fn remote_public_identity(&self) -> &PK {
198        self.reader.remote_public_identity()
199    }
200
201    /// the remote peer address we are receiving messages from
202    ///
203    /// beware that this may not reflect the peer's actual address as they may be
204    /// behind routers. Also this is not the same as the address that may be
205    /// advertised by the peer in the gossip as they might use a different port
206    /// number or a different routes for inbounds or outbound connections
207    /// depending on their IT configuration.
208    pub fn remote_address(&self) -> SocketAddr {
209        self.peer_addr
210    }
211
212    /// retrieve the unique identifier of the established session
213    ///
214    /// this is derived from the NOISE handshake and is the same
215    /// on both sides of the stream (here and for the remote).
216    pub fn session_id(&self) -> &SessionId {
217        self.reader.session_id()
218    }
219}
220
221impl<PK> ConnectionWriter<PK>
222where
223    PK: PublicKey,
224{
225    /// retrieve the public identity of the peer
226    ///
227    pub fn remote_public_identity(&self) -> &PK {
228        self.writer.remote_public_identity()
229    }
230
231    /// the remote address we are sending messages to
232    ///
233    /// beware that this may not reflect the peer's actual address as they may be
234    /// behind routers. Also this is not the same as the address that may be
235    /// advertised by the peer in the gossip as they might use a different port
236    /// number or a different routes for inbounds or outbound connections
237    /// depending on their IT configuration.
238    ///
239    pub fn remote_address(&self) -> SocketAddr {
240        self.peer_addr
241    }
242
243    /// retrieve the unique identifier of the established session
244    ///
245    /// this is derived from the NOISE handshake and is the same
246    /// on both sides of the stream (here and for the remote).
247    pub fn session_id(&self) -> &SessionId {
248        self.writer.session_id()
249    }
250}
251
252impl<PK> Connection<PK>
253where
254    PK: PublicKey + Debug + Display,
255{
256    /// retrieve the public identity of the peer
257    ///
258    pub fn remote_public_identity(&self) -> &PK {
259        self.writer.remote_public_identity()
260    }
261
262    /// the remote address we are sending/receiving messages to/from
263    ///
264    /// beware that this may not reflect the peer's actual address as they may be
265    /// behind routers. Also this is not the same as the address that may be
266    /// advertised by the peer in the gossip as they might use a different port
267    /// number or a different routes for inbounds or outbound connections
268    /// depending on their IT configuration.
269    ///
270    pub fn remote_address(&self) -> SocketAddr {
271        self.writer.remote_address()
272    }
273
274    /// retrieve the unique identifier of the established session
275    ///
276    /// this is derived from the NOISE handshake and is the same
277    /// on both sides of the stream (here and for the remote).
278    pub fn session_id(&self) -> &SessionId {
279        self.writer.session_id()
280    }
281
282    /// connect to the given socket address, expecting the remote to identify
283    /// with the [`PublicKey`] `rs`.
284    ///
285    /// The function will use the given `RNG` to generate an ephemeral private keys
286    /// that will be used only for this connection and the given key `k` to authenticate
287    /// ourself to the remote.
288    ///
289    #[tracing::instrument(skip(k, rng), level = "info")]
290    pub async fn connect_to<RNG, K>(
291        rng: RNG,
292        k: &K,
293        peer_addr: SocketAddr,
294        rs: K::Public,
295    ) -> Result<Self>
296    where
297        RNG: CryptoRng + RngCore,
298        K: Dh<Public = PK>,
299    {
300        let stream = TcpStream::connect(peer_addr)
301            .await
302            .with_context(|| format!("Cannot connect to peer {}", peer_addr))?;
303
304        let (reader, writer) = stream.into_split();
305
306        let handle = Handle::open(rng, k, rs, reader, writer)
307            .await
308            .with_context(|| format!("Failed to handshake with peer {}", peer_addr))?;
309
310        tracing::debug!(
311            session_id = %handle.session_id(),
312            id = %handle.remote_public_identity(),
313            "handshake succeed",
314        );
315
316        let (reader, writer) = handle.split();
317
318        let reader = ConnectionReader { reader, peer_addr };
319        let writer = ConnectionWriter { writer, peer_addr };
320        Ok(Self { reader, writer })
321    }
322
323    /// attempt to connect to any resolved [`lookup_host`] result of the given [`ToSocketAddrs`].
324    ///
325    /// The function will returns at the first successful attempt or once all the possible options
326    /// have been tried and failed.
327    ///
328    #[tracing::instrument(skip(k, rng), level = "info")]
329    pub async fn connect<RNG, K, A>(
330        mut rng: RNG,
331        k: &K,
332        peer_addr: A,
333        rs: K::Public,
334    ) -> Result<Self>
335    where
336        RNG: RngCore + CryptoRng,
337        A: ToSocketAddrs + Display + fmt::Debug,
338        K: Dh<Public = PK>,
339    {
340        let peer_addrs = lookup_host(&peer_addr)
341            .await
342            .context("Cannot connect to remote peer address")?;
343
344        for socket_addr in peer_addrs {
345            match Self::connect_to(&mut rng, k, socket_addr, rs.clone()).await {
346                Ok(connection) => return Ok(connection),
347                Err(error) => {
348                    tracing::info!(reason = ?error, "Failed to connect to {} with {}", peer_addr, socket_addr);
349                    continue;
350                }
351            }
352        }
353
354        bail!("Cannot connect to {}", peer_addr)
355    }
356
357    /// split the connections into 2 parts
358    ///
359    /// this allows to have 2 different independent objects to read or write to/from
360    /// the remote peer.
361    pub fn into_parts(self) -> (ConnectionReader<PK>, ConnectionWriter<PK>) {
362        let Self { reader, writer } = self;
363
364        (reader, writer)
365    }
366}
367
368impl<PK> Stream for Connection<PK>
369where
370    PK: PublicKey,
371{
372    type Item = Result<Bytes>;
373    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
374        let connection = self.get_mut();
375        Pin::new(&mut connection.reader).poll_next(cx)
376    }
377}
378
379impl<PK> Stream for ConnectionReader<PK>
380where
381    PK: PublicKey,
382{
383    type Item = Result<Bytes>;
384    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
385        let connection = self.get_mut();
386        match Pin::new(&mut connection.reader).poll_next(cx) {
387            Poll::Pending => Poll::Pending,
388            Poll::Ready(None) => Poll::Ready(None),
389            Poll::Ready(Some(Err(error))) => Poll::Ready(Some(
390                Err(error).context("Cannot receive message from connection"),
391            )),
392            Poll::Ready(Some(Ok(bytes))) => {
393                let r = bytes.freeze();
394
395                Poll::Ready(Some(Ok(r)))
396            }
397        }
398    }
399}
400
401impl<PK> stream::FusedStream for ConnectionReader<PK>
402where
403    PK: PublicKey,
404{
405    fn is_terminated(&self) -> bool {
406        self.reader.is_terminated()
407    }
408}
409
410impl<PK> stream::FusedStream for Connection<PK>
411where
412    PK: PublicKey,
413{
414    fn is_terminated(&self) -> bool {
415        self.reader.is_terminated()
416    }
417}
418
419impl<PK> Sink<Bytes> for Connection<PK>
420where
421    PK: PublicKey,
422{
423    type Error = anyhow::Error;
424
425    fn start_send(self: Pin<&mut Self>, item: Bytes) -> Result<(), Self::Error> {
426        let connection = self.get_mut();
427        Pin::new(&mut connection.writer).start_send(item)
428    }
429
430    fn poll_ready(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
431        let connection = self.get_mut();
432        Pin::new(&mut connection.writer).poll_ready(cx)
433    }
434
435    fn poll_close(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
436        let connection = self.get_mut();
437        Pin::new(&mut connection.writer).poll_close(cx)
438    }
439
440    fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
441        let connection = self.get_mut();
442        Pin::new(&mut connection.writer).poll_flush(cx)
443    }
444}
445
446impl<PK> Sink<Bytes> for ConnectionWriter<PK>
447where
448    PK: PublicKey,
449{
450    type Error = anyhow::Error;
451
452    fn start_send(self: Pin<&mut Self>, item: Bytes) -> Result<(), Self::Error> {
453        let connection = self.get_mut();
454        Pin::new(&mut connection.writer).start_send(item)
455    }
456
457    fn poll_ready(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
458        let connection = self.get_mut();
459        Pin::new(&mut connection.writer).poll_ready(cx)
460    }
461
462    fn poll_close(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
463        let connection = self.get_mut();
464        Pin::new(&mut connection.writer).poll_close(cx)
465    }
466
467    fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
468        let connection = self.get_mut();
469        Pin::new(&mut connection.writer).poll_flush(cx)
470    }
471}
472
473impl<RNG, K> fmt::Debug for Accepting<RNG, K>
474where
475    K: Dh,
476{
477    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
478        f.debug_struct("Accepting")
479            .field("remote_address", &self.peer_addr)
480            .finish()
481    }
482}
483
484impl<PK> fmt::Debug for Connection<PK>
485where
486    PK: PublicKey + Debug + Display,
487{
488    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
489        f.debug_struct("Connection")
490            .field("remote_address", &self.remote_address())
491            .field("session", &self.session_id())
492            .field("id", self.remote_public_identity())
493            .finish()
494    }
495}
keynesis_network/net.rs

keynesis_network/
net.rs
