sansmoraxz commented on code in PR #4055: URL: https://github.com/apache/iggy/pull/4055#discussion_r3982835381
########## core/server/src/external_auth.rs: ########## @@ -0,0 +1,495 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +//! External authentication callout. +//! +//! When enabled, the server POSTs credential and connection metadata to an +//! external HTTP service during login. The service decides whether to grant +//! access (with inline permissions or by mapping to an existing Iggy user) +//! or deny it. This module owns the request/response types, the HTTP +//! callout, and the session-scoped permission carrier. + +use std::fmt; +use std::sync::Arc; +use std::sync::atomic::{AtomicU32, Ordering}; + +use configs::external_auth::{ExternalAuthConfig, ExternalAuthErrorStrategy}; +use iggy_common::Permissions; +use serde::{Deserialize, Serialize}; +use tracing::warn; + +const MAX_RESPONSE_BODY_BYTES: usize = 1_048_576; + +thread_local! { + static HTTP_CLIENT: cyper::Client = + cyper::Client::new().expect("failed to build cyper HTTP client for external auth"); +} + +fn get_http_client() -> cyper::Client { + HTTP_CLIENT.with(cyper::Client::clone) +} + +/// Credential metadata sent to the external auth service. +/// +/// Manual `Debug` redacts the `credential` field so passwords and tokens +/// never appear in log output. +#[derive(Serialize)] +pub struct ExternalAuthRequest { + pub credential_type: CredentialType, + #[serde(skip_serializing_if = "Option::is_none")] + pub credential: Option<String>, + pub username: String, + pub transport: String, + pub client_address: String, +} + +impl fmt::Debug for ExternalAuthRequest { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("ExternalAuthRequest") + .field("credential_type", &self.credential_type) + .field( + "credential", + &self.credential.as_ref().map(|_| "[REDACTED]"), + ) + .field("username", &self.username) + .field("transport", &self.transport) + .field("client_address", &self.client_address) + .finish() + } +} + +#[derive(Debug, Serialize, Deserialize, Clone, Copy, PartialEq, Eq)] +#[serde(rename_all = "snake_case")] +pub enum CredentialType { + Password, + PersonalAccessToken, +} + +/// JSON response from the external auth service. +#[derive(Debug, Deserialize)] +struct ExternalAuthResponse { + decision: DecisionTag, + user_id: Option<u32>, + principal: Option<String>, + permissions: Option<Permissions>, + expires_at: Option<u64>, + reason: Option<String>, +} + +#[derive(Debug, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "snake_case")] +enum DecisionTag { + IggyUser, + InlineGrant, + Deny, +} + +/// Parsed decision from the external auth service. +#[derive(Debug)] +pub enum ExternalAuthDecision { + IggyUser { + user_id: u32, + }, + InlineGrant { + principal: String, + permissions: Permissions, + expires_at: u64, + }, + Deny { + reason: String, + }, +} + +/// Callout failure (network, timeout, bad response). +#[derive(Debug)] +pub enum ExternalAuthError { + HttpError(String), + Timeout, + BadResponse(String), +} + +impl fmt::Display for ExternalAuthError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::HttpError(msg) => write!(f, "external auth HTTP error: {msg}"), + Self::Timeout => write!(f, "external auth callout timed out"), + Self::BadResponse(msg) => write!(f, "external auth bad response: {msg}"), + } + } +} + +impl std::error::Error for ExternalAuthError {} + +/// # Errors +/// +/// Returns [`ServerError::InvalidExternalAuthConfig`](crate::server_error::ServerError::InvalidExternalAuthConfig) +/// when the URL is empty or uses an unsupported scheme. +pub fn validate_config( + config: &ExternalAuthConfig, +) -> Result<(), crate::server_error::ServerError> { + if !config.enabled { + return Ok(()); + } + if config.url.is_empty() { + return Err( + crate::server_error::ServerError::InvalidExternalAuthConfig { + reason: "external_auth.url must be set when external_auth.enabled = true" + .to_owned(), + }, + ); + } + if !config.url.starts_with("http://";) && !config.url.starts_with("https://";) { + return Err( + crate::server_error::ServerError::InvalidExternalAuthConfig { + reason: format!( + "external_auth.url must start with http:// or https://, got: {}", + config.url + ), + }, + ); + } + Ok(()) +} + +pub fn warn_insecure_url(config: &ExternalAuthConfig) { + if config.enabled && config.url.starts_with("http://";) { + tracing::warn!( + url = config.url, + "external auth URL uses plain HTTP; credentials will be sent in cleartext" + ); + } +} + +/// Session-scoped permissions from an external auth inline grant. +/// Carried on the connection/session, never persisted. +#[derive(Debug, Clone)] +pub struct SessionPermissions { + pub principal: String, + pub permissions: Permissions, + pub expires_at: u64, +} + +pub use iggy_common::{SYNTHETIC_USER_ID_THRESHOLD, is_synthetic_user_id}; + +/// Process-wide counter for minting synthetic user IDs. Wraps an +/// `Arc<AtomicU32>` so every transport (TCP, QUIC, WS, HTTP) draws from +/// the same sequence and no two transports can mint the same ID. +#[derive(Clone)] +pub struct SyntheticUserIdCounter(Arc<AtomicU32>); + +impl SyntheticUserIdCounter { + #[must_use] + pub fn new() -> Self { + Self(Arc::new(AtomicU32::new(u32::MAX))) + } + + #[must_use] + pub fn mint(&self) -> Option<u32> { + loop { + let current = self.0.load(Ordering::Relaxed); + if !is_synthetic_user_id(current) { + return None; + } + if self + .0 + .compare_exchange_weak(current, current - 1, Ordering::Relaxed, Ordering::Relaxed) + .is_ok() + { + return Some(current); + } + } + } +} + +impl Default for SyntheticUserIdCounter { + fn default() -> Self { + Self::new() + } +} + +/// Call the external auth service and parse the response. +/// +/// # Errors +/// +/// Returns [`ExternalAuthError`] on network/timeout/parse failure. The +/// caller applies the configured `on_error` strategy. +pub async fn callout_external_auth( + config: &ExternalAuthConfig, + request: ExternalAuthRequest, +) -> Result<ExternalAuthDecision, ExternalAuthError> { + let client = get_http_client(); + let timeout = config.timeout.get_duration(); + + let body = serde_json::to_vec(&request) + .map_err(|e| ExternalAuthError::BadResponse(format!("failed to serialize request: {e}")))?; + + let build = || -> Result<_, ExternalAuthError> { + Ok(client + .post(&config.url) + .map_err(|e| ExternalAuthError::HttpError(format!("failed to build request: {e}")))? + .header("content-type", "application/json") + .map_err(|e| ExternalAuthError::HttpError(format!("failed to set header: {e}")))? + .body(body)) + }; + let request_builder = build()?; + + let response = compio::time::timeout(timeout, request_builder.send()) Review Comment: using a single round_trip future object. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
