2017-11-02 11:58:02 +01:00
|
|
|
// Copyright 2017 Parity Technologies (UK) Ltd.
|
2018-03-07 16:20:55 +01:00
|
|
|
//
|
|
|
|
// Permission is hereby granted, free of charge, to any person obtaining a
|
|
|
|
// copy of this software and associated documentation files (the "Software"),
|
|
|
|
// to deal in the Software without restriction, including without limitation
|
|
|
|
// the rights to use, copy, modify, merge, publish, distribute, sublicense,
|
|
|
|
// and/or sell copies of the Software, and to permit persons to whom the
|
2017-11-02 11:58:02 +01:00
|
|
|
// Software is furnished to do so, subject to the following conditions:
|
|
|
|
//
|
2018-03-07 16:20:55 +01:00
|
|
|
// The above copyright notice and this permission notice shall be included in
|
2017-11-02 11:58:02 +01:00
|
|
|
// all copies or substantial portions of the Software.
|
|
|
|
//
|
2018-03-07 16:20:55 +01:00
|
|
|
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS
|
|
|
|
// OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
|
|
|
|
// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
|
|
|
|
// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
|
|
|
|
// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
|
|
|
|
// FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
|
2017-11-02 11:58:02 +01:00
|
|
|
// DEALINGS IN THE SOFTWARE.
|
|
|
|
|
|
|
|
//! Implementation of the libp2p `Transport` trait for TCP/IP.
|
2017-12-04 16:05:37 +01:00
|
|
|
//!
|
|
|
|
//! Uses [the *tokio* library](https://tokio.rs).
|
2018-03-07 16:20:55 +01:00
|
|
|
//!
|
2017-12-04 16:05:37 +01:00
|
|
|
//! # Usage
|
2018-03-07 16:20:55 +01:00
|
|
|
//!
|
2017-12-04 16:05:37 +01:00
|
|
|
//! Example:
|
|
|
|
//!
|
|
|
|
//! ```
|
2018-12-10 13:39:11 +01:00
|
|
|
//! extern crate libp2p_tcp;
|
|
|
|
//! use libp2p_tcp::TcpConfig;
|
2017-12-04 16:05:37 +01:00
|
|
|
//!
|
|
|
|
//! # fn main() {
|
2018-07-16 12:15:27 +02:00
|
|
|
//! let tcp = TcpConfig::new();
|
2017-12-04 16:05:37 +01:00
|
|
|
//! # }
|
|
|
|
//! ```
|
|
|
|
//!
|
|
|
|
//! The `TcpConfig` structs implements the `Transport` trait of the `swarm` library. See the
|
|
|
|
//! documentation of `swarm` and of libp2p in general to learn how to use the `Transport` trait.
|
2017-11-02 11:58:02 +01:00
|
|
|
|
2019-04-11 22:51:07 +02:00
|
|
|
use futures::{
|
|
|
|
future::{self, Either, FutureResult},
|
|
|
|
prelude::*,
|
|
|
|
stream::{self, Chain, IterOk, Once}
|
|
|
|
};
|
2019-06-03 17:55:15 +02:00
|
|
|
use get_if_addrs::{IfAddr, get_if_addrs};
|
|
|
|
use ipnet::{IpNet, Ipv4Net, Ipv6Net};
|
2019-04-30 21:14:57 +03:00
|
|
|
use libp2p_core::{
|
|
|
|
Transport,
|
|
|
|
multiaddr::{Protocol, Multiaddr},
|
|
|
|
transport::{ListenerEvent, TransportError}
|
|
|
|
};
|
2019-05-15 14:48:26 +02:00
|
|
|
use log::{debug, error, trace};
|
2019-04-11 22:51:07 +02:00
|
|
|
use std::{
|
2019-06-03 17:55:15 +02:00
|
|
|
collections::VecDeque,
|
2019-04-11 22:51:07 +02:00
|
|
|
fmt,
|
|
|
|
io::{self, Read, Write},
|
|
|
|
iter::{self, FromIterator},
|
|
|
|
net::{IpAddr, SocketAddr},
|
|
|
|
time::Duration,
|
|
|
|
vec::IntoIter
|
|
|
|
};
|
2018-08-09 14:04:15 +02:00
|
|
|
use tk_listen::{ListenExt, SleepOnError};
|
2018-07-21 13:01:59 +02:00
|
|
|
use tokio_io::{AsyncRead, AsyncWrite};
|
2019-04-11 22:51:07 +02:00
|
|
|
use tokio_tcp::{ConnectFuture, Incoming, TcpStream};
|
2017-09-18 16:52:51 +02:00
|
|
|
|
2017-12-04 16:05:37 +01:00
|
|
|
/// Represents the configuration for a TCP/IP transport capability for libp2p.
|
2017-11-02 11:58:02 +01:00
|
|
|
///
|
2018-07-16 12:15:27 +02:00
|
|
|
/// The TCP sockets created by libp2p will need to be progressed by running the futures and streams
|
|
|
|
/// obtained by libp2p through the tokio reactor.
|
2018-07-17 11:55:18 +02:00
|
|
|
#[derive(Debug, Clone, Default)]
|
2018-08-08 17:54:15 +02:00
|
|
|
pub struct TcpConfig {
|
2018-09-11 12:04:35 +02:00
|
|
|
/// How long a listener should sleep after receiving an error, before trying again.
|
2018-08-08 17:54:15 +02:00
|
|
|
sleep_on_error: Duration,
|
2018-09-11 12:04:35 +02:00
|
|
|
/// Size of the recv buffer size to set for opened sockets, or `None` to keep default.
|
|
|
|
recv_buffer_size: Option<usize>,
|
|
|
|
/// Size of the send buffer size to set for opened sockets, or `None` to keep default.
|
|
|
|
send_buffer_size: Option<usize>,
|
|
|
|
/// TTL to set for opened sockets, or `None` to keep default.
|
|
|
|
ttl: Option<u32>,
|
|
|
|
/// Keep alive duration to set for opened sockets, or `None` to keep default.
|
|
|
|
keepalive: Option<Option<Duration>>,
|
|
|
|
/// `TCP_NODELAY` to set for opened sockets, or `None` to keep default.
|
|
|
|
nodelay: Option<bool>,
|
2018-08-08 17:54:15 +02:00
|
|
|
}
|
2017-09-18 17:25:04 +02:00
|
|
|
|
2017-12-04 16:05:37 +01:00
|
|
|
impl TcpConfig {
|
2018-07-16 12:15:27 +02:00
|
|
|
/// Creates a new configuration object for TCP/IP.
|
|
|
|
pub fn new() -> TcpConfig {
|
2018-08-08 17:54:15 +02:00
|
|
|
TcpConfig {
|
|
|
|
sleep_on_error: Duration::from_millis(100),
|
2018-09-11 12:04:35 +02:00
|
|
|
recv_buffer_size: None,
|
|
|
|
send_buffer_size: None,
|
|
|
|
ttl: None,
|
|
|
|
keepalive: None,
|
|
|
|
nodelay: None,
|
2018-08-08 17:54:15 +02:00
|
|
|
}
|
2017-10-23 11:45:35 +02:00
|
|
|
}
|
2018-09-11 12:04:35 +02:00
|
|
|
|
|
|
|
/// Sets the size of the recv buffer size to set for opened sockets.
|
|
|
|
pub fn recv_buffer_size(mut self, value: usize) -> Self {
|
|
|
|
self.recv_buffer_size = Some(value);
|
|
|
|
self
|
|
|
|
}
|
|
|
|
|
|
|
|
/// Sets the size of the send buffer size to set for opened sockets.
|
|
|
|
pub fn send_buffer_size(mut self, value: usize) -> Self {
|
|
|
|
self.send_buffer_size = Some(value);
|
|
|
|
self
|
|
|
|
}
|
|
|
|
|
|
|
|
/// Sets the TTL to set for opened sockets.
|
|
|
|
pub fn ttl(mut self, value: u32) -> Self {
|
|
|
|
self.ttl = Some(value);
|
|
|
|
self
|
|
|
|
}
|
|
|
|
|
|
|
|
/// Sets the keep alive pinging duration to set for opened sockets.
|
|
|
|
pub fn keepalive(mut self, value: Option<Duration>) -> Self {
|
|
|
|
self.keepalive = Some(value);
|
|
|
|
self
|
|
|
|
}
|
|
|
|
|
|
|
|
/// Sets the `TCP_NODELAY` to set for opened sockets.
|
|
|
|
pub fn nodelay(mut self, value: bool) -> Self {
|
|
|
|
self.nodelay = Some(value);
|
|
|
|
self
|
|
|
|
}
|
2017-09-18 17:25:04 +02:00
|
|
|
}
|
2017-09-18 16:52:51 +02:00
|
|
|
|
2017-12-04 16:05:37 +01:00
|
|
|
impl Transport for TcpConfig {
|
2018-07-21 13:01:59 +02:00
|
|
|
type Output = TcpTransStream;
|
2019-01-10 11:27:06 +01:00
|
|
|
type Error = io::Error;
|
2019-04-11 22:51:07 +02:00
|
|
|
type Listener = TcpListener;
|
|
|
|
type ListenerUpgrade = FutureResult<Self::Output, Self::Error>;
|
2018-08-09 14:04:15 +02:00
|
|
|
type Dial = TcpDialFut;
|
2017-10-23 11:45:35 +02:00
|
|
|
|
2019-04-10 10:29:21 +02:00
|
|
|
fn listen_on(self, addr: Multiaddr) -> Result<Self::Listener, TransportError<Self::Error>> {
|
2019-04-11 22:51:07 +02:00
|
|
|
let socket_addr =
|
|
|
|
if let Ok(sa) = multiaddr_to_socketaddr(&addr) {
|
|
|
|
sa
|
|
|
|
} else {
|
|
|
|
return Err(TransportError::MultiaddrNotSupported(addr))
|
2017-11-24 16:10:34 +01:00
|
|
|
};
|
2017-12-19 18:09:17 +01:00
|
|
|
|
2019-04-11 22:51:07 +02:00
|
|
|
let listener = tokio_tcp::TcpListener::bind(&socket_addr).map_err(TransportError::Other)?;
|
|
|
|
let local_addr = listener.local_addr().map_err(TransportError::Other)?;
|
|
|
|
let port = local_addr.port();
|
|
|
|
|
|
|
|
// Determine all our listen addresses which is either a single local IP address
|
|
|
|
// or (if a wildcard IP address was used) the addresses of all our interfaces,
|
|
|
|
// as reported by `get_if_addrs`.
|
|
|
|
let addrs =
|
|
|
|
if socket_addr.ip().is_unspecified() {
|
|
|
|
let addrs = host_addresses(port).map_err(TransportError::Other)?;
|
2019-06-03 17:55:15 +02:00
|
|
|
debug!("Listening on {:?}", addrs.iter().map(|(_, _, ma)| ma).collect::<Vec<_>>());
|
2019-04-11 22:51:07 +02:00
|
|
|
Addresses::Many(addrs)
|
|
|
|
} else {
|
2019-06-03 17:55:15 +02:00
|
|
|
let ma = ip_to_multiaddr(local_addr.ip(), port);
|
2019-04-11 22:51:07 +02:00
|
|
|
debug!("Listening on {:?}", ma);
|
|
|
|
Addresses::One(ma)
|
|
|
|
};
|
|
|
|
|
|
|
|
// Generate `NewAddress` events for each new `Multiaddr`.
|
|
|
|
let events = match addrs {
|
|
|
|
Addresses::One(ref ma) => {
|
|
|
|
let event = ListenerEvent::NewAddress(ma.clone());
|
|
|
|
Either::A(stream::once(Ok(event)))
|
|
|
|
}
|
|
|
|
Addresses::Many(ref aa) => {
|
2019-06-03 17:55:15 +02:00
|
|
|
let events = aa.iter()
|
|
|
|
.map(|(_, _, ma)| ma)
|
2019-04-11 22:51:07 +02:00
|
|
|
.cloned()
|
|
|
|
.map(ListenerEvent::NewAddress)
|
|
|
|
.collect::<Vec<_>>();
|
|
|
|
Either::B(stream::iter_ok(events))
|
|
|
|
}
|
|
|
|
};
|
|
|
|
|
|
|
|
let stream = TcpListenStream {
|
|
|
|
inner: Ok(listener.incoming().sleep_on_error(self.sleep_on_error)),
|
|
|
|
port,
|
|
|
|
addrs,
|
|
|
|
pending: VecDeque::new(),
|
|
|
|
config: self
|
|
|
|
};
|
|
|
|
|
|
|
|
Ok(TcpListener {
|
|
|
|
inner: match events {
|
|
|
|
Either::A(e) => Either::A(e.chain(stream)),
|
|
|
|
Either::B(e) => Either::B(e.chain(stream))
|
|
|
|
}
|
|
|
|
})
|
2017-10-23 11:45:35 +02:00
|
|
|
}
|
|
|
|
|
2019-01-10 11:27:06 +01:00
|
|
|
fn dial(self, addr: Multiaddr) -> Result<Self::Dial, TransportError<Self::Error>> {
|
2019-04-11 22:51:07 +02:00
|
|
|
let socket_addr =
|
|
|
|
if let Ok(socket_addr) = multiaddr_to_socketaddr(&addr) {
|
|
|
|
if socket_addr.port() == 0 || socket_addr.ip().is_unspecified() {
|
|
|
|
debug!("Instantly refusing dialing {}, as it is invalid", addr);
|
|
|
|
return Err(TransportError::Other(io::ErrorKind::ConnectionRefused.into()))
|
|
|
|
}
|
|
|
|
socket_addr
|
2018-07-02 10:51:10 +02:00
|
|
|
} else {
|
2019-04-11 22:51:07 +02:00
|
|
|
return Err(TransportError::MultiaddrNotSupported(addr))
|
|
|
|
};
|
|
|
|
|
|
|
|
debug!("Dialing {}", addr);
|
|
|
|
|
|
|
|
let future = TcpDialFut {
|
|
|
|
inner: TcpStream::connect(&socket_addr),
|
|
|
|
config: self
|
|
|
|
};
|
|
|
|
|
|
|
|
Ok(future)
|
2017-10-23 11:45:35 +02:00
|
|
|
}
|
2017-09-18 16:52:51 +02:00
|
|
|
}
|
|
|
|
|
|
|
|
// This type of logic should probably be moved into the multiaddr package
|
2017-11-24 16:10:34 +01:00
|
|
|
fn multiaddr_to_socketaddr(addr: &Multiaddr) -> Result<SocketAddr, ()> {
|
2018-08-09 14:04:15 +02:00
|
|
|
let mut iter = addr.iter();
|
|
|
|
let proto1 = iter.next().ok_or(())?;
|
|
|
|
let proto2 = iter.next().ok_or(())?;
|
2017-10-23 11:45:35 +02:00
|
|
|
|
2018-08-09 14:04:15 +02:00
|
|
|
if iter.next().is_some() {
|
2017-12-15 17:29:54 +01:00
|
|
|
return Err(());
|
|
|
|
}
|
|
|
|
|
2018-08-09 14:04:15 +02:00
|
|
|
match (proto1, proto2) {
|
2018-09-20 19:51:00 +02:00
|
|
|
(Protocol::Ip4(ip), Protocol::Tcp(port)) => Ok(SocketAddr::new(ip.into(), port)),
|
|
|
|
(Protocol::Ip6(ip), Protocol::Tcp(port)) => Ok(SocketAddr::new(ip.into(), port)),
|
2018-08-09 14:04:15 +02:00
|
|
|
_ => Err(()),
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2019-04-11 22:51:07 +02:00
|
|
|
// Create a [`Multiaddr`] from the given IP address and port number.
|
2019-06-03 17:55:15 +02:00
|
|
|
fn ip_to_multiaddr(ip: IpAddr, port: u16) -> Multiaddr {
|
2019-04-11 22:51:07 +02:00
|
|
|
let proto = match ip {
|
|
|
|
IpAddr::V4(ip) => Protocol::Ip4(ip),
|
|
|
|
IpAddr::V6(ip) => Protocol::Ip6(ip)
|
|
|
|
};
|
|
|
|
let it = iter::once(proto).chain(iter::once(Protocol::Tcp(port)));
|
|
|
|
Multiaddr::from_iter(it)
|
|
|
|
}
|
|
|
|
|
|
|
|
// Collect all local host addresses and use the provided port number as listen port.
|
2019-06-03 17:55:15 +02:00
|
|
|
fn host_addresses(port: u16) -> io::Result<Vec<(IpAddr, IpNet, Multiaddr)>> {
|
|
|
|
let mut addrs = Vec::new();
|
2019-04-11 22:51:07 +02:00
|
|
|
for iface in get_if_addrs()? {
|
2019-06-03 17:55:15 +02:00
|
|
|
let ip = iface.ip();
|
|
|
|
let ma = ip_to_multiaddr(ip, port);
|
|
|
|
let ipn = match iface.addr {
|
|
|
|
IfAddr::V4(ip4) => {
|
|
|
|
let prefix_len = (!u32::from_be_bytes(ip4.netmask.octets())).leading_zeros();
|
|
|
|
let ipnet = Ipv4Net::new(ip4.ip, prefix_len as u8)
|
|
|
|
.expect("prefix_len is the number of bits in a u32, so can not exceed 32");
|
|
|
|
IpNet::V4(ipnet)
|
|
|
|
}
|
|
|
|
IfAddr::V6(ip6) => {
|
|
|
|
let prefix_len = (!u128::from_be_bytes(ip6.netmask.octets())).leading_zeros();
|
|
|
|
let ipnet = Ipv6Net::new(ip6.ip, prefix_len as u8)
|
|
|
|
.expect("prefix_len is the number of bits in a u128, so can not exceed 128");
|
|
|
|
IpNet::V6(ipnet)
|
|
|
|
}
|
|
|
|
};
|
|
|
|
addrs.push((ip, ipn, ma))
|
2019-04-11 22:51:07 +02:00
|
|
|
}
|
|
|
|
Ok(addrs)
|
|
|
|
}
|
|
|
|
|
2018-09-11 12:04:35 +02:00
|
|
|
/// Applies the socket configuration parameters to a socket.
|
2019-01-10 11:27:06 +01:00
|
|
|
fn apply_config(config: &TcpConfig, socket: &TcpStream) -> Result<(), io::Error> {
|
2018-09-11 12:04:35 +02:00
|
|
|
if let Some(recv_buffer_size) = config.recv_buffer_size {
|
|
|
|
socket.set_recv_buffer_size(recv_buffer_size)?;
|
|
|
|
}
|
|
|
|
|
|
|
|
if let Some(send_buffer_size) = config.send_buffer_size {
|
|
|
|
socket.set_send_buffer_size(send_buffer_size)?;
|
|
|
|
}
|
|
|
|
|
|
|
|
if let Some(ttl) = config.ttl {
|
|
|
|
socket.set_ttl(ttl)?;
|
|
|
|
}
|
|
|
|
|
|
|
|
if let Some(keepalive) = config.keepalive {
|
|
|
|
socket.set_keepalive(keepalive)?;
|
|
|
|
}
|
|
|
|
|
|
|
|
if let Some(nodelay) = config.nodelay {
|
|
|
|
socket.set_nodelay(nodelay)?;
|
|
|
|
}
|
|
|
|
|
|
|
|
Ok(())
|
|
|
|
}
|
|
|
|
|
2018-08-09 14:04:15 +02:00
|
|
|
/// Future that dials a TCP/IP address.
|
|
|
|
#[derive(Debug)]
|
2018-09-04 18:30:57 +08:00
|
|
|
#[must_use = "futures do nothing unless polled"]
|
2018-08-09 14:04:15 +02:00
|
|
|
pub struct TcpDialFut {
|
|
|
|
inner: ConnectFuture,
|
2018-09-11 12:04:35 +02:00
|
|
|
/// Original configuration.
|
|
|
|
config: TcpConfig,
|
2018-08-09 14:04:15 +02:00
|
|
|
}
|
|
|
|
|
|
|
|
impl Future for TcpDialFut {
|
2018-10-17 10:17:40 +01:00
|
|
|
type Item = TcpTransStream;
|
2019-01-10 11:27:06 +01:00
|
|
|
type Error = io::Error;
|
2018-08-09 14:04:15 +02:00
|
|
|
|
2019-01-10 11:27:06 +01:00
|
|
|
fn poll(&mut self) -> Poll<TcpTransStream, io::Error> {
|
2018-08-09 14:04:15 +02:00
|
|
|
match self.inner.poll() {
|
|
|
|
Ok(Async::Ready(stream)) => {
|
2018-09-11 12:04:35 +02:00
|
|
|
apply_config(&self.config, &stream)?;
|
2018-10-17 10:17:40 +01:00
|
|
|
Ok(Async::Ready(TcpTransStream { inner: stream }))
|
2018-08-09 14:04:15 +02:00
|
|
|
}
|
|
|
|
Ok(Async::NotReady) => Ok(Async::NotReady),
|
|
|
|
Err(err) => {
|
2018-10-17 10:17:40 +01:00
|
|
|
debug!("Error while dialing => {:?}", err);
|
2018-08-09 14:04:15 +02:00
|
|
|
Err(err)
|
|
|
|
}
|
2017-10-23 11:45:35 +02:00
|
|
|
}
|
2018-08-09 14:04:15 +02:00
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2019-04-11 22:51:07 +02:00
|
|
|
/// Stream of `ListenerEvent`s.
|
|
|
|
#[derive(Debug)]
|
|
|
|
pub struct TcpListener {
|
|
|
|
inner: Either<
|
|
|
|
Chain<Once<ListenerEvent<FutureResult<TcpTransStream, io::Error>>, io::Error>, TcpListenStream>,
|
|
|
|
Chain<IterOk<IntoIter<ListenerEvent<FutureResult<TcpTransStream, io::Error>>>, io::Error>, TcpListenStream>
|
|
|
|
>
|
|
|
|
}
|
|
|
|
|
|
|
|
impl Stream for TcpListener {
|
|
|
|
type Item = ListenerEvent<FutureResult<TcpTransStream, io::Error>>;
|
|
|
|
type Error = io::Error;
|
|
|
|
|
|
|
|
fn poll(&mut self) -> Poll<Option<Self::Item>, Self::Error> {
|
|
|
|
match self.inner {
|
|
|
|
Either::A(ref mut it) => it.poll(),
|
|
|
|
Either::B(ref mut it) => it.poll()
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
/// Listen address information.
|
|
|
|
#[derive(Debug)]
|
|
|
|
enum Addresses {
|
|
|
|
/// A specific address is used to listen.
|
|
|
|
One(Multiaddr),
|
|
|
|
/// A set of addresses is used to listen.
|
2019-06-03 17:55:15 +02:00
|
|
|
Many(Vec<(IpAddr, IpNet, Multiaddr)>)
|
2019-04-11 22:51:07 +02:00
|
|
|
}
|
|
|
|
|
2019-06-03 17:55:15 +02:00
|
|
|
type Buffer = VecDeque<ListenerEvent<FutureResult<TcpTransStream, io::Error>>>;
|
|
|
|
|
2018-08-09 14:04:15 +02:00
|
|
|
/// Stream that listens on an TCP/IP address.
|
|
|
|
pub struct TcpListenStream {
|
2019-04-11 22:51:07 +02:00
|
|
|
/// Stream of incoming sockets.
|
2019-01-10 11:27:06 +01:00
|
|
|
inner: Result<SleepOnError<Incoming>, Option<io::Error>>,
|
2019-04-11 22:51:07 +02:00
|
|
|
/// The port which we use as our listen port in listener event addresses.
|
|
|
|
port: u16,
|
|
|
|
/// The set of known addresses.
|
|
|
|
addrs: Addresses,
|
|
|
|
/// Temporary buffer of listener events.
|
2019-06-03 17:55:15 +02:00
|
|
|
pending: Buffer,
|
2018-09-11 12:04:35 +02:00
|
|
|
/// Original configuration.
|
2019-04-11 22:51:07 +02:00
|
|
|
config: TcpConfig
|
2018-08-09 14:04:15 +02:00
|
|
|
}
|
|
|
|
|
2019-06-03 17:55:15 +02:00
|
|
|
// Map a `SocketAddr` to the corresponding `Multiaddr`.
|
|
|
|
// If not found, check for host address changes.
|
|
|
|
// This is a function rather than a method due to borrowing issues.
|
|
|
|
fn map_addr(addr: &SocketAddr, addrs: &mut Addresses, pending: &mut Buffer, port: u16)
|
|
|
|
-> Result<Multiaddr, io::Error>
|
|
|
|
{
|
|
|
|
match addrs {
|
|
|
|
Addresses::One(ref ma) => Ok(ma.clone()),
|
|
|
|
Addresses::Many(ref mut addrs) => {
|
|
|
|
// Check for exact match:
|
|
|
|
if let Some((_, _, ma)) = addrs.iter().find(|(i, ..)| i == &addr.ip()) {
|
|
|
|
return Ok(ma.clone())
|
|
|
|
}
|
|
|
|
|
|
|
|
// No exact match => check netmask
|
|
|
|
if let Some((_, _, ma)) = addrs.iter().find(|(_, i, _)| i.contains(&addr.ip())) {
|
|
|
|
return Ok(ma.clone())
|
|
|
|
}
|
|
|
|
|
|
|
|
// The local IP address of this socket is new to us.
|
|
|
|
// We need to check for changes in the set of host addresses and report new
|
|
|
|
// and expired addresses.
|
|
|
|
//
|
|
|
|
// TODO: We do not detect expired addresses unless there is a new address.
|
|
|
|
|
|
|
|
let new_addrs = host_addresses(port)?;
|
|
|
|
let old_addrs = std::mem::replace(addrs, new_addrs);
|
|
|
|
|
|
|
|
// Check for addresses no longer in use.
|
|
|
|
for (ip, _, ma) in old_addrs.iter() {
|
|
|
|
if addrs.iter().find(|(i, ..)| i == ip).is_none() {
|
|
|
|
debug!("Expired listen address: {}", ma);
|
|
|
|
pending.push_back(ListenerEvent::AddressExpired(ma.clone()));
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
// Check for new addresses.
|
|
|
|
for (ip, _, ma) in addrs.iter() {
|
|
|
|
if old_addrs.iter().find(|(i, ..)| i == ip).is_none() {
|
|
|
|
debug!("New listen address: {}", ma);
|
|
|
|
pending.push_back(ListenerEvent::NewAddress(ma.clone()));
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
// We should now be able to find the listen address of the local socket address,
|
|
|
|
// if not something is seriously wrong and we report an error.
|
|
|
|
if addrs.iter().find(|(i, j, _)| i == &addr.ip() || j.contains(&addr.ip())).is_none() {
|
|
|
|
let msg = format!("{} does not match any listen address", addr.ip());
|
|
|
|
return Err(io::Error::new(io::ErrorKind::Other, msg))
|
|
|
|
}
|
|
|
|
|
|
|
|
Ok(ip_to_multiaddr(addr.ip(), port))
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2018-08-09 14:04:15 +02:00
|
|
|
impl Stream for TcpListenStream {
|
2019-04-10 10:29:21 +02:00
|
|
|
type Item = ListenerEvent<FutureResult<TcpTransStream, io::Error>>;
|
2019-01-10 11:27:06 +01:00
|
|
|
type Error = io::Error;
|
2018-08-09 14:04:15 +02:00
|
|
|
|
2019-04-10 10:29:21 +02:00
|
|
|
fn poll(&mut self) -> Poll<Option<Self::Item>, io::Error> {
|
2018-08-09 14:04:15 +02:00
|
|
|
let inner = match self.inner {
|
|
|
|
Ok(ref mut inc) => inc,
|
2019-04-11 22:51:07 +02:00
|
|
|
Err(ref mut err) => return Err(err.take().expect("poll called again after error"))
|
2018-08-09 14:04:15 +02:00
|
|
|
};
|
|
|
|
|
2018-10-17 10:17:40 +01:00
|
|
|
loop {
|
2019-04-11 22:51:07 +02:00
|
|
|
if let Some(event) = self.pending.pop_front() {
|
|
|
|
return Ok(Async::Ready(Some(event)))
|
|
|
|
}
|
|
|
|
|
|
|
|
let sock = match inner.poll() {
|
|
|
|
Ok(Async::Ready(Some(sock))) => sock,
|
|
|
|
Ok(Async::Ready(None)) => return Ok(Async::Ready(None)),
|
|
|
|
Ok(Async::NotReady) => return Ok(Async::NotReady),
|
|
|
|
Err(()) => unreachable!("sleep_on_error never produces an error")
|
|
|
|
};
|
|
|
|
|
|
|
|
let sock_addr = match sock.peer_addr() {
|
|
|
|
Ok(addr) => addr,
|
|
|
|
Err(err) => {
|
|
|
|
error!("Failed to get peer address: {:?}", err);
|
|
|
|
return Err(err)
|
|
|
|
}
|
|
|
|
};
|
|
|
|
|
|
|
|
let listen_addr = match sock.local_addr() {
|
2019-06-03 17:55:15 +02:00
|
|
|
Ok(addr) => map_addr(&addr, &mut self.addrs, &mut self.pending, self.port)?,
|
2019-04-11 22:51:07 +02:00
|
|
|
Err(err) => {
|
|
|
|
error!("Failed to get local address of incoming socket: {:?}", err);
|
|
|
|
return Err(err)
|
|
|
|
}
|
|
|
|
};
|
|
|
|
|
2019-06-03 17:55:15 +02:00
|
|
|
let remote_addr = ip_to_multiaddr(sock_addr.ip(), sock_addr.port());
|
2019-04-11 22:51:07 +02:00
|
|
|
|
|
|
|
match apply_config(&self.config, &sock) {
|
|
|
|
Ok(()) => {
|
2019-05-15 14:48:26 +02:00
|
|
|
trace!("Incoming connection from {} on {}", remote_addr, listen_addr);
|
2019-04-11 22:51:07 +02:00
|
|
|
self.pending.push_back(ListenerEvent::Upgrade {
|
2019-04-10 10:29:21 +02:00
|
|
|
upgrade: future::ok(TcpTransStream { inner: sock }),
|
2019-04-11 22:51:07 +02:00
|
|
|
listen_addr,
|
|
|
|
remote_addr
|
|
|
|
})
|
|
|
|
}
|
|
|
|
Err(err) => {
|
2019-05-15 14:48:26 +02:00
|
|
|
debug!("Error upgrading incoming connection from {}: {:?}", remote_addr, err);
|
2019-04-11 22:51:07 +02:00
|
|
|
self.pending.push_back(ListenerEvent::Upgrade {
|
|
|
|
upgrade: future::err(err),
|
|
|
|
listen_addr,
|
|
|
|
remote_addr
|
|
|
|
})
|
2018-10-17 10:17:40 +01:00
|
|
|
}
|
2018-08-09 14:04:15 +02:00
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
impl fmt::Debug for TcpListenStream {
|
2019-02-11 14:58:15 +01:00
|
|
|
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
|
2018-08-09 14:04:15 +02:00
|
|
|
match self.inner {
|
|
|
|
Ok(_) => write!(f, "TcpListenStream"),
|
|
|
|
Err(None) => write!(f, "TcpListenStream(Errored)"),
|
|
|
|
Err(Some(ref err)) => write!(f, "TcpListenStream({:?})", err),
|
2017-10-23 11:45:35 +02:00
|
|
|
}
|
|
|
|
}
|
2017-09-18 16:52:51 +02:00
|
|
|
}
|
|
|
|
|
2018-07-21 13:01:59 +02:00
|
|
|
/// Wraps around a `TcpStream` and adds logging for important events.
|
2018-08-09 14:04:15 +02:00
|
|
|
#[derive(Debug)]
|
2018-07-21 13:01:59 +02:00
|
|
|
pub struct TcpTransStream {
|
2018-08-09 14:04:15 +02:00
|
|
|
inner: TcpStream,
|
2018-07-21 13:01:59 +02:00
|
|
|
}
|
|
|
|
|
|
|
|
impl Read for TcpTransStream {
|
2019-01-10 11:27:06 +01:00
|
|
|
fn read(&mut self, buf: &mut [u8]) -> Result<usize, io::Error> {
|
2018-07-21 13:01:59 +02:00
|
|
|
self.inner.read(buf)
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2019-05-10 11:26:18 +02:00
|
|
|
impl AsyncRead for TcpTransStream {
|
|
|
|
unsafe fn prepare_uninitialized_buffer(&self, buf: &mut [u8]) -> bool {
|
|
|
|
self.inner.prepare_uninitialized_buffer(buf)
|
|
|
|
}
|
|
|
|
|
|
|
|
fn read_buf<B: bytes::BufMut>(&mut self, buf: &mut B) -> Poll<usize, io::Error> {
|
|
|
|
self.inner.read_buf(buf)
|
|
|
|
}
|
|
|
|
}
|
2018-07-21 13:01:59 +02:00
|
|
|
|
|
|
|
impl Write for TcpTransStream {
|
2019-01-10 11:27:06 +01:00
|
|
|
fn write(&mut self, buf: &[u8]) -> Result<usize, io::Error> {
|
2018-07-21 13:01:59 +02:00
|
|
|
self.inner.write(buf)
|
|
|
|
}
|
|
|
|
|
2019-01-10 11:27:06 +01:00
|
|
|
fn flush(&mut self) -> Result<(), io::Error> {
|
2018-07-21 13:01:59 +02:00
|
|
|
self.inner.flush()
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
impl AsyncWrite for TcpTransStream {
|
2019-01-10 11:27:06 +01:00
|
|
|
fn shutdown(&mut self) -> Poll<(), io::Error> {
|
2018-07-21 13:01:59 +02:00
|
|
|
AsyncWrite::shutdown(&mut self.inner)
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
impl Drop for TcpTransStream {
|
|
|
|
fn drop(&mut self) {
|
|
|
|
if let Ok(addr) = self.inner.peer_addr() {
|
|
|
|
debug!("Dropped TCP connection to {:?}", addr);
|
|
|
|
} else {
|
|
|
|
debug!("Dropped TCP connection to undeterminate peer");
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2017-09-30 15:55:57 +02:00
|
|
|
#[cfg(test)]
|
|
|
|
mod tests {
|
2019-06-03 17:55:15 +02:00
|
|
|
use futures::prelude::*;
|
|
|
|
use libp2p_core::{Transport, multiaddr::{Multiaddr, Protocol}, transport::ListenerEvent};
|
2018-05-14 15:55:16 +02:00
|
|
|
use std::net::{IpAddr, Ipv4Addr, SocketAddr};
|
2019-06-03 17:55:15 +02:00
|
|
|
use super::{multiaddr_to_socketaddr, TcpConfig};
|
|
|
|
use tokio::runtime::current_thread::Runtime;
|
2018-05-14 15:55:16 +02:00
|
|
|
use tokio_io;
|
2017-10-23 11:45:35 +02:00
|
|
|
|
2019-04-11 22:51:07 +02:00
|
|
|
#[test]
|
|
|
|
fn wildcard_expansion() {
|
|
|
|
let mut listener = TcpConfig::new()
|
|
|
|
.listen_on("/ip4/0.0.0.0/tcp/0".parse().unwrap())
|
|
|
|
.expect("listener");
|
|
|
|
|
|
|
|
// Get the first address.
|
|
|
|
let addr = listener.by_ref()
|
|
|
|
.wait()
|
|
|
|
.next()
|
|
|
|
.expect("some event")
|
|
|
|
.expect("no error")
|
|
|
|
.into_new_address()
|
|
|
|
.expect("listen address");
|
|
|
|
|
|
|
|
// Process all initial `NewAddress` events and make sure they
|
|
|
|
// do not contain wildcard address or port.
|
|
|
|
let server = listener
|
|
|
|
.take_while(|event| match event {
|
|
|
|
ListenerEvent::NewAddress(a) => {
|
|
|
|
let mut iter = a.iter();
|
|
|
|
match iter.next().expect("ip address") {
|
|
|
|
Protocol::Ip4(ip) => assert!(!ip.is_unspecified()),
|
|
|
|
Protocol::Ip6(ip) => assert!(!ip.is_unspecified()),
|
|
|
|
other => panic!("Unexpected protocol: {}", other)
|
|
|
|
}
|
|
|
|
if let Protocol::Tcp(port) = iter.next().expect("port") {
|
|
|
|
assert_ne!(0, port)
|
|
|
|
} else {
|
|
|
|
panic!("No TCP port in address: {}", a)
|
|
|
|
}
|
|
|
|
Ok(true)
|
|
|
|
}
|
|
|
|
_ => Ok(false)
|
|
|
|
})
|
|
|
|
.for_each(|_| Ok(()));
|
|
|
|
|
|
|
|
let client = TcpConfig::new().dial(addr).expect("dialer");
|
|
|
|
tokio::run(server.join(client).map(|_| ()).map_err(|e| panic!("error: {}", e)))
|
|
|
|
}
|
|
|
|
|
2017-10-23 11:45:35 +02:00
|
|
|
#[test]
|
|
|
|
fn multiaddr_to_tcp_conversion() {
|
|
|
|
use std::net::Ipv6Addr;
|
|
|
|
|
2018-03-07 16:20:55 +01:00
|
|
|
assert!(
|
|
|
|
multiaddr_to_socketaddr(&"/ip4/127.0.0.1/udp/1234".parse::<Multiaddr>().unwrap())
|
|
|
|
.is_err()
|
|
|
|
);
|
2017-11-16 23:59:38 +08:00
|
|
|
|
2017-10-23 11:45:35 +02:00
|
|
|
assert_eq!(
|
2017-12-28 18:07:49 +01:00
|
|
|
multiaddr_to_socketaddr(&"/ip4/127.0.0.1/tcp/12345".parse::<Multiaddr>().unwrap()),
|
2017-10-23 11:45:35 +02:00
|
|
|
Ok(SocketAddr::new(
|
|
|
|
IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)),
|
|
|
|
12345,
|
|
|
|
))
|
|
|
|
);
|
|
|
|
assert_eq!(
|
2018-08-09 14:04:15 +02:00
|
|
|
multiaddr_to_socketaddr(
|
|
|
|
&"/ip4/255.255.255.255/tcp/8080"
|
|
|
|
.parse::<Multiaddr>()
|
|
|
|
.unwrap()
|
|
|
|
),
|
2017-10-23 11:45:35 +02:00
|
|
|
Ok(SocketAddr::new(
|
|
|
|
IpAddr::V4(Ipv4Addr::new(255, 255, 255, 255)),
|
|
|
|
8080,
|
|
|
|
))
|
|
|
|
);
|
|
|
|
assert_eq!(
|
2017-12-28 18:07:49 +01:00
|
|
|
multiaddr_to_socketaddr(&"/ip6/::1/tcp/12345".parse::<Multiaddr>().unwrap()),
|
2017-10-23 11:45:35 +02:00
|
|
|
Ok(SocketAddr::new(
|
|
|
|
IpAddr::V6(Ipv6Addr::new(0, 0, 0, 0, 0, 0, 0, 1)),
|
|
|
|
12345,
|
|
|
|
))
|
|
|
|
);
|
|
|
|
assert_eq!(
|
2018-08-09 14:04:15 +02:00
|
|
|
multiaddr_to_socketaddr(
|
|
|
|
&"/ip6/ffff:ffff:ffff:ffff:ffff:ffff:ffff:ffff/tcp/8080"
|
|
|
|
.parse::<Multiaddr>()
|
|
|
|
.unwrap()
|
|
|
|
),
|
2017-10-23 11:45:35 +02:00
|
|
|
Ok(SocketAddr::new(
|
|
|
|
IpAddr::V6(Ipv6Addr::new(
|
2018-07-11 11:14:40 +02:00
|
|
|
65535, 65535, 65535, 65535, 65535, 65535, 65535, 65535,
|
2017-10-23 11:45:35 +02:00
|
|
|
)),
|
|
|
|
8080,
|
|
|
|
))
|
|
|
|
);
|
|
|
|
}
|
|
|
|
|
|
|
|
#[test]
|
|
|
|
fn communicating_between_dialer_and_listener() {
|
|
|
|
use std::io::Write;
|
|
|
|
|
|
|
|
std::thread::spawn(move || {
|
2017-12-28 18:07:49 +01:00
|
|
|
let addr = "/ip4/127.0.0.1/tcp/12345".parse::<Multiaddr>().unwrap();
|
2018-07-16 12:15:27 +02:00
|
|
|
let tcp = TcpConfig::new();
|
2018-10-25 05:26:37 -04:00
|
|
|
let mut rt = Runtime::new().unwrap();
|
|
|
|
let handle = rt.handle();
|
2019-04-10 10:29:21 +02:00
|
|
|
let listener = tcp.listen_on(addr).unwrap()
|
|
|
|
.filter_map(ListenerEvent::into_upgrade)
|
|
|
|
.for_each(|(sock, _)| {
|
|
|
|
sock.and_then(|sock| {
|
|
|
|
// Define what to do with the socket that just connected to us
|
|
|
|
// Which in this case is read 3 bytes
|
|
|
|
let handle_conn = tokio_io::io::read_exact(sock, [0; 3])
|
|
|
|
.map(|(_, buf)| assert_eq!(buf, [1, 2, 3]))
|
|
|
|
.map_err(|err| panic!("IO error {:?}", err));
|
|
|
|
|
|
|
|
// Spawn the future as a concurrent task
|
|
|
|
handle.spawn(handle_conn).unwrap();
|
|
|
|
|
|
|
|
Ok(())
|
|
|
|
})
|
|
|
|
});
|
2017-10-23 11:45:35 +02:00
|
|
|
|
2018-10-25 05:26:37 -04:00
|
|
|
rt.block_on(listener).unwrap();
|
|
|
|
rt.run().unwrap();
|
2017-10-23 11:45:35 +02:00
|
|
|
});
|
|
|
|
std::thread::sleep(std::time::Duration::from_millis(100));
|
2017-12-28 18:07:49 +01:00
|
|
|
let addr = "/ip4/127.0.0.1/tcp/12345".parse::<Multiaddr>().unwrap();
|
2018-07-16 12:15:27 +02:00
|
|
|
let tcp = TcpConfig::new();
|
2017-10-23 11:45:35 +02:00
|
|
|
// Obtain a future socket through dialing
|
|
|
|
let socket = tcp.dial(addr.clone()).unwrap();
|
|
|
|
// Define what to do with the socket once it's obtained
|
2018-07-16 12:15:27 +02:00
|
|
|
let action = socket.then(|sock| -> Result<(), ()> {
|
2018-10-17 10:17:40 +01:00
|
|
|
sock.unwrap().write(&[0x1, 0x2, 0x3]).unwrap();
|
2018-07-16 12:15:27 +02:00
|
|
|
Ok(())
|
2017-10-23 11:45:35 +02:00
|
|
|
});
|
|
|
|
// Execute the future in our event loop
|
2018-10-25 05:26:37 -04:00
|
|
|
let mut rt = Runtime::new().unwrap();
|
|
|
|
let _ = rt.block_on(action).unwrap();
|
2017-10-23 11:45:35 +02:00
|
|
|
}
|
2017-11-24 16:10:34 +01:00
|
|
|
|
|
|
|
#[test]
|
2018-01-02 16:00:08 +01:00
|
|
|
fn replace_port_0_in_returned_multiaddr_ipv4() {
|
2018-07-16 12:15:27 +02:00
|
|
|
let tcp = TcpConfig::new();
|
2017-11-24 16:10:34 +01:00
|
|
|
|
2017-12-28 18:07:49 +01:00
|
|
|
let addr = "/ip4/127.0.0.1/tcp/0".parse::<Multiaddr>().unwrap();
|
2017-11-24 16:10:34 +01:00
|
|
|
assert!(addr.to_string().contains("tcp/0"));
|
|
|
|
|
2019-04-10 10:29:21 +02:00
|
|
|
let new_addr = tcp.listen_on(addr).unwrap().wait()
|
|
|
|
.next()
|
|
|
|
.expect("some event")
|
|
|
|
.expect("no error")
|
|
|
|
.into_new_address()
|
|
|
|
.expect("listen address");
|
|
|
|
|
2017-11-24 16:10:34 +01:00
|
|
|
assert!(!new_addr.to_string().contains("tcp/0"));
|
|
|
|
}
|
2017-12-15 17:29:54 +01:00
|
|
|
|
2018-01-02 16:00:08 +01:00
|
|
|
#[test]
|
|
|
|
fn replace_port_0_in_returned_multiaddr_ipv6() {
|
2018-07-16 12:15:27 +02:00
|
|
|
let tcp = TcpConfig::new();
|
2018-01-02 16:00:08 +01:00
|
|
|
|
2017-12-28 18:07:49 +01:00
|
|
|
let addr: Multiaddr = "/ip6/::1/tcp/0".parse().unwrap();
|
2018-01-02 16:00:08 +01:00
|
|
|
assert!(addr.to_string().contains("tcp/0"));
|
|
|
|
|
2019-04-10 10:29:21 +02:00
|
|
|
let new_addr = tcp.listen_on(addr).unwrap().wait()
|
|
|
|
.next()
|
|
|
|
.expect("some event")
|
|
|
|
.expect("no error")
|
|
|
|
.into_new_address()
|
|
|
|
.expect("listen address");
|
|
|
|
|
2018-01-02 16:00:08 +01:00
|
|
|
assert!(!new_addr.to_string().contains("tcp/0"));
|
|
|
|
}
|
|
|
|
|
2017-12-15 17:29:54 +01:00
|
|
|
#[test]
|
|
|
|
fn larger_addr_denied() {
|
2018-07-16 12:15:27 +02:00
|
|
|
let tcp = TcpConfig::new();
|
2017-12-15 17:29:54 +01:00
|
|
|
|
2018-03-07 16:20:55 +01:00
|
|
|
let addr = "/ip4/127.0.0.1/tcp/12345/tcp/12345"
|
|
|
|
.parse::<Multiaddr>()
|
|
|
|
.unwrap();
|
2017-12-15 17:29:54 +01:00
|
|
|
assert!(tcp.listen_on(addr).is_err());
|
|
|
|
}
|
2017-09-18 16:52:51 +02:00
|
|
|
}
|