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
78impl SaslAuthenticateRequestV0 {
79    pub fn encode(&self) -> Result<Vec<u8>> {
80        let mut encoder = Encoder::new();
81        RequestHeader {
82            api_key: SASL_AUTHENTICATE_API_KEY,
83            api_version: 0,
84            correlation_id: self.correlation_id,
85            client_id: self.client_id.clone(),
86        }
87        .encode_v1(&mut encoder)?;
88        encoder.write_bytes(&self.auth_bytes)?;
89        Ok(encoder.into_bytes())
90    }
91}
92
93#[derive(Debug, Clone, PartialEq, Eq)]
94pub struct SaslAuthenticateResponseV0 {
95    pub error_code: i16,
96    pub error_message: Option<String>,
97    pub auth_bytes: Vec<u8>,
98}
99
100#[derive(Debug, Clone, PartialEq, Eq)]
101pub struct SaslAuthenticateResponseV1 {
102    pub error_code: i16,
103    pub error_message: Option<String>,
104    pub auth_bytes: Vec<u8>,
105    pub session_lifetime_ms: i64,
106}
107
108impl SaslAuthenticateResponseV1 {
109    pub fn decode_body(decoder: &mut Decoder<'_>) -> Result<Self> {
110        Ok(Self {
111            error_code: decoder.read_i16()?,
112            error_message: decoder.read_nullable_string()?,
113            auth_bytes: decoder.read_bytes()?,
114            session_lifetime_ms: decoder.read_i64()?,
115        })
116    }
117}
118
119impl SaslAuthenticateResponseV0 {
120    pub fn decode_body(decoder: &mut Decoder<'_>) -> Result<Self> {
121        Ok(Self {
122            error_code: decoder.read_i16()?,
123            error_message: decoder.read_nullable_string()?,
124            auth_bytes: decoder.read_bytes()?,
125        })
126    }
127}
128
129#[cfg(test)]
130#[allow(clippy::unwrap_used)]
131mod tests {
132    use super::{
133        SaslAuthenticateRequestV0, SaslAuthenticateRequestV1, SaslAuthenticateResponseV0,
134        SaslAuthenticateResponseV1, SaslHandshakeRequestV1, SaslHandshakeResponseV1,
135        SASL_AUTHENTICATE_API_KEY, SASL_HANDSHAKE_API_KEY,
136    };
137    use crate::codec::Decoder;
138
139    #[test]
140    fn encodes_sasl_handshake_v1_request() {
141        let request = SaslHandshakeRequestV1 {
142            correlation_id: 7,
143            client_id: Some("kafrust".to_owned()),
144            mechanism: "PLAIN".to_owned(),
145        };
146
147        assert_eq!(
148            request.encode().unwrap(),
149            [
150                0, 17, // api key
151                0, 1, // api version
152                0, 0, 0, 7, // correlation id
153                0, 7, b'k', b'a', b'f', b'r', b'u', b's', b't', // client id
154                0, 5, b'P', b'L', b'A', b'I', b'N', // mechanism
155            ]
156        );
157        assert_eq!(SASL_HANDSHAKE_API_KEY, 17);
158    }
159
160    #[test]
161    fn decodes_sasl_handshake_v1_response() {
162        let bytes = [
163            0, 0, // error code
164            0, 0, 0, 2, // mechanism count
165            0, 5, b'P', b'L', b'A', b'I', b'N', // mechanism
166            0, 13, b'S', b'C', b'R', b'A', b'M', b'-', b'S', b'H', b'A', b'-', b'2', b'5',
167            b'6', // mechanism
168        ];
169        let mut decoder = Decoder::new(&bytes);
170        let response = SaslHandshakeResponseV1::decode_body(&mut decoder).unwrap();
171
172        assert_eq!(response.error_code, 0);
173        assert_eq!(response.mechanisms, ["PLAIN", "SCRAM-SHA-256"]);
174        assert!(decoder.is_empty());
175    }
176
177    #[test]
178    fn encodes_sasl_authenticate_v0_request() {
179        let request = SaslAuthenticateRequestV0 {
180            correlation_id: 8,
181            client_id: Some("kafrust".to_owned()),
182            auth_bytes: b"\0user\0pass".to_vec(),
183        };
184
185        assert_eq!(
186            request.encode().unwrap(),
187            [
188                0, 36, // api key
189                0, 0, // api version
190                0, 0, 0, 8, // correlation id
191                0, 7, b'k', b'a', b'f', b'r', b'u', b's', b't', // client id
192                0, 0, 0, 10, // auth bytes length
193                0, b'u', b's', b'e', b'r', 0, b'p', b'a', b's', b's',
194            ]
195        );
196        assert_eq!(SASL_AUTHENTICATE_API_KEY, 36);
197    }
198
199    #[test]
200    fn decodes_sasl_authenticate_v0_response() {
201        let bytes = [
202            0, 0, // error code
203            0xff, 0xff, // null error message
204            0, 0, 0, 2, // auth bytes length
205            1, 2, // auth bytes
206        ];
207        let mut decoder = Decoder::new(&bytes);
208        let response = SaslAuthenticateResponseV0::decode_body(&mut decoder).unwrap();
209
210        assert_eq!(response.error_code, 0);
211        assert_eq!(response.error_message, None);
212        assert_eq!(response.auth_bytes, [1, 2]);
213        assert!(decoder.is_empty());
214    }
215
216    #[test]
217    fn encodes_sasl_authenticate_v1_request() {
218        let request = SaslAuthenticateRequestV1 {
219            correlation_id: 9,
220            client_id: Some("kafrust".to_owned()),
221            auth_bytes: b"n,,\x01auth=token\x01\x01".to_vec(),
222        };
223
224        assert_eq!(
225            request.encode().unwrap(),
226            [
227                0, 36, // api key
228                0, 1, // api version
229                0, 0, 0, 9, // correlation id
230                0, 7, b'k', b'a', b'f', b'r', b'u', b's', b't', // client id
231                0, 0, 0, 16, // auth bytes length
232                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,
233                1,
234            ]
235        );
236    }
237
238    #[test]
239    fn decodes_sasl_authenticate_v1_response() {
240        let bytes = [
241            0, 0, // error code
242            0xff, 0xff, // null error message
243            0, 0, 0, 2, // auth bytes length
244            1, 2, // auth bytes
245            0, 0, 0, 0, 0, 0, 0, 42, // session lifetime ms
246        ];
247        let mut decoder = Decoder::new(&bytes);
248        let response = SaslAuthenticateResponseV1::decode_body(&mut decoder).unwrap();
249
250        assert_eq!(response.error_code, 0);
251        assert_eq!(response.error_message, None);
252        assert_eq!(response.auth_bytes, [1, 2]);
253        assert_eq!(response.session_lifetime_ms, 42);
254        assert!(decoder.is_empty());
255    }
256}
kafrust_protocol/api/sasl.rs

kafrust_protocol/api/
sasl.rs
