kafrust_protocol/api/
sasl.rs

1use crate::codec::{Decoder, Encoder};
2use crate::error::Result;
3use crate::header::RequestHeader;
4
5pub const SASL_HANDSHAKE_API_KEY: i16 = 17;
6pub const SASL_AUTHENTICATE_API_KEY: i16 = 36;
7
8#[derive(Debug, Clone, PartialEq, Eq)]
9pub struct SaslHandshakeRequestV1 {
10    pub correlation_id: i32,
11    pub client_id: Option<String>,
12    pub mechanism: String,
13}
14
15impl SaslHandshakeRequestV1 {
16    pub fn encode(&self) -> Result<Vec<u8>> {
17        let mut encoder = Encoder::new();
18        RequestHeader {
19            api_key: SASL_HANDSHAKE_API_KEY,
20            api_version: 1,
21            correlation_id: self.correlation_id,
22            client_id: self.client_id.clone(),
23        }
24        .encode_v1(&mut encoder)?;
25        encoder.write_string(&self.mechanism)?;
26        Ok(encoder.into_bytes())
27    }
28}
29
30#[derive(Debug, Clone, PartialEq, Eq)]
31pub struct SaslHandshakeResponseV1 {
32    pub error_code: i16,
33    pub mechanisms: Vec<String>,
34}
35
36impl SaslHandshakeResponseV1 {
37    pub fn decode_body(decoder: &mut Decoder<'_>) -> Result<Self> {
38        let error_code = decoder.read_i16()?;
39        let mechanisms = decoder
40            .read_array("SASL mechanisms", |decoder| decoder.read_string())?
41            .unwrap_or_default();
42        Ok(Self {
43            error_code,
44            mechanisms,
45        })
46    }
47}
48
49#[derive(Debug, Clone, PartialEq, Eq)]
50pub struct SaslAuthenticateRequestV0 {
51    pub correlation_id: i32,
52    pub client_id: Option<String>,
53    pub auth_bytes: Vec<u8>,
54}
55
56#[derive(Debug, Clone, PartialEq, Eq)]
57pub struct SaslAuthenticateRequestV1 {
58    pub correlation_id: i32,
59    pub client_id: Option<String>,
60    pub auth_bytes: Vec<u8>,
61}
62
63impl SaslAuthenticateRequestV1 {
64    pub fn encode(&self) -> Result<Vec<u8>> {
65        let mut encoder = Encoder::new();
66        RequestHeader {
67            api_key: SASL_AUTHENTICATE_API_KEY,
68            api_version: 1,
69            correlation_id: self.correlation_id,
70            client_id: self.client_id.clone(),
71        }
72        .encode_v1(&mut encoder)?;
73        encoder.write_bytes(&self.auth_bytes)?;
74        Ok(encoder.into_bytes())
75    }
76}
77
78#[derive(Debug, Clone, PartialEq, Eq)]
79pub struct SaslAuthenticateRequestV2 {
80    pub correlation_id: i32,
81    pub client_id: Option<String>,
82    pub auth_bytes: Vec<u8>,
83}
84
85impl SaslAuthenticateRequestV2 {
86    pub fn encode(&self) -> Result<Vec<u8>> {
87        let mut encoder = Encoder::new();
88        RequestHeader {
89            api_key: SASL_AUTHENTICATE_API_KEY,
90            api_version: 2,
91            correlation_id: self.correlation_id,
92            client_id: self.client_id.clone(),
93        }
94        .encode_v2(&mut encoder)?;
95        encoder.write_compact_bytes(&self.auth_bytes)?;
96        encoder.write_empty_tagged_fields();
97        Ok(encoder.into_bytes())
98    }
99}
100
101impl SaslAuthenticateRequestV0 {
102    pub fn encode(&self) -> Result<Vec<u8>> {
103        let mut encoder = Encoder::new();
104        RequestHeader {
105            api_key: SASL_AUTHENTICATE_API_KEY,
106            api_version: 0,
107            correlation_id: self.correlation_id,
108            client_id: self.client_id.clone(),
109        }
110        .encode_v1(&mut encoder)?;
111        encoder.write_bytes(&self.auth_bytes)?;
112        Ok(encoder.into_bytes())
113    }
114}
115
116#[derive(Debug, Clone, PartialEq, Eq)]
117pub struct SaslAuthenticateResponseV0 {
118    pub error_code: i16,
119    pub error_message: Option<String>,
120    pub auth_bytes: Vec<u8>,
121}
122
123#[derive(Debug, Clone, PartialEq, Eq)]
124pub struct SaslAuthenticateResponseV1 {
125    pub error_code: i16,
126    pub error_message: Option<String>,
127    pub auth_bytes: Vec<u8>,
128    pub session_lifetime_ms: i64,
129}
130
131impl SaslAuthenticateResponseV1 {
132    pub fn decode_body(decoder: &mut Decoder<'_>) -> Result<Self> {
133        Ok(Self {
134            error_code: decoder.read_i16()?,
135            error_message: decoder.read_nullable_string()?,
136            auth_bytes: decoder.read_bytes()?,
137            session_lifetime_ms: decoder.read_i64()?,
138        })
139    }
140}
141
142#[derive(Debug, Clone, PartialEq, Eq)]
143pub struct SaslAuthenticateResponseV2 {
144    pub error_code: i16,
145    pub error_message: Option<String>,
146    pub auth_bytes: Vec<u8>,
147    pub session_lifetime_ms: i64,
148}
149
150impl SaslAuthenticateResponseV2 {
151    pub fn decode_body(decoder: &mut Decoder<'_>) -> Result<Self> {
152        let response = Self {
153            error_code: decoder.read_i16()?,
154            error_message: decoder.read_compact_nullable_string()?,
155            auth_bytes: decoder.read_compact_bytes()?,
156            session_lifetime_ms: decoder.read_i64()?,
157        };
158        decoder.read_tagged_fields()?;
159        Ok(response)
160    }
161}
162
163impl SaslAuthenticateResponseV0 {
164    pub fn decode_body(decoder: &mut Decoder<'_>) -> Result<Self> {
165        Ok(Self {
166            error_code: decoder.read_i16()?,
167            error_message: decoder.read_nullable_string()?,
168            auth_bytes: decoder.read_bytes()?,
169        })
170    }
171}
172
173#[cfg(test)]
174#[allow(clippy::unwrap_used)]
175mod tests {
176    use super::{
177        SaslAuthenticateRequestV0, SaslAuthenticateRequestV1, SaslAuthenticateRequestV2,
178        SaslAuthenticateResponseV0, SaslAuthenticateResponseV1, SaslAuthenticateResponseV2,
179        SaslHandshakeRequestV1, SaslHandshakeResponseV1, SASL_AUTHENTICATE_API_KEY,
180        SASL_HANDSHAKE_API_KEY,
181    };
182    use crate::codec::Decoder;
183
184    #[test]
185    fn encodes_sasl_handshake_v1_request() {
186        let request = SaslHandshakeRequestV1 {
187            correlation_id: 7,
188            client_id: Some("kafrust".to_owned()),
189            mechanism: "PLAIN".to_owned(),
190        };
191
192        assert_eq!(
193            request.encode().unwrap(),
194            [
195                0, 17, // api key
196                0, 1, // api version
197                0, 0, 0, 7, // correlation id
198                0, 7, b'k', b'a', b'f', b'r', b'u', b's', b't', // client id
199                0, 5, b'P', b'L', b'A', b'I', b'N', // mechanism
200            ]
201        );
202        assert_eq!(SASL_HANDSHAKE_API_KEY, 17);
203    }
204
205    #[test]
206    fn decodes_sasl_handshake_v1_response() {
207        let bytes = [
208            0, 0, // error code
209            0, 0, 0, 2, // mechanism count
210            0, 5, b'P', b'L', b'A', b'I', b'N', // mechanism
211            0, 13, b'S', b'C', b'R', b'A', b'M', b'-', b'S', b'H', b'A', b'-', b'2', b'5',
212            b'6', // mechanism
213        ];
214        let mut decoder = Decoder::new(&bytes);
215        let response = SaslHandshakeResponseV1::decode_body(&mut decoder).unwrap();
216
217        assert_eq!(response.error_code, 0);
218        assert_eq!(response.mechanisms, ["PLAIN", "SCRAM-SHA-256"]);
219        assert!(decoder.is_empty());
220    }
221
222    #[test]
223    fn encodes_sasl_authenticate_v0_request() {
224        let request = SaslAuthenticateRequestV0 {
225            correlation_id: 8,
226            client_id: Some("kafrust".to_owned()),
227            auth_bytes: b"\0user\0pass".to_vec(),
228        };
229
230        assert_eq!(
231            request.encode().unwrap(),
232            [
233                0, 36, // api key
234                0, 0, // api version
235                0, 0, 0, 8, // correlation id
236                0, 7, b'k', b'a', b'f', b'r', b'u', b's', b't', // client id
237                0, 0, 0, 10, // auth bytes length
238                0, b'u', b's', b'e', b'r', 0, b'p', b'a', b's', b's',
239            ]
240        );
241        assert_eq!(SASL_AUTHENTICATE_API_KEY, 36);
242    }
243
244    #[test]
245    fn decodes_sasl_authenticate_v0_response() {
246        let bytes = [
247            0, 0, // error code
248            0xff, 0xff, // null error message
249            0, 0, 0, 2, // auth bytes length
250            1, 2, // auth bytes
251        ];
252        let mut decoder = Decoder::new(&bytes);
253        let response = SaslAuthenticateResponseV0::decode_body(&mut decoder).unwrap();
254
255        assert_eq!(response.error_code, 0);
256        assert_eq!(response.error_message, None);
257        assert_eq!(response.auth_bytes, [1, 2]);
258        assert!(decoder.is_empty());
259    }
260
261    #[test]
262    fn encodes_sasl_authenticate_v1_request() {
263        let request = SaslAuthenticateRequestV1 {
264            correlation_id: 9,
265            client_id: Some("kafrust".to_owned()),
266            auth_bytes: b"n,,\x01auth=token\x01\x01".to_vec(),
267        };
268
269        assert_eq!(
270            request.encode().unwrap(),
271            [
272                0, 36, // api key
273                0, 1, // api version
274                0, 0, 0, 9, // correlation id
275                0, 7, b'k', b'a', b'f', b'r', b'u', b's', b't', // client id
276                0, 0, 0, 16, // auth bytes length
277                b'n', b',', b',', 1, b'a', b'u', b't', b'h', b'=', b't', b'o', b'k', b'e', b'n', 1,
278                1,
279            ]
280        );
281    }
282
283    #[test]
284    fn encodes_sasl_authenticate_v2_request() {
285        let request = SaslAuthenticateRequestV2 {
286            correlation_id: 7,
287            client_id: Some("kafrust".to_owned()),
288            auth_bytes: vec![1, 2],
289        };
290
291        assert_eq!(
292            request.encode().unwrap(),
293            [
294                0, 36, // api key
295                0, 2, // api version
296                0, 0, 0, 7, // correlation id
297                0, 7, b'k', b'a', b'f', b'r', b'u', b's', b't', // client id
298                0,    // request header tagged fields
299                3, 1, 2, // compact auth bytes
300                0, // request tagged fields
301            ]
302        );
303    }
304
305    #[test]
306    fn decodes_sasl_authenticate_v1_response() {
307        let bytes = [
308            0, 0, // error code
309            0xff, 0xff, // null error message
310            0, 0, 0, 2, // auth bytes length
311            1, 2, // auth bytes
312            0, 0, 0, 0, 0, 0, 0, 42, // session lifetime ms
313        ];
314        let mut decoder = Decoder::new(&bytes);
315        let response = SaslAuthenticateResponseV1::decode_body(&mut decoder).unwrap();
316
317        assert_eq!(response.error_code, 0);
318        assert_eq!(response.error_message, None);
319        assert_eq!(response.auth_bytes, [1, 2]);
320        assert_eq!(response.session_lifetime_ms, 42);
321        assert!(decoder.is_empty());
322    }
323
324    #[test]
325    fn decodes_sasl_authenticate_v2_response() {
326        let bytes = [
327            0, 0, // error code
328            0, // null error message
329            3, 1, 2, // compact auth bytes
330            0, 0, 0, 0, 0, 0, 0, 42, // session lifetime ms
331            0,  // response tagged fields
332        ];
333        let mut decoder = Decoder::new(&bytes);
334        let response = SaslAuthenticateResponseV2::decode_body(&mut decoder).unwrap();
335
336        assert_eq!(response.error_code, 0);
337        assert_eq!(response.error_message, None);
338        assert_eq!(response.auth_bytes, [1, 2]);
339        assert_eq!(response.session_lifetime_ms, 42);
340        assert!(decoder.is_empty());
341    }
342}
kafrust_protocol/api/sasl.rs

kafrust_protocol/api/
sasl.rs
