rocketmq_protocol/protocol/header/
maintenance_request_header.rs

1// Copyright 2023 The RocketMQ Rust Authors
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15use std::fmt;
16
17use cheetah_string::CheetahString;
18use rocketmq_error::Sensitive;
19use rocketmq_macros::RequestHeaderCodec;
20use serde::Deserialize;
21use serde::Serialize;
22
23/// Common header for every privileged maintenance request.
24///
25/// Authentication fields remain owned by the existing signed Remoting command.
26/// These fields bind the authenticated operation to a policy revision, absolute
27/// deadline, idempotency key, and Kubernetes Lease fencing token.
28#[derive(Clone, Deserialize, Eq, PartialEq, RequestHeaderCodec, Serialize)]
29#[header(type_id = "rocketmq_protocol::protocol::header::maintenance_request_header::MaintenanceRequestHeader")]
30#[serde(rename_all = "camelCase")]
31pub struct MaintenanceRequestHeader {
32    /// Idempotency key generated by the release coordinator.
33    #[header(required)]
34    pub operation_id: CheetahString,
35    /// Version of the maintenance policy used by the caller.
36    #[header(required)]
37    pub policy_version: u64,
38    /// Absolute Unix deadline in milliseconds.
39    #[header(required)]
40    pub deadline_unix_millis: u64,
41    /// Non-zero token derived from the current Kubernetes Lease revision.
42    #[header(required)]
43    pub fencing_token: u64,
44}
45
46impl fmt::Debug for MaintenanceRequestHeader {
47    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
48        formatter
49            .debug_struct("MaintenanceRequestHeader")
50            .field("operation_id", &Sensitive::new(&self.operation_id))
51            .field("policy_version", &self.policy_version)
52            .field("deadline_unix_millis", &self.deadline_unix_millis)
53            .field("fencing_token", &Sensitive::new(self.fencing_token))
54            .finish()
55    }
56}
57
58impl MaintenanceRequestHeader {
59    /// Validates request-local invariants before authorization.
60    ///
61    /// # Errors
62    ///
63    /// Returns a static reason when an idempotency key, policy version,
64    /// deadline, or fencing token is missing or invalid.
65    pub fn validate(&self) -> Result<(), &'static str> {
66        if self.operation_id.is_empty() {
67            return Err("operation_id cannot be empty");
68        }
69        if self.policy_version == 0 {
70            return Err("policy_version must be greater than zero");
71        }
72        if self.deadline_unix_millis == 0 {
73            return Err("deadline_unix_millis must be greater than zero");
74        }
75        if self.fencing_token == 0 {
76            return Err("fencing_token must be greater than zero");
77        }
78        Ok(())
79    }
80}
81
82#[cfg(test)]
83mod tests {
84    use std::collections::HashMap;
85
86    use super::MaintenanceRequestHeader;
87    use crate::protocol::command_custom_header::CommandCustomHeader;
88    use crate::protocol::command_custom_header::FromMap;
89    use cheetah_string::CheetahString;
90
91    #[test]
92    fn privileged_maintenance_header_round_trips_all_fencing_fields() {
93        let header = MaintenanceRequestHeader {
94            operation_id: "rollback-42-controller-snapshot".into(),
95            policy_version: 7,
96            deadline_unix_millis: 1_800_000_000_000,
97            fencing_token: 42,
98        };
99
100        let map = header.to_map().expect("encode maintenance header");
101        let decoded = <MaintenanceRequestHeader as FromMap>::from(&map).expect("decode maintenance header");
102
103        assert_eq!(decoded, header);
104        assert_eq!(
105            map.get(&CheetahString::from_static_str("fencingToken")),
106            Some(&CheetahString::from_static_str("42"))
107        );
108        assert!(decoded.validate().is_ok());
109    }
110
111    #[test]
112    fn privileged_maintenance_header_requires_every_field() {
113        let missing = HashMap::new();
114
115        assert!(<MaintenanceRequestHeader as FromMap>::from(&missing).is_err());
116    }
117
118    #[test]
119    fn rust_extension_preserves_the_full_unsigned_domain() {
120        let header = MaintenanceRequestHeader {
121            operation_id: "full-u64-domain".into(),
122            policy_version: u64::MAX,
123            deadline_unix_millis: u64::MAX,
124            fencing_token: u64::MAX,
125        };
126
127        let map = header.to_map().expect("encode full unsigned values");
128        let decoded = <MaintenanceRequestHeader as FromMap>::from(&map).expect("decode full unsigned values");
129
130        assert_eq!(decoded, header);
131    }
132
133    #[test]
134    fn privileged_maintenance_header_debug_redacts_sensitive_fields() {
135        let header = MaintenanceRequestHeader {
136            operation_id: "release-operation-secret".into(),
137            policy_version: 7,
138            deadline_unix_millis: 1_800_000_000_000,
139            fencing_token: 987_654_321,
140        };
141
142        let debug = format!("{header:?}");
143
144        assert!(debug.contains("<redacted>"));
145        assert!(!debug.contains("release-operation-secret"));
146        assert!(!debug.contains("987654321"));
147    }
148}
rocketmq_protocol/protocol/header/maintenance_request_header.rs

rocketmq_protocol/protocol/header/
maintenance_request_header.rs
