watermelon_proto/
server_error.rs

1use bytestring::ByteString;
2
3#[derive(Debug, PartialEq, Eq, thiserror::Error)]
4pub enum ServerError {
5    #[error("subject is invalid")]
6    InvalidSubject,
7    #[error("permissions violation for publish")]
8    PublishPermissionViolation,
9    #[error("permissions violation for subscription")]
10    SubscribePermissionViolation,
11
12    #[error("unknown protocol operation")]
13    UnknownProtocolOperation,
14
15    #[error("attempted to connect to route port")]
16    ConnectionAttemptedToWrongPort,
17
18    #[error("authorization violation")]
19    AuthorizationViolation,
20    #[error("authorization timeout")]
21    AuthorizationTimeout,
22    #[error("invalid client protocol")]
23    InvalidClientProtocol,
24    #[error("maximum control line exceeded")]
25    MaximumControlLineExceeded,
26    #[error("parser error")]
27    ParseError,
28    #[error("secure connection, tls required")]
29    TlsRequired,
30    #[error("stale connection")]
31    StaleConnection,
32    #[error("maximum connections exceeded")]
33    MaximumConnectionsExceeded,
34    #[error("slow consumer")]
35    SlowConsumer,
36    #[error("maximum payload violation")]
37    MaximumPayloadViolation,
38
39    #[error("unknown error: {raw_message}")]
40    Other { raw_message: ByteString },
41}
42
43impl ServerError {
44    pub fn is_fatal(&self) -> Option<bool> {
45        match self {
46            Self::InvalidSubject
47            | Self::PublishPermissionViolation
48            | Self::SubscribePermissionViolation => Some(false),
49
50            Self::UnknownProtocolOperation
51            | Self::ConnectionAttemptedToWrongPort
52            | Self::AuthorizationViolation
53            | Self::AuthorizationTimeout
54            | Self::InvalidClientProtocol
55            | Self::MaximumControlLineExceeded
56            | Self::ParseError
57            | Self::TlsRequired
58            | Self::StaleConnection
59            | Self::MaximumConnectionsExceeded
60            | Self::SlowConsumer
61            | Self::MaximumPayloadViolation => Some(true),
62
63            Self::Other { .. } => None,
64        }
65    }
66
67    pub(crate) fn parse(raw_message: ByteString) -> Self {
68        const PUBLISH_PERMISSIONS: &[u8] = b"Permissions Violation for Publish";
69        const SUBSCRIPTION_PERMISSIONS: &[u8] = b"Permissions Violation for Subscription";
70
71        // The message comes from the server, so the permission prefixes are
72        // compared byte-wise: a multi-byte character in the prefix region is
73        // just a byte mismatch, i.e. a non-match (`Other` below) — `&str`
74        // indexing at these offsets would panic instead.
75        let m = raw_message.trim();
76        let bytes = m.as_bytes();
77        let matches_prefix = |needle: &[u8]| {
78            bytes.len() > needle.len()
79                && bytes[..needle.len()]
80                    .iter()
81                    .zip(needle)
82                    .all(|(a, b)| a.eq_ignore_ascii_case(b))
83        };
84
85        if m.eq_ignore_ascii_case("Invalid Subject") {
86            Self::InvalidSubject
87        } else if matches_prefix(PUBLISH_PERMISSIONS) {
88            Self::PublishPermissionViolation
89        } else if matches_prefix(SUBSCRIPTION_PERMISSIONS) {
90            Self::SubscribePermissionViolation
91        } else if m.eq_ignore_ascii_case("Unknown Protocol Operation") {
92            Self::UnknownProtocolOperation
93        } else if m.eq_ignore_ascii_case("Attempted To Connect To Route Port") {
94            Self::ConnectionAttemptedToWrongPort
95        } else if m.eq_ignore_ascii_case("Authorization Violation") {
96            Self::AuthorizationViolation
97        } else if m.eq_ignore_ascii_case("Authorization Timeout") {
98            Self::AuthorizationTimeout
99        } else if m.eq_ignore_ascii_case("Invalid Client Protocol") {
100            Self::InvalidClientProtocol
101        } else if m.eq_ignore_ascii_case("Maximum Control Line Exceeded") {
102            Self::MaximumControlLineExceeded
103        } else if m.eq_ignore_ascii_case("Parser Error") {
104            Self::ParseError
105        } else if m.eq_ignore_ascii_case("Secure Connection - TLS Required") {
106            Self::TlsRequired
107        } else if m.eq_ignore_ascii_case("Stale Connection") {
108            Self::StaleConnection
109        } else if m.eq_ignore_ascii_case("Maximum Connections Exceeded") {
110            Self::MaximumConnectionsExceeded
111        } else if m.eq_ignore_ascii_case("Slow Consumer") {
112            Self::SlowConsumer
113        } else if m.eq_ignore_ascii_case("Maximum Payload Violation") {
114            Self::MaximumPayloadViolation
115        } else {
116            Self::Other { raw_message }
117        }
118    }
119}
120
121#[cfg(test)]
122mod tests {
123    use bytestring::ByteString;
124
125    use super::ServerError;
126
127    #[test]
128    fn parse_permission_violations() {
129        assert_eq!(
130            ServerError::parse(ByteString::from_static(
131                "Permissions Violation for Publish for subject \"a.b\"",
132            )),
133            ServerError::PublishPermissionViolation,
134        );
135        assert_eq!(
136            ServerError::parse(ByteString::from_static(
137                "Permissions Violation for Subscription for subject \"a.b\"",
138            )),
139            ServerError::SubscribePermissionViolation,
140        );
141    }
142
143    // A multi-byte character straddling the prefix length must not panic on
144    // byte indexing but be reported as an unknown error.
145    #[test]
146    fn parse_prefix_straddling_char_boundary() {
147        // `€` occupies bytes 32..35, so the 33-byte `Publish` prefix ends
148        // mid-character.
149        let msg = format!("{}€b", "a".repeat(32));
150        assert_eq!(
151            ServerError::parse(msg.clone().into()),
152            ServerError::Other {
153                raw_message: msg.clone().into(),
154            },
155        );
156
157        // Same for the 38-byte `Subscription` prefix.
158        let msg = format!("{}€b", "a".repeat(37));
159        assert_eq!(
160            ServerError::parse(msg.clone().into()),
161            ServerError::Other {
162                raw_message: msg.into(),
163            },
164        );
165    }
166}
watermelon_proto/server_error.rs

watermelon_proto/
server_error.rs
