krabka_protocol/krabka/break_glass/
propose.rs

1//! `ProposeBreakGlass`, api key 1017.
2//!
3//! The request asks the controller to open a break-glass proposal. The
4//! controller gives the proposal an id and an expiry, and it holds the proposal
5//! in the metadata log until an operator consumes it, an operator withdraws it,
6//! or it expires.
7//!
8//! A proposal names one action and one target. It carries no approval yet, so
9//! the proposer sends an [`ApproveBreakGlass`](super::approve) request from a
10//! second principal before the action can run.
11
12use bytes::{Buf, BufMut};
13
14#[cfg(test)]
15use crate::krabka::test_support::sample_tagged_fields;
16use crate::{
17    Decode, Encode, ProtocolError, ProtocolRequest, UnknownTaggedFields,
18    krabka::common::{
19        I8_LEN, I16_LEN, I32_LEN, I64_LEN, UUID_LEN, check_version, read_unknown_tagged_fields,
20        unknown_tagged_fields_len, write_unknown_tagged_fields,
21    },
22    primitives::{
23        fixed::{get_i8, get_i16, get_i32, get_i64, put_i8, put_i16, put_i32, put_i64},
24        string_bytes::{
25            compact_nullable_string_len, compact_string_len, get_compact_nullable_string_owned,
26            get_compact_string_owned, put_compact_nullable_string, put_compact_string,
27        },
28        uuid::{Uuid, get_uuid, put_uuid},
29    },
30};
31
32/// Api key of `ProposeBreakGlass`.
33pub const API_KEY: i16 = 1017;
34/// Lowest version of `ProposeBreakGlass` that this build speaks.
35pub const MIN_VERSION: i16 = 0;
36/// Highest version of `ProposeBreakGlass` that this build speaks.
37pub const MAX_VERSION: i16 = 0;
38/// First version with flexible framing from KIP-482.
39pub const FLEXIBLE_MIN: i16 = 0;
40
41/// Reports whether `version` uses flexible framing from KIP-482.
42///
43/// The request header takes the same framing as the request body, so a caller
44/// that writes the header needs this answer.
45#[inline]
46#[must_use]
47pub fn is_flexible(version: i16) -> bool {
48    version >= FLEXIBLE_MIN
49}
50
51/// Opens a break-glass proposal.
52#[derive(Debug, Clone, PartialEq, Eq, Default)]
53pub struct ProposeBreakGlassRequest {
54    /// The privileged action to gate. The value is the wire value of the
55    /// broker's break-glass action, as [`super`] describes.
56    pub action: i8,
57    /// What the action applies to. The shape of the text follows the action: a
58    /// topic name, a partition, a broker id, or a freeze scope.
59    pub target: String,
60    /// Text that says why the operator needs the action.
61    pub reason: String,
62    /// Lifetime of the proposal, in milliseconds. `0` asks for the lifetime
63    /// that the broker holds in its configuration. The broker caps a longer
64    /// value at that configured lifetime.
65    pub ttl_ms: i64,
66    /// Tagged fields that this build does not know.
67    pub unknown_tagged_fields: UnknownTaggedFields,
68}
69
70impl Encode for ProposeBreakGlassRequest {
71    fn encode<B: BufMut>(&self, buf: &mut B, version: i16) -> Result<(), ProtocolError> {
72        check_version(API_KEY, version, MIN_VERSION, MAX_VERSION)?;
73        put_i8(buf, self.action);
74        put_compact_string(buf, &self.target)?;
75        put_compact_string(buf, &self.reason)?;
76        put_i64(buf, self.ttl_ms);
77        write_unknown_tagged_fields(buf, &self.unknown_tagged_fields)?;
78        Ok(())
79    }
80
81    fn encoded_len(&self, _version: i16) -> usize {
82        I8_LEN
83            + compact_string_len(&self.target)
84            + compact_string_len(&self.reason)
85            + I64_LEN
86            + unknown_tagged_fields_len(&self.unknown_tagged_fields)
87    }
88}
89
90impl Decode<'_> for ProposeBreakGlassRequest {
91    fn decode<B: Buf>(buf: &mut B, version: i16) -> Result<Self, ProtocolError> {
92        check_version(API_KEY, version, MIN_VERSION, MAX_VERSION)?;
93        Ok(Self {
94            action: get_i8(buf)?,
95            target: get_compact_string_owned(buf)?,
96            reason: get_compact_string_owned(buf)?,
97            ttl_ms: get_i64(buf)?,
98            unknown_tagged_fields: read_unknown_tagged_fields(buf)?,
99        })
100    }
101}
102
103impl ProtocolRequest for ProposeBreakGlassRequest {
104    const API_KEY: i16 = API_KEY;
105    const MIN_VERSION: i16 = MIN_VERSION;
106    const MAX_VERSION: i16 = MAX_VERSION;
107    const LATEST_STABLE_VERSION: i16 = MAX_VERSION;
108    const FLEXIBLE_MIN: i16 = FLEXIBLE_MIN;
109
110    type Response = ProposeBreakGlassResponse;
111}
112
113/// Outcome of a [`ProposeBreakGlassRequest`].
114#[derive(Debug, Clone, PartialEq, Eq, Default)]
115pub struct ProposeBreakGlassResponse {
116    /// Milliseconds that the broker held the request back for a quota.
117    pub throttle_time_ms: i32,
118    /// Kafka error code. `0` means the controller opened the proposal.
119    pub error_code: i16,
120    /// Text that explains a non-zero `error_code`. `None` when the broker has
121    /// nothing to add to the code.
122    pub error_message: Option<String>,
123    /// Id of the new proposal. An approver names it, and so does the audit
124    /// event of the action that consumes it. It is nil for a non-zero
125    /// `error_code`.
126    pub proposal_id: Uuid,
127    /// Milliseconds since the Unix epoch, at the moment the proposal expires.
128    /// It is `0` for a non-zero `error_code`.
129    pub expires_at_ms: i64,
130    /// Tagged fields that this build does not know.
131    pub unknown_tagged_fields: UnknownTaggedFields,
132}
133
134impl Encode for ProposeBreakGlassResponse {
135    fn encode<B: BufMut>(&self, buf: &mut B, version: i16) -> Result<(), ProtocolError> {
136        check_version(API_KEY, version, MIN_VERSION, MAX_VERSION)?;
137        put_i32(buf, self.throttle_time_ms);
138        put_i16(buf, self.error_code);
139        put_compact_nullable_string(buf, self.error_message.as_deref())?;
140        put_uuid(buf, self.proposal_id);
141        put_i64(buf, self.expires_at_ms);
142        write_unknown_tagged_fields(buf, &self.unknown_tagged_fields)?;
143        Ok(())
144    }
145
146    fn encoded_len(&self, _version: i16) -> usize {
147        I32_LEN
148            + I16_LEN
149            + compact_nullable_string_len(self.error_message.as_deref())
150            + UUID_LEN
151            + I64_LEN
152            + unknown_tagged_fields_len(&self.unknown_tagged_fields)
153    }
154}
155
156impl Decode<'_> for ProposeBreakGlassResponse {
157    fn decode<B: Buf>(buf: &mut B, version: i16) -> Result<Self, ProtocolError> {
158        check_version(API_KEY, version, MIN_VERSION, MAX_VERSION)?;
159        Ok(Self {
160            throttle_time_ms: get_i32(buf)?,
161            error_code: get_i16(buf)?,
162            error_message: get_compact_nullable_string_owned(buf)?,
163            proposal_id: get_uuid(buf)?,
164            expires_at_ms: get_i64(buf)?,
165            unknown_tagged_fields: read_unknown_tagged_fields(buf)?,
166        })
167    }
168}
169
170#[cfg(test)]
171impl ProposeBreakGlassRequest {
172    /// Builds a request with every field set, for the round-trip tests.
173    fn populated() -> Self {
174        Self {
175            action: 6,
176            target: "doomed".to_owned(),
177            reason: "topic holds test data only".to_owned(),
178            ttl_ms: 1_800_000,
179            unknown_tagged_fields: sample_tagged_fields(),
180        }
181    }
182}
183
184#[cfg(test)]
185impl ProposeBreakGlassResponse {
186    /// Builds a response with every field set, for the round-trip tests.
187    fn populated() -> Self {
188        Self {
189            throttle_time_ms: 4,
190            error_code: 0,
191            error_message: None,
192            proposal_id: Uuid([
193                200, 201, 202, 203, 204, 205, 206, 207, 208, 209, 210, 211, 212, 213, 214, 215,
194            ]),
195            expires_at_ms: 1_770_000_180_000,
196            unknown_tagged_fields: sample_tagged_fields(),
197        }
198    }
199}
200
201#[cfg(test)]
202mod tests {
203    use assert2::assert;
204
205    use super::*;
206    use crate::krabka::test_support::{reject_truncated, roundtrip_case, unsupported_versions};
207
208    #[test]
209    fn requests_roundtrip() {
210        let cases = [
211            ("default", ProposeBreakGlassRequest::default()),
212            ("delete a topic", ProposeBreakGlassRequest::populated()),
213            (
214                "configured lifetime and no reason",
215                ProposeBreakGlassRequest {
216                    action: 1,
217                    target: "literal:orders".to_owned(),
218                    reason: String::new(),
219                    ttl_ms: 0,
220                    unknown_tagged_fields: UnknownTaggedFields::default(),
221                },
222            ),
223        ];
224        for (label, request) in &cases {
225            for version in MIN_VERSION..=MAX_VERSION {
226                roundtrip_case(label, request, version);
227            }
228        }
229    }
230
231    #[test]
232    fn responses_roundtrip() {
233        let cases = [
234            ("default", ProposeBreakGlassResponse::default()),
235            ("accepted proposal", ProposeBreakGlassResponse::populated()),
236            (
237                "refusal carries no proposal",
238                ProposeBreakGlassResponse {
239                    throttle_time_ms: 0,
240                    error_code: 1008,
241                    error_message: Some("principal is not an approver".to_owned()),
242                    proposal_id: Uuid::ZERO,
243                    expires_at_ms: 0,
244                    unknown_tagged_fields: UnknownTaggedFields::default(),
245                },
246            ),
247        ];
248        for (label, response) in &cases {
249            for version in MIN_VERSION..=MAX_VERSION {
250                roundtrip_case(label, response, version);
251            }
252        }
253    }
254
255    #[test]
256    fn a_default_request_asks_for_the_configured_lifetime() {
257        assert!(ProposeBreakGlassRequest::default().ttl_ms == 0);
258    }
259
260    #[test]
261    fn truncated_input_is_rejected() {
262        reject_truncated(&ProposeBreakGlassRequest::populated(), MIN_VERSION);
263        reject_truncated(&ProposeBreakGlassResponse::populated(), MIN_VERSION);
264    }
265
266    #[test]
267    fn rejects_unsupported_versions() {
268        for version in unsupported_versions(MIN_VERSION, MAX_VERSION) {
269            let request = ProposeBreakGlassRequest::populated();
270            assert!(matches!(
271                request.encode(&mut Vec::<u8>::new(), version),
272                Err(ProtocolError::UnsupportedVersion {
273                    api_key: API_KEY,
274                    ..
275                })
276            ));
277            let response = ProposeBreakGlassResponse::populated();
278            assert!(matches!(
279                response.encode(&mut Vec::<u8>::new(), version),
280                Err(ProtocolError::UnsupportedVersion {
281                    api_key: API_KEY,
282                    ..
283                })
284            ));
285        }
286    }
287}
krabka_protocol/krabka/break_glass/propose.rs

krabka_protocol/krabka/break_glass/
propose.rs
