You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by hg...@apache.org on 2026/01/15 11:01:48 UTC
(iggy) 01/01: refactor(server): consolidate permissions into metadata crate and fix inheritance bugs
This is an automated email from the ASF dual-hosted git repository.
hgruszecki pushed a commit to branch permissions-test
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 2c1e14b02eeefd411f13d4193317c6e699551699
Author: Hubert Gruszecki <h....@gmail.com>
AuthorDate: Wed Jan 14 20:29:19 2026 +0100
refactor(server): consolidate permissions into metadata crate and fix inheritance bugs
Permission checks were scattered across system layer, making enforcement
inconsistent. Moving them to handlers where auth context lives and using
metadata crate as single source of truth improves clarity.
Fixed broken inheritance - manage_stream now properly implies read_stream,
and similar fixes throughout the hierarchy.
Added resolve_* helpers to consolidate existence checks with ID resolution,
eliminating verbose repeated patterns in handlers.
Lastly, comprehensive permissions integration test was added.
---
Cargo.lock | 1 +
Cargo.toml | 1 +
.../src/client/binary_consumer_groups/mod.rs | 9 +-
core/integration/tests/server/general.rs | 3 +-
core/integration/tests/server/mod.rs | 6 +-
core/integration/tests/server/scenarios/mod.rs | 1 +
.../tests/server/scenarios/permissions_scenario.rs | 2512 ++++++++++++++++++++
core/metadata/src/lib.rs | 2 +-
core/metadata/src/permissioner/mod.rs | 8 +-
.../permissioner_rules/consumer_groups.rs | 4 -
.../permissioner_rules/consumer_offsets.rs | 4 -
.../permissioner/permissioner_rules/messages.rs | 80 +-
.../permissioner/permissioner_rules/partitions.rs | 4 -
.../permissioner/permissioner_rules/segments.rs | 4 -
.../src/permissioner/permissioner_rules/streams.rs | 4 -
.../src/permissioner/permissioner_rules/system.rs | 4 -
.../src/permissioner/permissioner_rules/topics.rs | 72 +-
.../src/permissioner/permissioner_rules/users.rs | 4 -
core/metadata/src/stm/user.rs | 6 +-
core/server/Cargo.toml | 1 +
.../cluster/get_cluster_metadata_handler.rs | 6 +-
.../create_consumer_group_handler.rs | 11 +-
.../delete_consumer_group_handler.rs | 26 +-
.../consumer_groups/get_consumer_group_handler.rs | 32 +-
.../consumer_groups/get_consumer_groups_handler.rs | 15 +-
.../consumer_groups/join_consumer_group_handler.rs | 10 +-
.../leave_consumer_group_handler.rs | 13 +-
.../delete_consumer_offset_handler.rs | 8 +-
.../get_consumer_offset_handler.rs | 8 +-
.../store_consumer_offset_handler.rs | 8 +-
.../handlers/messages/send_messages_handler.rs | 20 +-
.../partitions/create_partitions_handler.rs | 14 +-
.../partitions/delete_partitions_handler.rs | 14 +-
.../create_personal_access_token_handler.rs | 5 +-
.../delete_personal_access_token_handler.rs | 5 +-
.../get_personal_access_tokens_handler.rs | 7 +-
.../handlers/segments/delete_segments_handler.rs | 22 +-
.../handlers/streams/create_stream_handler.rs | 6 +-
.../handlers/streams/delete_stream_handler.rs | 15 +-
.../binary/handlers/streams/get_stream_handler.rs | 31 +-
.../handlers/streams/purge_stream_handler.rs | 8 +-
.../handlers/streams/update_stream_handler.rs | 13 +-
.../binary/handlers/system/get_client_handler.rs | 11 +-
.../binary/handlers/system/get_clients_handler.rs | 10 +-
.../src/binary/handlers/system/get_me_handler.rs | 11 +-
.../binary/handlers/system/get_snapshot_handler.rs | 2 +-
.../binary/handlers/system/get_stats_handler.rs | 4 +
.../binary/handlers/topics/create_topic_handler.rs | 7 +-
.../binary/handlers/topics/delete_topic_handler.rs | 7 +-
.../binary/handlers/topics/get_topic_handler.rs | 35 +-
.../binary/handlers/topics/get_topics_handler.rs | 9 +-
.../binary/handlers/topics/purge_topic_handler.rs | 7 +-
.../binary/handlers/topics/update_topic_handler.rs | 6 +-
.../handlers/users/change_password_handler.rs | 26 +-
.../binary/handlers/users/create_user_handler.rs | 6 +-
.../binary/handlers/users/delete_user_handler.rs | 6 +-
.../src/binary/handlers/users/get_user_handler.rs | 10 +-
.../src/binary/handlers/users/get_users_handler.rs | 10 +-
.../handlers/users/update_permissions_handler.rs | 19 +-
.../binary/handlers/users/update_user_handler.rs | 23 +-
core/server/src/http/consumer_groups.rs | 96 +-
core/server/src/http/consumer_offsets.rs | 46 +-
core/server/src/http/http_shard_wrapper.rs | 79 +-
core/server/src/http/messages.rs | 6 +-
core/server/src/http/partitions.rs | 27 +-
core/server/src/http/personal_access_tokens.rs | 14 +-
core/server/src/http/streams.rs | 78 +-
core/server/src/http/system.rs | 67 +-
core/server/src/http/topics.rs | 94 +-
core/server/src/http/users.rs | 98 +-
core/server/src/shard/handlers.rs | 71 +-
core/server/src/shard/mod.rs | 10 +-
core/server/src/shard/system/clients.rs | 35 +-
core/server/src/shard/system/cluster.rs | 11 +-
core/server/src/shard/system/consumer_groups.rs | 114 +-
core/server/src/shard/system/consumer_offsets.rs | 67 +-
core/server/src/shard/system/partitions.rs | 43 -
.../src/shard/system/personal_access_tokens.rs | 13 +-
core/server/src/shard/system/snapshot/mod.rs | 2 -
core/server/src/shard/system/streams.rs | 79 +-
core/server/src/shard/system/topics.rs | 68 -
core/server/src/shard/system/users.rs | 100 +-
core/server/src/shard/system/utils.rs | 116 +-
core/server/src/streaming/users/mod.rs | 2 -
core/server/src/streaming/users/permissioner.rs | 101 -
.../users/permissioner_rules/consumer_groups.rs | 76 -
.../users/permissioner_rules/consumer_offsets.rs | 49 -
.../streaming/users/permissioner_rules/messages.rs | 135 --
.../src/streaming/users/permissioner_rules/mod.rs | 27 -
.../users/permissioner_rules/partitions.rs | 40 -
.../streaming/users/permissioner_rules/segments.rs | 30 -
.../streaming/users/permissioner_rules/streams.rs | 86 -
.../streaming/users/permissioner_rules/system.rs | 44 -
.../streaming/users/permissioner_rules/topics.rs | 153 --
.../streaming/users/permissioner_rules/users.rs | 70 -
95 files changed, 3349 insertions(+), 2034 deletions(-)
diff --git a/Cargo.lock b/Cargo.lock
index a9e03e889..eb9e55d18 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -8329,6 +8329,7 @@ dependencies = [
"jsonwebtoken",
"left-right",
"lending-iterator",
+ "metadata",
"mimalloc",
"mime_guess",
"moka",
diff --git a/Cargo.toml b/Cargo.toml
index 3b68471ce..55019fc84 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -143,6 +143,7 @@ integration = { path = "core/integration" }
keyring = { version = "3.6.3", features = ["sync-secret-service", "vendored"] }
lazy_static = "1.5.0"
log = "0.4.29"
+metadata = { path = "core/metadata" }
mimalloc = "0.1"
mockall = "0.14.0"
nix = { version = "0.30.1", features = ["fs", "resource", "sched"] }
diff --git a/core/binary_protocol/src/client/binary_consumer_groups/mod.rs b/core/binary_protocol/src/client/binary_consumer_groups/mod.rs
index 6495665fc..b3a80b83e 100644
--- a/core/binary_protocol/src/client/binary_consumer_groups/mod.rs
+++ b/core/binary_protocol/src/client/binary_consumer_groups/mod.rs
@@ -37,13 +37,18 @@ impl<B: BinaryClient> ConsumerGroupClient for B {
group_id: &Identifier,
) -> Result<Option<ConsumerGroupDetails>, IggyError> {
fail_if_not_authenticated(self).await?;
- let response = self
+ let response = match self
.send_with_response(&GetConsumerGroup {
stream_id: stream_id.clone(),
topic_id: topic_id.clone(),
group_id: group_id.clone(),
})
- .await?;
+ .await
+ {
+ Ok(response) => response,
+ Err(IggyError::ConsumerGroupIdNotFound(_, _)) => return Ok(None),
+ Err(e) => return Err(e),
+ };
if response.is_empty() {
return Ok(None);
}
diff --git a/core/integration/tests/server/general.rs b/core/integration/tests/server/general.rs
index f2bb3de10..50277cc3c 100644
--- a/core/integration/tests/server/general.rs
+++ b/core/integration/tests/server/general.rs
@@ -17,7 +17,7 @@
use crate::server::{
ScenarioFn, authentication_scenario, bench_scenario, consumer_timestamp_polling_scenario,
- create_message_payload_scenario, message_headers_scenario, run_scenario,
+ create_message_payload_scenario, message_headers_scenario, permissions_scenario, run_scenario,
stream_size_validation_scenario, system_scenario, user_scenario,
};
use iggy_common::TransportProtocol;
@@ -30,6 +30,7 @@ use test_case::test_matrix;
authentication_scenario(),
system_scenario(),
user_scenario(),
+ permissions_scenario(),
message_headers_scenario(),
create_message_payload_scenario(),
stream_size_validation_scenario(),
diff --git a/core/integration/tests/server/mod.rs b/core/integration/tests/server/mod.rs
index 0b8fd502b..e8bd5d237 100644
--- a/core/integration/tests/server/mod.rs
+++ b/core/integration/tests/server/mod.rs
@@ -37,7 +37,7 @@ use scenarios::{
consumer_group_with_multiple_clients_polling_messages_scenario,
consumer_group_with_single_client_polling_messages_scenario,
consumer_timestamp_polling_scenario, create_message_payload, message_headers_scenario,
- stream_size_validation_scenario, system_scenario, user_scenario,
+ permissions_scenario, stream_size_validation_scenario, system_scenario, user_scenario,
};
use std::pin::Pin;
use std::sync::{Arc, Mutex};
@@ -98,6 +98,10 @@ fn bench_scenario() -> ScenarioFn {
|factory| Box::pin(bench_scenario::run(factory))
}
+fn permissions_scenario() -> ScenarioFn {
+ |factory| Box::pin(permissions_scenario::run(factory))
+}
+
fn consumer_timestamp_polling_scenario() -> ScenarioFn {
|factory| Box::pin(consumer_timestamp_polling_scenario::run(factory))
}
diff --git a/core/integration/tests/server/scenarios/mod.rs b/core/integration/tests/server/scenarios/mod.rs
index 1676cefa3..0c83470cb 100644
--- a/core/integration/tests/server/scenarios/mod.rs
+++ b/core/integration/tests/server/scenarios/mod.rs
@@ -30,6 +30,7 @@ pub mod delete_segments_scenario;
pub mod encryption_scenario;
pub mod message_headers_scenario;
pub mod message_size_scenario;
+pub mod permissions_scenario;
pub mod stale_client_consumer_group_scenario;
pub mod stream_size_validation_scenario;
pub mod system_scenario;
diff --git a/core/integration/tests/server/scenarios/permissions_scenario.rs b/core/integration/tests/server/scenarios/permissions_scenario.rs
new file mode 100644
index 000000000..178a3512d
--- /dev/null
+++ b/core/integration/tests/server/scenarios/permissions_scenario.rs
@@ -0,0 +1,2512 @@
+/* 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.
+ */
+
+//! Comprehensive permissions scenario test.
+//!
+//! Tests the full permission model including:
+//! - Global permissions (servers, users, streams, topics, messages)
+//! - Stream-specific permissions
+//! - Topic-specific permissions
+//! - Permission inheritance hierarchy
+//! - Positive cases (permission grants access)
+//! - Negative cases (missing permission denies access)
+
+use crate::server::scenarios::create_client;
+use ahash::AHashMap;
+use bytes::Bytes;
+use iggy::prelude::*;
+use integration::test_server::{ClientFactory, login_root};
+
+const STREAM_1: &str = "perm-stream-1";
+const STREAM_2: &str = "perm-stream-2";
+const TOPIC_1: &str = "perm-topic-1";
+const TOPIC_2: &str = "perm-topic-2";
+const PARTITIONS: u32 = 1;
+
+pub async fn run(client_factory: &dyn ClientFactory) {
+ let root_client = create_client(client_factory).await;
+ login_root(&root_client).await;
+
+ setup_test_resources(&root_client).await;
+
+ // Test categories
+ test_no_permissions(client_factory, &root_client).await;
+ test_system_permissions(client_factory, &root_client).await;
+ test_user_permissions(client_factory, &root_client).await;
+ test_stream_permissions(client_factory, &root_client).await;
+ test_topic_permissions(client_factory, &root_client).await;
+ test_message_permissions(client_factory, &root_client).await;
+ test_stream_specific_permissions(client_factory, &root_client).await;
+ test_topic_specific_permissions(client_factory, &root_client).await;
+
+ // Permission inheritance/implication tests
+ test_global_permission_inheritance(client_factory, &root_client).await;
+ test_stream_permission_inheritance(client_factory, &root_client).await;
+ test_topic_permission_inheritance(client_factory, &root_client).await;
+
+ // Consumer group operations matrix
+ test_consumer_group_operations(client_factory, &root_client).await;
+
+ // Union semantics tests
+ test_union_semantics(client_factory, &root_client).await;
+
+ // Missing resource behavior tests
+ test_missing_resource_behavior(client_factory, &root_client).await;
+
+ cleanup(&root_client).await;
+}
+
+async fn setup_test_resources(root_client: &IggyClient) {
+ root_client
+ .create_stream(STREAM_1)
+ .await
+ .expect("create stream 1");
+
+ root_client
+ .create_stream(STREAM_2)
+ .await
+ .expect("create stream 2");
+
+ for stream_name in [STREAM_1, STREAM_2] {
+ let stream_id = Identifier::named(stream_name).unwrap();
+ for topic_name in [TOPIC_1, TOPIC_2] {
+ root_client
+ .create_topic(
+ &stream_id,
+ topic_name,
+ PARTITIONS,
+ CompressionAlgorithm::None,
+ None,
+ IggyExpiry::NeverExpire,
+ MaxTopicSize::ServerDefault,
+ )
+ .await
+ .expect("create topic");
+ }
+ }
+}
+
+async fn cleanup(root_client: &IggyClient) {
+ for stream_name in [STREAM_1, STREAM_2] {
+ let _ = root_client
+ .delete_stream(&Identifier::named(stream_name).unwrap())
+ .await;
+ }
+}
+
+// =============================================================================
+// User Creation Helpers
+// =============================================================================
+
+async fn create_test_user(
+ root_client: &IggyClient,
+ username: &str,
+ permissions: Option<Permissions>,
+) {
+ let _ = root_client
+ .delete_user(&Identifier::named(username).unwrap())
+ .await;
+
+ root_client
+ .create_user(username, "password123", UserStatus::Active, permissions)
+ .await
+ .expect("create test user");
+}
+
+async fn delete_test_user(root_client: &IggyClient, username: &str) {
+ let _ = root_client
+ .delete_user(&Identifier::named(username).unwrap())
+ .await;
+}
+
+async fn login_user(client_factory: &dyn ClientFactory, username: &str) -> IggyClient {
+ let client = create_client(client_factory).await;
+ client
+ .login_user(username, "password123")
+ .await
+ .expect("login user");
+ client
+}
+
+fn no_permissions() -> Permissions {
+ Permissions {
+ global: GlobalPermissions::default(),
+ streams: None,
+ }
+}
+
+// =============================================================================
+// Test: User with no permissions
+// =============================================================================
+
+async fn test_no_permissions(client_factory: &dyn ClientFactory, root_client: &IggyClient) {
+ const USER: &str = "no-perms-user";
+ create_test_user(root_client, USER, Some(no_permissions())).await;
+ let client = login_user(client_factory, USER).await;
+
+ let stream_id = Identifier::named(STREAM_1).unwrap();
+ let topic_id = Identifier::named(TOPIC_1).unwrap();
+
+ // All operations should fail
+ assert_unauthorized(client.get_stats().await, "get_stats");
+ assert_unauthorized(client.get_clients().await, "get_clients");
+ assert_unauthorized(client.get_users().await, "get_users");
+ assert_unauthorized(client.get_streams().await, "get_streams");
+ assert_unauthorized(client.get_stream(&stream_id).await, "get_stream");
+ assert_unauthorized(client.get_topics(&stream_id).await, "get_topics");
+ assert_unauthorized(client.get_topic(&stream_id, &topic_id).await, "get_topic");
+ assert_unauthorized(client.create_stream("x").await, "create_stream");
+ assert_unauthorized(
+ client
+ .create_topic(
+ &stream_id,
+ "x",
+ 1,
+ CompressionAlgorithm::None,
+ None,
+ IggyExpiry::NeverExpire,
+ MaxTopicSize::ServerDefault,
+ )
+ .await,
+ "create_topic",
+ );
+
+ let mut msgs = test_messages();
+ assert_unauthorized(
+ client
+ .send_messages(
+ &stream_id,
+ &topic_id,
+ &Partitioning::partition_id(0),
+ &mut msgs,
+ )
+ .await,
+ "send_messages",
+ );
+ assert_unauthorized(
+ client
+ .poll_messages(
+ &stream_id,
+ &topic_id,
+ Some(0),
+ &Consumer::default(),
+ &PollingStrategy::offset(0),
+ 1,
+ false,
+ )
+ .await,
+ "poll_messages",
+ );
+
+ delete_test_user(root_client, USER).await;
+}
+
+// =============================================================================
+// Test: System permissions (read_servers, manage_servers)
+// =============================================================================
+
+async fn test_system_permissions(client_factory: &dyn ClientFactory, root_client: &IggyClient) {
+ // User with read_servers only
+ const READ_USER: &str = "read-servers-user";
+ create_test_user(
+ root_client,
+ READ_USER,
+ Some(Permissions {
+ global: GlobalPermissions {
+ read_servers: true,
+ ..Default::default()
+ },
+ streams: None,
+ }),
+ )
+ .await;
+
+ let client = login_user(client_factory, READ_USER).await;
+ client
+ .get_stats()
+ .await
+ .expect("read_servers: get_stats should work");
+ client
+ .get_clients()
+ .await
+ .expect("read_servers: get_clients should work");
+
+ // But cannot read users or streams
+ assert_unauthorized(client.get_users().await, "read_servers: get_users");
+ assert_unauthorized(client.get_streams().await, "read_servers: get_streams");
+
+ delete_test_user(root_client, READ_USER).await;
+
+ // User with manage_servers (implies read_servers capabilities)
+ const MANAGE_USER: &str = "manage-servers-user";
+ create_test_user(
+ root_client,
+ MANAGE_USER,
+ Some(Permissions {
+ global: GlobalPermissions {
+ manage_servers: true,
+ ..Default::default()
+ },
+ streams: None,
+ }),
+ )
+ .await;
+
+ let client = login_user(client_factory, MANAGE_USER).await;
+ client
+ .get_stats()
+ .await
+ .expect("manage_servers: get_stats should work");
+ client
+ .get_clients()
+ .await
+ .expect("manage_servers: get_clients should work");
+
+ delete_test_user(root_client, MANAGE_USER).await;
+}
+
+// =============================================================================
+// Test: User permissions (read_users, manage_users)
+// =============================================================================
+
+async fn test_user_permissions(client_factory: &dyn ClientFactory, root_client: &IggyClient) {
+ // User with read_users only
+ const READ_USER: &str = "read-users-user";
+ create_test_user(
+ root_client,
+ READ_USER,
+ Some(Permissions {
+ global: GlobalPermissions {
+ read_users: true,
+ ..Default::default()
+ },
+ streams: None,
+ }),
+ )
+ .await;
+
+ let client = login_user(client_factory, READ_USER).await;
+ client
+ .get_users()
+ .await
+ .expect("read_users: get_users should work");
+ client
+ .get_user(&Identifier::numeric(0).unwrap())
+ .await
+ .expect("read_users: get_user should work");
+
+ // Cannot create/delete users
+ assert_unauthorized(
+ client
+ .create_user(
+ "test-user-to-create",
+ "validpassword123",
+ UserStatus::Active,
+ None,
+ )
+ .await,
+ "read_users: create_user",
+ );
+ assert_unauthorized(
+ client
+ .delete_user(&Identifier::named(READ_USER).unwrap())
+ .await,
+ "read_users: delete_user",
+ );
+
+ delete_test_user(root_client, READ_USER).await;
+
+ // User with manage_users
+ const MANAGE_USER: &str = "manage-users-user";
+ create_test_user(
+ root_client,
+ MANAGE_USER,
+ Some(Permissions {
+ global: GlobalPermissions {
+ manage_users: true,
+ ..Default::default()
+ },
+ streams: None,
+ }),
+ )
+ .await;
+
+ let client = login_user(client_factory, MANAGE_USER).await;
+ client
+ .get_users()
+ .await
+ .expect("manage_users: get_users should work");
+
+ // Create a temporary user
+ client
+ .create_user("temp-user", "temp", UserStatus::Active, None)
+ .await
+ .expect("manage_users: create_user should work");
+
+ // Update and delete it
+ client
+ .update_user(
+ &Identifier::named("temp-user").unwrap(),
+ Some("temp-user-updated"),
+ None,
+ )
+ .await
+ .expect("manage_users: update_user should work");
+
+ client
+ .delete_user(&Identifier::named("temp-user-updated").unwrap())
+ .await
+ .expect("manage_users: delete_user should work");
+
+ delete_test_user(root_client, MANAGE_USER).await;
+}
+
+// =============================================================================
+// Test: Stream permissions (read_streams, manage_streams)
+// =============================================================================
+
+async fn test_stream_permissions(client_factory: &dyn ClientFactory, root_client: &IggyClient) {
+ let stream_id = Identifier::named(STREAM_1).unwrap();
+
+ // User with read_streams only
+ const READ_USER: &str = "read-streams-user";
+ create_test_user(
+ root_client,
+ READ_USER,
+ Some(Permissions {
+ global: GlobalPermissions {
+ read_streams: true,
+ ..Default::default()
+ },
+ streams: None,
+ }),
+ )
+ .await;
+
+ let client = login_user(client_factory, READ_USER).await;
+ client
+ .get_streams()
+ .await
+ .expect("read_streams: get_streams should work");
+ client
+ .get_stream(&stream_id)
+ .await
+ .expect("read_streams: get_stream should work");
+
+ // Cannot create/update/delete streams
+ assert_unauthorized(
+ client.create_stream("x").await,
+ "read_streams: create_stream",
+ );
+ assert_unauthorized(
+ client.update_stream(&stream_id, "new-name").await,
+ "read_streams: update_stream",
+ );
+ assert_unauthorized(
+ client.delete_stream(&stream_id).await,
+ "read_streams: delete_stream",
+ );
+
+ delete_test_user(root_client, READ_USER).await;
+
+ // User with manage_streams
+ const MANAGE_USER: &str = "manage-streams-user";
+ create_test_user(
+ root_client,
+ MANAGE_USER,
+ Some(Permissions {
+ global: GlobalPermissions {
+ manage_streams: true,
+ ..Default::default()
+ },
+ streams: None,
+ }),
+ )
+ .await;
+
+ let client = login_user(client_factory, MANAGE_USER).await;
+ client
+ .get_streams()
+ .await
+ .expect("manage_streams: get_streams should work");
+
+ // Create, update, delete stream
+ client
+ .create_stream("temp-stream")
+ .await
+ .expect("manage_streams: create_stream should work");
+
+ client
+ .update_stream(&Identifier::named("temp-stream").unwrap(), "temp-stream-v2")
+ .await
+ .expect("manage_streams: update_stream should work");
+
+ client
+ .delete_stream(&Identifier::named("temp-stream-v2").unwrap())
+ .await
+ .expect("manage_streams: delete_stream should work");
+
+ // manage_streams also implies manage_topics
+ let topic_id = Identifier::named(TOPIC_1).unwrap();
+ client
+ .get_topics(&stream_id)
+ .await
+ .expect("manage_streams: get_topics should work");
+ client
+ .get_topic(&stream_id, &topic_id)
+ .await
+ .expect("manage_streams: get_topic should work");
+
+ delete_test_user(root_client, MANAGE_USER).await;
+}
+
+// =============================================================================
+// Test: Topic permissions (read_topics, manage_topics)
+// =============================================================================
+
+async fn test_topic_permissions(client_factory: &dyn ClientFactory, root_client: &IggyClient) {
+ let stream_id = Identifier::named(STREAM_1).unwrap();
+ let topic_id = Identifier::named(TOPIC_1).unwrap();
+
+ // User with read_topics only
+ const READ_USER: &str = "read-topics-user";
+ create_test_user(
+ root_client,
+ READ_USER,
+ Some(Permissions {
+ global: GlobalPermissions {
+ read_topics: true,
+ ..Default::default()
+ },
+ streams: None,
+ }),
+ )
+ .await;
+
+ let client = login_user(client_factory, READ_USER).await;
+ client
+ .get_topics(&stream_id)
+ .await
+ .expect("read_topics: get_topics should work");
+ client
+ .get_topic(&stream_id, &topic_id)
+ .await
+ .expect("read_topics: get_topic should work");
+
+ // read_topics allows consumer group operations
+ client
+ .get_consumer_groups(&stream_id, &topic_id)
+ .await
+ .expect("read_topics: get_consumer_groups should work");
+
+ // Cannot create/update/delete topics
+ assert_unauthorized(
+ client
+ .create_topic(
+ &stream_id,
+ "x",
+ 1,
+ CompressionAlgorithm::None,
+ None,
+ IggyExpiry::NeverExpire,
+ MaxTopicSize::ServerDefault,
+ )
+ .await,
+ "read_topics: create_topic",
+ );
+ assert_unauthorized(
+ client
+ .update_topic(
+ &stream_id,
+ &topic_id,
+ "new-name",
+ CompressionAlgorithm::None,
+ None,
+ IggyExpiry::NeverExpire,
+ MaxTopicSize::ServerDefault,
+ )
+ .await,
+ "read_topics: update_topic",
+ );
+
+ delete_test_user(root_client, READ_USER).await;
+
+ // User with manage_topics
+ const MANAGE_USER: &str = "manage-topics-user";
+ create_test_user(
+ root_client,
+ MANAGE_USER,
+ Some(Permissions {
+ global: GlobalPermissions {
+ manage_topics: true,
+ ..Default::default()
+ },
+ streams: None,
+ }),
+ )
+ .await;
+
+ let client = login_user(client_factory, MANAGE_USER).await;
+
+ // Create topic
+ client
+ .create_topic(
+ &stream_id,
+ "temp-topic",
+ 1,
+ CompressionAlgorithm::None,
+ None,
+ IggyExpiry::NeverExpire,
+ MaxTopicSize::ServerDefault,
+ )
+ .await
+ .expect("manage_topics: create_topic should work");
+
+ // Update topic
+ client
+ .update_topic(
+ &stream_id,
+ &Identifier::named("temp-topic").unwrap(),
+ "temp-topic-v2",
+ CompressionAlgorithm::None,
+ None,
+ IggyExpiry::NeverExpire,
+ MaxTopicSize::ServerDefault,
+ )
+ .await
+ .expect("manage_topics: update_topic should work");
+
+ // Delete topic
+ client
+ .delete_topic(&stream_id, &Identifier::named("temp-topic-v2").unwrap())
+ .await
+ .expect("manage_topics: delete_topic should work");
+
+ // Cannot create streams
+ assert_unauthorized(
+ client.create_stream("x").await,
+ "manage_topics: create_stream",
+ );
+
+ delete_test_user(root_client, MANAGE_USER).await;
+}
+
+// =============================================================================
+// Test: Message permissions (poll_messages, send_messages)
+// =============================================================================
+
+async fn test_message_permissions(client_factory: &dyn ClientFactory, root_client: &IggyClient) {
+ let stream_id = Identifier::named(STREAM_1).unwrap();
+ let topic_id = Identifier::named(TOPIC_1).unwrap();
+
+ // User with poll_messages only
+ const POLL_USER: &str = "poll-messages-user";
+ create_test_user(
+ root_client,
+ POLL_USER,
+ Some(Permissions {
+ global: GlobalPermissions {
+ poll_messages: true,
+ ..Default::default()
+ },
+ streams: None,
+ }),
+ )
+ .await;
+
+ let client = login_user(client_factory, POLL_USER).await;
+
+ // Can poll messages
+ client
+ .poll_messages(
+ &stream_id,
+ &topic_id,
+ Some(0),
+ &Consumer::default(),
+ &PollingStrategy::offset(0),
+ 1,
+ false,
+ )
+ .await
+ .expect("poll_messages: poll_messages should work");
+
+ // Cannot send messages
+ let mut msgs = test_messages();
+ assert_unauthorized(
+ client
+ .send_messages(
+ &stream_id,
+ &topic_id,
+ &Partitioning::partition_id(0),
+ &mut msgs,
+ )
+ .await,
+ "poll_messages: send_messages",
+ );
+
+ // Cannot read streams/topics without those permissions
+ assert_unauthorized(client.get_streams().await, "poll_messages: get_streams");
+
+ delete_test_user(root_client, POLL_USER).await;
+
+ // User with send_messages only
+ const SEND_USER: &str = "send-messages-user";
+ create_test_user(
+ root_client,
+ SEND_USER,
+ Some(Permissions {
+ global: GlobalPermissions {
+ send_messages: true,
+ ..Default::default()
+ },
+ streams: None,
+ }),
+ )
+ .await;
+
+ let client = login_user(client_factory, SEND_USER).await;
+
+ // Can send messages
+ let mut msgs = test_messages();
+ client
+ .send_messages(
+ &stream_id,
+ &topic_id,
+ &Partitioning::partition_id(0),
+ &mut msgs,
+ )
+ .await
+ .expect("send_messages: send_messages should work");
+
+ // Cannot poll messages
+ assert_unauthorized(
+ client
+ .poll_messages(
+ &stream_id,
+ &topic_id,
+ Some(0),
+ &Consumer::default(),
+ &PollingStrategy::offset(0),
+ 1,
+ false,
+ )
+ .await,
+ "send_messages: poll_messages",
+ );
+
+ delete_test_user(root_client, SEND_USER).await;
+
+ // User with both poll and send
+ const BOTH_USER: &str = "poll-send-user";
+ create_test_user(
+ root_client,
+ BOTH_USER,
+ Some(Permissions {
+ global: GlobalPermissions {
+ poll_messages: true,
+ send_messages: true,
+ ..Default::default()
+ },
+ streams: None,
+ }),
+ )
+ .await;
+
+ let client = login_user(client_factory, BOTH_USER).await;
+
+ let mut msgs = test_messages();
+ client
+ .send_messages(
+ &stream_id,
+ &topic_id,
+ &Partitioning::partition_id(0),
+ &mut msgs,
+ )
+ .await
+ .expect("both: send_messages should work");
+
+ client
+ .poll_messages(
+ &stream_id,
+ &topic_id,
+ Some(0),
+ &Consumer::default(),
+ &PollingStrategy::offset(0),
+ 1,
+ false,
+ )
+ .await
+ .expect("both: poll_messages should work");
+
+ delete_test_user(root_client, BOTH_USER).await;
+}
+
+// =============================================================================
+// Test: Stream-specific permissions
+// =============================================================================
+
+async fn test_stream_specific_permissions(
+ client_factory: &dyn ClientFactory,
+ root_client: &IggyClient,
+) {
+ let stream1_id = Identifier::named(STREAM_1).unwrap();
+ let stream2_id = Identifier::named(STREAM_2).unwrap();
+ let topic_id = Identifier::named(TOPIC_1).unwrap();
+
+ // Get the actual stream IDs from the server
+ let streams = root_client.get_streams().await.expect("get streams");
+ let stream1_numeric_id = streams
+ .iter()
+ .find(|s| s.name == STREAM_1)
+ .map(|s| s.id as usize)
+ .expect("stream 1 should exist");
+
+ // User with permissions only for stream 1
+ const USER: &str = "stream-specific-user";
+ create_test_user(
+ root_client,
+ USER,
+ Some(Permissions {
+ global: GlobalPermissions::default(),
+ streams: Some(AHashMap::from([(
+ stream1_numeric_id,
+ StreamPermissions {
+ read_stream: true,
+ manage_topics: true,
+ poll_messages: true,
+ send_messages: true,
+ ..Default::default()
+ },
+ )])),
+ }),
+ )
+ .await;
+
+ let client = login_user(client_factory, USER).await;
+
+ // Can access stream 1
+ client
+ .get_stream(&stream1_id)
+ .await
+ .expect("stream-specific: get_stream 1 should work");
+ client
+ .get_topics(&stream1_id)
+ .await
+ .expect("stream-specific: get_topics 1 should work");
+ client
+ .get_topic(&stream1_id, &topic_id)
+ .await
+ .expect("stream-specific: get_topic 1 should work");
+
+ // Can send/poll messages on stream 1
+ let mut msgs = test_messages();
+ client
+ .send_messages(
+ &stream1_id,
+ &topic_id,
+ &Partitioning::partition_id(0),
+ &mut msgs,
+ )
+ .await
+ .expect("stream-specific: send_messages 1 should work");
+
+ client
+ .poll_messages(
+ &stream1_id,
+ &topic_id,
+ Some(0),
+ &Consumer::default(),
+ &PollingStrategy::offset(0),
+ 1,
+ false,
+ )
+ .await
+ .expect("stream-specific: poll_messages 1 should work");
+
+ // Cannot access stream 2
+ assert_unauthorized(
+ client.get_stream(&stream2_id).await,
+ "stream-specific: get_stream 2",
+ );
+ assert_unauthorized(
+ client.get_topics(&stream2_id).await,
+ "stream-specific: get_topics 2",
+ );
+ assert_unauthorized(
+ client.get_topic(&stream2_id, &topic_id).await,
+ "stream-specific: get_topic 2",
+ );
+
+ let mut msgs = test_messages();
+ assert_unauthorized(
+ client
+ .send_messages(
+ &stream2_id,
+ &topic_id,
+ &Partitioning::partition_id(0),
+ &mut msgs,
+ )
+ .await,
+ "stream-specific: send_messages 2",
+ );
+ assert_unauthorized(
+ client
+ .poll_messages(
+ &stream2_id,
+ &topic_id,
+ Some(0),
+ &Consumer::default(),
+ &PollingStrategy::offset(0),
+ 1,
+ false,
+ )
+ .await,
+ "stream-specific: poll_messages 2",
+ );
+
+ // Cannot get all streams (no global read_streams)
+ assert_unauthorized(client.get_streams().await, "stream-specific: get_streams");
+
+ delete_test_user(root_client, USER).await;
+}
+
+// =============================================================================
+// Test: Topic-specific permissions
+// =============================================================================
+
+async fn test_topic_specific_permissions(
+ client_factory: &dyn ClientFactory,
+ root_client: &IggyClient,
+) {
+ let stream_id = Identifier::named(STREAM_1).unwrap();
+ let topic1_id = Identifier::named(TOPIC_1).unwrap();
+ let topic2_id = Identifier::named(TOPIC_2).unwrap();
+
+ // Get the actual IDs from the server
+ let streams = root_client.get_streams().await.expect("get streams");
+ let stream1_numeric_id = streams
+ .iter()
+ .find(|s| s.name == STREAM_1)
+ .map(|s| s.id as usize)
+ .expect("stream 1 should exist");
+
+ let topics = root_client
+ .get_topics(&stream_id)
+ .await
+ .expect("get topics");
+ let topic1_numeric_id = topics
+ .iter()
+ .find(|t| t.name == TOPIC_1)
+ .map(|t| t.id as usize)
+ .expect("topic 1 should exist");
+
+ // User with permissions only for topic 1 in stream 1
+ const USER: &str = "topic-specific-user";
+ create_test_user(
+ root_client,
+ USER,
+ Some(Permissions {
+ global: GlobalPermissions::default(),
+ streams: Some(AHashMap::from([(
+ stream1_numeric_id,
+ StreamPermissions {
+ topics: Some(AHashMap::from([(
+ topic1_numeric_id,
+ TopicPermissions {
+ read_topic: true,
+ poll_messages: true,
+ send_messages: true,
+ ..Default::default()
+ },
+ )])),
+ ..Default::default()
+ },
+ )])),
+ }),
+ )
+ .await;
+
+ let client = login_user(client_factory, USER).await;
+
+ // Can access topic 1
+ client
+ .get_topic(&stream_id, &topic1_id)
+ .await
+ .expect("topic-specific: get_topic 1 should work");
+
+ // Can send/poll messages on topic 1
+ let mut msgs = test_messages();
+ client
+ .send_messages(
+ &stream_id,
+ &topic1_id,
+ &Partitioning::partition_id(0),
+ &mut msgs,
+ )
+ .await
+ .expect("topic-specific: send_messages 1 should work");
+
+ client
+ .poll_messages(
+ &stream_id,
+ &topic1_id,
+ Some(0),
+ &Consumer::default(),
+ &PollingStrategy::offset(0),
+ 1,
+ false,
+ )
+ .await
+ .expect("topic-specific: poll_messages 1 should work");
+
+ // Cannot access topic 2
+ assert_unauthorized(
+ client.get_topic(&stream_id, &topic2_id).await,
+ "topic-specific: get_topic 2",
+ );
+
+ let mut msgs = test_messages();
+ assert_unauthorized(
+ client
+ .send_messages(
+ &stream_id,
+ &topic2_id,
+ &Partitioning::partition_id(0),
+ &mut msgs,
+ )
+ .await,
+ "topic-specific: send_messages 2",
+ );
+ assert_unauthorized(
+ client
+ .poll_messages(
+ &stream_id,
+ &topic2_id,
+ Some(0),
+ &Consumer::default(),
+ &PollingStrategy::offset(0),
+ 1,
+ false,
+ )
+ .await,
+ "topic-specific: poll_messages 2",
+ );
+
+ // Cannot list all topics (no stream-level read_topics)
+ assert_unauthorized(
+ client.get_topics(&stream_id).await,
+ "topic-specific: get_topics",
+ );
+
+ delete_test_user(root_client, USER).await;
+}
+
+// =============================================================================
+// Test: Global permission inheritance/implication
+// Tests that higher-level permissions imply lower-level ones WITHOUT explicitly
+// setting the implied flags. This locks in the inheritance rules.
+// =============================================================================
+
+async fn test_global_permission_inheritance(
+ client_factory: &dyn ClientFactory,
+ root_client: &IggyClient,
+) {
+ let stream_id = Identifier::named(STREAM_1).unwrap();
+ let topic_id = Identifier::named(TOPIC_1).unwrap();
+
+ // Test: global read_streams implies read_topics, poll_messages, and CG ops
+ // Doc: "read_streams permission allows to read the streams and includes all the permissions of read_topics"
+ // Doc: "read_topics ... includes all the permissions of poll_messages" + CG ops
+ const READ_STREAMS_USER: &str = "inherit-read-streams-user";
+ create_test_user(
+ root_client,
+ READ_STREAMS_USER,
+ Some(Permissions {
+ global: GlobalPermissions {
+ read_streams: true,
+ ..Default::default()
+ },
+ streams: None,
+ }),
+ )
+ .await;
+
+ let client = login_user(client_factory, READ_STREAMS_USER).await;
+
+ // read_streams → read_topics
+ client
+ .get_topics(&stream_id)
+ .await
+ .expect("read_streams → read_topics: get_topics");
+ client
+ .get_topic(&stream_id, &topic_id)
+ .await
+ .expect("read_streams → read_topics: get_topic");
+
+ // read_streams → read_topics → poll_messages
+ client
+ .poll_messages(
+ &stream_id,
+ &topic_id,
+ Some(0),
+ &Consumer::default(),
+ &PollingStrategy::offset(0),
+ 1,
+ false,
+ )
+ .await
+ .expect("read_streams → poll_messages");
+
+ // read_streams → read_topics → CG ops
+ client
+ .get_consumer_groups(&stream_id, &topic_id)
+ .await
+ .expect("read_streams → CG ops");
+
+ // send_messages NOT implied
+ let mut msgs = test_messages();
+ assert_unauthorized(
+ client
+ .send_messages(
+ &stream_id,
+ &topic_id,
+ &Partitioning::partition_id(0),
+ &mut msgs,
+ )
+ .await,
+ "read_streams does NOT imply send_messages",
+ );
+
+ // manage operations NOT implied
+ assert_unauthorized(
+ client.create_stream("x").await,
+ "read_streams does NOT imply manage_streams",
+ );
+ assert_unauthorized(
+ client
+ .create_topic(
+ &stream_id,
+ "x",
+ 1,
+ CompressionAlgorithm::None,
+ None,
+ IggyExpiry::NeverExpire,
+ MaxTopicSize::ServerDefault,
+ )
+ .await,
+ "read_streams does NOT imply manage_topics",
+ );
+
+ delete_test_user(root_client, READ_STREAMS_USER).await;
+
+ // Test: global manage_streams implies read_streams, manage_topics, read_topics, poll_messages, CG ops
+ // Doc: "manage_streams permission ... includes all the permissions of read_streams"
+ // Doc: "Also, it allows to manage all the topics of a stream, thus it has all the permissions of manage_topics"
+ const MANAGE_STREAMS_USER: &str = "inherit-manage-streams-user";
+ create_test_user(
+ root_client,
+ MANAGE_STREAMS_USER,
+ Some(Permissions {
+ global: GlobalPermissions {
+ manage_streams: true,
+ ..Default::default()
+ },
+ streams: None,
+ }),
+ )
+ .await;
+
+ let client = login_user(client_factory, MANAGE_STREAMS_USER).await;
+
+ // manage_streams → read_streams
+ client
+ .get_streams()
+ .await
+ .expect("manage_streams → read_streams: get_streams");
+ client
+ .get_stream(&stream_id)
+ .await
+ .expect("manage_streams → read_streams: get_stream");
+
+ // manage_streams → manage_topics (topic CRUD)
+ client
+ .create_topic(
+ &stream_id,
+ "temp-inherit-topic",
+ 1,
+ CompressionAlgorithm::None,
+ None,
+ IggyExpiry::NeverExpire,
+ MaxTopicSize::ServerDefault,
+ )
+ .await
+ .expect("manage_streams → manage_topics: create_topic");
+ client
+ .delete_topic(
+ &stream_id,
+ &Identifier::named("temp-inherit-topic").unwrap(),
+ )
+ .await
+ .expect("manage_streams → manage_topics: delete_topic");
+
+ // manage_streams → read_topics
+ client
+ .get_topics(&stream_id)
+ .await
+ .expect("manage_streams → read_topics: get_topics");
+
+ // manage_streams → read_streams → read_topics → poll_messages
+ client
+ .poll_messages(
+ &stream_id,
+ &topic_id,
+ Some(0),
+ &Consumer::default(),
+ &PollingStrategy::offset(0),
+ 1,
+ false,
+ )
+ .await
+ .expect("manage_streams → poll_messages");
+
+ // manage_streams → read_topics → CG ops
+ client
+ .get_consumer_groups(&stream_id, &topic_id)
+ .await
+ .expect("manage_streams → CG ops");
+
+ // send_messages NOT implied
+ let mut msgs = test_messages();
+ assert_unauthorized(
+ client
+ .send_messages(
+ &stream_id,
+ &topic_id,
+ &Partitioning::partition_id(0),
+ &mut msgs,
+ )
+ .await,
+ "manage_streams does NOT imply send_messages",
+ );
+
+ delete_test_user(root_client, MANAGE_STREAMS_USER).await;
+
+ // Test: global manage_topics implies read_topics, poll_messages, CG ops
+ // Doc: "manage_topics permission allows to manage the topics and includes all the permissions of read_topics"
+ const MANAGE_TOPICS_USER: &str = "inherit-manage-topics-user";
+ create_test_user(
+ root_client,
+ MANAGE_TOPICS_USER,
+ Some(Permissions {
+ global: GlobalPermissions {
+ manage_topics: true,
+ ..Default::default()
+ },
+ streams: None,
+ }),
+ )
+ .await;
+
+ let client = login_user(client_factory, MANAGE_TOPICS_USER).await;
+
+ // manage_topics → read_topics
+ client
+ .get_topics(&stream_id)
+ .await
+ .expect("manage_topics → read_topics: get_topics");
+ client
+ .get_topic(&stream_id, &topic_id)
+ .await
+ .expect("manage_topics → read_topics: get_topic");
+
+ // manage_topics → read_topics → poll_messages
+ client
+ .poll_messages(
+ &stream_id,
+ &topic_id,
+ Some(0),
+ &Consumer::default(),
+ &PollingStrategy::offset(0),
+ 1,
+ false,
+ )
+ .await
+ .expect("manage_topics → poll_messages");
+
+ // manage_topics → read_topics → CG ops
+ client
+ .get_consumer_groups(&stream_id, &topic_id)
+ .await
+ .expect("manage_topics → CG ops");
+
+ // read_streams NOT implied
+ assert_unauthorized(
+ client.get_streams().await,
+ "manage_topics does NOT imply read_streams",
+ );
+
+ delete_test_user(root_client, MANAGE_TOPICS_USER).await;
+
+ // Test: global read_topics implies poll_messages and CG ops
+ // Doc: "read_topics permission allows to read the topics, manage consumer groups, and includes all the permissions of poll_messages"
+ const READ_TOPICS_USER: &str = "inherit-read-topics-user";
+ create_test_user(
+ root_client,
+ READ_TOPICS_USER,
+ Some(Permissions {
+ global: GlobalPermissions {
+ read_topics: true,
+ ..Default::default()
+ },
+ streams: None,
+ }),
+ )
+ .await;
+
+ let client = login_user(client_factory, READ_TOPICS_USER).await;
+
+ // read_topics allows reading topics
+ client
+ .get_topics(&stream_id)
+ .await
+ .expect("read_topics: get_topics");
+ client
+ .get_topic(&stream_id, &topic_id)
+ .await
+ .expect("read_topics: get_topic");
+
+ // read_topics → poll_messages
+ client
+ .poll_messages(
+ &stream_id,
+ &topic_id,
+ Some(0),
+ &Consumer::default(),
+ &PollingStrategy::offset(0),
+ 1,
+ false,
+ )
+ .await
+ .expect("read_topics → poll_messages");
+
+ // read_topics → CG ops
+ client
+ .get_consumer_groups(&stream_id, &topic_id)
+ .await
+ .expect("read_topics → CG ops");
+
+ delete_test_user(root_client, READ_TOPICS_USER).await;
+
+ // Test: manage_servers implies read_servers
+ const MANAGE_SERVERS_ONLY_USER: &str = "inherit-manage-servers-user";
+ create_test_user(
+ root_client,
+ MANAGE_SERVERS_ONLY_USER,
+ Some(Permissions {
+ global: GlobalPermissions {
+ manage_servers: true,
+ // NOT setting read_servers
+ ..Default::default()
+ },
+ streams: None,
+ }),
+ )
+ .await;
+
+ let client = login_user(client_factory, MANAGE_SERVERS_ONLY_USER).await;
+
+ // read_servers implied
+ client
+ .get_stats()
+ .await
+ .expect("manage_servers implies read_servers: get_stats should work");
+ client
+ .get_clients()
+ .await
+ .expect("manage_servers implies read_servers: get_clients should work");
+
+ delete_test_user(root_client, MANAGE_SERVERS_ONLY_USER).await;
+
+ // Test: manage_users implies read_users
+ const MANAGE_USERS_ONLY_USER: &str = "inherit-manage-users-user";
+ create_test_user(
+ root_client,
+ MANAGE_USERS_ONLY_USER,
+ Some(Permissions {
+ global: GlobalPermissions {
+ manage_users: true,
+ // NOT setting read_users
+ ..Default::default()
+ },
+ streams: None,
+ }),
+ )
+ .await;
+
+ let client = login_user(client_factory, MANAGE_USERS_ONLY_USER).await;
+
+ // read_users implied
+ client
+ .get_users()
+ .await
+ .expect("manage_users implies read_users: get_users should work");
+ client
+ .get_user(&Identifier::numeric(0).unwrap())
+ .await
+ .expect("manage_users implies read_users: get_user should work");
+
+ delete_test_user(root_client, MANAGE_USERS_ONLY_USER).await;
+}
+
+// =============================================================================
+// Test: Stream-scoped permission inheritance
+// Tests inheritance rules at the stream level without setting implied flags.
+// =============================================================================
+
+async fn test_stream_permission_inheritance(
+ client_factory: &dyn ClientFactory,
+ root_client: &IggyClient,
+) {
+ let stream_id = Identifier::named(STREAM_1).unwrap();
+ let topic_id = Identifier::named(TOPIC_1).unwrap();
+
+ // Get stream numeric ID
+ let streams = root_client.get_streams().await.expect("get streams");
+ let stream1_numeric_id = streams
+ .iter()
+ .find(|s| s.name == STREAM_1)
+ .map(|s| s.id as usize)
+ .expect("stream 1 should exist");
+
+ // Test: stream.read_stream implies read_topics, poll_messages, CG ops
+ // Doc: "read_stream permission allows to read the stream and includes all the permissions of read_topics"
+ // Doc: "Also, it allows to read all the messages of a topic, thus it has all the permissions of poll_messages"
+ const READ_STREAM_USER: &str = "inherit-read-stream-user";
+ create_test_user(
+ root_client,
+ READ_STREAM_USER,
+ Some(Permissions {
+ global: GlobalPermissions::default(),
+ streams: Some(AHashMap::from([(
+ stream1_numeric_id,
+ StreamPermissions {
+ read_stream: true,
+ ..Default::default()
+ },
+ )])),
+ }),
+ )
+ .await;
+
+ let client = login_user(client_factory, READ_STREAM_USER).await;
+
+ // read_stream allows get_stream
+ client
+ .get_stream(&stream_id)
+ .await
+ .expect("stream.read_stream: get_stream");
+
+ // read_stream → read_topics
+ client
+ .get_topics(&stream_id)
+ .await
+ .expect("stream.read_stream → read_topics: get_topics");
+ client
+ .get_topic(&stream_id, &topic_id)
+ .await
+ .expect("stream.read_stream → read_topics: get_topic");
+
+ // read_stream → poll_messages
+ client
+ .poll_messages(
+ &stream_id,
+ &topic_id,
+ Some(0),
+ &Consumer::default(),
+ &PollingStrategy::offset(0),
+ 1,
+ false,
+ )
+ .await
+ .expect("stream.read_stream → poll_messages");
+
+ // read_stream → read_topics → CG ops
+ client
+ .get_consumer_groups(&stream_id, &topic_id)
+ .await
+ .expect("stream.read_stream → CG ops");
+
+ // send_messages NOT implied
+ let mut msgs = test_messages();
+ assert_unauthorized(
+ client
+ .send_messages(
+ &stream_id,
+ &topic_id,
+ &Partitioning::partition_id(0),
+ &mut msgs,
+ )
+ .await,
+ "stream.read_stream does NOT imply send_messages",
+ );
+
+ // manage operations NOT implied
+ assert_unauthorized(
+ client
+ .create_topic(
+ &stream_id,
+ "x",
+ 1,
+ CompressionAlgorithm::None,
+ None,
+ IggyExpiry::NeverExpire,
+ MaxTopicSize::ServerDefault,
+ )
+ .await,
+ "stream.read_stream does NOT imply manage_topics",
+ );
+
+ delete_test_user(root_client, READ_STREAM_USER).await;
+
+ // Test: stream.manage_stream implies read_stream, manage_topics, read_topics, poll_messages, CG ops
+ // Doc: "manage_stream permission allows to manage the stream and includes all the permissions of read_stream"
+ // Doc: "Also, it allows to manage all the topics of a stream, thus it has all the permissions of manage_topics"
+ const MANAGE_STREAM_USER: &str = "inherit-manage-stream-user";
+ create_test_user(
+ root_client,
+ MANAGE_STREAM_USER,
+ Some(Permissions {
+ global: GlobalPermissions::default(),
+ streams: Some(AHashMap::from([(
+ stream1_numeric_id,
+ StreamPermissions {
+ manage_stream: true,
+ ..Default::default()
+ },
+ )])),
+ }),
+ )
+ .await;
+
+ let client = login_user(client_factory, MANAGE_STREAM_USER).await;
+
+ // manage_stream → read_stream
+ client
+ .get_stream(&stream_id)
+ .await
+ .expect("stream.manage_stream → read_stream: get_stream");
+
+ // manage_stream → manage_topics (topic CRUD)
+ client
+ .create_topic(
+ &stream_id,
+ "temp-manage-stream-topic",
+ 1,
+ CompressionAlgorithm::None,
+ None,
+ IggyExpiry::NeverExpire,
+ MaxTopicSize::ServerDefault,
+ )
+ .await
+ .expect("stream.manage_stream → manage_topics: create_topic");
+ client
+ .update_topic(
+ &stream_id,
+ &Identifier::named("temp-manage-stream-topic").unwrap(),
+ "temp-manage-stream-topic-v2",
+ CompressionAlgorithm::None,
+ None,
+ IggyExpiry::NeverExpire,
+ MaxTopicSize::ServerDefault,
+ )
+ .await
+ .expect("stream.manage_stream → manage_topics: update_topic");
+ client
+ .delete_topic(
+ &stream_id,
+ &Identifier::named("temp-manage-stream-topic-v2").unwrap(),
+ )
+ .await
+ .expect("stream.manage_stream → manage_topics: delete_topic");
+
+ // manage_stream → read_topics
+ client
+ .get_topics(&stream_id)
+ .await
+ .expect("stream.manage_stream → read_topics: get_topics");
+ client
+ .get_topic(&stream_id, &topic_id)
+ .await
+ .expect("stream.manage_stream → read_topics: get_topic");
+
+ // manage_stream → read_stream → poll_messages
+ client
+ .poll_messages(
+ &stream_id,
+ &topic_id,
+ Some(0),
+ &Consumer::default(),
+ &PollingStrategy::offset(0),
+ 1,
+ false,
+ )
+ .await
+ .expect("stream.manage_stream → poll_messages");
+
+ // manage_stream → read_topics → CG ops
+ client
+ .get_consumer_groups(&stream_id, &topic_id)
+ .await
+ .expect("stream.manage_stream → CG ops");
+
+ // send_messages NOT implied
+ let mut msgs = test_messages();
+ assert_unauthorized(
+ client
+ .send_messages(
+ &stream_id,
+ &topic_id,
+ &Partitioning::partition_id(0),
+ &mut msgs,
+ )
+ .await,
+ "stream.manage_stream does NOT imply send_messages",
+ );
+
+ delete_test_user(root_client, MANAGE_STREAM_USER).await;
+
+ // Test: stream.manage_topics implies read_topics and poll_messages
+ const MANAGE_TOPICS_STREAM_USER: &str = "inherit-manage-topics-stream-user";
+ create_test_user(
+ root_client,
+ MANAGE_TOPICS_STREAM_USER,
+ Some(Permissions {
+ global: GlobalPermissions::default(),
+ streams: Some(AHashMap::from([(
+ stream1_numeric_id,
+ StreamPermissions {
+ manage_topics: true,
+ // NOT setting read_topics, poll_messages
+ ..Default::default()
+ },
+ )])),
+ }),
+ )
+ .await;
+
+ let client = login_user(client_factory, MANAGE_TOPICS_STREAM_USER).await;
+
+ // manage_topics allows topic CRUD
+ client
+ .create_topic(
+ &stream_id,
+ "temp-manage-topic",
+ 1,
+ CompressionAlgorithm::None,
+ None,
+ IggyExpiry::NeverExpire,
+ MaxTopicSize::ServerDefault,
+ )
+ .await
+ .expect("stream.manage_topics: create_topic should work");
+ client
+ .delete_topic(&stream_id, &Identifier::named("temp-manage-topic").unwrap())
+ .await
+ .expect("stream.manage_topics: delete_topic should work");
+
+ // read_topics implied
+ client
+ .get_topics(&stream_id)
+ .await
+ .expect("stream.manage_topics implies read_topics: get_topics should work");
+
+ // poll_messages implied
+ client
+ .poll_messages(
+ &stream_id,
+ &topic_id,
+ Some(0),
+ &Consumer::default(),
+ &PollingStrategy::offset(0),
+ 1,
+ false,
+ )
+ .await
+ .expect("stream.manage_topics implies poll_messages: poll_messages should work");
+
+ // read_stream NOT implied
+ assert_unauthorized(
+ client.get_stream(&stream_id).await,
+ "stream.manage_topics does NOT imply read_stream",
+ );
+
+ delete_test_user(root_client, MANAGE_TOPICS_STREAM_USER).await;
+}
+
+// =============================================================================
+// Test: Topic-scoped permission inheritance
+// Tests inheritance rules at the topic level without setting implied flags.
+// =============================================================================
+
+async fn test_topic_permission_inheritance(
+ client_factory: &dyn ClientFactory,
+ root_client: &IggyClient,
+) {
+ let stream_id = Identifier::named(STREAM_1).unwrap();
+ let topic_id = Identifier::named(TOPIC_1).unwrap();
+
+ // Get IDs
+ let streams = root_client.get_streams().await.expect("get streams");
+ let stream1_numeric_id = streams
+ .iter()
+ .find(|s| s.name == STREAM_1)
+ .map(|s| s.id as usize)
+ .expect("stream 1 should exist");
+
+ let topics = root_client
+ .get_topics(&stream_id)
+ .await
+ .expect("get topics");
+ let topic1_numeric_id = topics
+ .iter()
+ .find(|t| t.name == TOPIC_1)
+ .map(|t| t.id as usize)
+ .expect("topic 1 should exist");
+
+ // Test: topic.read_topic implies poll_messages and CG operations
+ const READ_TOPIC_USER: &str = "inherit-read-topic-user";
+ create_test_user(
+ root_client,
+ READ_TOPIC_USER,
+ Some(Permissions {
+ global: GlobalPermissions::default(),
+ streams: Some(AHashMap::from([(
+ stream1_numeric_id,
+ StreamPermissions {
+ topics: Some(AHashMap::from([(
+ topic1_numeric_id,
+ TopicPermissions {
+ read_topic: true,
+ // NOT setting poll_messages
+ ..Default::default()
+ },
+ )])),
+ ..Default::default()
+ },
+ )])),
+ }),
+ )
+ .await;
+
+ let client = login_user(client_factory, READ_TOPIC_USER).await;
+
+ // get_topic should work
+ client
+ .get_topic(&stream_id, &topic_id)
+ .await
+ .expect("topic.read_topic: get_topic should work");
+
+ // poll_messages implied
+ client
+ .poll_messages(
+ &stream_id,
+ &topic_id,
+ Some(0),
+ &Consumer::default(),
+ &PollingStrategy::offset(0),
+ 1,
+ false,
+ )
+ .await
+ .expect("topic.read_topic implies poll_messages: poll_messages should work");
+
+ // CG operations implied
+ client
+ .get_consumer_groups(&stream_id, &topic_id)
+ .await
+ .expect("topic.read_topic implies CG ops: get_consumer_groups should work");
+
+ // send_messages NOT implied
+ let mut msgs = test_messages();
+ assert_unauthorized(
+ client
+ .send_messages(
+ &stream_id,
+ &topic_id,
+ &Partitioning::partition_id(0),
+ &mut msgs,
+ )
+ .await,
+ "topic.read_topic does NOT imply send_messages",
+ );
+
+ delete_test_user(root_client, READ_TOPIC_USER).await;
+
+ // Test: topic.manage_topic implies read_topic and poll_messages
+ const MANAGE_TOPIC_USER: &str = "inherit-manage-topic-user";
+ create_test_user(
+ root_client,
+ MANAGE_TOPIC_USER,
+ Some(Permissions {
+ global: GlobalPermissions::default(),
+ streams: Some(AHashMap::from([(
+ stream1_numeric_id,
+ StreamPermissions {
+ topics: Some(AHashMap::from([(
+ topic1_numeric_id,
+ TopicPermissions {
+ manage_topic: true,
+ // NOT setting read_topic, poll_messages
+ ..Default::default()
+ },
+ )])),
+ ..Default::default()
+ },
+ )])),
+ }),
+ )
+ .await;
+
+ let client = login_user(client_factory, MANAGE_TOPIC_USER).await;
+
+ // read_topic implied
+ client
+ .get_topic(&stream_id, &topic_id)
+ .await
+ .expect("topic.manage_topic implies read_topic: get_topic should work");
+
+ // poll_messages implied
+ client
+ .poll_messages(
+ &stream_id,
+ &topic_id,
+ Some(0),
+ &Consumer::default(),
+ &PollingStrategy::offset(0),
+ 1,
+ false,
+ )
+ .await
+ .expect("topic.manage_topic implies poll_messages: poll_messages should work");
+
+ // CG operations implied
+ client
+ .get_consumer_groups(&stream_id, &topic_id)
+ .await
+ .expect("topic.manage_topic implies CG ops: get_consumer_groups should work");
+
+ delete_test_user(root_client, MANAGE_TOPIC_USER).await;
+}
+
+// =============================================================================
+// Test: Consumer group operations matrix
+// Tests all CG operations under various permission levels and verifies they
+// are denied under poll_messages-only.
+// =============================================================================
+
+async fn test_consumer_group_operations(
+ client_factory: &dyn ClientFactory,
+ root_client: &IggyClient,
+) {
+ let stream_id = Identifier::named(STREAM_1).unwrap();
+ let topic_id = Identifier::named(TOPIC_1).unwrap();
+
+ // Test: poll_messages only does NOT allow CG operations
+ const POLL_ONLY_USER: &str = "cg-poll-only-user";
+ create_test_user(
+ root_client,
+ POLL_ONLY_USER,
+ Some(Permissions {
+ global: GlobalPermissions {
+ poll_messages: true,
+ ..Default::default()
+ },
+ streams: None,
+ }),
+ )
+ .await;
+
+ let client = login_user(client_factory, POLL_ONLY_USER).await;
+
+ // poll_messages should work
+ client
+ .poll_messages(
+ &stream_id,
+ &topic_id,
+ Some(0),
+ &Consumer::default(),
+ &PollingStrategy::offset(0),
+ 1,
+ false,
+ )
+ .await
+ .expect("poll_messages: poll should work");
+
+ // CG operations should be denied
+ assert_unauthorized(
+ client.get_consumer_groups(&stream_id, &topic_id).await,
+ "poll_messages only: get_consumer_groups should be denied",
+ );
+ assert_unauthorized(
+ client
+ .get_consumer_group(&stream_id, &topic_id, &Identifier::numeric(1).unwrap())
+ .await,
+ "poll_messages only: get_consumer_group should be denied",
+ );
+ assert_unauthorized(
+ client
+ .create_consumer_group(&stream_id, &topic_id, "test-cg")
+ .await,
+ "poll_messages only: create_consumer_group should be denied",
+ );
+
+ delete_test_user(root_client, POLL_ONLY_USER).await;
+
+ // Test: read_topics allows all CG operations
+ const READ_TOPICS_CG_USER: &str = "cg-read-topics-user";
+ create_test_user(
+ root_client,
+ READ_TOPICS_CG_USER,
+ Some(Permissions {
+ global: GlobalPermissions {
+ read_topics: true,
+ ..Default::default()
+ },
+ streams: None,
+ }),
+ )
+ .await;
+
+ let client = login_user(client_factory, READ_TOPICS_CG_USER).await;
+
+ // All CG operations should work
+ client
+ .get_consumer_groups(&stream_id, &topic_id)
+ .await
+ .expect("read_topics: get_consumer_groups should work");
+
+ let cg = client
+ .create_consumer_group(&stream_id, &topic_id, "test-cg-read-topics")
+ .await
+ .expect("read_topics: create_consumer_group should work");
+
+ let cg_id = Identifier::numeric(cg.id).unwrap();
+
+ client
+ .get_consumer_group(&stream_id, &topic_id, &cg_id)
+ .await
+ .expect("read_topics: get_consumer_group should work");
+
+ assert_ok_or_feature_unavailable(
+ client
+ .join_consumer_group(&stream_id, &topic_id, &cg_id)
+ .await,
+ "read_topics: join_consumer_group should work",
+ );
+
+ assert_ok_or_feature_unavailable(
+ client
+ .leave_consumer_group(&stream_id, &topic_id, &cg_id)
+ .await,
+ "read_topics: leave_consumer_group should work",
+ );
+
+ client
+ .delete_consumer_group(&stream_id, &topic_id, &cg_id)
+ .await
+ .expect("read_topics: delete_consumer_group should work");
+
+ delete_test_user(root_client, READ_TOPICS_CG_USER).await;
+
+ // Get IDs for scoped test
+ let streams = root_client.get_streams().await.expect("get streams");
+ let stream1_numeric_id = streams
+ .iter()
+ .find(|s| s.name == STREAM_1)
+ .map(|s| s.id as usize)
+ .expect("stream 1 should exist");
+ let topics = root_client
+ .get_topics(&stream_id)
+ .await
+ .expect("get topics");
+ let topic1_numeric_id = topics
+ .iter()
+ .find(|t| t.name == TOPIC_1)
+ .map(|t| t.id as usize)
+ .expect("topic 1 should exist");
+
+ // Test: topic.read_topic allows CG operations on that topic
+ const READ_TOPIC_CG_USER: &str = "cg-read-topic-user";
+ create_test_user(
+ root_client,
+ READ_TOPIC_CG_USER,
+ Some(Permissions {
+ global: GlobalPermissions::default(),
+ streams: Some(AHashMap::from([(
+ stream1_numeric_id,
+ StreamPermissions {
+ topics: Some(AHashMap::from([(
+ topic1_numeric_id,
+ TopicPermissions {
+ read_topic: true,
+ ..Default::default()
+ },
+ )])),
+ ..Default::default()
+ },
+ )])),
+ }),
+ )
+ .await;
+
+ let client = login_user(client_factory, READ_TOPIC_CG_USER).await;
+
+ // CG operations should work on topic 1
+ client
+ .get_consumer_groups(&stream_id, &topic_id)
+ .await
+ .expect("topic.read_topic: get_consumer_groups should work");
+
+ let cg = client
+ .create_consumer_group(&stream_id, &topic_id, "test-cg-read-topic")
+ .await
+ .expect("topic.read_topic: create_consumer_group should work");
+
+ let cg_id = Identifier::numeric(cg.id).unwrap();
+
+ assert_ok_or_feature_unavailable(
+ client
+ .join_consumer_group(&stream_id, &topic_id, &cg_id)
+ .await,
+ "topic.read_topic: join_consumer_group should work",
+ );
+
+ assert_ok_or_feature_unavailable(
+ client
+ .leave_consumer_group(&stream_id, &topic_id, &cg_id)
+ .await,
+ "topic.read_topic: leave_consumer_group should work",
+ );
+
+ client
+ .delete_consumer_group(&stream_id, &topic_id, &cg_id)
+ .await
+ .expect("topic.read_topic: delete_consumer_group should work");
+
+ // CG operations should be denied on topic 2
+ let topic2_id = Identifier::named(TOPIC_2).unwrap();
+ assert_unauthorized(
+ client.get_consumer_groups(&stream_id, &topic2_id).await,
+ "topic.read_topic: CG ops on other topic should be denied",
+ );
+
+ delete_test_user(root_client, READ_TOPIC_CG_USER).await;
+}
+
+// =============================================================================
+// Test: Union semantics (global + scoped should be OR, not AND)
+// Tests that stream permissions extend rather than restrict global permissions.
+// =============================================================================
+
+async fn test_union_semantics(client_factory: &dyn ClientFactory, root_client: &IggyClient) {
+ let stream1_id = Identifier::named(STREAM_1).unwrap();
+ let stream2_id = Identifier::named(STREAM_2).unwrap();
+ let topic_id = Identifier::named(TOPIC_1).unwrap();
+
+ let streams = root_client.get_streams().await.expect("get streams");
+ let stream1_numeric_id = streams
+ .iter()
+ .find(|s| s.name == STREAM_1)
+ .map(|s| s.id as usize)
+ .expect("stream 1 should exist");
+
+ let topics = root_client
+ .get_topics(&stream1_id)
+ .await
+ .expect("get topics");
+ let topic1_numeric_id = topics
+ .iter()
+ .find(|t| t.name == TOPIC_1)
+ .map(|t| t.id as usize)
+ .expect("topic 1 should exist");
+
+ // Test: global send_messages + scoped send_messages=false should still allow send
+ // (global permissions should not be restricted by scoped permissions)
+ const UNION_USER: &str = "union-semantics-user";
+ create_test_user(
+ root_client,
+ UNION_USER,
+ Some(Permissions {
+ global: GlobalPermissions {
+ send_messages: true,
+ ..Default::default()
+ },
+ streams: Some(AHashMap::from([(
+ stream1_numeric_id,
+ StreamPermissions {
+ send_messages: false, // Explicitly false at stream level
+ ..Default::default()
+ },
+ )])),
+ }),
+ )
+ .await;
+
+ let client = login_user(client_factory, UNION_USER).await;
+
+ // Should STILL be able to send (global overrides scoped restriction)
+ let mut msgs = test_messages();
+ client
+ .send_messages(
+ &stream1_id,
+ &topic_id,
+ &Partitioning::partition_id(0),
+ &mut msgs,
+ )
+ .await
+ .expect("union: global send_messages should work even with scoped=false");
+
+ // Should also work on stream 2 (not in scoped map)
+ let mut msgs = test_messages();
+ client
+ .send_messages(
+ &stream2_id,
+ &topic_id,
+ &Partitioning::partition_id(0),
+ &mut msgs,
+ )
+ .await
+ .expect("union: global send_messages should work on all streams");
+
+ delete_test_user(root_client, UNION_USER).await;
+
+ // Test: global read_streams + scoped map with one stream should still allow reading other streams
+ const GLOBAL_READ_SCOPED_MAP_USER: &str = "union-global-read-scoped-user";
+ create_test_user(
+ root_client,
+ GLOBAL_READ_SCOPED_MAP_USER,
+ Some(Permissions {
+ global: GlobalPermissions {
+ read_streams: true, // Global read for all streams
+ ..Default::default()
+ },
+ streams: Some(AHashMap::from([(
+ stream1_numeric_id,
+ StreamPermissions {
+ send_messages: true, // Extra permission for stream 1 only
+ ..Default::default()
+ },
+ )])),
+ }),
+ )
+ .await;
+
+ let client = login_user(client_factory, GLOBAL_READ_SCOPED_MAP_USER).await;
+
+ // Should be able to read all streams (global)
+ client
+ .get_streams()
+ .await
+ .expect("union: global read_streams should list all streams");
+ client
+ .get_stream(&stream1_id)
+ .await
+ .expect("union: should read stream 1");
+ client
+ .get_stream(&stream2_id)
+ .await
+ .expect("union: should read stream 2 (global, not in scoped map)");
+
+ // Should be able to send to stream 1 (scoped)
+ let mut msgs = test_messages();
+ client
+ .send_messages(
+ &stream1_id,
+ &topic_id,
+ &Partitioning::partition_id(0),
+ &mut msgs,
+ )
+ .await
+ .expect("union: scoped send_messages on stream 1 should work");
+
+ // Should NOT be able to send to stream 2 (no global send, no scoped send)
+ let mut msgs = test_messages();
+ assert_unauthorized(
+ client
+ .send_messages(
+ &stream2_id,
+ &topic_id,
+ &Partitioning::partition_id(0),
+ &mut msgs,
+ )
+ .await,
+ "union: no send permission for stream 2",
+ );
+
+ delete_test_user(root_client, GLOBAL_READ_SCOPED_MAP_USER).await;
+
+ // Test: topic-level permission should be additive with stream-level
+ const TOPIC_ADDITIVE_USER: &str = "union-topic-additive-user";
+ create_test_user(
+ root_client,
+ TOPIC_ADDITIVE_USER,
+ Some(Permissions {
+ global: GlobalPermissions::default(),
+ streams: Some(AHashMap::from([(
+ stream1_numeric_id,
+ StreamPermissions {
+ read_stream: true, // Can read stream and topics
+ topics: Some(AHashMap::from([(
+ topic1_numeric_id,
+ TopicPermissions {
+ send_messages: true, // Extra: can send to topic 1
+ ..Default::default()
+ },
+ )])),
+ ..Default::default()
+ },
+ )])),
+ }),
+ )
+ .await;
+
+ let client = login_user(client_factory, TOPIC_ADDITIVE_USER).await;
+
+ // Stream read operations should work
+ client
+ .get_stream(&stream1_id)
+ .await
+ .expect("additive: stream.read_stream should work");
+ client
+ .get_topics(&stream1_id)
+ .await
+ .expect("additive: should get all topics");
+
+ // Poll should work on both topics (implied by read_stream)
+ client
+ .poll_messages(
+ &stream1_id,
+ &topic_id,
+ Some(0),
+ &Consumer::default(),
+ &PollingStrategy::offset(0),
+ 1,
+ false,
+ )
+ .await
+ .expect("additive: poll topic 1 should work");
+
+ let topic2_id = Identifier::named(TOPIC_2).unwrap();
+ client
+ .poll_messages(
+ &stream1_id,
+ &topic2_id,
+ Some(0),
+ &Consumer::default(),
+ &PollingStrategy::offset(0),
+ 1,
+ false,
+ )
+ .await
+ .expect("additive: poll topic 2 should work (via stream.read_stream)");
+
+ // Send should ONLY work on topic 1 (explicit topic permission)
+ let mut msgs = test_messages();
+ client
+ .send_messages(
+ &stream1_id,
+ &topic_id,
+ &Partitioning::partition_id(0),
+ &mut msgs,
+ )
+ .await
+ .expect("additive: send to topic 1 should work (explicit topic permission)");
+
+ let mut msgs = test_messages();
+ assert_unauthorized(
+ client
+ .send_messages(
+ &stream1_id,
+ &topic2_id,
+ &Partitioning::partition_id(0),
+ &mut msgs,
+ )
+ .await,
+ "additive: send to topic 2 should be denied (no permission)",
+ );
+
+ delete_test_user(root_client, TOPIC_ADDITIVE_USER).await;
+}
+
+// =============================================================================
+// Test: Missing resource behavior
+// Tests behavior when resources don't exist, for both authorized and
+// unauthorized users.
+// =============================================================================
+
+async fn test_missing_resource_behavior(
+ client_factory: &dyn ClientFactory,
+ root_client: &IggyClient,
+) {
+ let nonexistent_stream = Identifier::named("nonexistent-stream").unwrap();
+ let nonexistent_topic = Identifier::named("nonexistent-topic").unwrap();
+ let existing_stream = Identifier::named(STREAM_1).unwrap();
+
+ // Test: Authorized user accessing non-existent resources
+ // Should get ResourceNotFound, not Unauthorized
+ const AUTHORIZED_USER: &str = "missing-resource-authorized-user";
+ create_test_user(
+ root_client,
+ AUTHORIZED_USER,
+ Some(Permissions {
+ global: GlobalPermissions {
+ read_streams: true,
+ read_topics: true,
+ poll_messages: true,
+ send_messages: true,
+ ..Default::default()
+ },
+ streams: None,
+ }),
+ )
+ .await;
+
+ let client = login_user(client_factory, AUTHORIZED_USER).await;
+
+ // Non-existent stream
+ assert_not_found_or_related(
+ client.get_stream(&nonexistent_stream).await,
+ "authorized: get non-existent stream",
+ );
+
+ // Non-existent topic in existing stream
+ assert_not_found_or_related(
+ client.get_topic(&existing_stream, &nonexistent_topic).await,
+ "authorized: get non-existent topic",
+ );
+
+ // Poll from non-existent stream/topic
+ assert_not_found_or_related(
+ client
+ .poll_messages(
+ &nonexistent_stream,
+ &nonexistent_topic,
+ Some(0),
+ &Consumer::default(),
+ &PollingStrategy::offset(0),
+ 1,
+ false,
+ )
+ .await,
+ "authorized: poll from non-existent stream/topic",
+ );
+
+ // Send to non-existent stream/topic
+ let mut msgs = test_messages();
+ assert_not_found_or_related(
+ client
+ .send_messages(
+ &nonexistent_stream,
+ &nonexistent_topic,
+ &Partitioning::partition_id(0),
+ &mut msgs,
+ )
+ .await,
+ "authorized: send to non-existent stream/topic",
+ );
+
+ delete_test_user(root_client, AUTHORIZED_USER).await;
+
+ // Test: Unauthorized user accessing non-existent resources
+ // Policy decision: should this return Unauthorized (anti-enumeration) or NotFound?
+ // We test for Unauthorized to prevent resource enumeration attacks.
+ const UNAUTHORIZED_USER: &str = "missing-resource-unauthorized-user";
+ create_test_user(root_client, UNAUTHORIZED_USER, Some(no_permissions())).await;
+
+ let client = login_user(client_factory, UNAUTHORIZED_USER).await;
+
+ // Non-existent stream - should be Unauthorized (don't reveal existence)
+ assert_unauthorized(
+ client.get_stream(&nonexistent_stream).await,
+ "unauthorized: get non-existent stream should be Unauthorized",
+ );
+
+ // Non-existent topic - should be Unauthorized
+ assert_unauthorized(
+ client.get_topic(&existing_stream, &nonexistent_topic).await,
+ "unauthorized: get non-existent topic should be Unauthorized",
+ );
+
+ // Poll from non-existent - should be Unauthorized
+ assert_unauthorized(
+ client
+ .poll_messages(
+ &nonexistent_stream,
+ &nonexistent_topic,
+ Some(0),
+ &Consumer::default(),
+ &PollingStrategy::offset(0),
+ 1,
+ false,
+ )
+ .await,
+ "unauthorized: poll from non-existent should be Unauthorized",
+ );
+
+ // Send to non-existent - should be Unauthorized
+ let mut msgs = test_messages();
+ assert_unauthorized(
+ client
+ .send_messages(
+ &nonexistent_stream,
+ &nonexistent_topic,
+ &Partitioning::partition_id(0),
+ &mut msgs,
+ )
+ .await,
+ "unauthorized: send to non-existent should be Unauthorized",
+ );
+
+ delete_test_user(root_client, UNAUTHORIZED_USER).await;
+
+ // Test: Authorized for some streams, accessing non-existent stream
+ // This tests the edge case where user has scoped permissions
+ let streams = root_client.get_streams().await.expect("get streams");
+ let stream1_numeric_id = streams
+ .iter()
+ .find(|s| s.name == STREAM_1)
+ .map(|s| s.id as usize)
+ .expect("stream 1 should exist");
+
+ const SCOPED_USER: &str = "missing-resource-scoped-user";
+ create_test_user(
+ root_client,
+ SCOPED_USER,
+ Some(Permissions {
+ global: GlobalPermissions::default(),
+ streams: Some(AHashMap::from([(
+ stream1_numeric_id,
+ StreamPermissions {
+ read_stream: true,
+ poll_messages: true,
+ send_messages: true,
+ ..Default::default()
+ },
+ )])),
+ }),
+ )
+ .await;
+
+ let client = login_user(client_factory, SCOPED_USER).await;
+
+ // Accessing non-existent stream should be Unauthorized (no permission for it)
+ assert_unauthorized(
+ client.get_stream(&nonexistent_stream).await,
+ "scoped: get non-existent stream should be Unauthorized",
+ );
+
+ // Accessing non-existent topic in authorized stream should be NotFound
+ assert_not_found_or_related(
+ client.get_topic(&existing_stream, &nonexistent_topic).await,
+ "scoped: get non-existent topic in authorized stream",
+ );
+
+ delete_test_user(root_client, SCOPED_USER).await;
+}
+
+// =============================================================================
+// Helpers
+// =============================================================================
+
+fn test_messages() -> Vec<IggyMessage> {
+ vec![
+ IggyMessage::builder()
+ .payload(Bytes::from("test-payload"))
+ .build()
+ .unwrap(),
+ ]
+}
+
+fn assert_unauthorized<T: std::fmt::Debug>(result: Result<T, IggyError>, context: &str) {
+ // Accept Unauthorized OR NotFound errors - we don't require hiding resource existence
+ // Also accept Ok(_) for HTTP which may return Ok(None) on 404
+ let dummy_id = Identifier::numeric(0).unwrap();
+ let acceptable_codes = [
+ IggyError::Unauthorized.as_code(),
+ IggyError::ResourceNotFound(String::new()).as_code(),
+ IggyError::StreamIdNotFound(dummy_id.clone()).as_code(),
+ IggyError::StreamNameNotFound(String::new()).as_code(),
+ IggyError::TopicIdNotFound(dummy_id.clone(), dummy_id.clone()).as_code(),
+ IggyError::TopicNameNotFound(String::new(), String::new()).as_code(),
+ ];
+
+ match result {
+ Err(e) if acceptable_codes.contains(&e.as_code()) => {}
+ Err(e) => panic!(
+ "{}: expected Unauthorized or NotFound, got {:?} ({})",
+ context,
+ e,
+ e.as_code()
+ ),
+ Ok(_) => {
+ // HTTP may return Ok(None) on 404 - this is acceptable
+ }
+ }
+}
+
+fn assert_not_found_or_related<T: std::fmt::Debug>(result: Result<T, IggyError>, context: &str) {
+ // Accept various "not found" error codes that indicate resource absence.
+ // This is used when an authorized user tries to access a non-existent resource.
+ let dummy_id = Identifier::numeric(0).unwrap();
+ let not_found_codes = [
+ IggyError::ResourceNotFound(String::new()).as_code(),
+ IggyError::StreamIdNotFound(dummy_id.clone()).as_code(),
+ IggyError::StreamNameNotFound(String::new()).as_code(),
+ IggyError::TopicIdNotFound(dummy_id.clone(), dummy_id.clone()).as_code(),
+ IggyError::TopicNameNotFound(String::new(), String::new()).as_code(),
+ ];
+
+ match result {
+ Err(e) if not_found_codes.contains(&e.as_code()) => {}
+ Err(e) => panic!(
+ "{}: expected NotFound-related error, got {:?} (code: {})",
+ context,
+ e,
+ e.as_code()
+ ),
+ Ok(_) => {
+ // For HTTP, some endpoints return Ok(None) wrapped in Ok instead of Err(NotFound)
+ // This is acceptable - the resource wasn't found
+ }
+ }
+}
+
+/// Assert that result is Ok or FeatureUnavailable.
+/// Used for stateful operations (join/leave consumer group) that are not supported on HTTP.
+fn assert_ok_or_feature_unavailable(result: Result<(), IggyError>, context: &str) {
+ match result {
+ Ok(()) => {}
+ Err(e) if e.as_code() == IggyError::FeatureUnavailable.as_code() => {}
+ Err(e) => panic!(
+ "{}: expected Ok or FeatureUnavailable, got {:?} (code: {})",
+ context,
+ e,
+ e.as_code()
+ ),
+ }
+}
diff --git a/core/metadata/src/lib.rs b/core/metadata/src/lib.rs
index 9b22a6211..824cf614f 100644
--- a/core/metadata/src/lib.rs
+++ b/core/metadata/src/lib.rs
@@ -18,7 +18,7 @@
//! Iggy metadata module
mod impls;
-mod permissioner;
+pub mod permissioner;
pub mod stm;
mod stats;
diff --git a/core/metadata/src/permissioner/mod.rs b/core/metadata/src/permissioner/mod.rs
index 36809a9d9..52b13443f 100644
--- a/core/metadata/src/permissioner/mod.rs
+++ b/core/metadata/src/permissioner/mod.rs
@@ -35,7 +35,7 @@ impl Permissioner {
Self::default()
}
- pub fn init_permissions(&mut self, user_id: UserId, permissions: Option<Permissions>) {
+ pub fn init_permissions_for_user(&mut self, user_id: UserId, permissions: Option<Permissions>) {
if permissions.is_none() {
return;
}
@@ -78,11 +78,11 @@ impl Permissioner {
user_id: UserId,
permissions: Option<Permissions>,
) {
- self.delete_permissions(user_id);
- self.init_permissions(user_id, permissions);
+ self.delete_permissions_for_user(user_id);
+ self.init_permissions_for_user(user_id, permissions);
}
- pub fn delete_permissions(&mut self, user_id: UserId) {
+ pub fn delete_permissions_for_user(&mut self, user_id: UserId) {
self.users_permissions.remove(&user_id);
self.users_that_can_poll_messages_from_all_streams
.remove(&user_id);
diff --git a/core/metadata/src/permissioner/permissioner_rules/consumer_groups.rs b/core/metadata/src/permissioner/permissioner_rules/consumer_groups.rs
index e73794d7c..66022ae2d 100644
--- a/core/metadata/src/permissioner/permissioner_rules/consumer_groups.rs
+++ b/core/metadata/src/permissioner/permissioner_rules/consumer_groups.rs
@@ -16,10 +16,6 @@
* under the License.
*/
-#![expect(
- unused,
- reason = "Methods are part of the state API and will be used once the implementation is complete"
-)]
use crate::permissioner::Permissioner;
use iggy_common::IggyError;
diff --git a/core/metadata/src/permissioner/permissioner_rules/consumer_offsets.rs b/core/metadata/src/permissioner/permissioner_rules/consumer_offsets.rs
index bd8bcaa53..f94e3fdc0 100644
--- a/core/metadata/src/permissioner/permissioner_rules/consumer_offsets.rs
+++ b/core/metadata/src/permissioner/permissioner_rules/consumer_offsets.rs
@@ -16,10 +16,6 @@
* under the License.
*/
-#![expect(
- unused,
- reason = "Methods are part of the state API and will be used once the implementation is complete"
-)]
use crate::permissioner::Permissioner;
use iggy_common::IggyError;
diff --git a/core/metadata/src/permissioner/permissioner_rules/messages.rs b/core/metadata/src/permissioner/permissioner_rules/messages.rs
index 520de0385..30e6c3395 100644
--- a/core/metadata/src/permissioner/permissioner_rules/messages.rs
+++ b/core/metadata/src/permissioner/permissioner_rules/messages.rs
@@ -16,14 +16,11 @@
* under the License.
*/
-#![expect(
- unused,
- reason = "Methods are part of the state API and will be used once the implementation is complete"
-)]
use crate::permissioner::Permissioner;
use iggy_common::IggyError;
impl Permissioner {
+ /// Inheritance: manage_streams → read_streams → read_topics → poll_messages
pub fn poll_messages(
&self,
user_id: u32,
@@ -37,6 +34,15 @@ impl Permissioner {
return Ok(());
}
+ if let Some(global) = self.users_permissions.get(&user_id)
+ && (global.read_topics
+ || global.manage_topics
+ || global.read_streams
+ || global.manage_streams)
+ {
+ return Ok(());
+ }
+
if self
.users_that_can_poll_messages_from_specific_streams
.contains(&(user_id, stream_id))
@@ -44,21 +50,16 @@ impl Permissioner {
return Ok(());
}
- let stream_permissions = self.users_streams_permissions.get(&(user_id, stream_id));
- if stream_permissions.is_none() {
+ let Some(stream_permissions) = self.users_streams_permissions.get(&(user_id, stream_id))
+ else {
return Err(IggyError::Unauthorized);
- }
+ };
- let stream_permissions = stream_permissions.unwrap();
- if stream_permissions.read_stream {
+ if stream_permissions.manage_stream || stream_permissions.read_stream {
return Ok(());
}
- if stream_permissions.manage_topics {
- return Ok(());
- }
-
- if stream_permissions.read_topics {
+ if stream_permissions.manage_topics || stream_permissions.read_topics {
return Ok(());
}
@@ -66,24 +67,19 @@ impl Permissioner {
return Ok(());
}
- if stream_permissions.topics.is_none() {
- return Err(IggyError::Unauthorized);
- }
-
- let topic_permissions = stream_permissions.topics.as_ref().unwrap();
- if let Some(topic_permissions) = topic_permissions.get(&topic_id) {
- return match topic_permissions.poll_messages
- | topic_permissions.read_topic
- | topic_permissions.manage_topic
- {
- true => Ok(()),
- false => Err(IggyError::Unauthorized),
- };
+ if let Some(topics) = &stream_permissions.topics
+ && let Some(topic_permissions) = topics.get(&topic_id)
+ && (topic_permissions.manage_topic
+ || topic_permissions.read_topic
+ || topic_permissions.poll_messages)
+ {
+ return Ok(());
}
Err(IggyError::Unauthorized)
}
+ /// send_messages is NOT inherited from manage permissions.
pub fn append_messages(
&self,
user_id: u32,
@@ -104,34 +100,20 @@ impl Permissioner {
return Ok(());
}
- let stream_permissions = self.users_streams_permissions.get(&(user_id, stream_id));
- if stream_permissions.is_none() {
+ let Some(stream_permissions) = self.users_streams_permissions.get(&(user_id, stream_id))
+ else {
return Err(IggyError::Unauthorized);
- }
-
- let stream_permissions = stream_permissions.unwrap();
- if stream_permissions.manage_stream {
- return Ok(());
- }
-
- if stream_permissions.manage_topics {
- return Ok(());
- }
+ };
if stream_permissions.send_messages {
return Ok(());
}
- if stream_permissions.topics.is_none() {
- return Err(IggyError::Unauthorized);
- }
-
- let topic_permissions = stream_permissions.topics.as_ref().unwrap();
- if let Some(topic_permissions) = topic_permissions.get(&topic_id) {
- return match topic_permissions.send_messages | topic_permissions.manage_topic {
- true => Ok(()),
- false => Err(IggyError::Unauthorized),
- };
+ if let Some(topics) = &stream_permissions.topics
+ && let Some(topic_permissions) = topics.get(&topic_id)
+ && topic_permissions.send_messages
+ {
+ return Ok(());
}
Err(IggyError::Unauthorized)
diff --git a/core/metadata/src/permissioner/permissioner_rules/partitions.rs b/core/metadata/src/permissioner/permissioner_rules/partitions.rs
index ad1829387..7f11007a1 100644
--- a/core/metadata/src/permissioner/permissioner_rules/partitions.rs
+++ b/core/metadata/src/permissioner/permissioner_rules/partitions.rs
@@ -16,10 +16,6 @@
* under the License.
*/
-#![expect(
- unused,
- reason = "Methods are part of the state API and will be used once the implementation is complete"
-)]
use crate::permissioner::Permissioner;
use iggy_common::IggyError;
diff --git a/core/metadata/src/permissioner/permissioner_rules/segments.rs b/core/metadata/src/permissioner/permissioner_rules/segments.rs
index 484a2c158..558e2b0b3 100644
--- a/core/metadata/src/permissioner/permissioner_rules/segments.rs
+++ b/core/metadata/src/permissioner/permissioner_rules/segments.rs
@@ -16,10 +16,6 @@
* under the License.
*/
-#![expect(
- unused,
- reason = "Methods are part of the state API and will be used once the implementation is complete"
-)]
use crate::permissioner::Permissioner;
use iggy_common::IggyError;
diff --git a/core/metadata/src/permissioner/permissioner_rules/streams.rs b/core/metadata/src/permissioner/permissioner_rules/streams.rs
index 70829c2fb..153d3d4d5 100644
--- a/core/metadata/src/permissioner/permissioner_rules/streams.rs
+++ b/core/metadata/src/permissioner/permissioner_rules/streams.rs
@@ -16,10 +16,6 @@
* under the License.
*/
-#![expect(
- unused,
- reason = "Methods are part of the state API and will be used once the implementation is complete"
-)]
use crate::permissioner::Permissioner;
use iggy_common::IggyError;
diff --git a/core/metadata/src/permissioner/permissioner_rules/system.rs b/core/metadata/src/permissioner/permissioner_rules/system.rs
index ce03c2dae..e2ef0d8c7 100644
--- a/core/metadata/src/permissioner/permissioner_rules/system.rs
+++ b/core/metadata/src/permissioner/permissioner_rules/system.rs
@@ -16,10 +16,6 @@
* under the License.
*/
-#![expect(
- unused,
- reason = "Methods are part of the state API and will be used once the implementation is complete"
-)]
use crate::permissioner::Permissioner;
use iggy_common::IggyError;
diff --git a/core/metadata/src/permissioner/permissioner_rules/topics.rs b/core/metadata/src/permissioner/permissioner_rules/topics.rs
index ca226a8d0..6e93655c3 100644
--- a/core/metadata/src/permissioner/permissioner_rules/topics.rs
+++ b/core/metadata/src/permissioner/permissioner_rules/topics.rs
@@ -16,37 +16,38 @@
* under the License.
*/
-#![expect(
- unused,
- reason = "Methods are part of the state API and will be used once the implementation is complete"
-)]
use crate::permissioner::Permissioner;
use iggy_common::IggyError;
impl Permissioner {
+ /// Inheritance: manage_streams → read_streams → read_topics
pub fn get_topic(
&self,
user_id: u32,
stream_id: usize,
topic_id: usize,
) -> Result<(), IggyError> {
- if let Some(global_permissions) = self.users_permissions.get(&user_id)
- && (global_permissions.read_streams
- || global_permissions.manage_streams
- || global_permissions.manage_topics
- || global_permissions.read_topics)
+ if let Some(global) = self.users_permissions.get(&user_id)
+ && (global.read_streams
+ || global.manage_streams
+ || global.manage_topics
+ || global.read_topics)
{
return Ok(());
}
if let Some(stream_permissions) = self.users_streams_permissions.get(&(user_id, stream_id))
{
- if stream_permissions.manage_topics || stream_permissions.read_topics {
+ if stream_permissions.manage_stream
+ || stream_permissions.read_stream
+ || stream_permissions.manage_topics
+ || stream_permissions.read_topics
+ {
return Ok(());
}
- if let Some(topic_permissions) =
- stream_permissions.topics.as_ref().unwrap().get(&topic_id)
+ if let Some(topics) = &stream_permissions.topics
+ && let Some(topic_permissions) = topics.get(&topic_id)
&& (topic_permissions.manage_topic || topic_permissions.read_topic)
{
return Ok(());
@@ -57,41 +58,37 @@ impl Permissioner {
}
pub fn get_topics(&self, user_id: u32, stream_id: usize) -> Result<(), IggyError> {
- if let Some(global_permissions) = self.users_permissions.get(&user_id)
- && (global_permissions.read_streams
- || global_permissions.manage_streams
- || global_permissions.manage_topics
- || global_permissions.read_topics)
+ if let Some(global) = self.users_permissions.get(&user_id)
+ && (global.read_streams
+ || global.manage_streams
+ || global.manage_topics
+ || global.read_topics)
{
return Ok(());
}
if let Some(stream_permissions) = self.users_streams_permissions.get(&(user_id, stream_id))
+ && (stream_permissions.manage_stream
+ || stream_permissions.read_stream
+ || stream_permissions.manage_topics
+ || stream_permissions.read_topics)
{
- if stream_permissions.manage_topics || stream_permissions.read_topics {
- return Ok(());
- }
-
- if let Some(topic_permissions) = stream_permissions.topics.as_ref() {
- for (topic_id, topic_permissions) in topic_permissions {
- if !topic_permissions.manage_topic || !topic_permissions.read_topic {
- return Err(IggyError::Unauthorized);
- }
- }
- }
+ return Ok(());
}
- Ok(())
+
+ Err(IggyError::Unauthorized)
}
+ /// Inheritance: manage_streams → manage_topics
pub fn create_topic(&self, user_id: u32, stream_id: usize) -> Result<(), IggyError> {
- if let Some(global_permissions) = self.users_permissions.get(&user_id)
- && (global_permissions.manage_streams || global_permissions.manage_topics)
+ if let Some(global) = self.users_permissions.get(&user_id)
+ && (global.manage_streams || global.manage_topics)
{
return Ok(());
}
if let Some(stream_permissions) = self.users_streams_permissions.get(&(user_id, stream_id))
- && stream_permissions.manage_topics
+ && (stream_permissions.manage_stream || stream_permissions.manage_topics)
{
return Ok(());
}
@@ -126,26 +123,27 @@ impl Permissioner {
self.manage_topic(user_id, stream_id, topic_id)
}
+ /// Inheritance: manage_streams → manage_topics
fn manage_topic(
&self,
user_id: u32,
stream_id: usize,
topic_id: usize,
) -> Result<(), IggyError> {
- if let Some(global_permissions) = self.users_permissions.get(&user_id)
- && (global_permissions.manage_streams || global_permissions.manage_topics)
+ if let Some(global) = self.users_permissions.get(&user_id)
+ && (global.manage_streams || global.manage_topics)
{
return Ok(());
}
if let Some(stream_permissions) = self.users_streams_permissions.get(&(user_id, stream_id))
{
- if stream_permissions.manage_topics {
+ if stream_permissions.manage_stream || stream_permissions.manage_topics {
return Ok(());
}
- if let Some(topic_permissions) =
- stream_permissions.topics.as_ref().unwrap().get(&topic_id)
+ if let Some(topics) = &stream_permissions.topics
+ && let Some(topic_permissions) = topics.get(&topic_id)
&& topic_permissions.manage_topic
{
return Ok(());
diff --git a/core/metadata/src/permissioner/permissioner_rules/users.rs b/core/metadata/src/permissioner/permissioner_rules/users.rs
index c084effcb..bba4b84a9 100644
--- a/core/metadata/src/permissioner/permissioner_rules/users.rs
+++ b/core/metadata/src/permissioner/permissioner_rules/users.rs
@@ -16,10 +16,6 @@
* under the License.
*/
-#![expect(
- unused,
- reason = "Methods are part of the state API and will be used once the implementation is complete"
-)]
use crate::permissioner::Permissioner;
use iggy_common::IggyError;
diff --git a/core/metadata/src/stm/user.rs b/core/metadata/src/stm/user.rs
index b31bdc764..635277965 100644
--- a/core/metadata/src/stm/user.rs
+++ b/core/metadata/src/stm/user.rs
@@ -185,7 +185,7 @@ impl Users {
pub fn init_permissions(&self, user_id: UserId, permissions: Option<Permissions>) {
self.permissioner
.borrow_mut()
- .init_permissions(user_id, permissions);
+ .init_permissions_for_user(user_id, permissions);
}
/// Update permissions for a user
@@ -197,7 +197,9 @@ impl Users {
/// Delete permissions for a user
pub fn delete_permissions(&self, user_id: UserId) {
- self.permissioner.borrow_mut().delete_permissions(user_id);
+ self.permissioner
+ .borrow_mut()
+ .delete_permissions_for_user(user_id);
}
/// Update username
diff --git a/core/server/Cargo.toml b/core/server/Cargo.toml
index fd72144be..f5c0deb5f 100644
--- a/core/server/Cargo.toml
+++ b/core/server/Cargo.toml
@@ -73,6 +73,7 @@ iggy_common = { workspace = true }
jsonwebtoken = { version = "10.2.0", features = ["rust_crypto"] }
left-right = "0.11"
lending-iterator = "0.1.7"
+metadata = { workspace = true }
mimalloc = { workspace = true, optional = true }
mime_guess = { version = "2.0", optional = true }
moka = { version = "0.12.12", features = ["future"] }
diff --git a/core/server/src/binary/handlers/cluster/get_cluster_metadata_handler.rs b/core/server/src/binary/handlers/cluster/get_cluster_metadata_handler.rs
index 501c697ba..395ed81f2 100644
--- a/core/server/src/binary/handlers/cluster/get_cluster_metadata_handler.rs
+++ b/core/server/src/binary/handlers/cluster/get_cluster_metadata_handler.rs
@@ -16,18 +16,16 @@
* under the License.
*/
-use std::rc::Rc;
-
use crate::binary::command::{
BinaryServerCommand, HandlerResult, ServerCommand, ServerCommandHandler,
};
use crate::binary::handlers::utils::receive_and_validate;
use crate::shard::IggyShard;
use crate::streaming::session::Session;
-//use crate::streaming::systems::system::SharedSystem;
use anyhow::Result;
use iggy_common::get_cluster_metadata::GetClusterMetadata;
use iggy_common::{BytesSerializable, IggyError, SenderKind};
+use std::rc::Rc;
use tracing::{debug, instrument};
impl ServerCommandHandler for GetClusterMetadata {
@@ -46,7 +44,7 @@ impl ServerCommandHandler for GetClusterMetadata {
debug!("session: {session}, command: {self}");
shard.ensure_authenticated(session)?;
- let cluster_metadata = shard.get_cluster_metadata(session)?;
+ let cluster_metadata = shard.get_cluster_metadata();
let response = cluster_metadata.to_bytes();
sender.send_ok_response(&response).await?;
diff --git a/core/server/src/binary/handlers/consumer_groups/create_consumer_group_handler.rs b/core/server/src/binary/handlers/consumer_groups/create_consumer_group_handler.rs
index fe1350a9f..a56c80551 100644
--- a/core/server/src/binary/handlers/consumer_groups/create_consumer_group_handler.rs
+++ b/core/server/src/binary/handlers/consumer_groups/create_consumer_group_handler.rs
@@ -50,12 +50,13 @@ impl ServerCommandHandler for CreateConsumerGroup {
) -> Result<HandlerResult, IggyError> {
debug!("session: {session}, command: {self}");
shard.ensure_authenticated(session)?;
- let cg = shard.create_consumer_group(
- session,
- &self.stream_id,
- &self.topic_id,
- self.name.clone(),
+ let (stream_id, topic_id) = shard.resolve_topic_id(&self.stream_id, &self.topic_id)?;
+ shard.permissioner.borrow().create_consumer_group(
+ session.get_user_id(),
+ stream_id,
+ topic_id,
)?;
+ let cg = shard.create_consumer_group(&self.stream_id, &self.topic_id, self.name.clone())?;
let cg_id = cg.id();
let event = ShardEvent::CreatedConsumerGroup {
diff --git a/core/server/src/binary/handlers/consumer_groups/delete_consumer_group_handler.rs b/core/server/src/binary/handlers/consumer_groups/delete_consumer_group_handler.rs
index 07751b591..6e372719d 100644
--- a/core/server/src/binary/handlers/consumer_groups/delete_consumer_group_handler.rs
+++ b/core/server/src/binary/handlers/consumer_groups/delete_consumer_group_handler.rs
@@ -50,7 +50,16 @@ impl ServerCommandHandler for DeleteConsumerGroup {
) -> Result<HandlerResult, IggyError> {
debug!("session: {session}, command: {self}");
shard.ensure_authenticated(session)?;
- let cg = shard.delete_consumer_group(session, &self.stream_id, &self.topic_id, &self.group_id).error(|e: &IggyError| {
+ let (stream_id_numeric, topic_id_numeric) =
+ shard.resolve_topic_id(&self.stream_id, &self.topic_id)?;
+ shard.permissioner.borrow().delete_consumer_group(
+ session.get_user_id(),
+ stream_id_numeric,
+ topic_id_numeric,
+ )?;
+ shard.ensure_consumer_group_exists(&self.stream_id, &self.topic_id, &self.group_id)?;
+
+ let cg = shard.delete_consumer_group(&self.stream_id, &self.topic_id, &self.group_id).error(|e: &IggyError| {
format!(
"{COMPONENT} (error: {e}) - failed to delete consumer group with ID: {} for topic with ID: {} in stream with ID: {} for session: {}",
self.group_id, self.topic_id, self.stream_id, session
@@ -59,24 +68,13 @@ impl ServerCommandHandler for DeleteConsumerGroup {
let cg_id = cg.id();
let partition_ids = cg.partitions();
- // Remove all consumer group members from ClientManager using helper functions to resolve identifiers
- let stream_id_usize = shard.streams.with_stream_by_id(
- &self.stream_id,
- crate::streaming::streams::helpers::get_stream_id(),
- );
- let topic_id_usize = shard.streams.with_topic_by_id(
- &self.stream_id,
- &self.topic_id,
- crate::streaming::topics::helpers::get_topic_id(),
- );
-
// Get members from the deleted consumer group and make them leave
let slab = cg.members().inner().shared_get();
for (_, member) in slab.iter() {
if let Err(err) = shard.client_manager.leave_consumer_group(
member.client_id,
- stream_id_usize,
- topic_id_usize,
+ stream_id_numeric,
+ topic_id_numeric,
cg_id,
) {
tracing::warn!(
diff --git a/core/server/src/binary/handlers/consumer_groups/get_consumer_group_handler.rs b/core/server/src/binary/handlers/consumer_groups/get_consumer_group_handler.rs
index 7156a4d53..27c39e420 100644
--- a/core/server/src/binary/handlers/consumer_groups/get_consumer_group_handler.rs
+++ b/core/server/src/binary/handlers/consumer_groups/get_consumer_group_handler.rs
@@ -23,7 +23,6 @@ use crate::binary::handlers::utils::receive_and_validate;
use crate::binary::mapper;
use crate::shard::IggyShard;
use crate::streaming::session::Session;
-use crate::streaming::{streams, topics};
use anyhow::Result;
use iggy_common::IggyError;
use iggy_common::SenderKind;
@@ -45,30 +44,23 @@ impl ServerCommandHandler for GetConsumerGroup {
) -> Result<HandlerResult, IggyError> {
debug!("session: {session}, command: {self}");
shard.ensure_authenticated(session)?;
- let exists = shard
- .ensure_consumer_group_exists(&self.stream_id, &self.topic_id, &self.group_id)
- .is_ok();
- if !exists {
+ let Ok((stream_id, topic_id)) = shard.resolve_topic_id(&self.stream_id, &self.topic_id)
+ else {
sender.send_empty_ok_response().await?;
return Ok(HandlerResult::Finished);
- }
- let numeric_topic_id = shard.streams.with_topic_by_id(
- &self.stream_id,
- &self.topic_id,
- topics::helpers::get_topic_id(),
- );
- let numeric_stream_id = shard
- .streams
- .with_stream_by_id(&self.stream_id, streams::helpers::get_stream_id());
- let has_permission = shard
- .permissioner
- .borrow()
- .get_consumer_group(session.get_user_id(), numeric_stream_id, numeric_topic_id)
- .is_ok();
- if !has_permission {
+ };
+ if shard
+ .ensure_consumer_group_exists(&self.stream_id, &self.topic_id, &self.group_id)
+ .is_err()
+ {
sender.send_empty_ok_response().await?;
return Ok(HandlerResult::Finished);
}
+ shard.permissioner.borrow().get_consumer_group(
+ session.get_user_id(),
+ stream_id,
+ topic_id,
+ )?;
let consumer_group = shard.streams.with_consumer_group_by_id(
&self.stream_id,
diff --git a/core/server/src/binary/handlers/consumer_groups/get_consumer_groups_handler.rs b/core/server/src/binary/handlers/consumer_groups/get_consumer_groups_handler.rs
index d59f33c97..8b3cab821 100644
--- a/core/server/src/binary/handlers/consumer_groups/get_consumer_groups_handler.rs
+++ b/core/server/src/binary/handlers/consumer_groups/get_consumer_groups_handler.rs
@@ -24,7 +24,6 @@ use crate::binary::mapper;
use crate::shard::IggyShard;
use crate::slab::traits_ext::{EntityComponentSystem, IntoComponents};
use crate::streaming::session::Session;
-use crate::streaming::{streams, topics};
use anyhow::Result;
use iggy_common::IggyError;
use iggy_common::SenderKind;
@@ -46,19 +45,11 @@ impl ServerCommandHandler for GetConsumerGroups {
) -> Result<HandlerResult, IggyError> {
debug!("session: {session}, command: {self}");
shard.ensure_authenticated(session)?;
- shard.ensure_topic_exists(&self.stream_id, &self.topic_id)?;
- let numeric_topic_id = shard.streams.with_topic_by_id(
- &self.stream_id,
- &self.topic_id,
- topics::helpers::get_topic_id(),
- );
- let numeric_stream_id = shard
- .streams
- .with_stream_by_id(&self.stream_id, streams::helpers::get_stream_id());
+ let (stream_id, topic_id) = shard.resolve_topic_id(&self.stream_id, &self.topic_id)?;
shard.permissioner.borrow().get_consumer_groups(
session.get_user_id(),
- numeric_stream_id,
- numeric_topic_id,
+ stream_id,
+ topic_id,
)?;
let consumer_groups =
diff --git a/core/server/src/binary/handlers/consumer_groups/join_consumer_group_handler.rs b/core/server/src/binary/handlers/consumer_groups/join_consumer_group_handler.rs
index 8aab2dca5..17722303c 100644
--- a/core/server/src/binary/handlers/consumer_groups/join_consumer_group_handler.rs
+++ b/core/server/src/binary/handlers/consumer_groups/join_consumer_group_handler.rs
@@ -21,7 +21,6 @@ use crate::binary::command::{
};
use crate::binary::handlers::consumer_groups::COMPONENT;
use crate::binary::handlers::utils::receive_and_validate;
-
use crate::shard::IggyShard;
use crate::streaming::session::Session;
use anyhow::Result;
@@ -46,9 +45,16 @@ impl ServerCommandHandler for JoinConsumerGroup {
) -> Result<HandlerResult, IggyError> {
debug!("session: {session}, command: {self}");
shard.ensure_authenticated(session)?;
+ let (stream_id, topic_id) = shard.resolve_topic_id(&self.stream_id, &self.topic_id)?;
+ shard.permissioner.borrow().join_consumer_group(
+ session.get_user_id(),
+ stream_id,
+ topic_id,
+ )?;
+ shard.ensure_consumer_group_exists(&self.stream_id, &self.topic_id, &self.group_id)?;
shard
.join_consumer_group(
- session,
+ session.client_id,
&self.stream_id,
&self.topic_id,
&self.group_id,
diff --git a/core/server/src/binary/handlers/consumer_groups/leave_consumer_group_handler.rs b/core/server/src/binary/handlers/consumer_groups/leave_consumer_group_handler.rs
index 32f1c4d5a..4a94fd99e 100644
--- a/core/server/src/binary/handlers/consumer_groups/leave_consumer_group_handler.rs
+++ b/core/server/src/binary/handlers/consumer_groups/leave_consumer_group_handler.rs
@@ -21,13 +21,12 @@ use crate::binary::command::{
BinaryServerCommand, HandlerResult, ServerCommand, ServerCommandHandler,
};
use crate::binary::handlers::utils::receive_and_validate;
-use iggy_common::SenderKind;
-
use crate::shard::IggyShard;
use crate::streaming::session::Session;
use anyhow::Result;
use err_trail::ErrContext;
use iggy_common::IggyError;
+use iggy_common::SenderKind;
use iggy_common::leave_consumer_group::LeaveConsumerGroup;
use std::rc::Rc;
use tracing::{debug, instrument};
@@ -47,10 +46,16 @@ impl ServerCommandHandler for LeaveConsumerGroup {
) -> Result<HandlerResult, IggyError> {
debug!("session: {session}, command: {self}");
shard.ensure_authenticated(session)?;
-
+ let (stream_id, topic_id) = shard.resolve_topic_id(&self.stream_id, &self.topic_id)?;
+ shard.permissioner.borrow().leave_consumer_group(
+ session.get_user_id(),
+ stream_id,
+ topic_id,
+ )?;
+ shard.ensure_consumer_group_exists(&self.stream_id, &self.topic_id, &self.group_id)?;
shard
.leave_consumer_group(
- session,
+ session.client_id,
&self.stream_id,
&self.topic_id,
&self.group_id,
diff --git a/core/server/src/binary/handlers/consumer_offsets/delete_consumer_offset_handler.rs b/core/server/src/binary/handlers/consumer_offsets/delete_consumer_offset_handler.rs
index 64d01255e..e3436d2e2 100644
--- a/core/server/src/binary/handlers/consumer_offsets/delete_consumer_offset_handler.rs
+++ b/core/server/src/binary/handlers/consumer_offsets/delete_consumer_offset_handler.rs
@@ -44,9 +44,15 @@ impl ServerCommandHandler for DeleteConsumerOffset {
) -> Result<HandlerResult, IggyError> {
debug!("session: {session}, command: {self}");
shard.ensure_authenticated(session)?;
+ let (stream_id, topic_id) = shard.resolve_topic_id(&self.stream_id, &self.topic_id)?;
+ shard.permissioner.borrow().delete_consumer_offset(
+ session.get_user_id(),
+ stream_id,
+ topic_id,
+ )?;
shard
.delete_consumer_offset(
- session,
+ session.client_id,
self.consumer,
&self.stream_id,
&self.topic_id,
diff --git a/core/server/src/binary/handlers/consumer_offsets/get_consumer_offset_handler.rs b/core/server/src/binary/handlers/consumer_offsets/get_consumer_offset_handler.rs
index 7b8991981..983d449d9 100644
--- a/core/server/src/binary/handlers/consumer_offsets/get_consumer_offset_handler.rs
+++ b/core/server/src/binary/handlers/consumer_offsets/get_consumer_offset_handler.rs
@@ -44,9 +44,15 @@ impl ServerCommandHandler for GetConsumerOffset {
) -> Result<HandlerResult, IggyError> {
debug!("session: {session}, command: {self}");
shard.ensure_authenticated(session)?;
+ let (stream_id, topic_id) = shard.resolve_topic_id(&self.stream_id, &self.topic_id)?;
+ shard.permissioner.borrow().get_consumer_offset(
+ session.get_user_id(),
+ stream_id,
+ topic_id,
+ )?;
let Ok(offset) = shard
.get_consumer_offset(
- session,
+ session.client_id,
self.consumer,
&self.stream_id,
&self.topic_id,
diff --git a/core/server/src/binary/handlers/consumer_offsets/store_consumer_offset_handler.rs b/core/server/src/binary/handlers/consumer_offsets/store_consumer_offset_handler.rs
index c603f4cb8..16b57b217 100644
--- a/core/server/src/binary/handlers/consumer_offsets/store_consumer_offset_handler.rs
+++ b/core/server/src/binary/handlers/consumer_offsets/store_consumer_offset_handler.rs
@@ -46,9 +46,15 @@ impl ServerCommandHandler for StoreConsumerOffset {
) -> Result<HandlerResult, IggyError> {
debug!("session: {session}, command: {self}");
shard.ensure_authenticated(session)?;
+ let (stream_id, topic_id) = shard.resolve_topic_id(&self.stream_id, &self.topic_id)?;
+ shard.permissioner.borrow().store_consumer_offset(
+ session.get_user_id(),
+ stream_id,
+ topic_id,
+ )?;
shard
.store_consumer_offset(
- session,
+ session.client_id,
self.consumer,
&self.stream_id,
&self.topic_id,
diff --git a/core/server/src/binary/handlers/messages/send_messages_handler.rs b/core/server/src/binary/handlers/messages/send_messages_handler.rs
index 6aa6786ac..2e2748776 100644
--- a/core/server/src/binary/handlers/messages/send_messages_handler.rs
+++ b/core/server/src/binary/handlers/messages/send_messages_handler.rs
@@ -21,7 +21,7 @@ use crate::shard::IggyShard;
use crate::shard::transmission::message::{ShardMessage, ShardRequest, ShardRequestPayload};
use crate::streaming::segments::{IggyIndexesMut, IggyMessagesBatchMut};
use crate::streaming::session::Session;
-use crate::streaming::{streams, topics};
+use crate::streaming::topics;
use anyhow::Result;
use compio::buf::{IntoInner as _, IoBuf};
use iggy_common::Identifier;
@@ -111,17 +111,13 @@ impl ServerCommandHandler for SendMessages {
);
batch.validate()?;
- shard.ensure_topic_exists(&self.stream_id, &self.topic_id)?;
-
- let numeric_stream_id = shard
- .streams
- .with_stream_by_id(&self.stream_id, streams::helpers::get_stream_id());
-
- let numeric_topic_id = shard.streams.with_topic_by_id(
- &self.stream_id,
- &self.topic_id,
- topics::helpers::get_topic_id(),
- );
+ let (numeric_stream_id, numeric_topic_id) =
+ shard.resolve_topic_id(&self.stream_id, &self.topic_id)?;
+ shard.permissioner.borrow().append_messages(
+ session.get_user_id(),
+ numeric_stream_id,
+ numeric_topic_id,
+ )?;
// TODO(tungtose): dry this code && get partition_id below have a side effect
let partition_id = shard.streams.with_topic_by_id(
diff --git a/core/server/src/binary/handlers/partitions/create_partitions_handler.rs b/core/server/src/binary/handlers/partitions/create_partitions_handler.rs
index 85b4e3030..df9cfee22 100644
--- a/core/server/src/binary/handlers/partitions/create_partitions_handler.rs
+++ b/core/server/src/binary/handlers/partitions/create_partitions_handler.rs
@@ -50,17 +50,19 @@ impl ServerCommandHandler for CreatePartitions {
) -> Result<HandlerResult, IggyError> {
debug!("session: {session}, command: {self}");
shard.ensure_authenticated(session)?;
+ let (stream_id_numeric, topic_id_numeric) =
+ shard.resolve_topic_id(&self.stream_id, &self.topic_id)?;
+ shard.permissioner.borrow().create_partitions(
+ session.get_user_id(),
+ stream_id_numeric,
+ topic_id_numeric,
+ )?;
// Acquire partition lock to serialize filesystem operations
let _partition_guard = shard.fs_locks.partition_lock.lock().await;
let partitions = shard
- .create_partitions(
- session,
- &self.stream_id,
- &self.topic_id,
- self.partitions_count,
- )
+ .create_partitions(&self.stream_id, &self.topic_id, self.partitions_count)
.await?;
let partition_ids = partitions.iter().map(|p| p.id()).collect::<Vec<_>>();
let event = ShardEvent::CreatedPartitions {
diff --git a/core/server/src/binary/handlers/partitions/delete_partitions_handler.rs b/core/server/src/binary/handlers/partitions/delete_partitions_handler.rs
index f4bed945f..4ce2566ad 100644
--- a/core/server/src/binary/handlers/partitions/delete_partitions_handler.rs
+++ b/core/server/src/binary/handlers/partitions/delete_partitions_handler.rs
@@ -48,6 +48,13 @@ impl ServerCommandHandler for DeletePartitions {
) -> Result<HandlerResult, IggyError> {
debug!("session: {session}, command: {self}");
shard.ensure_authenticated(session)?;
+ let (stream_id_numeric, topic_id_numeric) =
+ shard.resolve_topic_id(&self.stream_id, &self.topic_id)?;
+ shard.permissioner.borrow().delete_partitions(
+ session.get_user_id(),
+ stream_id_numeric,
+ topic_id_numeric,
+ )?;
let stream_id = self.stream_id.clone();
let topic_id = self.topic_id.clone();
@@ -55,12 +62,7 @@ impl ServerCommandHandler for DeletePartitions {
let _partition_guard = shard.fs_locks.partition_lock.lock().await;
let deleted_partition_ids = shard
- .delete_partitions(
- session,
- &self.stream_id,
- &self.topic_id,
- self.partitions_count,
- )
+ .delete_partitions(&self.stream_id, &self.topic_id, self.partitions_count)
.await?;
let event = ShardEvent::DeletedPartitions {
stream_id: self.stream_id.clone(),
diff --git a/core/server/src/binary/handlers/personal_access_tokens/create_personal_access_token_handler.rs b/core/server/src/binary/handlers/personal_access_tokens/create_personal_access_token_handler.rs
index 314f1821d..8f2e766dd 100644
--- a/core/server/src/binary/handlers/personal_access_tokens/create_personal_access_token_handler.rs
+++ b/core/server/src/binary/handlers/personal_access_tokens/create_personal_access_token_handler.rs
@@ -51,11 +51,12 @@ impl ServerCommandHandler for CreatePersonalAccessToken {
debug!("session: {session}, command: {self}");
shard.ensure_authenticated(session)?;
+ let user_id = session.get_user_id();
let (personal_access_token, token) = shard
- .create_personal_access_token(session, &self.name, self.expiry)
+ .create_personal_access_token(user_id, &self.name, self.expiry)
.error(|e: &IggyError| {
format!(
- "{COMPONENT} (error: {e}) - failed to create personal access token with name: {}, session: {session}",
+ "{COMPONENT} (error: {e}) - failed to create personal access token with name: {}, user: {user_id}",
self.name
)
})?;
diff --git a/core/server/src/binary/handlers/personal_access_tokens/delete_personal_access_token_handler.rs b/core/server/src/binary/handlers/personal_access_tokens/delete_personal_access_token_handler.rs
index edb1e8e9a..3e80f1dac 100644
--- a/core/server/src/binary/handlers/personal_access_tokens/delete_personal_access_token_handler.rs
+++ b/core/server/src/binary/handlers/personal_access_tokens/delete_personal_access_token_handler.rs
@@ -47,11 +47,12 @@ impl ServerCommandHandler for DeletePersonalAccessToken {
debug!("session: {session}, command: {self}");
shard.ensure_authenticated(session)?;
let token_name = self.name.clone();
+ let user_id = session.get_user_id();
shard
- .delete_personal_access_token(session, &self.name)
+ .delete_personal_access_token(user_id, &self.name)
.error(|e: &IggyError| {format!(
- "{COMPONENT} (error: {e}) - failed to delete personal access token with name: {token_name}, session: {session}"
+ "{COMPONENT} (error: {e}) - failed to delete personal access token with name: {token_name}, user: {user_id}"
)})?;
// Broadcast the event to other shards
diff --git a/core/server/src/binary/handlers/personal_access_tokens/get_personal_access_tokens_handler.rs b/core/server/src/binary/handlers/personal_access_tokens/get_personal_access_tokens_handler.rs
index 2f020c983..e4c3965f2 100644
--- a/core/server/src/binary/handlers/personal_access_tokens/get_personal_access_tokens_handler.rs
+++ b/core/server/src/binary/handlers/personal_access_tokens/get_personal_access_tokens_handler.rs
@@ -46,9 +46,12 @@ impl ServerCommandHandler for GetPersonalAccessTokens {
debug!("session: {session}, command: {self}");
shard.ensure_authenticated(session)?;
let personal_access_tokens = shard
- .get_personal_access_tokens(session)
+ .get_personal_access_tokens(session.get_user_id())
.error(|e: &IggyError| {
- format!("{COMPONENT} (error: {e}) - failed to get personal access tokens with session: {session}")
+ format!(
+ "{COMPONENT} (error: {e}) - failed to get personal access tokens for user: {}",
+ session.get_user_id()
+ )
})?;
let personal_access_tokens = mapper::map_personal_access_tokens(personal_access_tokens);
sender.send_ok_response(&personal_access_tokens).await?;
diff --git a/core/server/src/binary/handlers/segments/delete_segments_handler.rs b/core/server/src/binary/handlers/segments/delete_segments_handler.rs
index a4eb20654..b437af0bb 100644
--- a/core/server/src/binary/handlers/segments/delete_segments_handler.rs
+++ b/core/server/src/binary/handlers/segments/delete_segments_handler.rs
@@ -27,7 +27,6 @@ use crate::shard::transmission::message::{
ShardMessage, ShardRequest, ShardRequestPayload, ShardSendRequestResult,
};
use crate::state::command::EntryCommand;
-use crate::streaming;
use crate::streaming::session::Session;
use anyhow::Result;
use err_trail::ErrContext;
@@ -57,21 +56,14 @@ impl ServerCommandHandler for DeleteSegments {
let partition_id = self.partition_id as usize;
let segments_count = self.segments_count;
- // Ensure authentication and topic exists
shard.ensure_authenticated(session)?;
- shard.ensure_topic_exists(&stream_id, &topic_id)?;
- shard.ensure_partition_exists(&stream_id, &topic_id, partition_id)?;
-
- // Get numeric IDs for namespace
- let numeric_stream_id = shard
- .streams
- .with_stream_by_id(&stream_id, streaming::streams::helpers::get_stream_id());
-
- let numeric_topic_id = shard.streams.with_topic_by_id(
- &stream_id,
- &topic_id,
- streaming::topics::helpers::get_topic_id(),
- );
+ let (numeric_stream_id, numeric_topic_id, _) =
+ shard.resolve_partition_id(&stream_id, &topic_id, partition_id)?;
+ shard.permissioner.borrow().delete_segments(
+ session.get_user_id(),
+ numeric_stream_id,
+ numeric_topic_id,
+ )?;
// Route request to the correct shard
let namespace = IggyNamespace::new(numeric_stream_id, numeric_topic_id, partition_id);
diff --git a/core/server/src/binary/handlers/streams/create_stream_handler.rs b/core/server/src/binary/handlers/streams/create_stream_handler.rs
index 4d9e85d5d..4ff493517 100644
--- a/core/server/src/binary/handlers/streams/create_stream_handler.rs
+++ b/core/server/src/binary/handlers/streams/create_stream_handler.rs
@@ -55,6 +55,10 @@ impl ServerCommandHandler for CreateStream {
) -> Result<HandlerResult, IggyError> {
debug!("session: {session}, command: {self}");
shard.ensure_authenticated(session)?;
+ shard
+ .permissioner
+ .borrow()
+ .create_stream(session.get_user_id())?;
let request = ShardRequest {
stream_id: Identifier::default(),
@@ -75,7 +79,7 @@ impl ServerCommandHandler for CreateStream {
// Acquire stream lock to serialize filesystem operations
let _stream_guard = shard.fs_locks.stream_lock.lock().await;
- let stream = shard.create_stream(session, name).await?;
+ let stream = shard.create_stream(name).await?;
let created_stream_id = stream.id();
let event = ShardEvent::CreatedStream {
diff --git a/core/server/src/binary/handlers/streams/delete_stream_handler.rs b/core/server/src/binary/handlers/streams/delete_stream_handler.rs
index addc25af3..b4af2893c 100644
--- a/core/server/src/binary/handlers/streams/delete_stream_handler.rs
+++ b/core/server/src/binary/handlers/streams/delete_stream_handler.rs
@@ -53,6 +53,11 @@ impl ServerCommandHandler for DeleteStream {
) -> Result<HandlerResult, IggyError> {
debug!("session: {session}, command: {self}");
shard.ensure_authenticated(session)?;
+ let stream_id = shard.resolve_stream_id(&self.stream_id)?;
+ shard
+ .permissioner
+ .borrow()
+ .delete_stream(session.get_user_id(), stream_id)?;
let stream_id = self.stream_id.clone();
let request = ShardRequest {
stream_id: Identifier::default(),
@@ -73,11 +78,11 @@ impl ServerCommandHandler for DeleteStream {
// Acquire stream lock to serialize filesystem operations
let _stream_guard = shard.fs_locks.stream_lock.lock().await;
let stream = shard
- .delete_stream(session, &stream_id)
- .await
- .error(|e: &IggyError| {
- format!("{COMPONENT} (error: {e}) - failed to delete stream with ID: {stream_id}, session: {session}")
- })?;
+ .delete_stream(&stream_id)
+ .await
+ .error(|e: &IggyError| {
+ format!("{COMPONENT} (error: {e}) - failed to delete stream with ID: {stream_id}, session: {session}")
+ })?;
info!(
"Deleted stream with name: {}, ID: {}",
stream.root().name(),
diff --git a/core/server/src/binary/handlers/streams/get_stream_handler.rs b/core/server/src/binary/handlers/streams/get_stream_handler.rs
index fe9cd1f83..2f6be571b 100644
--- a/core/server/src/binary/handlers/streams/get_stream_handler.rs
+++ b/core/server/src/binary/handlers/streams/get_stream_handler.rs
@@ -24,9 +24,7 @@ use crate::binary::mapper;
use crate::shard::IggyShard;
use crate::slab::traits_ext::EntityComponentSystem;
use crate::streaming::session::Session;
-use crate::streaming::streams;
use anyhow::Result;
-use err_trail::ErrContext;
use iggy_common::IggyError;
use iggy_common::SenderKind;
use iggy_common::get_stream::GetStream;
@@ -47,30 +45,17 @@ impl ServerCommandHandler for GetStream {
) -> Result<HandlerResult, IggyError> {
debug!("session: {session}, command: {self}");
shard.ensure_authenticated(session)?;
- let exists = shard.ensure_stream_exists(&self.stream_id).is_ok();
- if !exists {
- sender.send_empty_ok_response().await?;
- return Ok(HandlerResult::Finished);
+ let stream_id = shard.resolve_stream_id(&self.stream_id);
+ if let Ok(id) = stream_id {
+ shard
+ .permissioner
+ .borrow()
+ .get_stream(session.get_user_id(), id)?;
}
- let stream_id = shard
- .streams
- .with_stream_by_id(&self.stream_id, streams::helpers::get_stream_id());
- let has_permission = shard
- .permissioner
- .borrow()
- .get_stream(session.get_user_id(), stream_id)
- .error(|e: &IggyError| {
- format!(
- "permission denied to get stream with ID: {} for user with ID: {}, error: {e}",
- self.stream_id,
- session.get_user_id(),
- )
- })
- .is_ok();
- if !has_permission {
+ let Ok(stream_id) = stream_id else {
sender.send_empty_ok_response().await?;
return Ok(HandlerResult::Finished);
- }
+ };
let response = shard
.streams
.with_components_by_id(stream_id, |(root, stats)| mapper::map_stream(&root, &stats));
diff --git a/core/server/src/binary/handlers/streams/purge_stream_handler.rs b/core/server/src/binary/handlers/streams/purge_stream_handler.rs
index c688ecd96..a352010dc 100644
--- a/core/server/src/binary/handlers/streams/purge_stream_handler.rs
+++ b/core/server/src/binary/handlers/streams/purge_stream_handler.rs
@@ -48,10 +48,16 @@ impl ServerCommandHandler for PurgeStream {
) -> Result<HandlerResult, IggyError> {
debug!("session: {session}, command: {self}");
shard.ensure_authenticated(session)?;
+ let stream_id = shard.resolve_stream_id(&self.stream_id)?;
+ shard
+ .permissioner
+ .borrow()
+ .purge_stream(session.get_user_id(), stream_id)?;
let stream_id = self.stream_id.clone();
shard
- .purge_stream(session, &self.stream_id).await
+ .purge_stream(&self.stream_id)
+ .await
.error(|e: &IggyError| {
format!("{COMPONENT} (error: {e}) - failed to purge stream with id: {stream_id}, session: {session}")
})?;
diff --git a/core/server/src/binary/handlers/streams/update_stream_handler.rs b/core/server/src/binary/handlers/streams/update_stream_handler.rs
index 3986f589c..eb9986666 100644
--- a/core/server/src/binary/handlers/streams/update_stream_handler.rs
+++ b/core/server/src/binary/handlers/streams/update_stream_handler.rs
@@ -48,12 +48,17 @@ impl ServerCommandHandler for UpdateStream {
) -> Result<HandlerResult, IggyError> {
debug!("session: {session}, command: {self}");
shard.ensure_authenticated(session)?;
+ let stream_id = shard.resolve_stream_id(&self.stream_id)?;
+ shard
+ .permissioner
+ .borrow()
+ .update_stream(session.get_user_id(), stream_id)?;
let stream_id = self.stream_id.clone();
shard
- .update_stream(session, &self.stream_id, self.name.clone())
- .error(|e: &IggyError| {
- format!("{COMPONENT} (error: {e}) - failed to update stream with id: {stream_id}, session: {session}")
- })?;
+ .update_stream(&self.stream_id, self.name.clone())
+ .error(|e: &IggyError| {
+ format!("{COMPONENT} (error: {e}) - failed to update stream with id: {stream_id}, session: {session}")
+ })?;
let event = ShardEvent::UpdatedStream {
stream_id: self.stream_id.clone(),
diff --git a/core/server/src/binary/handlers/system/get_client_handler.rs b/core/server/src/binary/handlers/system/get_client_handler.rs
index b62cd076e..40122d7b8 100644
--- a/core/server/src/binary/handlers/system/get_client_handler.rs
+++ b/core/server/src/binary/handlers/system/get_client_handler.rs
@@ -43,13 +43,12 @@ impl ServerCommandHandler for GetClient {
) -> Result<HandlerResult, IggyError> {
debug!("session: {session}, command: {self}");
shard.ensure_authenticated(session)?;
+ shard
+ .permissioner
+ .borrow()
+ .get_client(session.get_user_id())?;
- let Ok(client) = shard.get_client(session, self.client_id) else {
- sender.send_empty_ok_response().await?;
- return Ok(HandlerResult::Finished);
- };
-
- let Some(client) = client else {
+ let Some(client) = shard.get_client(self.client_id) else {
sender.send_empty_ok_response().await?;
return Ok(HandlerResult::Finished);
};
diff --git a/core/server/src/binary/handlers/system/get_clients_handler.rs b/core/server/src/binary/handlers/system/get_clients_handler.rs
index 5e365fa8b..4d1f1d779 100644
--- a/core/server/src/binary/handlers/system/get_clients_handler.rs
+++ b/core/server/src/binary/handlers/system/get_clients_handler.rs
@@ -19,12 +19,10 @@
use crate::binary::command::{
BinaryServerCommand, HandlerResult, ServerCommand, ServerCommandHandler,
};
-use crate::binary::handlers::system::COMPONENT;
use crate::binary::handlers::utils::receive_and_validate;
use crate::binary::mapper;
use crate::shard::IggyShard;
use crate::streaming::session::Session;
-use err_trail::ErrContext;
use iggy_common::IggyError;
use iggy_common::SenderKind;
use iggy_common::get_clients::GetClients;
@@ -45,10 +43,12 @@ impl ServerCommandHandler for GetClients {
) -> Result<HandlerResult, IggyError> {
debug!("session: {session}, command: {self}");
shard.ensure_authenticated(session)?;
+ shard
+ .permissioner
+ .borrow()
+ .get_clients(session.get_user_id())?;
- let clients = shard.get_clients(session).error(|e: &IggyError| {
- format!("{COMPONENT} (error: {e}) - failed to get clients, session: {session}")
- })?;
+ let clients = shard.get_clients();
let clients = mapper::map_clients(clients).await;
sender.send_ok_response(&clients).await?;
Ok(HandlerResult::Finished)
diff --git a/core/server/src/binary/handlers/system/get_me_handler.rs b/core/server/src/binary/handlers/system/get_me_handler.rs
index 6f9209bb1..6913a7529 100644
--- a/core/server/src/binary/handlers/system/get_me_handler.rs
+++ b/core/server/src/binary/handlers/system/get_me_handler.rs
@@ -19,12 +19,10 @@
use crate::binary::command::{
BinaryServerCommand, HandlerResult, ServerCommand, ServerCommandHandler,
};
-use crate::binary::handlers::system::COMPONENT;
use crate::binary::handlers::utils::receive_and_validate;
use crate::binary::mapper;
use crate::shard::IggyShard;
use crate::streaming::session::Session;
-use err_trail::ErrContext;
use iggy_common::IggyError;
use iggy_common::SenderKind;
use iggy_common::get_me::GetMe;
@@ -43,14 +41,7 @@ impl ServerCommandHandler for GetMe {
shard: &Rc<IggyShard>,
) -> Result<HandlerResult, IggyError> {
shard.ensure_authenticated(session)?;
- let Some(client) = shard.get_client(session, session.client_id).error(
- |e: &IggyError| {
- format!(
- "{COMPONENT} (error: {e}) - failed to get current client for session: {session}"
- )
- },
- )?
- else {
+ let Some(client) = shard.get_client(session.client_id) else {
return Err(IggyError::ClientNotFound(session.client_id));
};
diff --git a/core/server/src/binary/handlers/system/get_snapshot_handler.rs b/core/server/src/binary/handlers/system/get_snapshot_handler.rs
index 809f8fd60..082ef8b70 100644
--- a/core/server/src/binary/handlers/system/get_snapshot_handler.rs
+++ b/core/server/src/binary/handlers/system/get_snapshot_handler.rs
@@ -45,7 +45,7 @@ impl ServerCommandHandler for GetSnapshot {
shard.ensure_authenticated(session)?;
let snapshot = shard
- .get_snapshot(session, self.compression, &self.snapshot_types)
+ .get_snapshot(self.compression, &self.snapshot_types)
.await?;
let bytes = Bytes::copy_from_slice(&snapshot.0);
sender.send_ok_response(&bytes).await?;
diff --git a/core/server/src/binary/handlers/system/get_stats_handler.rs b/core/server/src/binary/handlers/system/get_stats_handler.rs
index ea787bea6..b0e9ddb1b 100644
--- a/core/server/src/binary/handlers/system/get_stats_handler.rs
+++ b/core/server/src/binary/handlers/system/get_stats_handler.rs
@@ -49,6 +49,10 @@ impl ServerCommandHandler for GetStats {
) -> Result<HandlerResult, IggyError> {
debug!("session: {session}, command: {self}");
shard.ensure_authenticated(session)?;
+ shard
+ .permissioner
+ .borrow()
+ .get_stats(session.get_user_id())?;
// Route GetStats to shard0 only
let request = ShardRequest {
diff --git a/core/server/src/binary/handlers/topics/create_topic_handler.rs b/core/server/src/binary/handlers/topics/create_topic_handler.rs
index 95cbbe4c5..7f66f2292 100644
--- a/core/server/src/binary/handlers/topics/create_topic_handler.rs
+++ b/core/server/src/binary/handlers/topics/create_topic_handler.rs
@@ -56,6 +56,11 @@ impl ServerCommandHandler for CreateTopic {
) -> Result<HandlerResult, IggyError> {
debug!("session: {session}, command: {self}");
shard.ensure_authenticated(session)?;
+ let stream_id = shard.resolve_stream_id(&self.stream_id)?;
+ shard
+ .permissioner
+ .borrow()
+ .create_topic(session.get_user_id(), stream_id)?;
let request = ShardRequest {
stream_id: Identifier::default(),
@@ -93,7 +98,6 @@ impl ServerCommandHandler for CreateTopic {
let topic = shard
.create_topic(
- session,
&stream_id,
name,
message_expiry,
@@ -118,7 +122,6 @@ impl ServerCommandHandler for CreateTopic {
shard.broadcast_event_to_all_shards(event).await?;
let partitions = shard
.create_partitions(
- session,
&stream_id,
&Identifier::numeric(topic_id as u32).unwrap(),
partitions_count,
diff --git a/core/server/src/binary/handlers/topics/delete_topic_handler.rs b/core/server/src/binary/handlers/topics/delete_topic_handler.rs
index 85db4d075..8b5d16ff8 100644
--- a/core/server/src/binary/handlers/topics/delete_topic_handler.rs
+++ b/core/server/src/binary/handlers/topics/delete_topic_handler.rs
@@ -55,6 +55,11 @@ impl ServerCommandHandler for DeleteTopic {
) -> Result<HandlerResult, IggyError> {
debug!("session: {session}, command: {self}");
shard.ensure_authenticated(session)?;
+ let (stream_id, topic_id) = shard.resolve_topic_id(&self.stream_id, &self.topic_id)?;
+ shard
+ .permissioner
+ .borrow()
+ .delete_topic(session.get_user_id(), stream_id, topic_id)?;
let request = ShardRequest {
stream_id: Identifier::default(),
@@ -80,7 +85,7 @@ impl ServerCommandHandler for DeleteTopic {
// Acquire topic lock to serialize filesystem operations
let _topic_guard = shard.fs_locks.topic_lock.lock().await;
- let topic = shard.delete_topic(session, &stream_id, &topic_id).await?;
+ let topic = shard.delete_topic(&stream_id, &topic_id).await?;
let stream_id_num = shard
.streams
.with_stream_by_id(&stream_id, streams::helpers::get_stream_id());
diff --git a/core/server/src/binary/handlers/topics/get_topic_handler.rs b/core/server/src/binary/handlers/topics/get_topic_handler.rs
index fcc42ac0b..1d6e01a2f 100644
--- a/core/server/src/binary/handlers/topics/get_topic_handler.rs
+++ b/core/server/src/binary/handlers/topics/get_topic_handler.rs
@@ -23,7 +23,6 @@ use crate::binary::handlers::utils::receive_and_validate;
use crate::binary::mapper;
use crate::shard::IggyShard;
use crate::streaming::session::Session;
-use crate::streaming::streams;
use anyhow::Result;
use iggy_common::IggyError;
use iggy_common::SenderKind;
@@ -45,35 +44,17 @@ impl ServerCommandHandler for GetTopic {
) -> Result<HandlerResult, IggyError> {
debug!("session: {session}, command: {self}");
shard.ensure_authenticated(session)?;
- let exists = shard
- .ensure_topic_exists(&self.stream_id, &self.topic_id)
- .is_ok();
- if !exists {
- sender.send_empty_ok_response().await?;
- return Ok(HandlerResult::Finished);
+ let topic_id = shard.resolve_topic_id(&self.stream_id, &self.topic_id);
+ if let Ok((stream_id, topic_id)) = topic_id {
+ shard
+ .permissioner
+ .borrow()
+ .get_topic(session.get_user_id(), stream_id, topic_id)?;
}
-
- let numeric_stream_id = shard
- .streams
- .with_stream_by_id(&self.stream_id, streams::helpers::get_stream_id());
- let has_permission = shard
- .permissioner
- .borrow()
- .get_topic(
- session.get_user_id(),
- numeric_stream_id,
- self.topic_id
- .get_u32_value()
- .unwrap_or(0)
- .try_into()
- .unwrap(),
- )
- .is_ok();
- if !has_permission {
+ let Ok(_) = topic_id else {
sender.send_empty_ok_response().await?;
return Ok(HandlerResult::Finished);
- }
-
+ };
let response =
shard
.streams
diff --git a/core/server/src/binary/handlers/topics/get_topics_handler.rs b/core/server/src/binary/handlers/topics/get_topics_handler.rs
index 96c3f9325..572edeb62 100644
--- a/core/server/src/binary/handlers/topics/get_topics_handler.rs
+++ b/core/server/src/binary/handlers/topics/get_topics_handler.rs
@@ -24,7 +24,6 @@ use crate::binary::mapper;
use crate::shard::IggyShard;
use crate::slab::traits_ext::{EntityComponentSystem, IntoComponents};
use crate::streaming::session::Session;
-use crate::streaming::streams;
use anyhow::Result;
use iggy_common::IggyError;
use iggy_common::SenderKind;
@@ -46,14 +45,12 @@ impl ServerCommandHandler for GetTopics {
) -> Result<HandlerResult, IggyError> {
debug!("session: {session}, command: {self}");
shard.ensure_authenticated(session)?;
- shard.ensure_stream_exists(&self.stream_id)?;
- let numeric_stream_id = shard
- .streams
- .with_stream_by_id(&self.stream_id, streams::helpers::get_stream_id());
+ let stream_id = shard.resolve_stream_id(&self.stream_id)?;
+
shard
.permissioner
.borrow()
- .get_topics(session.get_user_id(), numeric_stream_id)?;
+ .get_topics(session.get_user_id(), stream_id)?;
let response = shard.streams.with_topics(&self.stream_id, |topics| {
topics.with_components(|topics| {
diff --git a/core/server/src/binary/handlers/topics/purge_topic_handler.rs b/core/server/src/binary/handlers/topics/purge_topic_handler.rs
index 012795da6..6673a5ac3 100644
--- a/core/server/src/binary/handlers/topics/purge_topic_handler.rs
+++ b/core/server/src/binary/handlers/topics/purge_topic_handler.rs
@@ -46,11 +46,16 @@ impl ServerCommandHandler for PurgeTopic {
) -> Result<HandlerResult, IggyError> {
debug!("session: {session}, command: {self}");
shard.ensure_authenticated(session)?;
+ let (stream_id, topic_id) = shard.resolve_topic_id(&self.stream_id, &self.topic_id)?;
+ shard
+ .permissioner
+ .borrow()
+ .purge_topic(session.get_user_id(), stream_id, topic_id)?;
let topic_id = self.topic_id.clone();
let stream_id = self.stream_id.clone();
shard
- .purge_topic(session, &self.stream_id, &self.topic_id)
+ .purge_topic(&self.stream_id, &self.topic_id)
.await
.error(|e: &IggyError| {
format!(
diff --git a/core/server/src/binary/handlers/topics/update_topic_handler.rs b/core/server/src/binary/handlers/topics/update_topic_handler.rs
index aa8912fa7..41ba447ef 100644
--- a/core/server/src/binary/handlers/topics/update_topic_handler.rs
+++ b/core/server/src/binary/handlers/topics/update_topic_handler.rs
@@ -53,6 +53,11 @@ impl ServerCommandHandler for UpdateTopic {
) -> Result<HandlerResult, IggyError> {
debug!("session: {session}, command: {self}");
shard.ensure_authenticated(session)?;
+ let (stream_id, topic_id) = shard.resolve_topic_id(&self.stream_id, &self.topic_id)?;
+ shard
+ .permissioner
+ .borrow()
+ .update_topic(session.get_user_id(), stream_id, topic_id)?;
let request = ShardRequest {
stream_id: Identifier::default(),
@@ -86,7 +91,6 @@ impl ServerCommandHandler for UpdateTopic {
} = payload
{
shard.update_topic(
- session,
&stream_id,
&topic_id,
name.clone(),
diff --git a/core/server/src/binary/handlers/users/change_password_handler.rs b/core/server/src/binary/handlers/users/change_password_handler.rs
index 9b11969cc..ec8c80ece 100644
--- a/core/server/src/binary/handlers/users/change_password_handler.rs
+++ b/core/server/src/binary/handlers/users/change_password_handler.rs
@@ -51,20 +51,24 @@ impl ServerCommandHandler for ChangePassword {
debug!("session: {session}, command: {self}");
shard.ensure_authenticated(session)?;
+ // Check if user is changing someone else's password
+ let target_user = shard.get_user(&self.user_id)?;
+ if target_user.id != session.get_user_id() {
+ shard
+ .permissioner
+ .borrow()
+ .change_password(session.get_user_id())?;
+ }
+
info!("Changing password for user with ID: {}...", self.user_id);
shard
- .change_password(
- session,
- &self.user_id,
- &self.current_password,
- &self.new_password,
+ .change_password(&self.user_id, &self.current_password, &self.new_password)
+ .error(|e: &IggyError| {
+ format!(
+ "{COMPONENT} (error: {e}) - failed to change password for user_id: {}, session: {session}",
+ self.user_id
)
- .error(|e: &IggyError| {
- format!(
- "{COMPONENT} (error: {e}) - failed to change password for user_id: {}, session: {session}",
- self.user_id
- )
- })?;
+ })?;
info!("Changed password for user with ID: {}.", self.user_id);
diff --git a/core/server/src/binary/handlers/users/create_user_handler.rs b/core/server/src/binary/handlers/users/create_user_handler.rs
index 57501efc0..c49180b4b 100644
--- a/core/server/src/binary/handlers/users/create_user_handler.rs
+++ b/core/server/src/binary/handlers/users/create_user_handler.rs
@@ -55,6 +55,10 @@ impl ServerCommandHandler for CreateUser {
) -> Result<HandlerResult, IggyError> {
debug!("session: {session}, command: {self}");
shard.ensure_authenticated(session)?;
+ shard
+ .permissioner
+ .borrow()
+ .create_user(session.get_user_id())?;
let request = ShardRequest {
stream_id: Identifier::default(),
@@ -83,7 +87,7 @@ impl ServerCommandHandler for CreateUser {
{
let _user_guard = shard.fs_locks.user_lock.lock().await;
let user = shard
- .create_user(session, &username, &password, status, permissions.clone())
+ .create_user(&username, &password, status, permissions.clone())
.error(|e: &IggyError| {
format!(
"{COMPONENT} (error: {e}) - failed to create user with name: {}, session: {}",
diff --git a/core/server/src/binary/handlers/users/delete_user_handler.rs b/core/server/src/binary/handlers/users/delete_user_handler.rs
index 0c9151f91..a1d5c3538 100644
--- a/core/server/src/binary/handlers/users/delete_user_handler.rs
+++ b/core/server/src/binary/handlers/users/delete_user_handler.rs
@@ -54,6 +54,10 @@ impl ServerCommandHandler for DeleteUser {
) -> Result<HandlerResult, IggyError> {
debug!("session: {session}, command: {self}");
shard.ensure_authenticated(session)?;
+ shard
+ .permissioner
+ .borrow()
+ .delete_user(session.get_user_id())?;
let request = ShardRequest {
stream_id: Identifier::default(),
@@ -73,7 +77,7 @@ impl ServerCommandHandler for DeleteUser {
info!("Deleting user with ID: {}...", user_id);
let _user_guard = shard.fs_locks.user_lock.lock().await;
let user = shard
- .delete_user(session, &user_id)
+ .delete_user(&user_id)
.error(|e: &IggyError| {
format!(
"{COMPONENT} (error: {e}) - failed to delete user with ID: {}, session: {session}",
diff --git a/core/server/src/binary/handlers/users/get_user_handler.rs b/core/server/src/binary/handlers/users/get_user_handler.rs
index 4a60cdda4..3dc2e6bf1 100644
--- a/core/server/src/binary/handlers/users/get_user_handler.rs
+++ b/core/server/src/binary/handlers/users/get_user_handler.rs
@@ -44,7 +44,7 @@ impl ServerCommandHandler for GetUser {
) -> Result<HandlerResult, IggyError> {
debug!("session: {session}, command: {self}");
shard.ensure_authenticated(session)?;
- let Ok(user) = shard.find_user(session, &self.user_id) else {
+ let Ok(user) = shard.find_user(&self.user_id) else {
sender.send_empty_ok_response().await?;
return Ok(HandlerResult::Finished);
};
@@ -53,6 +53,14 @@ impl ServerCommandHandler for GetUser {
return Ok(HandlerResult::Finished);
};
+ // Permission check: only required if user is looking for someone else
+ if user.id != session.get_user_id() {
+ shard
+ .permissioner
+ .borrow()
+ .get_user(session.get_user_id())?;
+ }
+
let bytes = mapper::map_user(&user);
sender.send_ok_response(&bytes).await?;
Ok(HandlerResult::Finished)
diff --git a/core/server/src/binary/handlers/users/get_users_handler.rs b/core/server/src/binary/handlers/users/get_users_handler.rs
index b0ff5e76e..d783656e9 100644
--- a/core/server/src/binary/handlers/users/get_users_handler.rs
+++ b/core/server/src/binary/handlers/users/get_users_handler.rs
@@ -21,12 +21,10 @@ use std::rc::Rc;
use crate::binary::command::{
BinaryServerCommand, HandlerResult, ServerCommand, ServerCommandHandler,
};
-use crate::binary::handlers::users::COMPONENT;
use crate::binary::handlers::utils::receive_and_validate;
use crate::binary::mapper;
use crate::shard::IggyShard;
use crate::streaming::session::Session;
-use err_trail::ErrContext;
use iggy_common::IggyError;
use iggy_common::SenderKind;
use iggy_common::get_users::GetUsers;
@@ -46,9 +44,11 @@ impl ServerCommandHandler for GetUsers {
) -> Result<HandlerResult, IggyError> {
debug!("session: {session}, command: {self}");
shard.ensure_authenticated(session)?;
- let users = shard.get_users(session).await.error(|e: &IggyError| {
- format!("{COMPONENT} (error: {e}) - failed to get users, session: {session}")
- })?;
+ shard
+ .permissioner
+ .borrow()
+ .get_users(session.get_user_id())?;
+ let users = shard.get_users();
let users = mapper::map_users(users);
sender.send_ok_response(&users).await?;
Ok(HandlerResult::Finished)
diff --git a/core/server/src/binary/handlers/users/update_permissions_handler.rs b/core/server/src/binary/handlers/users/update_permissions_handler.rs
index 5e46b8a9b..5e6ee02f9 100644
--- a/core/server/src/binary/handlers/users/update_permissions_handler.rs
+++ b/core/server/src/binary/handlers/users/update_permissions_handler.rs
@@ -50,12 +50,25 @@ impl ServerCommandHandler for UpdatePermissions {
) -> Result<HandlerResult, IggyError> {
debug!("session: {session}, command: {self}");
shard.ensure_authenticated(session)?;
+ shard
+ .permissioner
+ .borrow()
+ .update_permissions(session.get_user_id())?;
+
+ // Check if target user is root - cannot change root user permissions
+ let target_user = shard.get_user(&self.user_id)?;
+ if target_user.is_root() {
+ return Err(IggyError::CannotChangePermissions(target_user.id));
+ }
shard
- .update_permissions(session, &self.user_id, self.permissions.clone())
- .error(|e: &IggyError| format!("{COMPONENT} (error: {e}) - failed to update permissions for user_id: {}, session: {session}",
+ .update_permissions(&self.user_id, self.permissions.clone())
+ .error(|e: &IggyError| {
+ format!(
+ "{COMPONENT} (error: {e}) - failed to update permissions for user_id: {}, session: {session}",
self.user_id
- ))?;
+ )
+ })?;
info!("Updated permissions for user with ID: {}.", self.user_id);
let event = ShardEvent::UpdatedPermissions {
user_id: self.user_id.clone(),
diff --git a/core/server/src/binary/handlers/users/update_user_handler.rs b/core/server/src/binary/handlers/users/update_user_handler.rs
index 433e024af..349d2bfe0 100644
--- a/core/server/src/binary/handlers/users/update_user_handler.rs
+++ b/core/server/src/binary/handlers/users/update_user_handler.rs
@@ -50,20 +50,19 @@ impl ServerCommandHandler for UpdateUser {
) -> Result<HandlerResult, IggyError> {
debug!("session: {session}, command: {self}");
shard.ensure_authenticated(session)?;
+ shard
+ .permissioner
+ .borrow()
+ .update_user(session.get_user_id())?;
- let user =shard
- .update_user(
- session,
- &self.user_id,
- self.username.clone(),
- self.status,
+ let user = shard
+ .update_user(&self.user_id, self.username.clone(), self.status)
+ .error(|e: &IggyError| {
+ format!(
+ "{COMPONENT} (error: {e}) - failed to update user with user_id: {}, session: {session}",
+ self.user_id
)
- .error(|e: &IggyError| {
- format!(
- "{COMPONENT} (error: {e}) - failed to update user with user_id: {}, session: {session}",
- self.user_id
- )
- })?;
+ })?;
info!("Updated user: {} with ID: {}.", user.username, user.id);
diff --git a/core/server/src/http/consumer_groups.rs b/core/server/src/http/consumer_groups.rs
index 6f66de799..11f381921 100644
--- a/core/server/src/http/consumer_groups.rs
+++ b/core/server/src/http/consumer_groups.rs
@@ -25,7 +25,6 @@ use crate::slab::traits_ext::{EntityComponentSystem, EntityMarker, IntoComponent
use crate::state::command::EntryCommand;
use crate::state::models::CreateConsumerGroupWithId;
use crate::streaming::polling_consumer::ConsumerGroupId;
-use crate::streaming::session::Session;
use axum::debug_handler;
use axum::extract::{Path, State};
use axum::http::StatusCode;
@@ -63,7 +62,16 @@ async fn get_consumer_group(
let identifier_topic_id = Identifier::from_str_value(&topic_id)?;
let identifier_group_id = Identifier::from_str_value(&group_id)?;
- let session = Session::stateless(identity.user_id, identity.ip_address);
+ let (numeric_stream_id, numeric_topic_id) = state
+ .shard
+ .shard()
+ .resolve_topic_id(&identifier_stream_id, &identifier_topic_id)?;
+ state
+ .shard
+ .shard()
+ .permissioner
+ .borrow()
+ .get_consumer_group(identity.user_id, numeric_stream_id, numeric_topic_id)?;
let exists = state
.shard
@@ -78,23 +86,6 @@ async fn get_consumer_group(
return Err(CustomError::ResourceNotFound);
}
- let numeric_topic_id = state.shard.shard().streams.with_topic_by_id(
- &identifier_stream_id,
- &identifier_topic_id,
- crate::streaming::topics::helpers::get_topic_id(),
- );
- let numeric_stream_id = state.shard.shard().streams.with_stream_by_id(
- &identifier_stream_id,
- crate::streaming::streams::helpers::get_stream_id(),
- );
-
- state
- .shard
- .shard()
- .permissioner
- .borrow()
- .get_consumer_group(session.get_user_id(), numeric_stream_id, numeric_topic_id)?;
-
let consumer_group = state.shard.shard().streams.with_consumer_group_by_id(
&identifier_stream_id,
&identifier_topic_id,
@@ -113,29 +104,16 @@ async fn get_consumer_groups(
let identifier_stream_id = Identifier::from_str_value(&stream_id)?;
let identifier_topic_id = Identifier::from_str_value(&topic_id)?;
- let session = Session::stateless(identity.user_id, identity.ip_address);
-
- state
+ let (numeric_stream_id, numeric_topic_id) = state
.shard
.shard()
- .ensure_topic_exists(&identifier_stream_id, &identifier_topic_id)?;
-
- let numeric_topic_id = state.shard.shard().streams.with_topic_by_id(
- &identifier_stream_id,
- &identifier_topic_id,
- crate::streaming::topics::helpers::get_topic_id(),
- );
- let numeric_stream_id = state.shard.shard().streams.with_stream_by_id(
- &identifier_stream_id,
- crate::streaming::streams::helpers::get_stream_id(),
- );
-
+ .resolve_topic_id(&identifier_stream_id, &identifier_topic_id)?;
state
.shard
.shard()
.permissioner
.borrow()
- .get_consumer_groups(session.get_user_id(), numeric_stream_id, numeric_topic_id)?;
+ .get_consumer_groups(identity.user_id, numeric_stream_id, numeric_topic_id)?;
let consumer_groups = state.shard.shard().streams.with_consumer_groups(
&identifier_stream_id,
@@ -163,11 +141,18 @@ async fn create_consumer_group(
command.topic_id = Identifier::from_str_value(&topic_id)?;
command.validate()?;
- let session = Session::stateless(identity.user_id, identity.ip_address);
+ let (numeric_stream_id, numeric_topic_id) = state
+ .shard
+ .shard()
+ .resolve_topic_id(&command.stream_id, &command.topic_id)?;
+ state
+ .shard
+ .shard()
+ .permissioner
+ .borrow()
+ .create_consumer_group(identity.user_id, numeric_stream_id, numeric_topic_id)?;
- // Create consumer group using the new API
let consumer_group = state.shard.shard().create_consumer_group(
- &session,
&command.stream_id,
&command.topic_id,
command.name.clone(),
@@ -233,11 +218,25 @@ async fn delete_consumer_group(
let identifier_group_id = Identifier::from_str_value(&group_id)?;
let result = SendWrapper::new(async move {
- let session = Session::stateless(identity.user_id, identity.ip_address);
+ let (stream_id_numeric, topic_id_numeric) = state
+ .shard
+ .shard()
+ .resolve_topic_id(&identifier_stream_id, &identifier_topic_id)?;
+ state
+ .shard
+ .shard()
+ .permissioner
+ .borrow()
+ .delete_consumer_group(identity.user_id, stream_id_numeric, topic_id_numeric)?;
+
+ // Now check if consumer group exists
+ state.shard.shard().ensure_consumer_group_exists(
+ &identifier_stream_id,
+ &identifier_topic_id,
+ &identifier_group_id,
+ )?;
- // Delete using the new API
let consumer_group = state.shard.shard().delete_consumer_group(
- &session,
&identifier_stream_id,
&identifier_topic_id,
&identifier_group_id
@@ -246,25 +245,14 @@ async fn delete_consumer_group(
let cg_id = consumer_group.id();
- // Remove all consumer group members from ClientManager
- let stream_id_usize = state.shard.shard().streams.with_stream_by_id(
- &identifier_stream_id,
- crate::streaming::streams::helpers::get_stream_id(),
- );
- let topic_id_usize = state.shard.shard().streams.with_topic_by_id(
- &identifier_stream_id,
- &identifier_topic_id,
- crate::streaming::topics::helpers::get_topic_id(),
- );
-
// TODO: Tech debt, repeated code from `delete_consumer_group_handler.rs`
// Get members from the deleted consumer group and make them leave
let slab = consumer_group.members().inner().shared_get();
for (_, member) in slab.iter() {
if let Err(err) = state.shard.shard().client_manager.leave_consumer_group(
member.client_id,
- stream_id_usize,
- topic_id_usize,
+ stream_id_numeric,
+ topic_id_numeric,
cg_id,
) {
tracing::warn!(
diff --git a/core/server/src/http/consumer_offsets.rs b/core/server/src/http/consumer_offsets.rs
index e7656a554..19fbfdb0a 100644
--- a/core/server/src/http/consumer_offsets.rs
+++ b/core/server/src/http/consumer_offsets.rs
@@ -20,7 +20,6 @@ use crate::http::COMPONENT;
use crate::http::error::CustomError;
use crate::http::jwt::json_web_token::Identity;
use crate::http::shared::AppState;
-use crate::streaming::session::Session;
use axum::debug_handler;
use axum::extract::{Path, Query, State};
use axum::http::StatusCode;
@@ -35,7 +34,6 @@ use iggy_common::Validatable;
use iggy_common::delete_consumer_offset::DeleteConsumerOffset;
use iggy_common::get_consumer_offset::GetConsumerOffset;
use iggy_common::store_consumer_offset::StoreConsumerOffset;
-use send_wrapper::SendWrapper;
use std::sync::Arc;
pub fn router(state: Arc<AppState>) -> Router {
@@ -62,12 +60,23 @@ async fn get_consumer_offset(
query.stream_id = Identifier::from_str_value(&stream_id)?;
query.topic_id = Identifier::from_str_value(&topic_id)?;
query.validate()?;
+
+ let (stream_id_numeric, topic_id_numeric) = state
+ .shard
+ .shard()
+ .resolve_topic_id(&query.stream_id, &query.topic_id)?;
+ state
+ .shard
+ .shard()
+ .permissioner
+ .borrow()
+ .get_consumer_offset(identity.user_id, stream_id_numeric, topic_id_numeric)?;
+
let consumer = Consumer::new(query.0.consumer.id);
- let session = SendWrapper::new(Session::stateless(identity.user_id, identity.ip_address));
let Ok(offset) = state
.shard
.get_consumer_offset(
- &session,
+ 0, // HTTP uses client_id 0 as it doesn't have persistent sessions
consumer,
&query.0.stream_id,
&query.0.topic_id,
@@ -95,11 +104,22 @@ async fn store_consumer_offset(
command.stream_id = Identifier::from_str_value(&stream_id)?;
command.topic_id = Identifier::from_str_value(&topic_id)?;
command.validate()?;
- let session = SendWrapper::new(Session::stateless(identity.user_id, identity.ip_address));
+
+ let (stream_id_numeric, topic_id_numeric) = state
+ .shard
+ .shard()
+ .resolve_topic_id(&command.stream_id, &command.topic_id)?;
+ state
+ .shard
+ .shard()
+ .permissioner
+ .borrow()
+ .store_consumer_offset(identity.user_id, stream_id_numeric, topic_id_numeric)?;
+
let consumer = Consumer::new(command.0.consumer.id);
state.shard
.store_consumer_offset(
- &session,
+ 0, // HTTP uses client_id 0 as it doesn't have persistent sessions
consumer,
&command.0.stream_id,
&command.0.topic_id,
@@ -118,12 +138,22 @@ async fn delete_consumer_offset(
Path((stream_id, topic_id, consumer_id)): Path<(String, String, String)>,
query: Query<DeleteConsumerOffset>,
) -> Result<StatusCode, CustomError> {
+ let (stream_id_numeric, topic_id_numeric) = state
+ .shard
+ .shard()
+ .resolve_topic_id(&query.stream_id, &query.topic_id)?;
+ state
+ .shard
+ .shard()
+ .permissioner
+ .borrow()
+ .delete_consumer_offset(identity.user_id, stream_id_numeric, topic_id_numeric)?;
+
let consumer = Consumer::new(consumer_id.try_into()?);
- let session = SendWrapper::new(Session::stateless(identity.user_id, identity.ip_address));
state
.shard
.delete_consumer_offset(
- &session,
+ 0, // HTTP uses client_id 0 as it doesn't have persistent sessions
consumer,
&query.stream_id,
&query.topic_id,
diff --git a/core/server/src/http/http_shard_wrapper.rs b/core/server/src/http/http_shard_wrapper.rs
index 7bcfdb65b..e48abd6c4 100644
--- a/core/server/src/http/http_shard_wrapper.rs
+++ b/core/server/src/http/http_shard_wrapper.rs
@@ -65,14 +65,14 @@ impl HttpSafeShard {
pub async fn get_consumer_offset(
&self,
- session: &SendWrapper<Session>,
+ client_id: u32,
consumer: Consumer,
stream_id: &Identifier,
topic_id: &Identifier,
partition_id: Option<u32>,
) -> Result<Option<ConsumerOffsetInfo>, IggyError> {
let future = SendWrapper::new(self.shard().get_consumer_offset(
- session,
+ client_id,
consumer,
stream_id,
topic_id,
@@ -83,7 +83,7 @@ impl HttpSafeShard {
pub async fn store_consumer_offset(
&self,
- session: &SendWrapper<Session>,
+ client_id: u32,
consumer: Consumer,
stream_id: &Identifier,
topic_id: &Identifier,
@@ -91,7 +91,7 @@ impl HttpSafeShard {
offset: u64,
) -> Result<(), IggyError> {
let future = SendWrapper::new(self.shard().store_consumer_offset(
- session,
+ client_id,
consumer,
stream_id,
topic_id,
@@ -104,14 +104,14 @@ impl HttpSafeShard {
pub async fn delete_consumer_offset(
&self,
- session: &SendWrapper<Session>,
+ client_id: u32,
consumer: Consumer,
stream_id: &Identifier,
topic_id: &Identifier,
partition_id: Option<u32>,
) -> Result<(), IggyError> {
let future = SendWrapper::new(self.shard().delete_consumer_offset(
- session,
+ client_id,
consumer,
stream_id,
topic_id,
@@ -121,40 +121,26 @@ impl HttpSafeShard {
Ok(())
}
- pub async fn delete_stream(
- &self,
- session: &Session,
- stream_id: &Identifier,
- ) -> Result<(), IggyError> {
- let future = SendWrapper::new(self.shard().delete_stream(session, stream_id));
+ pub async fn delete_stream(&self, stream_id: &Identifier) -> Result<(), IggyError> {
+ let future = SendWrapper::new(self.shard().delete_stream(stream_id));
future.await?;
Ok(())
}
- pub fn update_stream(
- &self,
- session: &Session,
- stream_id: &Identifier,
- name: String,
- ) -> Result<(), IggyError> {
- self.shard().update_stream(session, stream_id, name)
+ pub fn update_stream(&self, stream_id: &Identifier, name: String) -> Result<(), IggyError> {
+ self.shard().update_stream(stream_id, name)
}
- pub async fn purge_stream(
- &self,
- session: &Session,
- stream_id: &Identifier,
- ) -> Result<(), IggyError> {
- let future = SendWrapper::new(self.shard().purge_stream(session, stream_id));
+ pub async fn purge_stream(&self, stream_id: &Identifier) -> Result<(), IggyError> {
+ let future = SendWrapper::new(self.shard().purge_stream(stream_id));
future.await
}
pub async fn create_stream(
&self,
- session: &Session,
name: String,
) -> Result<crate::streaming::streams::stream::Stream, IggyError> {
- let future = SendWrapper::new(self.shard().create_stream(session, name));
+ let future = SendWrapper::new(self.shard().create_stream(name));
future.await
}
@@ -166,55 +152,50 @@ impl HttpSafeShard {
self.shard().state.apply(user_id, command).await
}
- pub async fn get_users(&self, session: &Session) -> Result<Vec<User>, IggyError> {
- self.shard().get_users(session).await
+ pub fn get_users(&self) -> Vec<User> {
+ self.shard().get_users()
}
pub fn create_user(
&self,
- session: &Session,
username: &str,
password: &str,
status: UserStatus,
permissions: Option<Permissions>,
) -> Result<User, IggyError> {
self.shard()
- .create_user(session, username, password, status, permissions)
+ .create_user(username, password, status, permissions)
}
- pub fn delete_user(&self, session: &Session, user_id: &Identifier) -> Result<User, IggyError> {
- self.shard().delete_user(session, user_id)
+ pub fn delete_user(&self, user_id: &Identifier) -> Result<User, IggyError> {
+ self.shard().delete_user(user_id)
}
pub fn update_user(
&self,
- session: &Session,
user_id: &Identifier,
username: Option<String>,
status: Option<UserStatus>,
) -> Result<User, IggyError> {
- self.shard().update_user(session, user_id, username, status)
+ self.shard().update_user(user_id, username, status)
}
pub fn update_permissions(
&self,
- session: &Session,
user_id: &Identifier,
permissions: Option<Permissions>,
) -> Result<(), IggyError> {
- self.shard()
- .update_permissions(session, user_id, permissions)
+ self.shard().update_permissions(user_id, permissions)
}
- pub async fn change_password(
+ pub fn change_password(
&self,
- session: &Session,
user_id: &Identifier,
current_password: &str,
new_password: &str,
) -> Result<(), IggyError> {
self.shard()
- .change_password(session, user_id, current_password, new_password)
+ .change_password(user_id, current_password, new_password)
}
pub fn login_user(
@@ -232,27 +213,23 @@ impl HttpSafeShard {
pub fn get_personal_access_tokens(
&self,
- session: &Session,
+ user_id: u32,
) -> Result<Vec<PersonalAccessToken>, IggyError> {
- self.shard().get_personal_access_tokens(session)
+ self.shard().get_personal_access_tokens(user_id)
}
pub fn create_personal_access_token(
&self,
- session: &Session,
+ user_id: u32,
name: &str,
expiry: IggyExpiry,
) -> Result<(PersonalAccessToken, String), IggyError> {
self.shard()
- .create_personal_access_token(session, name, expiry)
+ .create_personal_access_token(user_id, name, expiry)
}
- pub fn delete_personal_access_token(
- &self,
- session: &Session,
- name: &str,
- ) -> Result<(), IggyError> {
- self.shard().delete_personal_access_token(session, name)
+ pub fn delete_personal_access_token(&self, user_id: u32, name: &str) -> Result<(), IggyError> {
+ self.shard().delete_personal_access_token(user_id, name)
}
pub fn login_with_personal_access_token(
diff --git a/core/server/src/http/messages.rs b/core/server/src/http/messages.rs
index 5a459491f..6d73ff5bc 100644
--- a/core/server/src/http/messages.rs
+++ b/core/server/src/http/messages.rs
@@ -75,7 +75,8 @@ async fn poll_messages(
PollingArgs::new(query.0.strategy, query.0.count, query.0.auto_commit),
));
- let (metadata, messages) = poll_future.await
+ let (metadata, messages) = poll_future
+ .await
.error(|e: &IggyError| {
format!(
"{COMPONENT} (error: {e}) - failed to poll messages, stream ID: {}, topic ID: {}, partition ID: {:?}",
@@ -112,7 +113,8 @@ async fn send_messages(
batch,
));
- append_future.await
+ append_future
+ .await
.error(|e: &IggyError| {
format!(
"{COMPONENT} (error: {e}) - failed to append messages, stream ID: {stream_id}, topic ID: {topic_id}"
diff --git a/core/server/src/http/partitions.rs b/core/server/src/http/partitions.rs
index 959e1fd2c..e2ec4942c 100644
--- a/core/server/src/http/partitions.rs
+++ b/core/server/src/http/partitions.rs
@@ -22,7 +22,6 @@ use crate::http::jwt::json_web_token::Identity;
use crate::http::shared::AppState;
use crate::shard::transmission::event::ShardEvent;
use crate::state::command::EntryCommand;
-use crate::streaming::session::Session;
use axum::extract::{Path, Query, State};
use axum::http::StatusCode;
use axum::routing::post;
@@ -58,10 +57,19 @@ async fn create_partitions(
command.topic_id = Identifier::from_str_value(&topic_id)?;
command.validate()?;
+ let (numeric_stream_id, numeric_topic_id) = state
+ .shard
+ .shard()
+ .resolve_topic_id(&command.stream_id, &command.topic_id)?;
+ state
+ .shard
+ .shard()
+ .permissioner
+ .borrow()
+ .create_partitions(identity.user_id, numeric_stream_id, numeric_topic_id)?;
+
let _parititon_guard = state.shard.shard().fs_locks.partition_lock.lock().await;
- let session = Session::stateless(identity.user_id, identity.ip_address);
let partitions = SendWrapper::new(state.shard.shard().create_partitions(
- &session,
&command.stream_id,
&command.topic_id,
command.partitions_count,
@@ -112,10 +120,19 @@ async fn delete_partitions(
query.topic_id = Identifier::from_str_value(&topic_id)?;
query.validate()?;
- let session = Session::stateless(identity.user_id, identity.ip_address);
+ let (numeric_stream_id, numeric_topic_id) = state
+ .shard
+ .shard()
+ .resolve_topic_id(&query.stream_id, &query.topic_id)?;
+ state
+ .shard
+ .shard()
+ .permissioner
+ .borrow()
+ .delete_partitions(identity.user_id, numeric_stream_id, numeric_topic_id)?;
+
let deleted_partition_ids = {
let delete_future = SendWrapper::new(state.shard.shard().delete_partitions(
- &session,
&query.stream_id,
&query.topic_id,
query.partitions_count,
diff --git a/core/server/src/http/personal_access_tokens.rs b/core/server/src/http/personal_access_tokens.rs
index 75382fb76..94d89bdb4 100644
--- a/core/server/src/http/personal_access_tokens.rs
+++ b/core/server/src/http/personal_access_tokens.rs
@@ -24,7 +24,6 @@ use crate::http::mapper::map_generated_access_token_to_identity_info;
use crate::http::shared::AppState;
use crate::state::command::EntryCommand;
use crate::state::models::CreatePersonalAccessTokenWithHash;
-use crate::streaming::session::Session;
use axum::extract::{Path, State};
use axum::http::StatusCode;
use axum::routing::{delete, get, post};
@@ -66,7 +65,7 @@ async fn get_personal_access_tokens(
let personal_access_tokens = state
.shard
.shard()
- .get_personal_access_tokens(&Session::stateless(identity.user_id, identity.ip_address))
+ .get_personal_access_tokens(identity.user_id)
.error(|e: &IggyError| {
format!(
"{COMPONENT} (error: {e}) - failed to get personal access tokens, user ID: {}",
@@ -87,11 +86,7 @@ async fn create_personal_access_token(
command.validate()?;
let (_personal_access_token, token) = state
.shard
- .create_personal_access_token(
- &Session::stateless(identity.user_id, identity.ip_address),
- &command.name,
- command.expiry,
- )
+ .create_personal_access_token(identity.user_id, &command.name, command.expiry)
.error(|e: &IggyError| {
format!(
"{COMPONENT} (error: {e}) - failed to create personal access token, user ID: {}",
@@ -126,10 +121,7 @@ async fn delete_personal_access_token(
) -> Result<StatusCode, CustomError> {
state
.shard
- .delete_personal_access_token(
- &Session::stateless(identity.user_id, identity.ip_address),
- &name,
- )
+ .delete_personal_access_token(identity.user_id, &name)
.error(|e: &IggyError| {
format!(
"{COMPONENT} (error: {e}) - failed to delete personal access token, user ID: {}",
diff --git a/core/server/src/http/streams.rs b/core/server/src/http/streams.rs
index d96fd4e1e..79b09bd7e 100644
--- a/core/server/src/http/streams.rs
+++ b/core/server/src/http/streams.rs
@@ -21,7 +21,6 @@ use crate::http::error::CustomError;
use crate::http::jwt::json_web_token::Identity;
use crate::http::shared::AppState;
use crate::slab::traits_ext::{EntityComponentSystem, IntoComponents};
-use crate::streaming::session::Session;
use axum::extract::{Path, State};
use axum::http::StatusCode;
use axum::routing::{delete, get};
@@ -59,12 +58,14 @@ async fn get_stream(
Path(stream_id): Path<String>,
) -> Result<Json<StreamDetails>, CustomError> {
let stream_id = Identifier::from_str_value(&stream_id)?;
- let exists = state.shard.shard().ensure_stream_exists(&stream_id).is_ok();
- if !exists {
- return Err(CustomError::ResourceNotFound);
- }
- let _session = Session::stateless(identity.user_id, identity.ip_address);
+ let stream_id_numeric = state.shard.shard().resolve_stream_id(&stream_id)?;
+ state
+ .shard
+ .shard()
+ .permissioner
+ .borrow()
+ .get_stream(identity.user_id, stream_id_numeric)?;
// Use direct slab access for thread-safe stream retrieval
let stream_details = SendWrapper::new(|| {
@@ -85,7 +86,12 @@ async fn get_streams(
State(state): State<Arc<AppState>>,
Extension(identity): Extension<Identity>,
) -> Result<Json<Vec<Stream>>, CustomError> {
- let _session = Session::stateless(identity.user_id, identity.ip_address);
+ state
+ .shard
+ .shard()
+ .permissioner
+ .borrow()
+ .get_streams(identity.user_id)?;
// Use direct slab access for thread-safe streams retrieval
let streams = SendWrapper::new(|| {
@@ -106,15 +112,19 @@ async fn create_stream(
Json(command): Json<CreateStream>,
) -> Result<Json<StreamDetails>, CustomError> {
command.validate()?;
+ state
+ .shard
+ .shard()
+ .permissioner
+ .borrow()
+ .create_stream(identity.user_id)?;
let result = SendWrapper::new(async move {
- let session = Session::stateless(identity.user_id, identity.ip_address);
-
let _stream_guard = state.shard.shard().fs_locks.stream_lock.lock().await;
// Create stream using wrapper method
let stream = state
.shard
- .create_stream(&session, command.name.clone())
+ .create_stream(command.name.clone())
.await
.error(|e: &IggyError| {
format!(
@@ -184,13 +194,19 @@ async fn update_stream(
command.stream_id = Identifier::from_str_value(&stream_id)?;
command.validate()?;
- let result = SendWrapper::new(async move {
- let session = Session::stateless(identity.user_id, identity.ip_address);
+ let stream_id_numeric = state.shard.shard().resolve_stream_id(&command.stream_id)?;
+ state
+ .shard
+ .shard()
+ .permissioner
+ .borrow()
+ .update_stream(identity.user_id, stream_id_numeric)?;
+ let result = SendWrapper::new(async move {
// Update stream using wrapper method
state
.shard
- .update_stream(&session, &command.stream_id, command.name.clone())
+ .update_stream(&command.stream_id, command.name.clone())
.error(|e: &IggyError| {
format!(
"{COMPONENT} (error: {e}) - failed to update stream, stream ID: {stream_id}"
@@ -236,19 +252,24 @@ async fn delete_stream(
) -> Result<StatusCode, CustomError> {
let identifier_stream_id = Identifier::from_str_value(&stream_id)?;
+ let stream_id_numeric = state
+ .shard
+ .shard()
+ .resolve_stream_id(&identifier_stream_id)?;
+ state
+ .shard
+ .shard()
+ .permissioner
+ .borrow()
+ .delete_stream(identity.user_id, stream_id_numeric)?;
+
let result =
SendWrapper::new(async move {
- let session = Session::stateless(identity.user_id, identity.ip_address);
-
let _stream_guard = state.shard.shard().fs_locks.stream_lock.lock().await;
// Delete stream and get the stream entity
let stream = {
- let future = SendWrapper::new(
- state
- .shard
- .shard()
- .delete_stream(&session, &identifier_stream_id),
- );
+ let future =
+ SendWrapper::new(state.shard.shard().delete_stream(&identifier_stream_id));
future.await
}
.error(|e: &IggyError| {
@@ -299,13 +320,22 @@ async fn purge_stream(
) -> Result<StatusCode, CustomError> {
let identifier_stream_id = Identifier::from_str_value(&stream_id)?;
- let result = SendWrapper::new(async move {
- let session = Session::stateless(identity.user_id, identity.ip_address);
+ let stream_id_numeric = state
+ .shard
+ .shard()
+ .resolve_stream_id(&identifier_stream_id)?;
+ state
+ .shard
+ .shard()
+ .permissioner
+ .borrow()
+ .purge_stream(identity.user_id, stream_id_numeric)?;
+ let result = SendWrapper::new(async move {
// Purge stream using wrapper method
state
.shard
- .purge_stream(&session, &identifier_stream_id)
+ .purge_stream(&identifier_stream_id)
.await
.error(|e: &IggyError| {
format!("{COMPONENT} (error: {e}) - failed to purge stream, stream ID: {stream_id}")
diff --git a/core/server/src/http/system.rs b/core/server/src/http/system.rs
index af442a922..ead8b416b 100644
--- a/core/server/src/http/system.rs
+++ b/core/server/src/http/system.rs
@@ -22,7 +22,6 @@ use crate::http::error::CustomError;
use crate::http::jwt::json_web_token::Identity;
use crate::http::mapper;
use crate::http::shared::AppState;
-use crate::streaming::session::Session;
use axum::body::Body;
use axum::extract::{Path, State};
use axum::http::{HeaderMap, header};
@@ -69,7 +68,12 @@ async fn get_stats(
State(state): State<Arc<AppState>>,
Extension(identity): Extension<Identity>,
) -> Result<Json<Stats>, CustomError> {
- let _session = Session::stateless(identity.user_id, identity.ip_address);
+ state
+ .shard
+ .shard()
+ .permissioner
+ .borrow()
+ .get_stats(identity.user_id)?;
let stats_future = SendWrapper::new(state.shard.shard().get_stats());
let stats = stats_future.await.error(|e: &IggyError| {
format!(
@@ -84,16 +88,8 @@ async fn get_cluster_metadata(
State(state): State<Arc<AppState>>,
Extension(identity): Extension<Identity>,
) -> Result<Json<ClusterMetadata>, CustomError> {
- let shard = state.shard.shard();
- let session = Session::stateless(identity.user_id, identity.ip_address);
- let cluster_metadata = shard
- .get_cluster_metadata(&session)
- .error(|e: &IggyError| {
- format!(
- "{COMPONENT} (error: {e}) - failed to get cluster metadata, user ID: {}",
- identity.user_id
- )
- })?;
+ let _ = identity; // authenticated via middleware
+ let cluster_metadata = state.shard.shard().get_cluster_metadata();
Ok(Json(cluster_metadata))
}
@@ -102,23 +98,13 @@ async fn get_client(
Extension(identity): Extension<Identity>,
Path(client_id): Path<u32>,
) -> Result<Json<ClientInfoDetails>, CustomError> {
- let Ok(client) = state
+ state
.shard
.shard()
- .get_client(
- &Session::stateless(identity.user_id, identity.ip_address),
- client_id,
- )
- .error(|e: &IggyError| {
- format!(
- "{COMPONENT} (error: {e}) - failed to get client, user ID: {}",
- identity.user_id
- )
- })
- else {
- return Err(CustomError::ResourceNotFound);
- };
- let Some(client) = client else {
+ .permissioner
+ .borrow()
+ .get_client(identity.user_id)?;
+ let Some(client) = state.shard.shard().get_client(client_id) else {
return Err(CustomError::ResourceNotFound);
};
@@ -131,16 +117,13 @@ async fn get_clients(
State(state): State<Arc<AppState>>,
Extension(identity): Extension<Identity>,
) -> Result<Json<Vec<ClientInfo>>, CustomError> {
- let clients = state
+ state
.shard
.shard()
- .get_clients(&Session::stateless(identity.user_id, identity.ip_address))
- .error(|e: &IggyError| {
- format!(
- "{COMPONENT} (error: {e}) - failed to get clients, user ID: {}",
- identity.user_id
- )
- })?;
+ .permissioner
+ .borrow()
+ .get_clients(identity.user_id)?;
+ let clients = state.shard.shard().get_clients();
let clients = mapper::map_clients(&clients);
Ok(Json(clients))
}
@@ -148,17 +131,17 @@ async fn get_clients(
#[debug_handler]
async fn get_snapshot(
State(state): State<Arc<AppState>>,
- Extension(identity): Extension<Identity>,
+ Extension(_identity): Extension<Identity>,
Json(command): Json<GetSnapshot>,
) -> Result<impl IntoResponse, CustomError> {
command.validate()?;
- let session = Session::stateless(identity.user_id, identity.ip_address);
- let snapshot_future = SendWrapper::new(state.shard.shard().get_snapshot(
- &session,
- command.compression,
- &command.snapshot_types,
- ));
+ let snapshot_future = SendWrapper::new(
+ state
+ .shard
+ .shard()
+ .get_snapshot(command.compression, &command.snapshot_types),
+ );
let snapshot = snapshot_future
.await
diff --git a/core/server/src/http/topics.rs b/core/server/src/http/topics.rs
index cc686e69f..9cb62eb71 100644
--- a/core/server/src/http/topics.rs
+++ b/core/server/src/http/topics.rs
@@ -23,8 +23,6 @@ use crate::http::shared::AppState;
use crate::slab::traits_ext::{EntityComponentSystem, EntityMarker, IntoComponents};
use crate::state::command::EntryCommand;
use crate::state::models::CreateTopicWithId;
-use crate::streaming::session::Session;
-use crate::streaming::{streams, topics};
use axum::extract::{Path, State};
use axum::http::StatusCode;
use axum::routing::{delete, get};
@@ -67,45 +65,20 @@ async fn get_topic(
let identity_stream_id = Identifier::from_str_value(&stream_id)?;
let identity_topic_id = Identifier::from_str_value(&topic_id)?;
- let session = Session::stateless(identity.user_id, identity.ip_address);
-
- let stream_exists = state
+ let (numeric_stream_id, numeric_topic_id) = state
.shard
.shard()
- .ensure_stream_exists(&identity_stream_id)
- .is_ok();
- if !stream_exists {
- return Err(CustomError::ResourceNotFound);
- }
-
- let topic_exists = state
- .shard
- .shard()
- .ensure_topic_exists(&identity_stream_id, &identity_topic_id)
- .is_ok();
- if !topic_exists {
- return Err(CustomError::ResourceNotFound);
- }
-
- let numeric_stream_id = state
+ .resolve_topic_id(&identity_stream_id, &identity_topic_id)?;
+ state
.shard
.shard()
- .streams
- .with_stream_by_id(&identity_stream_id, streams::helpers::get_stream_id());
- let numeric_topic_id = state.shard.shard().streams.with_topic_by_id(
- &identity_stream_id,
- &identity_topic_id,
- topics::helpers::get_topic_id(),
- );
-
- state.shard.shard()
.permissioner
.borrow()
- .get_topic(session.get_user_id(), numeric_stream_id, numeric_topic_id)
+ .get_topic(identity.user_id, numeric_stream_id, numeric_topic_id)
.error(|e: &IggyError| {
format!(
"{COMPONENT} (error: {e}) - permission denied to get topic with ID: {topic_id} in stream with ID: {stream_id} for user with ID: {}",
- session.get_user_id(),
+ identity.user_id,
)
})?;
@@ -126,24 +99,18 @@ async fn get_topics(
Path(stream_id): Path<String>,
) -> Result<Json<Vec<Topic>>, CustomError> {
let stream_id = Identifier::from_str_value(&stream_id)?;
- let session = Session::stateless(identity.user_id, identity.ip_address);
-
- state.shard.shard().ensure_stream_exists(&stream_id)?;
- let numeric_stream_id = state
+ let numeric_stream_id = state.shard.shard().resolve_stream_id(&stream_id)?;
+ state
.shard
.shard()
- .streams
- .with_stream_by_id(&stream_id, streams::helpers::get_stream_id());
-
- state.shard.shard()
.permissioner
.borrow()
- .get_topics(session.get_user_id(), numeric_stream_id)
+ .get_topics(identity.user_id, numeric_stream_id)
.error(|e: &IggyError| {
format!(
"{COMPONENT} (error: {e}) - permission denied to get topics in stream with ID: {stream_id} for user with ID: {}",
- session.get_user_id(),
+ identity.user_id,
)
})?;
@@ -173,12 +140,17 @@ async fn create_topic(
command.stream_id = Identifier::from_str_value(&stream_id)?;
command.validate()?;
- let session = SendWrapper::new(Session::stateless(identity.user_id, identity.ip_address));
+ let numeric_stream_id = state.shard.shard().resolve_stream_id(&command.stream_id)?;
+ state
+ .shard
+ .shard()
+ .permissioner
+ .borrow()
+ .create_topic(identity.user_id, numeric_stream_id)?;
let _topic_guard = state.shard.shard().fs_locks.topic_lock.lock().await;
let topic = {
let future = SendWrapper::new(state.shard.shard().create_topic(
- &session,
&command.stream_id,
command.name.clone(),
command.message_expiry,
@@ -213,7 +185,6 @@ async fn create_topic(
// Create partitions
let partitions = shard
.create_partitions(
- &session,
&command.stream_id,
&Identifier::numeric(topic_id as u32).unwrap(),
command.partitions_count,
@@ -281,11 +252,18 @@ async fn update_topic(
command.topic_id = Identifier::from_str_value(&topic_id)?;
command.validate()?;
- let session = Session::stateless(identity.user_id, identity.ip_address);
+ let (numeric_stream_id, numeric_topic_id) = state
+ .shard
+ .shard()
+ .resolve_topic_id(&command.stream_id, &command.topic_id)?;
+ state.shard.shard().permissioner.borrow().update_topic(
+ identity.user_id,
+ numeric_stream_id,
+ numeric_topic_id,
+ )?;
let name_changed = !command.name.is_empty();
state.shard.shard().update_topic(
- &session,
&command.stream_id,
&command.topic_id,
command.name.clone(),
@@ -362,12 +340,19 @@ async fn delete_topic(
let identifier_stream_id = Identifier::from_str_value(&stream_id)?;
let identifier_topic_id = Identifier::from_str_value(&topic_id)?;
- let session = Session::stateless(identity.user_id, identity.ip_address);
+ let (numeric_stream_id, numeric_topic_id) = state
+ .shard
+ .shard()
+ .resolve_topic_id(&identifier_stream_id, &identifier_topic_id)?;
+ state.shard.shard().permissioner.borrow().delete_topic(
+ identity.user_id,
+ numeric_stream_id,
+ numeric_topic_id,
+ )?;
let _topic_guard = state.shard.shard().fs_locks.topic_lock.lock().await;
let topic = {
let future = SendWrapper::new(state.shard.shard().delete_topic(
- &session,
&identifier_stream_id,
&identifier_topic_id,
));
@@ -428,11 +413,18 @@ async fn purge_topic(
let identifier_stream_id = Identifier::from_str_value(&stream_id)?;
let identifier_topic_id = Identifier::from_str_value(&topic_id)?;
- let session = Session::stateless(identity.user_id, identity.ip_address);
+ let (numeric_stream_id, numeric_topic_id) = state
+ .shard
+ .shard()
+ .resolve_topic_id(&identifier_stream_id, &identifier_topic_id)?;
+ state.shard.shard().permissioner.borrow().purge_topic(
+ identity.user_id,
+ numeric_stream_id,
+ numeric_topic_id,
+ )?;
{
let future = SendWrapper::new(state.shard.shard().purge_topic(
- &session,
&identifier_stream_id,
&identifier_topic_id,
));
diff --git a/core/server/src/http/users.rs b/core/server/src/http/users.rs
index 29ae19230..ae7475ec4 100644
--- a/core/server/src/http/users.rs
+++ b/core/server/src/http/users.rs
@@ -69,16 +69,23 @@ async fn get_user(
Path(user_id): Path<String>,
) -> Result<Json<UserInfoDetails>, CustomError> {
let identifier_user_id = Identifier::from_str_value(&user_id)?;
- let Ok(user) = state.shard.shard().find_user(
- &Session::stateless(identity.user_id, identity.ip_address),
- &identifier_user_id,
- ) else {
+ let Ok(user) = state.shard.shard().find_user(&identifier_user_id) else {
return Err(CustomError::ResourceNotFound);
};
let Some(user) = user else {
return Err(CustomError::ResourceNotFound);
};
+ // Permission check: only required if user is looking for someone else
+ if user.id != identity.user_id {
+ state
+ .shard
+ .shard()
+ .permissioner
+ .borrow()
+ .get_user(identity.user_id)?;
+ }
+
let user = mapper::map_user(&user);
Ok(Json(user))
}
@@ -88,19 +95,14 @@ async fn get_users(
State(state): State<Arc<AppState>>,
Extension(identity): Extension<Identity>,
) -> Result<Json<Vec<UserInfo>>, CustomError> {
- let session = SendWrapper::new(Session::stateless(identity.user_id, identity.ip_address));
-
- let users = {
- let future = SendWrapper::new(state.shard.shard().get_users(&session));
- future.await
- }
- .error(|e: &IggyError| {
- format!(
- "{COMPONENT} (error: {e}) - failed to get users, user ID: {}",
- identity.user_id
- )
- })?;
+ state
+ .shard
+ .shard()
+ .permissioner
+ .borrow()
+ .get_users(identity.user_id)?;
+ let users = state.shard.shard().get_users();
let user_refs: Vec<&User> = users.iter().collect();
let users = mapper::map_users(&user_refs);
Ok(Json(users))
@@ -114,14 +116,17 @@ async fn create_user(
Json(command): Json<CreateUser>,
) -> Result<Json<UserInfoDetails>, CustomError> {
command.validate()?;
-
- let session = SendWrapper::new(Session::stateless(identity.user_id, identity.ip_address));
+ state
+ .shard
+ .shard()
+ .permissioner
+ .borrow()
+ .create_user(identity.user_id)?;
let user = state
.shard
.shard()
.create_user(
- &session,
&command.username,
&command.password,
command.status,
@@ -196,18 +201,17 @@ async fn update_user(
) -> Result<StatusCode, CustomError> {
command.user_id = Identifier::from_str_value(&user_id)?;
command.validate()?;
-
- let session = Session::stateless(identity.user_id, identity.ip_address);
+ state
+ .shard
+ .shard()
+ .permissioner
+ .borrow()
+ .update_user(identity.user_id)?;
state
.shard
.shard()
- .update_user(
- &session,
- &command.user_id,
- command.username.clone(),
- command.status,
- )
+ .update_user(&command.user_id, command.username.clone(), command.status)
.error(|e: &IggyError| {
format!("{COMPONENT} (error: {e}) - failed to update user, user ID: {user_id}")
})?;
@@ -261,12 +265,25 @@ async fn update_permissions(
) -> Result<StatusCode, CustomError> {
command.user_id = Identifier::from_str_value(&user_id)?;
command.validate()?;
+ state
+ .shard
+ .shard()
+ .permissioner
+ .borrow()
+ .update_permissions(identity.user_id)?;
+
+ // Check if target user is root - cannot change root user permissions
+ let target_user = state.shard.shard().get_user(&command.user_id)?;
+ if target_user.is_root() {
+ return Err(CustomError::from(IggyError::CannotChangePermissions(
+ target_user.id,
+ )));
+ }
- let session = Session::stateless(identity.user_id, identity.ip_address);
state
.shard
.shard()
- .update_permissions(&session, &command.user_id, command.permissions.clone())
+ .update_permissions(&command.user_id, command.permissions.clone())
.error(|e: &IggyError| {
format!("{COMPONENT} (error: {e}) - failed to update permissions, user ID: {user_id}")
})?;
@@ -318,12 +335,21 @@ async fn change_password(
command.user_id = Identifier::from_str_value(&user_id)?;
command.validate()?;
- let session = Session::stateless(identity.user_id, identity.ip_address);
+ // Check if user is changing someone else's password
+ let target_user = state.shard.shard().get_user(&command.user_id)?;
+ if target_user.id != identity.user_id {
+ state
+ .shard
+ .shard()
+ .permissioner
+ .borrow()
+ .change_password(identity.user_id)?;
+ }
+
state
.shard
.shard()
.change_password(
- &session,
&command.user_id,
&command.current_password,
&command.new_password,
@@ -377,13 +403,17 @@ async fn delete_user(
Path(user_id): Path<String>,
) -> Result<StatusCode, CustomError> {
let identifier_user_id = Identifier::from_str_value(&user_id)?;
-
- let session = Session::stateless(identity.user_id, identity.ip_address);
+ state
+ .shard
+ .shard()
+ .permissioner
+ .borrow()
+ .delete_user(identity.user_id)?;
state
.shard
.shard()
- .delete_user(&session, &identifier_user_id)
+ .delete_user(&identifier_user_id)
.error(|e: &IggyError| {
format!("{COMPONENT} (error: {e}) - failed to delete user with ID: {user_id}")
})?;
diff --git a/core/server/src/shard/handlers.rs b/core/server/src/shard/handlers.rs
index 087b7a54d..30ead4d62 100644
--- a/core/server/src/shard/handlers.rs
+++ b/core/server/src/shard/handlers.rs
@@ -26,7 +26,7 @@ use crate::{
message::{ShardMessage, ShardRequest, ShardRequestPayload},
},
},
- streaming::{session::Session, traits::MainOps},
+ streaming::traits::MainOps,
tcp::{
connection_handler::{ConnectionAction, handle_connection, handle_error},
tcp_listener::cleanup_connection,
@@ -108,18 +108,13 @@ async fn handle_request(
.await?;
Ok(ShardResponse::DeleteSegments)
}
- ShardRequestPayload::CreateStream { user_id, name } => {
+ ShardRequestPayload::CreateStream { user_id: _, name } => {
assert_eq!(shard.id, 0, "CreateStream should only be handled by shard0");
- let session = Session::stateless(
- user_id,
- std::net::SocketAddr::new(std::net::IpAddr::V4(std::net::Ipv4Addr::LOCALHOST), 0),
- );
-
// Acquire stream lock to serialize filesystem operations
let _stream_guard = shard.fs_locks.stream_lock.lock().await;
- let stream = shard.create_stream(&session, name.clone()).await?;
+ let stream = shard.create_stream(name.clone()).await?;
let created_stream_id = stream.id();
let event = ShardEvent::CreatedStream {
@@ -132,7 +127,7 @@ async fn handle_request(
Ok(ShardResponse::CreateStreamResponse(stream))
}
ShardRequestPayload::CreateTopic {
- user_id,
+ user_id: _,
stream_id,
name,
partitions_count,
@@ -143,17 +138,11 @@ async fn handle_request(
} => {
assert_eq!(shard.id, 0, "CreateTopic should only be handled by shard0");
- let session = Session::stateless(
- user_id,
- std::net::SocketAddr::new(std::net::IpAddr::V4(std::net::Ipv4Addr::LOCALHOST), 0),
- );
-
// Acquire topic lock to serialize filesystem operations
let _topic_guard = shard.fs_locks.topic_lock.lock().await;
let topic = shard
.create_topic(
- &session,
&stream_id,
name.clone(),
message_expiry,
@@ -171,7 +160,6 @@ async fn handle_request(
shard.broadcast_event_to_all_shards(event).await?;
let partitions = shard
.create_partitions(
- &session,
&stream_id,
&Identifier::numeric(topic_id as u32).unwrap(),
partitions_count,
@@ -188,7 +176,7 @@ async fn handle_request(
Ok(ShardResponse::CreateTopicResponse(topic))
}
ShardRequestPayload::UpdateTopic {
- user_id,
+ user_id: _,
stream_id,
topic_id,
name,
@@ -199,13 +187,7 @@ async fn handle_request(
} => {
assert_eq!(shard.id, 0, "UpdateTopic should only be handled by shard0");
- let session = Session::stateless(
- user_id,
- std::net::SocketAddr::new(std::net::IpAddr::V4(std::net::Ipv4Addr::LOCALHOST), 0),
- );
-
shard.update_topic(
- &session,
&stream_id,
&topic_id,
name.clone(),
@@ -229,19 +211,14 @@ async fn handle_request(
Ok(ShardResponse::UpdateTopicResponse)
}
ShardRequestPayload::DeleteTopic {
- user_id,
+ user_id: _,
stream_id,
topic_id,
} => {
assert_eq!(shard.id, 0, "DeleteTopic should only be handled by shard0");
- let session = Session::stateless(
- user_id,
- std::net::SocketAddr::new(std::net::IpAddr::V4(std::net::Ipv4Addr::LOCALHOST), 0),
- );
-
let _topic_guard = shard.fs_locks.topic_lock.lock().await;
- let topic = shard.delete_topic(&session, &stream_id, &topic_id).await?;
+ let topic = shard.delete_topic(&stream_id, &topic_id).await?;
let topic_id_num = topic.root().id();
let event = ShardEvent::DeletedTopic {
@@ -254,7 +231,7 @@ async fn handle_request(
Ok(ShardResponse::DeleteTopicResponse(topic))
}
ShardRequestPayload::CreateUser {
- user_id,
+ user_id: _,
username,
password,
status,
@@ -262,13 +239,8 @@ async fn handle_request(
} => {
assert_eq!(shard.id, 0, "CreateUser should only be handled by shard0");
- let session = Session::stateless(
- user_id,
- std::net::SocketAddr::new(std::net::IpAddr::V4(std::net::Ipv4Addr::LOCALHOST), 0),
- );
let _user_guard = shard.fs_locks.user_lock.lock().await;
- let user =
- shard.create_user(&session, &username, &password, status, permissions.clone())?;
+ let user = shard.create_user(&username, &password, status, permissions.clone())?;
let created_user_id = user.id;
@@ -288,30 +260,25 @@ async fn handle_request(
Ok(ShardResponse::GetStatsResponse(stats))
}
ShardRequestPayload::DeleteUser {
- session_user_id,
+ session_user_id: _,
user_id,
} => {
- assert_eq!(shard.id, 0, "CreateUser should only be handled by shard0");
+ assert_eq!(shard.id, 0, "DeleteUser should only be handled by shard0");
- let session = Session::stateless(
- session_user_id,
- std::net::SocketAddr::new(std::net::IpAddr::V4(std::net::Ipv4Addr::LOCALHOST), 0),
- );
let _user_guard = shard.fs_locks.user_lock.lock().await;
- let user = shard.delete_user(&session, &user_id)?;
+ let user = shard.delete_user(&user_id)?;
let event = ShardEvent::DeletedUser { user_id };
shard.broadcast_event_to_all_shards(event).await?;
Ok(ShardResponse::DeletedUser(user))
}
- ShardRequestPayload::DeleteStream { user_id, stream_id } => {
+ ShardRequestPayload::DeleteStream {
+ user_id: _,
+ stream_id,
+ } => {
assert_eq!(shard.id, 0, "DeleteStream should only be handled by shard0");
- let session = Session::stateless(
- user_id,
- std::net::SocketAddr::new(std::net::IpAddr::V4(std::net::Ipv4Addr::LOCALHOST), 0),
- );
let _stream_guard = shard.fs_locks.stream_lock.lock().await;
- let stream = shard.delete_stream(&session, &stream_id).await?;
+ let stream = shard.delete_stream(&stream_id).await?;
let event = ShardEvent::DeletedStream {
id: stream.id(),
stream_id,
@@ -571,7 +538,7 @@ pub async fn handle_event(shard: &Rc<IggyShard>, event: ShardEvent) -> Result<()
cg,
} => {
let cg_id = cg.id();
- let id = shard.create_consumer_group_bypass_auth(&stream_id, &topic_id, cg);
+ let id = shard.insert_consumer_group(&stream_id, &topic_id, cg);
assert_eq!(id, cg_id);
Ok(())
}
@@ -581,7 +548,7 @@ pub async fn handle_event(shard: &Rc<IggyShard>, event: ShardEvent) -> Result<()
topic_id,
group_id,
} => {
- let cg = shard.delete_consumer_group_bypass_auth(&stream_id, &topic_id, &group_id);
+ let cg = shard.delete_consumer_group(&stream_id, &topic_id, &group_id)?;
assert_eq!(cg.id(), id);
Ok(())
diff --git a/core/server/src/shard/mod.rs b/core/server/src/shard/mod.rs
index 8939e87b0..93fd5eaac 100644
--- a/core/server/src/shard/mod.rs
+++ b/core/server/src/shard/mod.rs
@@ -38,7 +38,7 @@ use crate::{
state::file::FileState,
streaming::{
clients::client_manager::ClientManager, diagnostics::metrics::Metrics, session::Session,
- users::permissioner::Permissioner, utils::ptr::EternalPtr,
+ utils::ptr::EternalPtr,
},
versioning::SemanticVersion,
};
@@ -46,6 +46,7 @@ use builder::IggyShardBuilder;
use dashmap::DashMap;
use iggy_common::sharding::{IggyNamespace, PartitionLocation};
use iggy_common::{EncryptorKind, Identifier, IggyError};
+use metadata::permissioner::Permissioner;
use std::{
cell::{Cell, RefCell},
net::SocketAddr,
@@ -251,9 +252,10 @@ impl IggyShard {
async fn load_users(&self) -> Result<(), IggyError> {
let users_list = self.users.values();
let users_count = users_list.len();
- self.permissioner
- .borrow_mut()
- .init(&users_list.iter().collect::<Vec<_>>());
+ let mut permissioner = self.permissioner.borrow_mut();
+ for user in &users_list {
+ permissioner.init_permissions_for_user(user.id, user.permissions.clone());
+ }
self.metrics.increment_users(users_count as u32);
info!("Initialized {} user(s).", users_count);
Ok(())
diff --git a/core/server/src/shard/system/clients.rs b/core/server/src/shard/system/clients.rs
index fd01862a6..d3a4d5a67 100644
--- a/core/server/src/shard/system/clients.rs
+++ b/core/server/src/shard/system/clients.rs
@@ -16,12 +16,9 @@
* under the License.
*/
-use super::COMPONENT;
use crate::shard::IggyShard;
use crate::streaming::clients::client_manager::Client;
use crate::streaming::session::Session;
-use err_trail::ErrContext;
-use iggy_common::IggyError;
use iggy_common::{Identifier, TransportProtocol};
use std::net::SocketAddr;
use tracing::{error, info};
@@ -68,35 +65,11 @@ impl IggyShard {
self.client_manager.delete_client(client_id);
}
- pub fn get_client(
- &self,
- session: &Session,
- client_id: u32,
- ) -> Result<Option<Client>, IggyError> {
- self.permissioner
- .borrow()
- .get_client(session.get_user_id())
- .error(|e: &IggyError| {
- format!(
- "{COMPONENT} (error: {e}) - permission denied to get client with ID: {client_id} by user ID: {}",
- session.get_user_id()
- )
- })?;
-
- Ok(self.client_manager.try_get_client(client_id))
+ pub fn get_client(&self, client_id: u32) -> Option<Client> {
+ self.client_manager.try_get_client(client_id)
}
- pub fn get_clients(&self, session: &Session) -> Result<Vec<Client>, IggyError> {
- self.permissioner
- .borrow()
- .get_clients(session.get_user_id())
- .error(|e: &IggyError| {
- format!(
- "{COMPONENT} (error: {e}) - failed to get clients by user ID {}",
- session.get_user_id()
- )
- })?;
-
- Ok(self.client_manager.get_clients())
+ pub fn get_clients(&self) -> Vec<Client> {
+ self.client_manager.get_clients()
}
}
diff --git a/core/server/src/shard/system/cluster.rs b/core/server/src/shard/system/cluster.rs
index 3ccb599b1..b9e086569 100644
--- a/core/server/src/shard/system/cluster.rs
+++ b/core/server/src/shard/system/cluster.rs
@@ -17,17 +17,14 @@
*/
use crate::shard::IggyShard;
-use crate::streaming::session::Session;
use crate::streaming::utils::address::{extract_ip, extract_port};
use iggy_common::{
- ClusterMetadata, ClusterNode, ClusterNodeRole, ClusterNodeStatus, IggyError, TransportEndpoints,
+ ClusterMetadata, ClusterNode, ClusterNodeRole, ClusterNodeStatus, TransportEndpoints,
};
use tracing::trace;
impl IggyShard {
- pub fn get_cluster_metadata(&self, session: &Session) -> Result<ClusterMetadata, IggyError> {
- trace!("Getting cluster metadata for session: {session}");
-
+ pub fn get_cluster_metadata(&self) -> ClusterMetadata {
let mut nodes = Vec::new();
// Determine cluster name and current node name based on whether clustering is enabled
@@ -105,10 +102,10 @@ impl IggyShard {
}
}
- Ok(ClusterMetadata {
+ ClusterMetadata {
name: cluster_name,
nodes,
- })
+ }
}
/// Get actual bound ports from the shard's bound addresses
diff --git a/core/server/src/shard/system/consumer_groups.rs b/core/server/src/shard/system/consumer_groups.rs
index 7f04e5f6e..970959e2d 100644
--- a/core/server/src/shard/system/consumer_groups.rs
+++ b/core/server/src/shard/system/consumer_groups.rs
@@ -16,32 +16,31 @@
* under the License.
*/
-use super::COMPONENT;
use crate::shard::IggyShard;
use crate::slab::consumer_groups;
use crate::slab::traits_ext::EntityMarker;
use crate::slab::traits_ext::Insert;
use crate::streaming::partitions;
-use crate::streaming::session::Session;
-use crate::streaming::streams;
-use crate::streaming::topics;
+use crate::streaming::streams::helpers::get_stream_id;
use crate::streaming::topics::consumer_group;
use crate::streaming::topics::consumer_group::MEMBERS_CAPACITY;
+use crate::streaming::topics::helpers::get_topic_id;
+use crate::streaming::{streams, topics};
use arcshift::ArcShift;
use err_trail::ErrContext;
use iggy_common::Identifier;
use iggy_common::IggyError;
use slab::Slab;
+use super::COMPONENT;
+
impl IggyShard {
pub fn create_consumer_group(
&self,
- session: &Session,
stream_id: &Identifier,
topic_id: &Identifier,
name: String,
) -> Result<consumer_group::ConsumerGroup, IggyError> {
- self.ensure_topic_exists(stream_id, topic_id)?;
let exists = self
.streams
.with_topic_by_id(stream_id, topic_id, |(root, ..)| {
@@ -55,19 +54,6 @@ impl IggyShard {
));
}
- {
- let topic_id =
- self.streams
- .with_topic_by_id(stream_id, topic_id, topics::helpers::get_topic_id());
- let stream_id = self
- .streams
- .with_stream_by_id(stream_id, streams::helpers::get_stream_id());
- self.permissioner.borrow().create_consumer_group(
- session.get_user_id(),
- stream_id,
- topic_id,
- ).error(|e: &IggyError| format!("{COMPONENT} (error: {e}) - permission denied to create consumer group for user {} on stream ID: {}, topic ID: {}", session.get_user_id(), stream_id, topic_id))?;
- }
let cg = self.create_and_insert_consumer_group_mem(stream_id, topic_id, name);
Ok(cg)
}
@@ -83,12 +69,12 @@ impl IggyShard {
});
let members = ArcShift::new(Slab::with_capacity(MEMBERS_CAPACITY));
let mut cg = consumer_group::ConsumerGroup::new(name, members, partitions);
- let id = self.insert_consumer_group_mem(stream_id, topic_id, cg.clone());
+ let id = self.insert_consumer_group(stream_id, topic_id, cg.clone());
cg.update_id(id);
cg
}
- fn insert_consumer_group_mem(
+ pub fn insert_consumer_group(
&self,
stream_id: &Identifier,
topic_id: &Identifier,
@@ -98,49 +84,17 @@ impl IggyShard {
.with_consumer_groups_mut(stream_id, topic_id, |container| container.insert(cg))
}
- pub fn create_consumer_group_bypass_auth(
- &self,
- stream_id: &Identifier,
- topic_id: &Identifier,
- cg: consumer_group::ConsumerGroup,
- ) -> usize {
- self.insert_consumer_group_mem(stream_id, topic_id, cg)
- }
-
pub fn delete_consumer_group(
&self,
- session: &Session,
stream_id: &Identifier,
topic_id: &Identifier,
group_id: &Identifier,
) -> Result<consumer_group::ConsumerGroup, IggyError> {
self.ensure_consumer_group_exists(stream_id, topic_id, group_id)?;
- {
- let topic_id =
- self.streams
- .with_topic_by_id(stream_id, topic_id, topics::helpers::get_topic_id());
- let stream_id = self
- .streams
- .with_stream_by_id(stream_id, streams::helpers::get_stream_id());
- self.permissioner.borrow().delete_consumer_group(
- session.get_user_id(),
- stream_id,
- topic_id,
- ).error(|e: &IggyError| format!("{COMPONENT} (error: {e}) - permission denied to delete consumer group for user {} on stream ID: {}, topic ID: {}", session.get_user_id(), stream_id, topic_id))?;
- }
let cg = self.delete_consumer_group_base(stream_id, topic_id, group_id);
Ok(cg)
}
- pub fn delete_consumer_group_bypass_auth(
- &self,
- stream_id: &Identifier,
- topic_id: &Identifier,
- group_id: &Identifier,
- ) -> consumer_group::ConsumerGroup {
- self.delete_consumer_group_base(stream_id, topic_id, group_id)
- }
-
fn delete_consumer_group_base(
&self,
stream_id: &Identifier,
@@ -148,13 +102,11 @@ impl IggyShard {
group_id: &Identifier,
) -> consumer_group::ConsumerGroup {
// Get numeric IDs before deletion for ClientManager cleanup
- let stream_id_value = self
+ let stream_id_numeric = self.streams.with_stream_by_id(stream_id, get_stream_id());
+ let topic_id_numeric = self
.streams
- .with_stream_by_id(stream_id, streams::helpers::get_stream_id());
- let topic_id_value =
- self.streams
- .with_topic_by_id(stream_id, topic_id, topics::helpers::get_topic_id());
- let group_id_value = self.streams.with_consumer_group_by_id(
+ .with_topic_by_id(stream_id, topic_id, get_topic_id());
+ let group_id_numeric = self.streams.with_consumer_group_by_id(
stream_id,
topic_id,
group_id,
@@ -168,34 +120,23 @@ impl IggyShard {
);
// Clean up ClientManager state
- self.client_manager
- .delete_consumer_group(stream_id_value, topic_id_value, group_id_value);
+ self.client_manager.delete_consumer_group(
+ stream_id_numeric,
+ topic_id_numeric,
+ group_id_numeric,
+ );
cg
}
pub fn join_consumer_group(
&self,
- session: &Session,
+ client_id: u32,
stream_id: &Identifier,
topic_id: &Identifier,
group_id: &Identifier,
) -> Result<(), IggyError> {
self.ensure_consumer_group_exists(stream_id, topic_id, group_id)?;
- {
- let topic_id =
- self.streams
- .with_topic_by_id(stream_id, topic_id, topics::helpers::get_topic_id());
- let stream_id = self
- .streams
- .with_stream_by_id(stream_id, streams::helpers::get_stream_id());
- self.permissioner.borrow().join_consumer_group(
- session.get_user_id(),
- stream_id,
- topic_id,
- ).error(|e: &IggyError| format!("{COMPONENT} (error: {e}) - permission denied to join consumer group for user {} on stream ID: {}, topic ID: {}", session.get_user_id(), stream_id, topic_id))?;
- }
- let client_id = session.client_id;
self.streams.with_consumer_group_by_id_mut(
stream_id,
topic_id,
@@ -218,7 +159,7 @@ impl IggyShard {
);
self.client_manager.join_consumer_group(
- session.client_id,
+ client_id,
stream_id_value,
topic_id_value,
group_id_value,
@@ -226,7 +167,7 @@ impl IggyShard {
.error(|e: &IggyError| {
format!(
"{COMPONENT} (error: {e}) - failed to make client join consumer group for client ID: {}",
- session.client_id
+ client_id
)
})?;
Ok(())
@@ -234,26 +175,13 @@ impl IggyShard {
pub fn leave_consumer_group(
&self,
- session: &Session,
+ client_id: u32,
stream_id: &Identifier,
topic_id: &Identifier,
group_id: &Identifier,
) -> Result<(), IggyError> {
self.ensure_consumer_group_exists(stream_id, topic_id, group_id)?;
- {
- let topic_id =
- self.streams
- .with_topic_by_id(stream_id, topic_id, topics::helpers::get_topic_id());
- let stream_id = self
- .streams
- .with_stream_by_id(stream_id, streams::helpers::get_stream_id());
- self.permissioner.borrow().leave_consumer_group(
- session.get_user_id(),
- stream_id,
- topic_id,
- ).error(|e: &IggyError| format!("{COMPONENT} (error: {e}) - permission denied to leave consumer group for user {} on stream ID: {}, topic ID: {}", session.get_user_id(), stream_id, topic_id))?;
- }
- self.leave_consumer_group_base(stream_id, topic_id, group_id, session.client_id)
+ self.leave_consumer_group_base(stream_id, topic_id, group_id, client_id)
}
pub fn leave_consumer_group_base(
diff --git a/core/server/src/shard/system/consumer_offsets.rs b/core/server/src/shard/system/consumer_offsets.rs
index da85b75ea..c7eb5b1eb 100644
--- a/core/server/src/shard/system/consumer_offsets.rs
+++ b/core/server/src/shard/system/consumer_offsets.rs
@@ -23,7 +23,6 @@ use crate::{
streaming::{
partitions,
polling_consumer::{ConsumerGroupId, PollingConsumer},
- session::Session,
streams, topics,
},
};
@@ -33,7 +32,7 @@ use iggy_common::{Consumer, ConsumerOffsetInfo, Identifier, IggyError};
impl IggyShard {
pub async fn store_consumer_offset(
&self,
- session: &Session,
+ client_id: u32,
consumer: Consumer,
stream_id: &Identifier,
topic_id: &Identifier,
@@ -41,29 +40,11 @@ impl IggyShard {
offset: u64,
) -> Result<(PollingConsumer, usize), IggyError> {
self.ensure_topic_exists(stream_id, topic_id)?;
- {
- let topic_id =
- self.streams
- .with_topic_by_id(stream_id, topic_id, topics::helpers::get_topic_id());
- let stream_id = self
- .streams
- .with_stream_by_id(stream_id, streams::helpers::get_stream_id());
- self.permissioner.borrow().store_consumer_offset(
- session.get_user_id(),
- stream_id,
- topic_id
- ).error(|e: &IggyError| {
- format!(
- "{COMPONENT} (error: {e}) - permission denied to store consumer offset for user with ID: {}, consumer: {consumer} in topic with ID: {topic_id} and stream with ID: {stream_id}",
- session.get_user_id(),
- )
- })?;
- }
let Some((polling_consumer, partition_id)) = self.resolve_consumer_with_partition_id(
stream_id,
topic_id,
&consumer,
- session.client_id,
+ client_id,
partition_id,
false,
)?
@@ -86,36 +67,18 @@ impl IggyShard {
pub async fn get_consumer_offset(
&self,
- session: &Session,
+ client_id: u32,
consumer: Consumer,
stream_id: &Identifier,
topic_id: &Identifier,
partition_id: Option<u32>,
) -> Result<Option<ConsumerOffsetInfo>, IggyError> {
self.ensure_topic_exists(stream_id, topic_id)?;
- {
- let topic_id =
- self.streams
- .with_topic_by_id(stream_id, topic_id, topics::helpers::get_topic_id());
- let stream_id = self
- .streams
- .with_stream_by_id(stream_id, streams::helpers::get_stream_id());
- self.permissioner.borrow().get_consumer_offset(
- session.get_user_id(),
- stream_id,
- topic_id
- ).error(|e: &IggyError| {
- format!(
- "{COMPONENT} (error: {e}) - permission denied to get consumer offset for user with ID: {}, consumer: {consumer} in topic with ID: {topic_id} and stream with ID: {stream_id}",
- session.get_user_id()
- )
- })?;
- }
let Some((polling_consumer, partition_id)) = self.resolve_consumer_with_partition_id(
stream_id,
topic_id,
&consumer,
- session.client_id,
+ client_id,
partition_id,
false,
)?
@@ -145,36 +108,18 @@ impl IggyShard {
pub async fn delete_consumer_offset(
&self,
- session: &Session,
+ client_id: u32,
consumer: Consumer,
stream_id: &Identifier,
topic_id: &Identifier,
partition_id: Option<u32>,
) -> Result<(PollingConsumer, usize), IggyError> {
self.ensure_topic_exists(stream_id, topic_id)?;
- {
- let topic_id =
- self.streams
- .with_topic_by_id(stream_id, topic_id, topics::helpers::get_topic_id());
- let stream_id = self
- .streams
- .with_stream_by_id(stream_id, streams::helpers::get_stream_id());
- self.permissioner.borrow().delete_consumer_offset(
- session.get_user_id(),
- stream_id,
- topic_id
- ).error(|e: &IggyError| {
- format!(
- "{COMPONENT} (error: {e}) - permission denied to delete consumer offset for user with ID: {}, consumer: {consumer} in topic with ID: {topic_id} and stream with ID: {stream_id}",
- session.get_user_id(),
- )
- })?;
- }
let Some((polling_consumer, partition_id)) = self.resolve_consumer_with_partition_id(
stream_id,
topic_id,
&consumer,
- session.client_id,
+ client_id,
partition_id,
false,
)?
diff --git a/core/server/src/shard/system/partitions.rs b/core/server/src/shard/system/partitions.rs
index 7607fd804..db37b1df9 100644
--- a/core/server/src/shard/system/partitions.rs
+++ b/core/server/src/shard/system/partitions.rs
@@ -16,7 +16,6 @@
* under the License.
*/
-use super::COMPONENT;
use crate::shard::IggyShard;
use crate::shard::calculate_shard_assignment;
use crate::slab::traits_ext::EntityMarker;
@@ -27,43 +26,16 @@ use crate::streaming::partitions::storage::create_partition_file_hierarchy;
use crate::streaming::partitions::storage::delete_partitions_from_disk;
use crate::streaming::segments::Segment;
use crate::streaming::segments::storage::create_segment_storage;
-use crate::streaming::session::Session;
use crate::streaming::streams;
use crate::streaming::topics;
-use err_trail::ErrContext;
use iggy_common::Identifier;
use iggy_common::IggyError;
use iggy_common::sharding::{IggyNamespace, LocalIdx, PartitionLocation, ShardId};
use tracing::info;
impl IggyShard {
- fn validate_partition_permissions(
- &self,
- session: &Session,
- stream_id: usize,
- topic_id: usize,
- operation: &str,
- ) -> Result<(), IggyError> {
- let permissioner = self.permissioner.borrow();
- let result = match operation {
- "create" => permissioner.create_partitions(session.get_user_id(), stream_id, topic_id),
- "delete" => permissioner.delete_partitions(session.get_user_id(), stream_id, topic_id),
- _ => return Err(IggyError::InvalidCommand),
- };
-
- result.error(|e: &IggyError| {
- format!(
- "{COMPONENT} (error: {e}) - permission denied to {operation} partitions for user {} on stream ID: {}, topic ID: {}",
- session.get_user_id(),
- stream_id,
- topic_id
- )
- })
- }
-
pub async fn create_partitions(
&self,
- session: &Session,
stream_id: &Identifier,
topic_id: &Identifier,
partitions_count: u32,
@@ -76,13 +48,6 @@ impl IggyShard {
self.streams
.with_topic_by_id(stream_id, topic_id, topics::helpers::get_topic_id());
- // Claude garbage, rework this.
- self.validate_partition_permissions(
- session,
- numeric_stream_id,
- numeric_topic_id,
- "create",
- )?;
let parent_stats =
self.streams
.with_topic_by_id(stream_id, topic_id, topics::helpers::get_stats());
@@ -222,7 +187,6 @@ impl IggyShard {
pub async fn delete_partitions(
&self,
- session: &Session,
stream_id: &Identifier,
topic_id: &Identifier,
partitions_count: u32,
@@ -236,13 +200,6 @@ impl IggyShard {
self.streams
.with_topic_by_id(stream_id, topic_id, topics::helpers::get_topic_id());
- self.validate_partition_permissions(
- session,
- numeric_stream_id,
- numeric_topic_id,
- "delete",
- )?;
-
let partitions = self.delete_partitions_base(stream_id, topic_id, partitions_count);
let parent = partitions
.first()
diff --git a/core/server/src/shard/system/personal_access_tokens.rs b/core/server/src/shard/system/personal_access_tokens.rs
index 6966193e4..c36e9b141 100644
--- a/core/server/src/shard/system/personal_access_tokens.rs
+++ b/core/server/src/shard/system/personal_access_tokens.rs
@@ -30,9 +30,8 @@ use tracing::{error, info};
impl IggyShard {
pub fn get_personal_access_tokens(
&self,
- session: &Session,
+ user_id: u32,
) -> Result<Vec<PersonalAccessToken>, IggyError> {
- let user_id = session.get_user_id();
let user = self.get_user(&user_id.try_into()?).error(|e: &IggyError| {
format!("{COMPONENT} (error: {e}) - failed to get user with id: {user_id}")
})?;
@@ -52,11 +51,10 @@ impl IggyShard {
pub fn create_personal_access_token(
&self,
- session: &Session,
+ user_id: u32,
name: &str,
expiry: IggyExpiry,
) -> Result<(PersonalAccessToken, String), IggyError> {
- let user_id = session.get_user_id();
let identifier = user_id.try_into()?;
{
let user = self.get_user(&identifier).error(|e: &IggyError| {
@@ -124,12 +122,7 @@ impl IggyShard {
Ok(())
}
- pub fn delete_personal_access_token(
- &self,
- session: &Session,
- name: &str,
- ) -> Result<(), IggyError> {
- let user_id = session.get_user_id();
+ pub fn delete_personal_access_token(&self, user_id: u32, name: &str) -> Result<(), IggyError> {
self.delete_personal_access_token_base(user_id, name)
}
diff --git a/core/server/src/shard/system/snapshot/mod.rs b/core/server/src/shard/system/snapshot/mod.rs
index d157da8ab..7047292e5 100644
--- a/core/server/src/shard/system/snapshot/mod.rs
+++ b/core/server/src/shard/system/snapshot/mod.rs
@@ -20,7 +20,6 @@ mod procdump;
use crate::configs::system::SystemConfig;
use crate::shard::IggyShard;
-use crate::streaming::session::Session;
use async_zip::base::write::ZipFileWriter;
use async_zip::{Compression, ZipEntryBuilder};
use compio::fs::OpenOptions;
@@ -43,7 +42,6 @@ use std::process::Command;
impl IggyShard {
pub async fn get_snapshot(
&self,
- _session: &Session,
compression: SnapshotCompression,
snapshot_types: &Vec<SystemSnapshotType>,
) -> Result<Snapshot, IggyError> {
diff --git a/core/server/src/shard/system/streams.rs b/core/server/src/shard/system/streams.rs
index f64131c76..eafaa2af9 100644
--- a/core/server/src/shard/system/streams.rs
+++ b/core/server/src/shard/system/streams.rs
@@ -16,24 +16,14 @@
* under the License.
*/
-use super::COMPONENT;
use crate::shard::IggyShard;
use crate::slab::traits_ext::{DeleteCell, EntityMarker, InsertCell};
-use crate::streaming::session::Session;
use crate::streaming::streams::storage::{create_stream_file_hierarchy, delete_stream_from_disk};
use crate::streaming::streams::{self, stream};
-use err_trail::ErrContext;
use iggy_common::{Identifier, IggyError};
impl IggyShard {
- pub async fn create_stream(
- &self,
- session: &Session,
- name: String,
- ) -> Result<stream::Stream, IggyError> {
- self.permissioner
- .borrow()
- .create_stream(session.get_user_id())?;
+ pub async fn create_stream(&self, name: String) -> Result<stream::Stream, IggyError> {
let exists = self
.streams
.exists(&Identifier::from_str_value(&name).unwrap());
@@ -56,27 +46,8 @@ impl IggyShard {
Ok(())
}
- pub fn update_stream(
- &self,
- session: &Session,
- stream_id: &Identifier,
- name: String,
- ) -> Result<(), IggyError> {
+ pub fn update_stream(&self, stream_id: &Identifier, name: String) -> Result<(), IggyError> {
self.ensure_stream_exists(stream_id)?;
- let id = self
- .streams
- .with_stream_by_id(stream_id, streams::helpers::get_stream_id());
-
- self.permissioner
- .borrow()
- .update_stream(session.get_user_id(), id)
- .error(|e: &IggyError| {
- format!(
- "{COMPONENT} (error: {e}) - failed to update stream, user ID: {}, stream ID: {}",
- session.get_user_id(),
- stream_id
- )
- })?;
self.update_stream_base(stream_id, name)?;
Ok(())
}
@@ -122,31 +93,14 @@ impl IggyShard {
stream
}
- pub async fn delete_stream(
- &self,
- session: &Session,
- id: &Identifier,
- ) -> Result<stream::Stream, IggyError> {
+ pub async fn delete_stream(&self, id: &Identifier) -> Result<stream::Stream, IggyError> {
self.ensure_stream_exists(id)?;
- let stream_id = self
- .streams
- .with_stream_by_id(id, streams::helpers::get_stream_id());
- self.permissioner
- .borrow()
- .delete_stream(session.get_user_id(), stream_id)
- .error(|e: &IggyError| {
- format!(
- "{COMPONENT} (error: {e}) - permission denied to delete stream for user {}, stream ID: {}",
- session.get_user_id(),
- stream_id,
- )
- })?;
let mut stream = self.delete_stream_base(id);
- let stream_id_usize = stream.id();
+ let stream_id_numeric = stream.id();
// Clean up consumer groups from ClientManager for this stream
self.client_manager
- .delete_consumer_groups_for_stream(stream_id_usize);
+ .delete_consumer_groups_for_stream(stream_id_numeric);
// Remove all entries from shards_table for this stream (all topics and partitions)
let namespaces_to_remove: Vec<_> = self
@@ -154,7 +108,7 @@ impl IggyShard {
.iter()
.filter_map(|entry| {
let (ns, _) = entry.pair();
- if ns.stream_id() == stream_id_usize {
+ if ns.stream_id() == stream_id_numeric {
Some(*ns)
} else {
None
@@ -170,27 +124,8 @@ impl IggyShard {
Ok(stream)
}
- pub async fn purge_stream(
- &self,
- session: &Session,
- stream_id: &Identifier,
- ) -> Result<(), IggyError> {
+ pub async fn purge_stream(&self, stream_id: &Identifier) -> Result<(), IggyError> {
self.ensure_stream_exists(stream_id)?;
- {
- let get_stream_id = crate::streaming::streams::helpers::get_stream_id();
- let stream_id = self.streams.with_stream_by_id(stream_id, get_stream_id);
- self.permissioner
- .borrow()
- .purge_stream(session.get_user_id(), stream_id)
- .error(|e: &IggyError| {
- format!(
- "{COMPONENT} (error: {e}) - permission denied to purge stream for user {}, stream ID: {}",
- session.get_user_id(),
- stream_id,
- )
- })?;
- }
-
self.purge_stream_base(stream_id).await
}
diff --git a/core/server/src/shard/system/topics.rs b/core/server/src/shard/system/topics.rs
index 295a7d409..2a2fbb9e9 100644
--- a/core/server/src/shard/system/topics.rs
+++ b/core/server/src/shard/system/topics.rs
@@ -16,14 +16,11 @@
* under the License.
*/
-use super::COMPONENT;
use crate::shard::IggyShard;
use crate::slab::traits_ext::{EntityComponentSystem, EntityMarker, InsertCell, IntoComponents};
-use crate::streaming::session::Session;
use crate::streaming::topics::storage::{create_topic_file_hierarchy, delete_topic_from_disk};
use crate::streaming::topics::topic::{self};
use crate::streaming::{partitions, streams, topics};
-use err_trail::ErrContext;
use iggy_common::{CompressionAlgorithm, Identifier, IggyError, IggyExpiry, MaxTopicSize};
use std::str::FromStr;
use tracing::info;
@@ -32,7 +29,6 @@ impl IggyShard {
#[allow(clippy::too_many_arguments)]
pub async fn create_topic(
&self,
- session: &Session,
stream_id: &Identifier,
name: String,
message_expiry: IggyExpiry,
@@ -42,17 +38,6 @@ impl IggyShard {
) -> Result<topic::Topic, IggyError> {
self.ensure_stream_exists(stream_id)?;
let numeric_stream_id = self.streams.get_index(stream_id);
- {
- self.permissioner
- .borrow()
- .create_topic(session.get_user_id(), numeric_stream_id)
- .error(|e: &IggyError| {
- format!(
- "{COMPONENT} (error: {e}) - permission denied to create topic with name: {name} in stream with ID: {stream_id} for user with ID: {}",
- session.get_user_id(),
- )
- })?;
- }
let exists = self.streams.with_topics(
stream_id,
topics::helpers::exists(&Identifier::from_str(&name).unwrap()),
@@ -93,7 +78,6 @@ impl IggyShard {
#[allow(clippy::too_many_arguments)]
pub fn update_topic(
&self,
- session: &Session,
stream_id: &Identifier,
topic_id: &Identifier,
name: String,
@@ -103,26 +87,6 @@ impl IggyShard {
replication_factor: Option<u8>,
) -> Result<(), IggyError> {
self.ensure_topic_exists(stream_id, topic_id)?;
- {
- let topic_id_val =
- self.streams
- .with_topic_by_id(stream_id, topic_id, topics::helpers::get_topic_id());
- let stream_id_val = self
- .streams
- .with_stream_by_id(stream_id, streams::helpers::get_stream_id());
- self.permissioner.borrow().update_topic(
- session.get_user_id(),
- stream_id_val,
- topic_id_val
- ).error(|e: &IggyError| {
- format!(
- "{COMPONENT} (error: {e}) - permission denied to update topic for user with id: {}, stream ID: {}, topic ID: {}",
- session.get_user_id(),
- stream_id_val,
- topic_id_val,
- )
- })?;
- }
let exists = self.streams.with_topics(
stream_id,
@@ -196,26 +160,13 @@ impl IggyShard {
pub async fn delete_topic(
&self,
- session: &Session,
stream_id: &Identifier,
topic_id: &Identifier,
) -> Result<topic::Topic, IggyError> {
self.ensure_topic_exists(stream_id, topic_id)?;
- let numeric_topic_id =
- self.streams
- .with_topic_by_id(stream_id, topic_id, topics::helpers::get_topic_id());
let numeric_stream_id = self
.streams
.with_stream_by_id(stream_id, streams::helpers::get_stream_id());
- self.permissioner
- .borrow()
- .delete_topic(session.get_user_id(), numeric_stream_id, numeric_topic_id)
- .error(|e: &IggyError| {
- format!(
- "{COMPONENT} (error: {e}) - permission denied to delete topic with ID: {topic_id} in stream with ID: {stream_id} for user with ID: {}",
- session.get_user_id(),
- )
- })?;
let mut topic = self.delete_topic_base(stream_id, topic_id);
let topic_id_numeric = topic.id();
@@ -267,29 +218,10 @@ impl IggyShard {
pub async fn purge_topic(
&self,
- session: &Session,
stream_id: &Identifier,
topic_id: &Identifier,
) -> Result<(), IggyError> {
self.ensure_topic_exists(stream_id, topic_id)?;
- {
- let topic_id =
- self.streams
- .with_topic_by_id(stream_id, topic_id, topics::helpers::get_topic_id());
- let stream_id = self
- .streams
- .with_stream_by_id(stream_id, streams::helpers::get_stream_id());
- self.permissioner.borrow().purge_topic(
- session.get_user_id(),
- stream_id,
- topic_id
- ).error(|e: &IggyError| {
- format!(
- "{COMPONENT} (error: {e}) - permission denied to purge topic with ID: {topic_id} in stream with ID: {stream_id} for user with ID: {}",
- session.get_user_id(),
- )
- })?;
- }
let (consumer_offset_paths, consumer_group_offset_paths) = self.streams.with_partitions(
stream_id,
diff --git a/core/server/src/shard/system/users.rs b/core/server/src/shard/system/users.rs
index 31266d496..100c850b1 100644
--- a/core/server/src/shard/system/users.rs
+++ b/core/server/src/shard/system/users.rs
@@ -31,25 +31,8 @@ use tracing::{error, warn};
const MAX_USERS: usize = u32::MAX as usize;
impl IggyShard {
- pub fn find_user(
- &self,
- session: &Session,
- user_id: &Identifier,
- ) -> Result<Option<User>, IggyError> {
- let Some(user) = self.try_get_user(user_id)? else {
- return Ok(None);
- };
-
- let session_user_id = session.get_user_id();
- if user.id != session_user_id {
- self.permissioner.borrow().get_user(session_user_id).error(|e: &IggyError| {
- format!(
- "{COMPONENT} (error: {e}) - permission denied to get user with ID: {user_id} for current user with ID: {session_user_id}"
- )
- })?;
- }
-
- Ok(Some(user))
+ pub fn find_user(&self, user_id: &Identifier) -> Result<Option<User>, IggyError> {
+ self.try_get_user(user_id)
}
pub fn get_user(&self, user_id: &Identifier) -> Result<User, IggyError> {
@@ -61,37 +44,17 @@ impl IggyShard {
self.users.get_by_identifier(user_id)
}
- pub async fn get_users(&self, session: &Session) -> Result<Vec<User>, IggyError> {
- self.permissioner
- .borrow()
- .get_users(session.get_user_id())
- .error(|e: &IggyError| {
- format!(
- "{COMPONENT} (error: {e}) - permission denied to get users for user with id: {}",
- session.get_user_id()
- )
- })?;
- Ok(self.users.values())
+ pub fn get_users(&self) -> Vec<User> {
+ self.users.values()
}
pub fn create_user(
&self,
- session: &Session,
username: &str,
password: &str,
status: UserStatus,
permissions: Option<Permissions>,
) -> Result<User, IggyError> {
- self.permissioner
- .borrow()
- .create_user(session.get_user_id())
- .error(|e: &IggyError| {
- format!(
- "{COMPONENT} (error: {e}) - permission denied to create user for user with id: {}",
- session.get_user_id()
- )
- })?;
-
if self.users.username_exists(username) {
error!("User: {username} already exists.");
return Err(IggyError::UserAlreadyExists);
@@ -144,17 +107,7 @@ impl IggyShard {
Ok(user_id)
}
- pub fn delete_user(&self, session: &Session, user_id: &Identifier) -> Result<User, IggyError> {
- self.permissioner
- .borrow()
- .delete_user(session.get_user_id())
- .error(|e: &IggyError| {
- format!(
- "{COMPONENT} (error: {e}) - permission denied to delete user for user with id: {}",
- session.get_user_id()
- )
- })?;
-
+ pub fn delete_user(&self, user_id: &Identifier) -> Result<User, IggyError> {
self.delete_user_base(user_id)
}
@@ -195,21 +148,10 @@ impl IggyShard {
pub fn update_user(
&self,
- session: &Session,
user_id: &Identifier,
username: Option<String>,
status: Option<UserStatus>,
) -> Result<User, IggyError> {
- self.permissioner
- .borrow()
- .update_user(session.get_user_id())
- .error(|e: &IggyError| {
- format!(
- "{COMPONENT} (error: {e}) - permission denied to update user for user with id: {}",
- session.get_user_id()
- )
- })?;
-
self.update_user_base(user_id, username, status)
}
@@ -257,28 +199,9 @@ impl IggyShard {
pub fn update_permissions(
&self,
- session: &Session,
user_id: &Identifier,
permissions: Option<Permissions>,
) -> Result<(), IggyError> {
- {
- self.permissioner
- .borrow()
- .update_permissions(session.get_user_id())
- .error(|e: &IggyError| {
- format!(
- "{COMPONENT} (error: {e}) - permission denied to update permissions for user with id: {}", session.get_user_id()
- )
- })?;
- let user: User = self.get_user(user_id).error(|e: &IggyError| {
- format!("{COMPONENT} (error: {e}) - failed to get user with id: {user_id}")
- })?;
- if user.is_root() {
- error!("Cannot change the root user permissions.");
- return Err(IggyError::CannotChangePermissions(user.id));
- }
- }
-
self.update_permissions_base(user_id, permissions)
}
@@ -314,23 +237,10 @@ impl IggyShard {
pub fn change_password(
&self,
- session: &Session,
user_id: &Identifier,
current_password: &str,
new_password: &str,
) -> Result<(), IggyError> {
- {
- let user = self.get_user(user_id).error(|e: &IggyError| {
- format!("{COMPONENT} (error: {e}) - failed to get user with id: {user_id}")
- })?;
- let session_user_id = session.get_user_id();
- if user.id != session_user_id {
- self.permissioner
- .borrow()
- .change_password(session_user_id)?;
- }
- }
-
self.change_password_base(user_id, current_password, new_password)
}
diff --git a/core/server/src/shard/system/utils.rs b/core/server/src/shard/system/utils.rs
index e7ad661b1..2d901d2f6 100644
--- a/core/server/src/shard/system/utils.rs
+++ b/core/server/src/shard/system/utils.rs
@@ -19,57 +19,109 @@ use iggy_common::{Consumer, ConsumerKind, Identifier, IggyError};
use crate::{
shard::IggyShard,
- streaming::{
- polling_consumer::PollingConsumer,
- topics::{self},
- },
+ streaming::{polling_consumer::PollingConsumer, streams, topics},
};
impl IggyShard {
- pub fn ensure_stream_exists(&self, stream_id: &Identifier) -> Result<(), IggyError> {
+ pub fn resolve_stream_id(&self, stream_id: &Identifier) -> Result<usize, IggyError> {
if !self.streams.exists(stream_id) {
return Err(IggyError::StreamIdNotFound(stream_id.clone()));
}
- Ok(())
+ let numeric_id = self
+ .streams
+ .with_stream_by_id(stream_id, streams::helpers::get_stream_id());
+ Ok(numeric_id)
}
- pub fn ensure_topic_exists(
+ pub fn resolve_topic_id(
&self,
stream_id: &Identifier,
topic_id: &Identifier,
- ) -> Result<(), IggyError> {
- self.ensure_stream_exists(stream_id)?;
- let exists = self
+ ) -> Result<(usize, usize), IggyError> {
+ let numeric_stream_id = self.resolve_stream_id(stream_id)?;
+ if !self
.streams
- .with_topics(stream_id, topics::helpers::exists(topic_id));
- if !exists {
+ .with_topics(stream_id, topics::helpers::exists(topic_id))
+ {
return Err(IggyError::TopicIdNotFound(
stream_id.clone(),
topic_id.clone(),
));
}
- Ok(())
+ let numeric_topic_id =
+ self.streams
+ .with_topic_by_id(stream_id, topic_id, topics::helpers::get_topic_id());
+ Ok((numeric_stream_id, numeric_topic_id))
}
- pub fn ensure_consumer_group_exists(
+ pub fn resolve_partition_id(
+ &self,
+ stream_id: &Identifier,
+ topic_id: &Identifier,
+ partition_id: usize,
+ ) -> Result<(usize, usize, usize), IggyError> {
+ let (numeric_stream_id, numeric_topic_id) = self.resolve_topic_id(stream_id, topic_id)?;
+ let partition_exists = self
+ .streams
+ .with_topic_by_id(stream_id, topic_id, |(root, ..)| {
+ root.partitions().exists(partition_id)
+ });
+ if !partition_exists {
+ return Err(IggyError::PartitionNotFound(
+ partition_id,
+ topic_id.clone(),
+ stream_id.clone(),
+ ));
+ }
+ Ok((numeric_stream_id, numeric_topic_id, partition_id))
+ }
+
+ pub fn resolve_consumer_group_id(
&self,
stream_id: &Identifier,
topic_id: &Identifier,
group_id: &Identifier,
- ) -> Result<(), IggyError> {
- self.ensure_stream_exists(stream_id)?;
- self.ensure_topic_exists(stream_id, topic_id)?;
- let exists = self.streams.with_topic_by_id(
- stream_id,
- topic_id,
- topics::helpers::cg_exists(group_id),
- );
- if !exists {
+ ) -> Result<(usize, usize, usize), IggyError> {
+ let (numeric_stream_id, numeric_topic_id) = self.resolve_topic_id(stream_id, topic_id)?;
+ if !self
+ .streams
+ .with_topic_by_id(stream_id, topic_id, topics::helpers::cg_exists(group_id))
+ {
return Err(IggyError::ConsumerGroupIdNotFound(
group_id.clone(),
topic_id.clone(),
));
}
+ let numeric_group_id = self.streams.with_consumer_group_by_id(
+ stream_id,
+ topic_id,
+ group_id,
+ topics::helpers::get_consumer_group_id(),
+ );
+ Ok((numeric_stream_id, numeric_topic_id, numeric_group_id))
+ }
+
+ pub fn ensure_stream_exists(&self, stream_id: &Identifier) -> Result<(), IggyError> {
+ self.resolve_stream_id(stream_id)?;
+ Ok(())
+ }
+
+ pub fn ensure_topic_exists(
+ &self,
+ stream_id: &Identifier,
+ topic_id: &Identifier,
+ ) -> Result<(), IggyError> {
+ self.resolve_topic_id(stream_id, topic_id)?;
+ Ok(())
+ }
+
+ pub fn ensure_consumer_group_exists(
+ &self,
+ stream_id: &Identifier,
+ topic_id: &Identifier,
+ group_id: &Identifier,
+ ) -> Result<(), IggyError> {
+ self.resolve_consumer_group_id(stream_id, topic_id, group_id)?;
Ok(())
}
@@ -79,7 +131,7 @@ impl IggyShard {
topic_id: &Identifier,
partitions_count: u32,
) -> Result<(), IggyError> {
- self.ensure_topic_exists(stream_id, topic_id)?;
+ self.resolve_topic_id(stream_id, topic_id)?;
let actual_partitions_count =
self.streams
.with_partitions(stream_id, topic_id, |partitions| partitions.len());
@@ -97,21 +149,7 @@ impl IggyShard {
topic_id: &Identifier,
partition_id: usize,
) -> Result<(), IggyError> {
- self.ensure_topic_exists(stream_id, topic_id)?;
- let partition_exists = self
- .streams
- .with_topic_by_id(stream_id, topic_id, |(root, ..)| {
- root.partitions().exists(partition_id)
- });
-
- if !partition_exists {
- return Err(IggyError::PartitionNotFound(
- partition_id,
- topic_id.clone(),
- stream_id.clone(),
- ));
- }
-
+ self.resolve_partition_id(stream_id, topic_id, partition_id)?;
Ok(())
}
diff --git a/core/server/src/streaming/users/mod.rs b/core/server/src/streaming/users/mod.rs
index 233d6c753..fa88e29ab 100644
--- a/core/server/src/streaming/users/mod.rs
+++ b/core/server/src/streaming/users/mod.rs
@@ -16,6 +16,4 @@
* under the License.
*/
-pub mod permissioner;
-pub mod permissioner_rules;
pub mod user;
diff --git a/core/server/src/streaming/users/permissioner.rs b/core/server/src/streaming/users/permissioner.rs
deleted file mode 100644
index 807dce465..000000000
--- a/core/server/src/streaming/users/permissioner.rs
+++ /dev/null
@@ -1,101 +0,0 @@
-/* Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-use crate::streaming::users::user::User;
-use ahash::{AHashMap, AHashSet};
-use iggy_common::UserId;
-use iggy_common::{GlobalPermissions, Permissions, StreamPermissions};
-
-#[derive(Debug, Default)]
-pub struct Permissioner {
- pub(super) users_permissions: AHashMap<UserId, GlobalPermissions>,
- pub(super) users_streams_permissions: AHashMap<(UserId, usize), StreamPermissions>,
- pub(super) users_that_can_poll_messages_from_all_streams: AHashSet<UserId>,
- pub(super) users_that_can_send_messages_to_all_streams: AHashSet<UserId>,
- pub(super) users_that_can_poll_messages_from_specific_streams: AHashSet<(UserId, usize)>,
- pub(super) users_that_can_send_messages_to_specific_streams: AHashSet<(UserId, usize)>,
-}
-
-impl Permissioner {
- pub fn init(&mut self, users: &[&User]) {
- for user in users {
- self.init_permissions_for_user(user.id, user.permissions.clone());
- }
- }
-
- pub fn init_permissions_for_user(&mut self, user_id: UserId, permissions: Option<Permissions>) {
- if permissions.is_none() {
- return;
- }
-
- let permissions = permissions.unwrap();
- if permissions.global.poll_messages {
- self.users_that_can_poll_messages_from_all_streams
- .insert(user_id);
- }
-
- if permissions.global.send_messages {
- self.users_that_can_send_messages_to_all_streams
- .insert(user_id);
- }
-
- self.users_permissions.insert(user_id, permissions.global);
- if permissions.streams.is_none() {
- return;
- }
-
- let streams = permissions.streams.unwrap();
- for (stream_id, stream) in streams {
- if stream.poll_messages {
- self.users_that_can_poll_messages_from_specific_streams
- .insert((user_id, stream_id));
- }
-
- if stream.send_messages {
- self.users_that_can_send_messages_to_specific_streams
- .insert((user_id, stream_id));
- }
-
- self.users_streams_permissions
- .insert((user_id, stream_id), stream);
- }
- }
-
- pub fn update_permissions_for_user(
- &mut self,
- user_id: UserId,
- permissions: Option<Permissions>,
- ) {
- self.delete_permissions_for_user(user_id);
- self.init_permissions_for_user(user_id, permissions);
- }
-
- pub fn delete_permissions_for_user(&mut self, user_id: UserId) {
- self.users_permissions.remove(&user_id);
- self.users_that_can_poll_messages_from_all_streams
- .remove(&user_id);
- self.users_that_can_send_messages_to_all_streams
- .remove(&user_id);
- self.users_streams_permissions
- .retain(|(id, _), _| *id != user_id);
- self.users_that_can_poll_messages_from_specific_streams
- .retain(|(id, _)| *id != user_id);
- self.users_that_can_send_messages_to_specific_streams
- .retain(|(id, _)| *id != user_id);
- }
-}
diff --git a/core/server/src/streaming/users/permissioner_rules/consumer_groups.rs b/core/server/src/streaming/users/permissioner_rules/consumer_groups.rs
deleted file mode 100644
index 230d2de0f..000000000
--- a/core/server/src/streaming/users/permissioner_rules/consumer_groups.rs
+++ /dev/null
@@ -1,76 +0,0 @@
-/* Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-use crate::streaming::users::permissioner::Permissioner;
-use iggy_common::IggyError;
-
-impl Permissioner {
- pub fn create_consumer_group(
- &self,
- user_id: u32,
- stream_id: usize,
- topic_id: usize,
- ) -> Result<(), IggyError> {
- self.get_topic(user_id, stream_id, topic_id)
- }
-
- pub fn delete_consumer_group(
- &self,
- user_id: u32,
- stream_id: usize,
- topic_id: usize,
- ) -> Result<(), IggyError> {
- self.get_topic(user_id, stream_id, topic_id)
- }
-
- pub fn get_consumer_group(
- &self,
- user_id: u32,
- stream_id: usize,
- topic_id: usize,
- ) -> Result<(), IggyError> {
- self.get_topic(user_id, stream_id, topic_id)
- }
-
- pub fn get_consumer_groups(
- &self,
- user_id: u32,
- stream_id: usize,
- topic_id: usize,
- ) -> Result<(), IggyError> {
- self.get_topic(user_id, stream_id, topic_id)
- }
-
- pub fn join_consumer_group(
- &self,
- user_id: u32,
- stream_id: usize,
- topic_id: usize,
- ) -> Result<(), IggyError> {
- self.get_topic(user_id, stream_id, topic_id)
- }
-
- pub fn leave_consumer_group(
- &self,
- user_id: u32,
- stream_id: usize,
- topic_id: usize,
- ) -> Result<(), IggyError> {
- self.get_topic(user_id, stream_id, topic_id)
- }
-}
diff --git a/core/server/src/streaming/users/permissioner_rules/consumer_offsets.rs b/core/server/src/streaming/users/permissioner_rules/consumer_offsets.rs
deleted file mode 100644
index dece246af..000000000
--- a/core/server/src/streaming/users/permissioner_rules/consumer_offsets.rs
+++ /dev/null
@@ -1,49 +0,0 @@
-/* Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-use crate::streaming::users::permissioner::Permissioner;
-use iggy_common::IggyError;
-
-impl Permissioner {
- pub fn get_consumer_offset(
- &self,
- user_id: u32,
- stream_id: usize,
- topic_id: usize,
- ) -> Result<(), IggyError> {
- self.poll_messages(user_id, stream_id, topic_id)
- }
-
- pub fn store_consumer_offset(
- &self,
- user_id: u32,
- stream_id: usize,
- topic_id: usize,
- ) -> Result<(), IggyError> {
- self.poll_messages(user_id, stream_id, topic_id)
- }
-
- pub fn delete_consumer_offset(
- &self,
- user_id: u32,
- stream_id: usize,
- topic_id: usize,
- ) -> Result<(), IggyError> {
- self.poll_messages(user_id, stream_id, topic_id)
- }
-}
diff --git a/core/server/src/streaming/users/permissioner_rules/messages.rs b/core/server/src/streaming/users/permissioner_rules/messages.rs
deleted file mode 100644
index 7e5a34af2..000000000
--- a/core/server/src/streaming/users/permissioner_rules/messages.rs
+++ /dev/null
@@ -1,135 +0,0 @@
-/* Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-use crate::streaming::users::permissioner::Permissioner;
-use iggy_common::IggyError;
-
-impl Permissioner {
- pub fn poll_messages(
- &self,
- user_id: u32,
- stream_id: usize,
- topic_id: usize,
- ) -> Result<(), IggyError> {
- if self
- .users_that_can_poll_messages_from_all_streams
- .contains(&user_id)
- {
- return Ok(());
- }
-
- if self
- .users_that_can_poll_messages_from_specific_streams
- .contains(&(user_id, stream_id))
- {
- return Ok(());
- }
-
- let stream_permissions = self.users_streams_permissions.get(&(user_id, stream_id));
- if stream_permissions.is_none() {
- return Err(IggyError::Unauthorized);
- }
-
- let stream_permissions = stream_permissions.unwrap();
- if stream_permissions.read_stream {
- return Ok(());
- }
-
- if stream_permissions.manage_topics {
- return Ok(());
- }
-
- if stream_permissions.read_topics {
- return Ok(());
- }
-
- if stream_permissions.poll_messages {
- return Ok(());
- }
-
- if stream_permissions.topics.is_none() {
- return Err(IggyError::Unauthorized);
- }
-
- let topic_permissions = stream_permissions.topics.as_ref().unwrap();
- if let Some(topic_permissions) = topic_permissions.get(&topic_id) {
- return match topic_permissions.poll_messages
- | topic_permissions.read_topic
- | topic_permissions.manage_topic
- {
- true => Ok(()),
- false => Err(IggyError::Unauthorized),
- };
- }
-
- Err(IggyError::Unauthorized)
- }
-
- pub fn append_messages(
- &self,
- user_id: u32,
- stream_id: usize,
- topic_id: usize,
- ) -> Result<(), IggyError> {
- if self
- .users_that_can_send_messages_to_all_streams
- .contains(&user_id)
- {
- return Ok(());
- }
-
- if self
- .users_that_can_send_messages_to_specific_streams
- .contains(&(user_id, stream_id))
- {
- return Ok(());
- }
-
- let stream_permissions = self.users_streams_permissions.get(&(user_id, stream_id));
- if stream_permissions.is_none() {
- return Err(IggyError::Unauthorized);
- }
-
- let stream_permissions = stream_permissions.unwrap();
- if stream_permissions.manage_stream {
- return Ok(());
- }
-
- if stream_permissions.manage_topics {
- return Ok(());
- }
-
- if stream_permissions.send_messages {
- return Ok(());
- }
-
- if stream_permissions.topics.is_none() {
- return Err(IggyError::Unauthorized);
- }
-
- let topic_permissions = stream_permissions.topics.as_ref().unwrap();
- if let Some(topic_permissions) = topic_permissions.get(&topic_id) {
- return match topic_permissions.send_messages | topic_permissions.manage_topic {
- true => Ok(()),
- false => Err(IggyError::Unauthorized),
- };
- }
-
- Err(IggyError::Unauthorized)
- }
-}
diff --git a/core/server/src/streaming/users/permissioner_rules/mod.rs b/core/server/src/streaming/users/permissioner_rules/mod.rs
deleted file mode 100644
index 0d08a7a1c..000000000
--- a/core/server/src/streaming/users/permissioner_rules/mod.rs
+++ /dev/null
@@ -1,27 +0,0 @@
-/* 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.
- */
-
-mod consumer_groups;
-pub mod consumer_offsets;
-mod messages;
-mod partitions;
-mod segments;
-mod streams;
-mod system;
-mod topics;
-mod users;
diff --git a/core/server/src/streaming/users/permissioner_rules/partitions.rs b/core/server/src/streaming/users/permissioner_rules/partitions.rs
deleted file mode 100644
index a82cd6368..000000000
--- a/core/server/src/streaming/users/permissioner_rules/partitions.rs
+++ /dev/null
@@ -1,40 +0,0 @@
-/* Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-use crate::streaming::users::permissioner::Permissioner;
-use iggy_common::IggyError;
-
-impl Permissioner {
- pub fn create_partitions(
- &self,
- user_id: u32,
- stream_id: usize,
- topic_id: usize,
- ) -> Result<(), IggyError> {
- self.update_topic(user_id, stream_id, topic_id)
- }
-
- pub fn delete_partitions(
- &self,
- user_id: u32,
- stream_id: usize,
- topic_id: usize,
- ) -> Result<(), IggyError> {
- self.update_topic(user_id, stream_id, topic_id)
- }
-}
diff --git a/core/server/src/streaming/users/permissioner_rules/segments.rs b/core/server/src/streaming/users/permissioner_rules/segments.rs
deleted file mode 100644
index d37350a45..000000000
--- a/core/server/src/streaming/users/permissioner_rules/segments.rs
+++ /dev/null
@@ -1,30 +0,0 @@
-/* Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-use crate::streaming::users::permissioner::Permissioner;
-use iggy_common::IggyError;
-
-impl Permissioner {
- pub fn delete_segments(
- &self,
- user_id: u32,
- stream_id: usize,
- topic_id: usize,
- ) -> Result<(), IggyError> {
- self.update_topic(user_id, stream_id, topic_id)
- }
-}
diff --git a/core/server/src/streaming/users/permissioner_rules/streams.rs b/core/server/src/streaming/users/permissioner_rules/streams.rs
deleted file mode 100644
index 0e7545ef3..000000000
--- a/core/server/src/streaming/users/permissioner_rules/streams.rs
+++ /dev/null
@@ -1,86 +0,0 @@
-/* Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-use crate::streaming::users::permissioner::Permissioner;
-use iggy_common::IggyError;
-
-impl Permissioner {
- pub fn get_stream(&self, user_id: u32, stream_id: usize) -> Result<(), IggyError> {
- if let Some(global_permissions) = self.users_permissions.get(&user_id)
- && (global_permissions.manage_streams || global_permissions.read_streams)
- {
- return Ok(());
- }
-
- if let Some(stream_permissions) = self.users_streams_permissions.get(&(user_id, stream_id))
- && (stream_permissions.manage_stream || stream_permissions.read_stream)
- {
- return Ok(());
- }
-
- Err(IggyError::Unauthorized)
- }
-
- pub fn get_streams(&self, user_id: u32) -> Result<(), IggyError> {
- if let Some(global_permissions) = self.users_permissions.get(&user_id)
- && (global_permissions.manage_streams || global_permissions.read_streams)
- {
- return Ok(());
- }
-
- Err(IggyError::Unauthorized)
- }
-
- pub fn create_stream(&self, user_id: u32) -> Result<(), IggyError> {
- if let Some(global_permissions) = self.users_permissions.get(&user_id)
- && global_permissions.manage_streams
- {
- return Ok(());
- }
-
- Err(IggyError::Unauthorized)
- }
-
- pub fn update_stream(&self, user_id: u32, stream_id: usize) -> Result<(), IggyError> {
- self.manage_stream(user_id, stream_id)
- }
-
- pub fn delete_stream(&self, user_id: u32, stream_id: usize) -> Result<(), IggyError> {
- self.manage_stream(user_id, stream_id)
- }
-
- pub fn purge_stream(&self, user_id: u32, stream_id: usize) -> Result<(), IggyError> {
- self.manage_stream(user_id, stream_id)
- }
-
- fn manage_stream(&self, user_id: u32, stream_id: usize) -> Result<(), IggyError> {
- if let Some(global_permissions) = self.users_permissions.get(&user_id)
- && global_permissions.manage_streams
- {
- return Ok(());
- }
-
- if let Some(stream_permissions) = self.users_streams_permissions.get(&(user_id, stream_id))
- && stream_permissions.manage_stream
- {
- return Ok(());
- }
-
- Err(IggyError::Unauthorized)
- }
-}
diff --git a/core/server/src/streaming/users/permissioner_rules/system.rs b/core/server/src/streaming/users/permissioner_rules/system.rs
deleted file mode 100644
index 86f1e0bd0..000000000
--- a/core/server/src/streaming/users/permissioner_rules/system.rs
+++ /dev/null
@@ -1,44 +0,0 @@
-/* Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-use crate::streaming::users::permissioner::Permissioner;
-use iggy_common::IggyError;
-
-impl Permissioner {
- pub fn get_stats(&self, user_id: u32) -> Result<(), IggyError> {
- self.get_server_info(user_id)
- }
-
- pub fn get_clients(&self, user_id: u32) -> Result<(), IggyError> {
- self.get_server_info(user_id)
- }
-
- pub fn get_client(&self, user_id: u32) -> Result<(), IggyError> {
- self.get_server_info(user_id)
- }
-
- fn get_server_info(&self, user_id: u32) -> Result<(), IggyError> {
- if let Some(global_permissions) = self.users_permissions.get(&user_id)
- && (global_permissions.manage_servers || global_permissions.read_servers)
- {
- return Ok(());
- }
-
- Err(IggyError::Unauthorized)
- }
-}
diff --git a/core/server/src/streaming/users/permissioner_rules/topics.rs b/core/server/src/streaming/users/permissioner_rules/topics.rs
deleted file mode 100644
index 28a85f058..000000000
--- a/core/server/src/streaming/users/permissioner_rules/topics.rs
+++ /dev/null
@@ -1,153 +0,0 @@
-/* Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-use crate::streaming::users::permissioner::Permissioner;
-use iggy_common::IggyError;
-
-impl Permissioner {
- pub fn get_topic(
- &self,
- user_id: u32,
- stream_id: usize,
- topic_id: usize,
- ) -> Result<(), IggyError> {
- if let Some(global_permissions) = self.users_permissions.get(&user_id)
- && (global_permissions.read_streams
- || global_permissions.manage_streams
- || global_permissions.manage_topics
- || global_permissions.read_topics)
- {
- return Ok(());
- }
-
- if let Some(stream_permissions) = self.users_streams_permissions.get(&(user_id, stream_id))
- {
- if stream_permissions.manage_topics || stream_permissions.read_topics {
- return Ok(());
- }
-
- if let Some(topic_permissions) =
- stream_permissions.topics.as_ref().unwrap().get(&topic_id)
- && (topic_permissions.manage_topic || topic_permissions.read_topic)
- {
- return Ok(());
- }
- }
-
- Err(IggyError::Unauthorized)
- }
-
- pub fn get_topics(&self, user_id: u32, stream_id: usize) -> Result<(), IggyError> {
- if let Some(global_permissions) = self.users_permissions.get(&user_id)
- && (global_permissions.read_streams
- || global_permissions.manage_streams
- || global_permissions.manage_topics
- || global_permissions.read_topics)
- {
- return Ok(());
- }
-
- if let Some(stream_permissions) = self.users_streams_permissions.get(&(user_id, stream_id))
- {
- if stream_permissions.manage_topics || stream_permissions.read_topics {
- return Ok(());
- }
-
- if let Some(topic_permissions) =
- stream_permissions.topics.as_ref().unwrap().get(&stream_id)
- && (topic_permissions.manage_topic || topic_permissions.read_topic)
- {
- return Ok(());
- }
- }
-
- Err(IggyError::Unauthorized)
- }
-
- pub fn create_topic(&self, user_id: u32, stream_id: usize) -> Result<(), IggyError> {
- if let Some(global_permissions) = self.users_permissions.get(&user_id)
- && (global_permissions.manage_streams || global_permissions.manage_topics)
- {
- return Ok(());
- }
-
- if let Some(stream_permissions) = self.users_streams_permissions.get(&(user_id, stream_id))
- && stream_permissions.manage_topics
- {
- return Ok(());
- }
-
- Err(IggyError::Unauthorized)
- }
-
- pub fn update_topic(
- &self,
- user_id: u32,
- stream_id: usize,
- topic_id: usize,
- ) -> Result<(), IggyError> {
- self.manage_topic(user_id, stream_id, topic_id)
- }
-
- pub fn delete_topic(
- &self,
- user_id: u32,
- stream_id: usize,
- topic_id: usize,
- ) -> Result<(), IggyError> {
- self.manage_topic(user_id, stream_id, topic_id)
- }
-
- pub fn purge_topic(
- &self,
- user_id: u32,
- stream_id: usize,
- topic_id: usize,
- ) -> Result<(), IggyError> {
- self.manage_topic(user_id, stream_id, topic_id)
- }
-
- fn manage_topic(
- &self,
- user_id: u32,
- stream_id: usize,
- topic_id: usize,
- ) -> Result<(), IggyError> {
- if let Some(global_permissions) = self.users_permissions.get(&user_id)
- && (global_permissions.manage_streams || global_permissions.manage_topics)
- {
- return Ok(());
- }
-
- if let Some(stream_permissions) = self.users_streams_permissions.get(&(user_id, stream_id))
- {
- if stream_permissions.manage_topics {
- return Ok(());
- }
-
- if let Some(topic_permissions) =
- stream_permissions.topics.as_ref().unwrap().get(&topic_id)
- && topic_permissions.manage_topic
- {
- return Ok(());
- }
- }
-
- Err(IggyError::Unauthorized)
- }
-}
diff --git a/core/server/src/streaming/users/permissioner_rules/users.rs b/core/server/src/streaming/users/permissioner_rules/users.rs
deleted file mode 100644
index 8c01d7c7f..000000000
--- a/core/server/src/streaming/users/permissioner_rules/users.rs
+++ /dev/null
@@ -1,70 +0,0 @@
-/* Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-use crate::streaming::users::permissioner::Permissioner;
-use iggy_common::IggyError;
-
-impl Permissioner {
- pub fn get_user(&self, user_id: u32) -> Result<(), IggyError> {
- self.read_users(user_id)
- }
-
- pub fn get_users(&self, user_id: u32) -> Result<(), IggyError> {
- self.read_users(user_id)
- }
-
- pub fn create_user(&self, user_id: u32) -> Result<(), IggyError> {
- self.manager_users(user_id)
- }
-
- pub fn delete_user(&self, user_id: u32) -> Result<(), IggyError> {
- self.manager_users(user_id)
- }
-
- pub fn update_user(&self, user_id: u32) -> Result<(), IggyError> {
- self.manager_users(user_id)
- }
-
- pub fn update_permissions(&self, user_id: u32) -> Result<(), IggyError> {
- self.manager_users(user_id)
- }
-
- pub fn change_password(&self, user_id: u32) -> Result<(), IggyError> {
- self.manager_users(user_id)
- }
-
- fn manager_users(&self, user_id: u32) -> Result<(), IggyError> {
- if let Some(global_permissions) = self.users_permissions.get(&user_id)
- && global_permissions.manage_users
- {
- return Ok(());
- }
-
- Err(IggyError::Unauthorized)
- }
-
- fn read_users(&self, user_id: u32) -> Result<(), IggyError> {
- if let Some(global_permissions) = self.users_permissions.get(&user_id)
- && (global_permissions.manage_users || global_permissions.read_users)
- {
- return Ok(());
- }
-
- Err(IggyError::Unauthorized)
- }
-}