store.rs 59 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127112811291130113111321133113411351136113711381139114011411142114311441145114611471148114911501151115211531154115511561157115811591160116111621163116411651166116711681169117011711172117311741175117611771178117911801181118211831184118511861187118811891190119111921193119411951196119711981199120012011202120312041205120612071208120912101211121212131214121512161217121812191220122112221223122412251226122712281229123012311232123312341235123612371238123912401241124212431244124512461247124812491250125112521253125412551256125712581259126012611262126312641265126612671268126912701271127212731274127512761277127812791280128112821283128412851286128712881289129012911292129312941295129612971298129913001301130213031304130513061307130813091310131113121313131413151316131713181319132013211322132313241325132613271328132913301331133213331334133513361337133813391340134113421343134413451346134713481349135013511352135313541355135613571358135913601361136213631364136513661367136813691370137113721373137413751376137713781379138013811382138313841385138613871388138913901391139213931394139513961397139813991400140114021403140414051406140714081409141014111412141314141415141614171418141914201421142214231424142514261427142814291430143114321433143414351436143714381439144014411442144314441445144614471448144914501451145214531454145514561457145814591460146114621463146414651466146714681469147014711472147314741475147614771478147914801481148214831484148514861487148814891490149114921493149414951496149714981499150015011502150315041505150615071508150915101511151215131514151515161517151815191520152115221523152415251526152715281529153015311532153315341535153615371538153915401541154215431544154515461547154815491550155115521553155415551556155715581559156015611562156315641565156615671568156915701571157215731574157515761577157815791580158115821583158415851586158715881589159015911592159315941595159615971598159916001601160216031604160516061607160816091610161116121613161416151616161716181619162016211622162316241625162616271628162916301631163216331634163516361637163816391640
  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::{
  19. collections::{HashMap, HashSet},
  20. fs,
  21. fs::File,
  22. sync::Arc,
  23. time::UNIX_EPOCH,
  24. };
  25. use log::{debug, error, info, trace, warn};
  26. use rand::{prelude::IteratorRandom, rngs::OsRng, Rng};
  27. use smol::lock::RwLock;
  28. use url::Url;
  29. use super::super::{p2p::P2pPtr, settings::SettingsPtr};
  30. use crate::{
  31. system::{Subscriber, SubscriberPtr, Subscription},
  32. util::{
  33. file::{load_file, save_file},
  34. path::expand_path,
  35. },
  36. Result,
  37. };
  38. /// Atomic pointer to hosts object
  39. pub type HostsPtr = Arc<Hosts>;
  40. // An array containing all possible local host strings
  41. // TODO: This could perhaps be more exhaustive?
  42. pub const LOCAL_HOST_STRS: [&str; 2] = ["localhost", "localhost.localdomain"];
  43. const WHITELIST_MAX_LEN: usize = 5000;
  44. const GREYLIST_MAX_LEN: usize = 2000;
  45. /// Manages a store of network addresses
  46. // TODO: Test the performance overhead of using vectors for white/grey/anchor lists.
  47. // TODO: Check whether anchorlist has a max size in Monero.
  48. // TODO: we can probably clean up a lot of the repetitive code in this module.
  49. pub struct Hosts {
  50. /// Intermediary node list that is periodically probed and updated to whitelist.
  51. pub greylist: RwLock<Vec<(Url, u64)>>,
  52. /// Recently seen hosts. Shared with other nodes.
  53. pub whitelist: RwLock<Vec<(Url, u64)>>,
  54. /// Nodes to which we have already been able to establish a connection.
  55. pub anchorlist: RwLock<Vec<(Url, u64)>>,
  56. /// Set of stored addresses that are quarantined.
  57. /// We quarantine peers we've been unable to connect to, but we keep them
  58. /// around so we can potentially try them again, up to n tries. This should
  59. /// be helpful in order to self-heal the p2p connections in case we have an
  60. /// Internet interrupt (goblins unplugging cables)
  61. quarantine: RwLock<HashMap<Url, usize>>,
  62. /// Peers we reject from connecting to
  63. rejected: RwLock<HashSet<String>>,
  64. /// Peers that are currently being removed from the hostlist
  65. migrating: RwLock<HashSet<Url>>,
  66. /// Subscriber listening for store updates
  67. store_subscriber: SubscriberPtr<usize>,
  68. /// Pointer to configured P2P settings
  69. settings: SettingsPtr,
  70. }
  71. impl Hosts {
  72. /// Create a new hosts list>
  73. pub fn new(settings: SettingsPtr) -> HostsPtr {
  74. Arc::new(Self {
  75. greylist: RwLock::new(Vec::new()),
  76. whitelist: RwLock::new(Vec::new()),
  77. anchorlist: RwLock::new(Vec::new()),
  78. quarantine: RwLock::new(HashMap::new()),
  79. rejected: RwLock::new(HashSet::new()),
  80. migrating: RwLock::new(HashSet::new()),
  81. store_subscriber: Subscriber::new(),
  82. settings,
  83. })
  84. }
  85. /// Loops through greylist addresses to find an outbound address that we can
  86. /// connect to. Check whether the address is valid by making sure it isn't
  87. /// our own inbound address, then checks whether it is already connected
  88. /// (exists) or connecting (pending).
  89. /// Lastly adds matching address to the pending list.
  90. pub async fn greylist_fetch_address(&self, transports: &[String]) -> Vec<(Url, u64)> {
  91. trace!(target: "store", "greylist_fetch_address() [START]");
  92. // Collect hosts
  93. let mut hosts = vec![];
  94. // If transport mixing is enabled, then for example we're allowed to
  95. // use tor:// to connect to tcp:// and tor+tls:// to connect to tcp+tls://.
  96. // However, **do not** mix tor:// and tcp+tls://, nor tor+tls:// and tcp://.
  97. let transport_mixing = self.settings.transport_mixing;
  98. macro_rules! mix_transport {
  99. ($a:expr, $b:expr) => {
  100. if transports.contains(&$a.to_string()) && transport_mixing {
  101. let mut a_to_b =
  102. self.greylist_fetch_with_schemes(&[$b.to_string()], None).await;
  103. for (addr, last_seen) in a_to_b.iter_mut() {
  104. addr.set_scheme($a).unwrap();
  105. hosts.push((addr.clone(), last_seen.clone()));
  106. }
  107. }
  108. };
  109. }
  110. mix_transport!("tor", "tcp");
  111. mix_transport!("tor+tls", "tcp+tls");
  112. mix_transport!("nym", "tcp");
  113. mix_transport!("nym+tls", "tcp+tls");
  114. // And now the actual requested transports
  115. for (addr, last_seen) in self.greylist_fetch_with_schemes(transports, None).await {
  116. hosts.push((addr, last_seen));
  117. }
  118. hosts
  119. }
  120. /// Loops through whitelist addresses to find an outbound address that we can
  121. /// connect to. Check whether the address is valid by making sure it isn't
  122. /// our own inbound address, then checks whether it is already connected
  123. /// (exists) or connecting (pending).
  124. /// Lastly adds matching address to the pending list.
  125. pub async fn whitelist_fetch_address(&self, transports: &[String]) -> Vec<(Url, u64)> {
  126. trace!(target: "store", "whitelist_fetch_address() [START]");
  127. // Collect hosts
  128. let mut hosts = vec![];
  129. // If transport mixing is enabled, then for example we're allowed to
  130. // use tor:// to connect to tcp:// and tor+tls:// to connect to tcp+tls://.
  131. // However, **do not** mix tor:// and tcp+tls://, nor tor+tls:// and tcp://.
  132. let transport_mixing = self.settings.transport_mixing;
  133. macro_rules! mix_transport {
  134. ($a:expr, $b:expr) => {
  135. if transports.contains(&$a.to_string()) && transport_mixing {
  136. let mut a_to_b =
  137. self.whitelist_fetch_with_schemes(&[$b.to_string()], None).await;
  138. for (addr, last_seen) in a_to_b.iter_mut() {
  139. addr.set_scheme($a).unwrap();
  140. hosts.push((addr.clone(), last_seen.clone()));
  141. }
  142. }
  143. };
  144. }
  145. mix_transport!("tor", "tcp");
  146. mix_transport!("tor+tls", "tcp+tls");
  147. mix_transport!("nym", "tcp");
  148. mix_transport!("nym+tls", "tcp+tls");
  149. // And now the actual requested transports
  150. for (addr, last_seen) in self.whitelist_fetch_with_schemes(transports, None).await {
  151. hosts.push((addr, last_seen));
  152. }
  153. trace!(target: "store::whitelist_fetch_address()",
  154. "Grabbed hosts, length: {}", hosts.len());
  155. hosts
  156. }
  157. /// Loops through anchorlist addresses to find an outbound address that we can
  158. /// connect to. Check whether the address is valid by making sure it isn't
  159. /// our own inbound address, then checks whether it is already connected
  160. /// (exists) or connecting (pending).
  161. /// Lastly adds matching address to the pending list.
  162. pub async fn anchorlist_fetch_address(&self, transports: &[String]) -> Vec<(Url, u64)> {
  163. trace!(target: "store", "anchorlist_fetch_address() [START]");
  164. // Collect hosts
  165. let mut hosts = vec![];
  166. // If transport mixing is enabled, then for example we're allowed to
  167. // use tor:// to connect to tcp:// and tor+tls:// to connect to tcp+tls://.
  168. // However, **do not** mix tor:// and tcp+tls://, nor tor+tls:// and tcp://.
  169. let transport_mixing = self.settings.transport_mixing;
  170. macro_rules! mix_transport {
  171. ($a:expr, $b:expr) => {
  172. if transports.contains(&$a.to_string()) && transport_mixing {
  173. let mut a_to_b =
  174. self.anchorlist_fetch_with_schemes(&[$b.to_string()], None).await;
  175. for (addr, last_seen) in a_to_b.iter_mut() {
  176. addr.set_scheme($a).unwrap();
  177. hosts.push((addr.clone(), last_seen.clone()));
  178. }
  179. }
  180. };
  181. }
  182. mix_transport!("tor", "tcp");
  183. mix_transport!("tor+tls", "tcp+tls");
  184. mix_transport!("nym", "tcp");
  185. mix_transport!("nym+tls", "tcp+tls");
  186. // And now the actual requested transports
  187. for (addr, last_seen) in self.anchorlist_fetch_with_schemes(transports, None).await {
  188. hosts.push((addr, last_seen));
  189. }
  190. trace!(target: "store::anchorlist_fetch_address()",
  191. "Grabbed hosts, length: {}", hosts.len());
  192. hosts
  193. }
  194. /// Check whether:
  195. /// * We already have this connection established
  196. /// * We already have this configured as a manual peer
  197. /// * This address is already pending a connection
  198. pub async fn check_address_with_lock(
  199. &self,
  200. p2p: P2pPtr,
  201. hosts: Vec<(Url, u64)>,
  202. ) -> Option<(Url, u64)> {
  203. // Try to find an unused host in the set.
  204. for (host, last_seen) in hosts {
  205. debug!(target: "store::check_address_with_lock()",
  206. "Starting checks");
  207. // Check if we already have this connection established
  208. if p2p.exists(&host).await {
  209. debug!(
  210. target: "store::check_address_with_lock()",
  211. "Host '{}' exists so skipping",
  212. host
  213. );
  214. continue
  215. }
  216. // Check if we already have this configured as a manual peer
  217. if self.settings.peers.contains(&host) {
  218. debug!(
  219. target: "store::check_address_with_lock()",
  220. "Host '{}' configured as manual peer so skipping",
  221. host
  222. );
  223. continue
  224. }
  225. // Obtain a lock on this address to prevent duplicate connection
  226. if !p2p.add_pending(&host).await {
  227. debug!(
  228. target: "store::check_address_with_lock()",
  229. "Host '{}' pending so skipping",
  230. host
  231. );
  232. continue
  233. }
  234. debug!(
  235. target: "store::check_address_with_lock()",
  236. "Found valid host {}",
  237. host
  238. );
  239. return Some((host.clone(), last_seen))
  240. }
  241. None
  242. }
  243. /// Upgrade a connection to the anchorlist. Called after a connection has been successfully
  244. /// established in Outbound and Manual sessions.
  245. pub async fn upgrade_host(&self, addr: &Url) {
  246. let last_seen = UNIX_EPOCH.elapsed().unwrap().as_secs();
  247. self.anchorlist_store_or_update(&[(addr.clone(), last_seen)]).await;
  248. }
  249. /// Remove an entry from the hostlist. Called when we cannot establish a connection to a host or
  250. /// when a pre-existing connection disconnects.
  251. pub async fn remove_host(&self, addr: &Url) {
  252. debug!(target: "store::remove_host", "Removing host {}", addr);
  253. self.mark_migrating(addr).await;
  254. // Remove channel from anchorlist
  255. if self.anchorlist_contains(addr).await {
  256. debug!(target: "store::remove_host", "Removing from anchorlist {}", addr);
  257. let index = self
  258. .get_anchorlist_index_at_addr(addr.clone())
  259. .await
  260. .expect("Expected anchorlist index to exist");
  261. self.anchorlist_remove(addr, index).await;
  262. }
  263. // Remove channel from whitelist
  264. if self.whitelist_contains(addr).await {
  265. debug!(target: "store::remove_host", "Removing from whitelist {}", addr);
  266. let index = self
  267. .get_whitelist_index_at_addr(addr.clone())
  268. .await
  269. .expect("Expected whitelist index to exist");
  270. self.whitelist_remove(addr, index).await;
  271. }
  272. // Remove channel the greylist
  273. if self.greylist_contains(addr).await {
  274. debug!(target: "store::remove_host", "Removing from greylist {}", addr);
  275. let index = self
  276. .get_greylist_index_at_addr(addr.clone())
  277. .await
  278. .expect("Expected greylist index to exist");
  279. self.greylist_remove(addr, index).await;
  280. }
  281. self.unmark_migrating(addr).await;
  282. }
  283. /// Stores an address on the greylist or updates its last_seen field if we already
  284. /// have the address.
  285. pub async fn greylist_store_or_update(&self, addrs: &[(Url, u64)]) {
  286. trace!(target: "store::greylist_store_or_update()", "[START]");
  287. // Filter addresses before writing to the greylist.
  288. let filtered_addrs = self.filter_addresses(addrs).await;
  289. let filtered_addrs_len = filtered_addrs.len();
  290. if filtered_addrs.is_empty() {
  291. debug!(target: "store::greylist_store_or_update()", "Filtered out all received addresses");
  292. }
  293. for (addr, last_seen) in filtered_addrs {
  294. if !self.greylist_contains(&addr).await {
  295. debug!(target: "store::greylist_store_or_update()",
  296. "We do not have this entry in the hostlist. Adding to store...");
  297. self.greylist_store(addr.clone(), last_seen).await;
  298. } else {
  299. debug!(target: "store::greylist_store_or_update()",
  300. "We have this entry in the greylist. Updating last seen...");
  301. let index = self
  302. .get_greylist_index_at_addr(addr.clone())
  303. .await
  304. .expect("Expected greylist entry to exist");
  305. self.greylist_update_last_seen(&addr, last_seen, index).await;
  306. self.store_subscriber.notify(filtered_addrs_len).await;
  307. }
  308. }
  309. }
  310. /// Stores an address on the whitelist or updates its last_seen field if we already
  311. /// have the address.
  312. pub async fn whitelist_store_or_update(&self, addrs: &[(Url, u64)]) {
  313. trace!(target: "store::whitelist_store_or_update()", "[START]");
  314. // No address filtering for whitelist (whitelist is created from greylist)
  315. for (addr, last_seen) in addrs {
  316. if !self.whitelist_contains(addr).await {
  317. debug!(target: "store::whitelist_store_or_update()",
  318. "We do not have this entry in the whitelist. Adding to store...");
  319. self.whitelist_store(addr.clone(), *last_seen).await;
  320. } else {
  321. debug!(target: "store::whitelist_store_or_update()",
  322. "We have this entry in the whitelist. Updating last seen...");
  323. let index = self
  324. .get_whitelist_index_at_addr(addr.clone())
  325. .await
  326. .expect("Expected whitelist entry to exist");
  327. self.whitelist_update_last_seen(addr, *last_seen, index).await;
  328. }
  329. }
  330. }
  331. /// Stores an address on the anchorlist or updates its last_seen field if we already
  332. /// have the address.
  333. pub async fn anchorlist_store_or_update(&self, addrs: &[(Url, u64)]) {
  334. trace!(target: "store::anchor_store_or_update()", "[START]");
  335. // No address filtering for anchorlist (contains addresses we have already connected to)
  336. for (addr, last_seen) in addrs {
  337. if !self.anchorlist_contains(addr).await {
  338. debug!(target: "store::anchorlist_store_or_update()",
  339. "We do not have this entry in the whitelist. Adding to store...");
  340. self.anchorlist_store(addr.clone(), *last_seen).await;
  341. } else {
  342. debug!(target: "store::anchorlist_store_or_update()",
  343. "We have this entry in the anchorlist. Updating last seen...");
  344. let index = self
  345. .get_anchorlist_index_at_addr(addr.clone())
  346. .await
  347. .expect("Expected anchorlist entry to exist");
  348. self.anchorlist_update_last_seen(addr, *last_seen, index).await;
  349. }
  350. }
  351. }
  352. /// Append host to the greylist. Called on learning of a new peer.
  353. pub async fn greylist_store(&self, addr: Url, last_seen: u64) {
  354. trace!(target: "store::greylist_store()", "hosts::greylist_store() [START]");
  355. let mut greylist = self.greylist.write().await;
  356. // Remove oldest element if the greylist reaches max size.
  357. if greylist.len() == GREYLIST_MAX_LEN {
  358. let last_entry = greylist.pop().unwrap();
  359. debug!(target: "store::greylist_store()", "Greylist reached max size. Removed {:?}", last_entry);
  360. }
  361. debug!(target: "store::greylist_store()", "Inserting {}", addr);
  362. greylist.push((addr, last_seen));
  363. // Sort the list by last_seen.
  364. greylist.sort_by_key(|entry| entry.1);
  365. greylist.reverse();
  366. trace!(target: "store::greylist_store()", "[END]");
  367. }
  368. /// Append host to the whitelist. Called after a successful interaction with an online peer.
  369. pub async fn whitelist_store(&self, addr: Url, last_seen: u64) {
  370. trace!(target: "store::whitelist_store()", "[START]");
  371. let mut whitelist = self.whitelist.write().await;
  372. // Remove oldest element if the whitelist reaches max size.
  373. if whitelist.len() == WHITELIST_MAX_LEN {
  374. let last_entry = whitelist.pop().unwrap();
  375. debug!(target: "store::whitelist_store()", "Whitelist reached max size. Removed {:?}", last_entry);
  376. }
  377. trace!(target: "store::whitelist_store()", "Inserting {}. Last seen {:?}", addr, last_seen);
  378. whitelist.push((addr, last_seen));
  379. // Sort the list by last_seen.
  380. whitelist.sort_by_key(|entry| entry.1);
  381. whitelist.reverse();
  382. trace!(target: "store::whitelist_store()", "[END]");
  383. }
  384. /// Append host to the anchorlist. Called after we have successfully established a connection
  385. /// to a peer.
  386. pub async fn anchorlist_store(&self, addr: Url, last_seen: u64) {
  387. trace!(target: "store::anchorlist_store()", "[START]");
  388. let mut anchorlist = self.anchorlist.write().await;
  389. trace!(target: "store::anchorlist_store()", "Inserting {}", addr);
  390. anchorlist.push((addr, last_seen));
  391. // Sort the list by last_seen.
  392. anchorlist.sort_by_key(|entry| entry.1);
  393. anchorlist.reverse();
  394. trace!(target: "store::anchorlist_store()", "[END]");
  395. }
  396. /// Update the last_seen field of a peer on the greylist.
  397. pub async fn greylist_update_last_seen(&self, addr: &Url, last_seen: u64, index: usize) {
  398. trace!(target: "store::greylist_update_last_seen()", "[START]");
  399. let mut greylist = self.greylist.write().await;
  400. greylist[index] = (addr.clone(), last_seen);
  401. // Sort the list by last_seen.
  402. greylist.sort_by_key(|entry| entry.1);
  403. greylist.reverse();
  404. trace!(target: "store::greylist_update_last_seen()", "[END]");
  405. }
  406. /// Update the last_seen field of a peer on the whitelist.
  407. pub async fn whitelist_update_last_seen(&self, addr: &Url, last_seen: u64, index: usize) {
  408. trace!(target: "store::whitelist_update_last_seen()", "[START]");
  409. let mut whitelist = self.whitelist.write().await;
  410. whitelist[index] = (addr.clone(), last_seen);
  411. // Sort the list by last_seen.
  412. whitelist.sort_by_key(|entry| entry.1);
  413. whitelist.reverse();
  414. trace!(target: "store::whitelist_update_last_seen()", "[END]");
  415. }
  416. /// Update the last_seen field of a peer on the anchorlist.
  417. pub async fn anchorlist_update_last_seen(&self, addr: &Url, last_seen: u64, index: usize) {
  418. trace!(target: "store::anchorlist_update_last_seen()", "[START]");
  419. let mut anchorlist = self.anchorlist.write().await;
  420. anchorlist[index] = (addr.clone(), last_seen);
  421. // Sort the list by last_seen.
  422. anchorlist.sort_by_key(|entry| entry.1);
  423. anchorlist.reverse();
  424. trace!(target: "store::anchorlist_update_last_seen()", "[END]");
  425. }
  426. /// Remove an entry from the greylist.
  427. pub async fn greylist_remove(&self, addr: &Url, index: usize) {
  428. debug!(target: "store::greylist_remove", "Removing peer {} from greylist", addr);
  429. self.greylist.write().await.remove(index);
  430. }
  431. /// Remove an entry from the whitelist.
  432. pub async fn whitelist_remove(&self, addr: &Url, index: usize) {
  433. debug!(target: "store::whitelist_remove", "Removing peer {} from whitelist", addr);
  434. self.whitelist.write().await.remove(index);
  435. }
  436. /// Remove an entry from the anchorlist.
  437. pub async fn anchorlist_remove(&self, addr: &Url, index: usize) {
  438. debug!(target: "store::anchorlist_remove", "Removing peer {} from anchorlist", addr);
  439. self.anchorlist.write().await.remove(index);
  440. }
  441. pub async fn subscribe_store(&self) -> Result<Subscription<usize>> {
  442. let sub = self.store_subscriber.clone().subscribe().await;
  443. Ok(sub)
  444. }
  445. // Verify whether a URL is local.
  446. // NOTE: This function is stateless and not specific to
  447. // `Hosts`. For this reason, it might make more sense
  448. // to move this function to a more appropriate location
  449. // in the codebase.
  450. /// Check whether a URL is local host
  451. pub async fn is_local_host(&self, url: Url) -> bool {
  452. // Reject Urls without host strings.
  453. if url.host_str().is_none() {
  454. return false
  455. }
  456. // We do this hack in order to parse IPs properly.
  457. // https://github.com/whatwg/url/issues/749
  458. let addr = Url::parse(&url.as_str().replace(url.scheme(), "http")).unwrap();
  459. // Filter private IP ranges
  460. match addr.host().unwrap() {
  461. url::Host::Ipv4(ip) => {
  462. if !ip.is_global() {
  463. return true
  464. }
  465. }
  466. url::Host::Ipv6(ip) => {
  467. if !ip.is_global() {
  468. return true
  469. }
  470. }
  471. url::Host::Domain(d) => {
  472. if LOCAL_HOST_STRS.contains(&d) {
  473. return true
  474. }
  475. }
  476. }
  477. false
  478. }
  479. /// Filter given addresses based on certain rulesets and validity.
  480. async fn filter_addresses(&self, addrs: &[(Url, u64)]) -> Vec<(Url, u64)> {
  481. trace!(target: "store::filter_addresses()", "Filtering addrs: {:?}", addrs);
  482. let mut ret = vec![];
  483. let localnet = self.settings.localnet;
  484. 'addr_loop: for (addr_, last_seen) in addrs {
  485. // Validate that the format is `scheme://host_str:port`
  486. if addr_.host_str().is_none() ||
  487. addr_.port().is_none() ||
  488. addr_.cannot_be_a_base() ||
  489. addr_.path_segments().is_some()
  490. {
  491. continue
  492. }
  493. if self.is_rejected(addr_).await {
  494. debug!(target: "store::filter_addresses()", "Peer {} is rejected", addr_);
  495. continue
  496. }
  497. let host_str = addr_.host_str().unwrap();
  498. if !localnet {
  499. // Our own external addresses should never enter the hosts set.
  500. for ext in &self.settings.external_addrs {
  501. if host_str == ext.host_str().unwrap() {
  502. continue 'addr_loop
  503. }
  504. }
  505. }
  506. // On localnet, make sure ours ports don't enter the host set.
  507. for ext in &self.settings.external_addrs {
  508. if addr_.port() == ext.port() {
  509. continue 'addr_loop
  510. }
  511. }
  512. // We do this hack in order to parse IPs properly.
  513. // https://github.com/whatwg/url/issues/749
  514. let addr = Url::parse(&addr_.as_str().replace(addr_.scheme(), "http")).unwrap();
  515. // Filter non-global ranges if we're not allowing localnet.
  516. // Should never be allowed in production, so we don't really care
  517. // about some of them (e.g. 0.0.0.0, or broadcast, etc.).
  518. if !localnet && self.is_local_host(addr).await {
  519. continue
  520. }
  521. match addr_.scheme() {
  522. // Validate that the address is an actual onion.
  523. #[cfg(feature = "p2p-tor")]
  524. "tor" | "tor+tls" => {
  525. use std::str::FromStr;
  526. if tor_hscrypto::pk::HsId::from_str(host_str).is_err() {
  527. continue
  528. }
  529. trace!(target: "store::filter_addresses()", "[Tor] Valid: {}", host_str);
  530. }
  531. #[cfg(feature = "p2p-nym")]
  532. "nym" | "nym+tls" => continue, // <-- Temp skip
  533. #[cfg(feature = "p2p-tcp")]
  534. "tcp" | "tcp+tls" => {
  535. trace!(target: "store::filter_addresses()", "[TCP] Valid: {}", host_str);
  536. }
  537. _ => continue,
  538. }
  539. ret.push((addr_.clone(), *last_seen));
  540. }
  541. ret
  542. }
  543. /// Quarantine a peer.
  544. /// If they've been quarantined for more than a configured limit, forget them.
  545. pub async fn quarantine(&self, url: &Url) {
  546. debug!(target: "store::remove()", "Quarantining peer {}", url);
  547. // Remove from the entire hosts set
  548. self.remove_host(url).await;
  549. let mut q = self.quarantine.write().await;
  550. if let Some(retries) = q.get_mut(url) {
  551. *retries += 1;
  552. debug!(target: "net::hosts::quarantine()", "Peer {} quarantined {} times", url, retries);
  553. if *retries == self.settings.hosts_quarantine_limit {
  554. debug!(target: "net::hosts::quarantine()", "Banning peer {}", url);
  555. q.remove(url);
  556. self.mark_rejected(url).await;
  557. }
  558. } else {
  559. debug!(target: "net::hosts::remove()", "Added peer {} to quarantine", url);
  560. q.insert(url.clone(), 0);
  561. }
  562. }
  563. /// Check if a given peer (URL) is in the set of rejected hosts
  564. pub async fn is_rejected(&self, peer: &Url) -> bool {
  565. // Skip lookup for UNIX sockets and localhost connections
  566. // as they should never belong to the list of rejected URLs.
  567. let Some(hostname) = peer.host_str() else { return false };
  568. if self.is_local_host(peer.clone()).await {
  569. return false
  570. }
  571. self.rejected.read().await.contains(hostname)
  572. }
  573. /// Mark a peer as rejected by adding it to the set of rejected URLs.
  574. pub async fn mark_rejected(&self, peer: &Url) {
  575. // We ignore UNIX sockets here so we will just work
  576. // with stuff that has host_str().
  577. if let Some(hostname) = peer.host_str() {
  578. // Localhost connections should not be rejected
  579. // This however allows any Tor and Nym connections.
  580. if self.is_local_host(peer.clone()).await {
  581. return
  582. }
  583. self.rejected.write().await.insert(hostname.to_string());
  584. }
  585. }
  586. /// Unmark a rejected peer
  587. pub async fn unmark_rejected(&self, peer: &Url) {
  588. if let Some(hostname) = peer.host_str() {
  589. self.rejected.write().await.remove(hostname);
  590. }
  591. }
  592. /// Peer that is currently being removed from hostlists.
  593. pub async fn is_migrating(&self, peer: &Url) -> bool {
  594. self.migrating.read().await.contains(peer)
  595. }
  596. /// Mark a peer as currently migrating.
  597. pub async fn mark_migrating(&self, peer: &Url) {
  598. self.migrating.write().await.insert(peer.clone());
  599. }
  600. /// Unmark a migrating peer.
  601. pub async fn unmark_migrating(&self, peer: &Url) {
  602. self.migrating.write().await.remove(peer);
  603. }
  604. /// Check if the greylist is empty.
  605. pub async fn is_empty_greylist(&self) -> bool {
  606. self.greylist.read().await.is_empty()
  607. }
  608. /// Check if the whitelist is empty.
  609. pub async fn is_empty_whitelist(&self) -> bool {
  610. self.whitelist.read().await.is_empty()
  611. }
  612. /// Check if the anchorlist is empty.
  613. pub async fn is_empty_anchorlist(&self) -> bool {
  614. self.anchorlist.read().await.is_empty()
  615. }
  616. /// Check if the hostlist is empty.
  617. pub async fn is_empty_hostlist(&self) -> bool {
  618. self.is_empty_greylist().await &&
  619. self.is_empty_whitelist().await &&
  620. self.is_empty_anchorlist().await
  621. }
  622. /// Check if host is in the greylist
  623. pub async fn greylist_contains(&self, addr: &Url) -> bool {
  624. self.greylist.read().await.iter().any(|(u, _t)| u == addr)
  625. }
  626. /// Check if host is in the whitelist
  627. pub async fn whitelist_contains(&self, addr: &Url) -> bool {
  628. self.whitelist.read().await.iter().any(|(u, _t)| u == addr)
  629. }
  630. /// Check if host is in the anchorlist
  631. pub async fn anchorlist_contains(&self, addr: &Url) -> bool {
  632. self.anchorlist.read().await.iter().any(|(u, _t)| u == addr)
  633. }
  634. /// Get the index for a given addr on the greylist.
  635. pub async fn get_greylist_index_at_addr(&self, addr: Url) -> Option<usize> {
  636. self.greylist.read().await.iter().position(|a| a.0 == addr)
  637. }
  638. /// Get the index for a given addr on the whitelist.
  639. pub async fn get_whitelist_index_at_addr(&self, addr: Url) -> Option<usize> {
  640. self.whitelist.read().await.iter().position(|a| a.0 == addr)
  641. }
  642. /// Get the index for a given addr on the anchorlist.
  643. pub async fn get_anchorlist_index_at_addr(&self, addr: Url) -> Option<usize> {
  644. self.anchorlist.read().await.iter().position(|a| a.0 == addr)
  645. }
  646. /// Get the entry for a given addr on the whitelist.
  647. pub async fn get_whitelist_entry_at_addr(&self, addr: &Url) -> Option<(Url, u64)> {
  648. self.whitelist
  649. .read()
  650. .await
  651. .iter()
  652. .find(|(url, _)| url == addr)
  653. .map(|(url, time)| (url.clone(), *time))
  654. }
  655. /// Get the entry for a given addr on the anchorlist.
  656. pub async fn get_anchorlist_entry_at_addr(&self, addr: &Url) -> Option<(Url, u64)> {
  657. self.anchorlist
  658. .read()
  659. .await
  660. .iter()
  661. .find(|(url, _)| url == addr)
  662. .map(|(url, time)| (url.clone(), *time))
  663. }
  664. /// Return all known whitelisted hosts
  665. pub async fn whitelist_fetch_all(&self) -> Vec<(Url, u64)> {
  666. self.whitelist.read().await.iter().cloned().collect()
  667. }
  668. /// Return all known greylisted hosts
  669. pub async fn greylist_fetch_all(&self) -> Vec<(Url, u64)> {
  670. self.greylist.read().await.iter().cloned().collect()
  671. }
  672. /// Return all known anchorlisted hosts
  673. pub async fn anchorlist_fetch_all(&self) -> Vec<(Url, u64)> {
  674. self.anchorlist.read().await.iter().cloned().collect()
  675. }
  676. /// Return all greylist and anchorlist hosts. Called on stop().
  677. /// Note: we do not return whitelist entries here since whitelist entries must go via the
  678. /// greylist refinery in the lifetime of the p2p network.
  679. pub async fn hostlist_fetch_safe(&self) -> HashMap<String, Vec<(Url, u64)>> {
  680. let mut hostlist = HashMap::new();
  681. hostlist.insert(
  682. "anchorlist".to_string(),
  683. self.anchorlist.read().await.iter().cloned().collect(),
  684. );
  685. hostlist
  686. .insert("greylist".to_string(), self.greylist.read().await.iter().cloned().collect());
  687. hostlist
  688. }
  689. /// Get up to n random peers from the whitelist.
  690. pub async fn whitelist_fetch_n_random(&self, n: u32) -> Vec<(Url, u64)> {
  691. let n = n as usize;
  692. if n == 0 {
  693. return vec![]
  694. }
  695. let addrs = self.whitelist.read().await;
  696. let urls = addrs.iter().choose_multiple(&mut OsRng, n.min(addrs.len()));
  697. urls.iter().map(|&url| url.clone()).collect()
  698. }
  699. /// Get a random peer from the greylist.
  700. pub async fn greylist_fetch_random(&self) -> ((Url, u64), usize) {
  701. let greylist = self.greylist.read().await;
  702. let position = rand::thread_rng().gen_range(0..greylist.len());
  703. let entry = &greylist[position];
  704. (entry.clone(), position)
  705. }
  706. /// Get a random peer from the whitelist.
  707. pub async fn whitelist_fetch_random(&self) -> ((Url, u64), usize) {
  708. let whitelist = self.whitelist.read().await;
  709. let position = rand::thread_rng().gen_range(0..whitelist.len());
  710. let entry = &whitelist[position];
  711. (entry.clone(), position)
  712. }
  713. /// Get the oldest entry from the whitelist.
  714. pub async fn whitelist_fetch_last(&self) -> ((Url, u64), usize) {
  715. let whitelist = self.whitelist.read().await;
  716. let position = whitelist.len() - 1;
  717. let entry = &whitelist[position];
  718. (entry.clone(), position)
  719. }
  720. /// Get a random greylist peer that matches the given transport schemes.
  721. pub async fn greylist_fetch_random_with_schemes(&self) -> Option<((Url, u64), usize)> {
  722. trace!(target: "store::greylist_fetch_random_with_schemes", "[START]");
  723. // Retrieve all peers corresponding to that transport schemes
  724. let schemes = &self.settings.allowed_transports;
  725. let greylist = self.greylist_fetch_with_schemes(schemes, None).await;
  726. if greylist.is_empty() {
  727. return None
  728. }
  729. let position = rand::thread_rng().gen_range(0..greylist.len());
  730. let entry = &greylist[position];
  731. Some((entry.clone(), position))
  732. }
  733. /// Get up to n random greylist peers. Schemes are not taken into account.
  734. pub async fn greylist_fetch_n_random(&self, n: u32) -> Vec<(Url, u64)> {
  735. trace!(target: "store::greylist_fetch_n_random", "[START]");
  736. let n = n as usize;
  737. if n == 0 {
  738. return vec![]
  739. }
  740. let mut hosts = vec![];
  741. let greylist = self.greylist.read().await;
  742. for (addr, last_seen) in greylist.iter() {
  743. hosts.push((addr.clone(), *last_seen));
  744. }
  745. if hosts.is_empty() {
  746. debug!(target: "store::greylist_fetch_n_random", "No greylist entries found!");
  747. return hosts
  748. }
  749. // Grab random ones
  750. let urls = hosts.iter().choose_multiple(&mut OsRng, n.min(hosts.len()));
  751. urls.iter().map(|&url| url.clone()).collect()
  752. }
  753. /// Get up to n random greylist peers that match the given transport schemes.
  754. pub async fn greylist_fetch_n_random_with_schemes(
  755. &self,
  756. schemes: &[String],
  757. n: u32,
  758. ) -> Vec<(Url, u64)> {
  759. let n = n as usize;
  760. if n == 0 {
  761. return vec![]
  762. }
  763. trace!(target: "store::greylist_fetch_n_random_with_schemes", "[START]");
  764. // Retrieve all peers corresponding to that transport schemes
  765. let hosts = self.greylist_fetch_with_schemes(schemes, None).await;
  766. if hosts.is_empty() {
  767. debug!(target: "store::greylist_fetch_n_random_with_schemes",
  768. "No such schemes found on greylist!");
  769. return hosts
  770. }
  771. // Grab random ones
  772. let urls = hosts.iter().choose_multiple(&mut OsRng, n.min(hosts.len()));
  773. urls.iter().map(|&url| url.clone()).collect()
  774. }
  775. /// Get up to n random whitelist peers that match the given transport schemes.
  776. pub async fn whitelist_fetch_n_random_with_schemes(
  777. &self,
  778. schemes: &[String],
  779. n: u32,
  780. ) -> Vec<(Url, u64)> {
  781. let n = n as usize;
  782. if n == 0 {
  783. return vec![]
  784. }
  785. trace!(target: "store::whitelist_fetch_n_random_with_schemes", "[START]");
  786. // Retrieve all peers corresponding to that transport schemes
  787. let hosts = self.whitelist_fetch_with_schemes(schemes, None).await;
  788. if hosts.is_empty() {
  789. debug!(target: "store::whitelist_fetch_n_random_with_schemes",
  790. "No such schemes found on whitelist!");
  791. return hosts
  792. }
  793. // Grab random ones
  794. let urls = hosts.iter().choose_multiple(&mut OsRng, n.min(hosts.len()));
  795. urls.iter().map(|&url| url.clone()).collect()
  796. }
  797. /// Get up to n random anchorlist peers that match the given transport schemes.
  798. pub async fn anchorlist_fetch_n_random_with_schemes(
  799. &self,
  800. schemes: &[String],
  801. n: u32,
  802. ) -> Vec<(Url, u64)> {
  803. let n = n as usize;
  804. if n == 0 {
  805. return vec![]
  806. }
  807. trace!(target: "store::anchorlist_fetch_n_random_with_schemes", "[START]");
  808. // Retrieve all peers corresponding to that transport schemes
  809. let hosts = self.anchorlist_fetch_with_schemes(schemes, None).await;
  810. if hosts.is_empty() {
  811. debug!(target: "store::anchorlist_fetch_n_random_with_schemes",
  812. "No such schemes found on anchorlist!");
  813. return hosts
  814. }
  815. // Grab random ones
  816. let urls = hosts.iter().choose_multiple(&mut OsRng, n.min(hosts.len()));
  817. urls.iter().map(|&url| url.clone()).collect()
  818. }
  819. /// Get up to limit peers that don't match the given transport schemes from the greylist.
  820. /// If limit was not provided, return all matching peers.
  821. pub async fn greylist_fetch_excluding_schemes(
  822. &self,
  823. schemes: &[String],
  824. limit: Option<usize>,
  825. ) -> Vec<(Url, u64)> {
  826. let greylist = self.greylist.read().await;
  827. let mut limit = match limit {
  828. Some(l) => l.min(greylist.len()),
  829. None => greylist.len(),
  830. };
  831. let mut ret = vec![];
  832. if limit == 0 {
  833. return ret
  834. }
  835. for (addr, last_seen) in greylist.iter() {
  836. if !schemes.contains(&addr.scheme().to_string()) {
  837. ret.push((addr.clone(), *last_seen));
  838. limit -= 1;
  839. if limit == 0 {
  840. return ret
  841. }
  842. }
  843. }
  844. if ret.is_empty() {
  845. debug!(target: "store::greylist_fetch_excluding_schemes",
  846. "No such schemes found on greylist!")
  847. }
  848. ret
  849. }
  850. /// Get up to limit peers that don't match the given transport schemes from the whitelist.
  851. /// If limit was not provided, return all matching peers.
  852. pub async fn whitelist_fetch_excluding_schemes(
  853. &self,
  854. schemes: &[String],
  855. limit: Option<usize>,
  856. ) -> Vec<(Url, u64)> {
  857. let addrs = self.whitelist.read().await;
  858. let mut limit = match limit {
  859. Some(l) => l.min(addrs.len()),
  860. None => addrs.len(),
  861. };
  862. let mut ret = vec![];
  863. if limit == 0 {
  864. return ret
  865. }
  866. for (addr, last_seen) in addrs.iter() {
  867. if !schemes.contains(&addr.scheme().to_string()) {
  868. ret.push((addr.clone(), *last_seen));
  869. limit -= 1;
  870. if limit == 0 {
  871. return ret
  872. }
  873. }
  874. }
  875. if ret.is_empty() {
  876. debug!(target: "store::whiteist_fetch_excluding_schemes",
  877. "No such schemes found on whitelist!")
  878. }
  879. ret
  880. }
  881. /// Get up to n random whitelisted peers that don't match the given transport schemes from the
  882. /// hosts set.
  883. pub async fn whitelist_fetch_n_random_excluding_schemes(
  884. &self,
  885. schemes: &[String],
  886. n: u32,
  887. ) -> Vec<(Url, u64)> {
  888. let n = n as usize;
  889. if n == 0 {
  890. return vec![]
  891. }
  892. trace!(target: "store::whitelist_fetch_excluding_schemes", "[START]");
  893. // Retrieve all peers not corresponding to that transport schemes
  894. let hosts = self.whitelist_fetch_excluding_schemes(schemes, None).await;
  895. if hosts.is_empty() {
  896. debug!(target: "store::whitelist_fetch_n_random_excluding_schemes",
  897. "No such schemes found on whitelist!");
  898. return hosts
  899. }
  900. // Grab random ones
  901. let urls = hosts.iter().choose_multiple(&mut OsRng, n.min(hosts.len()));
  902. urls.iter().map(|&url| url.clone()).collect()
  903. }
  904. /// Get up to limit peers that match the given transport schemes from the greylist.
  905. /// If limit was not provided, return all matching peers.
  906. async fn greylist_fetch_with_schemes(
  907. &self,
  908. schemes: &[String],
  909. limit: Option<usize>,
  910. ) -> Vec<(Url, u64)> {
  911. trace!(target: "store::greylist_fetch_with_schemes", "[START]");
  912. let greylist = self.greylist.read().await;
  913. let mut limit = match limit {
  914. Some(l) => l.min(greylist.len()),
  915. None => greylist.len(),
  916. };
  917. let mut ret = vec![];
  918. if limit == 0 {
  919. return ret
  920. }
  921. for (addr, last_seen) in greylist.iter() {
  922. if schemes.contains(&addr.scheme().to_string()) {
  923. ret.push((addr.clone(), *last_seen));
  924. limit -= 1;
  925. if limit == 0 {
  926. debug!(target: "store::greylist_fetch_with_schemes",
  927. "Found matching scheme, returning {} grey addresses",
  928. ret.len());
  929. return ret
  930. }
  931. }
  932. }
  933. if ret.is_empty() {
  934. debug!(target: "store::greylist_fetch_with_schemes",
  935. "No such schemes found on greylist!")
  936. }
  937. trace!(target: "store::greylist_fetch_with_schemes", "END");
  938. ret
  939. }
  940. /// Get up to limit peers that match the given transport schemes from the whitelist.
  941. /// If limit was not provided, return all matching peers.
  942. async fn whitelist_fetch_with_schemes(
  943. &self,
  944. schemes: &[String],
  945. limit: Option<usize>,
  946. ) -> Vec<(Url, u64)> {
  947. trace!(target: "store::whitelist_fetch_with_schemes", "[START]");
  948. let whitelist = self.whitelist.read().await;
  949. let mut limit = match limit {
  950. Some(l) => l.min(whitelist.len()),
  951. None => whitelist.len(),
  952. };
  953. let mut ret = vec![];
  954. if limit == 0 {
  955. return ret
  956. }
  957. for (addr, last_seen) in whitelist.iter() {
  958. if schemes.contains(&addr.scheme().to_string()) {
  959. ret.push((addr.clone(), *last_seen));
  960. limit -= 1;
  961. if limit == 0 {
  962. debug!(target: "store::whitelist_fetch_with_schemes",
  963. "Found matching scheme, returning {} white addresses",
  964. ret.len());
  965. return ret
  966. }
  967. }
  968. }
  969. if ret.is_empty() {
  970. debug!(target: "store::whitelist_fetch_with_schemes",
  971. "No such schemes found on whitelist!")
  972. }
  973. trace!(target: "store::whitelist_fetch_with_schemes", "END");
  974. ret
  975. }
  976. /// Get up to limit peers that match the given transport schemes from the anchorlist.
  977. /// If limit was not provided, return all matching peers.
  978. async fn anchorlist_fetch_with_schemes(
  979. &self,
  980. schemes: &[String],
  981. limit: Option<usize>,
  982. ) -> Vec<(Url, u64)> {
  983. //trace!(target: "store::anchorlist_fetch_with_schemes", "[START]");
  984. let anchorlist = self.anchorlist.read().await;
  985. let mut limit = match limit {
  986. Some(l) => l.min(anchorlist.len()),
  987. None => anchorlist.len(),
  988. };
  989. let mut ret = vec![];
  990. if limit == 0 {
  991. return ret
  992. }
  993. for (addr, last_seen) in anchorlist.iter() {
  994. if schemes.contains(&addr.scheme().to_string()) {
  995. ret.push((addr.clone(), *last_seen));
  996. limit -= 1;
  997. if limit == 0 {
  998. debug!(target: "store::anchorlist_fetch_with_schemes",
  999. "Found matching scheme, returning {} anchor addresses",
  1000. ret.len());
  1001. return ret
  1002. }
  1003. }
  1004. }
  1005. if ret.is_empty() {
  1006. warn!(target: "store::anchorlist_fetch_with_schemes",
  1007. "No matching schemes found on anchorlist")
  1008. }
  1009. trace!(target: "store::anchorlist_fetch_with_schemes", "END");
  1010. ret
  1011. }
  1012. /// Load the hostlist from a file.
  1013. pub async fn load_hosts(&self) -> Result<()> {
  1014. let path = expand_path(&self.settings.hostlist)?;
  1015. if !path.exists() {
  1016. if let Some(parent) = path.parent() {
  1017. fs::create_dir_all(parent)?;
  1018. }
  1019. File::create(path.clone())?;
  1020. }
  1021. let contents = load_file(&path);
  1022. if let Err(e) = contents {
  1023. warn!(target: "store", "Failed retrieving saved hosts: {}", e);
  1024. return Ok(())
  1025. }
  1026. for line in contents.unwrap().lines() {
  1027. let data: Vec<&str> = line.split('\t').collect();
  1028. let url = match Url::parse(data[1]) {
  1029. Ok(u) => u,
  1030. Err(e) => {
  1031. debug!(target: "store", "load_hosts(): Skipping malformed URL {}", e);
  1032. continue
  1033. }
  1034. };
  1035. let last_seen = match data[2].parse::<u64>() {
  1036. Ok(t) => t,
  1037. Err(e) => {
  1038. debug!(target: "store", "load_hosts(): Skipping malformed last seen {}", e);
  1039. continue
  1040. }
  1041. };
  1042. match data[0] {
  1043. "greylist" => {
  1044. self.greylist_store(url, last_seen).await;
  1045. }
  1046. "whitelist" => {
  1047. self.whitelist_store(url, last_seen).await;
  1048. }
  1049. "anchorlist" => {
  1050. self.anchorlist_store(url, last_seen).await;
  1051. }
  1052. _ => {
  1053. debug!(target: "store", "load_hosts(): Malformed list name...");
  1054. }
  1055. }
  1056. }
  1057. Ok(())
  1058. }
  1059. /// Save the hostlist to a file. Whitelist gets written to the greylist to force
  1060. /// whitelist entries through the refinery on start.
  1061. pub async fn save_hosts(&self) -> Result<()> {
  1062. let path = expand_path(&self.settings.hostlist)?;
  1063. let mut tsv = String::new();
  1064. let mut whitelist = vec![];
  1065. // First gather all the whitelist entries we don't have in greylist.
  1066. for (url, last_seen) in self.whitelist_fetch_all().await {
  1067. if !self.greylist_contains(&url).await {
  1068. whitelist.push((url, last_seen))
  1069. }
  1070. }
  1071. // Collect the greylist and anchorlist entries, and append any whitelist entries to the
  1072. // greylist before saving.
  1073. for (name, mut list) in self.hostlist_fetch_safe().await {
  1074. if name == *"greylist".to_string() {
  1075. list.append(&mut whitelist)
  1076. }
  1077. for (url, last_seen) in list {
  1078. tsv.push_str(&format!("{}\t{}\t{}\n", name, url, last_seen));
  1079. }
  1080. }
  1081. if !tsv.eq("") {
  1082. info!(target: "store", "Saving hosts to: {:?}",
  1083. path);
  1084. if let Err(e) = save_file(&path, &tsv) {
  1085. error!(target: "store", "Failed saving hosts: {}", e);
  1086. }
  1087. }
  1088. Ok(())
  1089. }
  1090. }
  1091. #[cfg(test)]
  1092. mod tests {
  1093. use super::{
  1094. super::super::{settings::Settings, P2p},
  1095. *,
  1096. };
  1097. use crate::{net::hosts::refinery::ping_node, system::sleep};
  1098. use smol::Executor;
  1099. use std::{sync::Arc, time::UNIX_EPOCH};
  1100. #[test]
  1101. fn test_ping_node() {
  1102. smol::block_on(async {
  1103. let settings = Settings {
  1104. localnet: false,
  1105. external_addrs: vec![
  1106. Url::parse("tcp://foo.bar:123").unwrap(),
  1107. Url::parse("tcp://lol.cat:321").unwrap(),
  1108. ],
  1109. ..Default::default()
  1110. };
  1111. let ex = Arc::new(Executor::new());
  1112. let p2p = P2p::new(settings, ex.clone()).await;
  1113. let url = Url::parse("tcp://xeno.systems.wtf").unwrap();
  1114. println!("Pinging node...");
  1115. let task = ex.spawn(ping_node(url.clone(), p2p));
  1116. ex.run(task).await;
  1117. println!("Ping node complete!");
  1118. });
  1119. }
  1120. #[test]
  1121. fn test_is_local_host() {
  1122. smol::block_on(async {
  1123. let settings = Settings {
  1124. localnet: false,
  1125. external_addrs: vec![
  1126. Url::parse("tcp://foo.bar:123").unwrap(),
  1127. Url::parse("tcp://lol.cat:321").unwrap(),
  1128. ],
  1129. ..Default::default()
  1130. };
  1131. let hosts = Hosts::new(Arc::new(settings.clone()));
  1132. let local_hosts: Vec<Url> = vec![
  1133. Url::parse("tcp://localhost").unwrap(),
  1134. Url::parse("tcp://127.0.0.1").unwrap(),
  1135. Url::parse("tcp+tls://[::1]").unwrap(),
  1136. Url::parse("tcp://localhost.localdomain").unwrap(),
  1137. Url::parse("tcp://192.168.10.65").unwrap(),
  1138. ];
  1139. for host in local_hosts {
  1140. eprintln!("{}", host);
  1141. assert!(hosts.is_local_host(host).await);
  1142. }
  1143. let remote_hosts: Vec<Url> = vec![
  1144. Url::parse("https://dyne.org").unwrap(),
  1145. Url::parse("tcp://77.168.10.65:2222").unwrap(),
  1146. Url::parse("tcp://[2345:0425:2CA1:0000:0000:0567:5673:23b5]").unwrap(),
  1147. Url::parse("http://eweiibe6tdjsdprb4px6rqrzzcsi22m4koia44kc5pcjr7nec2rlxyad.onion")
  1148. .unwrap(),
  1149. ];
  1150. for host in remote_hosts {
  1151. assert!(!(hosts.is_local_host(host).await))
  1152. }
  1153. });
  1154. }
  1155. #[test]
  1156. fn test_greylist_store() {
  1157. let last_seen = UNIX_EPOCH.elapsed().unwrap().as_secs();
  1158. smol::block_on(async {
  1159. let settings = Settings {
  1160. localnet: false,
  1161. external_addrs: vec![
  1162. Url::parse("tcp://foo.bar:123").unwrap(),
  1163. Url::parse("tcp://lol.cat:321").unwrap(),
  1164. ],
  1165. ..Default::default()
  1166. };
  1167. let hosts = Hosts::new(Arc::new(settings.clone()));
  1168. for addr in settings.external_addrs {
  1169. hosts.greylist_store(addr, last_seen).await;
  1170. }
  1171. assert!(!hosts.is_empty_greylist().await);
  1172. let local_hosts = vec![
  1173. (Url::parse("tcp://localhost:3921").unwrap()),
  1174. (Url::parse("tor://[::1]:21481").unwrap()),
  1175. (Url::parse("tcp://192.168.10.65:311").unwrap()),
  1176. (Url::parse("tcp+tls://0.0.0.0:2312").unwrap()),
  1177. (Url::parse("tcp://255.255.255.255:2131").unwrap()),
  1178. ];
  1179. for host in &local_hosts {
  1180. hosts.greylist_store(host.clone(), last_seen).await;
  1181. }
  1182. assert!(!hosts.is_empty_greylist().await);
  1183. let remote_hosts = vec![
  1184. (Url::parse("tcp://dark.fi:80").unwrap()),
  1185. (Url::parse("tcp://http.cat:401").unwrap()),
  1186. (Url::parse("tcp://foo.bar:111").unwrap()),
  1187. ];
  1188. for host in &remote_hosts {
  1189. hosts.greylist_store(host.clone(), last_seen).await;
  1190. }
  1191. assert!(hosts.greylist_contains(&remote_hosts[0]).await);
  1192. assert!(hosts.greylist_contains(&remote_hosts[1]).await);
  1193. assert!(hosts.greylist_contains(&remote_hosts[2]).await);
  1194. });
  1195. }
  1196. #[test]
  1197. fn test_whitelist_store() {
  1198. smol::block_on(async {
  1199. let settings = Settings {
  1200. localnet: false,
  1201. external_addrs: vec![
  1202. Url::parse("tcp://foo.bar:123").unwrap(),
  1203. Url::parse("tcp://lol.cat:321").unwrap(),
  1204. ],
  1205. ..Default::default()
  1206. };
  1207. let hosts = Hosts::new(Arc::new(settings.clone()));
  1208. assert!(hosts.is_empty_whitelist().await);
  1209. let url = Url::parse("tcp://dark.renaissance:333").unwrap();
  1210. let last_seen = UNIX_EPOCH.elapsed().unwrap().as_secs();
  1211. hosts.whitelist_store(url.clone(), last_seen).await;
  1212. assert!(!hosts.is_empty_whitelist().await);
  1213. assert!(hosts.whitelist_contains(&url).await);
  1214. });
  1215. }
  1216. #[test]
  1217. fn test_whitelist_get_last() {
  1218. smol::block_on(async {
  1219. let settings = Settings {
  1220. localnet: false,
  1221. external_addrs: vec![
  1222. Url::parse("tcp://foo.bar:123").unwrap(),
  1223. Url::parse("tcp://lol.cat:321").unwrap(),
  1224. ],
  1225. ..Default::default()
  1226. };
  1227. let hosts = Hosts::new(Arc::new(settings.clone()));
  1228. // Build up a hostlist
  1229. for i in 0..10 {
  1230. sleep(1).await;
  1231. let last_seen = UNIX_EPOCH.elapsed().unwrap().as_secs();
  1232. let url = Url::parse(&format!("tcp://whitelist{}:123", i)).unwrap();
  1233. hosts.whitelist_store(url.clone(), last_seen).await;
  1234. }
  1235. for (url, last_seen) in hosts.whitelist.read().await.iter() {
  1236. println!("{} {}", url, last_seen);
  1237. }
  1238. let (entry, _position) = hosts.whitelist_fetch_last().await;
  1239. println!("last entry: {} {}", entry.0, entry.1);
  1240. });
  1241. }
  1242. #[test]
  1243. fn test_hostlist_get_entry() {
  1244. smol::block_on(async {
  1245. let settings = Settings {
  1246. localnet: false,
  1247. external_addrs: vec![
  1248. Url::parse("tcp://foo.bar:123").unwrap(),
  1249. Url::parse("tcp://lol.cat:321").unwrap(),
  1250. ],
  1251. ..Default::default()
  1252. };
  1253. let hosts = Hosts::new(Arc::new(settings.clone()));
  1254. let url = Url::parse("tcp://dark.renaissance:333").unwrap();
  1255. let last_seen = UNIX_EPOCH.elapsed().unwrap().as_secs();
  1256. hosts.whitelist_store(url.clone(), last_seen).await;
  1257. hosts.anchorlist_store(url.clone(), last_seen).await;
  1258. assert!(hosts.get_whitelist_entry_at_addr(&url).await.is_some());
  1259. assert!(hosts.get_anchorlist_entry_at_addr(&url).await.is_some());
  1260. });
  1261. }
  1262. #[test]
  1263. fn test_remove() {
  1264. smol::block_on(async {
  1265. let settings = Settings {
  1266. outbound_connections: 8,
  1267. allowed_transports: vec!["tcp".to_string()],
  1268. ..Default::default()
  1269. };
  1270. let hosts = Hosts::new(Arc::new(settings.clone()));
  1271. let url = Url::parse("tcp://dark.renaissance:333").unwrap();
  1272. let last_seen = UNIX_EPOCH.elapsed().unwrap().as_secs();
  1273. hosts.whitelist_store(url.clone(), last_seen).await;
  1274. sleep(1).await;
  1275. let url = Url::parse("tcp://milady:333").unwrap();
  1276. let last_seen = UNIX_EPOCH.elapsed().unwrap().as_secs();
  1277. hosts.whitelist_store(url.clone(), last_seen).await;
  1278. sleep(1).await;
  1279. let url = Url::parse("tcp://king-ted:333").unwrap();
  1280. let last_seen = UNIX_EPOCH.elapsed().unwrap().as_secs();
  1281. hosts.whitelist_store(url.clone(), last_seen).await;
  1282. for (url, last_seen) in hosts.whitelist.read().await.iter() {
  1283. println!("{}, {}", url, last_seen);
  1284. }
  1285. let position = hosts.get_whitelist_index_at_addr(url.clone()).await.unwrap();
  1286. hosts.whitelist_remove(&url, position).await;
  1287. for (url, last_seen) in hosts.whitelist.read().await.iter() {
  1288. println!("{}, {}", url, last_seen);
  1289. }
  1290. });
  1291. }
  1292. #[test]
  1293. fn test_fetch_address() {
  1294. smol::block_on(async {
  1295. let mut hostlist = vec![];
  1296. let mut grey_urls = vec![];
  1297. let mut white_urls = vec![];
  1298. let mut anchor_urls = vec![];
  1299. let ex = Arc::new(Executor::new());
  1300. let settings = Settings {
  1301. outbound_connections: 8,
  1302. allowed_transports: vec!["tcp".to_string()],
  1303. ..Default::default()
  1304. };
  1305. let p2p = P2p::new(settings, ex.clone()).await;
  1306. let hosts = p2p.hosts();
  1307. // Build up a hostlist
  1308. for i in 0..5 {
  1309. let last_seen = UNIX_EPOCH.elapsed().unwrap().as_secs();
  1310. hosts
  1311. .anchorlist_store(
  1312. Url::parse(&format!("tcp://anchorlist{}:123", i)).unwrap(),
  1313. last_seen,
  1314. )
  1315. .await;
  1316. hosts
  1317. .whitelist_store(
  1318. Url::parse(&format!("tcp://whitelist{}:123", i)).unwrap(),
  1319. last_seen,
  1320. )
  1321. .await;
  1322. hosts
  1323. .greylist_store(
  1324. Url::parse(&format!("tcp://greylist{}:123", i)).unwrap(),
  1325. last_seen,
  1326. )
  1327. .await;
  1328. grey_urls
  1329. .push((Url::parse(&format!("tcp://greylist{}:123", i)).unwrap(), last_seen));
  1330. white_urls
  1331. .push((Url::parse(&format!("tcp://whitelist{}:123", i)).unwrap(), last_seen));
  1332. anchor_urls
  1333. .push((Url::parse(&format!("tcp://anchorlist{}:123", i)).unwrap(), last_seen));
  1334. }
  1335. assert!(!hosts.is_empty_anchorlist().await);
  1336. assert!(!hosts.is_empty_whitelist().await);
  1337. assert!(!hosts.is_empty_greylist().await);
  1338. let transports = &p2p.settings().allowed_transports;
  1339. let white_count =
  1340. p2p.settings().outbound_connections * p2p.settings().white_connection_percent / 100;
  1341. // Simulate the address selection logic found in outbound_session::fetch_address()
  1342. for i in 0..8 {
  1343. if i < p2p.settings().anchor_connection_count {
  1344. if !hosts.anchorlist_fetch_address(transports).await.is_empty() {
  1345. let addrs = hosts.anchorlist_fetch_address(transports).await;
  1346. hostlist.push(addrs);
  1347. }
  1348. if !hosts.whitelist_fetch_address(transports).await.is_empty() {
  1349. let addrs = hosts.whitelist_fetch_address(transports).await;
  1350. hostlist.push(addrs);
  1351. }
  1352. if !hosts.greylist_fetch_address(transports).await.is_empty() {
  1353. let addrs = hosts.greylist_fetch_address(transports).await;
  1354. hostlist.push(addrs);
  1355. }
  1356. } else if i < white_count {
  1357. if !hosts.whitelist_fetch_address(transports).await.is_empty() {
  1358. let addrs = hosts.whitelist_fetch_address(transports).await;
  1359. hostlist.push(addrs);
  1360. }
  1361. if !hosts.greylist_fetch_address(transports).await.is_empty() {
  1362. let addrs = hosts.greylist_fetch_address(transports).await;
  1363. hostlist.push(addrs);
  1364. }
  1365. } else if !hosts.greylist_fetch_address(transports).await.is_empty() {
  1366. let addrs = hosts.greylist_fetch_address(transports).await;
  1367. hostlist.push(addrs);
  1368. }
  1369. }
  1370. //// Check we're returning the correct addresses.
  1371. anchor_urls.sort();
  1372. white_urls.sort();
  1373. grey_urls.sort();
  1374. hostlist[0].sort();
  1375. hostlist[4].sort();
  1376. hostlist[7].sort();
  1377. assert!(anchor_urls == hostlist[0]);
  1378. assert!(white_urls == hostlist[4]);
  1379. assert!(grey_urls == hostlist[7]);
  1380. })
  1381. }
  1382. }