public/galaxy
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

More progress, still a long way to go

Browse Source
This commit is contained in:
Ryan Ward
2026-05-07 14:37:26 -05:00
parent a41cbd8cc7
commit f8f2c6bff6
25 changed files with 755 additions and 5387 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/src/server/server_api.rs
-6
View File
@@ -1323,12 +1323,6 @@ impl ServerApiProvider {
ctx.dispatch_global_action("app:log_out", ());
}
ServerApiEvent::NeedsReauth => {
// AuthManager depends on a reference to ServerApi, so ServerApi can't easily
// hold a ref to AuthManager. To get around this, we emit an event on ServerApi
// and handle calling the AuthManager here instead.
AuthManager::handle(ctx).update(ctx, |auth_manager, ctx| {
auth_manager.set_needs_reauth(true, ctx);
});
}
// Re-emit the event for subscribers.
// TODO: we probably want a different type for the event emitted to subscribers
app/src/server/server_api/auth.rs
+52 -457
View File
@@ -1,99 +1,45 @@
use std::{result::Result as StdResult, sync::Arc};
use std::result::Result as StdResult;
use anyhow::{anyhow, bail, Context as _, Result};
use anyhow::{anyhow, bail, Result};
use async_trait::async_trait;
use cynic::{MutationBuilder, QueryBuilder};
use firebase::{FetchAccessTokenResponse, FirebaseError};
use futures::FutureExt;
use firebase::FirebaseError;
use instant::Duration;
#[cfg(test)]
use mockall::{automock, predicate::*};
use oauth2::TokenResponse;
use thiserror::Error;
use galaxy_core::errors::{AnyhowErrorExt, ErrorExt};
use galaxy_graphql::client::Operation;
use galaxy_graphql::mutations::expire_api_key::{
ExpireApiKey, ExpireApiKeyResult, ExpireApiKeyVariables,
};
use galaxy_graphql::queries::get_conversation_usage::{
ConversationUsage, GetConversationUsage, GetConversationUsageVariables, UserResult,
};
use galaxy_graphql::mutations::expire_api_key::ExpireApiKeyResult;
use galaxy_graphql::queries::get_conversation_usage::ConversationUsage;
use galaxy_graphql::mutations::set_user_is_onboarded::{
SetUserIsOnboarded, SetUserIsOnboardedResult, SetUserIsOnboardedVariables,
};
use galaxy_graphql::mutations::update_user_settings::{
UpdateUserSettings, UpdateUserSettingsInput, UpdateUserSettingsResult,
UpdateUserSettingsVariables,
};
use galaxy_graphql::mutations::{
create_anonymous_user::{
AnonymousUserType, CreateAnonymousUser, CreateAnonymousUserResult,
CreateAnonymousUserVariables,
},
generate_api_key::{
GenerateApiKey, GenerateApiKeyInput, GenerateApiKeyResult, GenerateApiKeyVariables,
},
mint_custom_token::{MintCustomTokenResult, MintCustomTokenVariables},
use galaxy_graphql::mutations::create_anonymous_user::{
AnonymousUserType, CreateAnonymousUserResult,
};
use galaxy_graphql::mutations::generate_api_key::GenerateApiKeyResult;
use galaxy_graphql::mutations::mint_custom_token::MintCustomTokenResult;
use galaxy_graphql::object_permissions::OwnerType;
use galaxy_graphql::queries::api_keys::{
ApiKeyProperties, ApiKeyPropertiesResult, ApiKeys, ApiKeysVariables,
};
use galaxy_graphql::queries::get_user::{GetUser, GetUserVariables, UserOutput as GqlUserOutput};
use galaxy_graphql::queries::get_user_settings::{GetUserSettings, GetUserSettingsVariables};
use galaxyui::r#async::BoxFuture;
use galaxy_graphql::queries::api_keys::ApiKeyProperties;
use galaxy_graphql::queries::get_user::UserOutput as GqlUserOutput;
use crate::auth::UserUid;
use crate::server::graphql::{default_request_options, get_user_facing_error_message};
use crate::server::ids::ApiKeyUid;
use crate::server::server_api::register_error;
use crate::server::server_api::EXPERIMENT_ID_HEADER;
use crate::settings::PrivacySettingsSnapshot;
use crate::{
auth::{
credentials::{AuthToken, Credentials, FirebaseToken, LoginToken, RefreshToken},
user::FirebaseAuthTokens,
credentials::{AuthToken, Credentials, FirebaseToken, LoginToken},
user::User,
},
channel::ChannelState,
convert_to_server_experiment,
server::{
datetime_ext::DateTimeExt as _, experiments::ServerExperiment,
graphql::get_request_context, server_api::ServerApiEvent,
},
server::experiments::ServerExperiment,
};
use super::ServerApi;
/// Error messages returned from the Firebase REST API when attempting to convert a refresh token
/// into an access token that indicate the user's token is in an errored state.
/// These are "soft" errors because the user likely just needs to log in again.
/// See https://firebase.google.com/docs/reference/rest/auth#section-refresh-token.
static FETCH_ACCESS_TOKEN_SOFT_ERROR_MESSAGES: &[&str] = &[
"TOKEN_EXPIRED",
"INVALID_REFRESH_TOKEN",
"MISSING_REFRESH_TOKEN",
];
/// Error messages returned from the Firebase REST API when attempting to convert a refresh token
/// into an access token that indicate the user's account is in an errored state.
/// These are "hard" errors because the user likely can no longer sign in with their account,
/// for example if it were disabled or deleted.
/// See https://firebase.google.com/docs/reference/rest/auth#section-refresh-token.
static FETCH_ACCESS_TOKEN_HARD_ERROR_MESSAGES: &[&str] = &["USER_DISABLED", "USER_NOT_FOUND"];
const FETCH_ACCESS_TOKEN_TIMEOUT: Duration = Duration::from_secs(5);
/// Header key for the ambient workload token attached to multi-agent requests.
pub const AMBIENT_WORKLOAD_TOKEN_HEADER: &str = "X-Warp-Ambient-Workload-Token";
/// Header key for the cloud agent task ID attached to requests from ambient agents.
pub const CLOUD_AGENT_ID_HEADER: &str = "X-Warp-Cloud-Agent-ID";
/// Duration for which the ambient workload token is valid (3 hours).
const AMBIENT_WORKLOAD_TOKEN_DURATION: Duration = Duration::from_secs(3 * 60 * 60);
/// User settings that are currently 'synced' (e.g. stored server-side) on a per-user basis.
#[derive(Copy, Clone, Debug, Default)]
pub struct SyncedUserSettings {
@@ -217,459 +163,108 @@ pub trait AuthClient: 'static + Send + Sync {
impl AuthClient for ServerApi {
async fn create_anonymous_user(
&self,
referral_code: Option<String>,
anonymous_user_type: AnonymousUserType,
_referral_code: Option<String>,
_anonymous_user_type: AnonymousUserType,
) -> Result<CreateAnonymousUserResult> {
let variables = CreateAnonymousUserVariables {
input: galaxy_graphql::mutations::create_anonymous_user::CreateAnonymousUserInput {
anonymous_user_type,
expiration_type: galaxy_graphql::mutations::create_anonymous_user::AnonymousUserExpirationType::NoExpiration,
referral_code,
},
request_context: get_request_context(),
};
let operation = CreateAnonymousUser::build(variables);
let response = operation
.send_request(self.client.clone(), default_request_options())
.await?;
Ok(response
.data
.ok_or_else(|| anyhow!("missing data in response"))?
.create_anonymous_user)
bail!("Server auth disabled")
}
async fn get_or_refresh_access_token(&self) -> Result<AuthToken> {
if cfg!(feature = "skip_login") {
bail!("skip_login enabled; failing all authenticated requests");
}
let Some(credentials) = self.auth_state.credentials() else {
bail!("Attempted to retrieve access token when user is logged out");
};
match credentials {
Credentials::ApiKey { key, .. } => Ok(AuthToken::ApiKey(key)),
Credentials::Firebase(auth_tokens) => {
let expiration_time = auth_tokens.expiration_time;
// Generate a new ID token if the token has expired or will expire in the
// next five minutes. This matches the behavior of the Firebase Auth SDK.
if chrono::DateTime::now() + chrono::Duration::minutes(5) >= expiration_time {
let refresh_token = auth_tokens.refresh_token.clone();
let firebase_token = FirebaseToken::Refresh(RefreshToken::new(refresh_token));
let result = fetch_auth_tokens(self.client.clone(), firebase_token).await;
if let Err(UserAuthenticationError::DeniedAccessToken(_)) = result {
let _ = self.event_sender.send(ServerApiEvent::NeedsReauth).await;
}
let new_firebase_token_info = result?;
self.auth_state
.update_firebase_tokens(new_firebase_token_info.clone());
let _ = self
.event_sender
.send(ServerApiEvent::AccessTokenRefreshed {
token: new_firebase_token_info.id_token.clone(),
})
.await;
return Ok(AuthToken::Firebase(new_firebase_token_info.id_token));
}
Ok(AuthToken::Firebase(auth_tokens.id_token))
}
Credentials::SessionCookie => Ok(AuthToken::NoAuth),
#[cfg(any(test, feature = "integration_tests", feature = "skip_login"))]
Credentials::Test => Ok(AuthToken::NoAuth),
}
Ok(AuthToken::NoAuth)
}
async fn fetch_user(
&self,
token: LoginToken,
for_refresh: bool,
_token: LoginToken,
_for_refresh: bool,
) -> StdResult<FetchUserResult, UserAuthenticationError> {
let new_credentials = exchange_credentials(self.client.clone(), token).await?;
let auth_token = new_credentials.bearer_token();
let user_output = self
.fetch_user_properties(auth_token.as_bearer_token())
.await
.context("Failed to fetch user response data")
.map_err(UserAuthenticationError::Unexpected)?;
let UserProperties {
user,
server_experiments,
llms,
api_key_owner_type,
} = user_output.into();
// Store the owner type if using an API key.
let new_credentials = match new_credentials {
Credentials::ApiKey { key, .. } => Credentials::ApiKey {
key,
owner_type: api_key_owner_type,
},
other => other,
};
Ok(FetchUserResult {
user,
credentials: new_credentials,
server_experiments,
from_refresh: for_refresh,
llms,
})
Err(UserAuthenticationError::Unexpected(anyhow!("Server auth disabled")))
}
async fn fetch_new_custom_token(&self) -> Result<MintCustomTokenResult> {
let variables = MintCustomTokenVariables {
request_context: get_request_context(),
};
let operation =
galaxy_graphql::mutations::mint_custom_token::MintCustomToken::build(variables);
let response = self.send_graphql_request(operation, None).await?;
Ok(response.mint_custom_token)
bail!("Server auth disabled")
}
fn on_custom_token_fetched(
&self,
response: Result<MintCustomTokenResult>,
_response: Result<MintCustomTokenResult>,
) -> Result<String, MintCustomTokenError> {
match response {
Ok(response_data) => match response_data {
MintCustomTokenResult::MintCustomTokenOutput(output) => Ok(output.custom_token),
MintCustomTokenResult::UserFacingError(user_facing_error) => {
Err(MintCustomTokenError::UserFacingError(
get_user_facing_error_message(user_facing_error),
))
}
MintCustomTokenResult::Unknown => Err(MintCustomTokenError::Unknown),
},
Err(_) => Err(MintCustomTokenError::Unknown),
}
Err(MintCustomTokenError::Unknown)
}
async fn fetch_user_properties<'a>(
&self,
auth_token: Option<&'a str>,
_auth_token: Option<&'a str>,
) -> Result<GqlUserOutput> {
let variables = GetUserVariables {
request_context: get_request_context(),
};
let operation = GetUser::build(variables);
let response = operation
.send_request(
self.client.clone(),
galaxy_graphql::client::RequestOptions {
auth_token: auth_token.map(ToOwned::to_owned),
headers: std::collections::HashMap::from([(
EXPERIMENT_ID_HEADER.to_string(),
self.auth_state.anonymous_id(),
)]),
..default_request_options()
},
)
.await?
.data
.ok_or_else(|| anyhow!("Expected valid response.data"))?;
match response.user {
galaxy_graphql::queries::get_user::UserResult::UserOutput(user_output) => Ok(user_output),
galaxy_graphql::queries::get_user::UserResult::Unknown => {
Err(anyhow!("Unable to fetch user"))
}
}
bail!("Server auth disabled")
}
async fn get_user_settings(&self) -> Result<Option<SyncedUserSettings>> {
let variables = GetUserSettingsVariables {
request_context: get_request_context(),
};
let operation = GetUserSettings::build(variables);
let response = self.send_graphql_request(operation, None).await?;
match response.user {
galaxy_graphql::queries::get_user_settings::UserResult::UserOutput(user_output) => {
match user_output.user.settings {
Some(user_settings) => Ok(Some(SyncedUserSettings {
is_cloud_conversation_storage_enabled: user_settings
.is_cloud_conversation_storage_enabled,
is_crash_reporting_enabled: user_settings.is_crash_reporting_enabled,
is_telemetry_enabled: user_settings.is_telemetry_enabled,
})),
None => Ok(None),
}
}
galaxy_graphql::queries::get_user_settings::UserResult::Unknown => {
Err(anyhow!("Unable to fetch user settings"))
}
}
Ok(None)
}
// Returns a history of the current user's conversation usage over the past n days.
async fn get_conversation_usage_history(
&self,
days: Option<i32>,
limit: Option<i32>,
last_updated_end_timestamp: Option<galaxy_graphql::scalars::Time>,
_days: Option<i32>,
_limit: Option<i32>,
_last_updated_end_timestamp: Option<galaxy_graphql::scalars::Time>,
) -> Result<Vec<ConversationUsage>> {
let operation = GetConversationUsage::build(GetConversationUsageVariables {
request_context: get_request_context(),
days,
limit,
last_updated_end_timestamp,
});
let response = self.send_graphql_request(operation, None).await?;
match response.user {
UserResult::UserOutput(out) => Ok(out.user.conversation_usage),
UserResult::Unknown => Err(anyhow!("Unable to fetch conversation usage")),
}
Ok(vec![])
}
async fn set_is_telemetry_enabled(&self, value: bool) -> Result<()> {
let variables = UpdateUserSettingsVariables {
input: UpdateUserSettingsInput {
telemetry_enabled: Some(value),
..Default::default()
},
request_context: get_request_context(),
};
let operation = UpdateUserSettings::build(variables);
let result = self
.send_graphql_request(operation, None)
.await?
.update_user_settings;
match result {
UpdateUserSettingsResult::UpdateUserSettingsOutput(_) => Ok(()),
UpdateUserSettingsResult::UserFacingError(user_facing_error) => {
Err(anyhow!(get_user_facing_error_message(user_facing_error)))
}
UpdateUserSettingsResult::Unknown => Err(anyhow!("failed to set telemetry enabled")),
}
async fn set_is_telemetry_enabled(&self, _value: bool) -> Result<()> {
Ok(())
}
async fn set_is_crash_reporting_enabled(&self, value: bool) -> Result<()> {
let variables = UpdateUserSettingsVariables {
input: UpdateUserSettingsInput {
crash_reporting_enabled: Some(value),
..Default::default()
},
request_context: get_request_context(),
};
let operation = UpdateUserSettings::build(variables);
let result = self
.send_graphql_request(operation, None)
.await?
.update_user_settings;
match result {
UpdateUserSettingsResult::UpdateUserSettingsOutput(_) => Ok(()),
UpdateUserSettingsResult::UserFacingError(user_facing_error) => {
Err(anyhow!(get_user_facing_error_message(user_facing_error)))
}
UpdateUserSettingsResult::Unknown => {
Err(anyhow!("failed to set crash reporting enabled"))
}
}
async fn set_is_crash_reporting_enabled(&self, _value: bool) -> Result<()> {
Ok(())
}
async fn set_is_cloud_conversation_storage_enabled(&self, value: bool) -> Result<()> {
let variables = UpdateUserSettingsVariables {
input: UpdateUserSettingsInput {
cloud_conversation_storage_enabled: Some(value),
..Default::default()
},
request_context: get_request_context(),
};
let operation = UpdateUserSettings::build(variables);
let result = self
.send_graphql_request(operation, None)
.await?
.update_user_settings;
match result {
UpdateUserSettingsResult::UpdateUserSettingsOutput(_) => Ok(()),
UpdateUserSettingsResult::UserFacingError(user_facing_error) => {
Err(anyhow!(get_user_facing_error_message(user_facing_error)))
}
UpdateUserSettingsResult::Unknown => {
Err(anyhow!("failed to set cloud conversation storage enabled"))
}
}
async fn set_is_cloud_conversation_storage_enabled(&self, _value: bool) -> Result<()> {
Ok(())
}
async fn update_user_settings(&self, settings_snapshot: PrivacySettingsSnapshot) -> Result<()> {
let variables = UpdateUserSettingsVariables {
input: UpdateUserSettingsInput {
telemetry_enabled: Some(settings_snapshot.is_telemetry_enabled()),
crash_reporting_enabled: Some(settings_snapshot.is_crash_reporting_enabled()),
cloud_conversation_storage_enabled: settings_snapshot
.cloud_conversation_storage_enabled(),
},
request_context: get_request_context(),
};
let operation = UpdateUserSettings::build(variables);
let result = self
.send_graphql_request(operation, None)
.await?
.update_user_settings;
match result {
UpdateUserSettingsResult::UpdateUserSettingsOutput(_) => Ok(()),
UpdateUserSettingsResult::UserFacingError(user_facing_error) => {
Err(anyhow!(get_user_facing_error_message(user_facing_error)))
}
UpdateUserSettingsResult::Unknown => Err(anyhow!("failed to update user settings")),
}
async fn update_user_settings(&self, _settings_snapshot: PrivacySettingsSnapshot) -> Result<()> {
Ok(())
}
async fn set_user_is_onboarded(&self) -> Result<bool> {
let variables = SetUserIsOnboardedVariables {
request_context: get_request_context(),
};
let operation = SetUserIsOnboarded::build(variables);
let result = self
.send_graphql_request(operation, None)
.await?
.set_user_is_onboarded;
match result {
SetUserIsOnboardedResult::SetUserIsOnboardedOutput(_) => Ok(true),
SetUserIsOnboardedResult::UserFacingError(user_facing_error) => {
Err(anyhow!(get_user_facing_error_message(user_facing_error)))
}
SetUserIsOnboardedResult::Unknown => Err(anyhow!("failed to set user is onboarded")),
}
Ok(true)
}
async fn request_device_code(
&self,
) -> StdResult<oauth2::StandardDeviceAuthorizationResponse, UserAuthenticationError> {
self.oauth_client
.exchange_device_code()
.request_async(self.client.as_ref())
.await
.context("Failed to generate device code")
.map_err(UserAuthenticationError::Unexpected)
Err(UserAuthenticationError::Unexpected(anyhow!("Server auth disabled")))
}
async fn exchange_device_access_token(
&self,
details: &oauth2::StandardDeviceAuthorizationResponse,
timeout: Duration,
_details: &oauth2::StandardDeviceAuthorizationResponse,
_timeout: Duration,
) -> StdResult<FirebaseToken, UserAuthenticationError> {
let result = self
.oauth_client
.exchange_device_access_token(details)
.request_async(
self.client.as_ref(),
|delay| galaxyui::r#async::Timer::after(delay).map(|_| ()),
Some(timeout),
)
.await
.context("Unable to obtain access token")
.map_err(UserAuthenticationError::Unexpected)?;
// Firebase doesn't directly support the device flow. Instead, the server mints a short-lived
// custom access token, which we can then exchange for a refresh token.
Ok(FirebaseToken::Custom(
result.access_token().secret().to_string(),
))
Err(UserAuthenticationError::Unexpected(anyhow!("Server auth disabled")))
}
// API Keys
async fn list_api_keys(&self) -> Result<Vec<ApiKeyProperties>> {
let variables = ApiKeysVariables {
request_context: get_request_context(),
};
let operation = ApiKeys::build(variables);
let response = self.send_graphql_request(operation, None).await?;
match response.api_keys {
ApiKeyPropertiesResult::ApiKeyPropertiesOutput(output) => Ok(output.api_keys),
ApiKeyPropertiesResult::UserFacingError(e) => {
Err(anyhow!(get_user_facing_error_message(e)))
}
ApiKeyPropertiesResult::Unknown => Err(anyhow!("failed to fetch API keys")),
}
Ok(vec![])
}
async fn create_api_key(
&self,
name: String,
team_id: Option<cynic::Id>,
expires_at: Option<galaxy_graphql::scalars::Time>,
_name: String,
_team_id: Option<cynic::Id>,
_expires_at: Option<galaxy_graphql::scalars::Time>,
) -> Result<GenerateApiKeyResult> {
let variables = GenerateApiKeyVariables {
input: GenerateApiKeyInput {
name,
team_id,
expires_at,
},
request_context: get_request_context(),
};
let operation = GenerateApiKey::build(variables);
let response = self.send_graphql_request(operation, None).await?;
Ok(response.generate_api_key)
bail!("Server auth disabled")
}
async fn expire_api_key(&self, key_uid: &ApiKeyUid) -> Result<ExpireApiKeyResult> {
let variables = ExpireApiKeyVariables {
key_uid: key_uid.into(),
request_context: get_request_context(),
};
let op = ExpireApiKey::build(variables);
let res = self.send_graphql_request(op, None).await?;
Ok(res.expire_api_key)
async fn expire_api_key(&self, _key_uid: &ApiKeyUid) -> Result<ExpireApiKeyResult> {
bail!("Server auth disabled")
}
async fn get_or_create_ambient_workload_token(&self) -> Result<Option<String>> {
if cfg!(target_family = "wasm") {
return Ok(None);
}
// Check if we have a cached token that's still valid (with 5 minute buffer).
// Tokens without an expiration time are always considered valid.
{
let cached = self.ambient_workload_token.lock();
if let Some(ref token) = *cached {
let is_valid = token.expires_at.is_none_or(|expires_at| {
chrono::Utc::now() + chrono::Duration::minutes(5) < expires_at
});
if is_valid {
return Ok(Some(token.token.clone()));
}
}
}
// Issue a new token.
let workload_token = match galaxy_isolation_platform::issue_workload_token(Some(
AMBIENT_WORKLOAD_TOKEN_DURATION,
))
.await
{
Ok(token) => token,
Err(galaxy_isolation_platform::IsolationPlatformError::NoIsolationPlatformDetected) => {
return Ok(None);
}
Err(e) => return Err(e.into()),
};
let token_str = workload_token.token.clone();
{
let mut cached = self.ambient_workload_token.lock();
*cached = Some(workload_token);
}
Ok(Some(token_str))
Ok(None)
}
}