hubcio commented on code in PR #4055: URL: https://github.com/apache/iggy/pull/4055#discussion_r3952545168
########## 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"); Review Comment: critical: `cyper::Client::new()` follows redirects by default and resends the body on 307/308, so the plaintext credential goes to whatever host `Location` names and that host's reply becomes the auth decision. use `.redirect(Policy::none())`, same as `http/forward.rs:159`. ########## 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()) + .await + .map_err(|_| ExternalAuthError::Timeout)? + .map_err(|e| ExternalAuthError::HttpError(e.to_string()))?; + + let status = response.status(); + if !status.is_success() { + return Err(ExternalAuthError::HttpError(format!( + "non-success status: {status}" + ))); + } + + if let Some(len) = response + .headers() + .get("content-length") + .and_then(|v| v.to_str().ok()) + .and_then(|s| s.parse::<usize>().ok()) + && len > MAX_RESPONSE_BODY_BYTES + { + return Err(ExternalAuthError::BadResponse( + "response body too large".to_owned(), + )); + } + + let bytes = compio::time::timeout(timeout, response.bytes()) Review Comment: critical: `bytes()` buffers the whole body before the 1 MiB check below, and the content-length precheck is skipped on a chunked reply. there is no actual cap, though config.toml:1101 says there is. ########## 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: warning: the timeout wraps `send()` here and `bytes()` again below, so one login can take twice the configured budget. use a single deadline across both. ########## core/server/src/http/state.rs: ########## @@ -382,6 +426,15 @@ impl HttpInner { /// request re-register cleanly through the barrier. pub(in crate::http) fn forget_session(&self, session: &Rc<HttpSession>) { let torn = forget_if_same(&mut self.sessions.borrow_mut(), session); + if torn.is_some() Review Comment: critical: `forget_if_same` returns `None` when `registry_token` is unset even though it already removed the table entry, so this cleanup is skipped. any session that never did an acked produce leaks its permissions entry and burns an id. ########## core/server/src/dispatch/session_ops.rs: ########## @@ -1309,6 +1312,192 @@ pub(in crate::dispatch) async fn handle_login_register_request<B, MJ, S, SB>( } let body_tail = &body[prefix_len..]; + + if external_auth.enabled { + let ext_request = if let Ok((wire_request, _)) = + LoginRegisterRequest::decode_after_prefix(version_info.clone(), body_tail) + { + crate::external_auth::ExternalAuthRequest { + credential_type: crate::external_auth::CredentialType::Password, + credential: if external_auth.forward_credentials { + Some(wire_request.password.expose_secret().to_owned()) + } else { + None + }, + username: wire_request.username.to_string(), + transport: "binary".to_owned(), Review Comment: warning: `transport` is hardcoded `"binary"` for tcp, quic and websocket, but the documented contract is `tcp`/`quic`/`websocket`. the real kind sits on the connection - `get_connection()` was added for this and never called. ########## core/server/src/http/state.rs: ########## @@ -382,6 +426,15 @@ impl HttpInner { /// request re-register cleanly through the barrier. pub(in crate::http) fn forget_session(&self, session: &Rc<HttpSession>) { let torn = forget_if_same(&mut self.sessions.borrow_mut(), session); + if torn.is_some() + && self + .synthetic_permissions + .borrow_mut() + .remove(&session.user_id) + .is_some() + { + self.free_synthetic_ids.borrow_mut().insert(session.user_id); Review Comment: critical: the synthetic id goes back on the free list while its jwt is still valid, and there is no revocation list. the next inline grant gets that id, and the old token then authorizes as the new principal. ########## core/server/src/dispatch/session_ops.rs: ########## @@ -1309,6 +1312,192 @@ pub(in crate::dispatch) async fn handle_login_register_request<B, MJ, S, SB>( } let body_tail = &body[prefix_len..]; + + if external_auth.enabled { + let ext_request = if let Ok((wire_request, _)) = + LoginRegisterRequest::decode_after_prefix(version_info.clone(), body_tail) + { + crate::external_auth::ExternalAuthRequest { + credential_type: crate::external_auth::CredentialType::Password, + credential: if external_auth.forward_credentials { + Some(wire_request.password.expose_secret().to_owned()) + } else { + None + }, + username: wire_request.username.to_string(), + transport: "binary".to_owned(), + client_address: sessions + .borrow() + .connection_address(transport_client_id) + .map_or_else(String::new, |a| a.to_string()), + } + } else if let Ok((wire_request, _)) = + LoginRegisterWithPatRequest::decode_after_prefix(version_info.clone(), body_tail) + { + crate::external_auth::ExternalAuthRequest { + credential_type: crate::external_auth::CredentialType::PersonalAccessToken, + credential: if external_auth.forward_credentials { + Some(wire_request.token.expose_secret().to_owned()) + } else { + None + }, + username: String::new(), + transport: "binary".to_owned(), + client_address: sessions + .borrow() + .connection_address(transport_client_id) + .map_or_else(String::new, |a| a.to_string()), + } + } else { + warn!( + transport_client_id, + "rejecting register request with unsupported payload shape" + ); + send_login_eviction( + shard, + transport_client_id, + vsr_client_id, + EvictionReason::MalformedLogin, + ) + .await; + return; + }; + + match crate::external_auth::try_external_auth(external_auth, ext_request).await { + Ok(Some(crate::external_auth::ExternalAuthDecision::IggyUser { user_id })) => { + if user_id == 0 { + warn!( + transport_client_id, + "external auth attempted to map login to root user" + ); + send_login_eviction( + shard, + transport_client_id, + vsr_client_id, + EvictionReason::InvalidCredentials, + ) + .await; + return; + } + if crate::external_auth::is_synthetic_user_id(user_id) { + warn!( + transport_client_id, + user_id, "external auth returned synthetic user_id; rejecting" + ); + send_login_eviction( + shard, + transport_client_id, + vsr_client_id, + EvictionReason::InvalidCredentials, + ) + .await; + return; + } + if let Err(error) = complete_login_register( + shard, + sessions, + transport_client_id, + vsr_client_id, + request.header(), + user_id, + &version_info, + ) + .await + { + warn!(transport_client_id, error = %error, "external auth login failed"); + surface_login_failure(shard, transport_client_id, request.header(), &error) + .await; + } + return; + } + Ok(Some(crate::external_auth::ExternalAuthDecision::InlineGrant { + principal, + permissions, + expires_at, + })) => { + let synthetic_id = sessions.borrow_mut().mint_synthetic_user_id(); Review Comment: critical: a second `Register` on a bound connection is accepted, since `dispatch/mod.rs:702` runs before the bound check. it mints an id nothing frees, overwrites `conn.session_permissions`, and leaves the old bound id resolving to the new grant. ########## 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))) Review Comment: critical: this counter is per process but the jwt key is cluster-wide, so two nodes both mint `u32::MAX` first for different principals. a token minted on one node then resolves against the other node's grant. ########## core/server/src/dispatch/session_ops.rs: ########## @@ -1309,6 +1312,192 @@ pub(in crate::dispatch) async fn handle_login_register_request<B, MJ, S, SB>( } let body_tail = &body[prefix_len..]; + + if external_auth.enabled { + let ext_request = if let Ok((wire_request, _)) = + LoginRegisterRequest::decode_after_prefix(version_info.clone(), body_tail) + { + crate::external_auth::ExternalAuthRequest { + credential_type: crate::external_auth::CredentialType::Password, + credential: if external_auth.forward_credentials { + Some(wire_request.password.expose_secret().to_owned()) + } else { + None + }, + username: wire_request.username.to_string(), + transport: "binary".to_owned(), + client_address: sessions + .borrow() + .connection_address(transport_client_id) + .map_or_else(String::new, |a| a.to_string()), + } + } else if let Ok((wire_request, _)) = + LoginRegisterWithPatRequest::decode_after_prefix(version_info.clone(), body_tail) + { + crate::external_auth::ExternalAuthRequest { + credential_type: crate::external_auth::CredentialType::PersonalAccessToken, + credential: if external_auth.forward_credentials { + Some(wire_request.token.expose_secret().to_owned()) + } else { + None + }, + username: String::new(), + transport: "binary".to_owned(), + client_address: sessions + .borrow() + .connection_address(transport_client_id) + .map_or_else(String::new, |a| a.to_string()), + } + } else { + warn!( + transport_client_id, + "rejecting register request with unsupported payload shape" + ); + send_login_eviction( + shard, + transport_client_id, + vsr_client_id, + EvictionReason::MalformedLogin, + ) + .await; + return; + }; + + match crate::external_auth::try_external_auth(external_auth, ext_request).await { + Ok(Some(crate::external_auth::ExternalAuthDecision::IggyUser { user_id })) => { + if user_id == 0 { + warn!( + transport_client_id, + "external auth attempted to map login to root user" + ); + send_login_eviction( + shard, + transport_client_id, + vsr_client_id, + EvictionReason::InvalidCredentials, + ) + .await; + return; + } + if crate::external_auth::is_synthetic_user_id(user_id) { + warn!( + transport_client_id, + user_id, "external auth returned synthetic user_id; rejecting" + ); + send_login_eviction( + shard, + transport_client_id, + vsr_client_id, + EvictionReason::InvalidCredentials, + ) + .await; + return; + } + if let Err(error) = complete_login_register( + shard, + sessions, + transport_client_id, + vsr_client_id, + request.header(), + user_id, + &version_info, + ) + .await + { + warn!(transport_client_id, error = %error, "external auth login failed"); + surface_login_failure(shard, transport_client_id, request.header(), &error) + .await; + } + return; + } + Ok(Some(crate::external_auth::ExternalAuthDecision::InlineGrant { + principal, + permissions, + expires_at, + })) => { + let synthetic_id = sessions.borrow_mut().mint_synthetic_user_id(); + let Some(synthetic_id) = synthetic_id else { + warn!( + transport_client_id, + "synthetic user id pool exhausted; rejecting external auth login" + ); + send_login_eviction( + shard, + transport_client_id, + vsr_client_id, + EvictionReason::InvalidCredentials, + ) + .await; + return; + }; + // Set permissions BEFORE binding the session so that any + // request arriving immediately after bind already sees the + // grants. On bind failure we roll back. + sessions.borrow_mut().set_session_permissions( + transport_client_id, + synthetic_id, + crate::external_auth::SessionPermissions { + principal, + permissions, + expires_at, + }, + ); + if let Err(error) = complete_login_register( + shard, + sessions, + transport_client_id, + vsr_client_id, + request.header(), + synthetic_id, + &version_info, + ) + .await + { + warn!(transport_client_id, error = %error, "external auth inline grant login failed"); + sessions + .borrow_mut() + .clear_session_permissions(transport_client_id, synthetic_id); + surface_login_failure(shard, transport_client_id, request.header(), &error) + .await; + } + return; + } + Ok(Some(crate::external_auth::ExternalAuthDecision::Deny { reason })) => { + warn!( + transport_client_id, + reason = reason, + "external auth denied login" + ); + surface_login_failure( + shard, + transport_client_id, + request.header(), + &LoginRegisterError::ExternalAuthDenied(reason), + ) + .await; + return; + } + Ok(None) => { + // on_error = fallback: fall through to built-in credential check + } + Err(error) => { + warn!( Review Comment: warning: a timeout or a dead auth service reaches the client as `InvalidCredentials`, same as a wrong password, and `fail_sign_in` then wipes remembered credentials. add a non-terminal variant mapped to a transient code. ########## core/server/src/dispatch/session_ops.rs: ########## @@ -1309,6 +1312,192 @@ pub(in crate::dispatch) async fn handle_login_register_request<B, MJ, S, SB>( } let body_tail = &body[prefix_len..]; + + if external_auth.enabled { + let ext_request = if let Ok((wire_request, _)) = + LoginRegisterRequest::decode_after_prefix(version_info.clone(), body_tail) + { + crate::external_auth::ExternalAuthRequest { + credential_type: crate::external_auth::CredentialType::Password, + credential: if external_auth.forward_credentials { + Some(wire_request.password.expose_secret().to_owned()) + } else { + None + }, + username: wire_request.username.to_string(), + transport: "binary".to_owned(), + client_address: sessions + .borrow() + .connection_address(transport_client_id) + .map_or_else(String::new, |a| a.to_string()), + } + } else if let Ok((wire_request, _)) = + LoginRegisterWithPatRequest::decode_after_prefix(version_info.clone(), body_tail) + { + crate::external_auth::ExternalAuthRequest { + credential_type: crate::external_auth::CredentialType::PersonalAccessToken, + credential: if external_auth.forward_credentials { + Some(wire_request.token.expose_secret().to_owned()) + } else { + None + }, + username: String::new(), + transport: "binary".to_owned(), + client_address: sessions + .borrow() + .connection_address(transport_client_id) + .map_or_else(String::new, |a| a.to_string()), + } + } else { + warn!( + transport_client_id, + "rejecting register request with unsupported payload shape" + ); + send_login_eviction( + shard, + transport_client_id, + vsr_client_id, + EvictionReason::MalformedLogin, + ) + .await; + return; + }; + + match crate::external_auth::try_external_auth(external_auth, ext_request).await { + Ok(Some(crate::external_auth::ExternalAuthDecision::IggyUser { user_id })) => { + if user_id == 0 { + warn!( + transport_client_id, + "external auth attempted to map login to root user" + ); + send_login_eviction( + shard, + transport_client_id, + vsr_client_id, + EvictionReason::InvalidCredentials, + ) + .await; + return; + } + if crate::external_auth::is_synthetic_user_id(user_id) { Review Comment: critical: nothing checks that this user exists or is active. the built-in path and the http twin both check, config.toml:1085 promises it, and user ids get recycled - so a stale mapping logs in as whoever took that slot. ########## core/server/src/http/handlers.rs: ########## @@ -1799,6 +1866,103 @@ pub(in crate::http) async fn delete_pat( Ok(StatusCode::NO_CONTENT) } +/// Try external auth for an HTTP login. Returns `Some(result)` when the +/// external service responded (grant or deny) or when a callout failure +/// produces a terminal deny. Returns `None` when the caller should fall +/// through to built-in credential verification. +async fn try_external_auth_http_login( + state: &HttpInner, + credential_type: CredentialType, + username: &str, + credential_value: &str, + client_address: &str, +) -> Option<Result<Json<IdentityInfo>, CustomError>> { + use configs::external_auth::ExternalAuthErrorStrategy; + + let credential = state + .external_auth + .forward_credentials + .then(|| credential_value.to_owned()); + let request = ExternalAuthRequest { + credential_type, + credential, + username: username.to_owned(), + transport: "http".to_owned(), + client_address: client_address.to_owned(), + }; + let decision = match try_external_auth(&state.external_auth, request).await { + Ok(Some(decision)) => decision, + Ok(None) => return None, + Err(_) => { + return match state.external_auth.on_error { + ExternalAuthErrorStrategy::Fallback => None, + ExternalAuthErrorStrategy::Deny => Some(Err(IggyError::Unauthenticated.into())), + }; + } + }; + Some(handle_http_auth_decision(state, decision)) +} + +fn handle_http_auth_decision( + state: &HttpInner, + decision: ExternalAuthDecision, +) -> Result<Json<IdentityInfo>, CustomError> { + use consensus::MetadataHandle; + + match decision { + ExternalAuthDecision::IggyUser { user_id } => { + if user_id == 0 { + tracing::warn!("external auth attempted to map login to root user"); + return Err(IggyError::Unauthenticated.into()); + } + if is_synthetic_user_id(user_id) { + tracing::warn!( + user_id, + "external auth returned synthetic user_id in IggyUser response" + ); + return Err(IggyError::Unauthenticated.into()); + } + let user_valid = state.shard.plane.metadata().mux_stm.users().read(|users| { + users + .items + .get(user_id as usize) + .is_some_and(|u| u.status == iggy_common::UserStatus::Active) + }); + if !user_valid { + return Err(IggyError::Unauthenticated.into()); + } + issue_identity(state, user_id) + } + ExternalAuthDecision::InlineGrant { + principal: _, + permissions, + expires_at, + } => { + let Some(synthetic_user_id) = state.mint_synthetic_user_id() else { + tracing::error!("synthetic user ID space exhausted"); + return Err(IggyError::Unauthenticated.into()); + }; + state Review Comment: critical: this entry is only removed when a registered `HttpSession` expires or is forgotten, but read routes use `Identity` and register nothing. a poll-only client never creates a session, so its id leaks for the process lifetime. ########## core/metadata/src/stm/authz.rs: ########## @@ -136,6 +136,11 @@ pub(crate) fn authorize( if user_id == ROOT_USER_ID { return None; } + + if user_id > SYNTHETIC_USER_ID_THRESHOLD { Review Comment: critical: this runs before the operation match, so it also denies `JoinConsumerGroup` and `LeaveConsumerGroup`, which config.toml:1089 tells operators inline grants can do. the denial is post-commit, so each retry costs a journal append and a quorum round. ########## core/server/src/session_manager.rs: ########## @@ -351,6 +394,88 @@ impl SessionManager { .iter() .map(|(&id, conn)| record_from(id, conn)) } + + /// Mint a synthetic user ID for an external auth inline-grant session. + /// Reuses a previously freed ID when available, otherwise draws from the + /// shared counter. Returns `None` when the synthetic ID space is exhausted. + pub fn mint_synthetic_user_id(&mut self) -> Option<u32> { + if let Some(id) = self.free_synthetic_ids.pop_last() { + return Some(id); + } + self.synthetic_counter.mint() + } + + /// Store session-scoped permissions on a connection and index by synthetic + /// user ID for dispatch-time authorization lookups. + pub fn set_session_permissions( + &mut self, + connection_id: u128, + user_id: u32, + perms: SessionPermissions, + ) { + if let Some(conn) = self.connections.get_mut(&connection_id) { + conn.session_permissions = Some(perms); + self.synthetic_user_to_connection + .insert(user_id, connection_id); + } + } + + /// Remove session-scoped permissions and reclaim the synthetic user ID. + /// Used to roll back an inline-grant when `complete_login_register` fails. + /// Always reclaims the ID regardless of whether it was registered in the + /// reverse index (it may not be if the failure happened before + /// `set_session_permissions` populated the index, or if the connection + /// disappeared between mint and set). + pub fn clear_session_permissions(&mut self, connection_id: u128, user_id: u32) { + if let Some(conn) = self.connections.get_mut(&connection_id) { + conn.session_permissions = None; + } + self.synthetic_user_to_connection.remove(&user_id); + if is_synthetic_user_id(user_id) { + self.free_synthetic_ids.insert(user_id); + } + } + + /// Look up session-scoped permissions by synthetic user ID. + /// Used by the dispatch-time authorization layer. Returns `None` when + /// the session has expired (checked against the current wall clock), + /// which the authorization layer treats as `Unauthorized`. + #[must_use] + pub fn session_permissions_for_user(&self, user_id: u32) -> Option<&Permissions> { + let &conn_id = self.synthetic_user_to_connection.get(&user_id)?; + let conn = self.connections.get(&conn_id)?; + conn.session_permissions.as_ref().and_then(|sp| { + let now_secs = iggy_common::IggyTimestamp::now().to_secs(); + if sp.expires_at <= now_secs { + return None; + } + Some(&sp.permissions) + }) + } + + /// Connection IDs whose session permissions have expired. The heartbeat + /// verifier or a periodic sweep evicts these. + #[must_use] + pub fn collect_expired_sessions(&self, now_secs: u64) -> Vec<u128> { Review Comment: warning: `collect_expired_sessions` and `is_session_expired` have no callers, so nothing evicts a binary inline-grant connection at `expires_at`. gated ops fail closed, but the connection keeps its client-table slot and the auth-only codes. ########## core/configs/src/server_config/external_auth.rs: ########## @@ -0,0 +1,59 @@ +// 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. + +use configs::ConfigEnv; +use iggy_common::IggyDuration; +use serde::{Deserialize, Serialize}; +use serde_with::DisplayFromStr; +use serde_with::serde_as; + +/// Strategy when the external auth service is unreachable or returns an error. +#[derive(Debug, Default, Deserialize, Serialize, Clone, PartialEq, Eq, ConfigEnv)] +#[serde(rename_all = "snake_case")] +pub enum ExternalAuthErrorStrategy { + #[default] + Deny, + Fallback, +} + +/// External authentication callout configuration. +/// +/// When enabled, login attempts are forwarded to an external HTTP service +/// before (or instead of) built-in credential verification. The service +/// returns a grant (with inline permissions or by mapping to an existing +/// Iggy user) or a denial. +#[serde_as] +#[derive(Debug, Deserialize, Serialize, Clone, ConfigEnv)] +pub struct ExternalAuthConfig { + #[serde(default)] + pub enabled: bool, + #[serde(default, skip_serializing)] + #[config_env(secret)] + pub url: String, + #[serde_as(as = "DisplayFromStr")] + #[serde(default = "default_external_auth_timeout")] + #[config_env(leaf)] + pub timeout: IggyDuration, + #[serde(default)] + pub on_error: ExternalAuthErrorStrategy, Review Comment: warning: `on_error` has no `#[config_env(leaf)]`, and the enum's unit variants make `env_mappings()` empty, so `IGGY_EXTERNAL_AUTH_ON_ERROR` does not exist. setting it is ignored in release and panics boot in a debug build. ########## core/server/src/http/handlers.rs: ########## @@ -384,6 +426,7 @@ pub(in crate::http) async fn get_topics( |permissioner, uid| { scope.map_or(Ok(()), |stream_id| permissioner.get_topics(uid, stream_id)) }, + |p| scope.is_none_or(|sid| can_read_stream(p, sid)), Review Comment: warning: this passes `can_read_stream`, but the binary arm uses `can_list_topics`, which is what `Permissioner::get_topics` actually accepts. a grant holding only `read_topics` lists topics over tcp and gets 403 here. ########## 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 = Review Comment: warning: this `expect` fires when the rustls backend finds no system CA store, and the panic hook sets the shutdown flag - so the first login drops the whole server. build it in `bootstrap()` and return `ServerError`. ########## 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()) + .await + .map_err(|_| ExternalAuthError::Timeout)? + .map_err(|e| ExternalAuthError::HttpError(e.to_string()))?; + + let status = response.status(); + if !status.is_success() { + return Err(ExternalAuthError::HttpError(format!( + "non-success status: {status}" + ))); + } + + if let Some(len) = response + .headers() + .get("content-length") + .and_then(|v| v.to_str().ok()) + .and_then(|s| s.parse::<usize>().ok()) + && len > MAX_RESPONSE_BODY_BYTES + { + return Err(ExternalAuthError::BadResponse( + "response body too large".to_owned(), + )); + } + + let bytes = compio::time::timeout(timeout, response.bytes()) + .await + .map_err(|_| ExternalAuthError::Timeout)? + .map_err(|e| ExternalAuthError::HttpError(format!("failed to read body: {e}")))?; + + if bytes.len() > MAX_RESPONSE_BODY_BYTES { + return Err(ExternalAuthError::BadResponse( + "response body too large".into(), + )); + } + + let resp: ExternalAuthResponse = serde_json::from_slice(&bytes) + .map_err(|e| ExternalAuthError::BadResponse(format!("invalid JSON: {e}")))?; + + match resp.decision { + DecisionTag::IggyUser => { + let user_id = resp.user_id.ok_or_else(|| { + ExternalAuthError::BadResponse("iggy_user decision missing user_id".to_owned()) + })?; + Ok(ExternalAuthDecision::IggyUser { user_id }) + } + DecisionTag::InlineGrant => { + let principal = resp.principal.ok_or_else(|| { + ExternalAuthError::BadResponse("inline_grant decision missing principal".to_owned()) + })?; + let permissions = resp.permissions.ok_or_else(|| { + ExternalAuthError::BadResponse( + "inline_grant decision missing permissions".to_owned(), + ) + })?; + let expires_at = resp.expires_at.unwrap_or(u64::MAX); Review Comment: warning: a missing `expires_at` becomes `u64::MAX`, so the grant is bounded only by the connection or the default token ttl. fail-open default on a security field - bound it, or require the field. ########## core/server/src/http/state.rs: ########## @@ -122,6 +123,17 @@ pub(in crate::http) struct HttpInner { /// Legacy-parity metric registry served by the scrape route; the router's /// counting layer holds a clone of its request counter. pub(in crate::http) metrics: HttpMetrics, + /// External authentication callout config. Shared across all handlers. + pub(in crate::http) external_auth: Arc<ExternalAuthConfig>, + /// Session-scoped permissions for synthetic user IDs (external auth + /// inline-grant sessions). Keyed by synthetic `user_id`. Cleaned up on + /// session expiry sweep. + pub(in crate::http) synthetic_permissions: RefCell<HashMap<u32, Permissions>>, Review Comment: warning: this map is per node, so a follower-minted grant relayed to the primary misses and 403s with a misleading unauthorized. worse with the id collision, since the primary may hold a different principal under that id. ########## 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, Review Comment: warning: `principal` is written and never read anywhere, and the http path drops it entirely. it is the only link from a synthetic id back to a real identity, so log it at grant time or drop the field. ########## core/server/src/dispatch/authz.rs: ########## @@ -292,41 +460,54 @@ where // transport with an `Unauthenticated` Reply before it reaches the // builder, so this arm only ever fires if that gate is bypassed. GET_CLUSTER_METADATA_CODE => user_id.map(|_| ()).ok_or(IggyError::Unauthenticated), Review Comment: warning: this gates on authentication alone, so an inline grant with every bool false reads the cluster topology - node names, replica ips, endpoints. config.toml:1087 says these sessions are restricted to data-plane ops. ########## core/configs/src/server_config/external_auth.rs: ########## @@ -0,0 +1,59 @@ +// 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. + +use configs::ConfigEnv; +use iggy_common::IggyDuration; +use serde::{Deserialize, Serialize}; +use serde_with::DisplayFromStr; +use serde_with::serde_as; + +/// Strategy when the external auth service is unreachable or returns an error. +#[derive(Debug, Default, Deserialize, Serialize, Clone, PartialEq, Eq, ConfigEnv)] +#[serde(rename_all = "snake_case")] +pub enum ExternalAuthErrorStrategy { + #[default] + Deny, + Fallback, +} + +/// External authentication callout configuration. +/// +/// When enabled, login attempts are forwarded to an external HTTP service +/// before (or instead of) built-in credential verification. The service +/// returns a grant (with inline permissions or by mapping to an existing +/// Iggy user) or a denial. +#[serde_as] +#[derive(Debug, Deserialize, Serialize, Clone, ConfigEnv)] +pub struct ExternalAuthConfig { + #[serde(default)] + pub enabled: bool, + #[serde(default, skip_serializing)] + #[config_env(secret)] + pub url: String, + #[serde_as(as = "DisplayFromStr")] + #[serde(default = "default_external_auth_timeout")] + #[config_env(leaf)] + pub timeout: IggyDuration, Review Comment: warning: `timeout = "unlimited"` (also `0`, `none`, `disabled`) parses to `Duration::ZERO`, and the compio timeout then fires on the first poll. with the default `on_error = deny` that is a total login outage, and nothing validates it. ########## core/server/src/http/state.rs: ########## @@ -373,6 +382,41 @@ impl HttpInner { })) } + /// Mint a synthetic user ID for an external auth inline-grant session. + /// Reuses a previously freed ID when available, otherwise draws from the + /// shared counter. Returns `None` when the synthetic ID space is exhausted. + pub(in crate::http) fn mint_synthetic_user_id(&self) -> Option<u32> { + if let Some(id) = self.free_synthetic_ids.borrow_mut().pop_last() { + return Some(id); + } + self.synthetic_counter.mint() + } + + /// Return a synthetic user ID to the free list without removing a session. + pub(in crate::http) fn reclaim_synthetic_user_id(&self, id: u32) { + self.free_synthetic_ids.borrow_mut().insert(id); + } + + /// Look up session-scoped permissions for a synthetic user ID. Returns + /// `None` for non-synthetic users or when no permissions are stored. + pub(in crate::http) fn get_synthetic_permissions(&self, user_id: u32) -> Option<Permissions> { Review Comment: warning: this hands back the grant without checking `expires_at`, while the binary twin at `session_manager.rs:449` does check it. on http an expired grant keeps authorizing until the token itself expires. ########## core/server/src/session_manager.rs: ########## @@ -104,15 +112,31 @@ pub struct SessionManager { /// per-shard context already threaded to the non-replicated read path; /// installed once at bootstrap, disabled until then. cluster_roster: Rc<ClusterRoster>, + /// Shared counter for minting synthetic user IDs. All transports draw + /// from the same sequence so no two can mint the same ID. + synthetic_counter: SyntheticUserIdCounter, + /// Freed synthetic user IDs available for reuse. + free_synthetic_ids: BTreeSet<u32>, Review Comment: warning: each shard and the http listener keep their own free list over one shared counter, so freed ids only return to whoever minted them. once the counter drains, a busy shard mints `None` while ids sit free elsewhere. ########## core/server/src/dispatch/authz.rs: ########## @@ -42,30 +42,155 @@ use iggy_binary_protocol::requests::users::GetUserRequest; use iggy_binary_protocol::{ Operation, PrepareHeader, RoutedRequestHeader, WireDecode, WireIdentifier, }; -use iggy_common::IggyError; +use iggy_common::{IggyError, Permissions}; use journal::superblock::SuperblockStore; use journal::{Journal, JournalHandle}; use metadata::impls::metadata::StreamsFrontend; use metadata::permissioner::Permissioner; use server_common::Message; use tracing::warn; +use crate::external_auth::is_synthetic_user_id; use crate::responses::{ build_deny_reply, current_metadata_commit, resolve_stream_id, resolve_topic_id, }; use crate::shell::{ShellBus, ShellShard}; +/// Check session-scoped permissions for a synthetic user. Returns +/// `Some(Ok(()))` if permitted, `Some(Err(Unauthorized))` if denied, or `None` +/// if the user is not synthetic (caller falls through to the Permissioner). +pub(super) fn check_session_permission( + user_id: Option<u32>, + session_perms: Option<&Permissions>, + check: impl FnOnce(&Permissions) -> bool, +) -> Option<Result<(), IggyError>> { + let user_id = user_id?; + if !is_synthetic_user_id(user_id) { + return None; + } + match session_perms { + Some(perms) if check(perms) => Some(Ok(())), + _ => Some(Err(IggyError::Unauthorized)), + } +} + +/// Check if inline permissions allow sending messages to (stream, topic), +/// mirroring the `Permissioner::append_messages` inheritance chain. +pub fn can_send_messages(perms: &Permissions, stream_id: usize, topic_id: usize) -> bool { Review Comment: simplification: these five predicates re-implement `Permissioner` rules by hand. `init_permissions_for_user` builds a one-user permissioner from the same `Permissions`, dropping ~100 lines plus the closure param threaded through the gates. watch the `_ => false` at line 211 - the permissioner arm allows there. ########## core/server/src/http/handlers.rs: ########## @@ -1799,6 +1866,103 @@ pub(in crate::http) async fn delete_pat( Ok(StatusCode::NO_CONTENT) } +/// Try external auth for an HTTP login. Returns `Some(result)` when the +/// external service responded (grant or deny) or when a callout failure +/// produces a terminal deny. Returns `None` when the caller should fall +/// through to built-in credential verification. +async fn try_external_auth_http_login( + state: &HttpInner, + credential_type: CredentialType, + username: &str, + credential_value: &str, + client_address: &str, +) -> Option<Result<Json<IdentityInfo>, CustomError>> { + use configs::external_auth::ExternalAuthErrorStrategy; + + let credential = state + .external_auth + .forward_credentials + .then(|| credential_value.to_owned()); + let request = ExternalAuthRequest { + credential_type, + credential, + username: username.to_owned(), + transport: "http".to_owned(), + client_address: client_address.to_owned(), + }; + let decision = match try_external_auth(&state.external_auth, request).await { + Ok(Some(decision)) => decision, + Ok(None) => return None, + Err(_) => { Review Comment: simplification: `try_external_auth` already applied `on_error` - it returns `Ok(None)` for fallback and `Err` only for deny, so the fallback arm here is unreachable. collapse to `Err(_) => Some(Err(IggyError::Unauthenticated.into()))`. ########## core/server/src/http/handlers.rs: ########## @@ -1799,6 +1866,103 @@ pub(in crate::http) async fn delete_pat( Ok(StatusCode::NO_CONTENT) } +/// Try external auth for an HTTP login. Returns `Some(result)` when the +/// external service responded (grant or deny) or when a callout failure +/// produces a terminal deny. Returns `None` when the caller should fall +/// through to built-in credential verification. +async fn try_external_auth_http_login( + state: &HttpInner, + credential_type: CredentialType, + username: &str, + credential_value: &str, + client_address: &str, +) -> Option<Result<Json<IdentityInfo>, CustomError>> { + use configs::external_auth::ExternalAuthErrorStrategy; + + let credential = state + .external_auth + .forward_credentials + .then(|| credential_value.to_owned()); + let request = ExternalAuthRequest { + credential_type, + credential, + username: username.to_owned(), + transport: "http".to_owned(), + client_address: client_address.to_owned(), + }; + let decision = match try_external_auth(&state.external_auth, request).await { + Ok(Some(decision)) => decision, + Ok(None) => return None, + Err(_) => { + return match state.external_auth.on_error { + ExternalAuthErrorStrategy::Fallback => None, + ExternalAuthErrorStrategy::Deny => Some(Err(IggyError::Unauthenticated.into())), + }; + } + }; + Some(handle_http_auth_decision(state, decision)) +} + +fn handle_http_auth_decision( + state: &HttpInner, + decision: ExternalAuthDecision, +) -> Result<Json<IdentityInfo>, CustomError> { + use consensus::MetadataHandle; + + match decision { + ExternalAuthDecision::IggyUser { user_id } => { + if user_id == 0 { + tracing::warn!("external auth attempted to map login to root user"); + return Err(IggyError::Unauthenticated.into()); + } + if is_synthetic_user_id(user_id) { + tracing::warn!( + user_id, + "external auth returned synthetic user_id in IggyUser response" + ); + return Err(IggyError::Unauthenticated.into()); + } + let user_valid = state.shard.plane.metadata().mux_stm.users().read(|users| { + users + .items + .get(user_id as usize) + .is_some_and(|u| u.status == iggy_common::UserStatus::Active) + }); + if !user_valid { + return Err(IggyError::Unauthenticated.into()); + } + issue_identity(state, user_id) + } + ExternalAuthDecision::InlineGrant { + principal: _, + permissions, + expires_at, + } => { + let Some(synthetic_user_id) = state.mint_synthetic_user_id() else { + tracing::error!("synthetic user ID space exhausted"); + return Err(IggyError::Unauthenticated.into()); + }; + state + .synthetic_permissions + .borrow_mut() + .insert(synthetic_user_id, permissions); + let result = issue_identity_capped(state, synthetic_user_id, expires_at); Review Comment: warning: `expires_at` is never compared to now, so a past value mints an already-dead token, still returns 200, and skips the rollback below. that permanently burns a synthetic id per call. ########## 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( Review Comment: warning: every login attempt fires a POST before any credential check, with no in-flight cap and no single-flight. on the http route that is unauthenticated and unbounded - `jwks.rs:112` already solves this shape. ########## core/server/src/dispatch/reads.rs: ########## @@ -142,6 +142,9 @@ pub(in crate::dispatch) async fn handle_non_replicated_request<B, MJ, S, SB>( // connection lookup. `user_id` is `None` only on the pre-auth path // (PING), which serves ungated codes; the gated arms fail closed on it. let (user_id, client_address) = sessions.borrow().read_context(transport_client_id); + let session_perms = user_id + .filter(|&uid| crate::external_auth::is_synthetic_user_id(uid)) + .and_then(|uid| sessions.borrow().session_permissions_for_user(uid).cloned()); Review Comment: warning: this deep-clones `Permissions` on every request, including `PING` and `GET_ME` which never use it, and `streams` is a nested `BTreeMap`. store an `Rc` and clone that instead. -- 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]
