Add shared OAuth flows

apps/aether-gateway/src/oauth/http_executor.rs

@@ -0,0 +1,165 @@
+use crate::admin_api::AdminAppState;
+use crate::{AppState, GatewayError};
+use aether_contracts::{
+    ExecutionPlan, ExecutionResult, ExecutionTimeouts, RequestBody,
+    EXECUTION_REQUEST_FOLLOW_REDIRECTS_HEADER,
+};
+use aether_oauth::core::OAuthError;
+use aether_oauth::network::{OAuthHttpExecutor, OAuthHttpRequest, OAuthHttpResponse};
+use async_trait::async_trait;
+use base64::{engine::general_purpose::STANDARD, Engine as _};
+use flate2::read::{DeflateDecoder, GzDecoder};
+use std::collections::BTreeMap;
+use std::io::Read;
+
+#[derive(Clone)]
+pub(crate) struct GatewayOAuthHttpExecutor<'a> {
+    app: AppState,
+    _marker: std::marker::PhantomData<&'a AppState>,
+}
+
+impl<'a> GatewayOAuthHttpExecutor<'a> {
+    pub(crate) fn new(state: AdminAppState<'a>) -> Self {
+        Self {
+            app: state.cloned_app(),
+            _marker: std::marker::PhantomData,
+        }
+    }
+
+    pub(crate) fn from_app(app: &'a AppState) -> Self {
+        Self {
+            app: app.clone(),
+            _marker: std::marker::PhantomData,
+        }
+    }
+}
+
+#[async_trait]
+impl<'a> OAuthHttpExecutor for GatewayOAuthHttpExecutor<'a> {
+    async fn execute(&self, request: OAuthHttpRequest) -> Result<OAuthHttpResponse, OAuthError> {
+        let body = if let Some(json_body) = request.json_body {
+            RequestBody::from_json(json_body)
+        } else {
+            RequestBody {
+                json_body: None,
+                body_bytes_b64: request.body_bytes.map(|bytes| STANDARD.encode(bytes)),
+                body_ref: None,
+            }
+        };
+        let timeouts = request.network.timeouts;
+        let mut headers = request.headers;
+        headers
+            .entry(EXECUTION_REQUEST_FOLLOW_REDIRECTS_HEADER.to_string())
+            .or_insert_with(|| "true".to_string());
+        let plan = ExecutionPlan {
+            request_id: request.request_id,
+            candidate_id: None,
+            provider_name: Some("oauth".to_string()),
+            provider_id: String::new(),
+            endpoint_id: String::new(),
+            key_id: String::new(),
+            method: request.method.as_str().to_string(),
+            url: request.url,
+            headers,
+            content_type: request.content_type,
+            content_encoding: None,
+            body,
+            stream: false,
+            client_api_format: "oauth:exchange".to_string(),
+            provider_api_format: "oauth:exchange".to_string(),
+            model_name: Some("oauth-exchange".to_string()),
+            proxy: request.network.proxy,
+            tls_profile: None,
+            timeouts: Some(ExecutionTimeouts {
+                connect_ms: Some(timeouts.connect_ms),
+                read_ms: Some(timeouts.read_ms),
+                write_ms: Some(timeouts.write_ms),
+                pool_ms: Some(timeouts.connect_ms),
+                total_ms: Some(timeouts.total_ms),
+                ..ExecutionTimeouts::default()
+            }),
+        };
+        let result =
+            crate::execution_runtime::execute_execution_runtime_sync_plan(&self.app, None, &plan)
+                .await
+                .map_err(gateway_error_to_oauth_error)?;
+        Ok(OAuthHttpResponse {
+            status_code: result.status_code,
+            body_text: execution_body_text(&result),
+            json_body: execution_json_body(&result),
+        })
+    }
+}
+
+fn execution_json_body(result: &ExecutionResult) -> Option<serde_json::Value> {
+    result
+        .body
+        .as_ref()
+        .and_then(|body| body.json_body.clone())
+        .or_else(|| {
+            result
+                .body
+                .as_ref()
+                .and_then(|body| execution_body_bytes(&result.headers, body))
+                .and_then(|bytes| serde_json::from_slice::<serde_json::Value>(&bytes).ok())
+        })
+}
+
+fn execution_body_text(result: &ExecutionResult) -> String {
+    result
+        .body
+        .as_ref()
+        .and_then(|body| execution_body_bytes(&result.headers, body))
+        .map(|bytes| String::from_utf8_lossy(&bytes).to_string())
+        .or_else(|| {
+            result
+                .body
+                .as_ref()
+                .and_then(|body| body.json_body.as_ref())
+                .and_then(|value| serde_json::to_string(value).ok())
+        })
+        .unwrap_or_default()
+}
+
+fn execution_body_bytes(
+    headers: &BTreeMap<String, String>,
+    body: &aether_contracts::ResponseBody,
+) -> Option<Vec<u8>> {
+    let bytes = body
+        .body_bytes_b64
+        .as_deref()
+        .and_then(|value| STANDARD.decode(value).ok())?;
+    decode_response_bytes(&bytes, headers.get("content-encoding").map(String::as_str))
+        .or(Some(bytes))
+}
+
+fn decode_response_bytes(bytes: &[u8], content_encoding: Option<&str>) -> Option<Vec<u8>> {
+    match content_encoding
+        .map(str::trim)
+        .filter(|value| !value.is_empty())
+        .map(str::to_ascii_lowercase)
+        .as_deref()
+    {
+        Some("gzip") => {
+            let mut decoder = GzDecoder::new(bytes);
+            let mut out = Vec::new();
+            decoder.read_to_end(&mut out).ok()?;
+            Some(out)
+        }
+        Some("deflate") => {
+            let mut decoder = DeflateDecoder::new(bytes);
+            let mut out = Vec::new();
+            decoder.read_to_end(&mut out).ok()?;
+            Some(out)
+        }
+        _ => None,
+    }
+}
+
+fn gateway_error_to_oauth_error(error: GatewayError) -> OAuthError {
+    match error {
+        GatewayError::UpstreamUnavailable { message, .. }
+        | GatewayError::ControlUnavailable { message, .. }
+        | GatewayError::Internal(message) => OAuthError::Transport(message),
+    }
+}

apps/aether-gateway/src/oauth/identity_repo.rs

@@ -0,0 +1,803 @@
+use crate::handlers::shared::decrypt_catalog_secret_with_fallbacks;
+use crate::{AppState, GatewayError};
+use aether_data::repository::oauth_providers::StoredOAuthProviderConfig;
+use aether_data::repository::users::StoredUserAuthRecord;
+use aether_oauth::identity::{IdentityClaims, IdentityOAuthProviderConfig};
+use chrono::{DateTime, Utc};
+use serde::Serialize;
+use serde_json::{json, Value};
+use sqlx::Row;
+use uuid::Uuid;
+
+const LINUXDO_AUTHORIZE_URL: &str = "https://connect.linux.do/oauth2/authorize";
+const LINUXDO_TOKEN_URL: &str = "https://connect.linux.do/oauth2/token";
+const LINUXDO_USERINFO_URL: &str = "https://connect.linux.do/api/user";
+
+const FIND_OAUTH_LINKED_USER_SQL: &str = r#"
+SELECT
+    users.id,
+    users.email,
+    users.email_verified,
+    users.username,
+    users.password_hash,
+    users.role::text AS role,
+    users.auth_source::text AS auth_source,
+    users.allowed_providers,
+    users.allowed_api_formats,
+    users.allowed_models,
+    users.is_active,
+    users.is_deleted,
+    users.created_at,
+    users.last_login_at
+FROM user_oauth_links
+JOIN users ON users.id = user_oauth_links.user_id
+WHERE user_oauth_links.provider_type = $1
+  AND user_oauth_links.provider_user_id = $2
+LIMIT 1
+"#;
+
+const FIND_USER_BY_EMAIL_SQL: &str = r#"
+SELECT
+    id,
+    email,
+    email_verified,
+    username,
+    password_hash,
+    role::text AS role,
+    auth_source::text AS auth_source,
+    allowed_providers,
+    allowed_api_formats,
+    allowed_models,
+    is_active,
+    is_deleted,
+    created_at,
+    last_login_at
+FROM users
+WHERE LOWER(email) = LOWER($1)
+  AND is_deleted IS FALSE
+LIMIT 1
+"#;
+
+const CHECK_USERNAME_TAKEN_SQL: &str = r#"
+SELECT id
+FROM users
+WHERE username = $1
+LIMIT 1
+"#;
+
+const CREATE_OAUTH_USER_SQL: &str = r#"
+INSERT INTO users (
+    id,
+    email,
+    email_verified,
+    username,
+    password_hash,
+    role,
+    auth_source,
+    is_active,
+    is_deleted,
+    created_at,
+    updated_at,
+    last_login_at
+)
+VALUES (
+    $1,
+    $2,
+    TRUE,
+    $3,
+    NULL,
+    'user'::userrole,
+    'oauth'::authsource,
+    TRUE,
+    FALSE,
+    $4,
+    $4,
+    $4
+)
+RETURNING
+    id,
+    email,
+    email_verified,
+    username,
+    password_hash,
+    role::text AS role,
+    auth_source::text AS auth_source,
+    allowed_providers,
+    allowed_api_formats,
+    allowed_models,
+    is_active,
+    is_deleted,
+    created_at,
+    last_login_at
+"#;
+
+const UPSERT_OAUTH_LINK_SQL: &str = r#"
+INSERT INTO user_oauth_links (
+    id,
+    user_id,
+    provider_type,
+    provider_user_id,
+    provider_username,
+    provider_email,
+    extra_data,
+    linked_at,
+    last_login_at
+)
+VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $8)
+ON CONFLICT (user_id, provider_type) DO UPDATE
+SET provider_user_id = EXCLUDED.provider_user_id,
+    provider_username = EXCLUDED.provider_username,
+    provider_email = EXCLUDED.provider_email,
+    extra_data = EXCLUDED.extra_data,
+    last_login_at = EXCLUDED.last_login_at
+"#;
+
+const TOUCH_OAUTH_LINK_SQL: &str = r#"
+UPDATE user_oauth_links
+SET provider_username = COALESCE($3, provider_username),
+    provider_email = COALESCE($4, provider_email),
+    extra_data = COALESCE($5, extra_data),
+    last_login_at = $6
+WHERE provider_type = $1
+  AND provider_user_id = $2
+"#;
+
+const CREATE_AUTH_USER_WALLET_SQL: &str = r#"
+INSERT INTO wallets (
+    id,
+    user_id,
+    api_key_id,
+    balance,
+    gift_balance,
+    limit_mode,
+    currency,
+    status,
+    total_recharged,
+    total_consumed,
+    total_refunded,
+    total_adjusted,
+    created_at,
+    updated_at
+)
+VALUES (
+    $1,
+    $2,
+    NULL,
+    0,
+    $3,
+    $4,
+    'USD',
+    'active',
+    0,
+    0,
+    0,
+    $3,
+    NOW(),
+    NOW()
+)
+"#;
+
+const CREATE_AUTH_USER_WALLET_GIFT_TX_SQL: &str = r#"
+INSERT INTO wallet_transactions (
+    id,
+    wallet_id,
+    category,
+    reason_code,
+    amount,
+    balance_before,
+    balance_after,
+    recharge_balance_before,
+    recharge_balance_after,
+    gift_balance_before,
+    gift_balance_after,
+    link_type,
+    link_id,
+    operator_id,
+    description,
+    created_at
+)
+VALUES (
+    $1,
+    $2,
+    'gift',
+    'gift_initial',
+    $3,
+    0,
+    $3,
+    0,
+    0,
+    0,
+    $3,
+    'system_task',
+    $4,
+    NULL,
+    '用户初始赠款',
+    NOW()
+)
+"#;
+
+const LIST_OAUTH_LINKS_SQL: &str = r#"
+SELECT
+    user_oauth_links.provider_type,
+    oauth_providers.display_name,
+    user_oauth_links.provider_username,
+    user_oauth_links.provider_email,
+    user_oauth_links.linked_at,
+    user_oauth_links.last_login_at,
+    oauth_providers.is_enabled AS provider_enabled
+FROM user_oauth_links
+JOIN oauth_providers
+  ON oauth_providers.provider_type = user_oauth_links.provider_type
+WHERE user_oauth_links.user_id = $1
+ORDER BY user_oauth_links.linked_at ASC
+"#;
+
+const FIND_OAUTH_LINK_OWNER_SQL: &str = r#"
+SELECT user_id
+FROM user_oauth_links
+WHERE provider_type = $1
+  AND provider_user_id = $2
+LIMIT 1
+"#;
+
+const FIND_USER_PROVIDER_LINK_OWNER_SQL: &str = r#"
+SELECT user_id
+FROM user_oauth_links
+WHERE user_id = $1
+  AND provider_type = $2
+LIMIT 1
+"#;
+
+const COUNT_USER_OAUTH_LINKS_SQL: &str = r#"
+SELECT COUNT(*)::bigint AS link_count
+FROM user_oauth_links
+WHERE user_id = $1
+"#;
+
+const DELETE_USER_OAUTH_LINK_SQL: &str = r#"
+DELETE FROM user_oauth_links
+WHERE user_id = $1
+  AND provider_type = $2
+"#;
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
+pub(crate) struct IdentityOAuthProviderSummary {
+    pub(crate) provider_type: String,
+    pub(crate) display_name: String,
+}
+
+#[derive(Debug, Clone, PartialEq, Serialize)]
+pub(crate) struct IdentityOAuthLinkSummary {
+    pub(crate) provider_type: String,
+    pub(crate) display_name: String,
+    pub(crate) provider_username: Option<String>,
+    pub(crate) provider_email: Option<String>,
+    pub(crate) linked_at: Option<String>,
+    pub(crate) last_login_at: Option<String>,
+    pub(crate) provider_enabled: bool,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub(crate) enum IdentityOAuthAccountError {
+    ProviderUnavailable,
+    RegistrationDisabled,
+    EmailExistsLocal,
+    EmailIsLdap,
+    EmailIsOauth,
+    OAuthAlreadyBound,
+    AlreadyBoundProvider,
+    LastOAuthBinding,
+    LastLoginMethod,
+    Storage(String),
+}
+
+impl IdentityOAuthAccountError {
+    pub(crate) fn code(&self) -> &'static str {
+        match self {
+            Self::ProviderUnavailable | Self::Storage(_) => "provider_unavailable",
+            Self::RegistrationDisabled => "registration_disabled",
+            Self::EmailExistsLocal => "email_exists_local",
+            Self::EmailIsLdap => "email_is_ldap",
+            Self::EmailIsOauth => "email_is_oauth",
+            Self::OAuthAlreadyBound => "oauth_already_bound",
+            Self::AlreadyBoundProvider => "already_bound_provider",
+            Self::LastOAuthBinding => "last_oauth_binding",
+            Self::LastLoginMethod => "last_login_method",
+        }
+    }
+
+    pub(crate) fn detail(&self) -> String {
+        match self {
+            Self::Storage(message) => message.clone(),
+            _ => self.code().to_string(),
+        }
+    }
+}
+
+pub(crate) async fn list_enabled_identity_oauth_providers(
+    state: &AppState,
+) -> Result<Vec<IdentityOAuthProviderSummary>, GatewayError> {
+    let mut providers = state
+        .list_oauth_provider_configs()
+        .await?
+        .into_iter()
+        .filter(|provider| provider.is_enabled)
+        .map(|provider| IdentityOAuthProviderSummary {
+            provider_type: provider.provider_type,
+            display_name: provider.display_name,
+        })
+        .collect::<Vec<_>>();
+    providers.sort_by(|left, right| left.provider_type.cmp(&right.provider_type));
+    Ok(providers)
+}
+
+pub(crate) async fn get_enabled_identity_oauth_provider_config(
+    state: &AppState,
+    provider_type: &str,
+) -> Result<Option<IdentityOAuthProviderConfig>, IdentityOAuthAccountError> {
+    let provider_type = provider_type.trim().to_ascii_lowercase();
+    let config = state
+        .get_oauth_provider_config(&provider_type)
+        .await
+        .map_err(|err| IdentityOAuthAccountError::Storage(format!("{err:?}")))?;
+    let Some(config) = config.filter(|config| config.is_enabled) else {
+        return Ok(None);
+    };
+    stored_provider_config_to_identity_config(state, config).map(Some)
+}
+
+pub(crate) async fn list_identity_oauth_links(
+    state: &AppState,
+    user_id: &str,
+) -> Result<Vec<IdentityOAuthLinkSummary>, GatewayError> {
+    let Some(pool) = state.postgres_pool() else {
+        return Ok(Vec::new());
+    };
+    let rows = sqlx::query(LIST_OAUTH_LINKS_SQL)
+        .bind(user_id)
+        .fetch_all(&pool)
+        .await
+        .map_err(sql_gateway_error)?;
+    rows.iter().map(map_link_summary_row).collect()
+}
+
+pub(crate) async fn list_bindable_identity_oauth_providers(
+    state: &AppState,
+    user_id: &str,
+) -> Result<Vec<IdentityOAuthProviderSummary>, GatewayError> {
+    let linked = list_identity_oauth_links(state, user_id)
+        .await?
+        .into_iter()
+        .map(|link| link.provider_type)
+        .collect::<std::collections::BTreeSet<_>>();
+    let providers = list_enabled_identity_oauth_providers(state)
+        .await?
+        .into_iter()
+        .filter(|provider| !linked.contains(&provider.provider_type))
+        .collect();
+    Ok(providers)
+}
+
+pub(crate) async fn resolve_identity_oauth_login_user(
+    state: &AppState,
+    claims: &IdentityClaims,
+) -> Result<StoredUserAuthRecord, IdentityOAuthAccountError> {
+    let Some(pool) = state.postgres_pool() else {
+        return Err(IdentityOAuthAccountError::ProviderUnavailable);
+    };
+    let now = Utc::now();
+    if let Some(row) = sqlx::query(FIND_OAUTH_LINKED_USER_SQL)
+        .bind(&claims.provider_type)
+        .bind(&claims.subject)
+        .fetch_optional(&pool)
+        .await
+        .map_err(repo_sql_error)?
+    {
+        sqlx::query(TOUCH_OAUTH_LINK_SQL)
+            .bind(&claims.provider_type)
+            .bind(&claims.subject)
+            .bind(claims.username.as_deref())
+            .bind(claims.email.as_deref())
+            .bind(Some(claims.raw.clone()))
+            .bind(now)
+            .execute(&pool)
+            .await
+            .map_err(repo_sql_error)?;
+        return map_user_auth_row(&row).map_err(repo_data_error);
+    }
+
+    let email = normalize_identity_email(claims.email.as_deref());
+    if let Some(email) = email.as_deref() {
+        if let Some(row) = sqlx::query(FIND_USER_BY_EMAIL_SQL)
+            .bind(email)
+            .fetch_optional(&pool)
+            .await
+            .map_err(repo_sql_error)?
+        {
+            let existing = map_user_auth_row(&row).map_err(repo_data_error)?;
+            return Err(match existing.auth_source.to_ascii_lowercase().as_str() {
+                "local" => IdentityOAuthAccountError::EmailExistsLocal,
+                "ldap" => IdentityOAuthAccountError::EmailIsLdap,
+                _ => IdentityOAuthAccountError::EmailIsOauth,
+            });
+        }
+    }
+
+    let registration_enabled = state
+        .read_system_config_json_value("enable_registration")
+        .await
+        .map_err(|err| IdentityOAuthAccountError::Storage(format!("{err:?}")))?
+        .as_ref()
+        .map(system_config_bool)
+        .unwrap_or(false);
+    if !registration_enabled {
+        return Err(IdentityOAuthAccountError::RegistrationDisabled);
+    }
+
+    let initial_gift = state
+        .read_system_config_json_value("default_user_initial_gift_usd")
+        .await
+        .map_err(|err| IdentityOAuthAccountError::Storage(format!("{err:?}")))?
+        .as_ref()
+        .map(|value| system_config_f64(value, 10.0))
+        .unwrap_or(10.0);
+
+    let mut tx = pool.begin().await.map_err(repo_sql_error)?;
+    let username = unique_oauth_username(&mut tx, claims).await?;
+    let user_id = Uuid::new_v4().to_string();
+    let row = sqlx::query(CREATE_OAUTH_USER_SQL)
+        .bind(&user_id)
+        .bind(email.as_deref())
+        .bind(&username)
+        .bind(now)
+        .fetch_one(&mut *tx)
+        .await
+        .map_err(repo_sql_error)?;
+    let user = map_user_auth_row(&row).map_err(repo_data_error)?;
+
+    create_initial_wallet_in_tx(&mut tx, &user.id, initial_gift).await?;
+    upsert_oauth_link_in_tx(&mut tx, &user.id, claims, now).await?;
+    tx.commit().await.map_err(repo_sql_error)?;
+    Ok(user)
+}
+
+pub(crate) async fn bind_identity_oauth_to_user(
+    state: &AppState,
+    user: &StoredUserAuthRecord,
+    claims: &IdentityClaims,
+) -> Result<(), IdentityOAuthAccountError> {
+    if user.auth_source.eq_ignore_ascii_case("ldap") {
+        return Err(IdentityOAuthAccountError::EmailIsLdap);
+    }
+    let Some(pool) = state.postgres_pool() else {
+        return Err(IdentityOAuthAccountError::ProviderUnavailable);
+    };
+    if let Some(row) = sqlx::query(FIND_OAUTH_LINK_OWNER_SQL)
+        .bind(&claims.provider_type)
+        .bind(&claims.subject)
+        .fetch_optional(&pool)
+        .await
+        .map_err(repo_sql_error)?
+    {
+        let owner: String = row.try_get("user_id").map_err(repo_sql_error)?;
+        if owner != user.id {
+            return Err(IdentityOAuthAccountError::OAuthAlreadyBound);
+        }
+    }
+    if sqlx::query(FIND_USER_PROVIDER_LINK_OWNER_SQL)
+        .bind(&user.id)
+        .bind(&claims.provider_type)
+        .fetch_optional(&pool)
+        .await
+        .map_err(repo_sql_error)?
+        .is_some()
+    {
+        return Err(IdentityOAuthAccountError::AlreadyBoundProvider);
+    }
+    let mut tx = pool.begin().await.map_err(repo_sql_error)?;
+    upsert_oauth_link_in_tx(&mut tx, &user.id, claims, Utc::now()).await?;
+    tx.commit().await.map_err(repo_sql_error)?;
+    Ok(())
+}
+
+pub(crate) async fn unbind_identity_oauth(
+    state: &AppState,
+    user: &StoredUserAuthRecord,
+    provider_type: &str,
+) -> Result<bool, IdentityOAuthAccountError> {
+    if user.auth_source.eq_ignore_ascii_case("ldap") {
+        return Err(IdentityOAuthAccountError::EmailIsLdap);
+    }
+    let Some(pool) = state.postgres_pool() else {
+        return Err(IdentityOAuthAccountError::ProviderUnavailable);
+    };
+    let row = sqlx::query(COUNT_USER_OAUTH_LINKS_SQL)
+        .bind(&user.id)
+        .fetch_one(&pool)
+        .await
+        .map_err(repo_sql_error)?;
+    let link_count: i64 = row.try_get("link_count").map_err(repo_sql_error)?;
+    if user.auth_source.eq_ignore_ascii_case("oauth") && link_count <= 1 {
+        return Err(IdentityOAuthAccountError::LastOAuthBinding);
+    }
+    if !user.auth_source.eq_ignore_ascii_case("local") && link_count <= 1 {
+        return Err(IdentityOAuthAccountError::LastLoginMethod);
+    }
+    let result = sqlx::query(DELETE_USER_OAUTH_LINK_SQL)
+        .bind(&user.id)
+        .bind(provider_type.trim())
+        .execute(&pool)
+        .await
+        .map_err(repo_sql_error)?;
+    Ok(result.rows_affected() > 0)
+}
+
+fn stored_provider_config_to_identity_config(
+    state: &AppState,
+    config: StoredOAuthProviderConfig,
+) -> Result<IdentityOAuthProviderConfig, IdentityOAuthAccountError> {
+    let defaults = identity_provider_defaults(&config.provider_type);
+    let authorization_url = config
+        .authorization_url_override
+        .clone()
+        .or_else(|| defaults.map(|defaults| defaults.0.to_string()))
+        .filter(|value| !value.trim().is_empty())
+        .ok_or(IdentityOAuthAccountError::ProviderUnavailable)?;
+    let token_url = config
+        .token_url_override
+        .clone()
+        .or_else(|| defaults.map(|defaults| defaults.1.to_string()))
+        .filter(|value| !value.trim().is_empty())
+        .ok_or(IdentityOAuthAccountError::ProviderUnavailable)?;
+    let userinfo_url = config
+        .userinfo_url_override
+        .clone()
+        .or_else(|| defaults.map(|defaults| defaults.2.to_string()));
+    let client_secret = match config.client_secret_encrypted.as_deref() {
+        Some(ciphertext) => Some(
+            decrypt_catalog_secret_with_fallbacks(state.encryption_key(), ciphertext)
+                .ok_or(IdentityOAuthAccountError::ProviderUnavailable)?,
+        ),
+        None => None,
+    };
+
+    Ok(IdentityOAuthProviderConfig {
+        provider_type: config.provider_type,
+        display_name: config.display_name,
+        authorization_url,
+        token_url,
+        userinfo_url,
+        client_id: config.client_id,
+        client_secret,
+        scopes: config.scopes.unwrap_or_default(),
+        redirect_uri: config.redirect_uri,
+        frontend_callback_url: config.frontend_callback_url,
+        attribute_mapping: config.attribute_mapping,
+        extra_config: config.extra_config,
+    })
+}
+
+fn identity_provider_defaults(
+    provider_type: &str,
+) -> Option<(&'static str, &'static str, &'static str)> {
+    match provider_type.trim().to_ascii_lowercase().as_str() {
+        "linuxdo" => Some((
+            LINUXDO_AUTHORIZE_URL,
+            LINUXDO_TOKEN_URL,
+            LINUXDO_USERINFO_URL,
+        )),
+        _ => None,
+    }
+}
+
+fn map_link_summary_row(
+    row: &sqlx::postgres::PgRow,
+) -> Result<IdentityOAuthLinkSummary, GatewayError> {
+    Ok(IdentityOAuthLinkSummary {
+        provider_type: row.try_get("provider_type").map_err(sql_gateway_error)?,
+        display_name: row.try_get("display_name").map_err(sql_gateway_error)?,
+        provider_username: row
+            .try_get("provider_username")
+            .map_err(sql_gateway_error)?,
+        provider_email: row.try_get("provider_email").map_err(sql_gateway_error)?,
+        linked_at: row
+            .try_get::<Option<DateTime<Utc>>, _>("linked_at")
+            .map_err(sql_gateway_error)?
+            .map(|value| value.to_rfc3339()),
+        last_login_at: row
+            .try_get::<Option<DateTime<Utc>>, _>("last_login_at")
+            .map_err(sql_gateway_error)?
+            .map(|value| value.to_rfc3339()),
+        provider_enabled: row.try_get("provider_enabled").map_err(sql_gateway_error)?,
+    })
+}
+
+fn map_user_auth_row(
+    row: &sqlx::postgres::PgRow,
+) -> Result<StoredUserAuthRecord, aether_data::DataLayerError> {
+    StoredUserAuthRecord::new(
+        row.try_get("id").map_err(data_unexpected)?,
+        row.try_get("email").map_err(data_unexpected)?,
+        row.try_get("email_verified").map_err(data_unexpected)?,
+        row.try_get("username").map_err(data_unexpected)?,
+        row.try_get("password_hash").map_err(data_unexpected)?,
+        row.try_get("role").map_err(data_unexpected)?,
+        row.try_get("auth_source").map_err(data_unexpected)?,
+        row.try_get("allowed_providers").map_err(data_unexpected)?,
+        row.try_get("allowed_api_formats")
+            .map_err(data_unexpected)?,
+        row.try_get("allowed_models").map_err(data_unexpected)?,
+        row.try_get("is_active").map_err(data_unexpected)?,
+        row.try_get("is_deleted").map_err(data_unexpected)?,
+        row.try_get("created_at").map_err(data_unexpected)?,
+        row.try_get("last_login_at").map_err(data_unexpected)?,
+    )
+}
+
+async fn unique_oauth_username(
+    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
+    claims: &IdentityClaims,
+) -> Result<String, IdentityOAuthAccountError> {
+    let base = normalize_oauth_username(
+        claims
+            .username
+            .as_deref()
+            .or(claims.display_name.as_deref())
+            .or_else(|| {
+                claims
+                    .email
+                    .as_deref()
+                    .and_then(|email| email.split('@').next())
+            })
+            .unwrap_or("oauth_user"),
+    );
+    for attempt in 0..8 {
+        let candidate = if attempt == 0 {
+            base.clone()
+        } else {
+            format!(
+                "{}_{}",
+                base.chars().take(20).collect::<String>(),
+                short_uuid()
+            )
+        };
+        let taken = sqlx::query(CHECK_USERNAME_TAKEN_SQL)
+            .bind(&candidate)
+            .fetch_optional(&mut **tx)
+            .await
+            .map_err(repo_sql_error)?
+            .is_some();
+        if !taken {
+            return Ok(candidate);
+        }
+    }
+    Ok(format!("oauth_{}", short_uuid()))
+}
+
+async fn upsert_oauth_link_in_tx(
+    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
+    user_id: &str,
+    claims: &IdentityClaims,
+    now: DateTime<Utc>,
+) -> Result<(), IdentityOAuthAccountError> {
+    sqlx::query(UPSERT_OAUTH_LINK_SQL)
+        .bind(Uuid::new_v4().to_string())
+        .bind(user_id)
+        .bind(&claims.provider_type)
+        .bind(&claims.subject)
+        .bind(claims.username.as_deref())
+        .bind(claims.email.as_deref())
+        .bind(Some(claims.raw.clone()))
+        .bind(now)
+        .execute(&mut **tx)
+        .await
+        .map_err(repo_sql_error)?;
+    Ok(())
+}
+
+async fn create_initial_wallet_in_tx(
+    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
+    user_id: &str,
+    initial_gift_usd: f64,
+) -> Result<(), IdentityOAuthAccountError> {
+    let gift_amount = initial_gift_usd.max(0.0);
+    let wallet_id = Uuid::new_v4().to_string();
+    sqlx::query(CREATE_AUTH_USER_WALLET_SQL)
+        .bind(&wallet_id)
+        .bind(user_id)
+        .bind(gift_amount)
+        .bind("finite")
+        .execute(&mut **tx)
+        .await
+        .map_err(repo_sql_error)?;
+    if gift_amount > 0.0 {
+        sqlx::query(CREATE_AUTH_USER_WALLET_GIFT_TX_SQL)
+            .bind(Uuid::new_v4().to_string())
+            .bind(&wallet_id)
+            .bind(gift_amount)
+            .bind(user_id)
+            .execute(&mut **tx)
+            .await
+            .map_err(repo_sql_error)?;
+    }
+    Ok(())
+}
+
+fn normalize_identity_email(value: Option<&str>) -> Option<String> {
+    value
+        .map(str::trim)
+        .filter(|value| !value.is_empty())
+        .map(str::to_ascii_lowercase)
+}
+
+fn normalize_oauth_username(value: &str) -> String {
+    let mut normalized = value
+        .chars()
+        .map(|ch| {
+            if ch.is_ascii_alphanumeric() || matches!(ch, '_' | '-' | '.') {
+                ch
+            } else {
+                '_'
+            }
+        })
+        .collect::<String>();
+    while normalized.contains("__") {
+        normalized = normalized.replace("__", "_");
+    }
+    normalized = normalized
+        .trim_matches(|ch| matches!(ch, '_' | '-' | '.'))
+        .chars()
+        .take(30)
+        .collect();
+    if normalized.len() < 3 || is_reserved_username(&normalized) {
+        normalized = format!("oauth_{}", short_uuid());
+    }
+    normalized
+}
+
+fn is_reserved_username(value: &str) -> bool {
+    matches!(
+        value.to_ascii_lowercase().as_str(),
+        "admin" | "root" | "system" | "api" | "test" | "demo" | "user" | "guest" | "bot"
+    )
+}
+
+fn short_uuid() -> String {
+    Uuid::new_v4().simple().to_string()[..8].to_string()
+}
+
+fn system_config_bool(value: &Value) -> bool {
+    match value {
+        Value::Bool(value) => *value,
+        Value::Number(value) => value.as_i64().is_some_and(|value| value != 0),
+        Value::String(value) => matches!(
+            value.trim().to_ascii_lowercase().as_str(),
+            "1" | "true" | "yes" | "on"
+        ),
+        _ => false,
+    }
+}
+
+fn system_config_f64(value: &Value, default: f64) -> f64 {
+    match value {
+        Value::Number(value) => value.as_f64().unwrap_or(default),
+        Value::String(value) => value.trim().parse::<f64>().unwrap_or(default),
+        _ => default,
+    }
+}
+
+fn repo_sql_error(error: sqlx::Error) -> IdentityOAuthAccountError {
+    IdentityOAuthAccountError::Storage(error.to_string())
+}
+
+fn repo_data_error(error: aether_data::DataLayerError) -> IdentityOAuthAccountError {
+    IdentityOAuthAccountError::Storage(error.to_string())
+}
+
+fn sql_gateway_error(error: sqlx::Error) -> GatewayError {
+    GatewayError::Internal(error.to_string())
+}
+
+fn data_unexpected(error: sqlx::Error) -> aether_data::DataLayerError {
+    aether_data::DataLayerError::UnexpectedValue(error.to_string())
+}

apps/aether-gateway/src/oauth/mod.rs

@@ -0,0 +1,21 @@
+mod http_executor;
+mod identity_repo;
+mod provider_repo;
+mod proxy;
+mod state_store;
+
+pub(crate) use http_executor::GatewayOAuthHttpExecutor;
+pub(crate) use identity_repo::{
+    bind_identity_oauth_to_user, get_enabled_identity_oauth_provider_config,
+    list_bindable_identity_oauth_providers, list_enabled_identity_oauth_providers,
+    list_identity_oauth_links, resolve_identity_oauth_login_user, unbind_identity_oauth,
+    IdentityOAuthAccountError,
+};
+pub(crate) use provider_repo::ProviderOAuthRepository;
+pub(crate) use proxy::{
+    resolve_identity_oauth_network_context, resolve_provider_oauth_operation_proxy_snapshot,
+};
+pub(crate) use state_store::{
+    consume_identity_oauth_state, save_identity_oauth_state, IdentityOAuthStateMode,
+    StoredIdentityOAuthState,
+};

apps/aether-gateway/src/oauth/provider_repo.rs

@@ -0,0 +1,122 @@
+use crate::admin_api::{
+    create_provider_oauth_catalog_key, find_duplicate_provider_oauth_key,
+    refresh_provider_oauth_account_state_after_update, update_existing_provider_oauth_catalog_key,
+    AdminAppState, AdminGatewayProviderTransportSnapshot, AdminLocalOAuthRefreshError,
+};
+use crate::GatewayError;
+use aether_contracts::ProxySnapshot;
+use aether_data_contracts::repository::provider_catalog::{
+    StoredProviderCatalogKey, StoredProviderCatalogProvider,
+};
+
+#[derive(Debug, Clone, Copy, Default)]
+pub(crate) struct ProviderOAuthRepository;
+
+impl ProviderOAuthRepository {
+    pub(crate) async fn update_provider_catalog_key_oauth_credentials(
+        state: &AdminAppState<'_>,
+        key_id: &str,
+        encrypted_api_key: &str,
+        encrypted_auth_config: Option<&str>,
+        expires_at_unix_secs: Option<u64>,
+    ) -> Result<bool, GatewayError> {
+        state
+            .app()
+            .update_provider_catalog_key_oauth_credentials(
+                key_id,
+                encrypted_api_key,
+                encrypted_auth_config,
+                expires_at_unix_secs,
+            )
+            .await
+    }
+
+    pub(crate) async fn clear_provider_catalog_key_oauth_invalid_marker(
+        state: &AdminAppState<'_>,
+        key_id: &str,
+    ) -> Result<bool, GatewayError> {
+        state
+            .app()
+            .clear_provider_catalog_key_oauth_invalid_marker(key_id)
+            .await
+    }
+
+    pub(crate) async fn force_local_oauth_refresh_entry(
+        state: &AdminAppState<'_>,
+        transport: &AdminGatewayProviderTransportSnapshot,
+    ) -> Result<Option<crate::provider_transport::CachedOAuthEntry>, AdminLocalOAuthRefreshError>
+    {
+        state.app().force_local_oauth_refresh_entry(transport).await
+    }
+
+    pub(crate) async fn find_duplicate_provider_oauth_key(
+        state: &AdminAppState<'_>,
+        provider_id: &str,
+        auth_config: &serde_json::Map<String, serde_json::Value>,
+        exclude_key_id: Option<&str>,
+    ) -> Result<Option<StoredProviderCatalogKey>, String> {
+        find_duplicate_provider_oauth_key(state, provider_id, auth_config, exclude_key_id).await
+    }
+
+    pub(crate) async fn create_provider_oauth_catalog_key(
+        state: &AdminAppState<'_>,
+        provider_id: &str,
+        provider_type: &str,
+        name: &str,
+        access_token: &str,
+        auth_config: &serde_json::Map<String, serde_json::Value>,
+        api_formats: &[String],
+        proxy: Option<serde_json::Value>,
+        expires_at_unix_secs: Option<u64>,
+    ) -> Result<Option<StoredProviderCatalogKey>, GatewayError> {
+        create_provider_oauth_catalog_key(
+            state,
+            provider_id,
+            provider_type,
+            name,
+            access_token,
+            auth_config,
+            api_formats,
+            proxy,
+            expires_at_unix_secs,
+        )
+        .await
+    }
+
+    pub(crate) async fn update_existing_provider_oauth_catalog_key(
+        state: &AdminAppState<'_>,
+        existing_key: &StoredProviderCatalogKey,
+        provider_type: &str,
+        access_token: &str,
+        auth_config: &serde_json::Map<String, serde_json::Value>,
+        api_formats: &[String],
+        proxy: Option<serde_json::Value>,
+        expires_at_unix_secs: Option<u64>,
+    ) -> Result<Option<StoredProviderCatalogKey>, GatewayError> {
+        update_existing_provider_oauth_catalog_key(
+            state,
+            existing_key,
+            provider_type,
+            access_token,
+            auth_config,
+            api_formats,
+            proxy,
+            expires_at_unix_secs,
+        )
+        .await
+    }
+
+    pub(crate) async fn refresh_provider_oauth_account_state_after_update(
+        state: &AdminAppState<'_>,
+        provider: &StoredProviderCatalogProvider,
+        key_id: &str,
+        proxy_override: Option<&ProxySnapshot>,
+    ) -> Result<(bool, Option<String>), GatewayError> {
+        refresh_provider_oauth_account_state_after_update(state, provider, key_id, proxy_override)
+            .await
+    }
+
+    pub(crate) fn clear_transport_cache_after_write(state: &AdminAppState<'_>) {
+        state.app().clear_provider_transport_snapshot_cache();
+    }
+}

apps/aether-gateway/src/oauth/proxy.rs

@@ -0,0 +1,44 @@
+use crate::admin_api::AdminAppState;
+use crate::AppState;
+use aether_contracts::ProxySnapshot;
+use aether_oauth::network::{OAuthNetworkContext, OAuthNetworkPolicy, OAuthTimeouts};
+
+pub(crate) async fn resolve_identity_oauth_network_context(
+    state: &AppState,
+) -> OAuthNetworkContext {
+    let proxy = state.resolve_system_proxy_snapshot().await;
+    OAuthNetworkContext {
+        policy: OAuthNetworkPolicy::DirectOrSystemProxy,
+        requirement: aether_oauth::network::NetworkRequirement::Optional,
+        timeouts: if proxy.is_some() {
+            OAuthTimeouts::PROXY_DEFAULT
+        } else {
+            OAuthTimeouts::DIRECT_DEFAULT
+        },
+        proxy,
+    }
+}
+
+pub(crate) async fn resolve_provider_oauth_operation_proxy_snapshot(
+    state: &AdminAppState<'_>,
+    temporary_proxy_node_id: Option<&str>,
+    configured_proxies: &[Option<&serde_json::Value>],
+) -> Option<ProxySnapshot> {
+    if let Some(snapshot) = state
+        .resolve_admin_proxy_node_snapshot(temporary_proxy_node_id)
+        .await
+    {
+        return Some(snapshot);
+    }
+
+    for proxy in configured_proxies {
+        if let Some(snapshot) = state
+            .app()
+            .resolve_configured_proxy_snapshot_with_tunnel_affinity(*proxy)
+            .await
+        {
+            return Some(snapshot);
+        }
+    }
+    state.app().resolve_system_proxy_snapshot().await
+}

apps/aether-gateway/src/oauth/state_store.rs

@@ -0,0 +1,118 @@
+use crate::{AppState, GatewayError};
+use aether_oauth::core::{current_unix_secs, generate_oauth_nonce};
+use serde::{Deserialize, Serialize};
+
+const IDENTITY_OAUTH_STATE_TTL_SECS: u64 = 10 * 60;
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "snake_case")]
+pub(crate) enum IdentityOAuthStateMode {
+    Login,
+    Bind,
+}
+
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+pub(crate) struct StoredIdentityOAuthState {
+    pub(crate) nonce: String,
+    pub(crate) provider_type: String,
+    pub(crate) mode: IdentityOAuthStateMode,
+    pub(crate) client_device_id: String,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub(crate) pkce_verifier: Option<String>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub(crate) bind_user_id: Option<String>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub(crate) bind_session_id: Option<String>,
+    pub(crate) created_at: u64,
+}
+
+impl StoredIdentityOAuthState {
+    pub(crate) fn login(
+        provider_type: impl Into<String>,
+        client_device_id: impl Into<String>,
+        pkce_verifier: Option<String>,
+    ) -> Self {
+        Self {
+            nonce: generate_oauth_nonce(),
+            provider_type: provider_type.into(),
+            mode: IdentityOAuthStateMode::Login,
+            client_device_id: client_device_id.into(),
+            pkce_verifier,
+            bind_user_id: None,
+            bind_session_id: None,
+            created_at: current_unix_secs(),
+        }
+    }
+
+    pub(crate) fn bind(
+        provider_type: impl Into<String>,
+        client_device_id: impl Into<String>,
+        pkce_verifier: Option<String>,
+        user_id: impl Into<String>,
+        session_id: impl Into<String>,
+    ) -> Self {
+        Self {
+            nonce: generate_oauth_nonce(),
+            provider_type: provider_type.into(),
+            mode: IdentityOAuthStateMode::Bind,
+            client_device_id: client_device_id.into(),
+            pkce_verifier,
+            bind_user_id: Some(user_id.into()),
+            bind_session_id: Some(session_id.into()),
+            created_at: current_unix_secs(),
+        }
+    }
+}
+
+pub(crate) fn identity_oauth_state_storage_key(nonce: &str) -> String {
+    format!("identity_oauth_state:{}", nonce.trim())
+}
+
+pub(crate) async fn save_identity_oauth_state(
+    state: &AppState,
+    record: &StoredIdentityOAuthState,
+) -> Result<(), GatewayError> {
+    let key = identity_oauth_state_storage_key(&record.nonce);
+    let value =
+        serde_json::to_string(record).map_err(|err| GatewayError::Internal(err.to_string()))?;
+    if let Some(runner) = state.redis_kv_runner() {
+        runner
+            .setex(&key, &value, Some(IDENTITY_OAUTH_STATE_TTL_SECS))
+            .await
+            .map_err(|err| GatewayError::Internal(err.to_string()))?;
+        return Ok(());
+    }
+    if state.save_provider_oauth_state_for_tests(&key, &value) {
+        return Ok(());
+    }
+    Err(GatewayError::Internal(
+        "identity oauth state store unavailable".to_string(),
+    ))
+}
+
+pub(crate) async fn consume_identity_oauth_state(
+    state: &AppState,
+    nonce: &str,
+) -> Result<Option<StoredIdentityOAuthState>, GatewayError> {
+    let key = identity_oauth_state_storage_key(nonce);
+    let raw = if let Some(runner) = state.redis_kv_runner() {
+        let mut connection = runner
+            .client()
+            .get_multiplexed_async_connection()
+            .await
+            .map_err(|err| GatewayError::Internal(err.to_string()))?;
+        let namespaced_key = runner.keyspace().key(&key);
+        redis::cmd("GETDEL")
+            .arg(&namespaced_key)
+            .query_async::<Option<String>>(&mut connection)
+            .await
+            .map_err(|err| GatewayError::Internal(err.to_string()))?
+    } else {
+        state.take_provider_oauth_state_for_tests(&key)
+    };
+    raw.map(|value| {
+        serde_json::from_str::<StoredIdentityOAuthState>(&value)
+            .map_err(|err| GatewayError::Internal(err.to_string()))
+    })
+    .transpose()
+}