feat(targets): add extension hook diagnostics contracts (#3390)

Co-authored-by: houseme <housemecn@gmail.com>

crates/targets/src/runtime/mod.rs

@@ -13,6 +13,8 @@
 // limitations under the License.
 
 pub mod adapter;
+pub mod ops_diagnostics;
+pub mod s3_hooks;
 pub mod sidecar;
 pub mod sidecar_protocol;
 pub mod tls;

crates/targets/src/runtime/ops_diagnostics.rs

@@ -0,0 +1,214 @@
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use std::collections::BTreeMap;
+
+use rustfs_extension_schema::{
+    ExtensionContractError, ExtensionKind, ExtensionSchema, OPS_DIAGNOSTICS_CAPABILITY, OpsDiagnosticSurface,
+    OpsDiagnosticsContract, validate_ops_diagnostics_contract,
+};
+use thiserror::Error;
+
+#[derive(Debug, Error, PartialEq, Eq)]
+pub enum OpsDiagnosticsRegistryError {
+    #[error(transparent)]
+    InvalidContract(#[from] ExtensionContractError),
+
+    #[error("extension {extension_id} is {kind:?}, not an ops diagnostics extension")]
+    UnsupportedExtensionKind { extension_id: String, kind: ExtensionKind },
+
+    #[error("ops diagnostics extension {extension_id} is missing capability {capability}")]
+    MissingCapability { extension_id: String, capability: &'static str },
+}
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct OpsDiagnosticsRegistration {
+    pub extension_id: String,
+    pub surface: OpsDiagnosticSurface,
+}
+
+#[derive(Debug, Default, Clone, PartialEq, Eq)]
+pub struct OpsDiagnosticsRegistry {
+    registrations: BTreeMap<OpsDiagnosticSurface, Vec<OpsDiagnosticsRegistration>>,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub struct OpsDiagnosticsReadRequest<'a> {
+    pub surface: OpsDiagnosticSurface,
+    pub capability: &'a str,
+    pub admin_action_authorized: bool,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum OpsDiagnosticsAccessDecision {
+    AllowReadOnly,
+    DenyMissingAdminAction,
+    DenyMissingCapability,
+    DenyUnknownSurface,
+}
+
+impl OpsDiagnosticsRegistry {
+    pub fn new() -> Self {
+        Self::default()
+    }
+
+    pub fn register_schema(
+        &mut self,
+        schema: &ExtensionSchema,
+        contract: &OpsDiagnosticsContract,
+    ) -> Result<(), OpsDiagnosticsRegistryError> {
+        if schema.kind != ExtensionKind::OpsDiagnostics {
+            return Err(OpsDiagnosticsRegistryError::UnsupportedExtensionKind {
+                extension_id: schema.extension_id.clone(),
+                kind: schema.kind,
+            });
+        }
+
+        if !schema
+            .capabilities
+            .iter()
+            .any(|capability| capability.as_str() == OPS_DIAGNOSTICS_CAPABILITY)
+        {
+            return Err(OpsDiagnosticsRegistryError::MissingCapability {
+                extension_id: schema.extension_id.clone(),
+                capability: OPS_DIAGNOSTICS_CAPABILITY,
+            });
+        }
+
+        validate_ops_diagnostics_contract(contract)?;
+
+        for surface in &contract.surfaces {
+            self.registrations
+                .entry(*surface)
+                .or_default()
+                .push(OpsDiagnosticsRegistration {
+                    extension_id: schema.extension_id.clone(),
+                    surface: *surface,
+                });
+        }
+
+        Ok(())
+    }
+
+    pub fn registered_surface_count(&self) -> usize {
+        self.registrations.values().map(Vec::len).sum()
+    }
+
+    pub fn registrations_for(&self, surface: OpsDiagnosticSurface) -> impl Iterator<Item = &OpsDiagnosticsRegistration> {
+        self.registrations.get(&surface).into_iter().flatten()
+    }
+
+    pub fn authorize_read(&self, request: OpsDiagnosticsReadRequest<'_>) -> OpsDiagnosticsAccessDecision {
+        if !self.registrations.contains_key(&request.surface) {
+            return OpsDiagnosticsAccessDecision::DenyUnknownSurface;
+        }
+
+        if !request.admin_action_authorized {
+            return OpsDiagnosticsAccessDecision::DenyMissingAdminAction;
+        }
+
+        if request.capability != OPS_DIAGNOSTICS_CAPABILITY {
+            return OpsDiagnosticsAccessDecision::DenyMissingCapability;
+        }
+
+        OpsDiagnosticsAccessDecision::AllowReadOnly
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::{OpsDiagnosticsAccessDecision, OpsDiagnosticsReadRequest, OpsDiagnosticsRegistry, OpsDiagnosticsRegistryError};
+    use crate::{builtin_ops_diagnostics_contract, builtin_ops_diagnostics_extension_schema};
+    use rustfs_extension_schema::{ExtensionContractError, ExtensionKind, OPS_DIAGNOSTICS_CAPABILITY, OpsDiagnosticSurface};
+
+    #[test]
+    fn default_registry_denies_unknown_diagnostic_surface() {
+        let registry = OpsDiagnosticsRegistry::new();
+
+        assert_eq!(
+            registry.authorize_read(OpsDiagnosticsReadRequest {
+                surface: OpsDiagnosticSurface::Health,
+                capability: OPS_DIAGNOSTICS_CAPABILITY,
+                admin_action_authorized: true,
+            }),
+            OpsDiagnosticsAccessDecision::DenyUnknownSurface
+        );
+    }
+
+    #[test]
+    fn registered_ops_diagnostics_are_read_only_and_capability_limited() {
+        let mut registry = OpsDiagnosticsRegistry::new();
+        let schema = builtin_ops_diagnostics_extension_schema();
+        let contract = builtin_ops_diagnostics_contract();
+
+        registry
+            .register_schema(&schema, &contract)
+            .expect("builtin ops diagnostics contract should register");
+
+        assert_eq!(registry.registered_surface_count(), contract.surfaces.len());
+        assert_eq!(registry.registrations_for(OpsDiagnosticSurface::Metrics).count(), 1);
+        assert_eq!(
+            registry.authorize_read(OpsDiagnosticsReadRequest {
+                surface: OpsDiagnosticSurface::Health,
+                capability: OPS_DIAGNOSTICS_CAPABILITY,
+                admin_action_authorized: true,
+            }),
+            OpsDiagnosticsAccessDecision::AllowReadOnly
+        );
+        assert_eq!(
+            registry.authorize_read(OpsDiagnosticsReadRequest {
+                surface: OpsDiagnosticSurface::Health,
+                capability: "target.notify.v1",
+                admin_action_authorized: true,
+            }),
+            OpsDiagnosticsAccessDecision::DenyMissingCapability
+        );
+        assert_eq!(
+            registry.authorize_read(OpsDiagnosticsReadRequest {
+                surface: OpsDiagnosticSurface::Health,
+                capability: OPS_DIAGNOSTICS_CAPABILITY,
+                admin_action_authorized: false,
+            }),
+            OpsDiagnosticsAccessDecision::DenyMissingAdminAction
+        );
+    }
+
+    #[test]
+    fn rejects_non_diagnostics_schema_and_mutating_contracts() {
+        let mut registry = OpsDiagnosticsRegistry::new();
+        let mut schema = builtin_ops_diagnostics_extension_schema();
+        schema.kind = ExtensionKind::TargetPlugin;
+
+        assert_eq!(
+            registry
+                .register_schema(&schema, &builtin_ops_diagnostics_contract())
+                .expect_err("target plugin schema should not register as ops diagnostics"),
+            OpsDiagnosticsRegistryError::UnsupportedExtensionKind {
+                extension_id: "builtin:ops-diagnostics".to_string(),
+                kind: ExtensionKind::TargetPlugin
+            }
+        );
+
+        schema.kind = ExtensionKind::OpsDiagnostics;
+        let mut contract = builtin_ops_diagnostics_contract();
+        contract.mutates_object_data = true;
+
+        assert_eq!(
+            registry
+                .register_schema(&schema, &contract)
+                .expect_err("object mutation should be rejected"),
+            OpsDiagnosticsRegistryError::InvalidContract(ExtensionContractError::OpsDiagnosticsMutatesObjectData)
+        );
+    }
+}

crates/targets/src/runtime/s3_hooks.rs

@@ -0,0 +1,200 @@
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use std::collections::BTreeMap;
+
+use rustfs_extension_schema::{
+    ExtensionContractError, ExtensionKind, ExtensionSchema, S3_POST_AUTH_HOOK_CAPABILITY, S3HookContract, S3HookPoint,
+    validate_s3_hook_contract,
+};
+use thiserror::Error;
+
+#[derive(Debug, Error, PartialEq, Eq)]
+pub enum S3HookRegistryError {
+    #[error(transparent)]
+    InvalidContract(#[from] ExtensionContractError),
+
+    #[error("extension {extension_id} is {kind:?}, not an S3 hook")]
+    UnsupportedExtensionKind { extension_id: String, kind: ExtensionKind },
+
+    #[error("s3 hook extension {extension_id} is missing capability {capability}")]
+    MissingCapability { extension_id: String, capability: &'static str },
+}
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct S3HookRegistration {
+    pub extension_id: String,
+    pub hook_point: S3HookPoint,
+}
+
+#[derive(Debug, Default, Clone, PartialEq, Eq)]
+pub struct S3HookRegistry {
+    registrations: BTreeMap<S3HookPoint, Vec<S3HookRegistration>>,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub struct S3HookContext<'a> {
+    pub authenticated_principal: &'a str,
+    pub bucket: &'a str,
+    pub object: Option<&'a str>,
+}
+
+impl<'a> S3HookContext<'a> {
+    pub fn post_auth(authenticated_principal: &'a str, bucket: &'a str, object: Option<&'a str>) -> Option<Self> {
+        if authenticated_principal.trim().is_empty() {
+            return None;
+        }
+
+        Some(Self {
+            authenticated_principal,
+            bucket,
+            object,
+        })
+    }
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum S3HookDecision {
+    Continue,
+}
+
+impl S3HookRegistry {
+    pub fn new() -> Self {
+        Self::default()
+    }
+
+    pub fn register_schema(&mut self, schema: &ExtensionSchema, contract: &S3HookContract) -> Result<(), S3HookRegistryError> {
+        if schema.kind != ExtensionKind::S3Hook {
+            return Err(S3HookRegistryError::UnsupportedExtensionKind {
+                extension_id: schema.extension_id.clone(),
+                kind: schema.kind,
+            });
+        }
+
+        if !schema
+            .capabilities
+            .iter()
+            .any(|capability| capability.as_str() == S3_POST_AUTH_HOOK_CAPABILITY)
+        {
+            return Err(S3HookRegistryError::MissingCapability {
+                extension_id: schema.extension_id.clone(),
+                capability: S3_POST_AUTH_HOOK_CAPABILITY,
+            });
+        }
+
+        validate_s3_hook_contract(contract)?;
+
+        for hook_point in &contract.hook_points {
+            self.registrations.entry(*hook_point).or_default().push(S3HookRegistration {
+                extension_id: schema.extension_id.clone(),
+                hook_point: *hook_point,
+            });
+        }
+
+        Ok(())
+    }
+
+    pub fn is_empty(&self) -> bool {
+        self.registrations.values().all(Vec::is_empty)
+    }
+
+    pub fn registered_hook_count(&self) -> usize {
+        self.registrations.values().map(Vec::len).sum()
+    }
+
+    pub fn hooks_for(&self, hook_point: S3HookPoint) -> impl Iterator<Item = &S3HookRegistration> {
+        self.registrations.get(&hook_point).into_iter().flatten()
+    }
+
+    pub fn dispatch_post_auth(&self, _hook_point: S3HookPoint, _context: &S3HookContext<'_>) -> S3HookDecision {
+        S3HookDecision::Continue
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::{S3HookContext, S3HookDecision, S3HookRegistry, S3HookRegistryError};
+    use crate::{builtin_s3_hook_contract, builtin_s3_hook_extension_schema};
+    use rustfs_extension_schema::{ExtensionContractError, ExtensionKind, S3HookPoint};
+
+    #[test]
+    fn default_registry_has_identical_continue_behavior() {
+        let registry = S3HookRegistry::new();
+        let context = S3HookContext::post_auth("access-key", "photos", Some("2026/image.jpg"))
+            .expect("post-auth context should require an authenticated principal");
+
+        assert!(registry.is_empty());
+        assert_eq!(registry.registered_hook_count(), 0);
+        assert_eq!(
+            registry.dispatch_post_auth(S3HookPoint::PostAuthGetObject, &context),
+            S3HookDecision::Continue
+        );
+    }
+
+    #[test]
+    fn post_auth_context_rejects_missing_principal() {
+        assert!(S3HookContext::post_auth("", "photos", Some("2026/image.jpg")).is_none());
+    }
+
+    #[test]
+    fn registers_valid_post_auth_hook_contract_without_dispatch_side_effects() {
+        let mut registry = S3HookRegistry::new();
+        let schema = builtin_s3_hook_extension_schema();
+        let contract = builtin_s3_hook_contract();
+        let context = S3HookContext::post_auth("access-key", "photos", None).expect("post-auth context should be valid");
+
+        registry
+            .register_schema(&schema, &contract)
+            .expect("builtin s3 hook contract should register");
+
+        assert_eq!(registry.registered_hook_count(), contract.hook_points.len());
+        assert_eq!(
+            registry.hooks_for(S3HookPoint::PostAuthListObjects).count(),
+            1,
+            "registered hooks stay catalogued by allowlisted point"
+        );
+        assert_eq!(
+            registry.dispatch_post_auth(S3HookPoint::PostAuthListObjects, &context),
+            S3HookDecision::Continue
+        );
+    }
+
+    #[test]
+    fn rejects_non_s3_hook_schema_and_unsafe_contracts() {
+        let mut registry = S3HookRegistry::new();
+        let mut schema = builtin_s3_hook_extension_schema();
+        schema.kind = ExtensionKind::TargetPlugin;
+
+        assert_eq!(
+            registry
+                .register_schema(&schema, &builtin_s3_hook_contract())
+                .expect_err("target plugin schema should not register as s3 hook"),
+            S3HookRegistryError::UnsupportedExtensionKind {
+                extension_id: "builtin:s3-post-auth-hooks".to_string(),
+                kind: ExtensionKind::TargetPlugin
+            }
+        );
+
+        schema.kind = ExtensionKind::S3Hook;
+        let mut contract = builtin_s3_hook_contract();
+        contract.bypasses_iam = true;
+
+        assert_eq!(
+            registry
+                .register_schema(&schema, &contract)
+                .expect_err("IAM bypass should be rejected"),
+            S3HookRegistryError::InvalidContract(ExtensionContractError::S3HookBypassesIam)
+        );
+    }
+}