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

krabka_protocol/krabka/break_glass/
approve.rs
