diff --git a/crates/extension-schema/src/lib.rs b/crates/extension-schema/src/lib.rs index 929fde040..2bf03f98b 100644 --- a/crates/extension-schema/src/lib.rs +++ b/crates/extension-schema/src/lib.rs @@ -18,6 +18,8 @@ use serde::{Deserialize, Serialize}; use thiserror::Error; pub const EXTENSION_SCHEMA_VERSION: &str = "rustfs.extension-schema.v1"; +pub const OPS_DIAGNOSTICS_CAPABILITY: &str = "ops.diagnostics.v1"; +pub const S3_POST_AUTH_HOOK_CAPABILITY: &str = "s3.hook.post_auth.v1"; #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)] #[serde(rename_all = "snake_case")] @@ -74,6 +76,47 @@ pub struct ExtensionSchema { pub disabled_by_default: bool, } +#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum S3HookPoint { + PostAuthGetObject, + PostAuthPutObject, + PostAuthDeleteObject, + PostAuthListObjects, +} + +impl S3HookPoint { + pub const fn is_post_auth(self) -> bool { + true + } +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct S3HookContract { + pub hook_points: Vec, + pub mutates_object_data: bool, + pub bypasses_iam: bool, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum OpsDiagnosticSurface { + Metrics, + Trace, + Profile, + Health, + Diagnostics, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct OpsDiagnosticsContract { + pub surfaces: Vec, + pub mutates_object_data: bool, + pub requires_admin_action: bool, +} + #[derive(Debug, Error, PartialEq, Eq)] pub enum ExtensionSchemaError { #[error("extension schema at index {index} has an empty extension id")] @@ -110,6 +153,33 @@ pub enum ExtensionSchemaError { DuplicateExtension { extension_id: String }, } +#[derive(Debug, Error, PartialEq, Eq)] +pub enum ExtensionContractError { + #[error("s3 hook contract must declare at least one hook point")] + EmptyS3HookPoints, + + #[error("s3 hook contract duplicates hook point {hook_point:?}")] + DuplicateS3HookPoint { hook_point: S3HookPoint }, + + #[error("s3 hook contract cannot mutate object data")] + S3HookMutatesObjectData, + + #[error("s3 hook contract cannot bypass IAM")] + S3HookBypassesIam, + + #[error("ops diagnostics contract must declare at least one surface")] + EmptyOpsDiagnosticSurfaces, + + #[error("ops diagnostics contract duplicates surface {surface:?}")] + DuplicateOpsDiagnosticSurface { surface: OpsDiagnosticSurface }, + + #[error("ops diagnostics contract cannot mutate object data")] + OpsDiagnosticsMutatesObjectData, + + #[error("ops diagnostics contract must require an admin action")] + OpsDiagnosticsMissingAdminAction, +} + pub fn validate_extension_schemas(schemas: &[ExtensionSchema]) -> Result<(), ExtensionSchemaError> { let mut extension_ids = BTreeSet::new(); @@ -188,11 +258,59 @@ pub fn validate_extension_schemas(schemas: &[ExtensionSchema]) -> Result<(), Ext Ok(()) } +pub fn validate_s3_hook_contract(contract: &S3HookContract) -> Result<(), ExtensionContractError> { + if contract.hook_points.is_empty() { + return Err(ExtensionContractError::EmptyS3HookPoints); + } + + let mut hook_points = BTreeSet::new(); + for hook_point in &contract.hook_points { + if !hook_points.insert(*hook_point) { + return Err(ExtensionContractError::DuplicateS3HookPoint { hook_point: *hook_point }); + } + } + + if contract.mutates_object_data { + return Err(ExtensionContractError::S3HookMutatesObjectData); + } + + if contract.bypasses_iam { + return Err(ExtensionContractError::S3HookBypassesIam); + } + + Ok(()) +} + +pub fn validate_ops_diagnostics_contract(contract: &OpsDiagnosticsContract) -> Result<(), ExtensionContractError> { + if contract.surfaces.is_empty() { + return Err(ExtensionContractError::EmptyOpsDiagnosticSurfaces); + } + + let mut surfaces = BTreeSet::new(); + for surface in &contract.surfaces { + if !surfaces.insert(*surface) { + return Err(ExtensionContractError::DuplicateOpsDiagnosticSurface { surface: *surface }); + } + } + + if contract.mutates_object_data { + return Err(ExtensionContractError::OpsDiagnosticsMutatesObjectData); + } + + if !contract.requires_admin_action { + return Err(ExtensionContractError::OpsDiagnosticsMissingAdminAction); + } + + Ok(()) +} + #[cfg(test)] mod tests { use super::{ - EXTENSION_SCHEMA_VERSION, ExtensionCapabilityRef, ExtensionKind, ExtensionRuntimeBoundary, ExtensionRuntimeContract, - ExtensionSchema, ExtensionSchemaError, validate_extension_schemas, + EXTENSION_SCHEMA_VERSION, ExtensionCapabilityRef, ExtensionContractError, ExtensionKind, ExtensionRuntimeBoundary, + ExtensionRuntimeContract, ExtensionSchema, ExtensionSchemaError, OPS_DIAGNOSTICS_CAPABILITY, OpsDiagnosticSurface, + OpsDiagnosticsContract, S3_POST_AUTH_HOOK_CAPABILITY, S3HookContract, S3HookPoint, validate_extension_schemas, + validate_ops_diagnostics_contract, validate_s3_hook_contract, }; use serde_json::json; @@ -256,7 +374,7 @@ mod tests { api_version: "rustfs.extension.v1".to_string(), boundary: ExtensionRuntimeBoundary::Builtin, }, - capabilities: vec![ExtensionCapabilityRef::new("ops.diagnostics.v1")], + capabilities: vec![ExtensionCapabilityRef::new(OPS_DIAGNOSTICS_CAPABILITY)], disabled_by_default: false, }]; @@ -348,4 +466,95 @@ mod tests { } ); } + + #[test] + fn validates_s3_post_auth_hook_contract() { + let contract = S3HookContract { + hook_points: vec![S3HookPoint::PostAuthGetObject, S3HookPoint::PostAuthPutObject], + mutates_object_data: false, + bypasses_iam: false, + }; + + assert!(contract.hook_points.iter().all(|hook_point| hook_point.is_post_auth())); + assert!(validate_s3_hook_contract(&contract).is_ok()); + assert_eq!(S3_POST_AUTH_HOOK_CAPABILITY, "s3.hook.post_auth.v1"); + } + + #[test] + fn rejects_s3_hooks_that_mutate_or_bypass_iam() { + let mut contract = S3HookContract { + hook_points: vec![S3HookPoint::PostAuthGetObject], + mutates_object_data: true, + bypasses_iam: false, + }; + + assert_eq!( + validate_s3_hook_contract(&contract).expect_err("object mutation should be rejected"), + ExtensionContractError::S3HookMutatesObjectData + ); + + contract.mutates_object_data = false; + contract.bypasses_iam = true; + + assert_eq!( + validate_s3_hook_contract(&contract).expect_err("IAM bypass should be rejected"), + ExtensionContractError::S3HookBypassesIam + ); + } + + #[test] + fn rejects_duplicate_s3_hook_points() { + let contract = S3HookContract { + hook_points: vec![S3HookPoint::PostAuthDeleteObject, S3HookPoint::PostAuthDeleteObject], + mutates_object_data: false, + bypasses_iam: false, + }; + + assert_eq!( + validate_s3_hook_contract(&contract).expect_err("duplicate hook points should fail validation"), + ExtensionContractError::DuplicateS3HookPoint { + hook_point: S3HookPoint::PostAuthDeleteObject + } + ); + } + + #[test] + fn validates_ops_diagnostics_contract() { + let contract = OpsDiagnosticsContract { + surfaces: vec![ + OpsDiagnosticSurface::Metrics, + OpsDiagnosticSurface::Trace, + OpsDiagnosticSurface::Profile, + OpsDiagnosticSurface::Health, + OpsDiagnosticSurface::Diagnostics, + ], + mutates_object_data: false, + requires_admin_action: true, + }; + + assert!(validate_ops_diagnostics_contract(&contract).is_ok()); + assert_eq!(OPS_DIAGNOSTICS_CAPABILITY, "ops.diagnostics.v1"); + } + + #[test] + fn rejects_ops_diagnostics_without_admin_action_or_read_only_contract() { + let mut contract = OpsDiagnosticsContract { + surfaces: vec![OpsDiagnosticSurface::Metrics], + mutates_object_data: false, + requires_admin_action: false, + }; + + assert_eq!( + validate_ops_diagnostics_contract(&contract).expect_err("admin action should be required"), + ExtensionContractError::OpsDiagnosticsMissingAdminAction + ); + + contract.requires_admin_action = true; + contract.mutates_object_data = true; + + assert_eq!( + validate_ops_diagnostics_contract(&contract).expect_err("object mutation should be rejected"), + ExtensionContractError::OpsDiagnosticsMutatesObjectData + ); + } } diff --git a/crates/targets/src/catalog/extension.rs b/crates/targets/src/catalog/extension.rs index ab1ffd61b..8c4f2e948 100644 --- a/crates/targets/src/catalog/extension.rs +++ b/crates/targets/src/catalog/extension.rs @@ -21,12 +21,83 @@ use crate::manifest::{ }; use rustfs_extension_schema::{ EXTENSION_SCHEMA_VERSION, ExtensionCapabilityRef, ExtensionKind, ExtensionRuntimeBoundary, ExtensionRuntimeContract, - ExtensionSchema, + ExtensionSchema, OPS_DIAGNOSTICS_CAPABILITY, OpsDiagnosticSurface, OpsDiagnosticsContract, S3_POST_AUTH_HOOK_CAPABILITY, + S3HookContract, S3HookPoint, }; +pub const OPS_DIAGNOSTICS_EXTENSION_API_VERSION: &str = "rustfs.ops-diagnostics.v1"; +pub const S3_HOOK_EXTENSION_API_VERSION: &str = "rustfs.s3-hook.v1"; pub const TARGET_AUDIT_CAPABILITY: &str = "target.audit.v1"; pub const TARGET_NOTIFY_CAPABILITY: &str = "target.notify.v1"; +pub fn builtin_extension_schemas() -> Vec { + let mut schemas = builtin_target_extension_schemas(); + schemas.push(builtin_s3_hook_extension_schema()); + schemas.push(builtin_ops_diagnostics_extension_schema()); + schemas +} + +pub fn builtin_s3_hook_extension_schema() -> ExtensionSchema { + ExtensionSchema { + schema_version: EXTENSION_SCHEMA_VERSION.to_string(), + extension_id: "builtin:s3-post-auth-hooks".to_string(), + display_name: "S3 Post-Auth Hook Registry".to_string(), + provider: "rustfs".to_string(), + version: "1.0.0".to_string(), + kind: ExtensionKind::S3Hook, + runtime: ExtensionRuntimeContract { + api_version: S3_HOOK_EXTENSION_API_VERSION.to_string(), + boundary: ExtensionRuntimeBoundary::Builtin, + }, + capabilities: vec![ExtensionCapabilityRef::new(S3_POST_AUTH_HOOK_CAPABILITY)], + disabled_by_default: false, + } +} + +pub fn builtin_s3_hook_contract() -> S3HookContract { + S3HookContract { + hook_points: vec![ + S3HookPoint::PostAuthGetObject, + S3HookPoint::PostAuthPutObject, + S3HookPoint::PostAuthDeleteObject, + S3HookPoint::PostAuthListObjects, + ], + mutates_object_data: false, + bypasses_iam: false, + } +} + +pub fn builtin_ops_diagnostics_extension_schema() -> ExtensionSchema { + ExtensionSchema { + schema_version: EXTENSION_SCHEMA_VERSION.to_string(), + extension_id: "builtin:ops-diagnostics".to_string(), + display_name: "Ops Diagnostics".to_string(), + provider: "rustfs".to_string(), + version: "1.0.0".to_string(), + kind: ExtensionKind::OpsDiagnostics, + runtime: ExtensionRuntimeContract { + api_version: OPS_DIAGNOSTICS_EXTENSION_API_VERSION.to_string(), + boundary: ExtensionRuntimeBoundary::Builtin, + }, + capabilities: vec![ExtensionCapabilityRef::new(OPS_DIAGNOSTICS_CAPABILITY)], + disabled_by_default: false, + } +} + +pub fn builtin_ops_diagnostics_contract() -> OpsDiagnosticsContract { + OpsDiagnosticsContract { + surfaces: vec![ + OpsDiagnosticSurface::Metrics, + OpsDiagnosticSurface::Trace, + OpsDiagnosticSurface::Profile, + OpsDiagnosticSurface::Health, + OpsDiagnosticSurface::Diagnostics, + ], + mutates_object_data: false, + requires_admin_action: true, + } +} + pub fn target_marketplace_extension_schema(manifest: &TargetPluginMarketplaceManifest) -> ExtensionSchema { let boundary = target_runtime_boundary(manifest.entrypoint_kind); @@ -88,7 +159,10 @@ fn target_domain_capability(domain: TargetDomain) -> ExtensionCapabilityRef { #[cfg(test)] mod tests { use super::{ - TARGET_AUDIT_CAPABILITY, TARGET_NOTIFY_CAPABILITY, builtin_target_extension_schemas, target_marketplace_extension_schema, + OPS_DIAGNOSTICS_EXTENSION_API_VERSION, S3_HOOK_EXTENSION_API_VERSION, TARGET_AUDIT_CAPABILITY, TARGET_NOTIFY_CAPABILITY, + builtin_extension_schemas, builtin_ops_diagnostics_contract, builtin_ops_diagnostics_extension_schema, + builtin_s3_hook_contract, builtin_s3_hook_extension_schema, builtin_target_extension_schemas, + target_marketplace_extension_schema, }; use crate::catalog::example_external_webhook_plugin; use crate::domain::TargetDomain; @@ -97,7 +171,9 @@ mod tests { builtin_target_marketplace_manifest, }; use rustfs_extension_schema::{ - EXTENSION_SCHEMA_VERSION, ExtensionKind, ExtensionRuntimeBoundary, validate_extension_schemas, + EXTENSION_SCHEMA_VERSION, ExtensionKind, ExtensionRuntimeBoundary, OPS_DIAGNOSTICS_CAPABILITY, OpsDiagnosticSurface, + S3_POST_AUTH_HOOK_CAPABILITY, S3HookPoint, validate_extension_schemas, validate_ops_diagnostics_contract, + validate_s3_hook_contract, }; #[test] @@ -176,4 +252,66 @@ mod tests { assert!(schemas.iter().all(|schema| !schema.disabled_by_default)); assert!(validate_extension_schemas(&schemas).is_ok()); } + + #[test] + fn builtin_extension_catalog_includes_hooks_and_ops_diagnostics() { + let schemas = builtin_extension_schemas(); + let mut extension_ids = schemas.iter().map(|schema| schema.extension_id.as_str()).collect::>(); + extension_ids.sort_unstable(); + + assert_eq!(schemas.len(), 11); + assert!(extension_ids.contains(&"builtin:s3-post-auth-hooks")); + assert!(extension_ids.contains(&"builtin:ops-diagnostics")); + assert!(validate_extension_schemas(&schemas).is_ok()); + } + + #[test] + fn builtin_s3_hook_schema_declares_post_auth_noop_contract() { + let schema = builtin_s3_hook_extension_schema(); + let contract = builtin_s3_hook_contract(); + + assert_eq!(schema.kind, ExtensionKind::S3Hook); + assert_eq!(schema.runtime.boundary, ExtensionRuntimeBoundary::Builtin); + assert_eq!(schema.runtime.api_version, S3_HOOK_EXTENSION_API_VERSION); + assert_eq!(schema.capabilities[0].as_str(), S3_POST_AUTH_HOOK_CAPABILITY); + assert!(!schema.disabled_by_default); + assert_eq!( + contract.hook_points, + vec![ + S3HookPoint::PostAuthGetObject, + S3HookPoint::PostAuthPutObject, + S3HookPoint::PostAuthDeleteObject, + S3HookPoint::PostAuthListObjects, + ] + ); + assert!(contract.hook_points.iter().all(|hook_point| hook_point.is_post_auth())); + assert!(!contract.mutates_object_data); + assert!(!contract.bypasses_iam); + assert!(validate_s3_hook_contract(&contract).is_ok()); + } + + #[test] + fn builtin_ops_diagnostics_schema_is_admin_read_only() { + let schema = builtin_ops_diagnostics_extension_schema(); + let contract = builtin_ops_diagnostics_contract(); + + assert_eq!(schema.kind, ExtensionKind::OpsDiagnostics); + assert_eq!(schema.runtime.boundary, ExtensionRuntimeBoundary::Builtin); + assert_eq!(schema.runtime.api_version, OPS_DIAGNOSTICS_EXTENSION_API_VERSION); + assert_eq!(schema.capabilities[0].as_str(), OPS_DIAGNOSTICS_CAPABILITY); + assert!(!schema.disabled_by_default); + assert_eq!( + contract.surfaces, + vec![ + OpsDiagnosticSurface::Metrics, + OpsDiagnosticSurface::Trace, + OpsDiagnosticSurface::Profile, + OpsDiagnosticSurface::Health, + OpsDiagnosticSurface::Diagnostics, + ] + ); + assert!(!contract.mutates_object_data); + assert!(contract.requires_admin_action); + assert!(validate_ops_diagnostics_contract(&contract).is_ok()); + } } diff --git a/crates/targets/src/lib.rs b/crates/targets/src/lib.rs index bb3ebbdf9..b9a646c77 100644 --- a/crates/targets/src/lib.rs +++ b/crates/targets/src/lib.rs @@ -28,8 +28,10 @@ pub mod sys; pub mod target; pub use catalog::extension::{ - TARGET_AUDIT_CAPABILITY, TARGET_NOTIFY_CAPABILITY, builtin_target_extension_schemas, target_marketplace_extension_schema, - target_runtime_boundary, + OPS_DIAGNOSTICS_EXTENSION_API_VERSION, S3_HOOK_EXTENSION_API_VERSION, TARGET_AUDIT_CAPABILITY, TARGET_NOTIFY_CAPABILITY, + builtin_extension_schemas, builtin_ops_diagnostics_contract, builtin_ops_diagnostics_extension_schema, + builtin_s3_hook_contract, builtin_s3_hook_extension_schema, builtin_target_extension_schemas, + target_marketplace_extension_schema, target_runtime_boundary, }; pub use check::{ check_amqp_broker_available, check_kafka_broker_available, check_mqtt_broker_available, check_mqtt_broker_available_with_tls, @@ -63,6 +65,11 @@ pub use runtime::{ RuntimeTargetHealthState, RuntimeTargetSnapshot, SharedTarget, TargetRuntimeManager, activate_targets_with_replay, adapter::{BuiltinPluginRuntimeAdapter, PluginRuntimeAdapter}, init_target_and_optionally_start_replay, + ops_diagnostics::{ + OpsDiagnosticsAccessDecision, OpsDiagnosticsReadRequest, OpsDiagnosticsRegistration, OpsDiagnosticsRegistry, + OpsDiagnosticsRegistryError, + }, + s3_hooks::{S3HookContext, S3HookDecision, S3HookRegistration, S3HookRegistry, S3HookRegistryError}, sidecar::{SidecarPluginRuntime, SidecarRuntimePolicy, SidecarRuntimeSafetyChecks}, sidecar_protocol::{SIDECAR_RUNTIME_PROTOCOL_VERSION, SidecarHandshake, SidecarPluginCapability}, start_replay_worker, diff --git a/crates/targets/src/runtime/mod.rs b/crates/targets/src/runtime/mod.rs index 6ac58fb1c..a8df698f6 100644 --- a/crates/targets/src/runtime/mod.rs +++ b/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; diff --git a/crates/targets/src/runtime/ops_diagnostics.rs b/crates/targets/src/runtime/ops_diagnostics.rs new file mode 100644 index 000000000..abd45f1fa --- /dev/null +++ b/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>, +} + +#[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 { + 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) + ); + } +} diff --git a/crates/targets/src/runtime/s3_hooks.rs b/crates/targets/src/runtime/s3_hooks.rs new file mode 100644 index 000000000..235cc0f59 --- /dev/null +++ b/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>, +} + +#[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 { + 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 { + 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) + ); + } +} diff --git a/rustfs/src/admin/handlers/extensions.rs b/rustfs/src/admin/handlers/extensions.rs index ba6e758c4..0cb932d57 100644 --- a/rustfs/src/admin/handlers/extensions.rs +++ b/rustfs/src/admin/handlers/extensions.rs @@ -28,9 +28,7 @@ use hyper::Method; use matchit::Params; use rustfs_extension_schema::{ExtensionKind, ExtensionSchema}; use rustfs_policy::policy::action::{Action, AdminAction}; -use rustfs_targets::{ - builtin_target_extension_schemas, catalog::example_external_webhook_plugin, target_marketplace_extension_schema, -}; +use rustfs_targets::{builtin_extension_schemas, catalog::example_external_webhook_plugin, target_marketplace_extension_schema}; use s3s::header::CONTENT_TYPE; use s3s::{Body, S3Request, S3Response, S3Result, s3_error}; use serde::Serialize; @@ -86,7 +84,7 @@ pub(crate) struct ExtensionInstancesResponse { } fn build_extension_catalog_response() -> ExtensionCatalogResponse { - let mut extensions = builtin_target_extension_schemas(); + let mut extensions = builtin_extension_schemas(); let example = example_external_webhook_plugin(); extensions.push(target_marketplace_extension_schema(&example.manifest)); extensions.sort_by(|a, b| a.extension_id.cmp(&b.extension_id)); @@ -242,7 +240,7 @@ mod tests { } #[test] - fn extension_catalog_exposes_valid_target_plugin_schemas() { + fn extension_catalog_exposes_valid_extension_schemas() { let response = build_extension_catalog_response(); let webhook = response .extensions @@ -265,6 +263,23 @@ mod tests { .iter() .any(|capability| capability.as_str() == "target.notify.v1") ); + + let s3_hooks = response + .extensions + .iter() + .find(|schema| schema.extension_id == "builtin:s3-post-auth-hooks") + .expect("builtin s3 hook extension should be present"); + assert_eq!(s3_hooks.kind, ExtensionKind::S3Hook); + assert_eq!(s3_hooks.runtime.boundary, ExtensionRuntimeBoundary::Builtin); + + let diagnostics = response + .extensions + .iter() + .find(|schema| schema.extension_id == "builtin:ops-diagnostics") + .expect("builtin ops diagnostics extension should be present"); + assert_eq!(diagnostics.kind, ExtensionKind::OpsDiagnostics); + assert_eq!(diagnostics.runtime.boundary, ExtensionRuntimeBoundary::Builtin); + assert!(validate_extension_schemas(&response.extensions).is_ok()); }