From bb2d33bef08f7015acb88ebf49e91c495e7dff15 Mon Sep 17 00:00:00 2001 From: LynithDev <61880709+LynithDev@users.noreply.github.com> Date: Wed, 2 Apr 2025 17:43:59 +0200 Subject: [PATCH] feat: Discord RPC + HTTP and Crypto utils --- packages/core/examples/discord.rs | 25 ++++ packages/core/src/constants.rs | 6 +- packages/core/src/error.rs | 9 ++ packages/core/src/store/discord.rs | 95 ++++++++++++++ packages/core/src/store/mod.rs | 1 + packages/core/src/store/settings.rs | 6 +- packages/core/src/store/state.rs | 27 +++- packages/core/src/utils/crypto.rs | 46 +++++++ packages/core/src/utils/http.rs | 191 ++++++++++++++++++++++++++++ packages/core/src/utils/mod.rs | 2 + 10 files changed, 397 insertions(+), 11 deletions(-) create mode 100644 packages/core/examples/discord.rs create mode 100644 packages/core/src/store/discord.rs create mode 100644 packages/core/src/utils/crypto.rs create mode 100644 packages/core/src/utils/http.rs diff --git a/packages/core/examples/discord.rs b/packages/core/examples/discord.rs new file mode 100644 index 00000000..896875bc --- /dev/null +++ b/packages/core/examples/discord.rs @@ -0,0 +1,25 @@ +use onelauncher_core::{api::proxy::proxy_empty::ProxyEmpty, error::LauncherResult, initialize_core, store::State}; + +#[tokio::main] +async fn main() -> LauncherResult<()> { + initialize_core(ProxyEmpty::new()).await?; + + let state = State::get().await?; + let Some(rpc) = &state.rpc else { + println!("RPC is not initialized"); + return Ok(()); + }; + + rpc.set_message("Hello, Discord!").await; + println!("Set Discord message"); + + tokio::time::sleep(tokio::time::Duration::from_secs(5)).await; + rpc.set_message("This user is testing Discord RPC!").await; + println!("Set Discord message"); + + tokio::time::sleep(tokio::time::Duration::from_secs(5)).await; + rpc.clear_activity().await; + println!("Cleared Discord message"); + + Ok(()) +} \ No newline at end of file diff --git a/packages/core/src/constants.rs b/packages/core/src/constants.rs index 20b17852..ea70bc65 100644 --- a/packages/core/src/constants.rs +++ b/packages/core/src/constants.rs @@ -63,16 +63,12 @@ pub const SETTINGS_FILE: &str = "settings.json"; /// /// [`MinecraftAuth`]: crate::store::MinecraftAuth pub const AUTH_FILE: &str = "authentication.json"; -/// The public `processor.json` file used to store the global [`Processor`] state. -/// -/// [`Processor`]: crate::store::Processor -pub const PROCESSOR_FILE: &str = "processor.json"; /// The current [`Settings`] format version, bumped for breaking changes. /// If updated, a config file migration logic **NEEDS** to be implemented. /// /// [`Settings`]: crate::store::Settings -pub const CURRENT_SETTINGS_FORMAT_VERSION: u32 = 1; +pub const CURRENT_SETTINGS_FORMAT_VERSION: u32 = 2; // =========== Hacky Mojang-spec OS constants =========== #[cfg(target_os = "windows")] diff --git a/packages/core/src/error.rs b/packages/core/src/error.rs index 0871383e..3770e4fa 100644 --- a/packages/core/src/error.rs +++ b/packages/core/src/error.rs @@ -8,6 +8,11 @@ pub enum LauncherError { IngressError(#[from] crate::store::ingress::IngressError), #[error(transparent)] JavaError(#[from] crate::api::java::JavaError), + #[error(transparent)] + CryptoError(#[from] crate::utils::crypto::CryptoError), + + #[error("failed to connect to discord rpc socket")] + DiscordError, #[error(transparent)] SerdeError(#[from] serde_json::Error), @@ -15,6 +20,10 @@ pub enum LauncherError { AnyhowError(#[from] anyhow::Error), #[error(transparent)] DbError(#[from] sea_orm::DbErr), + #[error(transparent)] + ReqwestError(#[from] reqwest::Error), + #[error("couldn't acquire semaphore: {0}")] + SemaphoreError(#[from] tokio::sync::AcquireError), #[cfg(feature = "tauri")] #[error(transparent)] diff --git a/packages/core/src/store/discord.rs b/packages/core/src/store/discord.rs new file mode 100644 index 00000000..1bc71e09 --- /dev/null +++ b/packages/core/src/store/discord.rs @@ -0,0 +1,95 @@ +use std::sync::{atomic::AtomicBool, Arc}; + +use discord_rich_presence::{activity::{Activity, Assets, Button}, DiscordIpc, DiscordIpcClient}; +use tokio::sync::RwLock; + +use crate::{constants, error::{LauncherError, LauncherResult}}; + +pub struct DiscordRPC { + client: Arc>, + connected: Arc, +} + +impl DiscordRPC { + pub fn initialize() -> LauncherResult { + let mut client = DiscordIpcClient::new(constants::DISCORD_RPC_CLIENT_ID) + .map_err(|_| LauncherError::DiscordError)?; + + let connected = client.connect().is_ok(); + + Ok(Self { + client: Arc::new(RwLock::new(client)), + connected: Arc::new(AtomicBool::new(connected)), + }) + } + + pub async fn set_message(&self, msg: &str) -> bool { + self.set_activity( + Activity::new() + .state(msg) + .buttons(vec![Button::new("Website", "https://polyfrost.org/")]) + .assets( + Assets::new() + .large_image("onelauncher_logo_512") + .large_text("OneLauncher"), + ) + ).await + } + + pub async fn set_activity(&self, activity: Activity<'_>) -> bool { + if !self.is_connected() { + return false; + } + + let mut client = self.client.write().await; + + if client.set_activity(activity).is_err() { + return false; + } + + true + } + + pub async fn clear_activity(&self) -> bool { + if !self.is_connected() { + return false; + } + + let mut client = self.client.write().await; + + if client.clear_activity().is_err() { + return false; + } + + true + } + + pub async fn reconnect(&self) -> bool { + let mut client = self.client.write().await; + + let connected = client.reconnect().is_ok(); + + self.set_connected(connected); + connected + } + + pub async fn close(&self) -> bool { + let mut client = self.client.write().await; + + if client.close().is_err() { + return false; + } + + self.set_connected(false); + true + } + + #[must_use] + pub fn is_connected(&self) -> bool { + self.connected.load(std::sync::atomic::Ordering::Relaxed) + } + + fn set_connected(&self, connected: bool) { + self.connected.store(connected, std::sync::atomic::Ordering::Relaxed); + } +} \ No newline at end of file diff --git a/packages/core/src/store/mod.rs b/packages/core/src/store/mod.rs index 167e7de7..4023535d 100644 --- a/packages/core/src/store/mod.rs +++ b/packages/core/src/store/mod.rs @@ -1,4 +1,5 @@ pub(super) mod db; +pub mod discord; pub mod ingress; pub mod proxy; pub mod java; diff --git a/packages/core/src/store/settings.rs b/packages/core/src/store/settings.rs index 52bb0ca8..f45157d3 100644 --- a/packages/core/src/store/settings.rs +++ b/packages/core/src/store/settings.rs @@ -1,7 +1,7 @@ use onelauncher_entity::setting_profiles; use serde::{Deserialize, Serialize}; -use crate::{send_warning, utils::io, LauncherResult}; +use crate::{constants, send_warning, utils::io, LauncherResult}; use super::Dirs; @@ -13,6 +13,7 @@ pub struct Settings { pub allow_parallel_running_clusters: bool, pub enable_gamemode: bool, pub discord_enabled: bool, + pub max_concurrent_requests: usize, pub settings_version: u32, } @@ -23,7 +24,8 @@ impl Default for Settings { allow_parallel_running_clusters: false, discord_enabled: false, enable_gamemode: false, - settings_version: 2, + max_concurrent_requests: 25, + settings_version: constants::CURRENT_SETTINGS_FORMAT_VERSION, } } } diff --git a/packages/core/src/store/state.rs b/packages/core/src/store/state.rs index 2ec72bf7..2978316b 100644 --- a/packages/core/src/store/state.rs +++ b/packages/core/src/store/state.rs @@ -3,17 +3,21 @@ use std::sync::Arc; use sea_orm::DatabaseConnection; use tokio::sync::{OnceCell, RwLock}; -use crate::LauncherResult; +use crate::{utils::http::{self, FetchSemaphore}, LauncherResult}; -use super::{ingress::IngressProcessor, Settings}; +use super::{discord::DiscordRPC, ingress::IngressProcessor, Settings}; /// The static [`OnceCell>`] for storing the global runtime launcher state. static LAUNCHER_STATE: OnceCell> = OnceCell::const_new(); pub struct State { pub ingress_processor: IngressProcessor, - pub settings: Settings, + pub settings: RwLock, pub db: DatabaseConnection, + pub rpc: Option, + + pub client: Arc, + pub(crate) fetch_semaphore: FetchSemaphore, } impl State { @@ -46,11 +50,26 @@ impl State { let ingress_processor = IngressProcessor::new(); let settings = Settings::new().await; let db = super::db::create_pool().await?; + + let rpc = match DiscordRPC::initialize() { + Ok(rpc) => Some(rpc), + Err(err) => { + tracing::warn!("{}", err); + None + }, + }; + + let reqwest_client = http::create_client()?; + let fetch_semaphore = FetchSemaphore::new(settings.max_concurrent_requests); Ok(RwLock::new(Self { ingress_processor, - settings, + settings: RwLock::new(settings), db, + rpc, + + client: Arc::new(reqwest_client), + fetch_semaphore, })) } } \ No newline at end of file diff --git a/packages/core/src/utils/crypto.rs b/packages/core/src/utils/crypto.rs new file mode 100644 index 00000000..1de79b5f --- /dev/null +++ b/packages/core/src/utils/crypto.rs @@ -0,0 +1,46 @@ +use std::{fmt::Display, str::FromStr}; + +#[derive(Debug, thiserror::Error)] +pub enum CryptoError { + #[error("expected {algorithm} hash {expected}, but got {actual}")] + InvalidHash { + algorithm: HashAlgorithm, + expected: String, + actual: String, + }, + #[error("invalid hash algorithm")] + InvalidAlgorithm +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum HashAlgorithm { + Sha1, +} + +impl HashAlgorithm { + #[must_use] + pub fn hash(&self, data: &[u8]) -> String { + match self { + Self::Sha1 => sha1_smol::Sha1::from(data).digest().to_string(), + } + } +} + +impl Display for HashAlgorithm { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + Self::Sha1 => write!(f, "sha1"), + } + } +} + +impl FromStr for HashAlgorithm { + type Err = CryptoError; + + fn from_str(s: &str) -> Result { + match s.to_lowercase().as_str() { + "sha1" => Ok(Self::Sha1), + _ => Err(CryptoError::InvalidAlgorithm), + } + } +} \ No newline at end of file diff --git a/packages/core/src/utils/http.rs b/packages/core/src/utils/http.rs new file mode 100644 index 00000000..4e2ce9ce --- /dev/null +++ b/packages/core/src/utils/http.rs @@ -0,0 +1,191 @@ +#![allow(clippy::implicit_hasher)] + +use std::{collections::HashMap, path::Path, sync::Arc}; + +use bytes::Bytes; +use reqwest::Method; +use serde::Deserialize; +use tokio::sync::Semaphore; +use tokio_stream::StreamExt; + +use crate::{api::ingress::send_ingress, constants, error::LauncherResult, store::{ingress::IngressId, State}}; + +use super::{crypto::{CryptoError, HashAlgorithm}, io}; + +#[derive(Debug)] +pub struct FetchSemaphore(pub Arc); + +impl FetchSemaphore { + #[must_use] + pub fn new(limit: usize) -> Self { + Self(Arc::new(Semaphore::new(limit))) + } +} + +pub fn create_client() -> LauncherResult { + Ok(reqwest::ClientBuilder::new() + .tcp_keepalive(Some(std::time::Duration::from_secs(15))) + .use_rustls_tls() + .default_headers({ + let mut headers = reqwest::header::HeaderMap::new(); + let header = reqwest::header::HeaderValue::from_str(&format!( + "{}/{} (https://polyfrost.org)", + constants::NAME, + constants::VERSION, + )) + .expect("failed to build reqwest headers!"); + headers.insert(reqwest::header::USER_AGENT, header); + headers + }) + .build()?) +} + +#[tracing::instrument(skip(body, headers, ingress))] +pub async fn fetch_advanced( + method: Method, + url: &str, + body: Option, + hash: Option<(HashAlgorithm, &str)>, + headers: Option>, + ingress: Option<(&IngressId, f64)>, +) -> LauncherResult { + let state = State::get().await?; + let _permit = state.fetch_semaphore.0.acquire().await?; + let client = &state.client; + + for attempt in 0..constants::FETCH_ATTEMPTS { + let mut req = client.request(method.clone(), url); + + if let Some(body) = body.clone() { + req = req.json(&body); + } + + if let Some(headers) = &headers { + for (key, value) in headers { + req = req.header(*key, *value); + } + } + + let res = req.send().await; + + match res { + Err(_) if attempt <= 3 => {}, + Err(err) => return Err(err.into()), + Ok(res) => { + let bytes = if let Some((ingress_id, total)) = ingress { + let length = res.content_length(); + if let Some(total_size) = length { + let mut stream = res.bytes_stream(); + let mut bytes = Vec::new(); + + while let Some(item) = stream.next().await { + let chunk = item.or(Err(anyhow::anyhow!("no value for fetch bytes")))?; + bytes.append(&mut chunk.to_vec()); + + send_ingress( + ingress_id, + (chunk.len() as f64 / total_size as f64) * total, + ) + .await?; + } + + Ok(Bytes::from(bytes)) + } else { + res.bytes().await + } + } else { + res.bytes().await + }; + + if let Ok(bytes) = bytes { + if let Some((ref algorithm, expected_hash)) = hash { + let expected_hash = expected_hash.to_string(); + let calculated_hash = algorithm.hash(&bytes); + + if *calculated_hash != expected_hash { + if attempt <= 3 { + continue; + } + + return Err(CryptoError::InvalidHash { + algorithm: algorithm.clone(), + expected: expected_hash, + actual: calculated_hash + }.into()); + } + } + + tracing::debug!("finished downloading {url}"); + return Ok(bytes); + } else if let Err(err) = bytes { + return Err(err.into()); + } + }, + } + } + + unreachable!() +} + +pub async fn fetch(method: Method, url: &str) -> LauncherResult { + fetch_advanced(method, url, None, None, None, None).await +} + +pub async fn fetch_json Deserialize<'de>>( + method: Method, + url: &str, + body: Option, + headers: Option>, +) -> LauncherResult { + fetch_json_advanced(method, url, body, None, headers, None).await +} + +pub async fn fetch_json_advanced Deserialize<'de>>( + method: Method, + url: &str, + body: Option, + hash: Option<(HashAlgorithm, &str)>, + headers: Option>, + ingress: Option<(&IngressId, f64)>, +) -> LauncherResult { + let bytes = fetch_advanced(method, url, body, hash, headers, ingress).await?; + Ok(serde_json::from_slice(&bytes)?) +} + +pub async fn download_advanced( + method: Method, + url: &str, + path: impl AsRef, + body: Option, + hash: Option<(HashAlgorithm, &str)>, + headers: Option>, + ingress: Option<(&IngressId, f64)>, +) -> LauncherResult<()> { + let bytes = fetch_advanced(method, url, body, hash, headers, ingress.map(|(id, total)| (id, total / 2.0))).await?; + let path = path.as_ref(); + + if let Some((id, total)) = ingress { + send_ingress(id, total / 4.0).await?; + } + + if let Some(parent) = path.parent() { + io::create_dir_all(parent).await?; + } + + io::write(path, &bytes).await?; + + if let Some((id, total)) = ingress { + send_ingress(id, total / 4.0).await?; + } + + Ok(()) +} + +pub async fn download( + method: Method, + url: &str, + path: impl AsRef, + hash: Option<(HashAlgorithm, &str)> +) -> LauncherResult<()> { + download_advanced(method, url, path, None, hash, None, None).await +} diff --git a/packages/core/src/utils/mod.rs b/packages/core/src/utils/mod.rs index 678b90ea..46bbf137 100644 --- a/packages/core/src/utils/mod.rs +++ b/packages/core/src/utils/mod.rs @@ -1 +1,3 @@ +pub mod crypto; +pub mod http; pub mod io; \ No newline at end of file