store.rs 57 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127112811291130113111321133113411351136113711381139114011411142114311441145114611471148114911501151115211531154115511561157115811591160116111621163116411651166116711681169117011711172117311741175117611771178117911801181118211831184118511861187118811891190119111921193119411951196119711981199120012011202120312041205120612071208120912101211121212131214121512161217121812191220122112221223122412251226122712281229123012311232123312341235123612371238123912401241124212431244124512461247124812491250125112521253125412551256125712581259126012611262126312641265126612671268126912701271127212731274127512761277127812791280128112821283128412851286128712881289129012911292129312941295129612971298129913001301130213031304130513061307130813091310131113121313131413151316131713181319132013211322132313241325132613271328132913301331133213331334133513361337133813391340134113421343134413451346134713481349135013511352135313541355135613571358135913601361136213631364136513661367136813691370137113721373137413751376137713781379138013811382138313841385138613871388138913901391139213931394139513961397139813991400140114021403140414051406140714081409141014111412141314141415141614171418141914201421142214231424142514261427142814291430143114321433143414351436143714381439144014411442144314441445144614471448144914501451145214531454145514561457145814591460146114621463146414651466146714681469147014711472147314741475147614771478
  1. /* This file is part of DarkFi (https://dark.fi)
  2. *
  3. * Copyright (C) 2020-2024 Dyne.org foundation
  4. *
  5. * This program is free software: you can redistribute it and/or modify
  6. * it under the terms of the GNU Affero General Public License as
  7. * published by the Free Software Foundation, either version 3 of the
  8. * License, or (at your option) any later version.
  9. *
  10. * This program is distributed in the hope that it will be useful,
  11. * but WITHOUT ANY WARRANTY; without even the implied warranty of
  12. * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
  13. * GNU Affero General Public License for more details.
  14. *
  15. * You should have received a copy of the GNU Affero General Public License
  16. * along with this program. If not, see <https://www.gnu.org/licenses/>.
  17. */
  18. use std::{collections::HashMap, fmt, fs, fs::File, sync::Arc, time::Instant};
  19. use log::{debug, error, info, trace, warn};
  20. use rand::{prelude::IteratorRandom, rngs::OsRng, Rng};
  21. use smol::lock::RwLock;
  22. use url::Url;
  23. use super::super::{settings::SettingsPtr, ChannelPtr};
  24. use crate::{
  25. system::{Subscriber, SubscriberPtr, Subscription},
  26. util::{
  27. file::{load_file, save_file},
  28. path::expand_path,
  29. },
  30. Error, Result,
  31. };
  32. // An array containing all possible local host strings
  33. // TODO: This could perhaps be more exhaustive?
  34. pub const LOCAL_HOST_STRS: [&str; 2] = ["localhost", "localhost.localdomain"];
  35. const WHITELIST_MAX_LEN: usize = 5000;
  36. const GREYLIST_MAX_LEN: usize = 2000;
  37. /// Atomic pointer to hosts object
  38. pub type HostsPtr = Arc<Hosts>;
  39. /// Keeps track of hosts and their current state. Prevents race conditions
  40. /// where multiple threads are simultaneously trying to change the state of
  41. /// a given host.
  42. pub type HostRegistry = RwLock<HashMap<Url, HostState>>;
  43. /// HostState is a set of mutually exclusive states that can be Insert,
  44. /// Refine, Move, Connect, Suspend or Connected. The state is `None` when the
  45. /// corresponding host has been removed from the HostRegistry.
  46. /// ```
  47. /// +------+
  48. /// | None |
  49. /// +------+
  50. /// ^
  51. /// |
  52. /// +------+ +---------+
  53. /// +------> | move | ---> | suspend |
  54. /// | +------+ +---------+
  55. /// | ^ | +--------+
  56. /// | | | | insert |
  57. /// | v +--------+
  58. /// +---------+ | +--------+ |
  59. /// | connect | | | refine | |
  60. /// +---------+ | +--------+ |
  61. /// | v | v
  62. /// | +-----------+ | +------+
  63. /// +---> | connected | <-------+------> | None |
  64. /// +-----------+ +------+
  65. /// |
  66. /// v
  67. /// +------+
  68. /// | None |
  69. /// +------+
  70. ///
  71. /// ```
  72. /* NOTE: Currently if a user loses connectivity, they will be deleted from
  73. our hostlist by the refinery process and forgotten about until they regain
  74. connectivity and share their external address with the p2p network again.
  75. We may want to keep nodes with patchy connections in a `Red` list
  76. and periodically try to connect to them in Outbound Session, rather
  77. than sending them to the refinery (which will delete them if they are
  78. offline) as we do using `Suspend`. The current design favors reliability
  79. of connections but this may come at a risk for security since an attacker
  80. is likely to have good uptime. We want to insure that users with patchy
  81. connections or on mobile are still likely to be connected to.*/
  82. #[derive(Clone, Debug)]
  83. pub enum HostState {
  84. /// Hosts that are currently being inserting into the hostlist.
  85. Insert,
  86. /// Hosts that are migrating from the greylist to the whitelist or being
  87. /// removed from the greylist, as defined in `refinery.rs`.
  88. Refine,
  89. /// Hosts that are being connected to in Outbound and Manual Session.
  90. Connect,
  91. /// Hosts that we have just failed to connect to. Marking a host as
  92. /// Suspend effectively sends this host to refinery, since Suspend->
  93. /// Refine is an acceptable state transition. Being marked as Suspend does
  94. /// not increase a host's probability of being refined, since the refinery
  95. /// selects its subjects randomly (with the caveat that we cannot refine
  96. /// nodes marked as Connect, Connected, Insert or Move). It does however
  97. /// mean this host cannot be connected to unless it passes through the
  98. /// refinery successfully.
  99. Suspend,
  100. /// Hosts that have been successfully connected to.
  101. Connected(ChannelPtr),
  102. /// Host that are moving between hostlists, implemented in
  103. /// store::move_host(). Move takes a ChannelPtr so that Channels that
  104. /// are being promoted to the Gold list can be re-inserted into the
  105. /// Connected once the promotion is safely finalized.
  106. Move(Option<ChannelPtr>),
  107. }
  108. impl HostState {
  109. // Try to change state to Insert. Only possible if we are not yet
  110. // tracking this host in the HostRegistry.
  111. fn try_insert(&self) -> Result<Self> {
  112. let start = self.to_string();
  113. let end = HostState::Insert.to_string();
  114. match self {
  115. HostState::Insert => Err(Error::HostStateBlocked(start, end)),
  116. HostState::Refine => Err(Error::HostStateBlocked(start, end)),
  117. HostState::Connect => Err(Error::HostStateBlocked(start, end)),
  118. HostState::Suspend => Err(Error::HostStateBlocked(start, end)),
  119. HostState::Connected(_) => Err(Error::HostStateBlocked(start, end)),
  120. HostState::Move(_) => Err(Error::HostStateBlocked(start, end)),
  121. }
  122. }
  123. // Try to change state to Refine. Only possible if we are not yet
  124. // tracking this host in the HostRegistry or if the host is marked as
  125. // Suspend i.e. we have failed to connect to it.
  126. fn try_refine(&self) -> Result<Self> {
  127. let start = self.to_string();
  128. let end = HostState::Refine.to_string();
  129. match self {
  130. HostState::Insert => Err(Error::HostStateBlocked(start, end)),
  131. HostState::Refine => Err(Error::HostStateBlocked(start, end)),
  132. HostState::Connect => Err(Error::HostStateBlocked(start, end)),
  133. HostState::Suspend => Ok(HostState::Refine),
  134. HostState::Connected(_) => Err(Error::HostStateBlocked(start, end)),
  135. HostState::Move(_) => Err(Error::HostStateBlocked(start, end)),
  136. }
  137. }
  138. // Try to change state to Connect. Only possible if we are not yet
  139. // tracking this host in the HostRegistry.
  140. fn try_connect(&self) -> Result<Self> {
  141. let start = self.to_string();
  142. let end = HostState::Connect.to_string();
  143. match self {
  144. HostState::Insert => Err(Error::HostStateBlocked(start, end)),
  145. HostState::Refine => Err(Error::HostStateBlocked(start, end)),
  146. HostState::Connect => Err(Error::HostStateBlocked(start, end)),
  147. HostState::Suspend => Err(Error::HostStateBlocked(start, end)),
  148. HostState::Connected(_) => Err(Error::HostStateBlocked(start, end)),
  149. HostState::Move(_) => Err(Error::HostStateBlocked(start, end)),
  150. }
  151. }
  152. // Try to change state to Connected. Possible if this peer's state
  153. // is currently Connect, Refine or Move. Refine is necessary since the
  154. // refinery process requires us to establish a connection to a peer.
  155. // Move is necessary in the case that a host is being promoted to Gold list,
  156. // and must be re-added to the Connected() state after the promotion
  157. // has completed.
  158. fn try_connected(&self, channel: ChannelPtr) -> Result<Self> {
  159. let start = self.to_string();
  160. let end = HostState::Connected(channel.clone()).to_string();
  161. match self {
  162. HostState::Insert => Err(Error::HostStateBlocked(start, end)),
  163. HostState::Refine => Ok(HostState::Connected(channel)),
  164. HostState::Connect => Ok(HostState::Connected(channel)),
  165. HostState::Suspend => Err(Error::HostStateBlocked(start, end)),
  166. HostState::Connected(_) => Err(Error::HostStateBlocked(start, end)),
  167. HostState::Move(_) => Ok(HostState::Connected(channel)),
  168. }
  169. }
  170. // Try to change state to Move. Possibly if this host is currently
  171. // Connect i.e. it is being connected to, or if we are currently Connected
  172. // to this peer (necessary due to Gold list promotion sequence).
  173. fn try_move(&self, channel: Option<ChannelPtr>) -> Result<Self> {
  174. let start = self.to_string();
  175. let end = HostState::Move(channel.clone()).to_string();
  176. match self {
  177. HostState::Insert => Err(Error::HostStateBlocked(start, end)),
  178. HostState::Refine => Err(Error::HostStateBlocked(start, end)),
  179. HostState::Connect => Ok(HostState::Move(channel)),
  180. HostState::Suspend => Err(Error::HostStateBlocked(start, end)),
  181. HostState::Connected(_) => Ok(HostState::Move(channel)),
  182. HostState::Move(_) => Err(Error::HostStateBlocked(start, end)),
  183. }
  184. }
  185. // Try to change the state to Suspend. Only possible when we are
  186. // currently moving this host, since we suspend a host after failing
  187. // to connect to it and then downgrading in move_host.
  188. fn try_suspend(&self) -> Result<Self> {
  189. let start = self.to_string();
  190. let end = HostState::Suspend.to_string();
  191. match self {
  192. HostState::Insert => Err(Error::HostStateBlocked(start, end)),
  193. HostState::Refine => Err(Error::HostStateBlocked(start, end)),
  194. HostState::Connect => Err(Error::HostStateBlocked(start, end)),
  195. HostState::Suspend => Err(Error::HostStateBlocked(start, end)),
  196. HostState::Connected(_) => Err(Error::HostStateBlocked(start, end)),
  197. HostState::Move(_) => Ok(HostState::Suspend),
  198. }
  199. }
  200. }
  201. impl fmt::Display for HostState {
  202. fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
  203. fmt::Debug::fmt(self, f)
  204. }
  205. }
  206. #[repr(u8)]
  207. #[derive(Clone, Debug)]
  208. pub enum HostColor {
  209. /// Intermediary nodes that are periodically probed and updated
  210. /// to White.
  211. Grey = 0,
  212. /// Recently seen hosts. Shared with other nodes.
  213. White = 1,
  214. /// Nodes to which we have already been able to establish a
  215. /// connection.
  216. Gold = 2,
  217. /// Hostile peers that can neither be connected to nor establish
  218. /// connections to us for the duration of the program.
  219. Black = 3,
  220. }
  221. impl TryFrom<usize> for HostColor {
  222. type Error = Error;
  223. fn try_from(value: usize) -> Result<Self> {
  224. match value {
  225. 0 => Ok(HostColor::Grey),
  226. 1 => Ok(HostColor::White),
  227. 2 => Ok(HostColor::Gold),
  228. 3 => Ok(HostColor::Black),
  229. _ => Err(Error::InvalidHostColor),
  230. }
  231. }
  232. }
  233. /// A Container for managing Grey, White, Gold and Black hostlists. Exposes
  234. /// a common interface for writing to and querying hostlists.
  235. // TODO: Verify the performance overhead of using vectors for hostlists.
  236. // TODO: Check whether anchorlist (Gold) has a max size in Monero.
  237. pub struct HostContainer {
  238. pub hostlists: [RwLock<Vec<(Url, u64)>>; 4],
  239. }
  240. impl HostContainer {
  241. fn new() -> Self {
  242. let hostlists: [RwLock<Vec<(Url, u64)>>; 4] = [
  243. RwLock::new(Vec::new()),
  244. RwLock::new(Vec::new()),
  245. RwLock::new(Vec::new()),
  246. RwLock::new(Vec::new()),
  247. ];
  248. Self { hostlists }
  249. }
  250. /// Append host to a hostlist.
  251. async fn store(&self, color: usize, addr: Url, last_seen: u64) {
  252. trace!(target: "net::hosts::store()", "[START] list={:?}",
  253. HostColor::try_from(color).unwrap());
  254. let mut list = self.hostlists[color].write().await;
  255. list.push((addr, last_seen));
  256. if color == 0 && list.len() == GREYLIST_MAX_LEN {
  257. let last_entry = list.pop().unwrap();
  258. debug!(
  259. target: "net::hosts::store()",
  260. "Greylist reached max size. Removed {:?}", last_entry,
  261. );
  262. }
  263. if color == 1 && list.len() == WHITELIST_MAX_LEN {
  264. let last_entry = list.pop().unwrap();
  265. debug!(
  266. target: "net::hosts::store()",
  267. "Whitelist reached max size. Removed {:?}", last_entry,
  268. );
  269. }
  270. // Sort the list by last_seen.
  271. list.sort_by_key(|entry| entry.1);
  272. list.reverse();
  273. trace!(target: "net::hosts::store()", "[END] list={:?}",
  274. HostColor::try_from(color).unwrap());
  275. }
  276. /// Stores an address on a hostlist or updates its last_seen field if
  277. /// we already have the address.
  278. pub async fn store_or_update(&self, color: HostColor, addr: Url, last_seen: u64) {
  279. trace!(target: "net::hosts::store_or_update()", "[START] list={:?}", color);
  280. let color_int = color.clone() as usize;
  281. if !self.contains(color_int, &addr).await {
  282. debug!(target: "net::hosts::store_or_update()",
  283. "We do not have {} in {:?} list. Adding to store...", addr,
  284. color);
  285. self.store(color_int, addr, last_seen).await;
  286. } else {
  287. debug!(target: "net::hosts::store_or_update()",
  288. "We have {} in {:?} list. Updating last seen...", addr,
  289. color);
  290. self.update_last_seen(color_int, &addr, last_seen, None).await;
  291. }
  292. trace!(target: "net::hosts::store_or_update()", "[END] list={:?}", color);
  293. }
  294. /// Update the last_seen field of a peer on a hostlist.
  295. pub async fn update_last_seen(
  296. &self,
  297. color: usize,
  298. addr: &Url,
  299. last_seen: u64,
  300. position: Option<usize>,
  301. ) {
  302. trace!(target: "net::hosts::update_last_seen()", "[START] list={:?}",
  303. HostColor::try_from(color).unwrap());
  304. let i = match position {
  305. Some(i) => i,
  306. None => self.get_index_at_addr(color, addr.clone()).await.unwrap(),
  307. };
  308. let mut list = self.hostlists[color].write().await;
  309. list[i] = (addr.clone(), last_seen);
  310. list.sort_by_key(|entry| entry.1);
  311. list.reverse();
  312. trace!(target: "net::hosts::update_last_seen()", "[END] list={:?}",
  313. HostColor::try_from(color).unwrap());
  314. }
  315. /// Return all known hosts on a hostlist.
  316. pub async fn fetch_all(&self, color: HostColor) -> Vec<(Url, u64)> {
  317. self.hostlists[color as usize].read().await.iter().cloned().collect()
  318. }
  319. /// Get the oldest entry from a hostlist.
  320. pub async fn fetch_last(&self, color: HostColor) -> ((Url, u64), usize) {
  321. let list = self.hostlists[color as usize].read().await;
  322. let position = list.len() - 1;
  323. let entry = &list[position];
  324. (entry.clone(), position)
  325. }
  326. /// Fetch addresses that match the provided transports or acceptable
  327. /// mixed transports. Will return an empty Vector if no such addresses
  328. /// were found.
  329. pub async fn fetch_addrs(
  330. &self,
  331. color: HostColor,
  332. transports: &[String],
  333. transport_mixing: bool,
  334. ) -> Vec<(Url, u64)> {
  335. trace!(target: "net::hosts::fetch_addrs()", "[START] {:?}", color);
  336. let mut hosts = vec![];
  337. let index = color as usize;
  338. // If transport mixing is enabled, then for example we're allowed to
  339. // use tor:// to connect to tcp:// and tor+tls:// to connect to tcp+tls://.
  340. // However, **do not** mix tor:// and tcp+tls://, nor tor+tls:// and tcp://.
  341. macro_rules! mix_transport {
  342. ($a:expr, $b:expr) => {
  343. if transports.contains(&$a.to_string()) && transport_mixing {
  344. let mut a_to_b = self.fetch_with_schemes(index, &[$b.to_string()], None).await;
  345. for (addr, last_seen) in a_to_b.iter_mut() {
  346. addr.set_scheme($a).unwrap();
  347. hosts.push((addr.clone(), last_seen.clone()));
  348. }
  349. }
  350. };
  351. }
  352. mix_transport!("tor", "tcp");
  353. mix_transport!("tor+tls", "tcp+tls");
  354. mix_transport!("nym", "tcp");
  355. mix_transport!("nym+tls", "tcp+tls");
  356. // And now the actual requested transports
  357. for (addr, last_seen) in self.fetch_with_schemes(index, transports, None).await {
  358. hosts.push((addr, last_seen));
  359. }
  360. trace!(target: "net::hosts::fetch_addrs()", "Grabbed hosts, length: {}", hosts.len());
  361. hosts
  362. }
  363. /// Get up to limit peers that match the given transport schemes from
  364. /// a hostlist. If limit was not provided, return all matching peers.
  365. async fn fetch_with_schemes(
  366. &self,
  367. color: usize,
  368. schemes: &[String],
  369. limit: Option<usize>,
  370. ) -> Vec<(Url, u64)> {
  371. trace!(target: "net::hosts::fetch_with_schemes()", "[START] {:?}",
  372. HostColor::try_from(color).unwrap());
  373. let list = self.hostlists[color].read().await;
  374. let mut limit = match limit {
  375. Some(l) => l.min(list.len()),
  376. None => list.len(),
  377. };
  378. let mut ret = vec![];
  379. if limit == 0 {
  380. return ret
  381. }
  382. for (addr, last_seen) in list.iter() {
  383. if schemes.contains(&addr.scheme().to_string()) {
  384. ret.push((addr.clone(), *last_seen));
  385. limit -= 1;
  386. if limit == 0 {
  387. debug!(target: "net::hosts::fetch_with_schemes()",
  388. "Found matching {:?} scheme, returning {} addresses",
  389. HostColor::try_from(color).unwrap(), ret.len());
  390. return ret
  391. }
  392. }
  393. }
  394. if ret.is_empty() {
  395. debug!(target: "net::hosts::fetch_with_schemes()",
  396. "No such {:?} schemes found!", HostColor::try_from(color).unwrap())
  397. }
  398. ret
  399. }
  400. /// Get up to limit peers that don't match the given transport schemes
  401. /// from a hostlist. If limit was not provided, return all matching
  402. /// peers.
  403. async fn fetch_excluding_schemes(
  404. &self,
  405. color: usize,
  406. schemes: &[String],
  407. limit: Option<usize>,
  408. ) -> Vec<(Url, u64)> {
  409. trace!(target: "net::hosts::fetch_with_schemes()", "[START] {:?}",
  410. HostColor::try_from(color).unwrap());
  411. let list = self.hostlists[color].read().await;
  412. let mut limit = match limit {
  413. Some(l) => l.min(list.len()),
  414. None => list.len(),
  415. };
  416. let mut ret = vec![];
  417. if limit == 0 {
  418. return ret
  419. }
  420. for (addr, last_seen) in list.iter() {
  421. if !schemes.contains(&addr.scheme().to_string()) {
  422. ret.push((addr.clone(), *last_seen));
  423. limit -= 1;
  424. if limit == 0 {
  425. return ret
  426. }
  427. }
  428. }
  429. if ret.is_empty() {
  430. debug!(target: "net::hosts::fetch_excluding_schemes()",
  431. "No such schemes found!")
  432. }
  433. ret
  434. }
  435. /// Get a random peer from a hostlist.
  436. pub async fn fetch_random(&self, color: HostColor) -> ((Url, u64), usize) {
  437. let list = self.hostlists[color as usize].read().await;
  438. let position = rand::thread_rng().gen_range(0..list.len());
  439. let entry = &list[position];
  440. (entry.clone(), position)
  441. }
  442. /// Get a random peer from a hostlist that matches the given transport
  443. /// schemes.
  444. pub async fn fetch_random_with_schemes(
  445. &self,
  446. color: HostColor,
  447. schemes: &[String],
  448. ) -> Option<((Url, u64), usize)> {
  449. // Retrieve all peers corresponding to that transport schemes
  450. trace!(target: "net::hosts::fetch_random_with_schemes()", "[START] {:?}", color);
  451. let list = self.fetch_with_schemes(color as usize, schemes, None).await;
  452. if list.is_empty() {
  453. return None
  454. }
  455. let position = rand::thread_rng().gen_range(0..list.len());
  456. let entry = &list[position];
  457. Some((entry.clone(), position))
  458. }
  459. /// Get up to n random peers. Schemes are not taken into account.
  460. pub async fn fetch_n_random(&self, color: HostColor, n: u32) -> Vec<(Url, u64)> {
  461. trace!(target: "net::hosts::fetch_n_random()", "[START] {:?}", color);
  462. let n = n as usize;
  463. if n == 0 {
  464. return vec![]
  465. }
  466. let mut hosts = vec![];
  467. let list = self.hostlists[color as usize].read().await;
  468. for (addr, last_seen) in list.iter() {
  469. hosts.push((addr.clone(), *last_seen));
  470. }
  471. if hosts.is_empty() {
  472. debug!(target: "net::hosts::fetch_n_random()",
  473. "No entries found!");
  474. return hosts
  475. }
  476. // Grab random ones
  477. let urls = hosts.iter().choose_multiple(&mut OsRng, n.min(hosts.len()));
  478. urls.iter().map(|&url| url.clone()).collect()
  479. }
  480. /// Get up to n random peers that match the given transport schemes.
  481. pub async fn fetch_n_random_with_schemes(
  482. &self,
  483. color: HostColor,
  484. schemes: &[String],
  485. n: u32,
  486. ) -> Vec<(Url, u64)> {
  487. trace!(target: "net::hosts::fetch_n_random_with_schemes()", "[START] {:?}", color);
  488. let index = color as usize;
  489. let n = n as usize;
  490. if n == 0 {
  491. return vec![]
  492. }
  493. // Retrieve all peers corresponding to that transport schemes
  494. let hosts = self.fetch_with_schemes(index, schemes, None).await;
  495. if hosts.is_empty() {
  496. debug!(target: "net::hosts::fetch_n_random_with_schemes()",
  497. "No such schemes found!");
  498. return hosts
  499. }
  500. // Grab random ones
  501. let urls = hosts.iter().choose_multiple(&mut OsRng, n.min(hosts.len()));
  502. urls.iter().map(|&url| url.clone()).collect()
  503. }
  504. /// Get up to n random peers that don't match the given transport schemes
  505. /// from a hostlist.
  506. pub async fn fetch_n_random_excluding_schemes(
  507. &self,
  508. color: HostColor,
  509. schemes: &[String],
  510. n: u32,
  511. ) -> Vec<(Url, u64)> {
  512. trace!(target: "net::hosts::fetch_excluding_schemes()", "[START] {:?}", color);
  513. let index = color as usize;
  514. let n = n as usize;
  515. if n == 0 {
  516. return vec![]
  517. }
  518. // Retrieve all peers not corresponding to that transport schemes
  519. let hosts = self.fetch_excluding_schemes(index, schemes, None).await;
  520. if hosts.is_empty() {
  521. debug!(target: "net::hosts::fetch_n_random_excluding_schemes()",
  522. "No such schemes found!");
  523. return hosts
  524. }
  525. // Grab random ones
  526. let urls = hosts.iter().choose_multiple(&mut OsRng, n.min(hosts.len()));
  527. urls.iter().map(|&url| url.clone()).collect()
  528. }
  529. /// Remove an entry from a hostlist.
  530. pub async fn remove(&self, color: HostColor, addr: &Url, index: usize) {
  531. debug!(target: "net::hosts::remove()", "Removing peer {} from {:?}", addr, color);
  532. let mut list = self.hostlists[color as usize].write().await;
  533. list.remove(index);
  534. }
  535. /// Remove an entry from a hostlist if it exists.
  536. pub async fn remove_if_exists(&self, color: HostColor, addr: &Url) {
  537. trace!(target: "net::hosts::remove_if_exists()", "[START] addr={} list={:?}", addr, color);
  538. let index = color.clone() as usize;
  539. if self.contains(index, addr).await {
  540. let position =
  541. self.get_index_at_addr(index, addr.clone()).await.expect("Expected index to exist");
  542. self.remove(color.clone(), addr, position).await;
  543. }
  544. trace!(target: "net::hosts::remove_if_exists()", "[STOP] addr={} list={:?}", addr, color);
  545. }
  546. /// Check if a hostlist is empty.
  547. pub async fn is_empty(&self, color: HostColor) -> bool {
  548. self.hostlists[color as usize].read().await.is_empty()
  549. }
  550. /// Check if host is in a hostlist
  551. pub async fn contains(&self, color: usize, addr: &Url) -> bool {
  552. self.hostlists[color].read().await.iter().any(|(u, _t)| u == addr)
  553. }
  554. /// Get the index for a given addr on a hostlist.
  555. async fn get_index_at_addr(&self, color: usize, addr: Url) -> Option<usize> {
  556. self.hostlists[color].read().await.iter().position(|a| a.0 == addr)
  557. }
  558. /// Get the last_seen field for a given entry on a hostlist.
  559. pub async fn get_last_seen(&self, color: usize, addr: &Url) -> Option<u64> {
  560. self.hostlists[color]
  561. .read()
  562. .await
  563. .iter()
  564. .find(|(url, _)| url == addr)
  565. .map(|(_, last_seen)| *last_seen)
  566. }
  567. /// Load the hostlists from a file.
  568. pub async fn load_all(&self, path: &str) -> Result<()> {
  569. let path = expand_path(path)?;
  570. if !path.exists() {
  571. if let Some(parent) = path.parent() {
  572. fs::create_dir_all(parent)?;
  573. }
  574. File::create(path.clone())?;
  575. }
  576. let contents = load_file(&path);
  577. if let Err(e) = contents {
  578. warn!(target: "net::hosts::load_hosts()", "Failed retrieving saved hosts: {}", e);
  579. return Ok(())
  580. }
  581. for line in contents.unwrap().lines() {
  582. let data: Vec<&str> = line.split('\t').collect();
  583. let url = match Url::parse(data[1]) {
  584. Ok(u) => u,
  585. Err(e) => {
  586. debug!(target: "net::hosts::load_hosts()", "Skipping malformed URL {}", e);
  587. continue
  588. }
  589. };
  590. let last_seen = match data[2].parse::<u64>() {
  591. Ok(t) => t,
  592. Err(e) => {
  593. debug!(target: "net::hosts::load_hosts()", "Skipping malformed last seen {}", e);
  594. continue
  595. }
  596. };
  597. match data[0] {
  598. "greylist" => {
  599. self.store(HostColor::Grey as usize, url, last_seen).await;
  600. }
  601. "whitelist" => {
  602. self.store(HostColor::White as usize, url, last_seen).await;
  603. }
  604. "anchorlist" => {
  605. self.store(HostColor::Gold as usize, url, last_seen).await;
  606. }
  607. _ => {
  608. debug!(target: "net::hosts::load_hosts()", "Malformed list name...");
  609. }
  610. }
  611. }
  612. Ok(())
  613. }
  614. /// Save the hostlist to a file. Whitelist gets written to the greylist
  615. /// to force whitelist entries through the refinery on start.
  616. pub async fn save_all(&self, path: &str) -> Result<()> {
  617. let path = expand_path(path)?;
  618. let mut tsv = String::new();
  619. let mut white = vec![];
  620. let mut greygold: HashMap<String, Vec<(Url, u64)>> = HashMap::new();
  621. // First gather all the whitelist entries we don't have in greylist.
  622. for (url, last_seen) in self.fetch_all(HostColor::White).await {
  623. if !self.contains(HostColor::Grey as usize, &url).await {
  624. white.push((url, last_seen))
  625. }
  626. }
  627. // Then gather the greylist and anchorlist entries.
  628. greygold.insert("anchorlist".to_string(), self.fetch_all(HostColor::Gold).await);
  629. greygold.insert("greylist".to_string(), self.fetch_all(HostColor::Grey).await);
  630. // We write whitelist entries to the greylist on p2p.stop() to force
  631. // them through the refinery on start().
  632. for (name, mut list) in greygold {
  633. if name == *"greylist".to_string() {
  634. list.append(&mut white)
  635. }
  636. for (url, last_seen) in list {
  637. tsv.push_str(&format!("{}\t{}\t{}\n", name, url, last_seen));
  638. }
  639. }
  640. if !tsv.eq("") {
  641. info!(target: "net::hosts::save_hosts()", "Saving hosts to: {:?}",
  642. path);
  643. if let Err(e) = save_file(&path, &tsv) {
  644. error!(target: "net::hosts::save_hosts()", "Failed saving hosts: {}", e);
  645. }
  646. }
  647. Ok(())
  648. }
  649. }
  650. /// Main parent class for the management and manipulation of
  651. /// hostlists. Keeps track of hosts and their current state via the
  652. /// HostRegistry, and stores hostlists and associated methods in the
  653. /// HostContainer. Also operates two subscribers to notify other parts
  654. /// of the code base when new channels have been created or new hosts
  655. /// have been added to the hostlist.
  656. pub struct Hosts {
  657. /// A registry that tracks hosts and their current state.
  658. registry: HostRegistry,
  659. /// Hostlists and associated methods.
  660. pub container: HostContainer,
  661. /// Subscriber listening for store updates
  662. store_subscriber: SubscriberPtr<usize>,
  663. /// Subscriber for notifications of new channels
  664. pub channel_subscriber: SubscriberPtr<Result<ChannelPtr>>,
  665. /// Keeps track of the last time a connection was made.
  666. pub last_connection: RwLock<Instant>,
  667. /// Pointer to configured P2P settings
  668. settings: SettingsPtr,
  669. }
  670. impl Hosts {
  671. /// Create a new hosts list
  672. pub fn new(settings: SettingsPtr) -> HostsPtr {
  673. Arc::new(Self {
  674. registry: RwLock::new(HashMap::new()),
  675. container: HostContainer::new(),
  676. store_subscriber: Subscriber::new(),
  677. channel_subscriber: Subscriber::new(),
  678. last_connection: RwLock::new(Instant::now()),
  679. settings,
  680. })
  681. }
  682. /// Safely insert into the HostContainer. Filters the addresses first before storing and
  683. /// notifies the subscriber. Must be called when first receiving greylist addresses.
  684. pub async fn insert(&self, color: HostColor, addrs: &[(Url, u64)]) {
  685. trace!(target: "net::hosts:insert()", "[START]");
  686. // First filter these address to ensure this peer doesn't exist in our black, gold or
  687. // whitelist and apply transport filtering.
  688. let filtered_addrs = self.filter_addresses(self.settings.clone(), addrs).await;
  689. let mut addrs_len = 0;
  690. if filtered_addrs.is_empty() {
  691. debug!(target: "net::hosts::insert()", "Filtered out all addresses");
  692. }
  693. // Then ensure we aren't currently trying to add this peer to the hostlist.
  694. for (i, (addr, last_seen)) in filtered_addrs.iter().enumerate() {
  695. if let Err(e) = self.try_register(addr.clone(), HostState::Insert).await {
  696. debug!(target: "net::hosts::store_or_update", "Cannot insert addr={}, err={}",
  697. addr.clone(), e);
  698. continue
  699. }
  700. addrs_len += i + 1;
  701. self.container.store_or_update(color.clone(), addr.clone(), *last_seen).await;
  702. self.unregister(addr).await;
  703. }
  704. self.store_subscriber.notify(addrs_len).await;
  705. trace!(target: "net::hosts:insert()", "[END]");
  706. }
  707. /// Try to update the registry. If the host already exists, try to update its state.
  708. /// Otherwise add the host to the registry along with its state.
  709. pub async fn try_register(&self, addr: Url, new_state: HostState) -> Result<HostState> {
  710. let mut registry = self.registry.write().await;
  711. if registry.contains_key(&addr) {
  712. let current_state = registry.get(&addr).unwrap().clone();
  713. let result: Result<HostState> = match new_state {
  714. HostState::Insert => current_state.try_insert(),
  715. HostState::Refine => current_state.try_refine(),
  716. HostState::Connect => current_state.try_connect(),
  717. HostState::Suspend => current_state.try_suspend(),
  718. HostState::Connected(c) => current_state.try_connected(c),
  719. HostState::Move(c) => current_state.try_move(c),
  720. };
  721. if let Ok(state) = &result {
  722. registry.insert(addr.clone(), state.clone());
  723. }
  724. result
  725. } else {
  726. // We don't know this peer. We can safely update the state.
  727. debug!(target: "net::hosts::try_update_registry()", "Inserting addr={}, state={}",
  728. addr, new_state.to_string());
  729. registry.insert(addr.clone(), new_state.clone());
  730. Ok(new_state)
  731. }
  732. }
  733. // Loop through hosts selected by Outbound Session and see if any of them are
  734. // free to connect to.
  735. pub async fn check_addrs(&self, hosts: Vec<(Url, u64)>) -> Option<(Url, u64)> {
  736. for (host, last_seen) in hosts {
  737. debug!(target: "net::hosts::check_addrs()", "Starting checks");
  738. // Print a warning if we are trying to connect to a seed node in
  739. // Outbound session. This shouldn't happen as we reject configured
  740. // seed nodes from entering our hostlist in filter_addrs().
  741. if self.settings.seeds.contains(&host) {
  742. warn!(target: "net::hosts::check_addrs",
  743. "Seed addr={} has entered the hostlist! Skipping",
  744. host.clone());
  745. continue
  746. }
  747. if let Err(e) = self.try_register(host.clone(), HostState::Connect).await {
  748. debug!(target: "net::hosts::check_addrs", "Skipping addr={}, err={}",
  749. host.clone(), e);
  750. continue
  751. }
  752. debug!(
  753. target: "net::hosts::check_addrs()",
  754. "Found valid host {}",
  755. host
  756. );
  757. return Some((host.clone(), last_seen))
  758. }
  759. None
  760. }
  761. /// Remove a host from the HostRegistry. Must be called after move(), when the refinery
  762. /// process fails, or when a channel stops. Prevents hosts from getting trapped in the
  763. /// HostState logical machinery.
  764. pub async fn unregister(&self, addr: &Url) {
  765. debug!(target: "net::hosts::unregister()", "Removing {} from HostRegistry", addr);
  766. self.registry.write().await.remove(addr);
  767. }
  768. /// Returns the list of connected channels.
  769. pub async fn channels(&self) -> Vec<ChannelPtr> {
  770. let registry = self.registry.read().await;
  771. let mut channels = Vec::new();
  772. for (_, state) in registry.iter() {
  773. if let HostState::Connected(c) = state {
  774. channels.push(c.clone());
  775. }
  776. }
  777. channels
  778. }
  779. /// Returns the list of connected channels.
  780. pub async fn suspended(&self) -> Vec<Url> {
  781. let registry = self.registry.read().await;
  782. let mut addrs = Vec::new();
  783. for (url, state) in registry.iter() {
  784. if let HostState::Suspend = state {
  785. addrs.push(url.clone());
  786. }
  787. }
  788. addrs
  789. }
  790. /// Retrieve a random connected channel
  791. pub async fn random_channel(&self) -> ChannelPtr {
  792. let channels = self.channels().await;
  793. let position = rand::thread_rng().gen_range(0..channels.len());
  794. channels[position].clone()
  795. }
  796. /// Add a channel to the set of connected channels
  797. pub async fn register_channel(&self, channel: ChannelPtr) -> Result<()> {
  798. let address = channel.address().clone();
  799. self.try_register(address.clone(), HostState::Connected(channel.clone())).await.unwrap();
  800. // Notify that channel processing was successful
  801. self.channel_subscriber.notify(Ok(channel.clone())).await;
  802. let mut last_online = self.last_connection.write().await;
  803. *last_online = Instant::now();
  804. Ok(())
  805. }
  806. pub async fn subscribe_store(&self) -> Subscription<usize> {
  807. self.store_subscriber.clone().subscribe().await
  808. }
  809. pub async fn subscribe_channel(&self) -> Subscription<Result<ChannelPtr>> {
  810. self.channel_subscriber.clone().subscribe().await
  811. }
  812. // Verify whether a URL is local.
  813. // NOTE: This function is stateless and not specific to
  814. // `Hosts`. For this reason, it might make more sense
  815. // to move this function to a more appropriate location
  816. // in the codebase.
  817. /// Check whether a URL is local host
  818. pub async fn is_local_host(&self, url: Url) -> bool {
  819. // Reject Urls without host strings.
  820. if url.host_str().is_none() {
  821. return false
  822. }
  823. // We do this hack in order to parse IPs properly.
  824. // https://github.com/whatwg/url/issues/749
  825. let addr = Url::parse(&url.as_str().replace(url.scheme(), "http")).unwrap();
  826. // Filter private IP ranges
  827. match addr.host().unwrap() {
  828. url::Host::Ipv4(ip) => {
  829. if !ip.is_global() {
  830. return true
  831. }
  832. }
  833. url::Host::Ipv6(ip) => {
  834. if !ip.is_global() {
  835. return true
  836. }
  837. }
  838. url::Host::Domain(d) => {
  839. if LOCAL_HOST_STRS.contains(&d) {
  840. return true
  841. }
  842. }
  843. }
  844. false
  845. }
  846. /// Filter given addresses based on certain rulesets and validity. Strictly called only on
  847. /// the first time learning of a new peer.
  848. async fn filter_addresses(
  849. &self,
  850. settings: SettingsPtr,
  851. addrs: &[(Url, u64)],
  852. ) -> Vec<(Url, u64)> {
  853. trace!(target: "net::hosts::filter_addresses()", "Filtering addrs: {:?}", addrs);
  854. let mut ret = vec![];
  855. let localnet = self.settings.localnet;
  856. 'addr_loop: for (addr_, last_seen) in addrs {
  857. // Validate that the format is `scheme://host_str:port`
  858. if addr_.host_str().is_none() ||
  859. addr_.port().is_none() ||
  860. addr_.cannot_be_a_base() ||
  861. addr_.path_segments().is_some()
  862. {
  863. continue
  864. }
  865. // Configured seeds should never enter the hostlist.
  866. if self.settings.seeds.contains(addr_) {
  867. debug!(target: "net::hosts::filter_addresses()",
  868. "[{}] is a configured seed. Skipping", addr_);
  869. continue
  870. }
  871. // Blacklist peers should never enter the hostlist.
  872. if self.container.contains(HostColor::Black as usize, addr_).await {
  873. warn!(target: "net::hosts::filter_addresses()",
  874. "Peer {} is blacklisted", addr_);
  875. continue
  876. }
  877. // Reject this peer if it's already stored on the Gold or White list.
  878. // If it exists on the Grey list, we will simply update its last_seen
  879. // field.
  880. if self.container.contains(HostColor::Gold as usize, addr_).await ||
  881. self.container.contains(HostColor::White as usize, addr_).await
  882. {
  883. debug!(target: "net::hosts::filter_addresses()",
  884. "We already have {} in the hostlist. Skipping", addr_);
  885. continue
  886. }
  887. let host_str = addr_.host_str().unwrap();
  888. if !localnet {
  889. // Our own external addresses should never enter the hosts set.
  890. for ext in &settings.external_addrs {
  891. if host_str == ext.host_str().unwrap() {
  892. continue 'addr_loop
  893. }
  894. }
  895. }
  896. // On localnet, make sure ours ports don't enter the host set.
  897. for ext in &settings.external_addrs {
  898. if addr_.port() == ext.port() {
  899. continue 'addr_loop
  900. }
  901. }
  902. // We do this hack in order to parse IPs properly.
  903. // https://github.com/whatwg/url/issues/749
  904. let addr = Url::parse(&addr_.as_str().replace(addr_.scheme(), "http")).unwrap();
  905. // Filter non-global ranges if we're not allowing localnet.
  906. // Should never be allowed in production, so we don't really care
  907. // about some of them (e.g. 0.0.0.0, or broadcast, etc.).
  908. if !localnet && self.is_local_host(addr).await {
  909. continue
  910. }
  911. match addr_.scheme() {
  912. // Validate that the address is an actual onion.
  913. #[cfg(feature = "p2p-tor")]
  914. "tor" | "tor+tls" => {
  915. use std::str::FromStr;
  916. if tor_hscrypto::pk::HsId::from_str(host_str).is_err() {
  917. continue
  918. }
  919. trace!(target: "net::hosts::filter_addresses()",
  920. "[Tor] Valid: {}", host_str);
  921. }
  922. #[cfg(feature = "p2p-nym")]
  923. "nym" | "nym+tls" => continue, // <-- Temp skip
  924. #[cfg(feature = "p2p-tcp")]
  925. "tcp" | "tcp+tls" => {
  926. trace!(target: "net::hosts::filter_addresses()",
  927. "[TCP] Valid: {}", host_str);
  928. }
  929. _ => continue,
  930. }
  931. ret.push((addr_.clone(), *last_seen));
  932. }
  933. ret
  934. }
  935. /// A single atomic function for moving hosts between hostlists. Called on the following occasions:
  936. ///
  937. /// * When we cannot connect to a peer: move to grey, remove from white and gold.
  938. /// * When a peer disconnects from us: move to grey, remove from white and gold.
  939. /// * When the refinery passes successfully: move to white, remove from greylist.
  940. /// * When we connect to a peer, move to gold, remove from white or grey.
  941. /// * When we add a peer to the black list: move to black, remove from all other lists.
  942. pub async fn move_host(
  943. &self,
  944. addr: &Url,
  945. last_seen: u64,
  946. destination: HostColor,
  947. suspend: bool,
  948. channel: Option<ChannelPtr>,
  949. ) {
  950. debug!(target: "net::hosts::move_host()", "Trying to move addr={} node={} destination={:?}",
  951. addr, self.settings.node_id, destination);
  952. // This should never panic. Failure indicates a misuse of the HostState API.
  953. self.try_register(addr.clone(), HostState::Move(channel.clone())).await.unwrap();
  954. match destination {
  955. // Downgrade to grey. Remove from white and gold.
  956. HostColor::Grey => {
  957. // Remove from the gold list if it exists.
  958. let mut gold = self.container.hostlists[HostColor::Gold as usize].write().await;
  959. if gold.iter().any(|(u, _t)| u == addr) {
  960. let position = gold.iter().position(|a| a.0 == *addr).unwrap();
  961. gold.remove(position);
  962. }
  963. drop(gold);
  964. // Remove from the white list if it exists.
  965. let mut white = self.container.hostlists[HostColor::White as usize].write().await;
  966. if white.iter().any(|(u, _t)| u == addr) {
  967. let position = white.iter().position(|a| a.0 == *addr).unwrap();
  968. white.remove(position);
  969. }
  970. drop(white);
  971. // If it exists on the grey list, update its last seen field.
  972. // Otherwise, write to the greylist.
  973. let mut grey = self.container.hostlists[HostColor::Grey as usize].write().await;
  974. if grey.iter().any(|(u, _t)| u == addr) {
  975. let position = grey.iter().position(|a| a.0 == *addr).unwrap();
  976. grey[position] = (addr.clone(), last_seen);
  977. } else {
  978. // We don't have this entry.
  979. if grey.len() == GREYLIST_MAX_LEN {
  980. let last_entry = grey.pop().unwrap();
  981. debug!(
  982. target: "net::hosts::move_host()",
  983. "Greylist reached max size. Removed {:?}", last_entry,
  984. );
  985. }
  986. grey.push((addr.clone(), last_seen));
  987. // Sort the list by last_seen.
  988. grey.sort_by_key(|entry| entry.1);
  989. grey.reverse();
  990. }
  991. drop(grey);
  992. if suspend {
  993. // We mark this peer as Suspend which means we do not try to connect to it until it
  994. // has passed through the refinery. This should never panic.
  995. self.try_register(addr.clone(), HostState::Suspend).await.unwrap();
  996. return
  997. }
  998. }
  999. // Remove from Greylist, add to Whitelist. Called by the Refinery.
  1000. HostColor::White => {
  1001. // Remove from the grey list if it exists.
  1002. let mut grey = self.container.hostlists[HostColor::Grey as usize].write().await;
  1003. if grey.iter().any(|(u, _t)| u == addr) {
  1004. let position = grey.iter().position(|a| a.0 == *addr).unwrap();
  1005. grey.remove(position);
  1006. }
  1007. drop(grey);
  1008. // If it exists on the white list, update its last seen field.
  1009. // Otherwise, write to the white list.
  1010. let mut white = self.container.hostlists[HostColor::White as usize].write().await;
  1011. if white.iter().any(|(u, _t)| u == addr) {
  1012. let position = white.iter().position(|a| a.0 == *addr).unwrap();
  1013. white[position] = (addr.clone(), last_seen);
  1014. } else {
  1015. // We don't have this entry.
  1016. if white.len() == WHITELIST_MAX_LEN {
  1017. let last_entry = white.pop().unwrap();
  1018. debug!(
  1019. target: "net::hosts::move_host()",
  1020. "Whitelist reached max size. Removed {:?}", last_entry,
  1021. );
  1022. }
  1023. white.push((addr.clone(), last_seen));
  1024. // Sort the list by last_seen.
  1025. white.sort_by_key(|entry| entry.1);
  1026. white.reverse();
  1027. }
  1028. drop(white);
  1029. }
  1030. // Upgrade to gold. Remove from white or grey.
  1031. HostColor::Gold => {
  1032. // Remove from the grey list if it exists.
  1033. let mut grey = self.container.hostlists[HostColor::Grey as usize].write().await;
  1034. if grey.iter().any(|(u, _t)| u == addr) {
  1035. let position = grey.iter().position(|a| a.0 == *addr).unwrap();
  1036. grey.remove(position);
  1037. }
  1038. drop(grey);
  1039. // Remove from the white list if it exists.
  1040. let mut white = self.container.hostlists[HostColor::White as usize].write().await;
  1041. if white.iter().any(|(u, _t)| u == addr) {
  1042. let position = white.iter().position(|a| a.0 == *addr).unwrap();
  1043. white.remove(position);
  1044. }
  1045. drop(white);
  1046. // If it exists on the gold list, update its last seen field.
  1047. // Otherwise, write to the gold list.
  1048. let mut gold = self.container.hostlists[HostColor::Gold as usize].write().await;
  1049. if gold.iter().any(|(u, _t)| u == addr) {
  1050. let position = gold.iter().position(|a| a.0 == *addr).unwrap();
  1051. gold[position] = (addr.clone(), last_seen);
  1052. } else {
  1053. gold.push((addr.clone(), last_seen));
  1054. // Sort the list by last_seen.
  1055. gold.sort_by_key(|entry| entry.1);
  1056. gold.reverse();
  1057. }
  1058. drop(gold);
  1059. // Re-register this host as Connected. We want to keep track of all connected peers
  1060. // in the Connected() state.
  1061. self.try_register(addr.clone(), HostState::Connected(channel.unwrap()))
  1062. .await
  1063. .unwrap();
  1064. return
  1065. }
  1066. // Move to black. Remove from all other lists.
  1067. HostColor::Black => {
  1068. // We ignore UNIX sockets here so we will just work
  1069. // with stuff that has host_str().
  1070. if addr.host_str().is_some() {
  1071. // Localhost connections should never enter the blacklist
  1072. // This however allows any Tor and Nym connections.
  1073. if self.is_local_host(addr.clone()).await {
  1074. return
  1075. }
  1076. // Remove from the grey list if it exists.
  1077. let mut grey = self.container.hostlists[HostColor::Grey as usize].write().await;
  1078. if grey.iter().any(|(u, _t)| u == addr) {
  1079. let position = grey.iter().position(|a| a.0 == *addr).unwrap();
  1080. grey.remove(position);
  1081. }
  1082. drop(grey);
  1083. // Remove from the white list if it exists.
  1084. let mut white =
  1085. self.container.hostlists[HostColor::White as usize].write().await;
  1086. if white.iter().any(|(u, _t)| u == addr) {
  1087. let position = white.iter().position(|a| a.0 == *addr).unwrap();
  1088. white.remove(position);
  1089. }
  1090. drop(white);
  1091. // Remove from the gold list if it exists.
  1092. let mut gold =
  1093. self.container.hostlists[HostColor::White as usize].write().await;
  1094. if gold.iter().any(|(u, _t)| u == addr) {
  1095. let position = gold.iter().position(|a| a.0 == *addr).unwrap();
  1096. gold.remove(position);
  1097. }
  1098. drop(gold);
  1099. // Add to the black list.
  1100. let mut black =
  1101. self.container.hostlists[HostColor::Gold as usize].write().await;
  1102. if black.iter().any(|(u, _t)| u == addr) {
  1103. let position = black.iter().position(|a| a.0 == *addr).unwrap();
  1104. black[position] = (addr.clone(), last_seen);
  1105. } else {
  1106. black.push((addr.clone(), last_seen));
  1107. }
  1108. drop(black);
  1109. }
  1110. }
  1111. }
  1112. // Remove this entry from HostRegistry to avoid this host getting
  1113. // stuck in the Moving state.
  1114. self.unregister(addr).await;
  1115. }
  1116. }
  1117. #[cfg(test)]
  1118. mod tests {
  1119. use smol::Executor;
  1120. use std::time::UNIX_EPOCH;
  1121. use super::{
  1122. super::super::{settings::Settings, P2p},
  1123. *,
  1124. };
  1125. use crate::{net::hosts::refinery::ping_node, system::sleep};
  1126. #[test]
  1127. fn test_ping_node() {
  1128. smol::block_on(async {
  1129. let settings = Settings {
  1130. localnet: false,
  1131. external_addrs: vec![
  1132. Url::parse("tcp://foo.bar:123").unwrap(),
  1133. Url::parse("tcp://lol.cat:321").unwrap(),
  1134. ],
  1135. ..Default::default()
  1136. };
  1137. let ex = Arc::new(Executor::new());
  1138. let p2p = P2p::new(settings, ex.clone()).await;
  1139. let url = Url::parse("tcp://xeno.systems.wtf").unwrap();
  1140. println!("Pinging node...");
  1141. let task = ex.spawn(ping_node(url.clone(), p2p));
  1142. ex.run(task).await;
  1143. println!("Ping node complete!");
  1144. });
  1145. }
  1146. #[test]
  1147. fn test_is_local_host() {
  1148. smol::block_on(async {
  1149. let settings = Settings {
  1150. localnet: false,
  1151. external_addrs: vec![
  1152. Url::parse("tcp://foo.bar:123").unwrap(),
  1153. Url::parse("tcp://lol.cat:321").unwrap(),
  1154. ],
  1155. ..Default::default()
  1156. };
  1157. let hosts = Hosts::new(Arc::new(settings.clone()));
  1158. let local_hosts: Vec<Url> = vec![
  1159. Url::parse("tcp://localhost").unwrap(),
  1160. Url::parse("tcp://127.0.0.1").unwrap(),
  1161. Url::parse("tcp+tls://[::1]").unwrap(),
  1162. Url::parse("tcp://localhost.localdomain").unwrap(),
  1163. Url::parse("tcp://192.168.10.65").unwrap(),
  1164. ];
  1165. for host in local_hosts {
  1166. eprintln!("{}", host);
  1167. assert!(hosts.is_local_host(host).await);
  1168. }
  1169. let remote_hosts: Vec<Url> = vec![
  1170. Url::parse("https://dyne.org").unwrap(),
  1171. Url::parse("tcp://77.168.10.65:2222").unwrap(),
  1172. Url::parse("tcp://[2345:0425:2CA1:0000:0000:0567:5673:23b5]").unwrap(),
  1173. Url::parse("http://eweiibe6tdjsdprb4px6rqrzzcsi22m4koia44kc5pcjr7nec2rlxyad.onion")
  1174. .unwrap(),
  1175. ];
  1176. for host in remote_hosts {
  1177. assert!(!hosts.is_local_host(host).await)
  1178. }
  1179. });
  1180. }
  1181. #[test]
  1182. fn test_store() {
  1183. let last_seen = UNIX_EPOCH.elapsed().unwrap().as_secs();
  1184. smol::block_on(async {
  1185. let settings = Settings { ..Default::default() };
  1186. let hosts = Hosts::new(Arc::new(settings.clone()));
  1187. let grey_hosts = vec![
  1188. Url::parse("tcp://localhost:3921").unwrap(),
  1189. Url::parse("tor://[::1]:21481").unwrap(),
  1190. Url::parse("tcp://192.168.10.65:311").unwrap(),
  1191. Url::parse("tcp+tls://0.0.0.0:2312").unwrap(),
  1192. Url::parse("tcp://255.255.255.255:2131").unwrap(),
  1193. ];
  1194. for addr in &grey_hosts {
  1195. hosts.container.store(HostColor::Grey as usize, addr.clone(), last_seen).await;
  1196. }
  1197. assert!(!hosts.container.is_empty(HostColor::Grey).await);
  1198. let white_hosts = vec![
  1199. Url::parse("tcp://localhost:3921").unwrap(),
  1200. Url::parse("tor://[::1]:21481").unwrap(),
  1201. Url::parse("tcp://192.168.10.65:311").unwrap(),
  1202. Url::parse("tcp+tls://0.0.0.0:2312").unwrap(),
  1203. Url::parse("tcp://255.255.255.255:2131").unwrap(),
  1204. ];
  1205. for host in &white_hosts {
  1206. hosts.container.store(HostColor::White as usize, host.clone(), last_seen).await;
  1207. }
  1208. assert!(!hosts.container.is_empty(HostColor::White).await);
  1209. let gold_hosts = vec![
  1210. Url::parse("tcp://dark.fi:80").unwrap(),
  1211. Url::parse("tcp://http.cat:401").unwrap(),
  1212. Url::parse("tcp://foo.bar:111").unwrap(),
  1213. ];
  1214. for host in &gold_hosts {
  1215. hosts.container.store(HostColor::Gold as usize, host.clone(), last_seen).await;
  1216. }
  1217. assert!(hosts.container.contains(HostColor::Grey as usize, &grey_hosts[0]).await);
  1218. assert!(hosts.container.contains(HostColor::White as usize, &white_hosts[1]).await);
  1219. assert!(hosts.container.contains(HostColor::Gold as usize, &gold_hosts[2]).await);
  1220. });
  1221. }
  1222. #[test]
  1223. fn test_get_last() {
  1224. smol::block_on(async {
  1225. let settings = Settings { ..Default::default() };
  1226. let hosts = Hosts::new(Arc::new(settings.clone()));
  1227. // Build up a hostlist
  1228. for i in 0..10 {
  1229. sleep(1).await;
  1230. let last_seen = UNIX_EPOCH.elapsed().unwrap().as_secs();
  1231. let url = Url::parse(&format!("tcp://whitelist{}:123", i)).unwrap();
  1232. hosts.container.store(HostColor::White as usize, url.clone(), last_seen).await;
  1233. }
  1234. for (url, last_seen) in
  1235. hosts.container.hostlists[HostColor::White as usize].read().await.iter()
  1236. {
  1237. println!("{} {}", url, last_seen);
  1238. }
  1239. let (entry, _position) = hosts.container.fetch_last(HostColor::White).await;
  1240. println!("last entry: {} {}", entry.0, entry.1);
  1241. });
  1242. }
  1243. #[test]
  1244. fn test_remove() {
  1245. smol::block_on(async {
  1246. let settings = Settings { ..Default::default() };
  1247. let hosts = Hosts::new(Arc::new(settings.clone()));
  1248. let url = Url::parse("tcp://dark.renaissance:333").unwrap();
  1249. let last_seen = UNIX_EPOCH.elapsed().unwrap().as_secs();
  1250. hosts.container.store(HostColor::White as usize, url.clone(), last_seen).await;
  1251. sleep(1).await;
  1252. let url = Url::parse("tcp://milady:333").unwrap();
  1253. let last_seen = UNIX_EPOCH.elapsed().unwrap().as_secs();
  1254. hosts.container.store(HostColor::White as usize, url.clone(), last_seen).await;
  1255. sleep(1).await;
  1256. let url = Url::parse("tcp://king-ted:333").unwrap();
  1257. let last_seen = UNIX_EPOCH.elapsed().unwrap().as_secs();
  1258. hosts.container.store(HostColor::White as usize, url.clone(), last_seen).await;
  1259. for (url, last_seen) in
  1260. hosts.container.hostlists[HostColor::White as usize].read().await.iter()
  1261. {
  1262. println!("{}, {}", url, last_seen);
  1263. }
  1264. let position = hosts
  1265. .container
  1266. .get_index_at_addr(HostColor::White as usize, url.clone())
  1267. .await
  1268. .unwrap();
  1269. hosts.container.remove(HostColor::White, &url, position).await;
  1270. for (url, last_seen) in
  1271. hosts.container.hostlists[HostColor::White as usize].read().await.iter()
  1272. {
  1273. println!("{}, {}", url, last_seen);
  1274. }
  1275. });
  1276. }
  1277. }