/* This file is part of DarkFi (https://dark.fi)
*
* Copyright (C) 2020-2024 Dyne.org foundation
*
* This program is free software: you can redistribute it and/or modify
* it under the terms of the GNU Affero General Public License as
* published by the Free Software Foundation, either version 3 of the
* License, or (at your option) any later version.
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU Affero General Public License for more details.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see .
*/
use std::{sync::Arc, time::UNIX_EPOCH};
use async_trait::async_trait;
use log::debug;
use smol::Executor;
use super::{
super::{
channel::ChannelPtr,
hosts::{HostColor, HostsPtr},
message::{AddrsMessage, GetAddrsMessage},
message_publisher::MessageSubscription,
p2p::P2pPtr,
session::SESSION_OUTBOUND,
settings::SettingsPtr,
},
protocol_base::{ProtocolBase, ProtocolBasePtr},
protocol_jobs_manager::{ProtocolJobsManager, ProtocolJobsManagerPtr},
};
use crate::{Error, Result};
/// Defines address and get-address messages. On receiving GetAddr, nodes
/// reply an AddrMessage containing nodes from their hostlist. On receiving
/// an AddrMessage, nodes enter the info into their greylists.
///
/// The node selection logic for creating an AddrMessage is as follows:
///
/// 1. First select nodes matching the requested transports from the
/// anchorlist. These nodes have the highest guarantee of being reachable,
/// so we prioritize them first.
///
/// 2. Then select nodes matching the requested transports from the
/// whitelist.
///
/// 3. Next select whitelist nodes that don't match our transports. We do
/// this so that nodes share and propagate nodes of different transports,
/// even if they can't connect to them themselves.
///
/// 4. Finally, if there's still space available, fill the remaining vector
/// space with darklist entries. This is necessary to propagate transports
/// that neither this node nor the receiving node support.
pub struct ProtocolAddress {
channel: ChannelPtr,
addrs_sub: MessageSubscription,
get_addrs_sub: MessageSubscription,
hosts: HostsPtr,
settings: SettingsPtr,
jobsman: ProtocolJobsManagerPtr,
}
const PROTO_NAME: &str = "ProtocolAddress";
/// A vector of all currently accepted transports and valid transport
/// combinations. Should be updated if and when new transports are
/// added. Creates a upper bound on the number of transports a given peer
/// can request.
const TRANSPORT_COMBOS: [&str; 7] = ["tor", "tls", "tcp", "nym", "tor+tls", "nym+tls", "tcp+tls"];
impl ProtocolAddress {
/// Creates a new address protocol. Makes an address, an external address
/// and a get-address subscription and adds them to the address protocol
/// instance.
pub async fn init(channel: ChannelPtr, p2p: P2pPtr) -> ProtocolBasePtr {
let settings = p2p.settings();
let hosts = p2p.hosts();
// Creates a subscription to address message
let addrs_sub =
channel.subscribe_msg::().await.expect("Missing addrs dispatcher!");
// Creates a subscription to get-address message
let get_addrs_sub =
channel.subscribe_msg::().await.expect("Missing getaddrs dispatcher!");
Arc::new(Self {
channel: channel.clone(),
addrs_sub,
get_addrs_sub,
hosts,
jobsman: ProtocolJobsManager::new(PROTO_NAME, channel),
settings,
})
}
/// Handles receiving the address message. Loops to continually receive
/// address messages on the address subscription. Validates and adds the
/// received addresses to the greylist.
async fn handle_receive_addrs(self: Arc) -> Result<()> {
debug!(
target: "net::protocol_address::handle_receive_addrs()",
"[START] address={}", self.channel.address(),
);
loop {
let addrs_msg = self.addrs_sub.receive().await?;
debug!(
target: "net::protocol_address::handle_receive_addrs()",
"Received {} addrs from {}", addrs_msg.addrs.len(), self.channel.address(),
);
debug!(
target: "net::protocol_address::handle_receive_addrs()",
"Appending to greylist...",
);
self.hosts.insert(HostColor::Grey, &addrs_msg.addrs).await;
}
}
/// Handles receiving the get-address message. Continually receives
/// get-address messages on the get-address subscription. Then replies
/// with an address message.
async fn handle_receive_get_addrs(self: Arc) -> Result<()> {
debug!(
target: "net::protocol_address::handle_receive_get_addrs()",
"[START] address={}", self.channel.address(),
);
loop {
let get_addrs_msg = self.get_addrs_sub.receive().await?;
debug!(
target: "net::protocol_address::handle_receive_get_addrs()",
"Received GetAddrs({}) message from {}", get_addrs_msg.max, self.channel.address(),
);
// Check that this peer isn't requesting more transports than we support
// (the max number of all transports, plus mixing).
if get_addrs_msg.transports.len() > TRANSPORT_COMBOS.len() {
return Err(Error::InvalidTransportRequest);
}
// First we grab address with the requested transports from the gold list
debug!(target: "net::protocol_address::handle_receive_get_addrs()",
"Fetching gold entries with schemes");
let mut addrs = self
.hosts
.container
.fetch_n_random_with_schemes(
HostColor::Gold,
&get_addrs_msg.transports,
get_addrs_msg.max,
)
.await;
// Then we grab address with the requested transports from the whitelist
debug!(target: "net::protocol_address::handle_receive_get_addrs()",
"Fetching whitelist entries with schemes");
addrs.append(
&mut self
.hosts
.container
.fetch_n_random_with_schemes(
HostColor::White,
&get_addrs_msg.transports,
get_addrs_msg.max,
)
.await,
);
// Next we grab addresses without the requested transports
// to fill a 2 * max length vector.
// Then we grab address without the requested transports from the gold list
debug!(target: "net::protocol_address::handle_receive_get_addrs()",
"Fetching gold entries without schemes");
let remain = 2 * get_addrs_msg.max - addrs.len() as u32;
addrs.append(
&mut self
.hosts
.container
.fetch_n_random_excluding_schemes(
HostColor::Gold,
&get_addrs_msg.transports,
remain,
)
.await,
);
// Then we grab address without the requested transports from the white list
debug!(target: "net::protocol_address::handle_receive_get_addrs()",
"Fetching white entries without schemes");
let remain = 2 * get_addrs_msg.max - addrs.len() as u32;
addrs.append(
&mut self
.hosts
.container
.fetch_n_random_excluding_schemes(
HostColor::White,
&get_addrs_msg.transports,
remain,
)
.await,
);
// If there's still space available, take from the Dark list.
/* NOTE: We share peers from our Dark list because to ensure
that non-compatiable transports are shared with other nodes
so that they propagate on the network even if they're not
popular transports. */
debug!(target: "net::protocol_address::handle_receive_get_addrs()",
"Fetching dark entries");
let remain = 2 * get_addrs_msg.max - addrs.len() as u32;
addrs.append(&mut self.hosts.container.fetch_n_random(HostColor::Dark, remain).await);
debug!(
target: "net::protocol_address::handle_receive_get_addrs()",
"Sending {} addresses to {}", addrs.len(), self.channel.address(),
);
let addrs_msg = AddrsMessage { addrs };
self.channel.send(&addrs_msg).await?;
}
}
/// Send our own external addresses over a channel. Set the
/// last_seen field to now.
async fn send_my_addrs(self: Arc) -> Result<()> {
debug!(
target: "net::protocol_address::send_my_addrs()",
"[START] channel address={}", self.channel.address(),
);
let type_id = self.channel.session_type_id();
if type_id != SESSION_OUTBOUND {
debug!(target: "net::protocol_address::send_my_addrs()",
"Not an outbound session. Stopping");
return Ok(())
}
if self.settings.external_addrs.is_empty() {
debug!(target: "net::protocol_address::send_my_addrs()",
"External addr not configured. Stopping");
return Ok(())
}
let mut addrs = vec![];
for addr in self.settings.external_addrs.clone() {
let last_seen = UNIX_EPOCH.elapsed().unwrap().as_secs();
addrs.push((addr, last_seen));
}
debug!(target: "net::protocol_address::send_my_addrs()",
"Broadcasting {} addresses", addrs.len());
let ext_addr_msg = AddrsMessage { addrs };
self.channel.send(&ext_addr_msg).await?;
debug!(target: "net::protocol_address::send_my_addrs()",
"[END] channel address={}", self.channel.address());
Ok(())
}
}
#[async_trait]
impl ProtocolBase for ProtocolAddress {
/// Start the address protocol. If it's an outbound session and has an
/// external address, send our external address. Run receive address
/// and get address protocols on the protocol task manager. Then send
/// get-address msg.
async fn start(self: Arc, ex: Arc>) -> Result<()> {
debug!(target: "net::protocol_address::start()",
"START => address={}", self.channel.address());
self.jobsman.clone().start(ex.clone());
self.jobsman.clone().spawn(self.clone().send_my_addrs(), ex.clone()).await;
self.jobsman.clone().spawn(self.clone().handle_receive_addrs(), ex.clone()).await;
self.jobsman.spawn(self.clone().handle_receive_get_addrs(), ex).await;
// Send get_address message.
let get_addrs = GetAddrsMessage {
max: self.settings.outbound_connections as u32,
transports: self.settings.allowed_transports.clone(),
};
self.channel.send(&get_addrs).await?;
debug!(target: "net::protocol_address::start()",
"END => address={}", self.channel.address());
Ok(())
}
fn name(&self) -> &'static str {
PROTO_NAME
}
}