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
56impl SaslAuthenticateRequestV0 {
57    pub fn encode(&self) -> Result<Vec<u8>> {
58        let mut encoder = Encoder::new();
59        RequestHeader {
60            api_key: SASL_AUTHENTICATE_API_KEY,
61            api_version: 0,
62            correlation_id: self.correlation_id,
63            client_id: self.client_id.clone(),
64        }
65        .encode_v1(&mut encoder)?;
66        encoder.write_bytes(&self.auth_bytes)?;
67        Ok(encoder.into_bytes())
68    }
69}
70
71#[derive(Debug, Clone, PartialEq, Eq)]
72pub struct SaslAuthenticateResponseV0 {
73    pub error_code: i16,
74    pub error_message: Option<String>,
75    pub auth_bytes: Vec<u8>,
76}
77
78impl SaslAuthenticateResponseV0 {
79    pub fn decode_body(decoder: &mut Decoder<'_>) -> Result<Self> {
80        Ok(Self {
81            error_code: decoder.read_i16()?,
82            error_message: decoder.read_nullable_string()?,
83            auth_bytes: decoder.read_bytes()?,
84        })
85    }
86}
87
88#[cfg(test)]
89#[allow(clippy::unwrap_used)]
90mod tests {
91    use super::{
92        SaslAuthenticateRequestV0, SaslAuthenticateResponseV0, SaslHandshakeRequestV1,
93        SaslHandshakeResponseV1, SASL_AUTHENTICATE_API_KEY, SASL_HANDSHAKE_API_KEY,
94    };
95    use crate::codec::Decoder;
96
97    #[test]
98    fn encodes_sasl_handshake_v1_request() {
99        let request = SaslHandshakeRequestV1 {
100            correlation_id: 7,
101            client_id: Some("kafrust".to_owned()),
102            mechanism: "PLAIN".to_owned(),
103        };
104
105        assert_eq!(
106            request.encode().unwrap(),
107            [
108                0, 17, // api key
109                0, 1, // api version
110                0, 0, 0, 7, // correlation id
111                0, 7, b'k', b'a', b'f', b'r', b'u', b's', b't', // client id
112                0, 5, b'P', b'L', b'A', b'I', b'N', // mechanism
113            ]
114        );
115        assert_eq!(SASL_HANDSHAKE_API_KEY, 17);
116    }
117
118    #[test]
119    fn decodes_sasl_handshake_v1_response() {
120        let bytes = [
121            0, 0, // error code
122            0, 0, 0, 2, // mechanism count
123            0, 5, b'P', b'L', b'A', b'I', b'N', // mechanism
124            0, 13, b'S', b'C', b'R', b'A', b'M', b'-', b'S', b'H', b'A', b'-', b'2', b'5',
125            b'6', // mechanism
126        ];
127        let mut decoder = Decoder::new(&bytes);
128        let response = SaslHandshakeResponseV1::decode_body(&mut decoder).unwrap();
129
130        assert_eq!(response.error_code, 0);
131        assert_eq!(response.mechanisms, ["PLAIN", "SCRAM-SHA-256"]);
132        assert!(decoder.is_empty());
133    }
134
135    #[test]
136    fn encodes_sasl_authenticate_v0_request() {
137        let request = SaslAuthenticateRequestV0 {
138            correlation_id: 8,
139            client_id: Some("kafrust".to_owned()),
140            auth_bytes: b"\0user\0pass".to_vec(),
141        };
142
143        assert_eq!(
144            request.encode().unwrap(),
145            [
146                0, 36, // api key
147                0, 0, // api version
148                0, 0, 0, 8, // correlation id
149                0, 7, b'k', b'a', b'f', b'r', b'u', b's', b't', // client id
150                0, 0, 0, 10, // auth bytes length
151                0, b'u', b's', b'e', b'r', 0, b'p', b'a', b's', b's',
152            ]
153        );
154        assert_eq!(SASL_AUTHENTICATE_API_KEY, 36);
155    }
156
157    #[test]
158    fn decodes_sasl_authenticate_v0_response() {
159        let bytes = [
160            0, 0, // error code
161            0xff, 0xff, // null error message
162            0, 0, 0, 2, // auth bytes length
163            1, 2, // auth bytes
164        ];
165        let mut decoder = Decoder::new(&bytes);
166        let response = SaslAuthenticateResponseV0::decode_body(&mut decoder).unwrap();
167
168        assert_eq!(response.error_code, 0);
169        assert_eq!(response.error_message, None);
170        assert_eq!(response.auth_bytes, [1, 2]);
171        assert!(decoder.is_empty());
172    }
173}
kafrust_protocol/api/sasl.rs

kafrust_protocol/api/
sasl.rs
