From d1973a239efbc9ba5bb848cd412fc36c637852ab Mon Sep 17 00:00:00 2001 From: Serhii Mamontov Date: Sun, 11 Oct 2026 23:45:29 +0300 Subject: [PATCH 1/2] feat(access): add categories permissions to grant token Add `ChannelsCategoryPermission` and `UuidsCategoryPermission`, the `categories` option for `grant_token`, and parsing of category permissions in `parse_token`. Parsed token `resources` and `patterns` are optional now. fix(subscribe): fix `DataStream` and dispatcher deadlocks Fix `DataStream` deadlock between `push_data` and `poll_next` which took `queue` and `waker` locks in opposite order, and `EventDispatcher` deadlock between stream creation and `invalidate`. refactor(client): remove redundant `format!` references Remove redundant references in `format!` arguments of `PubNubClientInstance`, access and presence request builders. build(cargo): update manifest and dev dependencies Add `keywords`, enable `access` feature by default, update `wiremock` and `cucumber`, and register the `integration` test target. docs(readme): update PubNub Rust SDK header Update `README.md` PubNub Rust SDK header. test(access): cover category grants and token parsing Add unit and integration tests for categories-only grants, all permission kinds, token parsing and malformed tokens. test(subscribe): cover stream lock order and wake-ups Add tests for `DataStream` and `EventDispatcher` lock ordering, listener wake-ups and delivery of dispatched batches. --- Cargo.toml | 12 +- README.md | 13 +- src/core/data_stream.rs | 267 ++++++++++++++++- src/dx/access/builders/grant_token.rs | 36 ++- src/dx/access/builders/revoke.rs | 2 +- src/dx/access/mod.rs | 30 ++ src/dx/access/payloads.rs | 78 ++++- src/dx/access/permissions.rs | 167 ++++++++++- src/dx/parse_token.rs | 108 ++++++- .../presence/builders/get_presence_state.rs | 2 +- src/dx/presence/builders/heartbeat.rs | 2 +- src/dx/presence/builders/here_now.rs | 2 +- src/dx/presence/builders/leave.rs | 2 +- .../presence/builders/set_presence_state.rs | 2 +- src/dx/presence/builders/where_now.rs | 2 +- src/dx/pubnub_client.rs | 8 +- src/dx/subscribe/event_dispatcher.rs | 278 +++++++++++++++++- tests/access/access_steps.rs | 6 + tests/integration/access_integration.rs | 202 ++++++++----- 19 files changed, 1084 insertions(+), 135 deletions(-) diff --git a/Cargo.toml b/Cargo.toml index 4c2895fd..a9a34921 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -9,6 +9,7 @@ repository = "https://github.com/pubnub/rust" documentation = "https://docs.rs/pubnub/latest/pubnub" homepage = "https://www.pubnub.com" categories = ["api-bindings", "asynchronous", "network-programming", "wasm"] +keywords = ["pubnub", "realtime", "pubsub", "messaging", "long-poll"] build = "build.rs" [features] @@ -17,7 +18,7 @@ build = "build.rs" full = ["publish", "subscribe", "presence", "access", "serde", "reqwest", "crypto", "parse_token", "blocking", "std", "tokio"] # Enables all default features -default = ["publish", "subscribe", "serde", "reqwest", "std", "blocking", "tokio"] +default = ["publish", "subscribe", "serde", "reqwest", "std", "blocking", "tokio", "access"] # [PubNub features] @@ -121,9 +122,9 @@ getrandom = { version = "0.3", features = ["wasm_js"] } [dev-dependencies] async-trait = "0.1" tokio = { version = "1", features = ["rt-multi-thread", "macros", "time"] } -wiremock = "0.5" +wiremock = "0.6.5" env_logger = "0.10" -cucumber = { version = "0.20.2", features = ["output-junit"] } +cucumber = { version = "0.23.0", features = ["output-junit"] } reqwest = { version = "0.12", features = ["json"] } test-case = "3.0" hashbrown = { version = "0.14.0", features = ["serde"] } @@ -140,6 +141,11 @@ name = "contract_test" harness = false required-features = ["contract_test"] +[[test]] +name = "integration" +path = "tests/integration/mod.rs" +required-features = ["full"] + [[example]] name = "crypto" required-features = ["default", "crypto"] diff --git a/README.md b/README.md index bf8613c2..1b4f9891 100644 --- a/README.md +++ b/README.md @@ -1,16 +1,7 @@ -# PubNub Rust SDK - -
- -![PubNub](https://raw.githubusercontent.com/pubnub/rust/master/logo.svg) +rust-header -![Tests](https://github.com/pubnub/rust/actions/workflows/run-tests.yml/badge.svg) -![Validations](https://github.com/pubnub/rust/actions/workflows/run-validations.yml/badge.svg) -[![License: MIT](https://img.shields.io/badge/License-MIT-yellow.svg)](https://github.com/pubnub/rust/LICENSE) -**Make your app come alive with real-time experiences!** - -
+# PubNub Rust SDK ## Overview diff --git a/src/core/data_stream.rs b/src/core/data_stream.rs index 7ec8d1a3..821b35d2 100644 --- a/src/core/data_stream.rs +++ b/src/core/data_stream.rs @@ -152,28 +152,36 @@ impl DataStream { return; } - let mut queue_data_slot = self.queue.write(); + { + let mut queue_data_slot = self.queue.write(); - // Dropping the earliest entry to prevent the queue from growing too large. - if queue_data_slot.len() == queue_data_slot.capacity() { - queue_data_slot.pop_front(); - } + // Dropping the earliest entry to prevent the queue from growing too large. + if queue_data_slot.len() == queue_data_slot.capacity() { + queue_data_slot.pop_front(); + } - queue_data_slot.push_back(data); + queue_data_slot.push_back(data); + } + // Wake the listener only after the queue lock has been released. + // Waking while holding it can deadlock with a concurrent `poll_next`. self.wake_stream(); } #[cfg(all(feature = "subscribe", feature = "std"))] pub(crate) fn invalidate(&self) { - let mut is_valid = self.is_valid.write(); - *is_valid = false; + *self.is_valid.write() = false; self.wake_stream(); } + /// Wake up the stream listener (if any). + /// + /// The waker is taken out of its slot first, so no stream lock is held + /// while `wake()` runs. #[cfg(all(feature = "subscribe", feature = "std"))] fn wake_stream(&self) { - if let Some(waker) = self.waker.write().take() { + let waker = self.waker.write().take(); + if let Some(waker) = waker { waker.wake(); } } @@ -187,13 +195,248 @@ impl Stream for DataStream { return Poll::Ready(None); } - let mut waker_slot = self.waker.write(); - *waker_slot = Some(ctx.waker().clone()); + // Register the waker before checking the stream state, so data pushed + // (or invalidation) after this point wakes the listener. Locks are + // taken one at a time to avoid lock-order deadlocks with producers. + *self.waker.write() = Some(ctx.waker().clone()); - if let Some(data) = self.queue.write().pop_front() { + if !*self.is_valid.read() { + return Poll::Ready(None); + } + + let data = self.queue.write().pop_front(); + if let Some(data) = data { Poll::Ready(Some(data)) } else { Poll::Pending } } } + +#[cfg(all(test, feature = "subscribe", feature = "std"))] +mod should { + use super::*; + use std::{ + sync::{ + atomic::{AtomicBool, Ordering}, + mpsc, + }, + task::Wake, + thread, + time::{Duration, Instant}, + }; + + /// How long lock-order tests wait for the expected lock state. + const LOCK_WAIT: Duration = Duration::from_secs(2); + + /// Waker which records whether it has been called. + #[derive(Default)] + struct FlagWaker { + woken: AtomicBool, + } + + impl Wake for FlagWaker { + fn wake(self: Arc) { + self.woken.store(true, Ordering::SeqCst); + } + } + + /// Waker which records whether stream locks were free when it has been + /// called. + struct LockCheckWaker { + stream: DataStream, + woken: AtomicBool, + locks_free: AtomicBool, + } + + impl LockCheckWaker { + fn new(stream: DataStream) -> Self { + Self { + stream, + woken: AtomicBool::new(false), + locks_free: AtomicBool::new(false), + } + } + } + + impl Wake for LockCheckWaker { + fn wake(self: Arc) { + let locks_free = self.stream.queue.try_write().is_some() + && self.stream.waker.try_write().is_some() + && self.stream.is_valid.try_write().is_some(); + self.locks_free.store(locks_free, Ordering::SeqCst); + self.woken.store(true, Ordering::SeqCst); + } + } + + /// Waker which unparks the thread that created it. + struct ThreadWaker(thread::Thread); + + impl Wake for ThreadWaker { + fn wake(self: Arc) { + self.0.unpark(); + } + } + + fn poll_once(stream: &mut DataStream, waker: &Waker) -> Poll> { + let mut ctx = Context::from_waker(waker); + Pin::new(stream).poll_next(&mut ctx) + } + + #[test] + fn release_queue_lock_before_waking_listener() { + let stream = DataStream::::with_queue_size(10); + let waker_guard = stream.waker.write(); + + let producer_stream = stream.clone(); + let producer = thread::spawn(move || producer_stream.push_data(1)); + + let deadline = Instant::now() + LOCK_WAIT; + let mut queue_released = false; + while Instant::now() < deadline { + if stream + .queue + .try_read() + .is_some_and(|queue| queue.len() == 1) + { + queue_released = true; + break; + } + thread::yield_now(); + } + + drop(waker_guard); + producer.join().expect("producer thread panicked"); + + assert!( + queue_released, + "push_data() held the queue lock while waiting for the waker lock" + ); + } + + #[test] + fn register_waker_without_holding_queue_lock() { + let stream = DataStream::::with_queue_size(10); + let queue_guard = stream.queue.write(); + + let mut consumer_stream = stream.clone(); + let consumer = thread::spawn(move || { + let waker = Waker::from(Arc::new(FlagWaker::default())); + poll_once(&mut consumer_stream, &waker) + }); + + let deadline = Instant::now() + LOCK_WAIT; + let mut waker_released = false; + while Instant::now() < deadline { + if stream.waker.try_read().is_some_and(|slot| slot.is_some()) { + waker_released = true; + break; + } + thread::yield_now(); + } + + drop(queue_guard); + let poll = consumer.join().expect("consumer thread panicked"); + + assert!( + waker_released, + "poll_next() held the waker lock while waiting for the queue lock" + ); + assert!(poll.is_pending()); + } + + #[test] + fn wake_listener_on_push_without_holding_stream_locks() { + let mut stream = DataStream::::with_queue_size(10); + let check = Arc::new(LockCheckWaker::new(stream.clone())); + let waker = Waker::from(check.clone()); + + assert!(poll_once(&mut stream, &waker).is_pending()); + stream.push_data(1); + + assert!(check.woken.load(Ordering::SeqCst)); + assert!( + check.locks_free.load(Ordering::SeqCst), + "push_data() woke the listener while holding a stream lock" + ); + } + + #[test] + fn wake_listener_on_invalidate_without_holding_stream_locks() { + let mut stream = DataStream::::with_queue_size(10); + let check = Arc::new(LockCheckWaker::new(stream.clone())); + let waker = Waker::from(check.clone()); + + assert!(poll_once(&mut stream, &waker).is_pending()); + stream.invalidate(); + + assert!(check.woken.load(Ordering::SeqCst)); + assert!( + check.locks_free.load(Ordering::SeqCst), + "invalidate() woke the listener while holding a stream lock" + ); + } + + #[test] + fn wake_pending_listener_when_data_pushed() { + let mut stream = DataStream::::with_queue_size(10); + let flag = Arc::new(FlagWaker::default()); + let waker = Waker::from(flag.clone()); + + assert!(poll_once(&mut stream, &waker).is_pending()); + stream.push_data(7); + + assert!(flag.woken.load(Ordering::SeqCst)); + assert_eq!(poll_once(&mut stream, &waker), Poll::Ready(Some(7))); + assert!(poll_once(&mut stream, &waker).is_pending()); + } + + #[test] + fn wake_pending_listener_and_end_stream_when_invalidated() { + let mut stream = DataStream::::with_queue_size(10); + let flag = Arc::new(FlagWaker::default()); + let waker = Waker::from(flag.clone()); + + assert!(poll_once(&mut stream, &waker).is_pending()); + stream.invalidate(); + + assert!(flag.woken.load(Ordering::SeqCst)); + assert_eq!(poll_once(&mut stream, &waker), Poll::Ready(None)); + } + + #[test] + fn deliver_all_items_in_order_while_producer_and_listener_race() { + const ROUNDS: usize = 200; + const ITEMS: usize = 100; + + for round in 0..ROUNDS { + let mut consumer_stream = DataStream::::with_queue_size(ITEMS); + let producer_stream = consumer_stream.clone(); + let (result_tx, result_rx) = mpsc::channel(); + + thread::spawn(move || { + let waker = Waker::from(Arc::new(ThreadWaker(thread::current()))); + let mut received = Vec::with_capacity(ITEMS); + while received.len() < ITEMS { + match poll_once(&mut consumer_stream, &waker) { + Poll::Ready(Some(item)) => received.push(item), + Poll::Ready(None) => break, + Poll::Pending => thread::park(), + } + } + let _ = result_tx.send(received); + }); + let producer = + thread::spawn(move || (0..ITEMS).for_each(|item| producer_stream.push_data(item))); + + let received = result_rx + .recv_timeout(Duration::from_secs(10)) + .unwrap_or_else(|_| { + panic!("round {round}: listener stuck (deadlock or lost wake-up)") + }); + producer.join().expect("producer thread panicked"); + + assert_eq!(received, (0..ITEMS).collect::>(), "round {round}"); + } + } +} diff --git a/src/dx/access/builders/grant_token.rs b/src/dx/access/builders/grant_token.rs index 0d2f3af7..d3cd8bd3 100644 --- a/src/dx/access/builders/grant_token.rs +++ b/src/dx/access/builders/grant_token.rs @@ -82,6 +82,14 @@ where default = "None" )] pub patterns: Option<&'pa [Box]>, + + /// List of permissions for enumerate-able resource categories. + #[builder( + field(vis = "pub(in crate::dx::access)"), + setter(strip_option), + default = "None" + )] + pub categories: Option<&'pa [Box]>, } /// The [`GrantTokenRequestWithSerializerBuilder`] is used to build grant access @@ -113,7 +121,7 @@ where let body = self.serializer.serialize(&payload).unwrap_or(vec![]); TransportRequest { - path: format!("/v3/pam/{}/grant", &config.subscribe_key), + path: format!("/v3/pam/{}/grant", config.subscribe_key), query_parameters: Default::default(), method: TransportMethod::Post, headers: [(CONTENT_TYPE.to_string(), APPLICATION_JSON.to_string())].into(), @@ -133,7 +141,31 @@ where /// Validator ensure that list of provided data is enough to build valid /// request instance. fn validate(&self) -> Result<(), String> { - builders::validate_configuration(&self.pubnub_client) + builders::validate_configuration(&self.pubnub_client).and_then(|_| { + let has_resources = self.resources.flatten().is_some_and(|r| !r.is_empty()); + let has_patterns = self.patterns.flatten().is_some_and(|p| !p.is_empty()); + let has_categories = self.categories.flatten().is_some_and(|c| !c.is_empty()); + + if let Some(categories) = self.categories.flatten() { + categories.iter().try_for_each(|cat| { + if !["channels", "uuids"].contains(&cat.id().as_str()) { + return Err(format!("Unexpected category type: {}", cat.id())); + } + + if *cat.value() != permissions::GET { + return Err(format!("Unexpected category permission: {}", *cat.value())); + } + + Ok(()) + })?; + } + + if !has_resources && !has_patterns && !has_categories { + Err("At least one of the following should be provided: resources, patterns, categories.".into()) + } else { + Ok(()) + } + }) } } diff --git a/src/dx/access/builders/revoke.rs b/src/dx/access/builders/revoke.rs index 87fd3b0a..9333d390 100644 --- a/src/dx/access/builders/revoke.rs +++ b/src/dx/access/builders/revoke.rs @@ -52,7 +52,7 @@ impl RevokeTokenRequest { TransportRequest { path: format!( "/v3/pam/{}/grant/{}", - &config.subscribe_key, + config.subscribe_key, url_encode(self.token.as_bytes()) ), method: TransportMethod::Delete, diff --git a/src/dx/access/mod.rs b/src/dx/access/mod.rs index 77e41cf6..8bc712a3 100644 --- a/src/dx/access/mod.rs +++ b/src/dx/access/mod.rs @@ -323,6 +323,36 @@ mod it_should { assert!(request.is_err()); } + #[test] + fn grant_token_with_categories_only() { + let categories: Vec> = vec![ + permissions::channels_category().get(), + permissions::uuids_category().get(), + ]; + let client = client(true, true, None, None, None); + let request = client.grant_token(10).categories(&categories).build(); + + assert!(request.is_ok()); + } + + #[test] + fn not_grant_token_without_resources_patterns_or_categories() { + let client = client(true, true, None, None, None); + let request = client.grant_token(10).build(); + + assert!(request.is_err()); + } + + #[test] + fn not_grant_token_for_category_without_permission() { + let categories: Vec> = + vec![permissions::channels_category()]; + let client = client(true, true, None, None, None); + let request = client.grant_token(10).categories(&categories).build(); + + assert!(request.is_err()); + } + #[tokio::test] async fn grant_token() { let permissions = permissions(); diff --git a/src/dx/access/payloads.rs b/src/dx/access/payloads.rs index 64e3a70a..49c27b95 100644 --- a/src/dx/access/payloads.rs +++ b/src/dx/access/payloads.rs @@ -36,6 +36,21 @@ pub struct GrantTokenResourcesPayload { pub uuids: Option>, } +/// Categories permissions payload. +/// +/// This type used by [`GrantTokenPayload`] to store permissions for specific +/// resource type. +#[cfg_attr(feature = "serde", derive(serde::Serialize))] +pub struct GrantTokenCategoriesPayload { + /// Category permissions for `channels` enumeration endpoint access. + #[cfg_attr(feature = "serde", serde(skip_serializing_if = "Option::is_none"))] + pub channels: Option, + + /// Category permissions for `uuids` enumeration endpoint access. + #[cfg_attr(feature = "serde", serde(skip_serializing_if = "Option::is_none"))] + pub uuids: Option, +} + /// Token permissions. /// /// This type used by [`GrantTokenPayload`] to store information about requested @@ -47,7 +62,8 @@ pub struct GrantTokenPermissionsPayload<'request> { #[cfg_attr(feature = "serde", serde(skip_serializing_if = "Option::is_none"))] pub uuid: &'request Option, - /// Extra metadata to be published with the request. Values must be scalar only. + /// Extra metadata to be published with the request. Values must be scalar + /// only. #[cfg_attr(feature = "serde", serde(skip_serializing_if = "Option::is_none"))] pub meta: &'request Option>, @@ -58,6 +74,10 @@ pub struct GrantTokenPermissionsPayload<'request> { /// List of permissions mapped to RegExp match expressions. #[cfg_attr(feature = "serde", serde(skip_serializing_if = "Option::is_none"))] pub patterns: Option, + + /// List of permissions for enumerate-able resource categories. + #[cfg_attr(feature = "serde", serde(skip_serializing_if = "Option::is_none"))] + pub categories: Option, } /// Payload for grant token operation. @@ -88,6 +108,7 @@ impl<'request> GrantTokenPayload<'request> { permissions: Some(GrantTokenPermissionsPayload { patterns: resource_permissions(&request.patterns), resources: resource_permissions(&request.resources), + categories: categories_permissions(&request.categories), uuid: &request.authorized_user_id, meta: &request.meta, }), @@ -113,6 +134,7 @@ fn resource_permissions( ResourceType::UserId => { uuids.insert(perm.id().to_string(), *perm.value()); } + _ => {} }); GrantTokenResourcesPayload { @@ -122,3 +144,57 @@ fn resource_permissions( } }) } + +/// Extract permissions for categories. +fn categories_permissions( + resources: &Option<&[Box]>, +) -> Option { + resources.map(|res| { + let mut channels_category_perm: u8 = 0; + let mut uuids_category_perm: u8 = 0; + res.iter().for_each(|perm| match perm.resource_type() { + ResourceType::ChannelsCategory => channels_category_perm = *perm.value(), + ResourceType::UuidsCategory => uuids_category_perm = *perm.value(), + _ => {} + }); + + GrantTokenCategoriesPayload { + channels: channels_category_perm + .gt(&0) + .then_some(channels_category_perm), + uuids: uuids_category_perm.gt(&0).then_some(uuids_category_perm), + } + }) +} + +#[cfg(test)] +mod should { + use super::*; + use crate::dx::access::permissions::{self, GET}; + + #[test] + fn build_categories_payload_with_both_categories() { + let perms: [Box; 2] = [ + permissions::channels_category().get(), + permissions::uuids_category().get(), + ]; + let payload = categories_permissions(&Some(&perms)).unwrap(); + assert_eq!(payload.channels, Some(GET)); + assert_eq!(payload.uuids, Some(GET)); + } + + #[test] + fn omit_category_without_permission() { + let perms: [Box; 1] = [permissions::channels_category().get()]; + let payload = categories_permissions(&Some(&perms)).unwrap(); + assert_eq!(payload.channels, Some(GET)); + assert_eq!(payload.uuids, None); + } + + #[test] + fn not_treat_category_as_channel_resource() { + let perms: [Box; 1] = [permissions::channels_category().get()]; + let payload = resource_permissions(&Some(&perms)).unwrap(); + assert!(payload.channels.is_none()); + } +} diff --git a/src/dx/access/permissions.rs b/src/dx/access/permissions.rs index eef68fd1..2686c7b6 100644 --- a/src/dx/access/permissions.rs +++ b/src/dx/access/permissions.rs @@ -1,25 +1,26 @@ //! Resource permissions module. //! //! This module contains: [`Permission`] struct and it's implementation of -//! [`ChannelPermission`], [`ChannelGroupPermission`] and [`UserIdPermission`] +//! [`ChannelPermission`], [`ChannelGroupPermission`], [`UserIdPermission`], +//! [`ChannelsCategoryPermission`], and [`UuidsCategoryPermission`] //! traits. use crate::lib::{alloc::boxed::Box, alloc::string::String}; /// Resource **read** permissions. -const READ: u8 = 0b0000_0001; +pub(super) const READ: u8 = 0b0000_0001; /// Resource **write** permissions. -const WRITE: u8 = 0b0000_0010; +pub(super) const WRITE: u8 = 0b0000_0010; /// Resource **manage** permissions. -const MANAGE: u8 = 0b0000_0100; +pub(super) const MANAGE: u8 = 0b0000_0100; /// Resource **delete** permissions. -const DELETE: u8 = 0b0000_1000; +pub(super) const DELETE: u8 = 0b0000_1000; /// Resource **get** permissions. -const GET: u8 = 0b0010_0000; +pub(super) const GET: u8 = 0b0010_0000; /// Resource **update** permissions. -const UPDATE: u8 = 0b0100_0000; +pub(super) const UPDATE: u8 = 0b0100_0000; /// Resource **join** permissions. -const JOIN: u8 = 0b1000_0000; +pub(super) const JOIN: u8 = 0b1000_0000; /// Resource-based endpoint access permission. /// @@ -48,6 +49,10 @@ pub enum ResourceType { ChannelGroup, /// `UserId`-based resource permissions. UserId, + /// Channels category permissions. + ChannelsCategory, + /// Uuids category permissions. + UuidsCategory, } /// `Channel` permission. @@ -114,6 +119,48 @@ pub struct UserIdPermission { pub bits: u8, } +/// `Channels category` permission. +/// +/// Single `channels category` category permission information. +/// +/// # Example +/// ```rust, no_run +/// # use pubnub::dx::access::permissions::{self, ChannelsCategoryPermission, Permission}; +/// # +/// let channels_category_permission = // Channels category +/// # permissions::channels_category() +/// .get(); +/// ``` +#[derive(Debug)] +pub struct ChannelsCategoryPermission { + /// Name of channels category name for which permission level specified. + name: String, + + /// Bitmask with configured permission level. + pub bits: u8, +} + +/// `Uuids category` permission. +/// +/// Single `uuids category` permission information. +/// +/// # Example +/// ```rust, no_run +/// # use pubnub::dx::access::permissions::{self, UuidsCategoryPermission, Permission}; +/// # +/// let uuids_category_permission = // Uuids category +/// # permissions::uuids_category() +/// .get(); +/// ``` +#[derive(Debug)] +pub struct UuidsCategoryPermission { + /// Name of uuids category name for which permission level specified. + name: String, + + /// Bitmask with configured permission level. + pub bits: u8, +} + /// `Channel`-based endpoint permissions. /// /// Trait contains methods to configure permissions to access endpoints related @@ -309,6 +356,58 @@ impl Permission for UserIdPermission { } } +/// `Channels category` enumeration endpoint permissions. +impl ChannelsCategoryPermission { + /// Resource **get** permissions. + /// + /// This permission for channels category allows to: + /// * get all channels metadata + pub fn get(mut self) -> Box { + self.bits |= GET; + Box::new(self) + } +} + +impl Permission for ChannelsCategoryPermission { + fn id(&self) -> &String { + &self.name + } + + fn value(&self) -> &u8 { + &self.bits + } + + fn resource_type(&self) -> ResourceType { + ResourceType::ChannelsCategory + } +} + +/// `Uuids category` enumeration endpoint permissions. +impl UuidsCategoryPermission { + /// Resource **get** permissions. + /// + /// This permission for uuids category allows to: + /// * get all uuids metadata + pub fn get(mut self) -> Box { + self.bits |= GET; + Box::new(self) + } +} + +impl Permission for UuidsCategoryPermission { + fn id(&self) -> &String { + &self.name + } + + fn value(&self) -> &u8 { + &self.bits + } + + fn resource_type(&self) -> ResourceType { + ResourceType::UuidsCategory + } +} + /// Grant permission to the `channel`. /// /// Create a `channel` permission information object that can be used to specify @@ -377,6 +476,44 @@ where }) } +/// Grant permission to the `channel category`. +/// +/// Create a `channel category` permission information object that can be used +/// to specify required permissions. +/// +/// # Example +/// ```rust +/// # use pubnub::dx::access::permissions::{self, Permission, ChannelsCategoryPermission}; +/// # +/// let channels_category_permission = permissions::channels_category().get(); +/// # assert_eq!(channels_category_permission.value(), &0b0010_0000); +/// ``` +pub fn channels_category() -> Box { + Box::new(ChannelsCategoryPermission { + name: String::from("channels"), + bits: 0, + }) +} + +/// Grant permission to the `uuid category`. +/// +/// Create a `uuid category` permission information object that can be used +/// to specify required permissions. +/// +/// # Example +/// ```rust +/// # use pubnub::dx::access::permissions::{self, Permission, UuidsCategoryPermission}; +/// # +/// let uuids_category_permission = permissions::uuids_category().get(); +/// # assert_eq!(uuids_category_permission.value(), &0b0010_0000); +/// ``` +pub fn uuids_category() -> Box { + Box::new(UuidsCategoryPermission { + name: String::from("uuids"), + bits: 0, + }) +} + #[cfg(test)] mod it_should { use super::*; @@ -416,4 +553,18 @@ mod it_should { assert_eq!(permission.value(), &0b0110_0000); assert_eq!(permission.id(), user_id_value); } + + #[test] + fn create_channels_category_get_permission() { + let permission = channels_category().get(); + assert_eq!(permission.value(), &0b0010_0000); + assert_eq!(permission.id(), "channels"); + } + + #[test] + fn create_uuids_category_get_permission() { + let permission = uuids_category().get(); + assert_eq!(permission.value(), &0b0010_0000); + assert_eq!(permission.id(), "uuids"); + } } diff --git a/src/dx/parse_token.rs b/src/dx/parse_token.rs index 10fc9428..e4f8e87a 100644 --- a/src/dx/parse_token.rs +++ b/src/dx/parse_token.rs @@ -93,13 +93,18 @@ pub struct TokenV2 { /// Permissions for resources identified by their names. #[cfg_attr(feature = "serde", serde(rename = "res"))] - pub resources: TokenResources, + pub resources: Option, /// Permissions for resources identified by regular expressions. #[cfg_attr(feature = "serde", serde(rename = "pat"))] - pub patterns: TokenResources, + pub patterns: Option, + + /// Permissions for enumerate-able resource categories. + #[cfg_attr(feature = "serde", serde(rename = "cat"))] + pub categories: Option, /// Extra metadata to which has been included into access token. + #[cfg_attr(feature = "serde", serde(default))] pub meta: HashMap, } @@ -110,20 +115,34 @@ pub struct TokenV2 { pub struct TokenResources { /// `Channel`-based endpoints permission map between channel name / regexp /// and set of permissions. - #[cfg_attr(feature = "serde", serde(rename = "chan"))] + #[cfg_attr(feature = "serde", serde(rename = "chan", default))] pub channels: HashMap, /// `Channel group`-based endpoints permission map between channel /// name / regexp and set of permissions. - #[cfg_attr(feature = "serde", serde(rename = "grp"))] + #[cfg_attr(feature = "serde", serde(rename = "grp", default))] pub groups: HashMap, /// `UserId`-based endpoints permission map between channel /// name / regexp and set of permissions. - #[cfg_attr(feature = "serde", serde(rename = "uuid"))] + #[cfg_attr(feature = "serde", serde(rename = "uuid", default))] pub users: HashMap, } +/// Typed category permissions map. +#[derive(Debug, Clone)] +#[cfg_attr(feature = "serde", derive(serde::Deserialize))] +#[cfg_attr(test, derive(PartialEq, Eq))] +pub struct TokenCategories { + /// `Channels` enumeration endpoint permission. + #[cfg_attr(feature = "serde", serde(rename = "chan", default))] + pub channels: Option, + + /// `Uuids` enumeration endpoint permission. + #[cfg_attr(feature = "serde", serde(rename = "uuid", default))] + pub uuids: Option, +} + impl From for ResourcePermissions { fn from(int: u8) -> Self { ResourcePermissions { @@ -173,6 +192,25 @@ pub struct ResourcePermissions { pub join: bool, } +impl From for CategoryPermissions { + fn from(int: u8) -> Self { + CategoryPermissions { get: int & 32 != 0 } + } +} + +/// Category permissions map. +/// +/// This structure contains information about permissions which has been granted +/// to specific category. +#[derive(Debug, Clone)] +#[cfg_attr(feature = "serde", derive(serde::Deserialize))] +#[cfg_attr(test, derive(PartialEq, Eq))] +#[cfg_attr(feature = "serde", serde(from = "u8"))] +pub struct CategoryPermissions { + /// Whether or not the category has **get** permission. + pub get: bool, +} + /// Enum for values associated with token. #[derive(Debug, Clone)] #[cfg_attr(feature = "serde", derive(serde::Deserialize))] @@ -228,16 +266,17 @@ mod should { version: 2, ttl: 1337, timestamp: 1681219882, - patterns: TokenResources { + patterns: Some(TokenResources { channels: [("channel.*".into(), 2.into())].into(), groups: [("channelGroup.*".into(), 4.into())].into(), users: [("uuid.*".into(), 104.into())].into(), - }, - resources: TokenResources { + }), + resources: Some(TokenResources { users: [].into(), groups: [("channelGroup".into(), 1.into())].into(), channels: [("channelResource".into(), 255.into())].into(), - }, + }), + categories: None, authorized_user_id: Some("uuid".into()), meta: HashMap::from([ ("meta".into(), String("data".into())), @@ -249,4 +288,55 @@ mod should { token ); } + + #[test] + fn parse_token_with_categories_only() { + let Token::V2(token) = parse_token( + "qGF2AmF0GmFLd-NjdHRsGQWgY3Jlc6BjcGF0oGRtZXRhoGNjYXSiZGNoYW4YIGR1dWlkGCBjc2lnRAECAwQ=", + ) + .unwrap(); + assert_eq!(token.version, 2); + assert_eq!(token.timestamp, 1632335843); + assert_eq!(token.ttl, 1440); + let categories = token.categories.expect("categories should be present"); + assert_eq!(categories.channels, Some(CategoryPermissions { get: true })); + assert_eq!(categories.uuids, Some(CategoryPermissions { get: true })); + } + + #[test] + fn deserialize_categories_without_uuids() { + // CBOR for {"chan": 32} + let cbor = [0xa1, 0x64, b'c', b'h', b'a', b'n', 0x18, 0x20]; + let categories: TokenCategories = ciborium::de::from_reader(&cbor[..]).unwrap(); + assert_eq!(categories.channels, Some(CategoryPermissions { get: true })); + assert_eq!(categories.uuids, None); + } + + #[test] + fn deserialize_categories_without_channels() { + // CBOR for {"uuid": 32} + let cbor = [0xa1, 0x64, b'u', b'u', b'i', b'd', 0x18, 0x20]; + let categories: TokenCategories = ciborium::de::from_reader(&cbor[..]).unwrap(); + assert_eq!(categories.channels, None); + assert_eq!(categories.uuids, Some(CategoryPermissions { get: true })); + } + + #[test] + fn ignore_unknown_category_keys() { + // CBOR for {"chan": 32, "foo": 1} + let cbor = [ + 0xa2, 0x64, b'c', b'h', b'a', b'n', 0x18, 0x20, 0x63, b'f', b'o', b'o', 0x01, + ]; + let categories: TokenCategories = ciborium::de::from_reader(&cbor[..]).unwrap(); + assert_eq!(categories.channels, Some(CategoryPermissions { get: true })); + } + + #[test] + fn deserialize_empty_resources() { + // CBOR for {} + let resources: TokenResources = ciborium::de::from_reader(&[0xa0u8][..]).unwrap(); + assert!(resources.channels.is_empty()); + assert!(resources.groups.is_empty()); + assert!(resources.users.is_empty()); + } } diff --git a/src/dx/presence/builders/get_presence_state.rs b/src/dx/presence/builders/get_presence_state.rs index 8cab7e4d..a431f037 100644 --- a/src/dx/presence/builders/get_presence_state.rs +++ b/src/dx/presence/builders/get_presence_state.rs @@ -123,7 +123,7 @@ impl GetStateRequest { Ok(TransportRequest { path: format!( "/v2/presence/sub-key/{}/channel/{}/uuid/{}", - &config.subscribe_key, + config.subscribe_key, url_encoded_channels(&self.channels), url_encode_extended(self.user_id.as_bytes(), UrlEncodeExtension::NonChannelPath) ), diff --git a/src/dx/presence/builders/heartbeat.rs b/src/dx/presence/builders/heartbeat.rs index 68eb10fc..b8776508 100644 --- a/src/dx/presence/builders/heartbeat.rs +++ b/src/dx/presence/builders/heartbeat.rs @@ -234,7 +234,7 @@ impl HeartbeatRequest { Ok(TransportRequest { path: format!( "/v2/presence/sub_key/{}/channel/{}/heartbeat", - &config.subscribe_key, + config.subscribe_key, url_encoded_channels(&self.channels) ), query_parameters: query, diff --git a/src/dx/presence/builders/here_now.rs b/src/dx/presence/builders/here_now.rs index 99046e68..d4fb5711 100644 --- a/src/dx/presence/builders/here_now.rs +++ b/src/dx/presence/builders/here_now.rs @@ -154,7 +154,7 @@ impl HereNowRequest { Ok(TransportRequest { path: format!( "/v2/presence/sub-key/{}/channel/{}", - &config.subscribe_key, + config.subscribe_key, url_encoded_channels(&self.channels), ), query_parameters: query, diff --git a/src/dx/presence/builders/leave.rs b/src/dx/presence/builders/leave.rs index 8162745f..7be480ac 100644 --- a/src/dx/presence/builders/leave.rs +++ b/src/dx/presence/builders/leave.rs @@ -144,7 +144,7 @@ impl LeaveRequest { Ok(TransportRequest { path: format!( "/v2/presence/sub_key/{}/channel/{}/leave", - &config.subscribe_key, + config.subscribe_key, url_encoded_channels(&self.channels) ), query_parameters: query, diff --git a/src/dx/presence/builders/set_presence_state.rs b/src/dx/presence/builders/set_presence_state.rs index b3e8b9a1..f6c4247a 100644 --- a/src/dx/presence/builders/set_presence_state.rs +++ b/src/dx/presence/builders/set_presence_state.rs @@ -171,7 +171,7 @@ impl SetStateRequest { Ok(TransportRequest { path: format!( "/v2/presence/sub-key/{}/channel/{}/uuid/{}/data", - &config.subscribe_key, + config.subscribe_key, url_encoded_channels(&self.channels), url_encode_extended(self.user_id.as_bytes(), UrlEncodeExtension::NonChannelPath) ), diff --git a/src/dx/presence/builders/where_now.rs b/src/dx/presence/builders/where_now.rs index 5d018f77..c220c0b0 100644 --- a/src/dx/presence/builders/where_now.rs +++ b/src/dx/presence/builders/where_now.rs @@ -89,7 +89,7 @@ impl WhereNowRequest { Ok(TransportRequest { path: format!( "/v2/presence/sub-key/{}/uuid/{}", - &config.subscribe_key, + config.subscribe_key, url_encode_extended(user_id.as_bytes(), UrlEncodeExtension::NonChannelPath) ), query_parameters: HashMap::new(), diff --git a/src/dx/pubnub_client.rs b/src/dx/pubnub_client.rs index 4cfc1bf4..dbf7755b 100644 --- a/src/dx/pubnub_client.rs +++ b/src/dx/pubnub_client.rs @@ -441,7 +441,7 @@ impl PubNubClientInstance { let mut entities_slot = self.entities.write(); let name = name.into(); let entity = entities_slot - .entry(format!("{}_ch", &name)) + .entry(format!("{}_ch", name)) .or_insert(Channel::new(self, name).into()); match entity { @@ -541,7 +541,7 @@ impl PubNubClientInstance { let mut entities_slot = self.entities.write(); let name = name.into(); let entity = entities_slot - .entry(format!("{}_chg", &name)) + .entry(format!("{}_chg", name)) .or_insert(ChannelGroup::new(self, name).into()); match entity { @@ -644,7 +644,7 @@ impl PubNubClientInstance { let mut entities_slot = self.entities.write(); let id = id.into(); let entity = entities_slot - .entry(format!("{}_chm", &id)) + .entry(format!("{}_chm", id)) .or_insert(ChannelMetadata::new(self, id).into()); match entity { @@ -751,7 +751,7 @@ impl PubNubClientInstance { let mut entities_slot = self.entities.write(); let id = id.into(); let entity = entities_slot - .entry(format!("{}_uidm", &id)) + .entry(format!("{}_uidm", id)) .or_insert(UserMetadata::new(self, id).into()); match entity { diff --git a/src/dx/subscribe/event_dispatcher.rs b/src/dx/subscribe/event_dispatcher.rs index 697d8f16..f479e06f 100644 --- a/src/dx/subscribe/event_dispatcher.rs +++ b/src/dx/subscribe/event_dispatcher.rs @@ -286,11 +286,15 @@ impl EventDispatcher { /// Invalidates all streams in the instance. pub(crate) fn invalidate(&self) { - let mut has_streams_slot = self.has_streams.write(); - if !*has_streams_slot { - return; + // The flag guard must be released before stream lists are locked: + // stream creation locks a list first and the flag second. + { + let mut has_streams_slot = self.has_streams.write(); + if !*has_streams_slot { + return; + } + *has_streams_slot = false; } - *has_streams_slot = false; if let Some(streams) = self.message_streams.write().as_mut() { streams.iter_mut().for_each(|stream| stream.invalidate()); @@ -550,3 +554,269 @@ mod it_should { Ok(()) } } + +#[cfg(test)] +mod lock_order { + use super::*; + use std::{ + pin::Pin, + sync::{ + atomic::{AtomicBool, AtomicUsize, Ordering}, + mpsc, Arc, + }, + task::{Context, Poll, Wake, Waker}, + thread, + time::{Duration, Instant}, + }; + + use futures::Stream; + + /// How long to wait for the expected state before the test fails. + const WAIT: Duration = Duration::from_secs(2); + + fn wait_until(mut condition: impl FnMut() -> bool) -> bool { + let deadline = Instant::now() + WAIT; + while Instant::now() < deadline { + if condition() { + return true; + } + thread::yield_now(); + } + condition() + } + + /// Dispatcher with `has_streams` set, so `invalidate()` doesn't return early. + fn dispatcher_with_stream() -> Arc { + let dispatcher = Arc::new(EventDispatcher::new()); + let _stream = dispatcher.messages_stream(); + assert!(*dispatcher.has_streams.read()); + dispatcher + } + + #[test] + fn create_stream_while_invalidating_without_deadlock() { + let dispatcher = dispatcher_with_stream(); + let creator_holds_list = Arc::new(AtomicBool::new(false)); + let creator_may_continue = Arc::new(AtomicBool::new(false)); + let finished = Arc::new(AtomicUsize::new(0)); + + // Same two steps as `signals_stream()`: lock the list, then create. + { + let dispatcher = dispatcher.clone(); + let holds_list = creator_holds_list.clone(); + let may_continue = creator_may_continue.clone(); + let finished = finished.clone(); + thread::spawn(move || { + let list = dispatcher.signal_streams.write(); + holds_list.store(true, Ordering::SeqCst); + while !may_continue.load(Ordering::SeqCst) { + thread::yield_now(); + } + dispatcher.create_stream_in_list(list, None); + finished.fetch_add(1, Ordering::SeqCst); + }); + } + assert!(wait_until(|| creator_holds_list.load(Ordering::SeqCst))); + + { + let dispatcher = dispatcher.clone(); + let finished = finished.clone(); + thread::spawn(move || { + dispatcher.invalidate(); + finished.fetch_add(1, Ordering::SeqCst); + }); + } + + // `invalidate()` emptied the first list, so it is waiting for the + // `signal_streams` lock held by the creator. + assert!(wait_until(|| { + dispatcher + .message_streams + .try_read() + .is_some_and(|streams| match streams.as_ref() { + Some(list) => list.is_empty(), + None => true, + }) + })); + + creator_may_continue.store(true, Ordering::SeqCst); + + assert!( + wait_until(|| finished.load(Ordering::SeqCst) == 2), + "create_stream_in_list() and invalidate() deadlocked" + ); + } + + #[test] + fn dispatch_events_while_creating_stream_and_invalidating_without_deadlock() { + let dispatcher = dispatcher_with_stream(); + let creator_holds_list = Arc::new(AtomicBool::new(false)); + let creator_may_continue = Arc::new(AtomicBool::new(false)); + let dispatch_holds_list = Arc::new(AtomicBool::new(false)); + let dispatch_may_continue = Arc::new(AtomicBool::new(false)); + let finished = Arc::new(AtomicUsize::new(0)); + + // Stream creation: lock the list, then create. + { + let dispatcher = dispatcher.clone(); + let holds_list = creator_holds_list.clone(); + let may_continue = creator_may_continue.clone(); + let finished = finished.clone(); + thread::spawn(move || { + let list = dispatcher.signal_streams.write(); + holds_list.store(true, Ordering::SeqCst); + while !may_continue.load(Ordering::SeqCst) { + thread::yield_now(); + } + dispatcher.create_stream_in_list(list, None); + finished.fetch_add(1, Ordering::SeqCst); + }); + } + assert!(wait_until(|| creator_holds_list.load(Ordering::SeqCst))); + + // Event dispatch: `handle_events()` keeps read locks on several lists + // taken one after another (replicated with two of them). + { + let dispatcher = dispatcher.clone(); + let holds_list = dispatch_holds_list.clone(); + let may_continue = dispatch_may_continue.clone(); + let finished = finished.clone(); + thread::spawn(move || { + let _messages = dispatcher.message_streams.read(); + holds_list.store(true, Ordering::SeqCst); + while !may_continue.load(Ordering::SeqCst) { + thread::yield_now(); + } + let _signals = dispatcher.signal_streams.read(); + finished.fetch_add(1, Ordering::SeqCst); + }); + } + assert!(wait_until(|| dispatch_holds_list.load(Ordering::SeqCst))); + + { + let dispatcher = dispatcher.clone(); + let finished = finished.clone(); + thread::spawn(move || { + dispatcher.invalidate(); + finished.fetch_add(1, Ordering::SeqCst); + }); + } + + // `invalidate()` marked the dispatcher as having no streams. + assert!(wait_until(|| { + match dispatcher.has_streams.try_read() { + Some(has_streams) => !*has_streams, + None => true, + } + })); + + creator_may_continue.store(true, Ordering::SeqCst); + thread::sleep(Duration::from_millis(100)); + dispatch_may_continue.store(true, Ordering::SeqCst); + + assert!( + wait_until(|| finished.load(Ordering::SeqCst) == 3), + "dispatch, create_stream_in_list() and invalidate() deadlocked" + ); + } + + #[test] + fn create_and_invalidate_concurrently_without_deadlock() { + const ROUNDS: usize = 2_000; + + for round in 0..ROUNDS { + let dispatcher = dispatcher_with_stream(); + let start = Arc::new(AtomicBool::new(false)); + let finished = Arc::new(AtomicUsize::new(0)); + + for create in [true, false] { + let dispatcher = dispatcher.clone(); + let start = start.clone(); + let finished = finished.clone(); + thread::spawn(move || { + while !start.load(Ordering::SeqCst) { + std::hint::spin_loop(); + } + if create { + let _stream = dispatcher.signals_stream(); + } else { + dispatcher.invalidate(); + } + finished.fetch_add(1, Ordering::SeqCst); + }); + } + start.store(true, Ordering::SeqCst); + + assert!( + wait_until(|| finished.load(Ordering::SeqCst) == 2), + "signals_stream() and invalidate() deadlocked in round {round}" + ); + } + } + + /// Wakes the thread which owns the listener. + struct ThreadWaker(thread::Thread); + + impl Wake for ThreadWaker { + fn wake(self: Arc) { + self.0.unpark(); + } + } + + fn message(index: usize) -> Update { + Update::Message(Message { + sender: None, + timestamp: 0, + channel: "test-channel".to_string(), + subscription: "test-channel".to_string(), + data: index.to_string().into_bytes(), + r#type: None, + space_id: None, + user_metadata: None, + decryption_error: None, + }) + } + + #[test] + fn deliver_dispatched_batch_to_listener_without_deadlock() { + const ROUNDS: usize = 100; + const BATCH: usize = 100; + + for round in 0..ROUNDS { + let dispatcher = Arc::new(EventDispatcher::new()); + let mut stream = dispatcher.messages_stream(); + let (result_tx, result_rx) = mpsc::channel(); + + thread::spawn(move || { + let waker = Waker::from(Arc::new(ThreadWaker(thread::current()))); + let mut context = Context::from_waker(&waker); + let mut received = Vec::with_capacity(BATCH); + while received.len() < BATCH { + match Pin::new(&mut stream).poll_next(&mut context) { + Poll::Ready(Some(message)) => { + received.push(String::from_utf8(message.data).unwrap()) + } + Poll::Ready(None) => break, + Poll::Pending => thread::park(), + } + } + let _ = result_tx.send(received); + }); + + let producer = { + let dispatcher = dispatcher.clone(); + thread::spawn(move || dispatcher.handle_events((0..BATCH).map(message).collect())) + }; + + let received = result_rx + .recv_timeout(WAIT * 5) + .unwrap_or_else(|_| panic!("round {round}: listener stuck (deadlock)")); + producer.join().expect("producer thread panicked"); + + let expected = (0..BATCH) + .map(|index| index.to_string()) + .collect::>(); + assert_eq!(received, expected, "round {round}"); + } + } +} diff --git a/tests/access/access_steps.rs b/tests/access/access_steps.rs index 5cb87c7c..6a10e9a0 100644 --- a/tests/access/access_steps.rs +++ b/tests/access/access_steps.rs @@ -250,6 +250,9 @@ fn token_contains_resource( } else { token.patterns }; + let Some(resources) = resources else { + panic!("token has no {category} permissions"); + }; if resource == "CHANNEL" && resources.channels.contains_key(name.as_str()) { world.pam_state.resource_type = PAMCurrentResourceType::Channel; @@ -282,6 +285,9 @@ fn token_resource_has_permission(world: &mut PubNubWorld, category: String, perm } else { token.patterns }; + let Some(resources) = resources else { + panic!("token has no {category} permissions"); + }; let resource_permission = match world.pam_state.resource_type { PAMCurrentResourceType::Channel => resources.channels, PAMCurrentResourceType::ChannelGroup => resources.groups, diff --git a/tests/integration/access_integration.rs b/tests/integration/access_integration.rs index 9c87b824..442d95d7 100644 --- a/tests/integration/access_integration.rs +++ b/tests/integration/access_integration.rs @@ -1,36 +1,45 @@ #[cfg(test)] mod integration { use pubnub::core::PubNubError; - use pubnub::{access::*, Keyset, PubNubClientBuilder}; + use pubnub::dx::parse_token::MetaValue; + use pubnub::{access::*, parse_token, Keyset, PubNubClientBuilder, Token}; use std::collections::HashMap; - /// Keyset with default non-PAM enabled keys. - fn default_keyset() -> Keyset { - Keyset { - subscribe_key: option_env!("SDK_SUB_KEY").unwrap_or("demo").into(), - publish_key: Some(option_env!("SDK_PUB_KEY").unwrap_or("demo").into()), - secret_key: Some(option_env!("SDK_PUB_KEY").unwrap_or("demo").into()), - } + /// Read required keyset value from the environment. + /// + /// Values are provided by CI secrets. Nothing is hardcoded. + fn required_env(name: &str) -> String { + std::env::var(name) + .unwrap_or_else(|_| panic!("{name} must be set to run PAM integration tests")) } /// Keyset with PAM enabled keys. fn pam_keyset() -> Keyset { Keyset { - subscribe_key: option_env!("SDK_PAM_SUB_KEY").unwrap_or("demo").into(), - publish_key: Some(option_env!("SDK_PAM_PUB_KEY").unwrap_or("demo").into()), - secret_key: Some(option_env!("SDK_PAM_SEC_KEY").unwrap_or("demo").into()), + subscribe_key: required_env("SDK_PAM_SUB_KEY"), + publish_key: Some(required_env("SDK_PAM_PUB_KEY")), + secret_key: Some(required_env("SDK_PAM_SEC_KEY")), + } + } + + /// Keyset with PAM enabled subscribe / publish keys and wrong secret key. + fn pam_keyset_with_wrong_secret() -> Keyset { + Keyset { + subscribe_key: required_env("SDK_PAM_SUB_KEY"), + publish_key: Some(required_env("SDK_PAM_PUB_KEY")), + secret_key: Some("sec-c-invalid".into()), } } - /// Grant token success + /// Grant token with resources and meta, then parse it back. #[tokio::test] - async fn should_grant_token() -> Result<(), Box> { + async fn should_grant_token_and_parse_resources() -> Result<(), Box> { let client = PubNubClientBuilder::with_reqwest_transport() .with_keyset(pam_keyset()) .with_user_id("test") .build()?; - let result = client + let granted = client .grant_token(10) .resources(&[permissions::channel("test-channel").read().write()]) .meta(HashMap::from([ @@ -38,48 +47,120 @@ mod integration { ("null-val".into(), ().into()), ])) .execute() - .await; + .await?; + assert!(!granted.token.is_empty()); + + let Token::V2(token) = parse_token(&granted.token)?; + assert_eq!(token.version, 2); + assert_eq!(token.ttl, 10); + + let resources = token.resources.expect("resources should be present"); + let channel = resources + .channels + .get("test-channel") + .expect("test-channel should be granted"); + assert!(channel.read); + assert!(channel.write); + assert!(!channel.manage); + + assert!(token.categories.is_none()); + assert!(matches!( + token.meta.get("string-val"), + Some(MetaValue::String(value)) if value == "hello there" + )); - match result { - Ok(result) => { - assert!(!result.token.is_empty()); - Ok(()) - } - Err(err) => { - panic!("Unexpected error: {}", err); - } - } + Ok(()) + } + + /// Grant token with categories only, then parse it back. + #[tokio::test] + async fn should_grant_categories_only_token_and_parse_it( + ) -> Result<(), Box> { + let client = PubNubClientBuilder::with_reqwest_transport() + .with_keyset(pam_keyset()) + .with_user_id("test") + .build()?; + + let categories: Vec> = vec![ + permissions::channels_category().get(), + permissions::uuids_category().get(), + ]; + let granted = client + .grant_token(10) + .categories(&categories) + .execute() + .await?; + + let Token::V2(token) = parse_token(&granted.token)?; + let categories = token.categories.expect("categories should be present"); + assert!(categories.channels.expect("channels category").get); + assert!(categories.uuids.expect("uuids category").get); + + assert!(token.resources.is_none_or(|r| r.channels.is_empty())); + assert!(token.patterns.is_none_or(|p| p.channels.is_empty())); + + Ok(()) + } + + /// Grant token with resources, patterns and categories, then parse it back. + #[tokio::test] + async fn should_grant_all_permission_kinds_and_parse_them( + ) -> Result<(), Box> { + let client = PubNubClientBuilder::with_reqwest_transport() + .with_keyset(pam_keyset()) + .with_user_id("test") + .build()?; + + let categories: Vec> = + vec![permissions::uuids_category().get()]; + let granted = client + .grant_token(10) + .resources(&[permissions::channel("test-channel").read()]) + .patterns(&[permissions::channel("test-*").write()]) + .categories(&categories) + .execute() + .await?; + + let Token::V2(token) = parse_token(&granted.token)?; + + let resources = token.resources.expect("resources should be present"); + assert!(resources.channels["test-channel"].read); + + let patterns = token.patterns.expect("patterns should be present"); + assert!(patterns.channels["test-*"].write); + + let categories = token.categories.expect("categories should be present"); + assert!(categories.channels.is_none()); + assert!(categories.uuids.expect("uuids category").get); + + Ok(()) } /// Grant token failure because of signature mismatch. #[tokio::test] - async fn should_not_grant_token() -> Result<(), Box> { + async fn should_not_grant_token_with_wrong_secret() -> Result<(), Box> { let client = PubNubClientBuilder::with_reqwest_transport() - .with_keyset(default_keyset()) + .with_keyset(pam_keyset_with_wrong_secret()) .with_user_id("test") .build()?; let result = client .grant_token(10) .resources(&[permissions::channel("test-channel").read().write()]) - .meta(HashMap::from([("string-val".into(), "hello there".into())])) .execute() .await; match result { Ok(_) => panic!("Request should fail."), - Err(err) => { - if let PubNubError::API { status, .. } = err { - assert_eq!(status, 403); - Ok(()) - } else { - panic!("Unexpected error type."); - } + Err(PubNubError::API { status, .. }) => { + assert_eq!(status, 403); + Ok(()) } + Err(_) => panic!("Unexpected error type."), } } - /// Revoke token success + /// Grant token then revoke it. #[tokio::test] async fn should_revoke_token() -> Result<(), Box> { let client = PubNubClientBuilder::with_reqwest_transport() @@ -87,24 +168,18 @@ mod integration { .with_user_id("test") .build()?; - let result = client + let granted = client .grant_token(10) .resources(&[permissions::channel("test-channel").read().write()]) - .meta(HashMap::from([("string-val".into(), "hello there".into())])) .execute() .await?; - let result = client.revoke_token(result.token).execute().await; - - if let Err(err) = result { - eprintln!("Revoke token error: {err}"); - panic!("Request shouldn't fail"); - } + client.revoke_token(granted.token).execute().await?; Ok(()) } - /// Revoke token failure + /// Revoke token failure with expired / unknown token. #[tokio::test] async fn should_not_revoke_expired_token() -> Result<(), Box> { let client = PubNubClientBuilder::with_reqwest_transport() @@ -116,39 +191,18 @@ mod integration { match result { Ok(_) => panic!("Request should fail."), - Err(err) => { - if let PubNubError::API { status, .. } = err { - // We are expecting failure because of wrong access token. - assert_eq!(status, 400); - Ok(()) - } else { - panic!("Unexpected error type."); - } + Err(PubNubError::API { status, .. }) => { + // We are expecting failure because of wrong access token. + assert_eq!(status, 400); + Ok(()) } + Err(_) => panic!("Unexpected error type."), } } - /// Revoke token failure - #[tokio::test] - async fn should_not_revoke_malformed_token() -> Result<(), Box> { - let client = PubNubClientBuilder::with_reqwest_transport() - .with_keyset(pam_keyset()) - .with_user_id("demo") - .build()?; - - let result = client.revoke_token("some-token-here").execute().await; - - match result { - Ok(_) => panic!("Request should fail."), - Err(err) => { - if let PubNubError::API { status, .. } = err { - // We are expecting failure because of wrong access token. - assert_eq!(status, 400); - Ok(()) - } else { - panic!("Unexpected error type."); - } - } - } + /// Parse failure for token which is not base64 / CBOR. + #[test] + fn should_not_parse_malformed_token() { + assert!(parse_token("some-token-here").is_err()); } } From 07826968f04131ebe9b17c49e98d608296e631fd Mon Sep 17 00:00:00 2001 From: Serhii Mamontov Date: Thu, 15 Oct 2026 00:22:01 +0300 Subject: [PATCH 2/2] fix(features): gate modules and imports per feature set Gate `data_stream`, request helpers and `blocking` import so narrow feature builds compile. refactor(parse_token): decouple parse_token from access feature Move permission bits to `permission_bits` in core utils. `parse_token` builds without `access`. `access` re-exports the bits. refactor: address issues reported by PR reviewer docs(readme): standardize README to SDK template --- Cargo.toml | 11 +- README.md | 276 ++++++++++---------------- src/core/error.rs | 14 +- src/core/mod.rs | 2 + src/core/transport_request.rs | 6 + src/core/utils/mod.rs | 3 + src/core/utils/permission_bits.rs | 23 +++ src/dx/access/builders/grant_token.rs | 4 +- src/dx/access/payloads.rs | 6 +- src/dx/access/permissions.rs | 28 +-- src/dx/access/types.rs | 1 + src/dx/parse_token.rs | 21 +- src/dx/presence/builders/heartbeat.rs | 4 +- 13 files changed, 185 insertions(+), 214 deletions(-) create mode 100644 src/core/utils/permission_bits.rs diff --git a/Cargo.toml b/Cargo.toml index a9a34921..555740a4 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -18,7 +18,7 @@ build = "build.rs" full = ["publish", "subscribe", "presence", "access", "serde", "reqwest", "crypto", "parse_token", "blocking", "std", "tokio"] # Enables all default features -default = ["publish", "subscribe", "serde", "reqwest", "std", "blocking", "tokio", "access"] +default = ["publish", "subscribe", "serde", "reqwest", "std", "blocking", "tokio"] # [PubNub features] @@ -34,6 +34,12 @@ crypto = ["dep:aes", "dep:cbc", "getrandom"] ## Enables token parsing parse_token = ["dep:ciborium"] +## Enables subscribe API +subscribe = ["dep:futures"] + +## Enables presence API +presence = ["dep:futures"] + # [Implementation features] ## Enables serde implementation for serialization and deserialization @@ -68,9 +74,6 @@ contract_test = ["parse_token", "publish", "access", "crypto", "std", "subscribe full_no_std = ["serde", "reqwest", "crypto", "parse_token", "blocking", "publish", "access", "subscribe", "tokio", "presence"] full_no_std_platform_independent = ["serde", "crypto", "parse_token", "blocking", "publish", "access", "subscribe", "presence"] pubnub_only = ["crypto", "parse_token", "blocking", "publish", "access", "subscribe", "presence"] -# TODO: temporary treated as internal until we officially release it -subscribe = ["dep:futures"] -presence = ["dep:futures"] [dependencies] async-trait = "0.1" diff --git a/README.md b/README.md index 1b4f9891..cd61cc0f 100644 --- a/README.md +++ b/README.md @@ -1,248 +1,172 @@ rust-header - # PubNub Rust SDK -## Overview - -This is the official PubNub Rust SDK repository. +[![crates.io](https://img.shields.io/crates/v/pubnub.svg)](https://crates.io/crates/pubnub) -[PubNub](https://www.pubnub.com/) takes care of the infrastructure and APIs needed for the realtime -communication layer of your application. Work on your app's logic and let -PubNub handle sending and receiving data across the world in less than -100ms. +PubNub provides global infrastructure for real-time, interactive applications. -## Getting started +Publish and receive messages in Rust. Use this SDK for Rust server, desktop, WebAssembly, and embedded applications. -Below you can find everything you need to start messaging! +[Documentation](https://www.pubnub.com/docs/sdks/rust) · [API reference](https://www.pubnub.com/docs/sdks/rust/api-reference/publish-and-subscribe) · [Changelog](https://www.pubnub.com/docs/sdks/rust/changelog) · [Support](https://support.pubnub.com/) -### Get PubNub keys +## Requirements -You will need the publish and subscribe keys to authenticate your app. Get your keys from the [Admin Portal](https://dashboard.pubnub.com/login). +| Requirement | Supported version or setup | +| --- | --- | +| Rust | [Rust 1.65.0 or later](https://www.pubnub.com/docs/sdks/rust/platform-support) | +| Targets | [Windows, macOS, Unix/Linux, WebAssembly, and embedded systems](https://www.pubnub.com/docs/sdks/rust/platform-support) | +| Cargo dependencies | The async Quickstart uses PubNub, Tokio, Futures, and serde_json. | -### Import using [Cargo](https://doc.rust-lang.org/cargo/getting-started/installation.html) +## Installation -Add `pubnub` to your Rust project in the `Cargo.toml` file: +Add the following to `Cargo.toml`: ```toml -# default features [dependencies] pubnub = "0.8.0" - -# all features -[dependencies] -pubnub = { version = "0.8.0", features = ["full"] } +futures = "0.3" +serde_json = "1.0" +tokio = { version = "1", features = ["full"] } ``` -### Example +For other installation methods, see [environment setup](https://www.pubnub.com/docs/sdks/rust). + +## Quickstart + +This example runs in a Tokio-based Rust command-line application. It subscribes to `hello_world`, publishes one message, and prints the received text. + +### Get your keys -Try the following sample code to get up and running quickly! +1. Open the [PubNub Admin Portal](https://admin.pubnub.com/). +2. Create an app and a keyset for development, or select an existing development keyset. +3. Copy its publish key and subscribe key. + +If Access Manager is enabled for your keyset, obtain a token from your trusted backend and call `pubnub.set_token(token)` before accessing protected resources. The issued token is passed to the client. Keep the secret key on the backend and never embed it in distributed client code. + +### Send and receive a message + +Create a Cargo binary project and replace `src/main.rs` with: + +Replace the key placeholders. Use a User ID that identifies the user or device in your app. ```rust -use pubnub::subscribe::Subscriber; use futures::StreamExt; -use tokio::time::sleep; -use std::time::Duration; -use serde_json; use pubnub::{ - dx::subscribe::Update, - subscribe::EventSubscriber, + subscribe::{ConnectionStatus, EventEmitter, EventSubscriber, Subscriber}, Keyset, PubNubClientBuilder, }; + #[tokio::main] async fn main() -> Result<(), Box> { - use pubnub::subscribe::{EventEmitter, SubscriptionParams}; - let publish_key = "my_publish_key"; - let subscribe_key = "my_subscribe_key"; - let client = PubNubClientBuilder::with_reqwest_transport() + let pubnub = PubNubClientBuilder::with_reqwest_transport() .with_keyset(Keyset { - subscribe_key, - publish_key: Some(publish_key), + subscribe_key: "YOUR_SUBSCRIBE_KEY", + publish_key: Some("YOUR_PUBLISH_KEY"), secret_key: None, }) - .with_user_id("user_id") + .with_user_id("hello-world-user") .build()?; - println!("PubNub instance created"); - - let subscription = client.subscription(SubscriptionParams { - channels: Some(&["my_channel"]), - channel_groups: None, - options: None - }); + let subscription = pubnub.channel("hello_world").subscription(None); - let channel_entity = client.channel("my_channel_2"); - let channel_entity_subscription = channel_entity.subscription(None); + // Create the streams before subscribing so no event is missed. + let mut statuses = pubnub.status_stream(); + let mut messages = subscription.messages_stream(); subscription.subscribe(); - channel_entity_subscription.subscribe(); - - println!("Subscribed to channels"); - - // Launch a new task to print out each received message - tokio::spawn(client.status_stream().for_each(|status| async move { - println!("\nStatus: {:?}", status) - })); - tokio::spawn(subscription.stream().for_each(|event| async move { - match event { - Update::Message(message) | Update::Signal(message) => { - // Silently log if UTF-8 conversion fails - if let Ok(utf8_message) = String::from_utf8(message.data.clone()) { - if let Ok(cleaned) = serde_json::from_str::(&utf8_message) { - println!("message: {}", cleaned); - } - } - } - Update::Presence(presence) => { - println!("presence: {:?}", presence) - } - Update::AppContext(object) => { - println!("object: {:?}", object) - } - Update::MessageAction(action) => { - println!("message action: {:?}", action) - } - Update::File(file) => { - println!("file: {:?}", file) - } + + // Wait until the subscription is connected before publishing. + while let Some(status) = statuses.next().await { + match status { + ConnectionStatus::Connected => break, + ConnectionStatus::ConnectionError(error) => return Err(error.into()), + _ => {} } - })); - - // Explicitly listen only for real-time `message` updates. - tokio::spawn( - channel_entity_subscription - .messages_stream() - .for_each(|message| async move { - if let Ok(utf8_message) = String::from_utf8(message.data.clone()) { - if let Ok(cleaned) = serde_json::from_str::(&utf8_message) { - println!("message: {}", cleaned); - } - } - }), - ); - - sleep(Duration::from_secs(2)).await; - - // Send a message to the channel - client - .publish_message("hello world!") - .channel("my_channel") - .r#type("text-message") - .execute() - .await?; + } - // Send a message to another channel - client - .publish_message("hello world on the other channel!") - .channel("my_channel_2") - .r#type("text-message") + pubnub + .publish_message("Hello world") + .channel("hello_world") .execute() .await?; - sleep(Duration::from_secs(15)).await; + if let Some(message) = messages.next().await { + let text: String = serde_json::from_slice(&message.data)?; + println!("{text}"); + } + + subscription.unsubscribe(); + pubnub.terminate(); Ok(()) } ``` -You can find more examples in our [examples](https://github.com/pubnub/rust/tree/master/examples) directory! +### Run the example -## Features +``` +cargo run +``` -The `pubnub` crate is split into multiple features. You can enable or -disable them in the `Cargo.toml` file, like so: +The terminal should show: -```toml -# only blocking and access + default features -[dependencies] -pubnub = { version = "0.8.0", features = ["blocking", "access"] } - -# only parse_token + default features -[dependencies] -pubnub = { version = "0.8.0", features = ["parse_token"] } +``` +Hello world ``` -### Available features +The example calls `subscription.unsubscribe()` and `pubnub.terminate()` after it receives the message. In a long-running application, call them when you no longer need the subscription or when the client shuts down. -| Feature name | Description | Available PubNub APIs | -| :------------ | :---------- | :------------- | -| `full` | Enables all non-conflicting features | Configuration, Publish, Subscribe, Access Manager, Parse Token, Presence, Crypto Module | -| `default` | Enables default features: `publish`, `subscribe`, `serde`, `reqwest`, `std` | Configuration, Publish, Subscribe | -| `publish` | Enables Publish API | Configuration, Publish | -| `access` | Enables Access Manager API | Configuration, Access Manager | -| `parse_token` | Enables parsing Access Manager tokens | Configuration, Parse Token | -| `subscribe` | Enables Subscribe API | Configuration, Subscribe | -| `presence` | Enables Presence API | Configuration, Presence | -| `tokio` | Enables the [tokio](https://tokio.rs/) asynchronous runtime for Subscribe and Presence APIs | n/a | -| `serde` | Uses [serde](https://github.com/serde-rs/serde) for serialization | n/a | -| `reqwest` | Uses [reqwest](https://github.com/seanmonstar/reqwest) as a transport layer | n/a | -| `blocking` | Enables blocking executions of APIs | n/a | -| `crypto` | Enables crypto module for data encryption and decryption | n/a | -| `std` | Enables `std` library | n/a | +For a complete application, see the [getting started guide](https://www.pubnub.com/docs/sdks/rust). -## Documentation +## Next steps -* [API reference for Rust](https://www.pubnub.com/docs/sdks/rust) -* [Rust docs](https://www.docs.rs/pubnub/latest/pubnub) +| Task | Documentation | +| --- | --- | +| Configure the client | [Configuration](https://www.pubnub.com/docs/sdks/rust/api-reference/configuration) | +| Work with subscriptions and messages | [Publish & Subscribe](https://www.pubnub.com/docs/sdks/rust/api-reference/publish-and-subscribe) | +| Check channel occupancy | [Presence](https://www.pubnub.com/docs/sdks/rust/api-reference/presence) | +| Secure resources | [Access Manager](https://www.pubnub.com/docs/sdks/rust/api-reference/access-manager) | -## Wasm support +## Build with an AI coding assistant -The `pubnub` crate is compatible with WebAssembly. You can use it in your -Wasm project. +The [PubNub MCP server](https://www.pubnub.com/docs/ai/pubnub-mcp-server) gives an AI coding assistant access to PubNub SDK documentation and PubNub APIs. Connect the assistant to the hosted server at `https://mcp.pubnub.com`, or run `npx @pubnub/mcp@latest` locally. -## `no_std` support +The [server repository](https://github.com/pubnub/pubnub-mcp-server) has setup steps for VS Code, Cursor, Claude Code, Claude Desktop, Codex, Gemini CLI, and OpenCode. -The `pubnub` crate is `no_std` compatible. To use it in a `no_std` -environment, you have to disable the default features and enable the ones -you need, for example: +## Before production -```toml -[dependencies] -pubnub = { version = "0.8.0", default-features = false, features = ["serde", "publish", -"blocking"] } -``` +Use Access Manager to grant each client the permissions it needs. Keep the secret key on your backend. Never include it in a distributed client. -### Limitations +Each PubNub client holds its own transport and subscription state. Create one client per application or service and reuse it. Call `subscription.unsubscribe()` to stop an individual subscription, and call `pubnub.terminate()` when the client is shutting down. -The `no_std` support is limited by the implementation details of the SDK. +Live delivery through PubNub SDKs is at-most-once. A subscriber can miss messages while disconnected or if its buffer overflows. The current Rust SDK documentation does not expose a Message Persistence client API; check [Rust Feature Support](https://www.pubnub.com/docs/sdks/rust/feature-support) before designing history-based recovery around this SDK. -The SDK uses the `alloc` crate to allocate memory for some operations, which -means that certain targets aren't supported. Additionally, as we provide a -synchronous API, we use some parts of the `alloc::sync` module, which is -also not supported in certain `no_std` environments. +## Troubleshooting -Some SDK features aren't supported in a `no_std` environment: +| Symptom | Check | +| --- | --- | +| The Quickstart has unresolved imports for Tokio, Futures, or serde_json | Add every dependency shown in the Quickstart to `Cargo.toml`; the runnable example needs more than the `pubnub` crate alone. See the [Rust SDK guide](https://www.pubnub.com/docs/sdks/rust). | +| A publish succeeds but no message appears | Check the message handler, subscription readiness, keyset, and channel name. | +| A PubNub API is unavailable at compile time | Enable the crate feature that provides that API. See [Rust Configuration](https://www.pubnub.com/docs/sdks/rust/api-reference/configuration). | +| Extra clients or duplicate messages during development | Stop the previous process before restarting. Each live client that subscribes to a channel receives its own copy of every message, so a leftover process doubles them. | -* partially `access` module (because of lack of timestamp support) -* partially `reqwest` transport (because of the reqwest implementation - details) -* partially `subscribe` module (because of the spawning tasks and time - dependence) -* partially `presence` module (because of the spawning tasks and time - dependence) -* `std` feature (because of the `std` library) +For setup help, see [troubleshooting](https://www.pubnub.com/docs/sdks/rust). Check [network status](https://status.pubnub.com/) for service incidents. -We depend on a random number generator to generate data for debugging -purposes. If you want to use the SDK in a `no_std` environment, you'll have -to provide your own random number generator implementation for certain -targets. +## Releases -See more: +Read the [changelog](https://www.pubnub.com/docs/sdks/rust/changelog) before upgrading. -* [`getrandom` crate](https://docs.rs/getrandom/latest/getrandom/) -* [no_std examples](https://github.com/pubnub/rust/tree/master/examples/no_std/) +For a major version change, review the [Rust SDK changelog](https://www.pubnub.com/docs/sdks/rust/changelog) for breaking changes before upgrading. -If you're having problems compiling this crate for more exotic targets, you -can try to use the `extra_platforms` feature. Be aware that this feature is -**not supported** and we do not recommend using it. +## Support and contributions -For more information about this feature. refer to [Cargo.toml](https://github.com/pubnub/rust/blob/master/Cargo.toml) in the `[features]` section. +For setup or account help, contact [PubNub Support](https://support.pubnub.com/). -## Support +For a reproducible SDK bug, use [GitHub Issues](https://github.com/pubnub/rust/issues). Include the SDK version, runtime, and a small reproduction with credentials removed. -If you **need help** or have a **general question**, contact -support@pubnub.com. +Before opening a pull request, run `cargo fmt --check`, `cargo clippy --features full -- -D warnings`, and `cargo test --features full`. Include tests for behavioral changes. ## License -This project is licensed under the [MIT license](https://github.com/pubnub/rust/blob/master/LICENSE). +See the [PubNub Software Development Kit License Agreement](https://github.com/pubnub/rust/blob/master/LICENSE). diff --git a/src/core/error.rs b/src/core/error.rs index 17718e1d..d7b78514 100644 --- a/src/core/error.rs +++ b/src/core/error.rs @@ -161,7 +161,12 @@ impl PubNubError { /// /// This function used to inform about not initialized request parameters or /// validation failure. - #[cfg(any(feature = "publish", feature = "access", feature = "subscribe"))] + #[cfg(any( + feature = "publish", + feature = "access", + feature = "subscribe", + feature = "presence" + ))] pub(crate) fn general_api_error( message: S, status: Option, @@ -198,7 +203,12 @@ impl PubNubError { /// /// For better understanding some errors may provide additional information /// right from service response. - #[cfg(any(feature = "publish", feature = "access", feature = "subscribe"))] + #[cfg(any( + feature = "publish", + feature = "access", + feature = "subscribe", + feature = "presence" + ))] pub(crate) fn attach_response(self, service_response: TransportResponse) -> Self { match &self { PubNubError::API { diff --git a/src/core/mod.rs b/src/core/mod.rs index a44e8b85..5c8e0227 100644 --- a/src/core/mod.rs +++ b/src/core/mod.rs @@ -75,8 +75,10 @@ pub use runtime::Runtime; #[cfg(all(feature = "std", feature = "subscribe"))] pub mod runtime; +#[cfg(feature = "subscribe")] #[doc(inline)] pub use data_stream::DataStream; +#[cfg(feature = "subscribe")] pub mod data_stream; pub(crate) mod utils; diff --git a/src/core/transport_request.rs b/src/core/transport_request.rs index 483dca60..30da17cf 100644 --- a/src/core/transport_request.rs +++ b/src/core/transport_request.rs @@ -89,6 +89,12 @@ pub struct TransportRequest { pub timeout: u64, } +#[cfg(any( + feature = "publish", + feature = "access", + feature = "subscribe", + feature = "presence" +))] impl TransportRequest { /// Send async request and process [`PubNub API`] response. /// diff --git a/src/core/utils/mod.rs b/src/core/utils/mod.rs index 0a8a0625..76c008b1 100644 --- a/src/core/utils/mod.rs +++ b/src/core/utils/mod.rs @@ -14,3 +14,6 @@ pub mod encoding; pub mod headers; pub mod metadata; + +#[cfg(any(feature = "access", feature = "parse_token"))] +pub(crate) mod permission_bits; diff --git a/src/core/utils/permission_bits.rs b/src/core/utils/permission_bits.rs new file mode 100644 index 00000000..390119a1 --- /dev/null +++ b/src/core/utils/permission_bits.rs @@ -0,0 +1,23 @@ +//! Permission bit flags +//! +//! This module provides bit flags of Access Manager permissions shared between +//! the `access` and `parse_token` features. +//! + +/// Resource **read** permissions. +pub(crate) const READ: u8 = 0b0000_0001; +/// Resource **write** permissions. +pub(crate) const WRITE: u8 = 0b0000_0010; +/// Resource **manage** permissions. +pub(crate) const MANAGE: u8 = 0b0000_0100; +/// Resource **delete** permissions. +pub(crate) const DELETE: u8 = 0b0000_1000; +/// Resource **create** permissions. +#[cfg_attr(not(feature = "parse_token"), allow(dead_code))] +pub(crate) const CREATE: u8 = 0b0001_0000; +/// Resource **get** permissions. +pub(crate) const GET: u8 = 0b0010_0000; +/// Resource **update** permissions. +pub(crate) const UPDATE: u8 = 0b0100_0000; +/// Resource **join** permissions. +pub(crate) const JOIN: u8 = 0b1000_0000; diff --git a/src/dx/access/builders/grant_token.rs b/src/dx/access/builders/grant_token.rs index d3cd8bd3..ee698400 100644 --- a/src/dx/access/builders/grant_token.rs +++ b/src/dx/access/builders/grant_token.rs @@ -148,8 +148,8 @@ where if let Some(categories) = self.categories.flatten() { categories.iter().try_for_each(|cat| { - if !["channels", "uuids"].contains(&cat.id().as_str()) { - return Err(format!("Unexpected category type: {}", cat.id())); + if !matches!(cat.resource_type(), ResourceType::ChannelsCategory | ResourceType::UuidsCategory) { + return Err(format!("Unexpected category type: {:?}", cat.resource_type())); } if *cat.value() != permissions::GET { diff --git a/src/dx/access/payloads.rs b/src/dx/access/payloads.rs index 49c27b95..4cbbe152 100644 --- a/src/dx/access/payloads.rs +++ b/src/dx/access/payloads.rs @@ -16,11 +16,12 @@ use crate::{ }, }; -/// Resource and pattern-based permissions payload. +/// Resource and pattern-based permissions' payload. /// /// This type used by [`GrantTokenPayload`] to store permissions for specific /// resource type. #[cfg_attr(feature = "serde", derive(serde::Serialize))] +#[derive(Debug, Clone, PartialEq, Eq)] pub struct GrantTokenResourcesPayload { /// Specific channels permissions for `channel`-based endpoints access. #[cfg_attr(feature = "serde", serde(skip_serializing_if = "Option::is_none"))] @@ -41,6 +42,7 @@ pub struct GrantTokenResourcesPayload { /// This type used by [`GrantTokenPayload`] to store permissions for specific /// resource type. #[cfg_attr(feature = "serde", derive(serde::Serialize))] +#[derive(Debug, Clone, Copy, PartialEq, Eq)] pub struct GrantTokenCategoriesPayload { /// Category permissions for `channels` enumeration endpoint access. #[cfg_attr(feature = "serde", serde(skip_serializing_if = "Option::is_none"))] @@ -56,6 +58,7 @@ pub struct GrantTokenCategoriesPayload { /// This type used by [`GrantTokenPayload`] to store information about requested /// token permissions. #[cfg_attr(feature = "serde", derive(serde::Serialize))] +#[derive(Debug, Clone, PartialEq)] pub struct GrantTokenPermissionsPayload<'request> { /// A user ID, which is authorized to use the token to make API requests to /// PubNub. @@ -85,6 +88,7 @@ pub struct GrantTokenPermissionsPayload<'request> { /// A list of resource names and patterns and permissions which should be /// granted for them is contained in the payload. #[cfg_attr(feature = "serde", derive(serde::Serialize))] +#[derive(Debug, Clone, PartialEq)] pub struct GrantTokenPayload<'request> { /// How long (in minutes) the generated token should be valid. pub ttl: usize, diff --git a/src/dx/access/permissions.rs b/src/dx/access/permissions.rs index 2686c7b6..f96b6718 100644 --- a/src/dx/access/permissions.rs +++ b/src/dx/access/permissions.rs @@ -7,20 +7,9 @@ use crate::lib::{alloc::boxed::Box, alloc::string::String}; -/// Resource **read** permissions. -pub(super) const READ: u8 = 0b0000_0001; -/// Resource **write** permissions. -pub(super) const WRITE: u8 = 0b0000_0010; -/// Resource **manage** permissions. -pub(super) const MANAGE: u8 = 0b0000_0100; -/// Resource **delete** permissions. -pub(super) const DELETE: u8 = 0b0000_1000; -/// Resource **get** permissions. -pub(super) const GET: u8 = 0b0010_0000; -/// Resource **update** permissions. -pub(super) const UPDATE: u8 = 0b0100_0000; -/// Resource **join** permissions. -pub(super) const JOIN: u8 = 0b1000_0000; +pub(in crate::dx) use crate::core::utils::permission_bits::{ + DELETE, GET, JOIN, MANAGE, READ, UPDATE, WRITE, +}; /// Resource-based endpoint access permission. /// @@ -42,6 +31,7 @@ pub trait Permission { } /// Type of resource for permissions. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum ResourceType { /// `Channel`-based resource permissions. Channel, @@ -68,7 +58,7 @@ pub enum ResourceType { /// .get() /// .manage(); /// ``` -#[derive(Debug)] +#[derive(Debug, Clone, PartialEq, Eq)] pub struct ChannelPermission { /// Name of channel for which permission level specified. pub name: String, @@ -89,7 +79,7 @@ pub struct ChannelPermission { /// # permissions::channel_group("my-channel-group") /// .read(); /// ``` -#[derive(Debug)] +#[derive(Debug, Clone, PartialEq, Eq)] pub struct ChannelGroupPermission { /// Name of channel group for which permission level specified. pub name: String, @@ -110,7 +100,7 @@ pub struct ChannelGroupPermission { /// # permissions::user_id("my-user-id") /// .update(); /// ``` -#[derive(Debug)] +#[derive(Debug, Clone, PartialEq, Eq)] pub struct UserIdPermission { /// User id for which permission level specified. pub id: String, @@ -131,7 +121,7 @@ pub struct UserIdPermission { /// # permissions::channels_category() /// .get(); /// ``` -#[derive(Debug)] +#[derive(Debug, Clone, PartialEq, Eq)] pub struct ChannelsCategoryPermission { /// Name of channels category name for which permission level specified. name: String, @@ -152,7 +142,7 @@ pub struct ChannelsCategoryPermission { /// # permissions::uuids_category() /// .get(); /// ``` -#[derive(Debug)] +#[derive(Debug, Clone, PartialEq, Eq)] pub struct UuidsCategoryPermission { /// Name of uuids category name for which permission level specified. name: String, diff --git a/src/dx/access/types.rs b/src/dx/access/types.rs index 4572d8f4..f6aadc9a 100644 --- a/src/dx/access/types.rs +++ b/src/dx/access/types.rs @@ -9,6 +9,7 @@ use crate::lib::alloc::string::String; /// Enum for values associated with token. +#[derive(Debug, Clone, PartialEq)] pub enum MetaValue { /// `String` value. String(String), diff --git a/src/dx/parse_token.rs b/src/dx/parse_token.rs index e4f8e87a..b1cc553f 100644 --- a/src/dx/parse_token.rs +++ b/src/dx/parse_token.rs @@ -5,6 +5,7 @@ use crate::core::Deserializer; use crate::{ + core::utils::permission_bits as permissions, core::PubNubError, lib::{ alloc::{ @@ -146,14 +147,14 @@ pub struct TokenCategories { impl From for ResourcePermissions { fn from(int: u8) -> Self { ResourcePermissions { - read: int & 1 != 0, - write: int & 2 != 0, - manage: int & 4 != 0, - delete: int & 8 != 0, - create: int & 16 != 0, - get: int & 32 != 0, - update: int & 64 != 0, - join: int & 128 != 0, + read: int & permissions::READ != 0, + write: int & permissions::WRITE != 0, + manage: int & permissions::MANAGE != 0, + delete: int & permissions::DELETE != 0, + create: int & permissions::CREATE != 0, + get: int & permissions::GET != 0, + update: int & permissions::UPDATE != 0, + join: int & permissions::JOIN != 0, } } } @@ -194,7 +195,9 @@ pub struct ResourcePermissions { impl From for CategoryPermissions { fn from(int: u8) -> Self { - CategoryPermissions { get: int & 32 != 0 } + CategoryPermissions { + get: int & permissions::GET != 0, + } } } diff --git a/src/dx/presence/builders/heartbeat.rs b/src/dx/presence/builders/heartbeat.rs index b8776508..47ff3d25 100644 --- a/src/dx/presence/builders/heartbeat.rs +++ b/src/dx/presence/builders/heartbeat.rs @@ -7,7 +7,6 @@ use derive_builder::Builder; use crate::{ core::{ - blocking, utils::{ encoding::{url_encoded_channel_groups, url_encoded_channels}, headers::{APPLICATION_JSON, CONTENT_TYPE}, @@ -29,6 +28,9 @@ use crate::{ }, }; +#[cfg(feature = "blocking")] +use crate::core::blocking; + /// The [`HeartbeatRequestsBuilder`] is used to build a `user_id` presence /// announcement request that is sent to the [`PubNub`] network. ///