numinnex commented on code in PR #4550:
URL: https://github.com/apache/iggy/pull/4550#discussion_r4282814790
##########
core/metadata/src/impls/metadata.rs:
##########
@@ -849,12 +851,14 @@ where
// Normal op: apply, build the reply. `Err` is decode/corruption only; a
// business rejection commits as a deterministic no-op whose code rides
// the reply body, replayed on retry.
- let apply = gated_apply(mux_stm, prepare).unwrap_or_else(|err| {
- panic!(
- "apply_committed_prepare: committed metadata op={} failed to
apply: {err}",
- header.op
- );
- });
+ let apply = reject_stale_session(&header, &client_table.borrow())
Review Comment:
`Register` escapes this check: `apply_committed_prepare` returns for
`Operation::Register` above, and `register_preflight`
(`core/consensus/src/metadata_helpers.rs:246-291`) reads only the client table.
A login admitted while `DeleteUser(alice)` is in the pipeline therefore
commits a live entry for the freed id after `end_user_sessions` ran, and
`complete_session_binding` (`core/server/src/dispatch/mod.rs:1131-1141`) only
checks that `items[N]` is active.
Once bob takes id N, alice is bound as bob and her ops pass this check and
`authorize` under bob's grants. Without bob, the ghost entry survives in the
checkpointed table and `BIND_SESSION`
(`core/server/src/dispatch/reads.rs:640-671`) binds it to whoever takes N later.
HTTP closes the same race with `ensure_active_user(created_at)` after
Register (`core/server/src/http/state.rs:555`). Binary needs the same: carry
the verified `created_at` in the Register body, refuse at commit when
`items[N].created_at` differs, and compare it in `complete_session_binding`.
That also covers `CreatePersonalAccessTokenRequest::apply`
(`core/metadata/src/stm/user.rs:801-808`), which inserts under `user_id` with
no `items` check.
##########
core/shard/src/poll.rs:
##########
@@ -429,6 +429,23 @@ where
));
return;
}
+ if let Some(attachment) = &attachment
Review Comment:
An HTTP poll arrives as `UnboundPollOnPrimary` with no attachment
(`core/server/src/http/handlers.rs:1360-1368`), so this grant check never runs
for it. `complete_poll` then stores the `auto_commit` offset, and the gateway's
`authorize_read` after `partition_read` (`handlers.rs:1371`) drops the batch on
a revoke that landed during the read.
Binary has the same shape with a smaller window between this check and
`core/server/src/dispatch/partition.rs:701-711`.
Nit, since `auto_commit` is already at-most-once (`poll.rs:392-396`). Still,
carry `(user_id, created_at)` in `UnboundPollOnPrimary` and gate here before
`complete_poll`. Do not deliver after a deny; that reopens the earlier HTTP
poll leak.
##########
core/metadata/src/stm/authz.rs:
##########
@@ -78,7 +87,36 @@ where
if let Some(denied) = authorize(&prepare, mux.users(), mux.streams()) {
return Ok(denied);
}
- mux.update(prepare)
+ if let Some(rejected) = reject_dangling_grant(&prepare, mux.users(),
mux.streams()) {
+ return Ok(rejected);
+ }
+ let header = prepare.header();
+ let timestamp = IggyTimestamp::from(header.timestamp);
+ let context = ApplyContext::from(header);
+ let reply = mux.update(prepare)?;
+ if let Some(deleted) = reply.deleted_resource {
+ mux.users().remove_grants(deleted, timestamp, context);
+ }
+ Ok(reply)
+}
+
+/// An admitted request can commit after its user was deleted or disabled.
+/// Check the replicated session table before resolving the reused user ID.
+/// Partition completions use a synthetic session without a client-table entry.
+pub(crate) fn reject_stale_session(
+ header: &PrepareHeader,
+ client_table: &ClientTable,
+) -> Option<ApplyReply> {
+ if matches!(
+ header.operation,
+ Operation::CompleteConsumerGroupRevocation |
Operation::CompleteLifecycle
Review Comment:
Exemption by `Operation` allowlist. `submit_partition_completion` stamps
`session = 1` (`core/metadata/src/impls/metadata.rs:4471`), so the next
server-internal op that is not on this list commits as a silent `StaleClient`
no-op on every replica.
Exempt by `message_bus::is_server_internal_client(header.client)` instead,
as `metadata.rs:3501` does.
##########
core/metadata/src/stm/authz.rs:
##########
@@ -78,7 +87,36 @@ where
if let Some(denied) = authorize(&prepare, mux.users(), mux.streams()) {
return Ok(denied);
}
- mux.update(prepare)
+ if let Some(rejected) = reject_dangling_grant(&prepare, mux.users(),
mux.streams()) {
+ return Ok(rejected);
+ }
+ let header = prepare.header();
+ let timestamp = IggyTimestamp::from(header.timestamp);
+ let context = ApplyContext::from(header);
+ let reply = mux.update(prepare)?;
+ if let Some(deleted) = reply.deleted_resource {
+ mux.users().remove_grants(deleted, timestamp, context);
Review Comment:
This is a second `Users` publish after the `Streams` publish from
`mux.update`, so a peer-shard reader can observe the stream gone with the grant
still present for one op. Harmless today, the id cannot be reused before a
later op, but correct only by writer program order. Worth a comment here so a
future reorder does not break it.
##########
core/metadata/src/stm/user.rs:
##########
@@ -655,6 +707,82 @@ impl StateHandler for UpdatePermissionsRequest {
}
}
+/// Server-originated request that drops every grant on a deleted resource.
+///
+/// Never decoded from the wire: `gated_apply` applies it right after the
+/// delete, so every replica and WAL replay remove the same grants.
+#[derive(Debug, Clone)]
+pub struct RemoveGrantsRequest {
+ pub deleted: DeletedResource,
+}
+
+impl StateHandler for RemoveGrantsRequest {
+ type State = UsersInner;
+ fn apply(&self, state: &mut UsersInner, _timestamp: IggyTimestamp) ->
ApplyReply {
+ let mut pruned_any = false;
+ for (_, user) in &mut state.items {
+ let Some(pruned) = user
+ .permissions
+ .as_deref()
+ .and_then(|permissions| without_grant(permissions,
self.deleted))
+ else {
+ continue;
+ };
+ user.permissions = Some(Arc::new(pruned));
+ pruned_any = true;
+ }
+ if pruned_any {
+ state.permissioner = permissioner_of(&state.items);
Review Comment:
Any hit rebuilds the whole `Permissioner` and deep-clones every user's
`Permissions`, O(total grants), and left_right runs the apply twice
(`core/metadata/src/stm/mod.rs:559-565`), so a stream or topic delete under
many users stalls the shard 0 apply loop twice.
Collect the pruned user ids and patch the index for those users only; keep
the full rebuild for `from_snapshot`.
##########
core/server/src/session_manager.rs:
##########
@@ -199,6 +204,10 @@ impl SessionManager {
};
Some(ConnectionContext {
bound,
+ attachment: conn
+ .consumer_session
+ .as_ref()
+ .map(|(_, attachment)| attachment.clone()),
Review Comment:
This clones the `SessionAttachment` (atomic weak inc + dec) on every inbound
frame, `SendMessages` included, while only the `NonReplicatedRead` arm reads it
(`core/server/src/dispatch/mod.rs:817`). Clone it only for that class.
##########
core/metadata/src/stm/authz.rs:
##########
@@ -78,7 +87,36 @@ where
if let Some(denied) = authorize(&prepare, mux.users(), mux.streams()) {
return Ok(denied);
}
- mux.update(prepare)
+ if let Some(rejected) = reject_dangling_grant(&prepare, mux.users(),
mux.streams()) {
+ return Ok(rejected);
+ }
+ let header = prepare.header();
+ let timestamp = IggyTimestamp::from(header.timestamp);
+ let context = ApplyContext::from(header);
+ let reply = mux.update(prepare)?;
+ if let Some(deleted) = reply.deleted_resource {
+ mux.users().remove_grants(deleted, timestamp, context);
+ }
+ Ok(reply)
+}
+
+/// An admitted request can commit after its user was deleted or disabled.
+/// Check the replicated session table before resolving the reused user ID.
+/// Partition completions use a synthetic session without a client-table entry.
+pub(crate) fn reject_stale_session(
+ header: &PrepareHeader,
+ client_table: &ClientTable,
+) -> Option<ApplyReply> {
+ if matches!(
+ header.operation,
+ Operation::CompleteConsumerGroupRevocation |
Operation::CompleteLifecycle
+ ) {
+ return None;
+ }
+ (header.session != 0
+ && (client_table.get_epoch(header.client) != Some(header.session)
+ || client_table.get_user_id(header.client) !=
Some(header.user_id)))
Review Comment:
Dead clause. An epoch match implies a live entry, and `user_id` is immutable
per epoch and stamped from that same entry in `prepare_request`
(`core/metadata/src/impls/metadata.rs:3990-4010`), so `get_user_id` can never
disagree once `get_epoch` matched.
Drop it; that also removes the second slot lookup per committed op.
##########
core/server/src/http/state.rs:
##########
@@ -633,11 +677,29 @@ pub(in crate::http) fn insert_view_header(state:
&HttpInner, mut response: Respo
response
}
+/// `(id, created_at)` of `user` when a trusted issuer may act as it: active
and
+/// not root. A deleted user's id goes to the next user created, so the issuer
+/// is matched by username, and an external issuer never holds root.
+fn trusted_issuer_account(user: Option<&User>) -> Option<(u32, IggyTimestamp)>
{
+ let user = user?;
+ if user.id == DEFAULT_ROOT_USER_ID {
+ error!(
+ username = %user.username,
+ "a trusted issuer maps onto the root user, refusing its token"
Review Comment:
`error!` on every request once an issuer's username resolves to root, so one
token holder floods the log, while the missing or inactive case in
`trusted_issuer_user` is `debug!` only and a misspelled `username` is invisible
at the default level. One policy for both: `warn!` once per issuer.
##########
core/server/src/dispatch/partition.rs:
##########
@@ -694,11 +696,23 @@ where
consumer,
args,
attachment: ConsumerAttachment {
- session: attachment,
+ session: attachment.clone(),
Review Comment:
Three attachment clones per poll now: the `ConnectionContext` clone, the
second `consumer_session()` borrow and clone in `reads.rs`, and this one, which
was a move before. Derive the consumer session from the context, move it in
here, and keep a copied `user_id` for the recheck.
##########
core/server/src/http/jwt.rs:
##########
@@ -288,48 +284,66 @@ impl JwtManager {
validation.set_audience(std::slice::from_ref(&config.audience));
validation.validate_nbf = true;
validation.leeway = self.validation.leeway;
- let mut claims = decode::<JwtClaims>(token, &decoding_key, &validation)
+ let claims = decode::<JwtClaims>(token, &decoding_key, &validation)
.map(|data| data.claims)
.map_err(|error| {
error!(%error, "trusted-issuer JWT verification failed");
IggyError::Unauthenticated
})?;
- // Remap the external subject onto the fixed configured iggy user; the
- // raw external `sub` is discarded (no auto-provisioning).
- claims.sub = config.user_id.to_string();
- Ok(claims)
+ // The raw external `sub` is discarded (no auto-provisioning): every
+ // token from this issuer acts as the configured iggy user.
+ Ok(VerifiedToken {
+ jti: claims.jti,
+ exp: claims.exp,
+ subject: TokenSubject::Username(config.username.clone()),
Review Comment:
A `String` allocation per trusted-issuer request. An `Arc<str>` in the
manager's issuer map avoids it.
##########
core/common/src/traits/user_client.rs:
##########
@@ -35,6 +35,12 @@ pub trait UserClient {
/// Create a new user.
///
/// Authentication is required, and the permission to manage the users.
+ ///
+ /// Every stream and topic that `permissions` grants on must exist. If one
+ /// does not, a binary transport returns `IggyError::StreamIdNotFound` or
Review Comment:
The refusal reaches the client through `IggyError::from_code`, so it decodes
to `StreamIdNotFound(Identifier::default())` and prints "Stream with ID: 0 was
not found." (`core/server/src/http/reply.rs:145`, `core/sdk/src/vsr.rs:380`).
Stream 0 is a real id, and a multi-grant request gives no hint which grant
failed.
Same placeholder as every existing 1009 refusal, so a nit. Carry the missing
id in the reply body, or at least document the placeholder here.
The Rust `HttpClient` also drops the 1009/2010 id on 404
(`core/sdk/src/http/http_client.rs:367`), unlike the C# client, so the HTTP
branch documented here is coarser than it reads.
##########
core/server/src/http/extractor.rs:
##########
@@ -163,37 +169,56 @@ pub(in crate::http) fn bearer_token(headers: &HeaderMap)
-> Result<&str, IggyErr
.ok_or(IggyError::AccessTokenMissing)
}
-/// Map a bearer credential to its `(session key, user id, expiry secs)`.
+/// Map a bearer credential to its `(session key, user id, user created_at,
+/// expiry secs)`.
///
-/// A JWT is the primary credential, keyed by its `jti`. A raw PAT presented as
-/// the bearer is the documented fallback, keyed by the same 256-bit BLAKE3
-/// digest (`PersonalAccessToken::hash_token`) Iggy already indexes PATs by, so
-/// the key is stable and collision-free while the raw secret never enters the
-/// table. The JWT verify is `async` and `!Send` (a trusted-issuer token may
-/// fetch the issuer's JWKS on a cache miss), so both call sites drive it
inside
-/// a `SendWrapper`; the PAT check is local and synchronous. Also the forward
-/// middleware's verify-only gate: a follower proves the bearer before relaying
-/// a request, without minting a session.
+/// A JWT is the primary credential, keyed by its account and `jti`. A raw PAT
+/// presented as the bearer is the documented fallback, keyed by the same
+/// 256-bit BLAKE3 digest (`PersonalAccessToken::hash_token`) Iggy already
+/// indexes PATs by, so the key is stable and collision-free while the raw
+/// secret never enters the table. The JWT verify is `async` and `!Send` (a
+/// trusted-issuer token may fetch the issuer's JWKS on a cache miss), so both
+/// call sites drive it inside a `SendWrapper`; the PAT check is local and
+/// synchronous. Also the forward middleware's verify-only gate: a follower
+/// proves the bearer before relaying a request, without minting a session.
+///
+/// The `created_at` names the account behind the credential, because a deleted
+/// user's id goes to the next user created. A self-issued JWT must name the
+/// account that holds its id now, and a trusted-issuer JWT resolves the
+/// username its issuer config names.
pub(in crate::http) async fn resolve_credential(
state: &HttpState,
bearer: &str,
-) -> Result<(String, u32, u64), AuthError> {
- if let Ok(claims) = state.jwt.decode(bearer).await
- && let Ok(user_id) = claims.sub.parse::<u32>()
- {
- state.ensure_active_user(user_id)?;
- return Ok((
- format!("{JWT_KEY_PREFIX}{}", claims.jti),
- user_id,
- claims.exp,
- ));
+) -> Result<(String, u32, IggyTimestamp, u64), AuthError> {
+ if let Ok(token) = state.jwt.decode(bearer).await {
+ let (user_id, user_created_at) = match token.subject {
+ TokenSubject::Account {
+ user_id,
+ created_at,
+ } => (
+ user_id,
+ state.ensure_active_user(user_id, Some(created_at))?,
+ ),
+ TokenSubject::Username(username) =>
state.trusted_issuer_user(&username)?,
+ };
+ // A trusted-issuer token resolves its username on every request, so
one
+ // `jti` can name another account after a rename or a delete, and the
+ // issuer picks its own `jti`. The account in the key keeps a cached
+ // session from writing as any account but the one resolved here.
+ let key = format!(
+ "{JWT_KEY_PREFIX}{user_id}:{}:{}",
Review Comment:
The key is built on every request and discarded by the `Identity` extractor
below. Build it only in `Authenticated`, which is the sole consumer.
Pre-existing shape, but this PR made the string longer.
##########
core/server/README.md:
##########
@@ -44,7 +44,7 @@ Command-line arguments include `--replica-id`, `--fresh`
(`-f`), `--with-default
Deploy matching server and SDK releases together. Rolling upgrades and mixed
server versions are unsupported. Partition control frames from older servers
carry incarnation zero and are rejected, so a mixed-version cluster cannot
-complete partition view changes. The `IGGY-OWNERSHIP-HISTORY-5` storage gate
also
+complete partition view changes. The `IGGY-BOUND-GRANTS-1` storage gate also
Review Comment:
The upgrade note names only the storage gate. Two more things break for
operators: every pre-upgrade JWT, access and refresh, is refused
(`core/server/src/http/jwt.rs:340`), so HTTP clients and the web UI must log in
again; and `[[http.jwt.trusted_issuers]] user_id` became `username`, with the
env var `IGGY_HTTP_JWT_TRUSTED_ISSUERS_<n>_USER_ID` becoming `..._USERNAME`.
Both are in the commit body only.
##########
core/server/src/http/handlers.rs:
##########
@@ -1331,6 +1354,7 @@ pub(in crate::http) async fn poll_messages(
{
return Ok(response);
}
+ authorize_read(&state, &identity, consistency, rule)?;
Review Comment:
This is the last local gate for the routed case, but the `route_partition`
reply above returns before it, so a relayed poll is never re-gated after its
await. Safe, the target node re-verifies the same bearer and `created_at`, but
`reads.rs:55-58` lists partition routing among the rechecked routes. Either
reword that doc or gate the routed reply too. Same in `get_consumer_offset`.
##########
core/server/src/dispatch/reads.rs:
##########
@@ -708,6 +749,7 @@ where
else {
return Err(IggyError::TransientNotAccepted);
};
+ authorize_attached(attachment, authorize)?;
Review Comment:
The re-gate re-resolves by name, but the reply still carries the
`incarnation` captured before the `partition_read` await. A delete and recreate
under the same name during the wait returns an incarnation the client then
refreshes away as `HistoryUnavailable`. Fails closed, so a nit: re-read the
incarnation after the gate, or return `TransientNotAccepted` on mismatch.
##########
foreign/cpp/tests/e2e/client.cpp:
##########
@@ -354,8 +354,27 @@ TEST_F(E2E_Client,
CreateUserPreservesNestedPermissionsInCreateAndGetResponses)
RecordProperty("description",
"Creates a user with global and per-resource permissions,
then verifies create_user and get_user "
"return the same flags and numeric stream/topic IDs.");
- auto client = GetLoggedInHighLevelClient();
- const std::string username = GetRandomName(50);
+ auto client = GetLoggedInHighLevelClient();
+ const std::string username = GetRandomName(50);
+ const std::string first_stream_name = GetRandomName();
+ const std::string second_stream_name = GetRandomName();
+
+ // The server refuses a grant on a stream or topic that does not exist.
+ std::uint32_t first_stream_id = 0;
+ std::uint32_t second_stream_id = 0;
+ std::uint32_t first_topic_id = 0;
+ std::uint32_t second_topic_id = 0;
+ std::uint32_t third_topic_id = 0;
+ ASSERT_NO_THROW({
+ first_stream_id = client.CreateStream(first_stream_name).Id();
+ TrackStream(first_stream_name);
+ second_stream_id = client.CreateStream(second_stream_name).Id();
+ TrackStream(second_stream_name);
+ first_topic_id =
client.CreateTopic(iggy::Identifier::Numeric(first_stream_id),
GetRandomName()).Id();
+ second_topic_id =
client.CreateTopic(iggy::Identifier::Numeric(first_stream_id),
GetRandomName()).Id();
+ third_topic_id =
client.CreateTopic(iggy::Identifier::Numeric(second_stream_id),
GetRandomName()).Id();
Review Comment:
`first_topic_id` and `third_topic_id` are both 0 (first topic in each
stream), so a mix-up between the two streams' topic maps still passes the
assertions below. Create a filler topic in `second_stream` first, or assert the
ids are distinct.
##########
core/integration/tests/server/scenarios/permissions_scenario.rs:
##########
@@ -2626,6 +2628,168 @@ async fn test_live_permission_revocation(harness:
&TestHarness, root_client: &Ig
delete_test_user(root_client, USER).await;
}
+//
=============================================================================
+// Test: A deleted stream's grant does not reach the stream that reuses its id
+//
=============================================================================
+
+async fn test_deleted_stream_grant_not_inherited(harness: &TestHarness,
root_client: &IggyClient) {
+ const USER: &str = "deleted-stream-grant-user";
+ let topic_id = Identifier::named(TOPIC_1).unwrap();
+ let deleted = create_stream_with_topic(root_client,
"perm-deleted-stream").await;
+ let stream_id = Identifier::numeric(deleted.id).unwrap();
+ let granted_stream_id = deleted.id as usize;
+ create_test_user(
+ root_client,
+ USER,
+ Some(Permissions {
+ global: GlobalPermissions::default(),
+ streams: Some(BTreeMap::from([(
+ granted_stream_id,
+ StreamPermissions {
+ send_messages: true,
+ ..Default::default()
+ },
+ )])),
+ }),
+ )
+ .await;
+ let client = login_user(harness, USER).await;
+
+ let mut msgs = test_messages();
+ client
+ .send_messages(
+ &stream_id,
+ &topic_id,
+ &Partitioning::partition_id(0),
+ &mut msgs,
+ )
+ .await
+ .expect("send must work on the granted stream");
+
+ root_client
+ .delete_stream(&stream_id)
+ .await
+ .expect("delete the granted stream");
+ let user = root_client
+ .get_user(&Identifier::named(USER).unwrap())
+ .await
+ .expect("get user")
+ .expect("user exists");
+ assert!(
+ user.permissions
+ .and_then(|permissions| permissions.streams)
+ .is_none_or(|streams| !streams.contains_key(&granted_stream_id)),
+ "the user record must drop the grant on the deleted stream"
+ );
+
+ let reusing = create_stream_with_topic(root_client,
"perm-reusing-stream").await;
+ assert_eq!(
+ reusing.id, deleted.id,
+ "the new stream must reuse the freed id"
+ );
+ let mut msgs = test_messages();
+ assert_error_code(
+ client
+ .send_messages(
+ &stream_id,
+ &topic_id,
+ &Partitioning::partition_id(0),
+ &mut msgs,
+ )
+ .await,
+ IggyError::Unauthorized,
+ "send to the stream that reuses a deleted stream's id",
+ );
+
+ let _ = root_client.delete_stream(&stream_id).await;
+ delete_test_user(root_client, USER).await;
+}
+
+async fn create_stream_with_topic(root_client: &IggyClient, name: &str) ->
StreamDetails {
+ let stream = root_client
+ .create_stream(name)
+ .await
+ .expect("create stream");
+ root_client
+ .create_topic(
+ &Identifier::numeric(stream.id).unwrap(),
+ TOPIC_1,
+ &TopicCreateOptions {
+ partitions_count: Some(PARTITIONS),
+ message_expiry: Some(IggyExpiry::NeverExpire),
+ ..TopicCreateOptions::default()
+ },
+ )
+ .await
+ .expect("create topic");
+ stream
+}
+
+//
=============================================================================
+// Test: A grant on a stream or topic that does not exist is refused. Ids are
+// handed out lowest free, so such a grant would reach whatever is created at
+// that id later.
+//
=============================================================================
+
+async fn test_grant_on_missing_resource_refused(root_client: &IggyClient) {
Review Comment:
This and `http_rbac.rs` are the only coverage of the new contract. Nothing
in `bdd/` or the foreign SDK e2e suites asserts the 1009/2010 refusal or that a
grant is dropped when its stream or topic is deleted, although every SDK
surfaces both. One cross-SDK feature, or a follow-up issue.
--
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]
