krabka_protocol/krabka/break_glass/
approve.rs

1//! `ApproveBreakGlass`, api key 1018.
2//!
3//! One request adds an approval to a break-glass proposal, or it withdraws the
4//! proposal. `withdraw` picks between the two, which follows
5//! [`AlterableBarrierGroup`](crate::krabka::barrier::AlterableBarrierGroup) and
6//! its `delete` flag. One api key then serves both directions of the same
7//! standing authorization, and the krabka-private range keeps a key free.
8//!
9//! The broker checks the approver against the authenticated principal of the
10//! connection. It refuses the proposer, a principal that is not in the approver
11//! set, and a principal that already approved this proposal.
12//!
13//! # Signature
14//!
15//! `key_id` and `signature` carry a detached Ed25519 signature that the
16//! operator makes on their own machine. The private key never reaches a broker.
17//! Both fields are empty for an unsigned approval. The broker demands a
18//! signature for the actions that its configuration names.
19
20use bytes::{Buf, BufMut};
21
22#[cfg(test)]
23use crate::krabka::test_support::sample_tagged_fields;
24use crate::{
25    Decode, Encode, ProtocolError, ProtocolRequest, UnknownTaggedFields,
26    krabka::common::{
27        BOOL_LEN, I16_LEN, I32_LEN, UUID_LEN, check_version, read_unknown_tagged_fields,
28        unknown_tagged_fields_len, write_unknown_tagged_fields,
29    },
30    primitives::{
31        fixed::{get_bool, get_i16, get_i32, put_bool, put_i16, put_i32},
32        string_bytes::{
33            compact_bytes_len, compact_nullable_string_len, compact_string_len,
34            get_compact_bytes_owned, get_compact_nullable_string_owned, get_compact_string_owned,
35            put_compact_bytes, put_compact_nullable_string, put_compact_string,
36        },
37        uuid::{Uuid, get_uuid, put_uuid},
38    },
39};
40
41/// Api key of `ApproveBreakGlass`.
42pub const API_KEY: i16 = 1018;
43/// Lowest version of `ApproveBreakGlass` that this build speaks.
44pub const MIN_VERSION: i16 = 0;
45/// Highest version of `ApproveBreakGlass` that this build speaks.
46pub const MAX_VERSION: i16 = 0;
47/// First version with flexible framing from KIP-482.
48pub const FLEXIBLE_MIN: i16 = 0;
49
50/// Reports whether `version` uses flexible framing from KIP-482.
51///
52/// The request header takes the same framing as the request body, so a caller
53/// that writes the header needs this answer.
54#[inline]
55#[must_use]
56pub fn is_flexible(version: i16) -> bool {
57    version >= FLEXIBLE_MIN
58}
59
60/// Adds an approval to a break-glass proposal, or withdraws the proposal.
61#[derive(Debug, Clone, PartialEq, Eq, Default)]
62pub struct ApproveBreakGlassRequest {
63    /// The proposal to approve or to withdraw.
64    pub proposal_id: Uuid,
65    /// The operator key that made `signature`. It is empty for an unsigned
66    /// approval.
67    pub key_id: String,
68    /// Detached Ed25519 signature over the canonical bytes of the proposal. It
69    /// is empty for an unsigned approval.
70    pub signature: Vec<u8>,
71    /// `true` withdraws the proposal instead of approving it. The broker then
72    /// ignores `key_id` and `signature`.
73    pub withdraw: bool,
74    /// Tagged fields that this build does not know.
75    pub unknown_tagged_fields: UnknownTaggedFields,
76}
77
78impl Encode for ApproveBreakGlassRequest {
79    fn encode<B: BufMut>(&self, buf: &mut B, version: i16) -> Result<(), ProtocolError> {
80        check_version(API_KEY, version, MIN_VERSION, MAX_VERSION)?;
81        put_uuid(buf, self.proposal_id);
82        put_compact_string(buf, &self.key_id)?;
83        put_compact_bytes(buf, &self.signature)?;
84        put_bool(buf, self.withdraw);
85        write_unknown_tagged_fields(buf, &self.unknown_tagged_fields)?;
86        Ok(())
87    }
88
89    fn encoded_len(&self, _version: i16) -> usize {
90        UUID_LEN
91            + compact_string_len(&self.key_id)
92            + compact_bytes_len(&self.signature)
93            + BOOL_LEN
94            + unknown_tagged_fields_len(&self.unknown_tagged_fields)
95    }
96}
97
98impl Decode<'_> for ApproveBreakGlassRequest {
99    fn decode<B: Buf>(buf: &mut B, version: i16) -> Result<Self, ProtocolError> {
100        check_version(API_KEY, version, MIN_VERSION, MAX_VERSION)?;
101        Ok(Self {
102            proposal_id: get_uuid(buf)?,
103            key_id: get_compact_string_owned(buf)?,
104            signature: get_compact_bytes_owned(buf)?.to_vec(),
105            withdraw: get_bool(buf)?,
106            unknown_tagged_fields: read_unknown_tagged_fields(buf)?,
107        })
108    }
109}
110
111impl ProtocolRequest for ApproveBreakGlassRequest {
112    const API_KEY: i16 = API_KEY;
113    const MIN_VERSION: i16 = MIN_VERSION;
114    const MAX_VERSION: i16 = MAX_VERSION;
115    const LATEST_STABLE_VERSION: i16 = MAX_VERSION;
116    const FLEXIBLE_MIN: i16 = FLEXIBLE_MIN;
117
118    type Response = ApproveBreakGlassResponse;
119}
120
121/// Outcome of an [`ApproveBreakGlassRequest`].
122#[derive(Debug, Clone, PartialEq, Eq, Default)]
123pub struct ApproveBreakGlassResponse {
124    /// Milliseconds that the broker held the request back for a quota.
125    pub throttle_time_ms: i32,
126    /// Kafka error code. `0` means the controller recorded the approval, or
127    /// withdrew the proposal.
128    pub error_code: i16,
129    /// Text that explains a non-zero `error_code`. `None` when the broker has
130    /// nothing to add to the code.
131    pub error_message: Option<String>,
132    /// Number of distinct principals that have approved the proposal after this
133    /// request. A caller compares it with `approvals_required` to see whether
134    /// the proposal can now authorize its action.
135    pub approvals_held: i32,
136    /// Number of distinct principals that the proposal needs. The broker holds
137    /// this value in its configuration, and the minimum is two.
138    pub approvals_required: i32,
139    /// Tagged fields that this build does not know.
140    pub unknown_tagged_fields: UnknownTaggedFields,
141}
142
143impl Encode for ApproveBreakGlassResponse {
144    fn encode<B: BufMut>(&self, buf: &mut B, version: i16) -> Result<(), ProtocolError> {
145        check_version(API_KEY, version, MIN_VERSION, MAX_VERSION)?;
146        put_i32(buf, self.throttle_time_ms);
147        put_i16(buf, self.error_code);
148        put_compact_nullable_string(buf, self.error_message.as_deref())?;
149        put_i32(buf, self.approvals_held);
150        put_i32(buf, self.approvals_required);
151        write_unknown_tagged_fields(buf, &self.unknown_tagged_fields)?;
152        Ok(())
153    }
154
155    fn encoded_len(&self, _version: i16) -> usize {
156        I32_LEN
157            + I16_LEN
158            + compact_nullable_string_len(self.error_message.as_deref())
159            + I32_LEN
160            + I32_LEN
161            + unknown_tagged_fields_len(&self.unknown_tagged_fields)
162    }
163}
164
165impl Decode<'_> for ApproveBreakGlassResponse {
166    fn decode<B: Buf>(buf: &mut B, version: i16) -> Result<Self, ProtocolError> {
167        check_version(API_KEY, version, MIN_VERSION, MAX_VERSION)?;
168        Ok(Self {
169            throttle_time_ms: get_i32(buf)?,
170            error_code: get_i16(buf)?,
171            error_message: get_compact_nullable_string_owned(buf)?,
172            approvals_held: get_i32(buf)?,
173            approvals_required: get_i32(buf)?,
174            unknown_tagged_fields: read_unknown_tagged_fields(buf)?,
175        })
176    }
177}
178
179#[cfg(test)]
180impl ApproveBreakGlassRequest {
181    /// Builds a signed approval with every field set, for the round-trip tests.
182    fn populated() -> Self {
183        Self {
184            proposal_id: Uuid([
185                200, 201, 202, 203, 204, 205, 206, 207, 208, 209, 210, 211, 212, 213, 214, 215,
186            ]),
187            key_id: "bob-yubi".to_owned(),
188            signature: (0..64).collect(),
189            withdraw: false,
190            unknown_tagged_fields: sample_tagged_fields(),
191        }
192    }
193}
194
195#[cfg(test)]
196impl ApproveBreakGlassResponse {
197    /// Builds a response with every field set, for the round-trip tests.
198    fn populated() -> Self {
199        Self {
200            throttle_time_ms: 9,
201            error_code: 1007,
202            error_message: Some("principal already approved this proposal".to_owned()),
203            approvals_held: 1,
204            approvals_required: 2,
205            unknown_tagged_fields: sample_tagged_fields(),
206        }
207    }
208}
209
210#[cfg(test)]
211mod tests {
212    use assert2::assert;
213
214    use super::*;
215    use crate::krabka::test_support::{reject_truncated, roundtrip_case, unsupported_versions};
216
217    #[test]
218    fn requests_roundtrip() {
219        let cases = [
220            ("default", ApproveBreakGlassRequest::default()),
221            ("signed approval", ApproveBreakGlassRequest::populated()),
222            (
223                "unsigned approval",
224                ApproveBreakGlassRequest {
225                    key_id: String::new(),
226                    signature: Vec::new(),
227                    unknown_tagged_fields: UnknownTaggedFields::default(),
228                    ..ApproveBreakGlassRequest::populated()
229                },
230            ),
231            (
232                "withdraw",
233                ApproveBreakGlassRequest {
234                    key_id: String::new(),
235                    signature: Vec::new(),
236                    withdraw: true,
237                    ..ApproveBreakGlassRequest::populated()
238                },
239            ),
240        ];
241        for (label, request) in &cases {
242            for version in MIN_VERSION..=MAX_VERSION {
243                roundtrip_case(label, request, version);
244            }
245        }
246    }
247
248    #[test]
249    fn responses_roundtrip() {
250        let cases = [
251            ("default", ApproveBreakGlassResponse::default()),
252            (
253                "refusal with a message",
254                ApproveBreakGlassResponse::populated(),
255            ),
256            (
257                "second approval reaches the quorum",
258                ApproveBreakGlassResponse {
259                    throttle_time_ms: 0,
260                    error_code: 0,
261                    error_message: None,
262                    approvals_held: 2,
263                    approvals_required: 2,
264                    unknown_tagged_fields: UnknownTaggedFields::default(),
265                },
266            ),
267        ];
268        for (label, response) in &cases {
269            for version in MIN_VERSION..=MAX_VERSION {
270                roundtrip_case(label, response, version);
271            }
272        }
273    }
274
275    #[test]
276    fn a_default_request_approves_and_does_not_withdraw() {
277        assert!(!ApproveBreakGlassRequest::default().withdraw);
278    }
279
280    #[test]
281    fn truncated_input_is_rejected() {
282        reject_truncated(&ApproveBreakGlassRequest::populated(), MIN_VERSION);
283        reject_truncated(&ApproveBreakGlassResponse::populated(), MIN_VERSION);
284    }
285
286    #[test]
287    fn rejects_unsupported_versions() {
288        for version in unsupported_versions(MIN_VERSION, MAX_VERSION) {
289            let request = ApproveBreakGlassRequest::populated();
290            assert!(matches!(
291                request.encode(&mut Vec::<u8>::new(), version),
292                Err(ProtocolError::UnsupportedVersion {
293                    api_key: API_KEY,
294                    ..
295                })
296            ));
297            let response = ApproveBreakGlassResponse::populated();
298            assert!(matches!(
299                response.encode(&mut Vec::<u8>::new(), version),
300                Err(ProtocolError::UnsupportedVersion {
301                    api_key: API_KEY,
302                    ..
303                })
304            ));
305        }
306    }
307}
krabka_protocol/krabka/break_glass/approve.rs

krabka_protocol/krabka/break_glass/
approve.rs
