amq_protocol_uri/
lib.rs

1#![deny(missing_docs, missing_debug_implementations, unsafe_code)]
2#![warn(unreachable_pub, unused_qualifications, unused_lifetimes)]
3#![warn(
4    clippy::must_use_candidate,
5    clippy::unwrap_in_result,
6    clippy::panic_in_result_fn
7)]
8
9//! AMQP 0-9-1 URI parsing and construction.
10//!
11//! Parses `amqp://` and `amqps://` URIs into an [`AMQPUri`] struct and
12//! provides `Display` / `FromStr` implementations for round-tripping.
13
14use amq_protocol_types::{ChannelId, FrameSize, Heartbeat};
15use url::Url;
16
17use std::{fmt, num::ParseIntError, str::FromStr};
18
19/// An AMQP Uri
20#[derive(Clone, Debug, PartialEq, Eq)]
21pub struct AMQPUri {
22    /// The scheme used by the AMQP connection
23    pub scheme: AMQPScheme,
24    /// The connection information
25    pub authority: AMQPAuthority,
26    /// The target vhost
27    pub vhost: String,
28    /// The optional query string to pass parameters to the server
29    pub query: AMQPQueryString,
30}
31
32/// The scheme used by the AMQP connection
33#[derive(Clone, Debug, Default, PartialEq, Eq)]
34pub enum AMQPScheme {
35    /// Plain AMQP
36    #[default]
37    AMQP,
38    /// Encrypted AMQP over TLS
39    AMQPS,
40}
41
42impl FromStr for AMQPScheme {
43    type Err = String;
44
45    fn from_str(s: &str) -> Result<Self, Self::Err> {
46        match s {
47            "amqp" => Ok(AMQPScheme::AMQP),
48            "amqps" => Ok(AMQPScheme::AMQPS),
49            s => Err(format!("Invalid AMQP scheme: {s}")),
50        }
51    }
52}
53
54impl fmt::Display for AMQPScheme {
55    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
56        f.write_str(match self {
57            AMQPScheme::AMQP => "amqp",
58            AMQPScheme::AMQPS => "amqps",
59        })
60    }
61}
62
63/// The connection information
64#[derive(Clone, Debug, PartialEq, Eq)]
65pub struct AMQPAuthority {
66    /// The credentials used to connect to the server
67    pub userinfo: AMQPUserInfo,
68    /// The server's host
69    pub host: String,
70    /// The port the server listens on
71    pub port: u16,
72}
73
74/// The credentials used to connect to the server
75#[derive(Clone, Debug, PartialEq, Eq)]
76pub struct AMQPUserInfo {
77    /// The username
78    pub username: String,
79    /// The password
80    pub password: String,
81}
82
83/// The optional query string to pass parameters to the server
84#[derive(Clone, Debug, Default, PartialEq, Eq)]
85pub struct AMQPQueryString {
86    /// The maximum size of an AMQP Frame
87    pub frame_max: Option<FrameSize>,
88    /// The maximum number of open channels
89    pub channel_max: Option<ChannelId>,
90    /// The maximum time between two heartbeats
91    pub heartbeat: Option<Heartbeat>,
92    /// The maximum time to wait (in milliseconds) for the connection to succeed
93    pub connection_timeout: Option<u64>,
94    /// The SASL mechanism used for authentication
95    pub auth_mechanism: Option<SASLMechanism>,
96    // Fields available in Erlang implementation for SSL settings:
97    // cacertfile, certfile, keyfile, verify, fail_if_no_peer_cert, password,
98    // server_name_indication, depth
99}
100
101/// The SASL mechanisms supported by RabbitMQ
102#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
103pub enum SASLMechanism {
104    /// This is a legacy mechanism kept for backward compatibility
105    AMQPlain,
106    /// Anonymous authentication if supported by the RabbitMQ server
107    Anonymous,
108    /// Delegate all authentication to the transport instead of the RabbitMQ server
109    External,
110    /// Default plain login, this should be supported everywhere
111    #[default]
112    Plain,
113    /// A demo of RabbitMQ SecureOk mechanism, offers the same level of security as Plain
114    RabbitCrDemo,
115}
116
117impl SASLMechanism {
118    /// Get the name of the SASL mechanism as str
119    #[must_use]
120    pub fn name(&self) -> &'static str {
121        match self {
122            SASLMechanism::AMQPlain => "AMQPLAIN",
123            SASLMechanism::Anonymous => "ANONYMOUS",
124            SASLMechanism::External => "EXTERNAL",
125            SASLMechanism::Plain => "PLAIN",
126            SASLMechanism::RabbitCrDemo => "RABBIT-CR-DEMO",
127        }
128    }
129}
130
131impl fmt::Display for SASLMechanism {
132    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
133        f.write_str(self.name())
134    }
135}
136
137impl FromStr for SASLMechanism {
138    type Err = String;
139
140    fn from_str(s: &str) -> Result<Self, Self::Err> {
141        match s.to_lowercase().as_str() {
142            "amqplain" => Ok(SASLMechanism::AMQPlain),
143            "anonymous" => Ok(SASLMechanism::Anonymous),
144            "external" => Ok(SASLMechanism::External),
145            "plain" => Ok(SASLMechanism::Plain),
146            "rabbit-cr-demo" => Ok(SASLMechanism::RabbitCrDemo),
147            s => Err(format!("Invalid SASL mechanism: {s}")),
148        }
149    }
150}
151
152fn percent_decode(s: &str) -> Result<String, String> {
153    percent_encoding::percent_decode(s.as_bytes())
154        .decode_utf8()
155        .map(|s| s.to_string())
156        .map_err(|e| e.to_string())
157}
158
159fn percent_encode<'a>(s: &'a str) -> percent_encoding::PercentEncode<'a> {
160    percent_encoding::utf8_percent_encode(s, percent_encoding::NON_ALPHANUMERIC)
161}
162
163impl Default for AMQPUri {
164    fn default() -> Self {
165        AMQPUri {
166            scheme: Default::default(),
167            authority: Default::default(),
168            vhost: "/".to_string(),
169            query: Default::default(),
170        }
171    }
172}
173
174fn int_queryparam<T: FromStr<Err = ParseIntError>>(
175    url: &Url,
176    param: &str,
177) -> Result<Option<T>, String> {
178    url.query_pairs()
179        .find(|(key, _)| key == param)
180        .map_or(Ok(None), |(_, ref value)| value.parse::<T>().map(Some))
181        .map_err(|e: ParseIntError| e.to_string())
182}
183
184impl FromStr for AMQPUri {
185    type Err = String;
186
187    fn from_str(s: &str) -> Result<Self, Self::Err> {
188        let url = Url::parse(s).map_err(|e| e.to_string())?;
189        if url.cannot_be_a_base() {
190            return Err(format!("Invalid URL: '{s}'"));
191        }
192        let default = AMQPUri::default();
193        let scheme = url.scheme().parse::<AMQPScheme>()?;
194        let username = match url.username() {
195            "" => default.authority.userinfo.username,
196            username => percent_decode(username)?,
197        };
198        let password = url
199            .password()
200            .map_or(Ok(default.authority.userinfo.password), percent_decode)?;
201        let host = url
202            .domain()
203            .map_or(Ok(default.authority.host), percent_decode)?;
204        let port = url.port().unwrap_or_else(|| scheme.default_port());
205        let vhost = percent_decode(url.path().get(1..).unwrap_or("/"))?;
206        let frame_max = int_queryparam(&url, "frame_max")?;
207        let channel_max = int_queryparam(&url, "channel_max")?;
208        let heartbeat = int_queryparam(&url, "heartbeat")?;
209        let connection_timeout = int_queryparam(&url, "connection_timeout")?;
210        let auth_mechanism = url
211            .query_pairs()
212            .find(|(key, _)| key == "auth_mechanism")
213            .map_or(Ok(None), |(_, ref value)| value.parse().map(Some))?;
214
215        Ok(AMQPUri {
216            scheme,
217            authority: AMQPAuthority {
218                userinfo: AMQPUserInfo { username, password },
219                host,
220                port,
221            },
222            vhost,
223            query: AMQPQueryString {
224                frame_max,
225                channel_max,
226                heartbeat,
227                connection_timeout,
228                auth_mechanism,
229            },
230        })
231    }
232}
233
234impl fmt::Display for AMQPUri {
235    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
236        write!(
237            f,
238            "{}://{}:{}@{}:{}/{}",
239            self.scheme,
240            percent_encode(&self.authority.userinfo.username),
241            percent_encode(&self.authority.userinfo.password),
242            self.authority.host,
243            self.authority.port,
244            percent_encode(&self.vhost),
245        )?;
246        let mut sep = '?';
247        if let Some(v) = self.query.frame_max {
248            write!(f, "{sep}frame_max={v}")?;
249            sep = '&';
250        }
251        if let Some(v) = self.query.channel_max {
252            write!(f, "{sep}channel_max={v}")?;
253            sep = '&';
254        }
255        if let Some(v) = self.query.heartbeat {
256            write!(f, "{sep}heartbeat={v}")?;
257            sep = '&';
258        }
259        if let Some(v) = self.query.connection_timeout {
260            write!(f, "{sep}connection_timeout={v}")?;
261            sep = '&';
262        }
263        if let Some(v) = self.query.auth_mechanism {
264            write!(f, "{sep}auth_mechanism={v}")?;
265        }
266        Ok(())
267    }
268}
269
270impl AMQPScheme {
271    /// The default port for this scheme
272    #[must_use]
273    pub fn default_port(&self) -> u16 {
274        match *self {
275            AMQPScheme::AMQP => 5672,
276            AMQPScheme::AMQPS => 5671,
277        }
278    }
279}
280
281impl Default for AMQPAuthority {
282    fn default() -> Self {
283        AMQPAuthority {
284            userinfo: Default::default(),
285            host: "localhost".to_string(),
286            port: AMQPScheme::default().default_port(),
287        }
288    }
289}
290
291impl Default for AMQPUserInfo {
292    fn default() -> Self {
293        AMQPUserInfo {
294            username: "guest".to_string(),
295            password: "guest".to_string(),
296        }
297    }
298}
299
300#[cfg(test)]
301mod test {
302    use super::*;
303
304    #[test]
305    fn test_parse_amqp_no_path() {
306        let uri = "amqp://localhost".parse();
307        assert_eq!(uri, Ok(AMQPUri::default()));
308    }
309
310    #[test]
311    fn test_parse_amqp() {
312        let uri = "amqp://localhost/%2f".parse();
313        assert_eq!(uri, Ok(AMQPUri::default()));
314    }
315
316    #[test]
317    fn test_parse_amqps() {
318        let uri = "amqps://localhost/".parse();
319        assert_eq!(
320            uri,
321            Ok(AMQPUri {
322                scheme: AMQPScheme::AMQPS,
323                authority: AMQPAuthority {
324                    port: 5671,
325                    ..Default::default()
326                },
327                vhost: "".to_string(),
328                ..Default::default()
329            })
330        );
331    }
332
333    #[test]
334    fn test_parse_amqps_with_creds() {
335        let uri = "amqps://user:pass@hostname/v?foo=bar".parse();
336        assert_eq!(
337            uri,
338            Ok(AMQPUri {
339                scheme: AMQPScheme::AMQPS,
340                authority: AMQPAuthority {
341                    userinfo: AMQPUserInfo {
342                        username: "user".to_string(),
343                        password: "pass".to_string(),
344                    },
345                    host: "hostname".to_string(),
346                    port: 5671,
347                },
348                vhost: "v".to_string(),
349                ..Default::default()
350            })
351        );
352    }
353
354    #[test]
355    fn test_parse_amqps_with_creds_percent() {
356        let uri = "amqp://user%61:%61pass@ho%61st:10000/v%2fhost".parse();
357        assert_eq!(
358            uri,
359            Ok(AMQPUri {
360                scheme: AMQPScheme::AMQP,
361                authority: AMQPAuthority {
362                    userinfo: AMQPUserInfo {
363                        username: "usera".to_string(),
364                        password: "apass".to_string(),
365                    },
366                    host: "hoast".to_string(),
367                    port: 10000,
368                },
369                vhost: "v/host".to_string(),
370                ..Default::default()
371            })
372        );
373    }
374
375    #[test]
376    fn test_parse_with_heartbeat_frame_max() {
377        let uri = "amqp://localhost/%2f?heartbeat=42&frame_max=64&connection_timeout=30000".parse();
378        assert_eq!(
379            uri,
380            Ok(AMQPUri {
381                query: AMQPQueryString {
382                    frame_max: Some(64),
383                    heartbeat: Some(42),
384                    connection_timeout: Some(30000),
385                    ..Default::default()
386                },
387                ..Default::default()
388            })
389        );
390    }
391
392    #[test]
393    fn test_url_with_no_base() {
394        let uri: Result<AMQPUri, String> = "foo".parse();
395        assert_eq!(uri, Err("relative URL without a base".to_string()));
396    }
397
398    #[test]
399    fn test_invalid_url() {
400        let uri: Result<AMQPUri, String> = "foo:bar".parse();
401        assert_eq!(uri, Err("Invalid URL: 'foo:bar'".to_string()));
402    }
403
404    #[test]
405    fn test_invalid_scheme() {
406        let uri: Result<AMQPUri, String> = "http://localhost/".parse();
407        assert_eq!(uri, Err("Invalid AMQP scheme: http".to_string()));
408    }
409}
amq_protocol_uri/lib.rs

amq_protocol_uri/
lib.rs
