krabka_protocol/krabka/freeze/
describe_topic_freezes.rs

1//! `DescribeTopicFreezes`, api key 1016.
2//!
3//! The request reads the write-freeze registry. An empty `scope_filter` asks
4//! for every entry.
5//!
6//! Each entry carries the `key_id` and the `signature` that
7//! [`SetTopicFreeze`](super::set_topic_freeze) put in the metadata log. The
8//! response gives them back on purpose. An operator tool re-verifies each
9//! signature against the operator public keys on its own machine, so the
10//! broker's word is not the only evidence of who set a freeze.
11
12use bytes::{Buf, BufMut};
13
14use super::{PATTERN_TYPE_ANY, PATTERN_TYPE_LITERAL};
15#[cfg(test)]
16use crate::krabka::test_support::sample_tagged_fields;
17use crate::{
18    Decode, Encode, ProtocolError, ProtocolRequest, UnknownTaggedFields,
19    krabka::common::{
20        I8_LEN, I16_LEN, I32_LEN, I64_LEN, UUID_LEN, array_len, check_version, decode_array,
21        encode_array, read_unknown_tagged_fields, unknown_tagged_fields_len,
22        write_unknown_tagged_fields,
23    },
24    primitives::{
25        fixed::{get_i8, get_i16, get_i32, get_i64, put_i8, put_i16, put_i32, put_i64},
26        string_bytes::{
27            compact_bytes_len, compact_nullable_string_len, compact_string_len,
28            get_compact_bytes_owned, get_compact_nullable_string_owned, get_compact_string_owned,
29            put_compact_bytes, put_compact_nullable_string, put_compact_string,
30        },
31        uuid::{Uuid, get_uuid, put_uuid},
32    },
33};
34
35/// Api key of `DescribeTopicFreezes`.
36pub const API_KEY: i16 = 1016;
37/// Lowest version of `DescribeTopicFreezes` that this build speaks.
38pub const MIN_VERSION: i16 = 0;
39/// Highest version of `DescribeTopicFreezes` that this build speaks.
40pub const MAX_VERSION: i16 = 0;
41/// First version with flexible framing from KIP-482.
42pub const FLEXIBLE_MIN: i16 = 0;
43
44/// Reports whether `version` uses flexible framing from KIP-482.
45///
46/// The request header takes the same framing as the request body, so a caller
47/// that writes the header needs this answer.
48#[inline]
49#[must_use]
50pub fn is_flexible(version: i16) -> bool {
51    version >= FLEXIBLE_MIN
52}
53
54/// Reads the write-freeze registry.
55#[derive(Debug, Clone, PartialEq, Eq)]
56pub struct DescribeTopicFreezesRequest {
57    /// The scope to read. `None` asks for every entry. A value matches the
58    /// `scope` of an entry exactly, and it is not itself a prefix match.
59    pub scope_filter: Option<String>,
60    /// [`PATTERN_TYPE_LITERAL`] or
61    /// [`PATTERN_TYPE_PREFIXED`](super::PATTERN_TYPE_PREFIXED) reads entries of
62    /// that pattern type alone. [`PATTERN_TYPE_ANY`] and `0` both read every
63    /// pattern type.
64    pub pattern_type_filter: i8,
65    /// Tagged fields that this build does not know.
66    pub unknown_tagged_fields: UnknownTaggedFields,
67}
68
69impl Default for DescribeTopicFreezesRequest {
70    fn default() -> Self {
71        Self {
72            scope_filter: None,
73            pattern_type_filter: PATTERN_TYPE_ANY,
74            unknown_tagged_fields: UnknownTaggedFields::default(),
75        }
76    }
77}
78
79impl Encode for DescribeTopicFreezesRequest {
80    fn encode<B: BufMut>(&self, buf: &mut B, version: i16) -> Result<(), ProtocolError> {
81        check_version(API_KEY, version, MIN_VERSION, MAX_VERSION)?;
82        put_compact_nullable_string(buf, self.scope_filter.as_deref())?;
83        put_i8(buf, self.pattern_type_filter);
84        write_unknown_tagged_fields(buf, &self.unknown_tagged_fields)?;
85        Ok(())
86    }
87
88    fn encoded_len(&self, _version: i16) -> usize {
89        compact_nullable_string_len(self.scope_filter.as_deref())
90            + I8_LEN
91            + unknown_tagged_fields_len(&self.unknown_tagged_fields)
92    }
93}
94
95impl Decode<'_> for DescribeTopicFreezesRequest {
96    fn decode<B: Buf>(buf: &mut B, version: i16) -> Result<Self, ProtocolError> {
97        check_version(API_KEY, version, MIN_VERSION, MAX_VERSION)?;
98        Ok(Self {
99            scope_filter: get_compact_nullable_string_owned(buf)?,
100            pattern_type_filter: get_i8(buf)?,
101            unknown_tagged_fields: read_unknown_tagged_fields(buf)?,
102        })
103    }
104}
105
106impl ProtocolRequest for DescribeTopicFreezesRequest {
107    const API_KEY: i16 = API_KEY;
108    const MIN_VERSION: i16 = MIN_VERSION;
109    const MAX_VERSION: i16 = MAX_VERSION;
110    const LATEST_STABLE_VERSION: i16 = MAX_VERSION;
111    const FLEXIBLE_MIN: i16 = FLEXIBLE_MIN;
112
113    type Response = DescribeTopicFreezesResponse;
114}
115
116/// One live entry of the write-freeze registry.
117#[derive(Debug, Clone, PartialEq, Eq)]
118pub struct DescribedTopicFreeze {
119    /// A literal topic name, or a topic-name prefix. `pattern_type` says which.
120    pub scope: String,
121    /// [`PATTERN_TYPE_LITERAL`] or
122    /// [`PATTERN_TYPE_PREFIXED`](super::PATTERN_TYPE_PREFIXED).
123    pub pattern_type: i8,
124    /// Text that says why the operator set the freeze.
125    pub reason: String,
126    /// The principal that the broker authenticated when it took the request.
127    pub set_by: String,
128    /// Milliseconds since the Unix epoch, at the moment the operator signed the
129    /// request that set this entry.
130    pub set_at_ms: i64,
131    /// The break-glass proposal that approved this entry. It is nil for a
132    /// freeze, because a freeze needs no approval.
133    pub proposal_id: Uuid,
134    /// The operator key that made `signature`. It is empty for an entry that an
135    /// unsigned request set.
136    pub key_id: String,
137    /// Detached Ed25519 signature over the canonical bytes of the freeze. It is
138    /// empty for an entry that an unsigned request set. A tool re-verifies it
139    /// locally, and an empty signature is an attestation and not a proof.
140    pub signature: Vec<u8>,
141    /// Tagged fields that this build does not know.
142    pub unknown_tagged_fields: UnknownTaggedFields,
143}
144
145impl Default for DescribedTopicFreeze {
146    fn default() -> Self {
147        Self {
148            scope: String::new(),
149            pattern_type: PATTERN_TYPE_LITERAL,
150            reason: String::new(),
151            set_by: String::new(),
152            set_at_ms: 0,
153            proposal_id: Uuid::ZERO,
154            key_id: String::new(),
155            signature: Vec::new(),
156            unknown_tagged_fields: UnknownTaggedFields::default(),
157        }
158    }
159}
160
161impl Encode for DescribedTopicFreeze {
162    fn encode<B: BufMut>(&self, buf: &mut B, _version: i16) -> Result<(), ProtocolError> {
163        put_compact_string(buf, &self.scope)?;
164        put_i8(buf, self.pattern_type);
165        put_compact_string(buf, &self.reason)?;
166        put_compact_string(buf, &self.set_by)?;
167        put_i64(buf, self.set_at_ms);
168        put_uuid(buf, self.proposal_id);
169        put_compact_string(buf, &self.key_id)?;
170        put_compact_bytes(buf, &self.signature)?;
171        write_unknown_tagged_fields(buf, &self.unknown_tagged_fields)?;
172        Ok(())
173    }
174
175    fn encoded_len(&self, _version: i16) -> usize {
176        compact_string_len(&self.scope)
177            + I8_LEN
178            + compact_string_len(&self.reason)
179            + compact_string_len(&self.set_by)
180            + I64_LEN
181            + UUID_LEN
182            + compact_string_len(&self.key_id)
183            + compact_bytes_len(&self.signature)
184            + unknown_tagged_fields_len(&self.unknown_tagged_fields)
185    }
186}
187
188impl Decode<'_> for DescribedTopicFreeze {
189    fn decode<B: Buf>(buf: &mut B, _version: i16) -> Result<Self, ProtocolError> {
190        Ok(Self {
191            scope: get_compact_string_owned(buf)?,
192            pattern_type: get_i8(buf)?,
193            reason: get_compact_string_owned(buf)?,
194            set_by: get_compact_string_owned(buf)?,
195            set_at_ms: get_i64(buf)?,
196            proposal_id: get_uuid(buf)?,
197            key_id: get_compact_string_owned(buf)?,
198            signature: get_compact_bytes_owned(buf)?.to_vec(),
199            unknown_tagged_fields: read_unknown_tagged_fields(buf)?,
200        })
201    }
202}
203
204/// The registry entries that a [`DescribeTopicFreezesRequest`] asked for.
205#[derive(Debug, Clone, PartialEq, Eq, Default)]
206pub struct DescribeTopicFreezesResponse {
207    /// Milliseconds that the broker held the request back for a quota.
208    pub throttle_time_ms: i32,
209    /// Kafka error code. `0` means `freezes` holds the answer.
210    pub error_code: i16,
211    /// Text that explains a non-zero `error_code`. `None` when the broker has
212    /// nothing to add to the code.
213    pub error_message: Option<String>,
214    /// One entry for each freeze that the filter matched.
215    pub freezes: Vec<DescribedTopicFreeze>,
216    /// Tagged fields that this build does not know.
217    pub unknown_tagged_fields: UnknownTaggedFields,
218}
219
220impl Encode for DescribeTopicFreezesResponse {
221    fn encode<B: BufMut>(&self, buf: &mut B, version: i16) -> Result<(), ProtocolError> {
222        check_version(API_KEY, version, MIN_VERSION, MAX_VERSION)?;
223        put_i32(buf, self.throttle_time_ms);
224        put_i16(buf, self.error_code);
225        put_compact_nullable_string(buf, self.error_message.as_deref())?;
226        encode_array(buf, &self.freezes, version)?;
227        write_unknown_tagged_fields(buf, &self.unknown_tagged_fields)?;
228        Ok(())
229    }
230
231    fn encoded_len(&self, version: i16) -> usize {
232        I32_LEN
233            + I16_LEN
234            + compact_nullable_string_len(self.error_message.as_deref())
235            + array_len(&self.freezes, version)
236            + unknown_tagged_fields_len(&self.unknown_tagged_fields)
237    }
238}
239
240impl Decode<'_> for DescribeTopicFreezesResponse {
241    fn decode<B: Buf>(buf: &mut B, version: i16) -> Result<Self, ProtocolError> {
242        check_version(API_KEY, version, MIN_VERSION, MAX_VERSION)?;
243        Ok(Self {
244            throttle_time_ms: get_i32(buf)?,
245            error_code: get_i16(buf)?,
246            error_message: get_compact_nullable_string_owned(buf)?,
247            freezes: decode_array(buf, version)?,
248            unknown_tagged_fields: read_unknown_tagged_fields(buf)?,
249        })
250    }
251}
252
253#[cfg(test)]
254impl DescribeTopicFreezesRequest {
255    /// Builds a request with every field set, for the round-trip tests.
256    fn populated() -> Self {
257        Self {
258            scope_filter: Some("tenant-a.".to_owned()),
259            pattern_type_filter: super::PATTERN_TYPE_PREFIXED,
260            unknown_tagged_fields: sample_tagged_fields(),
261        }
262    }
263}
264
265#[cfg(test)]
266impl DescribeTopicFreezesResponse {
267    /// Builds a response with every field set, for the round-trip tests.
268    fn populated() -> Self {
269        Self {
270            throttle_time_ms: 3,
271            error_code: 0,
272            error_message: None,
273            freezes: vec![
274                DescribedTopicFreeze {
275                    scope: "orders".to_owned(),
276                    pattern_type: PATTERN_TYPE_LITERAL,
277                    reason: "DR cutover".to_owned(),
278                    set_by: "User:alice".to_owned(),
279                    set_at_ms: 1_770_000_000_000,
280                    proposal_id: Uuid::ZERO,
281                    key_id: "alice-yubi".to_owned(),
282                    signature: (0..64).collect(),
283                    unknown_tagged_fields: sample_tagged_fields(),
284                },
285                DescribedTopicFreeze {
286                    scope: "tenant-a.".to_owned(),
287                    pattern_type: super::PATTERN_TYPE_PREFIXED,
288                    reason: "tenant offboarding".to_owned(),
289                    set_by: "User:bob".to_owned(),
290                    set_at_ms: 1_770_000_100_000,
291                    ..DescribedTopicFreeze::default()
292                },
293            ],
294            unknown_tagged_fields: sample_tagged_fields(),
295        }
296    }
297}
298
299#[cfg(test)]
300mod tests {
301    use assert2::assert;
302
303    use super::*;
304    use crate::krabka::{
305        freeze::PATTERN_TYPE_PREFIXED,
306        test_support::{reject_truncated, roundtrip_case, unsupported_versions},
307    };
308
309    #[test]
310    fn requests_roundtrip() {
311        let cases = [
312            (
313                "default asks for every entry",
314                DescribeTopicFreezesRequest::default(),
315            ),
316            (
317                "prefixed scope filter",
318                DescribeTopicFreezesRequest::populated(),
319            ),
320            (
321                "literal scope filter",
322                DescribeTopicFreezesRequest {
323                    scope_filter: Some("orders".to_owned()),
324                    pattern_type_filter: PATTERN_TYPE_LITERAL,
325                    unknown_tagged_fields: UnknownTaggedFields::default(),
326                },
327            ),
328            (
329                "unknown pattern type reads every entry",
330                DescribeTopicFreezesRequest {
331                    scope_filter: None,
332                    pattern_type_filter: 0,
333                    unknown_tagged_fields: UnknownTaggedFields::default(),
334                },
335            ),
336        ];
337        for (label, request) in &cases {
338            for version in MIN_VERSION..=MAX_VERSION {
339                roundtrip_case(label, request, version);
340            }
341        }
342    }
343
344    #[test]
345    fn responses_roundtrip() {
346        let cases = [
347            ("default", DescribeTopicFreezesResponse::default()),
348            (
349                "signed and unsigned entries",
350                DescribeTopicFreezesResponse::populated(),
351            ),
352            (
353                "refusal with no entries",
354                DescribeTopicFreezesResponse {
355                    throttle_time_ms: 11,
356                    error_code: 31,
357                    error_message: Some("cluster authorization failed".to_owned()),
358                    freezes: Vec::new(),
359                    unknown_tagged_fields: UnknownTaggedFields::default(),
360                },
361            ),
362        ];
363        for (label, response) in &cases {
364            for version in MIN_VERSION..=MAX_VERSION {
365                roundtrip_case(label, response, version);
366            }
367        }
368    }
369
370    #[test]
371    fn no_filter_is_the_request_default() {
372        assert!(
373            DescribeTopicFreezesRequest::default()
374                == DescribeTopicFreezesRequest {
375                    scope_filter: None,
376                    pattern_type_filter: PATTERN_TYPE_ANY,
377                    unknown_tagged_fields: UnknownTaggedFields::default(),
378                }
379        );
380    }
381
382    #[test]
383    fn an_unsigned_entry_carries_an_empty_signature() {
384        let response = DescribeTopicFreezesResponse::populated();
385        let unsigned = &response.freezes[1];
386        assert!(unsigned.key_id.is_empty());
387        assert!(unsigned.signature.is_empty());
388        assert!(unsigned.pattern_type == PATTERN_TYPE_PREFIXED);
389    }
390
391    #[test]
392    fn truncated_input_is_rejected() {
393        reject_truncated(&DescribeTopicFreezesRequest::populated(), MIN_VERSION);
394        reject_truncated(&DescribeTopicFreezesResponse::populated(), MIN_VERSION);
395    }
396
397    #[test]
398    fn rejects_unsupported_versions() {
399        for version in unsupported_versions(MIN_VERSION, MAX_VERSION) {
400            let request = DescribeTopicFreezesRequest::populated();
401            assert!(matches!(
402                request.encode(&mut Vec::<u8>::new(), version),
403                Err(ProtocolError::UnsupportedVersion {
404                    api_key: API_KEY,
405                    ..
406                })
407            ));
408            let response = DescribeTopicFreezesResponse::populated();
409            assert!(matches!(
410                response.encode(&mut Vec::<u8>::new(), version),
411                Err(ProtocolError::UnsupportedVersion {
412                    api_key: API_KEY,
413                    ..
414                })
415            ));
416        }
417    }
418}
krabka_protocol/krabka/freeze/describe_topic_freezes.rs

krabka_protocol/krabka/freeze/
describe_topic_freezes.rs
