eightequal/AstralRinth
1
0
Fork 0
forked from didirus/AstralRinth
Code Pull Requests Activity

Auth retrying, std logs (#879)

Browse Source
This commit is contained in:
Wyatt Verchere
2023-11-17 20:49:32 -08:00
committed by GitHub
parent 01ab507e3a
commit 25662d1402
20 changed files with 379 additions and 114 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
theseus/src/api/hydra/init.rs
View File

@@ -5,7 +5,7 @@ use serde::{Deserialize, Serialize};
use crate::{hydra::MicrosoftError, util::fetch::REQWEST_CLIENT};
use super::MICROSOFT_CLIENT_ID;
use super::{stages::auth_retry, MICROSOFT_CLIENT_ID};
#[derive(Serialize, Deserialize, Debug)]
pub struct DeviceLoginSuccess {
@@ -28,13 +28,13 @@ pub async fn init() -> crate::Result<DeviceLoginSuccess> {
params.insert("scope", "XboxLive.signin offline_access");
// urlencoding::encode("XboxLive.signin offline_access"));
let req = REQWEST_CLIENT.post("https://login.microsoftonline.com/consumers/oauth2/v2.0/devicecode")
.header("Content-Type", "application/x-www-form-urlencoded").form(&params).send().await?;
let resp = auth_retry(|| REQWEST_CLIENT.post("https://login.microsoftonline.com/consumers/oauth2/v2.0/devicecode")
.header("Content-Type", "application/x-www-form-urlencoded").form(&params).send()).await?;
match req.status() {
reqwest::StatusCode::OK => Ok(req.json().await?),
match resp.status() {
reqwest::StatusCode::OK => Ok(resp.json().await?),
_ => {
let microsoft_error = req.json::<MicrosoftError>().await?;
let microsoft_error = resp.json::<MicrosoftError>().await?;
Err(crate::ErrorKind::HydraError(format!(
"Error from Microsoft: {:?}",
microsoft_error.error_description

7
theseus/src/api/hydra/refresh.rs
View File

@@ -8,6 +8,8 @@ use crate::{
util::fetch::REQWEST_CLIENT,
};
use super::stages::auth_retry;
#[derive(Debug, Deserialize)]
pub struct OauthSuccess {
pub token_type: String,
@@ -25,11 +27,14 @@ pub async fn refresh(refresh_token: String) -> crate::Result<OauthSuccess> {
// Poll the URL in a loop until we are successful.
// On an authorization_pending response, wait 5 seconds and try again.
let resp = REQWEST_CLIENT
let resp =
auth_retry(|| {
REQWEST_CLIENT
.post("https://login.microsoftonline.com/consumers/oauth2/v2.0/token")
.header("Content-Type", "application/x-www-form-urlencoded")
.form(&params)
.send()
})
.await?;
match resp.status() {

27
theseus/src/api/hydra/stages/bearer_token.rs
View File

@@ -1,20 +1,25 @@
use serde_json::json;
use super::auth_retry;
const MCSERVICES_AUTH_URL: &str =
"https://api.minecraftservices.com/launcher/login";
#[tracing::instrument]
pub async fn fetch_bearer(token: &str, uhs: &str) -> crate::Result<String> {
let client = reqwest::Client::new();
let body = client
.post(MCSERVICES_AUTH_URL)
.json(&json!({
"xtoken": format!("XBL3.0 x={};{}", uhs, token),
"platform": "PC_LAUNCHER"
}))
.send()
.await?
.text()
.await?;
let body = auth_retry(|| {
let client = reqwest::Client::new();
client
.post(MCSERVICES_AUTH_URL)
.json(&json!({
"xtoken": format!("XBL3.0 x={};{}", uhs, token),
"platform": "PC_LAUNCHER"
}))
.send()
})
.await?
.text()
.await?;
serde_json::from_str::<serde_json::Value>(&body)?
.get("access_token")

30
theseus/src/api/hydra/stages/mod.rs
View File

@@ -1,7 +1,37 @@
//! MSA authentication stages
use futures::Future;
use reqwest::Response;
const RETRY_COUNT: usize = 2; // Does command 3 times
const RETRY_WAIT: std::time::Duration = std::time::Duration::from_secs(2);
pub mod bearer_token;
pub mod player_info;
pub mod poll_response;
pub mod xbl_signin;
pub mod xsts_token;
#[tracing::instrument(skip(reqwest_request))]
pub async fn auth_retry<F>(
reqwest_request: impl Fn() -> F,
) -> crate::Result<reqwest::Response>
where
F: Future<Output = Result<Response, reqwest::Error>>,
{
let mut resp = reqwest_request().await?;
for i in 0..RETRY_COUNT {
if resp.status().is_success() {
break;
}
tracing::debug!(
"Request failed with status code {}, retrying...",
resp.status()
);
if i < RETRY_COUNT - 1 {
tokio::time::sleep(RETRY_WAIT).await;
}
resp = reqwest_request().await?;
}
Ok(resp)
}

23
theseus/src/api/hydra/stages/player_info.rs
View File

@@ -1,6 +1,10 @@
//! Fetch player info for display
use serde::Deserialize;
use crate::util::fetch::REQWEST_CLIENT;
use super::auth_retry;
const PROFILE_URL: &str = "https://api.minecraftservices.com/minecraft/profile";
#[derive(Deserialize)]
@@ -18,16 +22,17 @@ impl Default for PlayerInfo {
}
}
#[tracing::instrument]
pub async fn fetch_info(token: &str) -> crate::Result<PlayerInfo> {
let client = reqwest::Client::new();
let resp = client
.get(PROFILE_URL)
.header(reqwest::header::AUTHORIZATION, format!("Bearer {token}"))
.send()
.await?
.error_for_status()?
.json()
.await?;
let response = auth_retry(|| {
REQWEST_CLIENT
.get(PROFILE_URL)
.header(reqwest::header::AUTHORIZATION, format!("Bearer {token}"))
.send()
})
.await?;
let resp = response.error_for_status()?.json().await?;
Ok(resp)
}

9
theseus/src/api/hydra/stages/poll_response.rs
View File

@@ -8,6 +8,8 @@ use crate::{
util::fetch::REQWEST_CLIENT,
};
use super::auth_retry;
#[derive(Debug, Deserialize)]
pub struct OauthSuccess {
pub token_type: String,
@@ -17,6 +19,7 @@ pub struct OauthSuccess {
pub refresh_token: String,
}
#[tracing::instrument]
pub async fn poll_response(device_code: String) -> crate::Result<OauthSuccess> {
let mut params = HashMap::new();
params.insert("grant_type", "urn:ietf:params:oauth:grant-type:device_code");
@@ -26,14 +29,16 @@ pub async fn poll_response(device_code: String) -> crate::Result<OauthSuccess> {
// Poll the URL in a loop until we are successful.
// On an authorization_pending response, wait 5 seconds and try again.
loop {
let resp = REQWEST_CLIENT
let resp = auth_retry(|| {
REQWEST_CLIENT
.post(
"https://login.microsoftonline.com/consumers/oauth2/v2.0/token",
)
.header("Content-Type", "application/x-www-form-urlencoded")
.form(&params)
.send()
.await?;
})
.await?;
match resp.status() {
StatusCode::OK => {

41
theseus/src/api/hydra/stages/xbl_signin.rs
View File

@@ -1,5 +1,9 @@
use serde_json::json;
use crate::util::fetch::REQWEST_CLIENT;
use super::auth_retry;
const XBL_AUTH_URL: &str = "https://user.auth.xboxlive.com/user/authenticate";
// Deserialization
@@ -9,25 +13,26 @@ pub struct XBLLogin {
}
// Impl
#[tracing::instrument]
pub async fn login_xbl(token: &str) -> crate::Result<XBLLogin> {
let client = reqwest::Client::new();
let body = client
.post(XBL_AUTH_URL)
.header(reqwest::header::ACCEPT, "application/json")
.header("x-xbl-contract-version", "1")
.json(&json!({
"Properties": {
"AuthMethod": "RPS",
"SiteName": "user.auth.xboxlive.com",
"RpsTicket": format!("d={token}")
},
"RelyingParty": "http://auth.xboxlive.com",
"TokenType": "JWT"
}))
.send()
.await?
.text()
.await?;
let response = auth_retry(|| {
REQWEST_CLIENT
.post(XBL_AUTH_URL)
.header(reqwest::header::ACCEPT, "application/json")
.header("x-xbl-contract-version", "1")
.json(&json!({
"Properties": {
"AuthMethod": "RPS",
"SiteName": "user.auth.xboxlive.com",
"RpsTicket": format!("d={token}")
},
"RelyingParty": "http://auth.xboxlive.com",
"TokenType": "JWT"
}))
.send()
})
.await?;
let body = response.text().await?;
let json = serde_json::from_str::<serde_json::Value>(&body)?;
let token = Some(&json)

38
theseus/src/api/hydra/stages/xsts_token.rs
View File

@@ -1,5 +1,9 @@
use serde_json::json;
use crate::util::fetch::REQWEST_CLIENT;
use super::auth_retry;
const XSTS_AUTH_URL: &str = "https://xsts.auth.xboxlive.com/xsts/authorize";
pub enum XSTSResponse {
@@ -7,23 +11,25 @@ pub enum XSTSResponse {
Success { token: String },
}
#[tracing::instrument]
pub async fn fetch_token(token: &str) -> crate::Result<XSTSResponse> {
let client = reqwest::Client::new();
let resp = client
.post(XSTS_AUTH_URL)
.header(reqwest::header::ACCEPT, "application/json")
.json(&json!({
"Properties": {
"SandboxId": "RETAIL",
"UserTokens": [
token
]
},
"RelyingParty": "rp://api.minecraftservices.com/",
"TokenType": "JWT"
}))
.send()
.await?;
let resp = auth_retry(|| {
REQWEST_CLIENT
.post(XSTS_AUTH_URL)
.header(reqwest::header::ACCEPT, "application/json")
.json(&json!({
"Properties": {
"SandboxId": "RETAIL",
"UserTokens": [
token
]
},
"RelyingParty": "rp://api.minecraftservices.com/",
"TokenType": "JWT"
}))
.send()
})
.await?;
let status = resp.status();
let body = resp.text().await?;

35
theseus/src/api/logs.rs
View File

@@ -1,7 +1,7 @@
use std::io::{Read, SeekFrom};
use crate::{
prelude::Credentials,
prelude::{Credentials, DirectoryInfo},
util::io::{self, IOError},
{state::ProfilePathId, State},
};
@@ -74,7 +74,6 @@ pub async fn get_logs(
profile_path: ProfilePathId,
clear_contents: Option<bool>,
) -> crate::Result<Vec<Logs>> {
let state = State::get().await?;
let profile_path =
if let Some(p) = crate::profile::get(&profile_path, None).await? {
p.profile_id()
@@ -85,7 +84,7 @@ pub async fn get_logs(
.into());
};
let logs_folder = state.directories.profile_logs_dir(&profile_path).await?;
let logs_folder = DirectoryInfo::profile_logs_dir(&profile_path).await?;
let mut logs = Vec::new();
if logs_folder.exists() {
for entry in std::fs::read_dir(&logs_folder)
@@ -138,8 +137,7 @@ pub async fn get_output_by_filename(
file_name: &str,
) -> crate::Result<CensoredString> {
let state = State::get().await?;
let logs_folder =
state.directories.profile_logs_dir(profile_subpath).await?;
let logs_folder = DirectoryInfo::profile_logs_dir(profile_subpath).await?;
let path = logs_folder.join(file_name);
let credentials: Vec<Credentials> =
@@ -201,8 +199,7 @@ pub async fn delete_logs(profile_path: ProfilePathId) -> crate::Result<()> {
.into());
};
let state = State::get().await?;
let logs_folder = state.directories.profile_logs_dir(&profile_path).await?;
let logs_folder = DirectoryInfo::profile_logs_dir(&profile_path).await?;
for entry in std::fs::read_dir(&logs_folder)
.map_err(|e| IOError::with_path(e, &logs_folder))?
{
@@ -230,8 +227,7 @@ pub async fn delete_logs_by_filename(
.into());
};
let state = State::get().await?;
let logs_folder = state.directories.profile_logs_dir(&profile_path).await?;
let logs_folder = DirectoryInfo::profile_logs_dir(&profile_path).await?;
let path = logs_folder.join(filename);
io::remove_dir_all(&path).await?;
Ok(())
@@ -240,6 +236,23 @@ pub async fn delete_logs_by_filename(
#[tracing::instrument]
pub async fn get_latest_log_cursor(
profile_path: ProfilePathId,
cursor: u64, // 0 to start at beginning of file
) -> crate::Result<LatestLogCursor> {
get_generic_live_log_cursor(profile_path, "latest.log", cursor).await
}
#[tracing::instrument]
pub async fn get_std_log_cursor(
profile_path: ProfilePathId,
cursor: u64, // 0 to start at beginning of file
) -> crate::Result<LatestLogCursor> {
get_generic_live_log_cursor(profile_path, "latest_stdout.log", cursor).await
}
#[tracing::instrument]
pub async fn get_generic_live_log_cursor(
profile_path: ProfilePathId,
log_file_name: &str,
mut cursor: u64, // 0 to start at beginning of file
) -> crate::Result<LatestLogCursor> {
let profile_path =
@@ -253,8 +266,8 @@ pub async fn get_latest_log_cursor(
};
let state = State::get().await?;
let logs_folder = state.directories.profile_logs_dir(&profile_path).await?;
let path = logs_folder.join("latest.log");
let logs_folder = DirectoryInfo::profile_logs_dir(&profile_path).await?;
let path = logs_folder.join(log_file_name);
if !path.exists() {
// Allow silent failure if latest.log doesn't exist (as the instance may have been launched, but not yet created the file)
return Ok(LatestLogCursor {

4
theseus/src/launcher/mod.rs
View File

@@ -512,8 +512,8 @@ pub async fn launch_minecraft(
.collect::<Vec<_>>(),
)
.current_dir(instance_path.clone())
.stdout(Stdio::null())
.stderr(Stdio::null());
.stdout(Stdio::piped())
.stderr(Stdio::piped());
// CARGO-set DYLD_LIBRARY_PATH breaks Minecraft on macOS during testing on playground
#[cfg(target_os = "macos")]

164
theseus/src/state/children.rs
View File

@@ -1,12 +1,21 @@
use super::DirectoryInfo;
use super::{Profile, ProfilePathId};
use chrono::{DateTime, Utc};
use serde::Deserialize;
use serde::Serialize;
use std::path::Path;
use std::{collections::HashMap, sync::Arc};
use sysinfo::PidExt;
use tokio::fs::File;
use tokio::io::AsyncBufReadExt;
use tokio::io::AsyncWriteExt;
use tokio::io::BufReader;
use tokio::process::Child;
use tokio::process::ChildStderr;
use tokio::process::ChildStdout;
use tokio::process::Command;
use tokio::sync::RwLock;
use tracing::error;
use crate::event::emit::emit_process;
use crate::event::ProcessPayloadType;
@@ -192,6 +201,7 @@ impl ChildType {
pub struct MinecraftChild {
pub uuid: Uuid,
pub profile_relative_path: ProfilePathId,
pub output: Option<SharedOutput>,
pub manager: Option<JoinHandle<crate::Result<i32>>>, // None when future has completed and been handled
pub current_child: Arc<RwLock<ChildType>>,
pub last_updated_playtime: DateTime<Utc>, // The last time we updated the playtime for the associated profile
@@ -271,7 +281,43 @@ impl Children {
censor_strings: HashMap<String, String>,
) -> crate::Result<Arc<RwLock<MinecraftChild>>> {
// Takes the first element of the commands vector and spawns it
let child = mc_command.spawn().map_err(IOError::from)?;
let mut child = mc_command.spawn().map_err(IOError::from)?;
// Create std watcher threads for stdout and stderr
let log_path = DirectoryInfo::profile_logs_dir(&profile_relative_path)
.await?
.join("latest_stdout.log");
let shared_output =
SharedOutput::build(&log_path, censor_strings).await?;
if let Some(child_stdout) = child.stdout.take() {
let stdout_clone = shared_output.clone();
tokio::spawn(async move {
if let Err(e) = stdout_clone.read_stdout(child_stdout).await {
error!("Stdout process died with error: {}", e);
let _ = stdout_clone
.push_line(format!(
"Stdout process died with error: {}",
e
))
.await;
}
});
}
if let Some(child_stderr) = child.stderr.take() {
let stderr_clone = shared_output.clone();
tokio::spawn(async move {
if let Err(e) = stderr_clone.read_stderr(child_stderr).await {
error!("Stderr process died with error: {}", e);
let _ = stderr_clone
.push_line(format!(
"Stderr process died with error: {}",
e
))
.await;
}
});
}
let child = ChildType::TokioChild(child);
// Slots child into manager
@@ -312,6 +358,7 @@ impl Children {
let mchild = MinecraftChild {
uuid,
profile_relative_path,
output: Some(shared_output),
current_child,
manager,
last_updated_playtime,
@@ -402,6 +449,7 @@ impl Children {
let mchild = MinecraftChild {
uuid: cached_process.uuid,
profile_relative_path: cached_process.profile_relative_path,
output: None, // No output for cached/rescued processes
current_child,
manager,
last_updated_playtime,
@@ -710,3 +758,117 @@ impl Default for Children {
Self::new()
}
}
// SharedOutput, a wrapper around a String that can be read from and written to concurrently
// Designed to be used with ChildStdout and ChildStderr in a tokio thread to have a simple String storage for the output of a child process
#[derive(Debug, Clone)]
pub struct SharedOutput {
log_file: Arc<RwLock<File>>,
censor_strings: HashMap<String, String>,
}
impl SharedOutput {
#[tracing::instrument(skip(censor_strings))]
async fn build(
log_file_path: &Path,
censor_strings: HashMap<String, String>,
) -> crate::Result<Self> {
// create log_file_path parent if it doesn't exist
let parent_folder = log_file_path.parent().ok_or_else(|| {
crate::ErrorKind::LauncherError(format!(
"Could not get parent folder of {:?}",
log_file_path
))
})?;
tokio::fs::create_dir_all(parent_folder)
.await
.map_err(|e| IOError::with_path(e, parent_folder))?;
Ok(SharedOutput {
log_file: Arc::new(RwLock::new(
File::create(log_file_path)
.await
.map_err(|e| IOError::with_path(e, log_file_path))?,
)),
censor_strings,
})
}
async fn read_stdout(
&self,
child_stdout: ChildStdout,
) -> crate::Result<()> {
let mut buf_reader = BufReader::new(child_stdout);
let mut buf = Vec::new();
while buf_reader
.read_until(b'\n', &mut buf)
.await
.map_err(IOError::from)?
> 0
{
let line = String::from_utf8_lossy(&buf).into_owned();
let val_line = self.censor_log(line.clone());
{
let mut log_file = self.log_file.write().await;
log_file
.write_all(val_line.as_bytes())
.await
.map_err(IOError::from)?;
}
buf.clear();
}
Ok(())
}
async fn read_stderr(
&self,
child_stderr: ChildStderr,
) -> crate::Result<()> {
let mut buf_reader = BufReader::new(child_stderr);
let mut buf = Vec::new();
// TODO: these can be asbtracted into noe function
while buf_reader
.read_until(b'\n', &mut buf)
.await
.map_err(IOError::from)?
> 0
{
let line = String::from_utf8_lossy(&buf).into_owned();
let val_line = self.censor_log(line.clone());
{
let mut log_file = self.log_file.write().await;
log_file
.write_all(val_line.as_bytes())
.await
.map_err(IOError::from)?;
}
buf.clear();
}
Ok(())
}
async fn push_line(&self, line: String) -> crate::Result<()> {
let val_line = self.censor_log(line.clone());
{
let mut log_file = self.log_file.write().await;
log_file
.write_all(val_line.as_bytes())
.await
.map_err(IOError::from)?;
}
Ok(())
}
fn censor_log(&self, mut val: String) -> String {
for (find, replace) in &self.censor_strings {
val = val.replace(find, replace);
}
val
}
}

1
theseus/src/state/dirs.rs
View File

@@ -159,7 +159,6 @@ impl DirectoryInfo {
/// Gets the logs dir for a given profile
#[inline]
pub async fn profile_logs_dir(
&self,
profile_id: &ProfilePathId,
) -> crate::Result<PathBuf> {
Ok(profile_id.get_full_path().await?.join("logs"))