use bytes::Bytes;
use log::*;
use serde_derive::{Deserialize, Serialize};
use std::{io, net::SocketAddr, time::Duration};
use tokio::{net::TcpListener, prelude::*, reactor::Handle, runtime::Runtime};
use tokio_codec::{BytesCodec, Decoder};
pub type IpEchoServer = Runtime;
pub const MAX_PORT_COUNT_PER_MESSAGE: usize = 4;
#[derive(Serialize, Deserialize, Default)]
pub(crate) struct IpEchoServerMessage {
tcp_ports: [u16; MAX_PORT_COUNT_PER_MESSAGE], // Fixed size list of ports to avoid vec serde
udp_ports: [u16; MAX_PORT_COUNT_PER_MESSAGE], // Fixed size list of ports to avoid vec serde
}
impl IpEchoServerMessage {
pub fn new(tcp_ports: &[u16], udp_ports: &[u16]) -> Self {
let mut msg = Self::default();
assert!(tcp_ports.len() <= msg.tcp_ports.len());
assert!(udp_ports.len() <= msg.udp_ports.len());
msg.tcp_ports[..tcp_ports.len()].copy_from_slice(tcp_ports);
msg.udp_ports[..udp_ports.len()].copy_from_slice(udp_ports);
msg
}
}
/// Starts a simple TCP server on the given port that echos the IP address of any peer that
/// connects. Used by |get_public_ip_addr|
pub fn ip_echo_server(tcp: std::net::TcpListener) -> IpEchoServer {
info!("bound to {:?}", tcp.local_addr());
let tcp =
TcpListener::from_std(tcp, &Handle::default()).expect("Failed to convert std::TcpListener");
let server = tcp
.incoming()
.map_err(|err| warn!("accept failed: {:?}", err))
.filter_map(|socket| match socket.peer_addr() {
Ok(peer_addr) => {
info!("connection from {:?}", peer_addr);
Some((peer_addr, BytesCodec::new().framed(socket)))
}
Err(err) => {
info!("peer_addr failed for {:?}: {:?}", socket, err);
None
}
})
.for_each(move |(peer_addr, framed)| {
let (writer, reader) = framed.split();
let processor = reader
.and_then(move |data| {
if data.len() < 4 {
return Err(io::Error::new(
io::ErrorKind::Other,
format!("Request too short, received {} bytes", data.len()),
));
}
let request_header: String = data[0..4].iter().map(|b| *b as char).collect();
if request_header != "\0\0\0\0" {
// Explicitly check for HTTP GET/POST requests to more gracefully handle
// the case where a user accidentally tried to use a gossip entrypoint in
// place of a JSON RPC URL:
if request_header == "GET " || request_header == "POST" {
return Ok(None); // None -> Send HTTP error response
}
return Err(io::Error::new(
io::ErrorKind::Other,
format!("Bad request header: {}", request_header),
));
}
let expected_len =
bincode::serialized_size(&IpEchoServerMessage::default()).unwrap() as usize;
let actual_len = data[4..].len();
if actual_len < expected_len {
return Err(io::Error::new(
io::ErrorKind::Other,
format!(
"Request too short, actual {} < expected {}",
actual_len, expected_len
),
));
}
bincode::deserialize::<IpEchoServerMessage>(&data[4..])
.map(Some)
.map_err(|err| {
io::Error::new(
io::ErrorKind::Other,
format!("Failed to deserialize IpEchoServerMessage: {:?}", err),
)
})
})
.and_then(move |maybe_msg| {
match maybe_msg {
None => None, // Send HTTP error response
Some(msg) => {
// Fire a datagram at each non-zero UDP port
if !msg.udp_ports.is_empty() {
match std::net::UdpSocket::bind("0.0.0.0:0") {
Ok(udp_socket) => {
for udp_port in &msg.udp_ports {
if *udp_port != 0 {
match udp_socket.send_to(
&[0],
SocketAddr::from((peer_addr.ip(), *udp_port)),
) {
Ok(_) => debug!(
"Successful send_to udp/{}",
udp_port
),
Err(err) => info!(
"Failed to send_to udp/{}: {}",
udp_port, err
),
}
}
}
}
Err(err) => {
warn!("Failed to bind local udp socket: {}", err);
}
}
}
// Try to connect to each non-zero TCP port
let tcp_futures: Vec<_> =
msg.tcp_ports
.iter()
.filter_map(|tcp_port| {
let tcp_port = *tcp_port;
if tcp_port == 0 {
None
} else {
Some(
tokio::net::TcpStream::connect(&SocketAddr::new(
peer_addr.ip(),
tcp_port,
))
.and_then(move |tcp_stream| {
debug!(
"Connection established to tcp/{}",
tcp_port
);
let _ = tcp_stream
.shutdown(std::net::Shutdown::Both);
Ok(())
})
.timeout(Duration::from_secs(5))
.or_else(move |err| {
Err(io::Error::new(
io::ErrorKind::Other,
format!(
"Connection timeout to {}: {:?}",
tcp_port, err
),
))
}),
)
}
})
.collect();
Some(future::join_all(tcp_futures))
}
}
})
.and_then(move |valid_request| {
if valid_request.is_none() {
Ok(Bytes::from(
"HTTP/1.1 400 Bad Request\nContent-length: 0\n\n",
))
} else {
// "\0\0\0\0" header is added to ensure a valid response will never
// conflict with the first four bytes of a valid HTTP response.
let mut bytes = vec![
0;
4 + bincode::serialized_size(&peer_addr.ip()).unwrap()
as usize
];
bincode::serialize_into(&mut bytes[4..], &peer_addr.ip()).unwrap();
Ok(Bytes::from(bytes))
}
});
let connection = writer
.send_all(processor)
.timeout(Duration::from_secs(5))
.then(|result| {
if let Err(err) = result {
info!("Session failed: {:?}", err);
}
Ok(())
});
tokio::spawn(connection)
});
let mut rt = Runtime::new().expect("Failed to create Runtime");
rt.spawn(server);
rt
}