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

krabka_protocol/krabka/freeze/
describe_topic_freezes.rs
