hubcio commented on code in PR #4055:
URL: https://github.com/apache/iggy/pull/4055#discussion_r4061043072
##########
core/metadata/src/stm/authz.rs:
##########
@@ -129,15 +134,39 @@ pub(crate) fn authorize(
prepare: &Message<PrepareHeader>,
users: &Users,
streams: &Streams,
+ external_auth_user_id: Option<u32>,
) -> Option<ApplyReply> {
let header = prepare.header();
let user_id = header.user_id;
if user_id == ROOT_USER_ID {
return None;
}
+
let body = prepare.body();
+ // The external auth inline-grant user may execute the replicated
+ // data-plane ops listed below. Each is also gated at dispatch time by
+ // session-scoped permissions (authorize_partition_op for SendMessages
+ // and consumer offsets; authorize_consumer_group_op for CG join/leave).
+ // Non-replicated reads (PollMessages, GetConsumerOffset) never reach
+ // the STM and are gated only at dispatch time.
+ // SYNC: if a new REPLICATED data-plane operation is added, add it
+ // here AND add a dispatch-time session-scoped permission check in
+ // dispatch/authz.rs or dispatch/mod.rs.
+ if external_auth_user_id.is_some_and(|id| user_id == id) {
Review Comment:
critical: this reads node-local config inside a replicated apply - a node
with external auth off, or a snapshot restart that replays before boot sets the
id, denies the join a peer applied and group state diverges. carry the id in
replicated state instead.
##########
core/server/src/http/state.rs:
##########
@@ -283,16 +296,21 @@ impl HttpInner {
let mut table = self.sessions.borrow_mut();
let torn = sweep_expired(&mut table, now);
if table.len() >= self.max_http_sessions {
- // Still full after dropping expired entries: too
many
- // genuinely live sessions. Refuse rather than
evict a
- // live one (its `fresh` client id is orphaned on
the
- // peers until they evict it - a rare at-cap cost).
(None, torn)
} else {
table.insert(key.clone(), Rc::clone(&fresh));
(Some(fresh), torn)
}
};
+ // Reclaim orphaned grants: either the expiry has passed
+ // or the session table no longer holds a matching entry
+ // (the JWT expired and was swept above).
+ {
+ let sessions = self.sessions.borrow();
+ self.session_grants.borrow_mut().retain(|key, grant| {
Review Comment:
critical: a login grant has no session entry until its holder's first write,
so this retain drops it as soon as any other client registers - the holder then
gets 403 on every read. drop the session-table clause and let expiry reclaim.
##########
core/common/src/types/permissions/permissions_global.rs:
##########
@@ -26,6 +26,7 @@ use std::fmt::Display;
/// Global permissions are applied to all streams.
/// Stream permissions are applied to a specific stream.
#[derive(Debug, Serialize, Deserialize, PartialEq, Clone, Default)]
+#[serde(default)]
Review Comment:
warning: `#[serde(default)]` here also relaxes the create-user and
update-permissions REST bodies, which parse this same type - a body with
`global` missing or misspelled now parses as all-false instead of failing.
scope the default to the callout's grant type or document the break.
##########
core/server/src/http/state.rs:
##########
@@ -193,6 +198,14 @@ pub(in crate::http) struct HttpInner {
/// Behind `Rc` because the write path records from a detached task that
/// outlives its handler by design (see `submit_committed`).
pub(in crate::http) metadata_watermarks: Rc<MetadataWatermarks>,
+ /// External authentication callout config. Shared across all handlers.
+ pub(in crate::http) external_auth: Arc<ExternalAuthConfig>,
+ /// Pending inline-grant permissions keyed by session key (`jwt:{jti}`).
+ /// Populated at login when an inline grant is issued; consumed by the
+ /// session resolution path to attach the grant to the `HttpSession`.
+ pub(in crate::http) session_grants: RefCell<
Review Comment:
warning: grants live only on the node that issued the token, and the token
verifies on every node, so a request that lands elsewhere gets 403. carry the
grant in the token or replicate it.
##########
core/server/src/dispatch/authz.rs:
##########
@@ -128,12 +264,60 @@ where
decision.err().map(|error| error.as_code())
}
+/// Pre-submit check for external auth users on CG join/leave. The STM
+/// allow-list passes these ops (it has no session-scoped permissions), so
+/// this is the only topic-level gate on the binary transport. Returns
+/// `Some(status_code)` to deny, `None` to proceed.
+pub(in crate::dispatch) fn authorize_consumer_group_op<B, MJ, S, SB>(
+ shard: &Rc<ShellShard<B, MJ, S, SB>>,
+ operation: Operation,
+ body: &[u8],
+ session_perms: Option<&Permissions>,
+) -> Option<u32>
+where
+ B: ShellBus,
+ MJ: JournalHandle + 'static,
+ MJ::Target: Journal<Entry = Message<PrepareHeader>, Header =
PrepareHeader>,
+ S: 'static,
+ SB: SuperblockStore + 'static,
+{
+ let Some(perms) = session_perms else {
+ return Some(IggyError::Unauthorized.as_code());
+ };
+ let (stream_id, topic_id) = match operation {
+ Operation::JoinConsumerGroup => {
+ let Ok(req) = JoinConsumerGroupRequest::decode_from(body) else {
+ return None;
+ };
+ (req.stream_id, req.topic_id)
+ }
+ Operation::LeaveConsumerGroup => {
+ let Ok(req) = LeaveConsumerGroupRequest::decode_from(body) else {
+ return None;
+ };
+ (req.stream_id, req.topic_id)
+ }
+ _ => return None,
+ };
+ let (sid, tid) = resolve_topic_scope(shard, &stream_id, &topic_id)?;
Review Comment:
critical: the `?` here returns `None`, which means proceed, so a join on a
topic this shard's mirror has not seen yet reaches the primary unchecked and
the STM allow-list admits it. deny on a miss like the other ext-auth gates do.
##########
core/server/src/dispatch/reads.rs:
##########
@@ -399,12 +411,27 @@ pub(in crate::dispatch) async fn
handle_non_replicated_request<B, MJ, S, SB>(
GET_ME_CODE => {
// Self-scoped, so no permissioner rule -- but the consumer-group
// list it carries is read off the streams STM, so it is gated like
- // any other metadata read.
+ // any other metadata read. External auth sessions with no stream
+ // grants are denied: an all-false inline grant should not learn
+ // transport details or consumer-group memberships.
if let Err(error) = authorize_and_hold_read(shard, code,
watermark, || Ok(())).await {
send_non_replicated_deny(shard, &request, transport_client_id,
error.as_code())
.await;
return;
}
+ if session_perms
Review Comment:
warning: this checks the grant's shape, not its bits - a global-only
`read_streams` grant can list every stream yet GET_ME denies it, while a
missing or expired grant passes. gate on read permission, deny an all-false
grant, and deny a missing one.
also at line 636.
##########
core/server/src/dispatch/authz.rs:
##########
@@ -128,12 +264,60 @@ where
decision.err().map(|error| error.as_code())
}
+/// Pre-submit check for external auth users on CG join/leave. The STM
+/// allow-list passes these ops (it has no session-scoped permissions), so
+/// this is the only topic-level gate on the binary transport. Returns
+/// `Some(status_code)` to deny, `None` to proceed.
+pub(in crate::dispatch) fn authorize_consumer_group_op<B, MJ, S, SB>(
+ shard: &Rc<ShellShard<B, MJ, S, SB>>,
+ operation: Operation,
+ body: &[u8],
+ session_perms: Option<&Permissions>,
+) -> Option<u32>
+where
+ B: ShellBus,
+ MJ: JournalHandle + 'static,
+ MJ::Target: Journal<Entry = Message<PrepareHeader>, Header =
PrepareHeader>,
+ S: 'static,
+ SB: SuperblockStore + 'static,
+{
+ let Some(perms) = session_perms else {
+ return Some(IggyError::Unauthorized.as_code());
+ };
+ let (stream_id, topic_id) = match operation {
+ Operation::JoinConsumerGroup => {
+ let Ok(req) = JoinConsumerGroupRequest::decode_from(body) else {
+ return None;
+ };
+ (req.stream_id, req.topic_id)
+ }
+ Operation::LeaveConsumerGroup => {
+ let Ok(req) = LeaveConsumerGroupRequest::decode_from(body) else {
+ return None;
+ };
+ (req.stream_id, req.topic_id)
+ }
+ _ => return None,
+ };
+ let (sid, tid) = resolve_topic_scope(shard, &stream_id, &topic_id)?;
+ if can_poll_messages(perms, sid, tid) {
Review Comment:
warning: join and leave map to `get_topic` in the permissioner, but this
tests `can_poll_messages`, which also admits a bare `poll_messages` grant. use
`can_read_topic` for both.
##########
core/server/src/external_auth.rs:
##########
@@ -0,0 +1,772 @@
+// 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.
+//!
+//! Known limitation: there is no per-client rate limit on callout attempts.
+//! Every failed built-in login triggers a callout, so a brute-force attack
+//! can amplify traffic to the external service. The shard reactor's
+//! single-threaded model bounds binary-transport concurrency, but the HTTP
+//! transport is concurrent. Operators should rate-limit at the network
+//! layer or inside the external auth service itself.
+
+use std::fmt;
+
+use configs::external_auth::ExternalAuthConfig;
+use iggy_common::Permissions;
+use serde::{Deserialize, Serialize};
+use tracing::warn;
+
+const MAX_RESPONSE_BODY_BYTES: usize = 1_048_576;
+
+/// Reject credentials longer than this before serializing them into the
+/// callout body. A multi-MB PAT passes the built-in hash check (mismatch)
+/// and would otherwise be forwarded verbatim to the external service.
+const MAX_CREDENTIAL_BYTES: usize = 8_192;
+
+thread_local! {
+ static HTTP_CLIENT: cyper::Client =
+ cyper::Client::builder()
+ .redirect(cyper::redirect::Policy::none())
+ .build()
+ .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.
+///
+/// For `PersonalAccessToken` logins, `username` is empty (PATs are not
+/// associated with a username on the wire). The auth service should use
+/// the `credential` value to identify the caller.
+///
+/// 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. Fields are `Option`
+/// because different decisions use different subsets; `into_decision`
+/// validates the required fields per variant. A tagged enum would be
+/// cleaner but serde's internal-tag path buffers through `Value`,
+/// which loses string-to-integer map key parsing for
+/// `Permissions.streams: BTreeMap<u32, _>`.
+#[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, or rejected request).
+#[derive(Debug)]
+pub enum ExternalAuthError {
+ HttpError(String),
+ Timeout,
+ BadResponse(String),
+ /// The request was rejected before the callout was made (e.g. credential
+ /// too large). Distinguished from `BadResponse` so callers/logs can tell
+ /// which side was at fault.
+ BadRequest(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}"),
+ Self::BadRequest(msg) => write!(f, "external auth bad request:
{msg}"),
+ }
+ }
+}
+
+impl std::error::Error for ExternalAuthError {}
+
+fn redact_url_userinfo(url: &str) -> std::borrow::Cow<'_, str> {
+ let Some((scheme, rest)) = url.split_once("://") else {
+ return std::borrow::Cow::Borrowed(url);
+ };
+ match rest.split_once('@') {
+ Some((_, after_at)) =>
std::borrow::Cow::Owned(format!("{scheme}://{after_at}")),
+ None => std::borrow::Cow::Borrowed(url),
+ }
+}
+
+/// # 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: {}",
+ redact_url_userinfo(&config.url)
+ ),
+ },
+ );
+ }
+ // Catch bare schemes ("http://";) with no host at boot instead of on
+ // first callout. The scheme prefix check above already guarantees the
+ // split will succeed.
+ let after_scheme = config.url.split_once("://").map_or("", |(_, rest)|
rest);
+ if after_scheme.is_empty() || after_scheme.starts_with('/') {
+ return Err(
+ crate::server_error::ServerError::InvalidExternalAuthConfig {
+ reason: format!(
+ "external_auth.url has no host: {}",
+ redact_url_userinfo(&config.url)
+ ),
+ },
+ );
+ }
+ if config.timeout.get_duration().is_zero() {
+ return Err(
+ crate::server_error::ServerError::InvalidExternalAuthConfig {
+ reason: "external_auth.timeout must be greater than
zero".to_owned(),
+ },
+ );
+ }
+ // The callout runs inline on the single-threaded shard reactor; a very
+ // large timeout blocks the entire shard for its duration. Warn above 30 s.
+ if config.timeout.get_duration().as_secs() > 30 {
+ tracing::warn!(
+ timeout = %config.timeout,
+ "external_auth.timeout is unusually large (> 30 s); \
+ the callout blocks the shard reactor for its full duration"
+ );
+ }
+ if config.user_id == 0 {
+ return Err(
+ crate::server_error::ServerError::InvalidExternalAuthConfig {
+ reason: "external_auth.user_id must not be 0 (reserved for
root)".to_owned(),
+ },
+ );
+ }
+ // The slab allocator assigns user IDs sequentially from 0 (root).
+ // A low user_id will collide with a real user once enough are created,
+ // silently restricting that user to data-plane ops. The default
+ // (u32::MAX) avoids this; warn if the operator picked a low value.
+ if config.user_id < 1_000_000 {
+ tracing::warn!(
+ user_id = config.user_id,
+ "external_auth.user_id is low; it may collide with a future Iggy
user ID. \
+ Consider using a large value (the default is u32::MAX)."
+ );
+ }
+ Ok(())
+}
+
+pub fn warn_insecure_url(config: &ExternalAuthConfig) {
+ if config.enabled && config.url.starts_with("http://";) {
+ tracing::warn!(
+ url = %redact_url_userinfo(&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. The permissions
+/// are `Arc`-wrapped so dispatch-time lookups share rather than clone
+/// the full `BTreeMap` tree on every request.
+#[derive(Debug, Clone)]
+pub struct SessionPermissions {
+ pub principal: String,
+ pub permissions: std::sync::Arc<Permissions>,
+ pub expires_at: u64,
+}
+
+/// Call the external auth service and parse the response.
+///
+/// # Errors
+///
+/// Returns [`ExternalAuthError`] on network/timeout/parse failure.
+/// Fail-closed: every error variant denies the login.
+pub async fn callout_external_auth(
Review Comment:
warning: nothing bounds concurrent callouts - every failed login on an
unauthenticated route opens an outbound socket and parks a handler for up to
the timeout. cap in-flight callouts the way the partition-write admission guard
does.
##########
core/server/src/external_auth.rs:
##########
@@ -0,0 +1,772 @@
+// 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.
+//!
+//! Known limitation: there is no per-client rate limit on callout attempts.
+//! Every failed built-in login triggers a callout, so a brute-force attack
+//! can amplify traffic to the external service. The shard reactor's
+//! single-threaded model bounds binary-transport concurrency, but the HTTP
+//! transport is concurrent. Operators should rate-limit at the network
+//! layer or inside the external auth service itself.
+
+use std::fmt;
+
+use configs::external_auth::ExternalAuthConfig;
+use iggy_common::Permissions;
+use serde::{Deserialize, Serialize};
+use tracing::warn;
+
+const MAX_RESPONSE_BODY_BYTES: usize = 1_048_576;
+
+/// Reject credentials longer than this before serializing them into the
+/// callout body. A multi-MB PAT passes the built-in hash check (mismatch)
+/// and would otherwise be forwarded verbatim to the external service.
+const MAX_CREDENTIAL_BYTES: usize = 8_192;
+
+thread_local! {
+ static HTTP_CLIENT: cyper::Client =
+ cyper::Client::builder()
+ .redirect(cyper::redirect::Policy::none())
+ .build()
+ .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.
+///
+/// For `PersonalAccessToken` logins, `username` is empty (PATs are not
+/// associated with a username on the wire). The auth service should use
+/// the `credential` value to identify the caller.
+///
+/// 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. Fields are `Option`
+/// because different decisions use different subsets; `into_decision`
+/// validates the required fields per variant. A tagged enum would be
+/// cleaner but serde's internal-tag path buffers through `Value`,
+/// which loses string-to-integer map key parsing for
+/// `Permissions.streams: BTreeMap<u32, _>`.
+#[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, or rejected request).
+#[derive(Debug)]
+pub enum ExternalAuthError {
+ HttpError(String),
+ Timeout,
+ BadResponse(String),
+ /// The request was rejected before the callout was made (e.g. credential
+ /// too large). Distinguished from `BadResponse` so callers/logs can tell
+ /// which side was at fault.
+ BadRequest(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}"),
+ Self::BadRequest(msg) => write!(f, "external auth bad request:
{msg}"),
+ }
+ }
+}
+
+impl std::error::Error for ExternalAuthError {}
+
+fn redact_url_userinfo(url: &str) -> std::borrow::Cow<'_, str> {
+ let Some((scheme, rest)) = url.split_once("://") else {
+ return std::borrow::Cow::Borrowed(url);
+ };
+ match rest.split_once('@') {
Review Comment:
warning: this splits at the first `@`, so a password with one leaks its tail
into the boot error and the warn. use `rsplit_once`.
##########
core/server/src/external_auth.rs:
##########
@@ -0,0 +1,772 @@
+// 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.
+//!
+//! Known limitation: there is no per-client rate limit on callout attempts.
+//! Every failed built-in login triggers a callout, so a brute-force attack
+//! can amplify traffic to the external service. The shard reactor's
Review Comment:
nit: every binary client drains on its own task, so the reactor does not
bound callout concurrency either. drop that sentence.
##########
core/server/src/http/handlers.rs:
##########
@@ -1870,6 +1964,110 @@ pub(in crate::http) async fn delete_pat(
Ok(StatusCode::NO_CONTENT)
}
+/// Try external auth for an HTTP login. Returns `Some(result)` when the
Review Comment:
nit: this doc promises `Some`/`None` but the function returns a `Result` and
never falls through. drop the stale paragraph.
##########
core/server/src/boot/mod.rs:
##########
@@ -624,6 +645,29 @@ async fn shard_main(
} else {
(None, None, None, None, (0, 0), None)
};
+ // Peer shards (Waiter path) and snapshot-restored mux_stm need the
+ // external auth user_id too. The Owner path sets it in seed_baseline
Review Comment:
nit: only true without a snapshot - recovery skips the seed closure when it
restores one, so the id stays unset for that whole replay. fix the comment.
##########
core/server/src/http/state.rs:
##########
@@ -466,6 +523,9 @@ 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 let Some(sk) =
crate::http::extractor::SessionKey::from_table_key(&session.key) {
Review Comment:
warning: eviction is transient, but this drops the grant for good - the
bearer keeps a valid token and gets 403 until it expires. keep the grant here
and let expiry reclaim it.
##########
core/server/src/external_auth.rs:
##########
@@ -0,0 +1,772 @@
+// 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.
+//!
+//! Known limitation: there is no per-client rate limit on callout attempts.
+//! Every failed built-in login triggers a callout, so a brute-force attack
+//! can amplify traffic to the external service. The shard reactor's
+//! single-threaded model bounds binary-transport concurrency, but the HTTP
+//! transport is concurrent. Operators should rate-limit at the network
+//! layer or inside the external auth service itself.
+
+use std::fmt;
+
+use configs::external_auth::ExternalAuthConfig;
+use iggy_common::Permissions;
+use serde::{Deserialize, Serialize};
+use tracing::warn;
+
+const MAX_RESPONSE_BODY_BYTES: usize = 1_048_576;
+
+/// Reject credentials longer than this before serializing them into the
+/// callout body. A multi-MB PAT passes the built-in hash check (mismatch)
+/// and would otherwise be forwarded verbatim to the external service.
+const MAX_CREDENTIAL_BYTES: usize = 8_192;
+
+thread_local! {
+ static HTTP_CLIENT: cyper::Client =
+ cyper::Client::builder()
+ .redirect(cyper::redirect::Policy::none())
+ .build()
+ .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.
+///
+/// For `PersonalAccessToken` logins, `username` is empty (PATs are not
+/// associated with a username on the wire). The auth service should use
+/// the `credential` value to identify the caller.
+///
+/// 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. Fields are `Option`
+/// because different decisions use different subsets; `into_decision`
+/// validates the required fields per variant. A tagged enum would be
+/// cleaner but serde's internal-tag path buffers through `Value`,
+/// which loses string-to-integer map key parsing for
+/// `Permissions.streams: BTreeMap<u32, _>`.
+#[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, or rejected request).
+#[derive(Debug)]
+pub enum ExternalAuthError {
+ HttpError(String),
+ Timeout,
+ BadResponse(String),
+ /// The request was rejected before the callout was made (e.g. credential
+ /// too large). Distinguished from `BadResponse` so callers/logs can tell
+ /// which side was at fault.
+ BadRequest(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}"),
+ Self::BadRequest(msg) => write!(f, "external auth bad request:
{msg}"),
+ }
+ }
+}
+
+impl std::error::Error for ExternalAuthError {}
+
+fn redact_url_userinfo(url: &str) -> std::borrow::Cow<'_, str> {
+ let Some((scheme, rest)) = url.split_once("://") else {
+ return std::borrow::Cow::Borrowed(url);
+ };
+ match rest.split_once('@') {
+ Some((_, after_at)) =>
std::borrow::Cow::Owned(format!("{scheme}://{after_at}")),
+ None => std::borrow::Cow::Borrowed(url),
+ }
+}
+
+/// # 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: {}",
+ redact_url_userinfo(&config.url)
+ ),
+ },
+ );
+ }
+ // Catch bare schemes ("http://";) with no host at boot instead of on
+ // first callout. The scheme prefix check above already guarantees the
+ // split will succeed.
+ let after_scheme = config.url.split_once("://").map_or("", |(_, rest)|
rest);
+ if after_scheme.is_empty() || after_scheme.starts_with('/') {
+ return Err(
+ crate::server_error::ServerError::InvalidExternalAuthConfig {
+ reason: format!(
+ "external_auth.url has no host: {}",
+ redact_url_userinfo(&config.url)
+ ),
+ },
+ );
+ }
+ if config.timeout.get_duration().is_zero() {
+ return Err(
+ crate::server_error::ServerError::InvalidExternalAuthConfig {
+ reason: "external_auth.timeout must be greater than
zero".to_owned(),
+ },
+ );
+ }
+ // The callout runs inline on the single-threaded shard reactor; a very
+ // large timeout blocks the entire shard for its duration. Warn above 30 s.
+ if config.timeout.get_duration().as_secs() > 30 {
+ tracing::warn!(
+ timeout = %config.timeout,
+ "external_auth.timeout is unusually large (> 30 s); \
+ the callout blocks the shard reactor for its full duration"
+ );
+ }
+ if config.user_id == 0 {
+ return Err(
+ crate::server_error::ServerError::InvalidExternalAuthConfig {
+ reason: "external_auth.user_id must not be 0 (reserved for
root)".to_owned(),
+ },
+ );
+ }
+ // The slab allocator assigns user IDs sequentially from 0 (root).
+ // A low user_id will collide with a real user once enough are created,
+ // silently restricting that user to data-plane ops. The default
+ // (u32::MAX) avoids this; warn if the operator picked a low value.
+ if config.user_id < 1_000_000 {
+ tracing::warn!(
+ user_id = config.user_id,
+ "external_auth.user_id is low; it may collide with a future Iggy
user ID. \
+ Consider using a large value (the default is u32::MAX)."
+ );
+ }
+ Ok(())
+}
+
+pub fn warn_insecure_url(config: &ExternalAuthConfig) {
+ if config.enabled && config.url.starts_with("http://";) {
+ tracing::warn!(
+ url = %redact_url_userinfo(&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. The permissions
+/// are `Arc`-wrapped so dispatch-time lookups share rather than clone
+/// the full `BTreeMap` tree on every request.
+#[derive(Debug, Clone)]
+pub struct SessionPermissions {
+ pub principal: String,
+ pub permissions: std::sync::Arc<Permissions>,
+ pub expires_at: u64,
+}
+
+/// Call the external auth service and parse the response.
+///
+/// # Errors
+///
+/// Returns [`ExternalAuthError`] on network/timeout/parse failure.
+/// Fail-closed: every error variant denies the login.
+pub async fn callout_external_auth(
+ config: &ExternalAuthConfig,
+ request: ExternalAuthRequest,
+) -> Result<ExternalAuthDecision, ExternalAuthError> {
+ use futures::StreamExt;
+
+ if let Some(ref cred) = request.credential
Review Comment:
warning: only `credential` is capped. on HTTP an over-long `username` fails
as invalid credentials, which is exactly the arm that calls out, so it rides
the whole 2 MB body to the service - cap it the same way.
##########
core/simulator/src/replica.rs:
##########
@@ -372,6 +372,12 @@ pub fn new_shard(
}
// Same seed the server bootstrap runs after its own replay.
metadata.seed_applied_frontier_from_consensus();
+ // NOTE: The simulator uses ExternalAuthConfig::default() (enabled: false),
Review Comment:
nit: backwards - with the id unset the gate routes the reserved user to the
permissioner, which denies its join and leave, so the sim diverges from nodes
that set it. and there is no `seed_baseline` closure here, the mux is built
inline above.
##########
core/server/src/http/handlers.rs:
##########
@@ -1271,6 +1350,10 @@ pub(in crate::http) async fn poll_messages(
permissioner.poll_messages(uid, stream_id, topic_id)
})
},
+ |p| {
+ resolve_gate_topic_ids(&state, &stream_id, &topic_id)
+ .is_none_or(|(sid, tid)| can_poll_messages(p, sid, tid))
Review Comment:
nit: a resolution miss passes here but the same miss on the binary path
denies, so one unknown topic answers 404 over HTTP and 403 over TCP. pick one.
also at line 1441.
##########
core/server/src/http/reads.rs:
##########
@@ -383,10 +395,36 @@ pub(in crate::http) fn resolve_gate_topic_ids(
pub(in crate::http) fn authorize_data_plane(
state: &HttpInner,
user_id: u32,
+ session_key: &str,
stream_id: &Identifier,
topic_id: &Identifier,
rule: impl FnOnce(&Permissioner, u32, usize, usize) -> Result<(),
IggyError>,
+ inline_grant_check: impl FnOnce(&Permissions, usize, usize) -> bool,
) -> Result<(), IggyError> {
+ if state.external_auth.enabled && user_id == state.external_auth.user_id {
+ let sk =
crate::http::extractor::SessionKey::from_table_key(session_key)
+ .ok_or(IggyError::Unauthorized)?;
+ let perms = state
+ .session_grant_permissions(&sk)
+ .ok_or(IggyError::Unauthorized)?;
+ // Fail-closed: for external auth users the session-scoped check below
+ // is the only topic-level gate (the STM authz gate allows all
+ // data-plane ops by op-code). A resolution miss must deny, not fall
+ // through to the handler's own not-found path.
+ let (Ok(wire_stream), Ok(wire_topic)) =
Review Comment:
simplification: both arms convert and resolve the same two ids. resolve once
above the branch and keep only the miss mapping per arm.
##########
core/server/src/http/extractor.rs:
##########
@@ -42,6 +42,35 @@ const BEARER: &str = "Bearer ";
const JWT_KEY_PREFIX: &str = "jwt:";
const PAT_KEY_PREFIX: &str = "pat:";
+/// Session table key identifying a credential. Two variants so JWTs and PATs
+/// occupy disjoint key spaces.
+#[derive(Debug, Clone, PartialEq, Eq, Hash)]
Review Comment:
simplification: `SessionKey` is formatted to a string on every write and
parsed back at four sites, and every `from_table_key` miss branch is dead.
return the prefixed string from `resolve_credential` and pass `&str` around.
##########
core/server/src/external_auth.rs:
##########
@@ -0,0 +1,772 @@
+// 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.
+//!
+//! Known limitation: there is no per-client rate limit on callout attempts.
+//! Every failed built-in login triggers a callout, so a brute-force attack
+//! can amplify traffic to the external service. The shard reactor's
+//! single-threaded model bounds binary-transport concurrency, but the HTTP
+//! transport is concurrent. Operators should rate-limit at the network
+//! layer or inside the external auth service itself.
+
+use std::fmt;
+
+use configs::external_auth::ExternalAuthConfig;
+use iggy_common::Permissions;
+use serde::{Deserialize, Serialize};
+use tracing::warn;
+
+const MAX_RESPONSE_BODY_BYTES: usize = 1_048_576;
+
+/// Reject credentials longer than this before serializing them into the
+/// callout body. A multi-MB PAT passes the built-in hash check (mismatch)
+/// and would otherwise be forwarded verbatim to the external service.
+const MAX_CREDENTIAL_BYTES: usize = 8_192;
+
+thread_local! {
+ static HTTP_CLIENT: cyper::Client =
+ cyper::Client::builder()
+ .redirect(cyper::redirect::Policy::none())
+ .build()
+ .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.
+///
+/// For `PersonalAccessToken` logins, `username` is empty (PATs are not
+/// associated with a username on the wire). The auth service should use
+/// the `credential` value to identify the caller.
+///
+/// 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. Fields are `Option`
+/// because different decisions use different subsets; `into_decision`
+/// validates the required fields per variant. A tagged enum would be
+/// cleaner but serde's internal-tag path buffers through `Value`,
+/// which loses string-to-integer map key parsing for
+/// `Permissions.streams: BTreeMap<u32, _>`.
+#[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, or rejected request).
+#[derive(Debug)]
+pub enum ExternalAuthError {
+ HttpError(String),
+ Timeout,
+ BadResponse(String),
+ /// The request was rejected before the callout was made (e.g. credential
+ /// too large). Distinguished from `BadResponse` so callers/logs can tell
+ /// which side was at fault.
+ BadRequest(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}"),
+ Self::BadRequest(msg) => write!(f, "external auth bad request:
{msg}"),
+ }
+ }
+}
+
+impl std::error::Error for ExternalAuthError {}
+
+fn redact_url_userinfo(url: &str) -> std::borrow::Cow<'_, str> {
+ let Some((scheme, rest)) = url.split_once("://") else {
+ return std::borrow::Cow::Borrowed(url);
+ };
+ match rest.split_once('@') {
+ Some((_, after_at)) =>
std::borrow::Cow::Owned(format!("{scheme}://{after_at}")),
+ None => std::borrow::Cow::Borrowed(url),
+ }
+}
+
+/// # 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: {}",
+ redact_url_userinfo(&config.url)
+ ),
+ },
+ );
+ }
+ // Catch bare schemes ("http://";) with no host at boot instead of on
+ // first callout. The scheme prefix check above already guarantees the
+ // split will succeed.
+ let after_scheme = config.url.split_once("://").map_or("", |(_, rest)|
rest);
+ if after_scheme.is_empty() || after_scheme.starts_with('/') {
+ return Err(
+ crate::server_error::ServerError::InvalidExternalAuthConfig {
+ reason: format!(
+ "external_auth.url has no host: {}",
+ redact_url_userinfo(&config.url)
+ ),
+ },
+ );
+ }
+ if config.timeout.get_duration().is_zero() {
+ return Err(
+ crate::server_error::ServerError::InvalidExternalAuthConfig {
+ reason: "external_auth.timeout must be greater than
zero".to_owned(),
+ },
+ );
+ }
+ // The callout runs inline on the single-threaded shard reactor; a very
+ // large timeout blocks the entire shard for its duration. Warn above 30 s.
+ if config.timeout.get_duration().as_secs() > 30 {
+ tracing::warn!(
+ timeout = %config.timeout,
+ "external_auth.timeout is unusually large (> 30 s); \
+ the callout blocks the shard reactor for its full duration"
+ );
+ }
+ if config.user_id == 0 {
+ return Err(
+ crate::server_error::ServerError::InvalidExternalAuthConfig {
+ reason: "external_auth.user_id must not be 0 (reserved for
root)".to_owned(),
+ },
+ );
+ }
+ // The slab allocator assigns user IDs sequentially from 0 (root).
+ // A low user_id will collide with a real user once enough are created,
+ // silently restricting that user to data-plane ops. The default
+ // (u32::MAX) avoids this; warn if the operator picked a low value.
+ if config.user_id < 1_000_000 {
+ tracing::warn!(
+ user_id = config.user_id,
+ "external_auth.user_id is low; it may collide with a future Iggy
user ID. \
+ Consider using a large value (the default is u32::MAX)."
+ );
+ }
+ Ok(())
+}
+
+pub fn warn_insecure_url(config: &ExternalAuthConfig) {
+ if config.enabled && config.url.starts_with("http://";) {
+ tracing::warn!(
+ url = %redact_url_userinfo(&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. The permissions
+/// are `Arc`-wrapped so dispatch-time lookups share rather than clone
+/// the full `BTreeMap` tree on every request.
+#[derive(Debug, Clone)]
+pub struct SessionPermissions {
+ pub principal: String,
+ pub permissions: std::sync::Arc<Permissions>,
+ pub expires_at: u64,
+}
+
+/// Call the external auth service and parse the response.
+///
+/// # Errors
+///
+/// Returns [`ExternalAuthError`] on network/timeout/parse failure.
+/// Fail-closed: every error variant denies the login.
+pub async fn callout_external_auth(
+ config: &ExternalAuthConfig,
+ request: ExternalAuthRequest,
+) -> Result<ExternalAuthDecision, ExternalAuthError> {
+ use futures::StreamExt;
+
+ if let Some(ref cred) = request.credential
+ && cred.len() > MAX_CREDENTIAL_BYTES
+ {
+ return Err(ExternalAuthError::BadRequest(format!(
+ "credential too large ({} bytes, limit {MAX_CREDENTIAL_BYTES})",
+ cred.len()
+ )));
+ }
+
+ 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}")))?;
+
+ // Single timeout wrapping the entire round-trip (connect + headers +
+ // body read) so a slow-drip body cannot extend the window to 2x.
+ let round_trip = async {
+ let request_builder = 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 response = request_builder
+ .send()
+ .await
+ .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(),
+ ));
+ }
+
+ // Stream the body with a running cap so a length-less response cannot
+ // OOM the server. Matches the forward.rs pattern.
+ let mut buf = Vec::new();
+ let mut stream = response.bytes_stream();
+ while let Some(chunk) = stream.next().await {
+ let chunk =
+ chunk.map_err(|e| ExternalAuthError::HttpError(format!("body
read: {e}")))?;
+ if buf.len() + chunk.len() > MAX_RESPONSE_BODY_BYTES {
+ return Err(ExternalAuthError::BadResponse(
+ "response body too large".to_owned(),
+ ));
+ }
+ buf.extend_from_slice(&chunk);
+ }
+ Ok(buf)
+ };
+ let bytes = compio::time::timeout(timeout, round_trip)
+ .await
+ .map_err(|_| ExternalAuthError::Timeout)??;
+
+ let resp: ExternalAuthResponse = serde_json::from_slice(&bytes)
+ .map_err(|e| ExternalAuthError::BadResponse(format!("invalid JSON:
{e}")))?;
+
+ into_decision(resp)
+}
+
+/// Validate required fields per decision variant and convert to the
+/// public type. Extracted so tests can exercise the mapping without
+/// an HTTP round-trip.
+fn into_decision(resp: ExternalAuthResponse) -> Result<ExternalAuthDecision,
ExternalAuthError> {
+ 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.ok_or_else(|| {
+ ExternalAuthError::BadResponse(
+ "inline_grant decision missing expires_at".to_owned(),
+ )
+ })?;
+ Ok(ExternalAuthDecision::InlineGrant {
+ principal,
+ permissions,
+ expires_at,
+ })
+ }
+ DecisionTag::Deny => {
+ let reason = resp
+ .reason
+ .unwrap_or_else(|| "denied by external auth".to_owned());
+ Ok(ExternalAuthDecision::Deny { reason })
+ }
+ }
+}
+
+/// Try external auth and map the result to a decision the login flow
+/// can act on. Fail-closed: a callout failure (timeout, network error,
+/// bad response) denies the login.
+///
+/// # Errors
+///
+/// Returns [`ExternalAuthError`] when the callout itself failed.
+pub async fn try_external_auth(
Review Comment:
simplification: this only logs and re-raises, and the binary caller logs the
same message again with more context. delete it and put the warn in the HTTP
caller, which drops the error today.
##########
core/server/src/http/jwt.rs:
##########
@@ -79,6 +79,9 @@ pub struct JwtManager {
/// `[[http.jwt.trusted_issuers]]` is configured, in which case `decode`
/// takes the JWKS verification path for tokens from these issuers.
trusted_issuers: HashMap<String, TrustedIssuerConfig>,
+ /// User IDs that trusted issuers must never map to (defense in depth).
+ /// Contains the external auth reserved `user_id` when that feature is
enabled.
+ forbidden_user_ids: Vec<u32>,
Review Comment:
simplification: this holds at most one id - the only writer collects an
`Option<u32>`. store the `Option` and compare it at line 282.
##########
core/server/src/external_auth.rs:
##########
@@ -0,0 +1,772 @@
+// 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.
+//!
+//! Known limitation: there is no per-client rate limit on callout attempts.
+//! Every failed built-in login triggers a callout, so a brute-force attack
+//! can amplify traffic to the external service. The shard reactor's
+//! single-threaded model bounds binary-transport concurrency, but the HTTP
+//! transport is concurrent. Operators should rate-limit at the network
+//! layer or inside the external auth service itself.
+
+use std::fmt;
+
+use configs::external_auth::ExternalAuthConfig;
+use iggy_common::Permissions;
+use serde::{Deserialize, Serialize};
+use tracing::warn;
+
+const MAX_RESPONSE_BODY_BYTES: usize = 1_048_576;
+
+/// Reject credentials longer than this before serializing them into the
+/// callout body. A multi-MB PAT passes the built-in hash check (mismatch)
+/// and would otherwise be forwarded verbatim to the external service.
+const MAX_CREDENTIAL_BYTES: usize = 8_192;
+
+thread_local! {
+ static HTTP_CLIENT: cyper::Client =
+ cyper::Client::builder()
+ .redirect(cyper::redirect::Policy::none())
+ .build()
+ .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.
+///
+/// For `PersonalAccessToken` logins, `username` is empty (PATs are not
+/// associated with a username on the wire). The auth service should use
+/// the `credential` value to identify the caller.
+///
+/// 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. Fields are `Option`
+/// because different decisions use different subsets; `into_decision`
+/// validates the required fields per variant. A tagged enum would be
+/// cleaner but serde's internal-tag path buffers through `Value`,
+/// which loses string-to-integer map key parsing for
+/// `Permissions.streams: BTreeMap<u32, _>`.
+#[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, or rejected request).
+#[derive(Debug)]
+pub enum ExternalAuthError {
+ HttpError(String),
+ Timeout,
+ BadResponse(String),
+ /// The request was rejected before the callout was made (e.g. credential
+ /// too large). Distinguished from `BadResponse` so callers/logs can tell
+ /// which side was at fault.
+ BadRequest(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}"),
+ Self::BadRequest(msg) => write!(f, "external auth bad request:
{msg}"),
+ }
+ }
+}
+
+impl std::error::Error for ExternalAuthError {}
+
+fn redact_url_userinfo(url: &str) -> std::borrow::Cow<'_, str> {
+ let Some((scheme, rest)) = url.split_once("://") else {
+ return std::borrow::Cow::Borrowed(url);
+ };
+ match rest.split_once('@') {
+ Some((_, after_at)) =>
std::borrow::Cow::Owned(format!("{scheme}://{after_at}")),
+ None => std::borrow::Cow::Borrowed(url),
+ }
+}
+
+/// # 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: {}",
+ redact_url_userinfo(&config.url)
+ ),
+ },
+ );
+ }
+ // Catch bare schemes ("http://";) with no host at boot instead of on
+ // first callout. The scheme prefix check above already guarantees the
+ // split will succeed.
+ let after_scheme = config.url.split_once("://").map_or("", |(_, rest)|
rest);
+ if after_scheme.is_empty() || after_scheme.starts_with('/') {
+ return Err(
+ crate::server_error::ServerError::InvalidExternalAuthConfig {
+ reason: format!(
+ "external_auth.url has no host: {}",
+ redact_url_userinfo(&config.url)
+ ),
+ },
+ );
+ }
+ if config.timeout.get_duration().is_zero() {
+ return Err(
+ crate::server_error::ServerError::InvalidExternalAuthConfig {
+ reason: "external_auth.timeout must be greater than
zero".to_owned(),
+ },
+ );
+ }
+ // The callout runs inline on the single-threaded shard reactor; a very
+ // large timeout blocks the entire shard for its duration. Warn above 30 s.
+ if config.timeout.get_duration().as_secs() > 30 {
+ tracing::warn!(
+ timeout = %config.timeout,
+ "external_auth.timeout is unusually large (> 30 s); \
+ the callout blocks the shard reactor for its full duration"
Review Comment:
nit: an awaited callout parks its own task, not the reactor, so this warning
overstates the cost. say it stalls that login for the timeout instead.
##########
core/server/src/http/submit.rs:
##########
@@ -406,6 +406,13 @@ pub(in crate::http) async fn partition_write_replicated(
);
PartitionWriteError::Unavailable
})?;
+ let session_perms = (state.external_auth.enabled
Review Comment:
simplification: this block is repeated verbatim at line 474. pull it into
one `HttpInner` helper.
##########
core/server/src/external_auth.rs:
##########
@@ -0,0 +1,772 @@
+// 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.
+//!
+//! Known limitation: there is no per-client rate limit on callout attempts.
+//! Every failed built-in login triggers a callout, so a brute-force attack
+//! can amplify traffic to the external service. The shard reactor's
+//! single-threaded model bounds binary-transport concurrency, but the HTTP
+//! transport is concurrent. Operators should rate-limit at the network
+//! layer or inside the external auth service itself.
+
+use std::fmt;
+
+use configs::external_auth::ExternalAuthConfig;
+use iggy_common::Permissions;
+use serde::{Deserialize, Serialize};
+use tracing::warn;
+
+const MAX_RESPONSE_BODY_BYTES: usize = 1_048_576;
+
+/// Reject credentials longer than this before serializing them into the
+/// callout body. A multi-MB PAT passes the built-in hash check (mismatch)
+/// and would otherwise be forwarded verbatim to the external service.
+const MAX_CREDENTIAL_BYTES: usize = 8_192;
+
+thread_local! {
+ static HTTP_CLIENT: cyper::Client =
+ cyper::Client::builder()
+ .redirect(cyper::redirect::Policy::none())
+ .build()
+ .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.
+///
+/// For `PersonalAccessToken` logins, `username` is empty (PATs are not
+/// associated with a username on the wire). The auth service should use
+/// the `credential` value to identify the caller.
+///
+/// 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. Fields are `Option`
+/// because different decisions use different subsets; `into_decision`
+/// validates the required fields per variant. A tagged enum would be
+/// cleaner but serde's internal-tag path buffers through `Value`,
+/// which loses string-to-integer map key parsing for
+/// `Permissions.streams: BTreeMap<u32, _>`.
+#[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, or rejected request).
+#[derive(Debug)]
+pub enum ExternalAuthError {
+ HttpError(String),
+ Timeout,
+ BadResponse(String),
+ /// The request was rejected before the callout was made (e.g. credential
+ /// too large). Distinguished from `BadResponse` so callers/logs can tell
+ /// which side was at fault.
+ BadRequest(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}"),
+ Self::BadRequest(msg) => write!(f, "external auth bad request:
{msg}"),
+ }
+ }
+}
+
+impl std::error::Error for ExternalAuthError {}
+
+fn redact_url_userinfo(url: &str) -> std::borrow::Cow<'_, str> {
+ let Some((scheme, rest)) = url.split_once("://") else {
+ return std::borrow::Cow::Borrowed(url);
+ };
+ match rest.split_once('@') {
+ Some((_, after_at)) =>
std::borrow::Cow::Owned(format!("{scheme}://{after_at}")),
+ None => std::borrow::Cow::Borrowed(url),
+ }
+}
+
+/// # 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: {}",
+ redact_url_userinfo(&config.url)
+ ),
+ },
+ );
+ }
+ // Catch bare schemes ("http://";) with no host at boot instead of on
+ // first callout. The scheme prefix check above already guarantees the
+ // split will succeed.
+ let after_scheme = config.url.split_once("://").map_or("", |(_, rest)|
rest);
+ if after_scheme.is_empty() || after_scheme.starts_with('/') {
+ return Err(
+ crate::server_error::ServerError::InvalidExternalAuthConfig {
+ reason: format!(
+ "external_auth.url has no host: {}",
+ redact_url_userinfo(&config.url)
+ ),
+ },
+ );
+ }
+ if config.timeout.get_duration().is_zero() {
+ return Err(
+ crate::server_error::ServerError::InvalidExternalAuthConfig {
+ reason: "external_auth.timeout must be greater than
zero".to_owned(),
+ },
+ );
+ }
+ // The callout runs inline on the single-threaded shard reactor; a very
+ // large timeout blocks the entire shard for its duration. Warn above 30 s.
+ if config.timeout.get_duration().as_secs() > 30 {
+ tracing::warn!(
+ timeout = %config.timeout,
+ "external_auth.timeout is unusually large (> 30 s); \
+ the callout blocks the shard reactor for its full duration"
+ );
+ }
+ if config.user_id == 0 {
+ return Err(
+ crate::server_error::ServerError::InvalidExternalAuthConfig {
+ reason: "external_auth.user_id must not be 0 (reserved for
root)".to_owned(),
+ },
+ );
+ }
+ // The slab allocator assigns user IDs sequentially from 0 (root).
+ // A low user_id will collide with a real user once enough are created,
+ // silently restricting that user to data-plane ops. The default
+ // (u32::MAX) avoids this; warn if the operator picked a low value.
+ if config.user_id < 1_000_000 {
+ tracing::warn!(
+ user_id = config.user_id,
+ "external_auth.user_id is low; it may collide with a future Iggy
user ID. \
+ Consider using a large value (the default is u32::MAX)."
+ );
+ }
+ Ok(())
+}
+
+pub fn warn_insecure_url(config: &ExternalAuthConfig) {
+ if config.enabled && config.url.starts_with("http://";) {
+ tracing::warn!(
+ url = %redact_url_userinfo(&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. The permissions
+/// are `Arc`-wrapped so dispatch-time lookups share rather than clone
+/// the full `BTreeMap` tree on every request.
+#[derive(Debug, Clone)]
+pub struct SessionPermissions {
+ pub principal: String,
Review Comment:
simplification: `principal` is written by both login paths and read by
nothing. drop the field.
##########
core/server/src/dispatch/authz.rs:
##########
@@ -128,12 +264,60 @@ where
decision.err().map(|error| error.as_code())
}
+/// Pre-submit check for external auth users on CG join/leave. The STM
+/// allow-list passes these ops (it has no session-scoped permissions), so
+/// this is the only topic-level gate on the binary transport. Returns
+/// `Some(status_code)` to deny, `None` to proceed.
+pub(in crate::dispatch) fn authorize_consumer_group_op<B, MJ, S, SB>(
+ shard: &Rc<ShellShard<B, MJ, S, SB>>,
+ operation: Operation,
+ body: &[u8],
+ session_perms: Option<&Permissions>,
+) -> Option<u32>
+where
+ B: ShellBus,
+ MJ: JournalHandle + 'static,
+ MJ::Target: Journal<Entry = Message<PrepareHeader>, Header =
PrepareHeader>,
+ S: 'static,
+ SB: SuperblockStore + 'static,
+{
+ let Some(perms) = session_perms else {
+ return Some(IggyError::Unauthorized.as_code());
+ };
+ let (stream_id, topic_id) = match operation {
+ Operation::JoinConsumerGroup => {
+ let Ok(req) = JoinConsumerGroupRequest::decode_from(body) else {
+ return None;
+ };
+ (req.stream_id, req.topic_id)
+ }
+ Operation::LeaveConsumerGroup => {
+ let Ok(req) = LeaveConsumerGroupRequest::decode_from(body) else {
+ return None;
+ };
+ (req.stream_id, req.topic_id)
+ }
+ _ => return None,
Review Comment:
simplification: unreachable, the caller only routes join and leave here.
drop the arm.
##########
core/server/src/dispatch/reads.rs:
##########
@@ -565,17 +613,39 @@ pub(in crate::dispatch) async fn
handle_non_replicated_request<B, MJ, S, SB>(
}
}
GET_CONSUMER_OFFSET_CODE => {
- handle_get_consumer_offset(shard, transport_client_id, &request,
user_id).await;
+ handle_get_consumer_offset(
+ shard,
+ transport_client_id,
+ &request,
+ user_id,
+ session_perms.as_deref(),
+ )
+ .await;
}
SYNC_CONSUMER_GROUP_CODE => {
// Self-scoped: serves the caller's own assignment keyed by the
// header client id, so it carries no permissioner rule. The
// assignment itself is metadata-STM state, hence the gate.
+ // External auth sessions with no stream grants are denied (same
+ // rationale as GET_ME above).
if let Err(error) = authorize_and_hold_read(shard, code,
watermark, || Ok(())).await {
send_non_replicated_deny(shard, &request, transport_client_id,
error.as_code())
.await;
return;
}
+ if session_perms
Review Comment:
simplification: same predicate as line 422. one helper, so the fix lands
once.
##########
core/server/src/dispatch/authz.rs:
##########
@@ -51,17 +52,143 @@ use server_common::Message;
use crate::responses::{resolve_stream_id, resolve_topic_id};
use crate::shell::{ShellBus, ShellShard};
+/// Check session-scoped permissions for an external auth inline-grant user.
+/// Returns `Some(Ok(()))` if permitted, `Some(Err(Unauthorized))` if denied,
+/// or `None` if `session_perms` is absent (the caller is a regular user and
+/// falls through to the Permissioner).
+pub(super) fn check_session_permission(
+ session_perms: Option<&Permissions>,
+ check: impl FnOnce(&Permissions) -> bool,
+) -> Option<Result<(), IggyError>> {
+ match session_perms {
+ Some(perms) if check(perms) => Some(Ok(())),
+ Some(_) => Some(Err(IggyError::Unauthorized)),
+ None => None,
+ }
+}
+
+/// Check if inline permissions allow sending messages to (stream, topic),
+/// mirroring the `Permissioner::append_messages` inheritance chain.
+// SYNC: keep in lockstep with Permissioner::append_messages
+pub fn can_send_messages(perms: &Permissions, stream_id: usize, topic_id:
usize) -> bool {
Review Comment:
simplification: these five predicates re-implement the permissioner rules by
hand and already drift, see the join/leave gate. pull the rule bodies out as
free functions over `Option<&GlobalPermissions>` and
`Option<&StreamPermissions>` and call them from both sides.
--
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]
