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

krabka_protocol/krabka/break_glass/
propose.rs
