| use super::constants::{ |
| MANAGE_STREAM_LONG, MANAGE_STREAM_SHORT, MANAGE_TOPICS_LONG, MANAGE_TOPICS_SHORT, |
| POLL_MESSAGES_LONG, POLL_MESSAGES_SHORT, READ_STREAM_LONG, READ_STREAM_SHORT, READ_TOPICS_LONG, |
| READ_TOPICS_SHORT, SEND_MESSAGES_LONG, SEND_MESSAGES_SHORT, |
| }; |
| use crate::args::permissions::topic::TopicPermissionsArg; |
| use ahash::AHashMap; |
| use iggy::models::permissions::StreamPermissions; |
| use std::str::FromStr; |
| |
| #[derive(Debug, PartialEq)] |
| pub(super) enum StreamPermission { |
| ManageStream, |
| ReadStream, |
| ManageTopics, |
| ReadTopics, |
| PollMessages, |
| SendMessages, |
| } |
| |
| #[derive(Clone, Debug, PartialEq)] |
| pub(crate) struct StreamPermissionError(String); |
| |
| impl FromStr for StreamPermission { |
| type Err = StreamPermissionError; |
| |
| fn from_str(s: &str) -> Result<Self, Self::Err> { |
| match s { |
| MANAGE_STREAM_SHORT | MANAGE_STREAM_LONG => Ok(StreamPermission::ManageStream), |
| READ_STREAM_SHORT | READ_STREAM_LONG => Ok(StreamPermission::ReadStream), |
| MANAGE_TOPICS_SHORT | MANAGE_TOPICS_LONG => Ok(StreamPermission::ManageTopics), |
| READ_TOPICS_SHORT | READ_TOPICS_LONG => Ok(StreamPermission::ReadTopics), |
| POLL_MESSAGES_SHORT | POLL_MESSAGES_LONG => Ok(StreamPermission::PollMessages), |
| SEND_MESSAGES_SHORT | SEND_MESSAGES_LONG => Ok(StreamPermission::SendMessages), |
| "" => Err(StreamPermissionError("[empty]".to_owned())), |
| _ => Err(StreamPermissionError(s.to_owned())), |
| } |
| } |
| } |
| |
| #[derive(Clone, Debug, PartialEq)] |
| pub(crate) struct StreamPermissionsArg { |
| pub(crate) stream_id: u32, |
| pub(crate) permissions: StreamPermissions, |
| } |
| |
| impl From<StreamPermissionsArg> for StreamPermissions { |
| fn from(cmd: StreamPermissionsArg) -> Self { |
| cmd.permissions |
| } |
| } |
| |
| impl StreamPermissionsArg { |
| pub(super) fn new( |
| stream_id: u32, |
| stream_permissions: Vec<StreamPermission>, |
| topic_permissions: Vec<TopicPermissionsArg>, |
| ) -> Self { |
| let mut result = Self { |
| stream_id, |
| permissions: StreamPermissions::default(), |
| }; |
| |
| for permission in stream_permissions { |
| result.set_permission(permission); |
| } |
| |
| if !topic_permissions.is_empty() { |
| let mut permissions = AHashMap::new(); |
| for permission in topic_permissions { |
| permissions.insert(permission.topic_id, permission.permissions); |
| } |
| |
| result.permissions.topics = Some(permissions); |
| } |
| |
| result |
| } |
| |
| fn set_permission(&mut self, permission: StreamPermission) { |
| match permission { |
| StreamPermission::ManageStream => self.permissions.manage_stream = true, |
| StreamPermission::ReadStream => self.permissions.read_stream = true, |
| StreamPermission::ManageTopics => self.permissions.manage_topics = true, |
| StreamPermission::ReadTopics => self.permissions.read_topics = true, |
| StreamPermission::PollMessages => self.permissions.poll_messages = true, |
| StreamPermission::SendMessages => self.permissions.send_messages = true, |
| } |
| } |
| } |
| |
| impl FromStr for StreamPermissionsArg { |
| type Err = String; |
| |
| fn from_str(s: &str) -> Result<Self, Self::Err> { |
| let mut p = s.split('#'); |
| |
| let stream_part = match p.next() { |
| Some(part) => part, |
| None => return Err("Missing stream permissions part".to_string()), |
| }; |
| |
| let mut parts = stream_part.split(':'); |
| let stream_id = parts |
| .next() |
| .ok_or("Missing stream ID".to_string()) |
| .and_then(|id| { |
| id.parse() |
| .map_err(|error| format!("Invalid stream ID - {}", error)) |
| })?; |
| |
| let (stream_permissions, stream_errors): (Vec<StreamPermission>, Option<String>) = |
| match parts.next() { |
| Some(permissions_str) => { |
| let (values, errors): (Vec<_>, Vec<_>) = permissions_str |
| .split(',') |
| .map(|s| s.parse::<StreamPermission>()) |
| .partition(Result::is_ok); |
| |
| let error = if !errors.is_empty() { |
| let errors = errors |
| .into_iter() |
| .map(|e| format!("\"{}\"", e.err().unwrap().0)) |
| .collect::<Vec<String>>(); |
| |
| Some(format!( |
| "Unknown permission{} {} for stream ID: {}", |
| match errors.len() { |
| 1 => "", |
| _ => "s", |
| }, |
| errors.join(", "), |
| stream_id |
| )) |
| } else { |
| None |
| }; |
| |
| (values.into_iter().map(|p| p.unwrap()).collect(), error) |
| } |
| None => (vec![], None), |
| }; |
| |
| let (topic_permissions, topic_errors): (Vec<TopicPermissionsArg>, Option<String>) = { |
| let (permissions, errors): (Vec<_>, Vec<_>) = p |
| .map(|s| s.parse::<TopicPermissionsArg>()) |
| .partition(Result::is_ok); |
| |
| let errors = if !errors.is_empty() { |
| Some( |
| errors |
| .into_iter() |
| .map(|e| e.err().unwrap()) |
| .collect::<Vec<String>>() |
| .join("; "), |
| ) |
| } else { |
| None |
| }; |
| |
| ( |
| permissions.into_iter().map(|p| p.unwrap()).collect(), |
| errors, |
| ) |
| }; |
| |
| match (stream_errors, topic_errors) { |
| (Some(e), Some(f)) => return Err(format!("{}; {}", e, f)), |
| (Some(e), None) => return Err(e), |
| (None, Some(f)) => return Err(f), |
| (None, None) => (), |
| } |
| |
| Ok(StreamPermissionsArg::new( |
| stream_id, |
| stream_permissions, |
| topic_permissions, |
| )) |
| } |
| } |
| |
| #[cfg(test)] |
| mod tests { |
| use iggy::models::permissions::TopicPermissions; |
| |
| use super::*; |
| |
| #[test] |
| fn should_deserialize_single_permission() { |
| assert_eq!( |
| StreamPermission::from_str("manage_stream").unwrap(), |
| StreamPermission::ManageStream |
| ); |
| assert_eq!( |
| StreamPermission::from_str("read_stream").unwrap(), |
| StreamPermission::ReadStream |
| ); |
| assert_eq!( |
| StreamPermission::from_str("manage_topics").unwrap(), |
| StreamPermission::ManageTopics |
| ); |
| assert_eq!( |
| StreamPermission::from_str("read_topics").unwrap(), |
| StreamPermission::ReadTopics |
| ); |
| assert_eq!( |
| StreamPermission::from_str("poll_messages").unwrap(), |
| StreamPermission::PollMessages |
| ); |
| assert_eq!( |
| StreamPermission::from_str("send_messages").unwrap(), |
| StreamPermission::SendMessages |
| ); |
| } |
| |
| #[test] |
| fn should_deserialize_single_short_permission() { |
| assert_eq!( |
| StreamPermission::from_str("m_str").unwrap(), |
| StreamPermission::ManageStream |
| ); |
| assert_eq!( |
| StreamPermission::from_str("r_str").unwrap(), |
| StreamPermission::ReadStream |
| ); |
| assert_eq!( |
| StreamPermission::from_str("m_top").unwrap(), |
| StreamPermission::ManageTopics |
| ); |
| assert_eq!( |
| StreamPermission::from_str("r_top").unwrap(), |
| StreamPermission::ReadTopics |
| ); |
| assert_eq!( |
| StreamPermission::from_str("p_msg").unwrap(), |
| StreamPermission::PollMessages |
| ); |
| assert_eq!( |
| StreamPermission::from_str("s_msg").unwrap(), |
| StreamPermission::SendMessages |
| ); |
| } |
| |
| #[test] |
| fn should_not_deserialize_single_permission() { |
| let wrong_permission = StreamPermission::from_str("read_topic"); |
| assert!(wrong_permission.is_err()); |
| assert_eq!( |
| wrong_permission.unwrap_err(), |
| StreamPermissionError("read_topic".to_owned()) |
| ); |
| let empty_permission = StreamPermission::from_str(""); |
| assert!(empty_permission.is_err()); |
| assert_eq!( |
| empty_permission.unwrap_err(), |
| StreamPermissionError("[empty]".to_owned()) |
| ); |
| } |
| |
| #[test] |
| fn should_not_deserialize_single_short_permission() { |
| let wrong_permission = StreamPermission::from_str("w_top"); |
| assert!(wrong_permission.is_err()); |
| assert_eq!( |
| wrong_permission.unwrap_err(), |
| StreamPermissionError("w_top".to_owned()) |
| ); |
| let wrong_permission = StreamPermission::from_str("m_msg"); |
| assert!(wrong_permission.is_err()); |
| assert_eq!( |
| wrong_permission.unwrap_err(), |
| StreamPermissionError("m_msg".to_owned()) |
| ); |
| } |
| |
| #[test] |
| fn should_deserialize_permissions() { |
| assert_eq!( |
| StreamPermissionsArg::from_str( |
| "1:manage_stream,read_stream,manage_topics,read_topics,poll_messages,send_messages" |
| ) |
| .unwrap(), |
| StreamPermissionsArg { |
| stream_id: 1, |
| permissions: StreamPermissions { |
| manage_stream: true, |
| read_stream: true, |
| manage_topics: true, |
| read_topics: true, |
| poll_messages: true, |
| send_messages: true, |
| topics: None, |
| } |
| } |
| ); |
| assert_eq!( |
| StreamPermissionsArg::from_str("1:manage_topics,read_topics").unwrap(), |
| StreamPermissionsArg { |
| stream_id: 1, |
| permissions: StreamPermissions { |
| manage_stream: false, |
| read_stream: false, |
| manage_topics: true, |
| read_topics: true, |
| poll_messages: false, |
| send_messages: false, |
| topics: None, |
| } |
| } |
| ); |
| assert_eq!( |
| StreamPermissionsArg::from_str("39:send_messages,read_topics,read_stream").unwrap(), |
| StreamPermissionsArg { |
| stream_id: 39, |
| permissions: StreamPermissions { |
| manage_stream: false, |
| read_stream: true, |
| manage_topics: false, |
| read_topics: true, |
| poll_messages: false, |
| send_messages: true, |
| topics: None, |
| } |
| } |
| ); |
| assert_eq!( |
| StreamPermissionsArg::from_str("71").unwrap(), |
| StreamPermissionsArg { |
| stream_id: 71, |
| permissions: StreamPermissions { |
| manage_stream: false, |
| read_stream: false, |
| manage_topics: false, |
| read_topics: false, |
| poll_messages: false, |
| send_messages: false, |
| topics: None, |
| } |
| } |
| ); |
| assert_eq!( |
| StreamPermissionsArg::from_str("9:send_messages").unwrap(), |
| StreamPermissionsArg { |
| stream_id: 9, |
| permissions: StreamPermissions { |
| manage_stream: false, |
| read_stream: false, |
| manage_topics: false, |
| read_topics: false, |
| poll_messages: false, |
| send_messages: true, |
| topics: None, |
| } |
| } |
| ); |
| assert_eq!( |
| StreamPermissionsArg::from_str("9#1#2").unwrap(), |
| StreamPermissionsArg { |
| stream_id: 9, |
| permissions: StreamPermissions { |
| manage_stream: false, |
| read_stream: false, |
| manage_topics: false, |
| read_topics: false, |
| poll_messages: false, |
| send_messages: false, |
| topics: Some(AHashMap::from([ |
| ( |
| 2, |
| TopicPermissions { |
| manage_topic: false, |
| read_topic: false, |
| poll_messages: false, |
| send_messages: false, |
| } |
| ), |
| ( |
| 1, |
| TopicPermissions { |
| manage_topic: false, |
| read_topic: false, |
| poll_messages: false, |
| send_messages: false, |
| } |
| ) |
| ])), |
| } |
| } |
| ); |
| assert_eq!( |
| StreamPermissionsArg::from_str("4:manage_topics#1:manage_topic#2:manage_topic") |
| .unwrap(), |
| StreamPermissionsArg { |
| stream_id: 4, |
| permissions: StreamPermissions { |
| manage_stream: false, |
| read_stream: false, |
| manage_topics: true, |
| read_topics: false, |
| poll_messages: false, |
| send_messages: false, |
| topics: Some(AHashMap::from([ |
| ( |
| 2, |
| TopicPermissions { |
| manage_topic: true, |
| read_topic: false, |
| poll_messages: false, |
| send_messages: false, |
| } |
| ), |
| ( |
| 1, |
| TopicPermissions { |
| manage_topic: true, |
| read_topic: false, |
| poll_messages: false, |
| send_messages: false, |
| } |
| ) |
| ])), |
| } |
| } |
| ); |
| } |
| |
| #[test] |
| fn should_deserialize_short_permissions() { |
| assert_eq!( |
| StreamPermissionsArg::from_str("111:m_str,r_str,m_top,r_top,p_msg,s_msg").unwrap(), |
| StreamPermissionsArg { |
| stream_id: 111, |
| permissions: StreamPermissions { |
| manage_stream: true, |
| read_stream: true, |
| manage_topics: true, |
| read_topics: true, |
| poll_messages: true, |
| send_messages: true, |
| topics: None, |
| } |
| } |
| ); |
| assert_eq!( |
| StreamPermissionsArg::from_str("27:m_top,r_top").unwrap(), |
| StreamPermissionsArg { |
| stream_id: 27, |
| permissions: StreamPermissions { |
| manage_stream: false, |
| read_stream: false, |
| manage_topics: true, |
| read_topics: true, |
| poll_messages: false, |
| send_messages: false, |
| topics: None, |
| } |
| } |
| ); |
| assert_eq!( |
| StreamPermissionsArg::from_str("9:s_msg,r_top,r_str").unwrap(), |
| StreamPermissionsArg { |
| stream_id: 9, |
| permissions: StreamPermissions { |
| manage_stream: false, |
| read_stream: true, |
| manage_topics: false, |
| read_topics: true, |
| poll_messages: false, |
| send_messages: true, |
| topics: None, |
| } |
| } |
| ); |
| assert_eq!( |
| StreamPermissionsArg::from_str("4:s_msg").unwrap(), |
| StreamPermissionsArg { |
| stream_id: 4, |
| permissions: StreamPermissions { |
| manage_stream: false, |
| read_stream: false, |
| manage_topics: false, |
| read_topics: false, |
| poll_messages: false, |
| send_messages: true, |
| topics: None, |
| } |
| } |
| ); |
| assert_eq!( |
| StreamPermissionsArg::from_str("6:m_top#1:m_top#2:m_top").unwrap(), |
| StreamPermissionsArg { |
| stream_id: 6, |
| permissions: StreamPermissions { |
| manage_stream: false, |
| read_stream: false, |
| manage_topics: true, |
| read_topics: false, |
| poll_messages: false, |
| send_messages: false, |
| topics: Some(AHashMap::from([ |
| ( |
| 2, |
| TopicPermissions { |
| manage_topic: true, |
| read_topic: false, |
| poll_messages: false, |
| send_messages: false, |
| } |
| ), |
| ( |
| 1, |
| TopicPermissions { |
| manage_topic: true, |
| read_topic: false, |
| poll_messages: false, |
| send_messages: false, |
| } |
| ) |
| ])), |
| } |
| } |
| ); |
| } |
| |
| #[test] |
| fn should_not_deserialize_permissions() { |
| let wrong_id = StreamPermissionsArg::from_str("4a"); |
| assert!(wrong_id.is_err()); |
| assert_eq!( |
| wrong_id.unwrap_err(), |
| "Invalid stream ID - invalid digit found in string" |
| ); |
| let wrong_permission = StreamPermissionsArg::from_str("4:read_topic"); |
| assert!(wrong_permission.is_err()); |
| assert_eq!( |
| wrong_permission.unwrap_err(), |
| "Unknown permission \"read_topic\" for stream ID: 4" |
| ); |
| let multiple_wrong = StreamPermissionsArg::from_str("55:read_topic,sent_messages"); |
| assert!(multiple_wrong.is_err()); |
| assert_eq!( |
| multiple_wrong.unwrap_err(), |
| "Unknown permissions \"read_topic\", \"sent_messages\" for stream ID: 55" |
| ); |
| let multiple_wrong_with_topic = |
| StreamPermissionsArg::from_str("55:read_topic,sent_messages#4:reed_topic"); |
| assert!(multiple_wrong_with_topic.is_err()); |
| assert_eq!( |
| multiple_wrong_with_topic.unwrap_err(), |
| "Unknown permissions \"read_topic\", \"sent_messages\" for stream ID: 55; Unknown permission \"reed_topic\" for topic ID: 4" |
| ); |
| } |
| |
| #[test] |
| fn should_not_deserialize_short_permissions() { |
| let wrong_id = StreamPermissionsArg::from_str("1x"); |
| assert!(wrong_id.is_err()); |
| assert_eq!( |
| wrong_id.unwrap_err(), |
| "Invalid stream ID - invalid digit found in string" |
| ); |
| let wrong_permission = StreamPermissionsArg::from_str("7:x_topic"); |
| assert!(wrong_permission.is_err()); |
| assert_eq!( |
| wrong_permission.unwrap_err(), |
| "Unknown permission \"x_topic\" for stream ID: 7" |
| ); |
| let multiple_wrong = StreamPermissionsArg::from_str("37:w_top,s_msgs"); |
| assert!(multiple_wrong.is_err()); |
| assert_eq!( |
| multiple_wrong.unwrap_err(), |
| "Unknown permissions \"w_top\", \"s_msgs\" for stream ID: 37" |
| ); |
| let multiple_wrong_with_topic = StreamPermissionsArg::from_str("11:r_to,s_msg#4:r_to"); |
| assert!(multiple_wrong_with_topic.is_err()); |
| assert_eq!( |
| multiple_wrong_with_topic.unwrap_err(), |
| "Unknown permission \"r_to\" for stream ID: 11; Unknown permission \"r_to\" for topic ID: 4" |
| ); |
| let multiple_wrong_topics = |
| StreamPermissionsArg::from_str("11:r_top,s_msg#7:r_tox#5:w_top,s_top"); |
| assert!(multiple_wrong_topics.is_err()); |
| assert_eq!( |
| multiple_wrong_topics.unwrap_err(), |
| "Unknown permission \"r_tox\" for topic ID: 7; Unknown permissions \"w_top\", \"s_top\" for topic ID: 5" |
| ); |
| } |
| } |
