darkirc.rs 27 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743
  1. /* This file is part of DarkFi (https://dark.fi)
  2. *
  3. * Copyright (C) 2020-2026 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,
  20. io::Cursor,
  21. sync::{Arc, Mutex as SyncMutex, OnceLock, Weak},
  22. time::UNIX_EPOCH,
  23. };
  24. use async_lock::RwLock;
  25. use async_trait::async_trait;
  26. use darkfi::{
  27. event_graph::{
  28. self,
  29. proto::{EventPut, ProtocolEventGraph},
  30. EventGraph, EventGraphConfig, EventGraphPtr,
  31. },
  32. net::{
  33. session::SESSION_DEFAULT,
  34. settings::{MagicBytes, NetworkProfile, Settings as NetSettings},
  35. ChannelPtr, P2p, P2pPtr,
  36. },
  37. system::{sleep, Subscription},
  38. Result as DarkFiResult,
  39. };
  40. use darkfi_serial::{
  41. deserialize_async, serialize, serialize_async, AsyncEncodable, Decodable, Encodable,
  42. SerialDecodable, SerialEncodable,
  43. };
  44. use irc2::{
  45. crypto::saltbox,
  46. irc::{server::MAX_NICK_LEN, IrcChannel, IrcContact},
  47. pad, unpad, Privmsg,
  48. };
  49. use sled_overlay::sled;
  50. use crate::{
  51. error::{Error, Result},
  52. prop::{BatchGuardPtr, PropertyAtomicGuard, PropertyStr, Role},
  53. scene::{MethodCallSub, Pimpl, SceneNode, SceneNodePtr, SceneNodeType, SceneNodeWeak, Slot},
  54. ui::{
  55. chatview::{MessageId, Timestamp},
  56. OnModify,
  57. },
  58. ExecutorPtr,
  59. };
  60. use super::PluginSettings;
  61. const P2P_RETRY_TIME: u64 = 20;
  62. const COOLOFF_SLEEP_TIME: u64 = 20;
  63. const COOLOFF_SYNC_ATTEMPTS: usize = 6;
  64. const SYNC_MIN_PEERS: usize = 2;
  65. const P2P_OUTBOUND_ACTIVE: usize = 6;
  66. const P2P_OUTBOUND_SLEEP: usize = 1;
  67. /// Due to drift between different machine's clocks, if the message timestamp is recent
  68. /// then we will just correct it to the current time so messages appear sequential in the UI.
  69. const RECENT_TIME_DIST: u64 = 25_000;
  70. #[cfg(target_os = "android")]
  71. mod paths {
  72. use crate::android::{get_appdata_path, get_external_storage_path};
  73. use std::path::PathBuf;
  74. pub fn get_evgrdb_path() -> PathBuf {
  75. get_external_storage_path().join("evgr")
  76. }
  77. pub fn get_use_tor_filename() -> PathBuf {
  78. get_external_storage_path().join("use_tor.txt")
  79. }
  80. pub fn nick_filename() -> PathBuf {
  81. get_appdata_path().join("/nick.txt")
  82. }
  83. pub fn p2p_datastore_path() -> PathBuf {
  84. get_appdata_path().join("darkirc_p2p")
  85. }
  86. pub fn hostlist_path() -> PathBuf {
  87. get_appdata_path().join("hostlist.tsv")
  88. }
  89. }
  90. #[cfg(not(target_os = "android"))]
  91. mod paths {
  92. use std::path::PathBuf;
  93. pub fn get_evgrdb_path() -> PathBuf {
  94. dirs::data_local_dir().unwrap().join("darkfi/app/evgr")
  95. }
  96. pub fn get_use_tor_filename() -> PathBuf {
  97. dirs::data_local_dir().unwrap().join("darkfi/app/use_tor.txt")
  98. }
  99. pub fn nick_filename() -> PathBuf {
  100. dirs::cache_dir().unwrap().join("darkfi/app/nick.txt")
  101. }
  102. pub fn p2p_datastore_path() -> PathBuf {
  103. dirs::cache_dir().unwrap().join("darkfi/app/darkirc_p2p")
  104. }
  105. pub fn hostlist_path() -> PathBuf {
  106. dirs::cache_dir().unwrap().join("darkfi/app/hostlist.tsv")
  107. }
  108. }
  109. use paths::*;
  110. macro_rules! t { ($($arg:tt)*) => { trace!(target: "plugin::darkirc", $($arg)*); } }
  111. macro_rules! d { ($($arg:tt)*) => { debug!(target: "plugin::darkirc", $($arg)*); } }
  112. macro_rules! i { ($($arg:tt)*) => { info!(target: "plugin::darkirc", $($arg)*); } }
  113. macro_rules! e { ($($arg:tt)*) => { error!(target: "plugin::darkirc", $($arg)*); } }
  114. macro_rules! w { ($($arg:tt)*) => { warn!(target: "plugin::darkirc", $($arg)*); } }
  115. struct SeenMsg {
  116. id: MessageId,
  117. is_self: bool,
  118. seen_times: usize,
  119. }
  120. struct SeenMessages {
  121. seen: Vec<SeenMsg>,
  122. }
  123. impl SeenMessages {
  124. fn new() -> Self {
  125. Self { seen: vec![] }
  126. }
  127. fn get_status(&self, id: &MessageId) -> Option<&SeenMsg> {
  128. self.seen.iter().find(|s| s.id == *id)
  129. }
  130. fn push(&mut self, id: MessageId, is_self: bool) {
  131. self.seen.push(SeenMsg { id, is_self, seen_times: 0 });
  132. }
  133. }
  134. pub type DarkIrcPtr = Arc<DarkIrc>;
  135. pub struct DarkIrc {
  136. node: SceneNodeWeak,
  137. tasks: OnceLock<Vec<smol::Task<()>>>,
  138. p2p: P2pPtr,
  139. event_graph: EventGraphPtr,
  140. seen_msgs: SyncMutex<SeenMessages>,
  141. nick: PropertyStr,
  142. /// Configured channels
  143. pub channels: RwLock<HashMap<String, IrcChannel>>,
  144. /// Configured contacts
  145. pub contacts: RwLock<HashMap<String, IrcContact>>,
  146. settings: PluginSettings,
  147. }
  148. impl DarkIrc {
  149. pub async fn new(node: SceneNodeWeak, sg_root: SceneNodePtr, ex: ExecutorPtr) -> Result<Pimpl> {
  150. let node_ref = &node.upgrade().unwrap();
  151. let nick = PropertyStr::wrap(node_ref, Role::Internal, "nick", 0).unwrap();
  152. let setting_root = Arc::new(SceneNode::new("setting", SceneNodeType::SettingRoot));
  153. node_ref.link(setting_root.clone());
  154. i!("Starting DarkIRC backend");
  155. let evgr_path = get_evgrdb_path();
  156. let db = match sled::open(&evgr_path) {
  157. Ok(db) => db,
  158. Err(err) => {
  159. e!("Sled database '{}' failed to open: {err}!", evgr_path.display());
  160. return Err(Error::SledDbErr)
  161. }
  162. };
  163. let setting_tree = db.open_tree("settings")?;
  164. let settings = PluginSettings { setting_root, sled_tree: setting_tree };
  165. let mut p2p_settings: NetSettings = Default::default();
  166. p2p_settings.magic_bytes = MagicBytes([251, 229, 199, 181]);
  167. p2p_settings.app_version = semver::Version::parse("0.5.0").unwrap();
  168. p2p_settings.app_name = "darkirc".to_string();
  169. if get_use_tor_filename().exists() {
  170. i!("Setup P2P network [tor]");
  171. let mut tor_profile = NetworkProfile::tor_default();
  172. tor_profile.outbound_connect_timeout = 60;
  173. p2p_settings.profiles.insert("tor".to_string(), tor_profile);
  174. p2p_settings.outbound_peer_discovery_cooloff_time = 60;
  175. p2p_settings.seeds.push(
  176. url::Url::parse(
  177. "tor://g7fxelebievvpr27w7gt24lflptpw3jeeuvafovgliq5utdst6xyruyd.onion:25552",
  178. )
  179. .unwrap(),
  180. );
  181. p2p_settings.seeds.push(
  182. url::Url::parse(
  183. "tor://yvklzjnfmwxhyodhrkpomawjcdvcaushsj6torjz2gyd7e25f3gfunyd.onion:25552",
  184. )
  185. .unwrap(),
  186. );
  187. p2p_settings.active_profiles = vec!["tor".to_string()];
  188. } else {
  189. i!("Setup P2P network [clearnet]");
  190. let mut profile = NetworkProfile::default();
  191. profile.outbound_connect_timeout = 40;
  192. profile.channel_handshake_timeout = 30;
  193. p2p_settings.profiles.insert("tcp+tls".to_string(), profile);
  194. p2p_settings.outbound_connections = 5;
  195. p2p_settings.inbound_connections = 2;
  196. p2p_settings.seeds.push(url::Url::parse("tcp+tls://lilith0.dark.fi:25551").unwrap());
  197. p2p_settings.seeds.push(url::Url::parse("tcp+tls://lilith1.dark.fi:25551").unwrap());
  198. p2p_settings.active_profiles = vec!["tcp+tls".to_string()];
  199. }
  200. p2p_settings.p2p_datastore = p2p_datastore_path().into_os_string().into_string().ok();
  201. p2p_settings.hostlist = hostlist_path().into_os_string().into_string().ok();
  202. settings.add_p2p_settings(&p2p_settings);
  203. settings.load_settings();
  204. settings.update_p2p_settings(&mut p2p_settings);
  205. let p2p = match P2p::new(p2p_settings.clone(), ex.clone()).await {
  206. Ok(p2p) => p2p,
  207. Err(err) => {
  208. e!("Create p2p network failed: {err}!");
  209. return Err(Error::ServiceFailed)
  210. }
  211. };
  212. let event_graph = match EventGraph::new(
  213. p2p.clone(),
  214. db.clone(),
  215. std::path::PathBuf::new(),
  216. false,
  217. EventGraphConfig {
  218. initial_genesis: 1_704_067_200_000,
  219. hours_rotation: 1,
  220. genesis_contents: b"darkirc".to_vec(),
  221. rln_enabled: false,
  222. pregenerated_identity_commitments: vec![],
  223. max_dags: Some(24),
  224. },
  225. ex.clone(),
  226. )
  227. .await
  228. {
  229. Ok(evgr) => evgr,
  230. Err(err) => {
  231. e!("Create event graph failed: {err}!");
  232. return Err(Error::ServiceFailed)
  233. }
  234. };
  235. if let Ok(prev_nick) = std::fs::read_to_string(nick_filename()) {
  236. nick.set(&mut PropertyAtomicGuard::none(), prev_nick);
  237. }
  238. let self_ = Arc::new(Self {
  239. node: node.clone(),
  240. tasks: OnceLock::new(),
  241. p2p,
  242. event_graph,
  243. seen_msgs: SyncMutex::new(SeenMessages::new()),
  244. nick,
  245. channels: RwLock::new(HashMap::new()),
  246. contacts: RwLock::new(HashMap::new()),
  247. settings,
  248. });
  249. self_.clone().start(sg_root, ex).await;
  250. Ok(Pimpl::DarkIrc(self_))
  251. }
  252. async fn dag_sync(self: Arc<Self>, channel_sub: Subscription<DarkFiResult<ChannelPtr>>) {
  253. i!("Starting p2p network");
  254. while let Err(err) = self.p2p.clone().start().await {
  255. // This usually means we cannot listen on the inbound ports
  256. e!("Failed to start p2p network: {err}!");
  257. e!("Usually this means there is another process listening on the same ports.");
  258. e!("Trying again in {P2P_RETRY_TIME} secs");
  259. sleep(P2P_RETRY_TIME).await;
  260. }
  261. i!("Waiting for some P2P connections...");
  262. let mut sync_attempt = 0;
  263. loop {
  264. // Wait for a channel
  265. if let Err(err) = channel_sub.receive().await {
  266. w!("There was an error listening for channels. The service closed unexpectedly with error: {err}");
  267. continue
  268. }
  269. let peers_count = self.p2p.peers_count();
  270. self.notify_connect(peers_count, false).await;
  271. // Wait until we have enough connections
  272. if peers_count < SYNC_MIN_PEERS {
  273. i!("Connected to {peers_count} peers. Waiting for more connections.");
  274. continue
  275. }
  276. sync_attempt += 1;
  277. // Cool off periodically
  278. if sync_attempt > COOLOFF_SYNC_ATTEMPTS {
  279. i!("Wasn't able to sync yet. Cooling off for {COOLOFF_SLEEP_TIME} then will try again.");
  280. sleep(COOLOFF_SLEEP_TIME).await;
  281. sync_attempt = 0;
  282. }
  283. i!("Syncing event DAG (attempt #{sync_attempt})");
  284. // TODO: sync_selected args should be configurable
  285. match self.event_graph.sync_selected(24).await {
  286. Ok(()) => break,
  287. Err(e) => {
  288. // TODO: Maybe at this point we should prune or something?
  289. // TODO: Or maybe just tell the user to delete the DAG from FS.
  290. w!("Failed DAG sync: ({e}). Waiting for more connections before retry.");
  291. }
  292. }
  293. }
  294. let peers_count = self.p2p.peers_count();
  295. self.notify_connect(peers_count, true).await;
  296. // Initial sync finished. Now just notify of connection changes
  297. loop {
  298. // Wait for a channel
  299. if let Err(err) = channel_sub.receive().await {
  300. w!("There was an error listening for channels. The service closed unexpectedly with error: {err}");
  301. continue
  302. }
  303. let peers_count = self.p2p.peers_count();
  304. self.notify_connect(peers_count, true).await;
  305. }
  306. }
  307. async fn notify_connect(&self, peers_count: usize, is_dag_synced: bool) {
  308. let node = self.node.upgrade().unwrap();
  309. node.trigger("connect", serialize(&(peers_count as u32, is_dag_synced))).await.unwrap();
  310. }
  311. async fn relay_events(self: Arc<Self>, ev_sub: Subscription<event_graph::Event>) {
  312. loop {
  313. let ev = ev_sub.receive().await;
  314. // Try to deserialize the `Event`'s content into a `Privmsg`
  315. let privmsg: Privmsg = match deserialize_async(ev.content()).await {
  316. Ok(v) => v,
  317. Err(e) => {
  318. e!("[IRC CLIENT] Failed deserializing incoming Privmsg event: {}", e);
  319. continue
  320. }
  321. };
  322. // TODO: decrypt messages here:
  323. // self.try_decrypt(&mut privmsg, &self.nick.get()).await;
  324. let mut timest = ev.header.timestamp;
  325. let msg_id = msg_id(&privmsg, timest);
  326. t!(
  327. "Relaying ev_id={:?}, ev={ev:?}, msg_id={msg_id}, privmsg={privmsg:?}, timest={timest}",
  328. ev.id(),
  329. );
  330. let is_self = {
  331. let mut is_self = false;
  332. let mut seen = self.seen_msgs.lock().unwrap();
  333. match seen.get_status(&msg_id) {
  334. Some(msg) => {
  335. is_self = msg.is_self;
  336. if !msg.is_self || msg.seen_times > 1 {
  337. w!("Skipping duplicate seen message: {msg_id}");
  338. continue
  339. }
  340. }
  341. None => {
  342. seen.push(msg_id.clone(), false);
  343. }
  344. }
  345. is_self
  346. };
  347. // This is a hack to make messages appear sequentially in the UI
  348. let now_timest = UNIX_EPOCH.elapsed().unwrap().as_millis() as u64;
  349. if !is_self && timest.abs_diff(now_timest) < RECENT_TIME_DIST {
  350. d!("Applied timestamp correction: <{timest}> => <{now_timest}>");
  351. timest = now_timest;
  352. }
  353. // Strip off starting #
  354. let mut channel = privmsg.channel;
  355. if channel.is_empty() {
  356. w!("Received privmsg with empty channel!");
  357. continue
  358. }
  359. if channel.chars().next().unwrap() != '#' {
  360. w!("Skipping encrypted channel: {channel}");
  361. continue
  362. }
  363. channel.remove(0);
  364. // Workaround for the chatview hack. This nick is off limits!
  365. let mut nick = privmsg.nick;
  366. if nick == "NOTICE" {
  367. nick = "noticer".to_string();
  368. }
  369. let mut arg_data = vec![];
  370. channel.encode(&mut arg_data).unwrap();
  371. timest.encode(&mut arg_data).unwrap();
  372. msg_id.encode(&mut arg_data).unwrap();
  373. nick.encode(&mut arg_data).unwrap();
  374. privmsg.msg.encode(&mut arg_data).unwrap();
  375. let node = self.node.upgrade().unwrap();
  376. node.trigger("recv", arg_data).await.unwrap();
  377. }
  378. }
  379. async fn process_send(me: &Weak<Self>, sub: &MethodCallSub) -> bool {
  380. let Ok(method_call) = sub.receive().await else {
  381. d!("Event relayer closed");
  382. return false
  383. };
  384. t!("method called: send({method_call:?})");
  385. assert!(method_call.send_res.is_none());
  386. fn decode_data(data: &[u8]) -> std::io::Result<(Timestamp, String, String)> {
  387. let mut cur = Cursor::new(&data);
  388. let timest = Timestamp::decode(&mut cur).unwrap();
  389. let channel = String::decode(&mut cur)?;
  390. let msg = String::decode(&mut cur)?;
  391. Ok((timest, channel, msg))
  392. }
  393. let Ok((timest, channel, msg)) = decode_data(&method_call.data) else {
  394. e!("send() method invalid arg data");
  395. return true
  396. };
  397. let Some(self_) = me.upgrade() else {
  398. // Should not happen
  399. panic!("self destroyed before send_method_task was stopped!");
  400. };
  401. self_.handle_send(timest, channel, msg).await;
  402. true
  403. }
  404. async fn handle_send(&self, timest: Timestamp, channel: String, msg: String) {
  405. let nick = self.nick.get();
  406. // Send text to channel
  407. d!("Sending privmsg: {timest} {channel}: <{nick}> {msg}");
  408. let msg = Privmsg { version: 0, msg_type: 0, channel, nick, msg };
  409. // TODO: messages should be encrypted here with:
  410. // self.try_encrypt(&mut msg).await;
  411. let evgr = self.event_graph.clone();
  412. let mut event = event_graph::Event::new(serialize_async(&msg).await, &evgr).await.unwrap();
  413. event.header.timestamp = timest;
  414. let msg_id = msg_id(&msg, timest);
  415. // Keep track of our own messages so we don't apply timestamp correction to them
  416. // which messes up the msg id.
  417. {
  418. let mut seen = self.seen_msgs.lock().unwrap();
  419. seen.push(msg_id.clone(), true);
  420. }
  421. let mut arg_data = vec![];
  422. timest.encode_async(&mut arg_data).await.unwrap();
  423. msg_id.encode_async(&mut arg_data).await.unwrap();
  424. msg.nick.encode_async(&mut arg_data).await.unwrap();
  425. msg.msg.encode_async(&mut arg_data).await.unwrap();
  426. // Broadcast the msg
  427. let current_genesis = self.event_graph.current_genesis.read().await;
  428. let dag_name = current_genesis.header.timestamp.to_string();
  429. if let Err(e) = evgr.insert_signal_with_blob(&event, &[], &dag_name).await {
  430. error!(target: "darkirc", "Failed inserting new event to DAG: {}", e);
  431. }
  432. if let Err(e) = self.p2p.broadcast(&EventPut(event, vec![])).await {
  433. error!(target: "darkirc", "Event broadcast was not admitted: {e}");
  434. }
  435. }
  436. async fn apply_settings(self_: Arc<Self>, _: BatchGuardPtr) {
  437. self_.settings.save_settings();
  438. let p2p_settings = self_.p2p.settings();
  439. let mut write_guard = p2p_settings.write().await;
  440. self_.settings.update_p2p_settings(&mut write_guard);
  441. }
  442. async fn process_reconnect(me: &Weak<Self>, sub: &MethodCallSub) -> bool {
  443. let Ok(method_call) = sub.receive().await else {
  444. d!("Reconnect method closed");
  445. return false
  446. };
  447. t!("method called: reconnect({method_call:?})");
  448. let Some(self_) = me.upgrade() else {
  449. e!("DarkIrc destroyed before reconnect completed");
  450. return false
  451. };
  452. i!("Manual P2P reconnection triggered");
  453. self_.p2p.clone().stop().await;
  454. while let Err(err) = self_.p2p.clone().start().await {
  455. e!("Failed to start P2P network: {err}!");
  456. e!("Retrying in {P2P_RETRY_TIME} secs");
  457. sleep(P2P_RETRY_TIME).await;
  458. }
  459. i!("P2P reconnection completed");
  460. true
  461. }
  462. async fn start(self: Arc<Self>, sg_root: SceneNodePtr, ex: ExecutorPtr) {
  463. i!("Registering EventGraph P2P protocol");
  464. let event_graph_ = Arc::clone(&self.event_graph);
  465. let registry = self.p2p.protocol_registry();
  466. registry
  467. .register(SESSION_DEFAULT, move |channel, _| {
  468. let event_graph_ = event_graph_.clone();
  469. async move { ProtocolEventGraph::init(event_graph_, channel).await.unwrap() }
  470. })
  471. .await;
  472. let me = Arc::downgrade(&self);
  473. let node = &self.node.upgrade().unwrap();
  474. let method_sub = node.subscribe_method_call("send").unwrap();
  475. let me2 = me.clone();
  476. let send_method_task =
  477. ex.spawn(async move { while Self::process_send(&me2, &method_sub).await {} });
  478. let reconnect_method_sub = node.subscribe_method_call("reconnect").unwrap();
  479. let me2 = me.clone();
  480. let reconnect_method_task =
  481. ex.spawn(
  482. async move { while Self::process_reconnect(&me2, &reconnect_method_sub).await {} },
  483. );
  484. let mut on_modify = OnModify::new(ex.clone(), self.node.clone(), me.clone());
  485. async fn save_nick(self_: Arc<DarkIrc>, _batch: BatchGuardPtr) {
  486. let _ = std::fs::write(nick_filename(), self_.nick.get());
  487. }
  488. on_modify.when_change(self.nick.prop(), save_nick);
  489. // `apply_settings` is triggered if any setting changes
  490. for setting_node in self.settings.setting_root.get_children().iter() {
  491. on_modify.when_change(
  492. setting_node.get_property("value").clone().unwrap(),
  493. Self::apply_settings,
  494. );
  495. }
  496. let ev_sub = self.event_graph.event_pub.clone().subscribe().await;
  497. let ev_task = ex.spawn(self.clone().relay_events(ev_sub));
  498. // Sync the DAG
  499. let channel_sub = self.p2p.hosts().subscribe_channel().await;
  500. let dag_task = ex.spawn(self.clone().dag_sync(channel_sub));
  501. // Subscribe to window start/stop signals for dynamic outbound connections
  502. let window_node = sg_root.lookup_node("/window").unwrap();
  503. let (start_slot, start_recv) = Slot::new("app_start");
  504. window_node.register("start", start_slot).unwrap();
  505. let p2p = self.p2p.clone();
  506. let start_task = ex.spawn(async move {
  507. while let Ok(_) = start_recv.recv().await {
  508. i!("App started: set outbound connections to {P2P_OUTBOUND_ACTIVE}");
  509. p2p.settings().write().await.outbound_connections = P2P_OUTBOUND_ACTIVE;
  510. p2p.clone().reload().await;
  511. }
  512. });
  513. let (stop_slot, stop_recv) = Slot::new("app_stop");
  514. window_node.register("stop", stop_slot).unwrap();
  515. let p2p = self.p2p.clone();
  516. let stop_task = ex.spawn(async move {
  517. while let Ok(_) = stop_recv.recv().await {
  518. i!("App stopped: set outbound connections to {P2P_OUTBOUND_SLEEP}");
  519. p2p.settings().write().await.outbound_connections = P2P_OUTBOUND_SLEEP;
  520. p2p.clone().reload().await;
  521. }
  522. });
  523. let mut tasks =
  524. vec![send_method_task, reconnect_method_task, ev_task, dag_task, start_task, stop_task];
  525. tasks.append(&mut on_modify.tasks);
  526. self.tasks.set(tasks).unwrap();
  527. }
  528. /// Try encrypting a given `Privmsg` if there is such a channel/contact.
  529. pub async fn try_encrypt(&self, privmsg: &mut Privmsg) {
  530. if let Some((name, channel)) = self.channels.read().await.get_key_value(&privmsg.channel) {
  531. if let Some(saltbox) = &channel.saltbox {
  532. // We will use a dummy channel value of MAX_NICK_LEN,
  533. // since its not used, so all encrypted messages look the same.
  534. privmsg.channel = saltbox::encrypt(saltbox, &[0x00; MAX_NICK_LEN]);
  535. // We will pad the name to MAX_NICK_LEN so they all look the same
  536. privmsg.nick = saltbox::encrypt(saltbox, &pad(&privmsg.nick));
  537. privmsg.msg = saltbox::encrypt(saltbox, privmsg.msg.as_bytes());
  538. d!("Successfully encrypted message for {name}");
  539. return
  540. }
  541. };
  542. if let Some((name, contact)) = self.contacts.read().await.get_key_value(&privmsg.channel) {
  543. // We will use dummy channel and nick values of MAX_NICK_LEN,
  544. // since they are not used, so all encrypted messages look the same.
  545. privmsg.channel = saltbox::encrypt(&contact.saltbox, &[0x00; MAX_NICK_LEN]);
  546. // We will encrypt the dummy nick value using our own self saltbox,
  547. // so we can identify our messages.
  548. privmsg.nick = saltbox::encrypt(&contact.self_saltbox, &[0x00; MAX_NICK_LEN]);
  549. privmsg.msg = saltbox::encrypt(&contact.saltbox, privmsg.msg.as_bytes());
  550. d!("Successfully encrypted message for {name}");
  551. };
  552. }
  553. /// Try decrypting a given potentially encrypted `Privmsg` object.
  554. pub async fn try_decrypt(&self, privmsg: &mut Privmsg, self_nickname: &str) {
  555. // If all fields have base58, then we can consider decrypting.
  556. let channel_ciphertext = match bs58::decode(&privmsg.channel).into_vec() {
  557. Ok(v) => v,
  558. Err(_) => return,
  559. };
  560. let nick_ciphertext = match bs58::decode(&privmsg.nick).into_vec() {
  561. Ok(v) => v,
  562. Err(_) => return,
  563. };
  564. let msg_ciphertext = match bs58::decode(&privmsg.msg).into_vec() {
  565. Ok(v) => v,
  566. Err(_) => return,
  567. };
  568. // Now go through all 3 ciphertexts. We'll use intermediate buffers
  569. // for decryption, if all passes, we will return a modified
  570. // (i.e. decrypted) privmsg, otherwise we return the original.
  571. for (name, channel) in self.channels.read().await.iter() {
  572. let Some(saltbox) = &channel.saltbox else { continue };
  573. if saltbox::try_decrypt(saltbox, &channel_ciphertext).is_none() {
  574. continue
  575. };
  576. let Some(mut nick_dec) = saltbox::try_decrypt(saltbox, &nick_ciphertext) else {
  577. w!("Could not decrypt nick ciphertext for channel: {name}");
  578. continue
  579. };
  580. let Some(msg_dec) = saltbox::try_decrypt(saltbox, &msg_ciphertext) else {
  581. w!("Could not decrypt message ciphertext for channel: {name}");
  582. continue
  583. };
  584. unpad(&mut nick_dec);
  585. privmsg.channel = name.to_string();
  586. privmsg.nick = String::from_utf8_lossy(&nick_dec).into();
  587. privmsg.msg = String::from_utf8_lossy(&msg_dec).into();
  588. d!("Successfully decrypted message for {name}");
  589. return
  590. }
  591. for (name, contact) in self.contacts.read().await.iter() {
  592. if saltbox::try_decrypt(&contact.saltbox, &channel_ciphertext).is_none() {
  593. continue
  594. };
  595. // Since everyone encrypts the dummy nick value with their self saltbox,
  596. // we try to decrypt using our, to identify our messages.
  597. let nick = if saltbox::try_decrypt(&contact.self_saltbox, &nick_ciphertext).is_some() {
  598. String::from(self_nickname)
  599. } else {
  600. name.to_string()
  601. };
  602. let Some(msg_dec) = saltbox::try_decrypt(&contact.saltbox, &msg_ciphertext) else {
  603. w!("Could not decrypt message ciphertext for contact: {name}");
  604. continue
  605. };
  606. privmsg.channel = name.to_string();
  607. privmsg.nick = nick;
  608. privmsg.msg = String::from_utf8_lossy(&msg_dec).into();
  609. d!("Successfully decrypted message from {name}");
  610. return
  611. }
  612. }
  613. }
  614. pub fn msg_id(privmsg: &Privmsg, timest: u64) -> MessageId {
  615. let mut hasher = blake3::Hasher::new();
  616. 0u8.encode(&mut hasher).unwrap();
  617. 0u8.encode(&mut hasher).unwrap();
  618. timest.encode(&mut hasher).unwrap();
  619. privmsg.channel.encode(&mut hasher).unwrap();
  620. privmsg.nick.encode(&mut hasher).unwrap();
  621. privmsg.msg.encode(&mut hasher).unwrap();
  622. MessageId(hasher.finalize().into())
  623. }