/* 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::{
collections::{HashMap, HashSet},
sync::Arc,
};
use log::debug;
use rand::{prelude::IteratorRandom, rngs::OsRng};
use smol::lock::RwLock;
use url::Url;
use super::settings::SettingsPtr;
use crate::{
system::{Subscriber, SubscriberPtr, Subscription},
Result,
};
/// Atomic pointer to hosts object
pub type HostsPtr = Arc;
// An array containing all possible local host strings
// TODO: This could perhaps be more exhaustive?
pub const LOCAL_HOST_STRS: [&str; 2] = ["localhost", "localhost.localdomain"];
/// Manages a store of network addresses
pub struct Hosts {
/// Set of stored addresses
addrs: RwLock>,
/// Set of stored addresses that are quarantined.
/// We quarantine peers we've been unable to connect to, but we keep them
/// around so we can potentially try them again, up to n tries. This should
/// be helpful in order to self-heal the p2p connections in case we have an
/// Internet interrupt (goblins unplugging cables)
quarantine: RwLock>,
/// Peers we reject from connecting
rejected: RwLock>,
/// Subscriber listening for store updates
store_subscriber: SubscriberPtr,
/// Pointer to configured P2P settings
settings: SettingsPtr,
}
impl Hosts {
/// Create a new hosts list>
pub fn new(settings: SettingsPtr) -> HostsPtr {
Arc::new(Self {
addrs: RwLock::new(HashSet::new()),
quarantine: RwLock::new(HashMap::new()),
rejected: RwLock::new(HashSet::new()),
store_subscriber: Subscriber::new(),
settings,
})
}
/// Append given addrs to the known set.
pub async fn store(&self, addrs: &[Url]) {
debug!(target: "net::hosts::store()", "hosts::store() [START]");
let filtered_addrs = self.filter_addresses(addrs).await;
let filtered_addrs_len = filtered_addrs.len();
if !filtered_addrs.is_empty() {
let mut addrs_map = self.addrs.write().await;
for addr in filtered_addrs {
debug!(target: "net::hosts::store()", "Inserting {}", addr);
addrs_map.insert(addr);
}
}
self.store_subscriber.notify(filtered_addrs_len).await;
debug!(target: "net::hosts::store()", "hosts::store() [END]");
}
pub async fn subscribe_store(&self) -> Result> {
let sub = self.store_subscriber.clone().subscribe().await;
Ok(sub)
}
// Verify whether a URL is local.
// NOTE: This function is stateless and not specific to
// `Hosts`. For this reason, it might make more sense
// to move this function to a more appropriate location
// in the codebase.
pub async fn is_local_host(&self, url: Url) -> bool {
// Reject Urls without host strings.
if url.host_str().is_none() {
return false
}
// We do this hack in order to parse IPs properly.
// https://github.com/whatwg/url/issues/749
let addr = Url::parse(&url.as_str().replace(url.scheme(), "http")).unwrap();
// Filter private IP ranges
match addr.host().unwrap() {
url::Host::Ipv4(ip) => {
if !ip.is_global() {
return true
}
}
url::Host::Ipv6(ip) => {
if !ip.is_global() {
return true
}
}
url::Host::Domain(d) => {
if LOCAL_HOST_STRS.contains(&d) {
return true
}
}
}
false
}
/// Filter given addresses based on certain rulesets and validity.
async fn filter_addresses(&self, addrs: &[Url]) -> Vec {
debug!(target: "net::hosts::filter_addresses()", "Filtering addrs: {:?}", addrs);
let mut ret = vec![];
let localnet = self.settings.localnet;
'addr_loop: for addr_ in addrs {
// Validate that the format is `scheme://host_str:port`
if addr_.host_str().is_none() ||
addr_.port().is_none() ||
addr_.cannot_be_a_base() ||
addr_.path_segments().is_some()
{
continue
}
if self.is_rejected(addr_).await {
debug!(target: "net::hosts::filter_addresses()", "Peer {} is rejected", addr_);
continue
}
let host_str = addr_.host_str().unwrap();
if !localnet {
// Our own external addresses should never enter the hosts set.
for ext in &self.settings.external_addrs {
if host_str == ext.host_str().unwrap() {
continue 'addr_loop
}
}
}
// We do this hack in order to parse IPs properly.
// https://github.com/whatwg/url/issues/749
let addr = Url::parse(&addr_.as_str().replace(addr_.scheme(), "http")).unwrap();
// Filter non-global ranges if we're not allowing localnet.
// Should never be allowed in production, so we don't really care
// about some of them (e.g. 0.0.0.0, or broadcast, etc.).
if !localnet && self.is_local_host(addr).await {
continue
}
match addr_.scheme() {
// Validate that the address is an actual onion.
#[cfg(feature = "p2p-tor")]
"tor" | "tor+tls" => {
use std::str::FromStr;
if tor_hscrypto::pk::HsId::from_str(host_str).is_err() {
continue
}
debug!(target: "net::hosts::filter_addresses()", "[Tor] Valid: {}", host_str);
}
#[cfg(feature = "p2p-nym")]
"nym" | "nym+tls" => continue, // <-- Temp skip
#[cfg(feature = "p2p-tcp")]
"tcp" | "tcp+tls" => {
debug!(target: "net::hosts::filter_addresses()", "[TCP] Valid: {}", host_str);
}
_ => continue,
}
ret.push(addr_.clone());
}
ret
}
pub async fn remove(&self, url: &Url) {
debug!(target: "net::hosts::remove()", "Removing peer {}", url);
self.addrs.write().await.remove(url);
self.quarantine.write().await.remove(url);
}
/// Quarantine a peer.
/// If they've been quarantined for more than a configured limit, forget them.
pub async fn quarantine(&self, url: &Url) {
debug!(target: "net::hosts::remove()", "Quarantining peer {}", url);
// Remove from main hosts set
self.addrs.write().await.remove(url);
let mut q = self.quarantine.write().await;
if let Some(retries) = q.get_mut(url) {
*retries += 1;
debug!(target: "net::hosts::quarantine()", "Peer {} quarantined {} times", url, retries);
if *retries == self.settings.hosts_quarantine_limit {
debug!(target: "net::hosts::quarantine()", "Banning peer {}", url);
q.remove(url);
self.mark_rejected(url).await;
}
} else {
debug!(target: "net::hosts::remove()", "Added peer {} to quarantine", url);
q.insert(url.clone(), 0);
}
}
/// Check if a given peer (URL) is in the set of rejected hosts
pub async fn is_rejected(&self, peer: &Url) -> bool {
// Skip lookup for UNIX sockets and localhost connections
// as they should never belong to the list of rejected URLs.
let Some(hostname) = peer.host_str() else { return false };
if self.is_local_host(peer.clone()).await {
return false
}
self.rejected.read().await.contains(hostname)
}
/// Mark a peer as rejected by adding it to the set of rejected URLs.
pub async fn mark_rejected(&self, peer: &Url) {
// We ignore UNIX sockets here so we will just work
// with stuff that has host_str().
if let Some(hostname) = peer.host_str() {
// Localhost connections should not be rejected
// This however allows any Tor and Nym connections.
if self.is_local_host(peer.clone()).await {
return
}
self.rejected.write().await.insert(hostname.to_string());
}
}
/// Unmark a rejected peer
pub async fn unmark_rejected(&self, peer: &Url) {
if let Some(hostname) = peer.host_str() {
self.rejected.write().await.remove(hostname);
}
}
/// Check if the host list is empty.
pub async fn is_empty(&self) -> bool {
self.addrs.read().await.is_empty()
}
/// Check if host is already in the set
pub async fn contains(&self, addr: &Url) -> bool {
self.addrs.read().await.contains(addr)
}
/// Return all known hosts
pub async fn fetch_all(&self) -> Vec {
self.addrs.read().await.iter().cloned().collect()
}
/// Get up to n random peers from the hosts set.
pub async fn fetch_n_random(&self, n: u32) -> Vec {
let n = n as usize;
if n == 0 {
return vec![]
}
let addrs = self.addrs.read().await;
let urls = addrs.iter().choose_multiple(&mut OsRng, n.min(addrs.len()));
urls.iter().map(|&url| url.clone()).collect()
}
/// Get up to n random peers that match the given transport schemes from the hosts set.
pub async fn fetch_n_random_with_schemes(&self, schemes: &[String], n: u32) -> Vec {
let n = n as usize;
if n == 0 {
return vec![]
}
// Retrieve all peers corresponding to that transport schemes
let hosts = self.fetch_with_schemes(schemes, None).await;
if hosts.is_empty() {
return hosts
}
// Grab random ones
let urls = hosts.iter().choose_multiple(&mut OsRng, n.min(hosts.len()));
urls.iter().map(|&url| url.clone()).collect()
}
/// Get up to n random peers that don't match the given transport schemes from the hosts set.
pub async fn fetch_n_random_excluding_schemes(&self, schemes: &[String], n: u32) -> Vec {
let n = n as usize;
if n == 0 {
return vec![]
}
// Retrieve all peers not corresponding to that transport schemes
let hosts = self.fetch_exluding_schemes(schemes, None).await;
if hosts.is_empty() {
return hosts
}
// Grab random ones
let urls = hosts.iter().choose_multiple(&mut OsRng, n.min(hosts.len()));
urls.iter().map(|&url| url.clone()).collect()
}
/// Get up to limit peers that match the given transport schemes from the hosts set.
/// If limit was not provided, return all matching peers.
pub async fn fetch_with_schemes(&self, schemes: &[String], limit: Option) -> Vec {
let addrs = self.addrs.read().await;
let mut limit = match limit {
Some(l) => l.min(addrs.len()),
None => addrs.len(),
};
let mut ret = vec![];
if limit == 0 {
return ret
}
for addr in addrs.iter() {
if schemes.contains(&addr.scheme().to_string()) {
ret.push(addr.clone());
limit -= 1;
if limit == 0 {
return ret
}
}
}
// If we didn't find any, pick some from the quarantine zone
if ret.is_empty() {
for addr in self.quarantine.read().await.keys() {
if schemes.contains(&addr.scheme().to_string()) {
ret.push(addr.clone());
limit -= 1;
if limit == 0 {
break
}
}
}
}
ret
}
/// Get up to limit peers that don't match the given transport schemes from the hosts set.
/// If limit was not provided, return all matching peers.
pub async fn fetch_exluding_schemes(
&self,
schemes: &[String],
limit: Option,
) -> Vec {
let addrs = self.addrs.read().await;
let mut limit = match limit {
Some(l) => l.min(addrs.len()),
None => addrs.len(),
};
let mut ret = vec![];
if limit == 0 {
return ret
}
for addr in addrs.iter() {
if !schemes.contains(&addr.scheme().to_string()) {
ret.push(addr.clone());
limit -= 1;
if limit == 0 {
return ret
}
}
}
// If we didn't find any, pick some from the quarantine zone
if ret.is_empty() {
for addr in self.quarantine.read().await.keys() {
if !schemes.contains(&addr.scheme().to_string()) {
ret.push(addr.clone());
limit -= 1;
if limit == 0 {
break
}
}
}
}
ret
}
}
#[cfg(test)]
mod tests {
use super::{super::settings::Settings, *};
#[test]
fn test_store_localnet() {
smol::block_on(async {
let settings = Settings {
localnet: true,
external_addrs: vec![
Url::parse("tcp://foo.bar:123").unwrap(),
Url::parse("tcp://lol.cat:321").unwrap(),
],
..Default::default()
};
let hosts = Hosts::new(Arc::new(settings.clone()));
hosts.store(&settings.external_addrs).await;
for i in settings.external_addrs {
assert!(hosts.contains(&i).await);
}
let local_hosts = vec![
Url::parse("tcp://localhost:3921").unwrap(),
Url::parse("tcp://127.0.0.1:23957").unwrap(),
Url::parse("tcp://[::1]:21481").unwrap(),
Url::parse("tcp://192.168.10.65:311").unwrap(),
Url::parse("tcp://0.0.0.0:2312").unwrap(),
Url::parse("tcp://255.255.255.255:2131").unwrap(),
];
hosts.store(&local_hosts).await;
for i in local_hosts {
assert!(hosts.contains(&i).await);
}
let remote_hosts = vec![
Url::parse("tcp://dark.fi:80").unwrap(),
Url::parse("tcp://top.kek:111").unwrap(),
Url::parse("tcp://http.cat:401").unwrap(),
];
hosts.store(&remote_hosts).await;
for i in remote_hosts {
assert!(hosts.contains(&i).await);
}
});
}
#[test]
fn test_store() {
smol::block_on(async {
let settings = Settings {
localnet: false,
external_addrs: vec![
Url::parse("tcp://foo.bar:123").unwrap(),
Url::parse("tcp://lol.cat:321").unwrap(),
],
..Default::default()
};
let hosts = Hosts::new(Arc::new(settings.clone()));
hosts.store(&settings.external_addrs).await;
assert!(hosts.is_empty().await);
let local_hosts = vec![
Url::parse("tcp://localhost:3921").unwrap(),
Url::parse("tor://[::1]:21481").unwrap(),
Url::parse("tcp://192.168.10.65:311").unwrap(),
Url::parse("tcp+tls://0.0.0.0:2312").unwrap(),
Url::parse("tcp://255.255.255.255:2131").unwrap(),
];
hosts.store(&local_hosts).await;
assert!(hosts.is_empty().await);
let remote_hosts = vec![
Url::parse("tcp://dark.fi:80").unwrap(),
Url::parse("tcp://http.cat:401").unwrap(),
Url::parse("tcp://foo.bar:111").unwrap(),
];
hosts.store(&remote_hosts).await;
assert!(hosts.contains(&remote_hosts[0]).await);
assert!(hosts.contains(&remote_hosts[1]).await);
assert!(!hosts.contains(&remote_hosts[2]).await);
});
}
#[test]
fn test_is_local_host() {
smol::block_on(async {
let settings = Settings {
localnet: false,
external_addrs: vec![
Url::parse("tcp://foo.bar:123").unwrap(),
Url::parse("tcp://lol.cat:321").unwrap(),
],
..Default::default()
};
let hosts = Hosts::new(Arc::new(settings.clone()));
let local_hosts: Vec = vec![
Url::parse("tcp://localhost").unwrap(),
Url::parse("tcp://127.0.0.1").unwrap(),
Url::parse("tcp+tls://[::1]").unwrap(),
Url::parse("tcp://localhost.localdomain").unwrap(),
Url::parse("tcp://192.168.10.65").unwrap(),
];
for host in local_hosts {
eprintln!("{}", host);
assert!(hosts.is_local_host(host).await);
}
let remote_hosts: Vec = vec![
Url::parse("https://dyne.org").unwrap(),
Url::parse("tcp://77.168.10.65:2222").unwrap(),
Url::parse("tcp://[2345:0425:2CA1:0000:0000:0567:5673:23b5]").unwrap(),
Url::parse("http://eweiibe6tdjsdprb4px6rqrzzcsi22m4koia44kc5pcjr7nec2rlxyad.onion")
.unwrap(),
];
for host in remote_hosts {
assert!(!(hosts.is_local_host(host).await))
}
});
}
}