rocketmq_security_api/
maintenance.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
15//! Runtime-neutral, fail-closed authorization for maintenance operations.
16
17use std::collections::BTreeSet;
18use std::fmt;
19
20use serde::Deserialize;
21use serde::Serialize;
22
23use crate::MaintenancePolicyRule;
24use crate::Secret;
25use crate::SecurityContractViolation;
26
27/// Current maintenance policy schema.
28pub const MAINTENANCE_POLICY_SCHEMA_VERSION: u16 = 1;
29
30/// A privileged production operation controlled by the maintenance policy.
31#[derive(Clone, Copy, Debug, Deserialize, Eq, Hash, Ord, PartialEq, PartialOrd, Serialize)]
32#[serde(rename_all = "snake_case")]
33pub enum MaintenanceCapability {
34    /// Create, verify, or restore-verify a release checkpoint set.
35    ReleaseCheckpoint,
36}
37
38/// A role that can be bound by the maintenance policy.
39#[derive(Clone, Copy, Debug, Deserialize, Eq, Hash, Ord, PartialEq, PartialOrd, Serialize)]
40#[serde(rename_all = "snake_case")]
41pub enum MaintenanceRole {
42    /// An ordinary administrator. This role never implies release checkpoint access.
43    Administrator,
44    /// A separately delegated production release operator.
45    ReleaseOperator,
46}
47
48/// Classification applied to every maintenance request before dispatch.
49#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
50#[serde(rename_all = "snake_case")]
51pub enum MaintenanceRequestClass {
52    /// A fail-closed operation that requires a dedicated role and capability.
53    PrivilegedMaintenance,
54}
55
56/// Bounded resources available to one authorized checkpoint operation.
57#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
58#[serde(deny_unknown_fields, rename_all = "snake_case")]
59pub struct MaintenanceResourceBudget {
60    /// Maximum uncompressed bytes accepted for one checkpoint artifact.
61    pub max_checkpoint_bytes: u64,
62    /// Maximum number of Store members in one checkpoint set.
63    pub max_store_members: u32,
64    /// Maximum number of maintenance operations admitted at once.
65    pub max_concurrent_operations: u32,
66}
67
68/// A principal-to-role binding.
69#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
70#[serde(deny_unknown_fields, rename_all = "snake_case")]
71pub struct MaintenancePrincipalBinding {
72    /// Authenticated access-key identity.
73    pub principal: String,
74    /// Explicit maintenance roles assigned to this identity.
75    pub roles: BTreeSet<MaintenanceRole>,
76}
77
78/// Capabilities delegated to one maintenance role.
79#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
80#[serde(deny_unknown_fields, rename_all = "snake_case")]
81pub struct MaintenanceRoleGrant {
82    /// Role receiving the capability set.
83    pub role: MaintenanceRole,
84    /// Capabilities granted to the role.
85    pub capabilities: BTreeSet<MaintenanceCapability>,
86}
87
88/// Versioned fail-closed maintenance policy.
89#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
90#[serde(deny_unknown_fields, rename_all = "snake_case")]
91pub struct MaintenancePolicy {
92    /// Policy schema version.
93    pub schema_version: u16,
94    /// Stable policy identifier used by audit records.
95    pub policy_id: String,
96    /// Monotonically increasing policy version.
97    pub policy_version: u64,
98    /// Authentication must remain enabled for maintenance.
99    pub require_authentication: bool,
100    /// Authorization must remain enabled for maintenance.
101    pub require_authorization: bool,
102    /// Every state-changing request must carry a non-zero fencing token.
103    pub require_fencing_token: bool,
104    /// Maximum allowed request lifetime from authorization time.
105    pub max_request_lifetime_millis: u64,
106    /// Resource budget enforced by checkpoint coordinators.
107    pub resource_budget: MaintenanceResourceBudget,
108    /// Authenticated identities and their independent maintenance roles.
109    pub principal_bindings: Vec<MaintenancePrincipalBinding>,
110    /// Role-to-capability grants.
111    pub role_grants: Vec<MaintenanceRoleGrant>,
112}
113
114impl MaintenancePolicy {
115    /// Validates every policy invariant without granting authorization.
116    ///
117    /// # Errors
118    ///
119    /// Returns [`SecurityContractViolation`] when the policy could authorize an
120    /// operation without authentication, authorization, fencing, or a bounded
121    /// resource budget.
122    pub fn validate(&self) -> Result<(), SecurityContractViolation> {
123        validate_policy(self)
124    }
125
126    /// Consumes a completely validated policy for use by an authorizer.
127    ///
128    /// # Errors
129    ///
130    /// Returns [`SecurityContractViolation`] for the same fail-closed invariants
131    /// checked by [`Self::validate`].
132    pub fn into_validated(self) -> Result<ValidatedMaintenancePolicy, SecurityContractViolation> {
133        validate_policy(&self)?;
134        Ok(ValidatedMaintenancePolicy(self))
135    }
136
137    fn roles_for(&self, principal: &str) -> Option<&BTreeSet<MaintenanceRole>> {
138        self.principal_bindings
139            .iter()
140            .find(|binding| binding.principal == principal)
141            .map(|binding| &binding.roles)
142    }
143
144    fn role_allows(&self, role: MaintenanceRole, capability: MaintenanceCapability) -> bool {
145        self.role_grants
146            .iter()
147            .find(|grant| grant.role == role)
148            .is_some_and(|grant| grant.capabilities.contains(&capability))
149    }
150}
151
152/// A maintenance policy whose complete semantic contract has been validated.
153#[derive(Clone, Debug)]
154pub struct ValidatedMaintenancePolicy(MaintenancePolicy);
155
156impl ValidatedMaintenancePolicy {
157    /// Returns the immutable validated policy.
158    pub const fn policy(&self) -> &MaintenancePolicy {
159        &self.0
160    }
161}
162
163/// Authenticated request facts supplied by a service composition root.
164#[derive(Clone, Eq, PartialEq)]
165pub struct MaintenanceAuthorizationContext {
166    /// Whether the authentication runtime is enabled.
167    pub authentication_enabled: bool,
168    /// Whether authorization is enabled for the service.
169    pub authorization_enabled: bool,
170    /// Identity produced by successful credential verification.
171    pub principal: Option<String>,
172    /// Privileged classification assigned by the protocol router.
173    pub request_class: MaintenanceRequestClass,
174    /// Capability requested by the operation.
175    pub capability: MaintenanceCapability,
176    /// Absolute request deadline.
177    pub deadline_unix_millis: u64,
178    /// Non-zero lease fencing token.
179    pub fencing_token: Option<u64>,
180}
181
182impl fmt::Debug for MaintenanceAuthorizationContext {
183    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
184        formatter
185            .debug_struct("MaintenanceAuthorizationContext")
186            .field("authentication_enabled", &self.authentication_enabled)
187            .field("authorization_enabled", &self.authorization_enabled)
188            .field("principal", &Secret::new(self.principal.as_deref()))
189            .field("request_class", &self.request_class)
190            .field("capability", &self.capability)
191            .field("deadline_unix_millis", &self.deadline_unix_millis)
192            .field("fencing_token", &Secret::new(self.fencing_token))
193            .finish()
194    }
195}
196
197/// Auditable authorization result carried into a maintenance operation.
198///
199/// All fields are private and there is intentionally no unchecked constructor.
200/// A grant can only be produced by [`MaintenanceAuthorizer::authorize`].
201#[derive(Clone, Eq, PartialEq)]
202pub struct MaintenanceAuthorizationGrant {
203    principal: String,
204    role: MaintenanceRole,
205    capability: MaintenanceCapability,
206    policy_version: u64,
207    deadline_unix_millis: u64,
208    fencing_token: u64,
209    resource_budget: MaintenanceResourceBudget,
210}
211
212impl fmt::Debug for MaintenanceAuthorizationGrant {
213    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
214        formatter
215            .debug_struct("MaintenanceAuthorizationGrant")
216            .field("principal", &Secret::new(&self.principal))
217            .field("role", &self.role)
218            .field("capability", &self.capability)
219            .field("policy_version", &self.policy_version)
220            .field("deadline_unix_millis", &self.deadline_unix_millis)
221            .field("fencing_token", &Secret::new(self.fencing_token))
222            .field("resource_budget", &self.resource_budget)
223            .finish()
224    }
225}
226
227impl MaintenanceAuthorizationGrant {
228    /// Returns the authenticated identity.
229    pub fn principal(&self) -> &str {
230        &self.principal
231    }
232
233    /// Returns the dedicated maintenance role.
234    pub const fn role(&self) -> MaintenanceRole {
235        self.role
236    }
237
238    /// Returns the authorized capability.
239    pub const fn capability(&self) -> MaintenanceCapability {
240        self.capability
241    }
242
243    /// Returns the policy version used for the decision.
244    pub const fn policy_version(&self) -> u64 {
245        self.policy_version
246    }
247
248    /// Returns the absolute request deadline.
249    pub const fn deadline_unix_millis(&self) -> u64 {
250        self.deadline_unix_millis
251    }
252
253    /// Returns the non-zero lease fencing token.
254    pub const fn fencing_token(&self) -> u64 {
255        self.fencing_token
256    }
257
258    /// Returns the resource budget pinned by the authorizing policy.
259    pub const fn resource_budget(&self) -> &MaintenanceResourceBudget {
260        &self.resource_budget
261    }
262}
263
264/// Stateless fail-closed maintenance authorization engine.
265#[derive(Clone, Debug)]
266pub struct MaintenanceAuthorizer {
267    policy: ValidatedMaintenancePolicy,
268}
269
270impl MaintenanceAuthorizer {
271    /// Creates an authorizer from a policy that has already passed validation.
272    pub fn new(policy: impl Into<ValidatedMaintenancePolicy>) -> Self {
273        Self { policy: policy.into() }
274    }
275
276    /// Returns the validated policy used for decisions and resource budgets.
277    pub const fn policy(&self) -> &MaintenancePolicy {
278        self.policy.policy()
279    }
280
281    /// Authorizes one privileged maintenance request.
282    ///
283    /// # Errors
284    ///
285    /// Returns [`MaintenanceAuthorizationDenial`] for a missing context, disabled
286    /// auth, anonymous or unbound identity, ordinary administrator, absent
287    /// capability, invalid deadline, or missing fencing token.
288    pub fn authorize(
289        &self,
290        context: Option<&MaintenanceAuthorizationContext>,
291        now_unix_millis: u64,
292    ) -> Result<MaintenanceAuthorizationGrant, MaintenanceAuthorizationDenial> {
293        let context = context.ok_or(MaintenanceAuthorizationDenial::MissingAuthorizationContext)?;
294        if !context.authentication_enabled {
295            return Err(MaintenanceAuthorizationDenial::AuthenticationDisabled);
296        }
297        if !context.authorization_enabled {
298            return Err(MaintenanceAuthorizationDenial::AuthorizationDisabled);
299        }
300        if context.request_class != MaintenanceRequestClass::PrivilegedMaintenance {
301            return Err(MaintenanceAuthorizationDenial::InvalidRequestClass);
302        }
303        let principal = context
304            .principal
305            .as_deref()
306            .filter(|principal| !principal.trim().is_empty())
307            .ok_or(MaintenanceAuthorizationDenial::Anonymous)?;
308        let roles = self
309            .policy()
310            .roles_for(principal)
311            .ok_or(MaintenanceAuthorizationDenial::PrincipalUnbound)?;
312        if !roles.contains(&MaintenanceRole::ReleaseOperator) {
313            return Err(MaintenanceAuthorizationDenial::MissingRole(
314                MaintenanceRole::ReleaseOperator,
315            ));
316        }
317        if !self
318            .policy()
319            .role_allows(MaintenanceRole::ReleaseOperator, context.capability)
320        {
321            return Err(MaintenanceAuthorizationDenial::CapabilityDenied(context.capability));
322        }
323        if context.deadline_unix_millis <= now_unix_millis {
324            return Err(MaintenanceAuthorizationDenial::DeadlineExpired);
325        }
326        let lifetime = context.deadline_unix_millis - now_unix_millis;
327        if lifetime > self.policy().max_request_lifetime_millis {
328            return Err(MaintenanceAuthorizationDenial::DeadlineTooFar {
329                requested_millis: lifetime,
330                maximum_millis: self.policy().max_request_lifetime_millis,
331            });
332        }
333        let fencing_token = context
334            .fencing_token
335            .filter(|token| *token != 0)
336            .ok_or(MaintenanceAuthorizationDenial::MissingFencingToken)?;
337
338        Ok(MaintenanceAuthorizationGrant {
339            principal: principal.to_string(),
340            role: MaintenanceRole::ReleaseOperator,
341            capability: context.capability,
342            policy_version: self.policy().policy_version,
343            deadline_unix_millis: context.deadline_unix_millis,
344            fencing_token,
345            resource_budget: self.policy().resource_budget.clone(),
346        })
347    }
348}
349
350/// Fail-closed maintenance authorization denial.
351///
352/// Denials are normal outcomes rather than operational errors. Variants retain
353/// only closed, non-sensitive categories and never store the principal.
354#[derive(Clone, Debug, Eq, PartialEq)]
355pub enum MaintenanceAuthorizationDenial {
356    /// Represents the missing authorization context case.
357    MissingAuthorizationContext,
358    /// Represents the authentication disabled case.
359    AuthenticationDisabled,
360    /// Represents the authorization disabled case.
361    AuthorizationDisabled,
362    /// Represents the invalid request class case.
363    InvalidRequestClass,
364    /// Represents the anonymous case.
365    Anonymous,
366    /// Represents the principal unbound case.
367    PrincipalUnbound,
368    /// Represents the missing role case.
369    MissingRole(MaintenanceRole),
370    /// Represents the capability denied case.
371    CapabilityDenied(MaintenanceCapability),
372    /// Represents the deadline expired case.
373    DeadlineExpired,
374    /// Represents the deadline too far case.
375    DeadlineTooFar {
376        /// The requested duration in milliseconds.
377        requested_millis: u64,
378        /// The maximum duration in milliseconds.
379        maximum_millis: u64,
380    },
381    /// Represents the missing fencing token case.
382    MissingFencingToken,
383}
384
385fn validate_policy(policy: &MaintenancePolicy) -> Result<(), SecurityContractViolation> {
386    if policy.schema_version != MAINTENANCE_POLICY_SCHEMA_VERSION {
387        return invalid_policy(MaintenancePolicyRule::SchemaVersion);
388    }
389    if !is_canonical_identifier(&policy.policy_id) {
390        return invalid_policy(MaintenancePolicyRule::PolicyIdentifier);
391    }
392    if policy.policy_version == 0 {
393        return invalid_policy(MaintenancePolicyRule::PolicyVersion);
394    }
395    if !policy.require_authentication {
396        return invalid_policy(MaintenancePolicyRule::AuthenticationRequired);
397    }
398    if !policy.require_authorization {
399        return invalid_policy(MaintenancePolicyRule::AuthorizationRequired);
400    }
401    if !policy.require_fencing_token {
402        return invalid_policy(MaintenancePolicyRule::FencingTokenRequired);
403    }
404    if !(1_000..=86_400_000).contains(&policy.max_request_lifetime_millis) {
405        return invalid_policy(MaintenancePolicyRule::RequestLifetime);
406    }
407    if policy.resource_budget.max_checkpoint_bytes == 0
408        || policy.resource_budget.max_store_members == 0
409        || policy.resource_budget.max_concurrent_operations == 0
410    {
411        return invalid_policy(MaintenancePolicyRule::ResourceLimits);
412    }
413
414    let mut principals = BTreeSet::new();
415    let mut has_release_operator = false;
416    for binding in &policy.principal_bindings {
417        if !is_canonical_principal(&binding.principal) {
418            return invalid_policy(MaintenancePolicyRule::PrincipalIdentifier);
419        }
420        if !principals.insert(binding.principal.as_str()) {
421            return invalid_policy(MaintenancePolicyRule::UniquePrincipalBinding);
422        }
423        if binding.roles.is_empty() {
424            return invalid_policy(MaintenancePolicyRule::PrincipalRoleRequired);
425        }
426        has_release_operator |= binding.roles.contains(&MaintenanceRole::ReleaseOperator);
427    }
428    if !has_release_operator {
429        return invalid_policy(MaintenancePolicyRule::ReleaseOperatorRequired);
430    }
431
432    let mut granted_roles = BTreeSet::new();
433    let mut release_checkpoint_granted = false;
434    for grant in &policy.role_grants {
435        if !granted_roles.insert(grant.role) {
436            return invalid_policy(MaintenancePolicyRule::UniqueRoleGrant);
437        }
438        if grant.capabilities.is_empty() {
439            return invalid_policy(MaintenancePolicyRule::RoleCapabilityRequired);
440        }
441        if grant.role == MaintenanceRole::Administrator
442            && grant.capabilities.contains(&MaintenanceCapability::ReleaseCheckpoint)
443        {
444            return invalid_policy(MaintenancePolicyRule::AdministratorCheckpointForbidden);
445        }
446        release_checkpoint_granted |= grant.role == MaintenanceRole::ReleaseOperator
447            && grant.capabilities.contains(&MaintenanceCapability::ReleaseCheckpoint);
448    }
449    if !release_checkpoint_granted {
450        return invalid_policy(MaintenancePolicyRule::ReleaseCheckpointRequired);
451    }
452    Ok(())
453}
454
455fn invalid_policy<T>(rule: MaintenancePolicyRule) -> Result<T, SecurityContractViolation> {
456    Err(SecurityContractViolation::MaintenancePolicy { rule })
457}
458
459fn is_canonical_identifier(value: &str) -> bool {
460    !value.is_empty()
461        && value.len() <= 128
462        && value
463            .bytes()
464            .all(|byte| byte.is_ascii_lowercase() || byte.is_ascii_digit() || matches!(byte, b'.' | b'_' | b'-'))
465}
466
467fn is_canonical_principal(value: &str) -> bool {
468    !value.is_empty()
469        && value.len() <= 256
470        && value
471            .bytes()
472            .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'.' | b'_' | b'-' | b'@' | b':' | b'/'))
473}
474
475#[cfg(test)]
476mod tests {
477    use super::*;
478
479    fn policy() -> MaintenancePolicy {
480        MaintenancePolicy {
481            schema_version: MAINTENANCE_POLICY_SCHEMA_VERSION,
482            policy_id: "rocketmq.production-maintenance".to_string(),
483            policy_version: 7,
484            require_authentication: true,
485            require_authorization: true,
486            require_fencing_token: true,
487            max_request_lifetime_millis: 30_000,
488            resource_budget: MaintenanceResourceBudget {
489                max_checkpoint_bytes: 1024,
490                max_store_members: 3,
491                max_concurrent_operations: 1,
492            },
493            principal_bindings: vec![MaintenancePrincipalBinding {
494                principal: "release-operator".to_string(),
495                roles: BTreeSet::from([MaintenanceRole::ReleaseOperator]),
496            }],
497            role_grants: vec![MaintenanceRoleGrant {
498                role: MaintenanceRole::ReleaseOperator,
499                capabilities: BTreeSet::from([MaintenanceCapability::ReleaseCheckpoint]),
500            }],
501        }
502    }
503
504    fn context() -> MaintenanceAuthorizationContext {
505        MaintenanceAuthorizationContext {
506            authentication_enabled: true,
507            authorization_enabled: true,
508            principal: Some("release-operator".to_string()),
509            request_class: MaintenanceRequestClass::PrivilegedMaintenance,
510            capability: MaintenanceCapability::ReleaseCheckpoint,
511            deadline_unix_millis: 120_000,
512            fencing_token: Some(42),
513        }
514    }
515
516    #[test]
517    fn grant_pins_policy_deadline_fencing_and_budget() {
518        let authorizer = MaintenanceAuthorizer::new(policy().into_validated().expect("valid policy"));
519        let grant = authorizer.authorize(Some(&context()), 100_000).expect("authorized");
520
521        assert_eq!(grant.policy_version(), 7);
522        assert_eq!(grant.deadline_unix_millis(), 120_000);
523        assert_eq!(grant.fencing_token(), 42);
524        assert_eq!(grant.resource_budget().max_checkpoint_bytes, 1024);
525    }
526
527    #[test]
528    fn grant_cannot_bypass_fail_closed_inputs() {
529        let authorizer = MaintenanceAuthorizer::new(policy().into_validated().expect("valid policy"));
530        let mut request = context();
531        request.authentication_enabled = false;
532        assert_eq!(
533            authorizer.authorize(Some(&request), 100_000),
534            Err(MaintenanceAuthorizationDenial::AuthenticationDisabled)
535        );
536        request.authentication_enabled = true;
537        request.fencing_token = None;
538        assert_eq!(
539            authorizer.authorize(Some(&request), 100_000),
540            Err(MaintenanceAuthorizationDenial::MissingFencingToken)
541        );
542        request.fencing_token = Some(42);
543        request.deadline_unix_millis = 100_000;
544        assert_eq!(
545            authorizer.authorize(Some(&request), 100_000),
546            Err(MaintenanceAuthorizationDenial::DeadlineExpired)
547        );
548    }
549
550    #[test]
551    fn policy_and_denial_errors_never_retain_principals() {
552        let sentinel = "principal-secret-sentinel";
553        let mut invalid = policy();
554        invalid.principal_bindings = vec![
555            MaintenancePrincipalBinding {
556                principal: sentinel.to_string(),
557                roles: BTreeSet::from([MaintenanceRole::ReleaseOperator]),
558            },
559            MaintenancePrincipalBinding {
560                principal: sentinel.to_string(),
561                roles: BTreeSet::from([MaintenanceRole::ReleaseOperator]),
562            },
563        ];
564        let error = invalid.validate().expect_err("duplicate principals must fail closed");
565        assert_eq!(
566            error,
567            SecurityContractViolation::MaintenancePolicy {
568                rule: MaintenancePolicyRule::UniquePrincipalBinding,
569            }
570        );
571        assert!(!error.to_string().contains(sentinel));
572        assert!(!format!("{error:?}").contains(sentinel));
573
574        let authorizer = MaintenanceAuthorizer::new(policy().into_validated().expect("valid policy"));
575        let mut request = context();
576        request.principal = Some(sentinel.to_string());
577        let denial = authorizer
578            .authorize(Some(&request), 100_000)
579            .expect_err("unbound principal must be denied");
580        assert_eq!(denial, MaintenanceAuthorizationDenial::PrincipalUnbound);
581        assert!(!format!("{denial:?}").contains(sentinel));
582    }
583}
rocketmq_security_api/maintenance.rs

rocketmq_security_api/
maintenance.rs
