From fb0e2d6d8f3accf4c404b22f5acc53c5ceeb290e Mon Sep 17 00:00:00 2001 From: Zhengchao An Date: Fri, 3 Jul 2026 10:18:21 +0800 Subject: [PATCH] refactor(plugins): harden target-plugin system and admin plugin contract (#4217) --- .agents/skills/plugin-contract-guard/SKILL.md | 83 +++++ crates/audit/src/system.rs | 3 +- crates/extension-schema/src/lib.rs | 9 +- crates/notify/src/notifier.rs | 3 +- crates/notify/src/runtime_facade.rs | 3 +- crates/notify/src/runtime_view.rs | 3 +- crates/targets/src/catalog/builtin.rs | 9 +- crates/targets/src/catalog/extension.rs | 1 - crates/targets/src/catalog/mod.rs | 12 +- crates/targets/src/control_plane.rs | 194 ++++++++++-- crates/targets/src/lib.rs | 4 +- crates/targets/src/manifest.rs | 29 +- crates/targets/src/plugin.rs | 62 +++- crates/targets/src/runtime/adapter.rs | 15 +- crates/targets/src/runtime/mod.rs | 29 +- crates/targets/src/target/amqp.rs | 11 +- crates/targets/src/target/kafka.rs | 11 +- crates/targets/src/target/mod.rs | 8 +- crates/targets/src/target/mqtt.rs | 11 +- crates/targets/src/target/mysql.rs | 11 +- crates/targets/src/target/nats.rs | 11 +- crates/targets/src/target/postgres.rs | 11 +- crates/targets/src/target/pulsar.rs | 11 +- crates/targets/src/target/redis.rs | 11 +- crates/targets/src/target/webhook.rs | 11 +- rustfs/src/admin/handlers/extensions.rs | 60 ++-- rustfs/src/admin/handlers/plugins_catalog.rs | 52 ++- .../src/admin/handlers/plugins_instances.rs | 295 +++++++++++------- .../src/admin/handlers/target_descriptor.rs | 4 + rustfs/src/admin/plugin_contract.rs | 34 +- 30 files changed, 683 insertions(+), 328 deletions(-) create mode 100644 .agents/skills/plugin-contract-guard/SKILL.md diff --git a/.agents/skills/plugin-contract-guard/SKILL.md b/.agents/skills/plugin-contract-guard/SKILL.md new file mode 100644 index 000000000..6c061e555 --- /dev/null +++ b/.agents/skills/plugin-contract-guard/SKILL.md @@ -0,0 +1,83 @@ +--- +name: plugin-contract-guard +description: Invariants and change procedure for the target-plugin / extension system — plugin manifests, admin plugin/extension catalog and instance APIs, secret redaction, external-plugin install policy. Use when editing crates/targets (manifest, plugin, control_plane, catalog, runtime), crates/extension-schema, or rustfs/src/admin plugin_contract.rs / plugins_*.rs / extensions.rs / target_descriptor.rs. +--- + +# Plugin & Extension Contract Guard + +The "plugin system" spans four surfaces that must stay consistent: + +| Surface | Location | +|---|---| +| Manifests & registry | `crates/targets/src/{manifest,plugin}.rs` | +| Install/enable planning (control plane) | `crates/targets/src/control_plane.rs` | +| Extension schemas | `crates/extension-schema/src/lib.rs`, `crates/targets/src/catalog/extension.rs` | +| Admin API contract | `rustfs/src/admin/plugin_contract.rs`, `handlers/{plugins_catalog,plugins_instances,extensions,target_descriptor}.rs` | + +## Hard invariants (verify before merging) + +1. **Secrets have one source of truth.** Secret config keys are declared only + in the plugin manifest (`TargetPluginManifest.secret_fields`, + `crates/targets/src/manifest.rs`) and flow to admin via + `AdminTargetSpec.secret_fields`. Never add a hand-maintained per-service + secret table in a handler; if redaction misses a field, fix the manifest. + +2. **Redaction must round-trip.** Instance GET responses replace secret values + with `***redacted***` (`REDACTED_SECRET_VALUE` in `plugins_instances.rs`). + Instance PUT restores the stored secret when it receives that placeholder + back (`restore_redacted_secret_values`). Any new read or write path for + target config must keep both halves: redact on the way out, restore the + placeholder on the way in. The placeholder literal must never be persisted. + +3. **Fixtures never reach production responses.** + `example_external_webhook_plugin()` (`crates/targets/src/catalog/mod.rs`) + is a test/demo fixture for control-plane planning tests. Production + catalog/extension handlers must not include it; regression tests + (`plugin_catalog_never_exposes_example_or_external_fixtures`, + `extension_catalog_never_exposes_example_or_external_fixtures`) enforce it. + +4. **External plugin flow is planning-only and deny-by-default.** + `plan_external_target_plugin_action` returns decisions, it executes + nothing. `TargetPluginExternalFlowGate::default()` is fully closed and + `TargetPluginInstallPolicy::default().allowed_download_hosts` is empty — + keep it that way; tests opt in via explicit policies. Install validation + requires https, an allowlisted host, a full 64-hex-char sha256 digest, + signature and provenance URIs, and an artifact matching the host + `target_triple`. + +5. **Custom target types must not collide.** Unknown target types get an + interned unique `custom:` plugin id (`custom_plugin_id` in + `manifest.rs`). Custom plugins with secrets must register via + `TargetPluginDescriptor::with_manifest` and declare `secret_fields`; + `::new` derives a manifest with no secrets. + +## Changing the admin JSON contract + +- Shapes are locked twice in `plugin_contract.rs` tests: insta snapshots + (`rustfs/src/admin/snapshots/`) plus literal `json!` assertions. Update + both deliberately; a shape change is a console-facing API change. +- Field naming is `snake_case`, except discovery blocks + (`runtimeCapabilities`, `clusterSnapshot`, `extensionsCatalog`) which are + camelCase **by cross-endpoint convention** (same shape in `system.rs`, + `console.rs`, `pools.rs`). Do not "fix" that inconsistency locally. +- Contract types deliberately duplicate `rustfs_targets` types + (anti-corruption layer). Add a `From` impl; do not serialize internal + types directly. + +## Handler conventions + +- Every new admin plugin/extension route needs authorization at the top of + `call` and an `include_str!` guard test asserting it (repo-wide pattern — + see `plugin_instance_handlers_require_admin_authorization_contract`). +- Reads use `GetBucketTargetAction` (instances) or `ServerInfoAdminAction` + (catalogs); writes use `SetBucketTargetAction`. +- Refresh persisted module switches once per request + (`refresh_persisted_module_switches`), then evaluate the sync + `module_disabled_block_reason` per domain — do not re-read the store per + domain or per instance. + +## Generic bounds + +Event-payload generics use the `PluginEvent` blanket trait +(`crates/targets/src/plugin.rs`). Do not respell +`Send + Sync + 'static + Clone + Serialize + DeserializeOwned`. diff --git a/crates/audit/src/system.rs b/crates/audit/src/system.rs index 065bbc151..7ae914fdc 100644 --- a/crates/audit/src/system.rs +++ b/crates/audit/src/system.rs @@ -550,7 +550,6 @@ mod tests { use rustfs_targets::store::{Key, Store}; use rustfs_targets::target::{EntityTarget, QueuedPayload, QueuedPayloadMeta}; use rustfs_targets::{StoreError, Target, TargetError}; - use serde::{Serialize, de::DeserializeOwned}; use std::collections::HashMap; use std::sync::Arc; use std::sync::atomic::{AtomicUsize, Ordering}; @@ -574,7 +573,7 @@ mod tests { #[async_trait] impl Target for TestTarget where - E: Send + Sync + 'static + Clone + Serialize + DeserializeOwned, + E: rustfs_targets::PluginEvent, { fn id(&self) -> TargetID { self.id.clone() diff --git a/crates/extension-schema/src/lib.rs b/crates/extension-schema/src/lib.rs index 74ec6e0e9..c09e030fe 100644 --- a/crates/extension-schema/src/lib.rs +++ b/crates/extension-schema/src/lib.rs @@ -78,6 +78,8 @@ pub struct ExtensionSchema { pub disabled_by_default: bool, } +/// Every supported hook point runs strictly after authentication and +/// authorization; pre-auth hook points are intentionally not representable. #[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)] #[serde(rename_all = "snake_case")] pub enum S3HookPoint { @@ -87,12 +89,6 @@ pub enum S3HookPoint { 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 { @@ -693,7 +689,6 @@ mod tests { 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"); } diff --git a/crates/notify/src/notifier.rs b/crates/notify/src/notifier.rs index a7244dc73..ba71b9698 100644 --- a/crates/notify/src/notifier.rs +++ b/crates/notify/src/notifier.rs @@ -420,7 +420,6 @@ mod tests { store::{Key, Store}, target::{EntityTarget, QueuedPayload, QueuedPayloadMeta}, }; - use serde::{Serialize, de::DeserializeOwned}; use std::sync::{ Arc, atomic::{AtomicUsize, Ordering}, @@ -502,7 +501,7 @@ mod tests { #[async_trait] impl Target for TestTarget where - E: Send + Sync + 'static + Clone + Serialize + DeserializeOwned, + E: rustfs_targets::PluginEvent, { fn id(&self) -> TargetID { self.id.clone() diff --git a/crates/notify/src/runtime_facade.rs b/crates/notify/src/runtime_facade.rs index 6624b5257..f59f20972 100644 --- a/crates/notify/src/runtime_facade.rs +++ b/crates/notify/src/runtime_facade.rs @@ -177,7 +177,6 @@ mod tests { use rustfs_targets::store::{Key, Store}; use rustfs_targets::target::{EntityTarget, QueuedPayload, QueuedPayloadMeta}; use rustfs_targets::{ReplayWorkerManager, SharedTarget, StoreError, Target, TargetError}; - use serde::{Serialize, de::DeserializeOwned}; use std::sync::Arc; use std::sync::atomic::{AtomicUsize, Ordering}; use tokio::sync::{RwLock, Semaphore}; @@ -200,7 +199,7 @@ mod tests { #[async_trait] impl Target for TestTarget where - E: Send + Sync + 'static + Clone + Serialize + DeserializeOwned, + E: rustfs_targets::PluginEvent, { fn id(&self) -> TargetID { self.id.clone() diff --git a/crates/notify/src/runtime_view.rs b/crates/notify/src/runtime_view.rs index 00e3f3767..0ad80baa1 100644 --- a/crates/notify/src/runtime_view.rs +++ b/crates/notify/src/runtime_view.rs @@ -79,7 +79,6 @@ mod tests { use rustfs_targets::store::{Key, Store}; use rustfs_targets::target::{EntityTarget, QueuedPayload, QueuedPayloadMeta, TargetDeliverySnapshot}; use rustfs_targets::{ReplayWorkerManager, StoreError, Target, TargetError}; - use serde::{Serialize, de::DeserializeOwned}; use std::sync::{ Arc, atomic::{AtomicU64, Ordering}, @@ -128,7 +127,7 @@ mod tests { #[async_trait] impl Target for TestTarget where - E: Send + Sync + 'static + Clone + Serialize + DeserializeOwned, + E: rustfs_targets::PluginEvent, { fn id(&self) -> TargetID { self.id.clone() diff --git a/crates/targets/src/catalog/builtin.rs b/crates/targets/src/catalog/builtin.rs index 18738e01c..d88d6bb92 100644 --- a/crates/targets/src/catalog/builtin.rs +++ b/crates/targets/src/catalog/builtin.rs @@ -12,6 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. +use crate::plugin::PluginEvent; use crate::plugin::{ BuiltinTargetAdminDescriptor, BuiltinTargetDescriptor, TargetAdminMetadata, TargetPluginDescriptor, TargetRequestValidator, boxed_target, @@ -36,8 +37,6 @@ use rustfs_config::{ AUDIT_POSTGRES_SUB_SYS, AUDIT_PULSAR_SUB_SYS, AUDIT_REDIS_SUB_SYS, AUDIT_WEBHOOK_SUB_SYS, }, }; -use serde::Serialize; -use serde::de::DeserializeOwned; use crate::config::{ build_amqp_args, build_kafka_args, build_mqtt_args, build_mysql_args, build_nats_args, build_postgres_args, @@ -57,7 +56,7 @@ fn build_descriptor( create_target: Create, ) -> BuiltinTargetDescriptor where - E: Send + Sync + 'static + Clone + Serialize + DeserializeOwned, + E: PluginEvent, Create: Fn(String, &KVS) -> Result, TargetError> + Send + Sync + 'static, Validate: Fn(&KVS) -> Result<(), TargetError> + Send + Sync + 'static, { @@ -145,7 +144,7 @@ pub fn builtin_audit_target_admin_descriptors() -> Vec() -> Vec> where - E: Send + Sync + 'static + Clone + Serialize + DeserializeOwned, + E: PluginEvent, { vec![ build_descriptor( @@ -317,7 +316,7 @@ pub fn builtin_notify_target_admin_descriptors() -> Vec() -> Vec> where - E: Send + Sync + 'static + Clone + Serialize + DeserializeOwned, + E: PluginEvent, { vec![ build_descriptor( diff --git a/crates/targets/src/catalog/extension.rs b/crates/targets/src/catalog/extension.rs index 81f901c3c..cb595b4cf 100644 --- a/crates/targets/src/catalog/extension.rs +++ b/crates/targets/src/catalog/extension.rs @@ -341,7 +341,6 @@ mod tests { 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()); diff --git a/crates/targets/src/catalog/mod.rs b/crates/targets/src/catalog/mod.rs index f821d7a87..b235d148c 100644 --- a/crates/targets/src/catalog/mod.rs +++ b/crates/targets/src/catalog/mod.rs @@ -33,6 +33,14 @@ pub struct ExampleInstallableTargetPlugin { pub valid_fields: Vec, } +/// Test/demo fixture describing what an installable external plugin would +/// look like. It must never be exposed through production admin responses; +/// it exists so control-plane planning and contract tests can exercise the +/// external-plugin shape before a real marketplace source exists. +/// Test/demo fixture describing what an installable external plugin would +/// look like. It must never be exposed through production admin responses; +/// it exists so control-plane planning and contract tests can exercise the +/// external-plugin shape before a real marketplace source exists. pub fn example_external_webhook_plugin() -> ExampleInstallableTargetPlugin { let base = TargetPluginManifest { plugin_id: "external:webhook-sidecar", @@ -55,7 +63,7 @@ pub fn example_external_webhook_plugin() -> ExampleInstallableTargetPlugin { artifact_id: "sidecar-linux-amd64", target_triple: "x86_64-unknown-linux-gnu", download_uri: "https://plugins.example.test/webhook-sidecar.tar.zst", - digest_sha256: "0123456789abcdef0123456789abcdef", + digest_sha256: "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef", signature_uri: "https://plugins.example.test/webhook-sidecar.tar.zst.sig", provenance_uri: "https://plugins.example.test/webhook-sidecar.tar.zst.intoto.jsonl", size_bytes: 8192, @@ -88,7 +96,7 @@ pub fn example_external_webhook_plugin() -> ExampleInstallableTargetPlugin { manifest, installation: external_target_plugin_installation( base.version, - "0123456789abcdef0123456789abcdef", + "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef", "sidecar-linux-amd64", Some("2026-05-13T20:00:00Z".to_string()), ), diff --git a/crates/targets/src/control_plane.rs b/crates/targets/src/control_plane.rs index 226b4ab47..14b6d5ce2 100644 --- a/crates/targets/src/control_plane.rs +++ b/crates/targets/src/control_plane.rs @@ -22,6 +22,8 @@ use serde::{Deserialize, Serialize}; use thiserror::Error; use url::Url; +const SHA256_HEX_DIGEST_LEN: usize = 64; + #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] #[serde(rename_all = "snake_case")] pub enum TargetPluginInstallState { @@ -274,8 +276,8 @@ pub enum TargetPluginExternalActionError { #[error("external plugin circuit breaker is open")] CircuitBreakerOpen, - #[error("external plugin {plugin_id} has no installable artifact")] - MissingArtifact { plugin_id: String }, + #[error("external plugin {plugin_id} has no installable artifact for host target triple {target_triple}")] + MissingArtifactForHost { plugin_id: String, target_triple: String }, } pub fn plan_external_target_plugin_action( @@ -283,12 +285,13 @@ pub fn plan_external_target_plugin_action( action: TargetPluginExternalAction, installation: &TargetPluginInstallation, gate: &TargetPluginExternalFlowGate, + host_target_triple: &str, ) -> Result { validate_external_action_subject(manifest)?; validate_external_action_gate(gate)?; match action { - TargetPluginExternalAction::Install => plan_external_install(manifest, action, gate), + TargetPluginExternalAction::Install => plan_external_install(manifest, action, gate, host_target_triple), TargetPluginExternalAction::Enable => { require_installed(manifest.plugin_id, installation)?; Ok(TargetPluginExternalActionDecision { @@ -332,7 +335,9 @@ impl Default for TargetPluginInstallPolicy { fn default() -> Self { Self { allowed_providers: vec!["rustfs".to_string(), "rustfs-labs".to_string()], - allowed_download_hosts: vec!["plugins.example.test".to_string()], + // Deny-by-default: operators must explicitly allow the hosts + // artifacts may be downloaded from before any install can plan. + allowed_download_hosts: Vec::new(), require_https: true, require_signature: true, require_provenance: true, @@ -382,10 +387,12 @@ pub fn validate_external_plugin_installation( if artifact.size_bytes == 0 { return Err(format!("artifact {} must declare a non-zero size", artifact.artifact_id)); } - if artifact.digest_sha256.len() < 16 || !artifact.digest_sha256.chars().all(|ch| ch.is_ascii_hexdigit()) { + if artifact.digest_sha256.len() != SHA256_HEX_DIGEST_LEN + || !artifact.digest_sha256.chars().all(|ch| ch.is_ascii_hexdigit()) + { return Err(format!( - "artifact {} has invalid digest_sha256 {}", - artifact.artifact_id, artifact.digest_sha256 + "artifact {} has invalid digest_sha256 {} (expected {} hex characters)", + artifact.artifact_id, artifact.digest_sha256, SHA256_HEX_DIGEST_LEN )); } if policy.require_signature && artifact.signature_uri.is_empty() { @@ -429,6 +436,7 @@ fn plan_external_install( manifest: &TargetPluginMarketplaceManifest, action: TargetPluginExternalAction, gate: &TargetPluginExternalFlowGate, + host_target_triple: &str, ) -> Result { let install_manifest = TargetPluginManifest { plugin_id: manifest.plugin_id, @@ -449,9 +457,15 @@ fn plan_external_install( let artifact = manifest .distribution - .and_then(|distribution| distribution.artifacts.first()) - .ok_or_else(|| TargetPluginExternalActionError::MissingArtifact { + .and_then(|distribution| { + distribution + .artifacts + .iter() + .find(|artifact| artifact.target_triple == host_target_triple) + }) + .ok_or_else(|| TargetPluginExternalActionError::MissingArtifactForHost { plugin_id: manifest.plugin_id.to_string(), + target_triple: host_target_triple.to_string(), })?; Ok(TargetPluginExternalActionDecision { @@ -542,6 +556,24 @@ mod tests { use crate::{SidecarRuntimePolicy, SidecarRuntimeSafetyChecks}; use std::time::Duration; + const TEST_HOST_TRIPLE: &str = "x86_64-unknown-linux-gnu"; + + fn policy_allowing_example_host() -> TargetPluginInstallPolicy { + TargetPluginInstallPolicy { + allowed_download_hosts: vec!["plugins.example.test".to_string()], + ..TargetPluginInstallPolicy::default() + } + } + + fn verified_gate_with_example_host() -> TargetPluginExternalFlowGate { + let mut gate = TargetPluginExternalFlowGate::verified( + SidecarRuntimePolicy::verified_external(16, Duration::from_secs(5), 3), + SidecarRuntimeSafetyChecks::verified(0), + ); + gate.install_policy = policy_allowing_example_host(); + gate + } + #[test] fn builtin_installation_maps_to_virtual_installed_revision() { let installation = builtin_target_plugin_installation(&builtin_target_manifest("webhook")); @@ -656,6 +688,7 @@ mod tests { validation_error: None, }, &gate, + TEST_HOST_TRIPLE, ); assert_eq!(result, Err(TargetPluginExternalActionError::ExternalFlowDisabled)); @@ -681,6 +714,7 @@ mod tests { validation_error: None, }, &gate, + TEST_HOST_TRIPLE, ); assert_eq!( @@ -697,7 +731,7 @@ mod tests { artifact_id: "sidecar-linux-amd64", target_triple: "x86_64-unknown-linux-gnu", download_uri: "https://plugins.example.test/webhook-sidecar.tar.zst", - digest_sha256: "0123456789abcdef0123456789abcdef", + digest_sha256: "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef", signature_uri: "https://plugins.example.test/webhook-sidecar.tar.zst.sig", provenance_uri: "", size_bytes: 8192, @@ -707,10 +741,7 @@ mod tests { example.manifest.distribution = Some(TargetPluginDistributionManifest { artifacts: MISSING_PROVENANCE_ARTIFACTS, }); - let gate = TargetPluginExternalFlowGate::verified( - SidecarRuntimePolicy::verified_external(16, Duration::from_secs(5), 3), - SidecarRuntimeSafetyChecks::verified(0), - ); + let gate = verified_gate_with_example_host(); let result = plan_external_target_plugin_action( &example.manifest, @@ -722,6 +753,7 @@ mod tests { validation_error: None, }, &gate, + TEST_HOST_TRIPLE, ); assert_eq!( @@ -749,6 +781,7 @@ mod tests { TargetPluginExternalAction::Enable, &example.installation, &gate, + TEST_HOST_TRIPLE, ); assert_eq!( @@ -773,6 +806,7 @@ mod tests { TargetPluginExternalAction::Enable, &example.installation, &gate, + TEST_HOST_TRIPLE, ); assert_eq!(result, Err(TargetPluginExternalActionError::CircuitBreakerOpen)); @@ -781,10 +815,7 @@ mod tests { #[test] fn external_actions_plan_install_disable_and_rollback_without_execution() { let example = example_external_webhook_plugin(); - let gate = TargetPluginExternalFlowGate::verified( - SidecarRuntimePolicy::verified_external(16, Duration::from_secs(5), 3), - SidecarRuntimeSafetyChecks::verified(0), - ); + let gate = verified_gate_with_example_host(); let install = plan_external_target_plugin_action( &example.manifest, @@ -796,6 +827,7 @@ mod tests { validation_error: None, }, &gate, + TEST_HOST_TRIPLE, ) .expect("verified external install action should plan"); assert_eq!(install.installation.install_state, TargetPluginInstallState::Installed); @@ -807,6 +839,7 @@ mod tests { TargetPluginExternalAction::Disable, &example.installation, &gate, + TEST_HOST_TRIPLE, ) .expect("verified external disable action should plan"); assert_eq!(disable.operational_state.enable_state, TargetPluginEnableState::Disabled); @@ -836,6 +869,7 @@ mod tests { validation_error: None, }, &gate, + TEST_HOST_TRIPLE, ) .expect("verified external rollback action should plan"); @@ -860,13 +894,13 @@ mod tests { artifact_id: "sidecar-linux-amd64", target_triple: "x86_64-unknown-linux-gnu", download_uri: "https://plugins.example.test/webhook-sidecar.tar.zst", - digest_sha256: "0123456789abcdef0123456789abcdef", + digest_sha256: "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef", signature_uri: "https://plugins.example.test/webhook-sidecar.tar.zst.sig", provenance_uri: "https://plugins.example.test/webhook-sidecar.tar.zst.intoto.jsonl", size_bytes: 8192, }], }; - let policy = TargetPluginInstallPolicy::default(); + let policy = policy_allowing_example_host(); let result = validate_external_plugin_installation( &manifest, @@ -882,18 +916,56 @@ mod tests { } #[test] - fn validate_external_installation_rejects_disallowed_provider() { + fn default_install_policy_denies_all_download_hosts() { + let policy = TargetPluginInstallPolicy::default(); + assert!(policy.allowed_download_hosts.is_empty()); + let manifest = TargetPluginManifest { plugin_id: "external:webhook-sidecar", display_name: "Webhook Sidecar", - provider: "unknown-vendor", + provider: "rustfs-labs", version: "1.0.0", target_type: "webhook", supported_domains: &[], secret_fields: &[], }; - let policy = TargetPluginInstallPolicy::default(); + let result = validate_external_plugin_installation( + &manifest, + &TargetPluginExternalRuntimeContract { + protocol_version: crate::SIDECAR_RUNTIME_PROTOCOL_VERSION, + transport: TargetPluginRuntimeTransport::Grpc, + }, + Some(TargetPluginDistributionManifest { + artifacts: &[TargetPluginArtifactManifest { + artifact_id: "sidecar-linux-amd64", + target_triple: "x86_64-unknown-linux-gnu", + download_uri: "https://plugins.example.test/webhook-sidecar.tar.zst", + digest_sha256: "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef", + signature_uri: "https://plugins.example.test/webhook-sidecar.tar.zst.sig", + provenance_uri: "https://plugins.example.test/webhook-sidecar.tar.zst.intoto.jsonl", + size_bytes: 8192, + }], + }), + &policy, + ); + assert_eq!( + result.as_ref().map_err(String::as_str), + Err("artifact sidecar-linux-amd64 download host plugins.example.test is not allowed") + ); + } + + #[test] + fn validate_external_installation_rejects_truncated_digest() { + let manifest = TargetPluginManifest { + plugin_id: "external:webhook-sidecar", + display_name: "Webhook Sidecar", + provider: "rustfs-labs", + version: "1.0.0", + target_type: "webhook", + supported_domains: &[], + secret_fields: &[], + }; let result = validate_external_plugin_installation( &manifest, &TargetPluginExternalRuntimeContract { @@ -911,6 +983,74 @@ mod tests { size_bytes: 8192, }], }), + &policy_allowing_example_host(), + ); + + assert_eq!( + result.as_ref().map_err(String::as_str), + Err( + "artifact sidecar-linux-amd64 has invalid digest_sha256 0123456789abcdef0123456789abcdef (expected 64 hex characters)" + ) + ); + } + + #[test] + fn external_install_requires_artifact_for_host_target_triple() { + let example = example_external_webhook_plugin(); + let gate = verified_gate_with_example_host(); + + let result = plan_external_target_plugin_action( + &example.manifest, + TargetPluginExternalAction::Install, + &TargetPluginInstallation { + install_state: TargetPluginInstallState::NotInstalled, + current_revision: None, + previous_revision: None, + validation_error: None, + }, + &gate, + "aarch64-apple-darwin", + ); + + assert_eq!( + result, + Err(TargetPluginExternalActionError::MissingArtifactForHost { + plugin_id: "external:webhook-sidecar".to_string(), + target_triple: "aarch64-apple-darwin".to_string(), + }) + ); + } + + #[test] + fn validate_external_installation_rejects_disallowed_provider() { + let manifest = TargetPluginManifest { + plugin_id: "external:webhook-sidecar", + display_name: "Webhook Sidecar", + provider: "unknown-vendor", + version: "1.0.0", + target_type: "webhook", + supported_domains: &[], + secret_fields: &[], + }; + let policy = policy_allowing_example_host(); + + let result = validate_external_plugin_installation( + &manifest, + &TargetPluginExternalRuntimeContract { + protocol_version: crate::SIDECAR_RUNTIME_PROTOCOL_VERSION, + transport: TargetPluginRuntimeTransport::Grpc, + }, + Some(TargetPluginDistributionManifest { + artifacts: &[TargetPluginArtifactManifest { + artifact_id: "sidecar-linux-amd64", + target_triple: "x86_64-unknown-linux-gnu", + download_uri: "https://plugins.example.test/webhook-sidecar.tar.zst", + digest_sha256: "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef", + signature_uri: "https://plugins.example.test/webhook-sidecar.tar.zst.sig", + provenance_uri: "https://plugins.example.test/webhook-sidecar.tar.zst.intoto.jsonl", + size_bytes: 8192, + }], + }), &policy, ); @@ -928,7 +1068,7 @@ mod tests { supported_domains: &[], secret_fields: &[], }; - let policy = TargetPluginInstallPolicy::default(); + let policy = policy_allowing_example_host(); let result = validate_external_plugin_installation( &manifest, @@ -941,7 +1081,7 @@ mod tests { artifact_id: "sidecar-linux-amd64", target_triple: "x86_64-unknown-linux-gnu", download_uri: "https://plugins.example.test/webhook-sidecar.tar.zst", - digest_sha256: "0123456789abcdef0123456789abcdef", + digest_sha256: "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef", signature_uri: "", provenance_uri: "https://plugins.example.test/webhook-sidecar.intoto.jsonl", size_bytes: 8192, @@ -967,7 +1107,7 @@ mod tests { supported_domains: &[], secret_fields: &[], }; - let policy = TargetPluginInstallPolicy::default(); + let policy = policy_allowing_example_host(); let result = validate_external_plugin_installation( &manifest, @@ -980,7 +1120,7 @@ mod tests { artifact_id: "sidecar-linux-amd64", target_triple: "x86_64-unknown-linux-gnu", download_uri: "https://plugins.example.test/webhook-sidecar.tar.zst", - digest_sha256: "0123456789abcdef0123456789abcdef", + digest_sha256: "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef", signature_uri: "https://plugins.example.test/webhook-sidecar.sig", provenance_uri: "", size_bytes: 8192, diff --git a/crates/targets/src/lib.rs b/crates/targets/src/lib.rs index 04f000c58..e9af8e94f 100644 --- a/crates/targets/src/lib.rs +++ b/crates/targets/src/lib.rs @@ -60,8 +60,8 @@ pub use manifest::{ }; pub use net::*; pub use plugin::{ - BuiltinTargetAdminDescriptor, BuiltinTargetDescriptor, TargetAdminMetadata, TargetPluginDescriptor, TargetPluginRegistry, - TargetRequestValidator, boxed_target, + BuiltinTargetAdminDescriptor, BuiltinTargetDescriptor, PluginEvent, TargetAdminMetadata, TargetPluginDescriptor, + TargetPluginRegistry, TargetRequestValidator, boxed_target, }; pub use runtime::{ ReplayEvent, ReplayWorkerManager, RuntimeActivation, RuntimeStatusSnapshot, RuntimeTargetHealthSnapshot, diff --git a/crates/targets/src/manifest.rs b/crates/targets/src/manifest.rs index 0d7ef3b17..1dcd3f7f9 100644 --- a/crates/targets/src/manifest.rs +++ b/crates/targets/src/manifest.rs @@ -13,6 +13,9 @@ // limitations under the License. use crate::domain::TargetDomain; +use std::collections::HashMap; +use std::sync::{Mutex, OnceLock}; + use rustfs_config::{ AMQP_PASSWORD, AMQP_TLS_CLIENT_CERT, AMQP_TLS_CLIENT_KEY, KAFKA_SASL_PASSWORD, KAFKA_TLS_CLIENT_CERT, KAFKA_TLS_CLIENT_KEY, MQTT_PASSWORD, MQTT_TLS_CLIENT_CERT, MQTT_TLS_CLIENT_KEY, MYSQL_DSN_STRING, MYSQL_TLS_CLIENT_CERT, MYSQL_TLS_CLIENT_KEY, @@ -210,10 +213,22 @@ fn builtin_plugin_id(target_type: &'static str) -> &'static str { "mysql" => "builtin:mysql", "redis" => "builtin:redis", "postgres" => "builtin:postgres", - _ => "custom:target", + _ => custom_plugin_id(target_type), } } +/// Interns a unique `custom:` plugin id per non-builtin target type. +/// A shared fallback id would make distinct custom plugins collide in every +/// identity-keyed surface (registry keys, canonical instance ids, admin catalog). +/// The leak is bounded by the number of distinct registered target types. +fn custom_plugin_id(target_type: &'static str) -> &'static str { + static CUSTOM_PLUGIN_IDS: OnceLock>> = OnceLock::new(); + let ids = CUSTOM_PLUGIN_IDS.get_or_init(|| Mutex::new(HashMap::new())); + let mut ids = ids.lock().unwrap_or_else(std::sync::PoisonError::into_inner); + ids.entry(target_type) + .or_insert_with(|| Box::leak(format!("custom:{target_type}").into_boxed_str())) +} + #[cfg(test)] mod tests { use super::{ @@ -263,6 +278,18 @@ mod tests { assert_eq!(manifest.supported_domains, &[TargetDomain::Audit, TargetDomain::Notify]); } + #[test] + fn custom_target_types_get_unique_stable_plugin_ids() { + let first = builtin_target_manifest("custom-a"); + let second = builtin_target_manifest("custom-b"); + + assert_eq!(first.plugin_id, "custom:custom-a"); + assert_eq!(second.plugin_id, "custom:custom-b"); + assert_ne!(first.plugin_id, second.plugin_id); + // Interned: repeated lookups return the same &'static str. + assert!(std::ptr::eq(first.plugin_id, builtin_target_manifest("custom-a").plugin_id)); + } + #[test] fn builtin_kafka_manifest_marks_sasl_password_secret() { let manifest = builtin_target_manifest("kafka"); diff --git a/crates/targets/src/plugin.rs b/crates/targets/src/plugin.rs index 3b8f5dce3..ca7e94108 100644 --- a/crates/targets/src/plugin.rs +++ b/crates/targets/src/plugin.rs @@ -23,12 +23,20 @@ use serde::Serialize; use serde::de::DeserializeOwned; use std::collections::HashSet; use std::sync::Arc; -use tracing::{error, info}; +use tracing::{error, info, warn}; type BoxedTarget = Box + Send + Sync>; type TargetCreateFn = Arc Result, TargetError> + Send + Sync>; type TargetValidateFn = Arc Result<(), TargetError> + Send + Sync>; +/// Event payload contract shared by all target plugin machinery. +/// +/// Blanket-implemented for every type meeting the bounds; it exists solely to +/// keep this composite bound spelled in one place instead of on every generic. +pub trait PluginEvent: Send + Sync + Clone + Serialize + DeserializeOwned + 'static {} + +impl PluginEvent for T where T: Send + Sync + Clone + Serialize + DeserializeOwned + 'static {} + #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum TargetRequestValidator { Webhook, @@ -105,7 +113,7 @@ impl BuiltinTargetAdminDescriptor { #[derive(Clone)] pub struct TargetPluginDescriptor where - E: Send + Sync + 'static + Clone + Serialize + DeserializeOwned, + E: PluginEvent, { create_target: TargetCreateFn, manifest: TargetPluginManifest, @@ -117,7 +125,7 @@ where impl TargetPluginDescriptor where - E: Send + Sync + 'static + Clone + Serialize + DeserializeOwned, + E: PluginEvent, { pub fn new( target_type: &'static str, @@ -186,7 +194,7 @@ where #[derive(Clone)] pub struct BuiltinTargetDescriptor where - E: Send + Sync + 'static + Clone + Serialize + DeserializeOwned, + E: PluginEvent, { plugin: TargetPluginDescriptor, admin: TargetAdminMetadata, @@ -194,7 +202,7 @@ where impl BuiltinTargetDescriptor where - E: Send + Sync + 'static + Clone + Serialize + DeserializeOwned, + E: PluginEvent, { pub fn new(subsystem: &'static str, request_validator: TargetRequestValidator, plugin: TargetPluginDescriptor) -> Self { Self { @@ -226,7 +234,7 @@ where impl From> for BuiltinTargetAdminDescriptor where - E: Send + Sync + 'static + Clone + Serialize + DeserializeOwned, + E: PluginEvent, { fn from(descriptor: BuiltinTargetDescriptor) -> Self { Self::new( @@ -239,14 +247,14 @@ where pub struct TargetPluginRegistry where - E: Send + Sync + 'static + Clone + Serialize + DeserializeOwned, + E: PluginEvent, { plugins: HashMap>, } impl Default for TargetPluginRegistry where - E: Send + Sync + 'static + Clone + Serialize + DeserializeOwned, + E: PluginEvent, { fn default() -> Self { Self::new() @@ -255,14 +263,22 @@ where impl TargetPluginRegistry where - E: Send + Sync + 'static + Clone + Serialize + DeserializeOwned, + E: PluginEvent, { pub fn new() -> Self { Self { plugins: HashMap::new() } } pub fn register(&mut self, plugin: TargetPluginDescriptor) -> Option> { - self.plugins.insert(plugin.target_type().to_string(), plugin) + let replaced = self.plugins.insert(plugin.target_type().to_string(), plugin); + if let Some(previous) = &replaced { + warn!( + target_type = %previous.target_type(), + plugin_id = %previous.manifest().plugin_id, + "replacing previously registered target plugin descriptor" + ); + } + replaced } pub fn register_all(&mut self, plugins: I) @@ -291,12 +307,18 @@ where plugin.create_target(id, config) } + /// Creates every enabled target instance found in `config`. + /// + /// Creation is fault-isolated per instance: one broken target must not + /// prevent the remaining targets from activating, so failures are logged + /// and summarized instead of aborting the whole activation. pub async fn create_targets_from_config( &self, config: &Config, route_prefix: &str, ) -> Result>, TargetError> { let mut successful_targets = Vec::new(); + let mut failed_targets = 0usize; for (target_type, plugin) in &self.plugins { info!(target_type = %target_type, "Start working on target type"); @@ -308,13 +330,25 @@ where successful_targets.push(target); } Err(err) => { + failed_targets += 1; error!(target_type = %target_type, instance_id = %id, error = %err, "Failed to create target"); } } } } - info!(count = successful_targets.len(), "All target processing completed"); + if failed_targets > 0 { + warn!( + created = successful_targets.len(), + failed = failed_targets, + "Some configured targets failed to create and were skipped" + ); + } + info!( + count = successful_targets.len(), + failed = failed_targets, + "All target processing completed" + ); Ok(successful_targets) } @@ -334,7 +368,7 @@ where pub fn boxed_target(target: T) -> BoxedTarget where - E: Send + Sync + 'static + Clone + Serialize + DeserializeOwned, + E: PluginEvent, T: Target + Send + Sync + 'static, { Box::new(target) @@ -343,6 +377,7 @@ where #[cfg(test)] mod tests { use super::{TargetPluginDescriptor, TargetPluginRegistry}; + use crate::PluginEvent; use crate::runtime::adapter::BuiltinPluginRuntimeAdapter; use crate::store::{Key, Store}; use crate::target::{EntityTarget, QueuedPayload, QueuedPayloadMeta}; @@ -350,7 +385,6 @@ mod tests { use async_trait::async_trait; use rustfs_config::ENABLE_KEY; use rustfs_config::server_config::{Config, KVS}; - use serde::{Serialize, de::DeserializeOwned}; use std::collections::HashMap; use std::sync::Arc; use std::time::Duration; @@ -363,7 +397,7 @@ mod tests { #[async_trait] impl Target for TestTarget where - E: Send + Sync + 'static + Clone + Serialize + DeserializeOwned, + E: PluginEvent, { fn id(&self) -> crate::arn::TargetID { self.id.clone() diff --git a/crates/targets/src/runtime/adapter.rs b/crates/targets/src/runtime/adapter.rs index 084efd0e5..5b3649b8c 100644 --- a/crates/targets/src/runtime/adapter.rs +++ b/crates/targets/src/runtime/adapter.rs @@ -16,10 +16,9 @@ use super::{ ReplayEvent, ReplayWorkerManager, RuntimeActivation, RuntimeStatusSnapshot, RuntimeTargetHealthSnapshot, TargetRuntimeManager, activate_targets_with_replay, init_target_and_optionally_start_replay, start_replay_worker, }; +use crate::plugin::PluginEvent; use crate::{Target, TargetError}; use async_trait::async_trait; -use serde::Serialize; -use serde::de::DeserializeOwned; use std::future::Future; use std::pin::Pin; use std::sync::Arc; @@ -33,7 +32,7 @@ type ReplayStartObserver = Arc; #[async_trait] pub trait PluginRuntimeAdapter: Send + Sync where - E: Send + Sync + 'static + Clone + Serialize + DeserializeOwned, + E: PluginEvent, { async fn activate_with_replay(&self, targets: Vec + Send + Sync>>) -> RuntimeActivation; @@ -66,7 +65,7 @@ where #[derive(Clone)] pub struct BuiltinPluginRuntimeAdapter where - E: Send + Sync + 'static + Clone + Serialize + DeserializeOwned, + E: PluginEvent, { replay_hook: ReplayHook, replay_start_observer: ReplayStartObserver, @@ -78,7 +77,7 @@ where impl BuiltinPluginRuntimeAdapter where - E: Send + Sync + 'static + Clone + Serialize + DeserializeOwned, + E: PluginEvent, { pub fn new( replay_hook: ReplayHook, @@ -102,7 +101,7 @@ where #[async_trait] impl PluginRuntimeAdapter for BuiltinPluginRuntimeAdapter where - E: Send + Sync + 'static + Clone + Serialize + DeserializeOwned, + E: PluginEvent, { async fn activate_with_replay(&self, targets: Vec + Send + Sync>>) -> RuntimeActivation { let replay_hook = Arc::clone(&self.replay_hook); @@ -184,12 +183,12 @@ where #[cfg(test)] mod tests { use super::{BuiltinPluginRuntimeAdapter, PluginRuntimeAdapter}; + use crate::PluginEvent; use crate::arn::TargetID; use crate::store::{Key, QueueStore, Store}; use crate::target::{EntityTarget, QueuedPayload, QueuedPayloadMeta}; use crate::{StoreError, Target, TargetError}; use async_trait::async_trait; - use serde::{Serialize, de::DeserializeOwned}; use std::sync::Arc; use std::sync::atomic::{AtomicUsize, Ordering}; use std::time::Duration; @@ -230,7 +229,7 @@ mod tests { #[async_trait] impl Target for TestTarget where - E: Send + Sync + 'static + Clone + Serialize + DeserializeOwned, + E: PluginEvent, { fn id(&self) -> TargetID { self.id.clone() diff --git a/crates/targets/src/runtime/mod.rs b/crates/targets/src/runtime/mod.rs index 4dce1f831..24d1e9061 100644 --- a/crates/targets/src/runtime/mod.rs +++ b/crates/targets/src/runtime/mod.rs @@ -22,12 +22,11 @@ pub mod tls; use crate::Target; use crate::arn::TargetID; +use crate::plugin::PluginEvent; use crate::store::{Key, Store, ensure_store_entry_raw_readable}; use crate::target::QueuedPayload; use crate::target::TargetDeliverySnapshot; use crate::{StoreError, TargetError}; -use serde::Serialize; -use serde::de::DeserializeOwned; use std::sync::Arc; use std::{collections::HashMap, fmt::Debug}; use std::{future::Future, pin::Pin, time::Duration}; @@ -78,7 +77,7 @@ impl ReplayWorkerManager { pub struct RuntimeActivation where - E: Send + Sync + 'static + Clone + Serialize + DeserializeOwned, + E: PluginEvent, { pub replay_workers: ReplayWorkerManager, pub targets: Vec>, @@ -119,7 +118,7 @@ pub struct RuntimeTargetHealthSnapshot { pub enum ReplayEvent where - E: Send + Sync + 'static + Clone + Serialize + DeserializeOwned, + E: PluginEvent, { Delivered { key: Key, @@ -159,14 +158,14 @@ where /// be layered on top in later phases. pub struct TargetRuntimeManager where - E: Send + Sync + 'static + Clone + Serialize + DeserializeOwned, + E: PluginEvent, { targets: HashMap>, } impl Default for TargetRuntimeManager where - E: Send + Sync + 'static + Clone + Serialize + DeserializeOwned, + E: PluginEvent, { fn default() -> Self { Self::new() @@ -175,7 +174,7 @@ where impl Debug for TargetRuntimeManager where - E: Send + Sync + 'static + Clone + Serialize + DeserializeOwned, + E: PluginEvent, { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { f.debug_struct("TargetRuntimeManager") @@ -186,7 +185,7 @@ where impl TargetRuntimeManager where - E: Send + Sync + 'static + Clone + Serialize + DeserializeOwned, + E: PluginEvent, { pub fn new() -> Self { Self { targets: HashMap::new() } @@ -320,7 +319,7 @@ pub async fn init_target_and_optionally_start_replay( start_replay: G, ) -> Option<(SharedTarget, Option>)> where - E: Send + Sync + 'static + Clone + Serialize + DeserializeOwned, + E: PluginEvent, F: FnOnce(&str, bool), G: FnOnce(Box + Send>, SharedTarget) -> mpsc::Sender<()>, { @@ -356,7 +355,7 @@ pub async fn activate_targets_with_replay( mut activate_one: F, ) -> RuntimeActivation where - E: Send + Sync + 'static + Clone + Serialize + DeserializeOwned, + E: PluginEvent, F: FnMut(Box + Send + Sync>) -> Fut, Fut: Future, Option>)>>, { @@ -388,7 +387,7 @@ pub fn start_replay_worker( idle_sleep: Duration, ) -> mpsc::Sender<()> where - E: Send + Sync + 'static + Clone + Serialize + DeserializeOwned, + E: PluginEvent, { let (cancel_tx, cancel_rx) = mpsc::channel(1); @@ -408,7 +407,7 @@ async fn stream_replay_worker( batch_timeout: Duration, idle_sleep: Duration, ) where - E: Send + Sync + 'static + Clone + Serialize + DeserializeOwned, + E: PluginEvent, { const MAX_RETRIES: usize = 5; const BASE_RETRY_DELAY: Duration = Duration::from_secs(2); @@ -469,7 +468,7 @@ async fn stream_replay_worker( hook: &ReplayHook, semaphore: Option>, ) where - E: Send + Sync + 'static + Clone + Serialize + DeserializeOwned, + E: PluginEvent, { if batch_keys.is_empty() { return; @@ -553,13 +552,13 @@ async fn stream_replay_worker( #[cfg(test)] mod tests { use super::TargetRuntimeManager; + use crate::PluginEvent; use crate::StoreError; use crate::arn::TargetID; use crate::store::{Key, Store}; use crate::target::{EntityTarget, QueuedPayload, QueuedPayloadMeta}; use crate::{Target, TargetError}; use async_trait::async_trait; - use serde::{Serialize, de::DeserializeOwned}; use std::sync::Arc; use std::sync::atomic::{AtomicUsize, Ordering}; @@ -581,7 +580,7 @@ mod tests { #[async_trait] impl Target for TestTarget where - E: Send + Sync + 'static + Clone + Serialize + DeserializeOwned, + E: PluginEvent, { fn id(&self) -> TargetID { self.id.clone() diff --git a/crates/targets/src/target/amqp.rs b/crates/targets/src/target/amqp.rs index 0fd56fa9a..faefa3fde 100644 --- a/crates/targets/src/target/amqp.rs +++ b/crates/targets/src/target/amqp.rs @@ -18,6 +18,7 @@ //! Queue-store mode uses the shared target store and replays the same raw JSON //! body through `send_raw_from_store`. +use crate::plugin::PluginEvent; use crate::{ StoreError, Target, arn::TargetID, @@ -42,8 +43,6 @@ use lapin::{ use parking_lot::Mutex; use rustfs_config::{AMQP_TLS_CA, AMQP_TLS_CLIENT_CERT, AMQP_TLS_CLIENT_KEY}; use rustfs_tls_runtime::load_cert_bundle_der_bytes; -use serde::Serialize; -use serde::de::DeserializeOwned; use std::fmt; use std::path::Path; use std::sync::Arc; @@ -298,7 +297,7 @@ pub struct AMQPConnection { pub struct AMQPTarget where - E: Send + Sync + 'static + Clone + Serialize + DeserializeOwned, + E: PluginEvent, { id: TargetID, args: AMQPArgs, @@ -314,7 +313,7 @@ where impl AMQPTarget where - E: Send + Sync + 'static + Clone + Serialize + DeserializeOwned, + E: PluginEvent, { pub fn clone_box(&self) -> Box + Send + Sync> { Box::new(AMQPTarget:: { @@ -464,7 +463,7 @@ where #[async_trait] impl ReloadableTargetTls for AMQPTarget where - E: Send + Sync + 'static + Clone + Serialize + DeserializeOwned, + E: PluginEvent, { type Material = AMQPConnection; @@ -500,7 +499,7 @@ where #[async_trait] impl Target for AMQPTarget where - E: Send + Sync + 'static + Clone + Serialize + DeserializeOwned, + E: PluginEvent, { fn id(&self) -> TargetID { self.id.clone() diff --git a/crates/targets/src/target/kafka.rs b/crates/targets/src/target/kafka.rs index 8baff55a2..da942e24d 100644 --- a/crates/targets/src/target/kafka.rs +++ b/crates/targets/src/target/kafka.rs @@ -12,6 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. +use crate::plugin::PluginEvent; use crate::{ StoreError, Target, arn::TargetID, @@ -31,8 +32,6 @@ use async_trait::async_trait; use rustfs_kafka_async::error::{ConnectionError, Error as KafkaError}; use rustfs_kafka_async::{AsyncProducer, AsyncProducerConfig, Record, RequiredAcks, SaslConfig, SecurityConfig}; use rustfs_tls_runtime::{load_cert_bundle_der_bytes, load_private_key}; -use serde::Serialize; -use serde::de::DeserializeOwned; use std::{fmt, marker::PhantomData, sync::Arc, time::Duration}; use tokio::sync::Mutex; use tracing::{debug, error, info, instrument, warn}; @@ -228,7 +227,7 @@ impl KafkaArgs { /// A target that sends events to an Apache Kafka topic pub struct KafkaTarget where - E: Send + Sync + 'static + Clone + Serialize + DeserializeOwned, + E: PluginEvent, { id: TargetID, args: KafkaArgs, @@ -245,7 +244,7 @@ where impl KafkaTarget where - E: Send + Sync + 'static + Clone + Serialize + DeserializeOwned, + E: PluginEvent, { fn map_kafka_error(err: KafkaError, context: &str) -> TargetError { match err { @@ -397,7 +396,7 @@ where #[async_trait] impl Target for KafkaTarget where - E: Send + Sync + 'static + Clone + Serialize + DeserializeOwned, + E: PluginEvent, { fn id(&self) -> TargetID { self.id.clone() @@ -490,7 +489,7 @@ where #[async_trait] impl ReloadableTargetTls for KafkaTarget where - E: Send + Sync + 'static + Clone + Serialize + DeserializeOwned, + E: PluginEvent, { type Material = Arc; diff --git a/crates/targets/src/target/mod.rs b/crates/targets/src/target/mod.rs index 00d0edaa9..60975b877 100644 --- a/crates/targets/src/target/mod.rs +++ b/crates/targets/src/target/mod.rs @@ -13,11 +13,11 @@ // limitations under the License. use crate::arn::TargetID; +use crate::plugin::PluginEvent; use crate::store::{Key, QueueStore, Store}; use crate::{StoreError, TargetError, TargetLog}; use async_trait::async_trait; use rustfs_s3_types::EventName; -use serde::de::DeserializeOwned; use serde::{Deserialize, Serialize}; use std::fmt::Formatter; use std::future::Future; @@ -96,7 +96,7 @@ impl TargetDeliveryCounters { #[async_trait] pub trait Target: Send + Sync + 'static where - E: Send + Sync + 'static + Clone + Serialize + DeserializeOwned, + E: PluginEvent, { /// Returns the ID of the target fn id(&self) -> TargetID; @@ -428,7 +428,7 @@ pub fn decode_object_name(encoded: &str) -> Result { pub(crate) fn build_queued_payload(event: &EntityTarget) -> Result where - E: Send + Sync + 'static + Clone + Serialize + DeserializeOwned, + E: PluginEvent, { build_queued_payload_with_records(event, vec![event.data.clone()]) } @@ -438,7 +438,7 @@ pub(crate) fn build_queued_payload_with_records( records: Vec, ) -> Result where - E: Send + Sync + 'static + Clone + Serialize + DeserializeOwned, + E: PluginEvent, R: Serialize, { let object_name = decode_object_name(&event.object_name)?; diff --git a/crates/targets/src/target/mqtt.rs b/crates/targets/src/target/mqtt.rs index 0113882a6..2b25f6a63 100644 --- a/crates/targets/src/target/mqtt.rs +++ b/crates/targets/src/target/mqtt.rs @@ -12,6 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. +use crate::plugin::PluginEvent; use crate::{ StoreError, Target, arn::TargetID, @@ -39,8 +40,6 @@ use rustfs_config::{ }; use rustfs_tls_runtime::{load_certs, load_private_key}; use rustls::ClientConfig; -use serde::Serialize; -use serde::de::DeserializeOwned; use std::fmt; use std::sync::Arc; use std::{ @@ -524,7 +523,7 @@ struct BgTaskManager { /// A target that sends events to an MQTT broker pub struct MQTTTarget where - E: Send + Sync + 'static + Clone + Serialize + DeserializeOwned, + E: PluginEvent, { id: TargetID, args: MQTTArgs, @@ -544,7 +543,7 @@ where impl MQTTTarget where - E: Send + Sync + 'static + Clone + Serialize + DeserializeOwned, + E: PluginEvent, { /// Creates a new MQTTTarget #[instrument(skip(args), fields(target_id_as_string = %id))] @@ -844,7 +843,7 @@ where #[async_trait] impl ReloadableTargetTls for MQTTTarget where - E: Send + Sync + 'static + Clone + Serialize + DeserializeOwned, + E: PluginEvent, { type Material = MqttOptions; @@ -1130,7 +1129,7 @@ fn is_fatal_mqtt_error(err: &ConnectionError) -> bool { #[async_trait] impl Target for MQTTTarget where - E: Send + Sync + 'static + Clone + Serialize + DeserializeOwned, + E: PluginEvent, { fn id(&self) -> TargetID { self.id.clone() diff --git a/crates/targets/src/target/mysql.rs b/crates/targets/src/target/mysql.rs index 8607b528e..efa984611 100644 --- a/crates/targets/src/target/mysql.rs +++ b/crates/targets/src/target/mysql.rs @@ -12,6 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. +use crate::plugin::PluginEvent; use crate::{ StoreError, Target, arn::TargetID, @@ -31,8 +32,6 @@ use async_trait::async_trait; use mysql_async::{Conn, Opts, OptsBuilder, Pool, PoolConstraints, PoolOpts, SslOpts, prelude::Queryable}; use rustfs_config::{MYSQL_TLS_CA, MYSQL_TLS_CLIENT_CERT, MYSQL_TLS_CLIENT_KEY}; use rustfs_tls_runtime::{load_certs, load_private_key}; -use serde::Serialize; -use serde::de::DeserializeOwned; use std::fmt; use std::marker::PhantomData; use std::path::{Path, PathBuf}; @@ -505,7 +504,7 @@ async fn validate_existing_schema(conn: &mut Conn, table: &str) -> Result<(), Ta /// ``` pub struct MySqlTarget where - E: Send + Sync + 'static + Clone + Serialize + DeserializeOwned, + E: PluginEvent, { /// Unique target identifier (name + type) id: TargetID, @@ -528,7 +527,7 @@ where impl MySqlTarget where - E: Send + Sync + 'static + Clone + Serialize + DeserializeOwned, + E: PluginEvent, { /// Creates a new MySqlTarget. /// @@ -770,7 +769,7 @@ pub(crate) fn map_mysql_error(err: mysql_async::Error, operation: &str) -> Targe #[async_trait] impl Target for MySqlTarget where - E: Send + Sync + 'static + Clone + Serialize + DeserializeOwned, + E: PluginEvent, { fn id(&self) -> TargetID { self.id.clone() @@ -927,7 +926,7 @@ where #[async_trait] impl ReloadableTargetTls for MySqlTarget where - E: Send + Sync + 'static + Clone + Serialize + DeserializeOwned, + E: PluginEvent, { type Material = Pool; diff --git a/crates/targets/src/target/nats.rs b/crates/targets/src/target/nats.rs index b32ba4c6a..569292537 100644 --- a/crates/targets/src/target/nats.rs +++ b/crates/targets/src/target/nats.rs @@ -12,6 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. +use crate::plugin::PluginEvent; use crate::{ StoreError, Target, arn::TargetID, @@ -29,8 +30,6 @@ use crate::{ }; use async_trait::async_trait; use rustfs_config::{NATS_CREDENTIALS_FILE, NATS_TLS_CA, NATS_TLS_CLIENT_CERT, NATS_TLS_CLIENT_KEY}; -use serde::Serialize; -use serde::de::DeserializeOwned; use std::fmt; use std::path::{Path, PathBuf}; use std::str::FromStr; @@ -192,7 +191,7 @@ pub async fn connect_nats(args: &NATSArgs) -> Result where - E: Send + Sync + 'static + Clone + Serialize + DeserializeOwned, + E: PluginEvent, { id: TargetID, args: NATSArgs, @@ -210,7 +209,7 @@ where impl NATSTarget where - E: Send + Sync + 'static + Clone + Serialize + DeserializeOwned, + E: PluginEvent, { pub fn clone_box(&self) -> Box + Send + Sync> { Box::new(NATSTarget:: { @@ -318,7 +317,7 @@ where #[async_trait] impl Target for NATSTarget where - E: Send + Sync + 'static + Clone + Serialize + DeserializeOwned, + E: PluginEvent, { fn id(&self) -> TargetID { self.id.clone() @@ -415,7 +414,7 @@ where #[async_trait] impl ReloadableTargetTls for NATSTarget where - E: Send + Sync + 'static + Clone + Serialize + DeserializeOwned, + E: PluginEvent, { type Material = async_nats::Client; diff --git a/crates/targets/src/target/postgres.rs b/crates/targets/src/target/postgres.rs index c0dc49aa7..b9c7543e1 100644 --- a/crates/targets/src/target/postgres.rs +++ b/crates/targets/src/target/postgres.rs @@ -25,6 +25,7 @@ //! Connection pooling is delegated to `deadpool-postgres`; the pool itself //! is `Clone`, so no `Mutex` is required around it. +use crate::plugin::PluginEvent; use crate::{ StoreError, Target, arn::TargetID, @@ -44,8 +45,6 @@ use async_trait::async_trait; use deadpool_postgres::{Manager, ManagerConfig, Pool, RecyclingMethod}; use rustfs_config::{POSTGRES_DSN_STRING, POSTGRES_TLS_CA, POSTGRES_TLS_CLIENT_CERT, POSTGRES_TLS_CLIENT_KEY}; use rustfs_tls_runtime::{load_certs, load_private_key}; -use serde::Serialize; -use serde::de::DeserializeOwned; use std::fmt; use std::path::Path; use std::sync::Arc; @@ -555,7 +554,7 @@ fn resolve_payload_key(payload: &serde_json::Value, meta: &QueuedPayloadMeta) -> /// the legacy inline fingerprint check is used as a fallback. pub struct PostgresTarget where - E: Send + Sync + 'static + Clone + Serialize + DeserializeOwned, + E: PluginEvent, { id: TargetID, args: PostgresArgs, @@ -573,7 +572,7 @@ where impl PostgresTarget where - E: Send + Sync + 'static + Clone + Serialize + DeserializeOwned, + E: PluginEvent, { pub fn clone_box(&self) -> Box + Send + Sync> { Box::new(PostgresTarget:: { @@ -692,7 +691,7 @@ where #[async_trait] impl ReloadableTargetTls for PostgresTarget where - E: Send + Sync + 'static + Clone + Serialize + DeserializeOwned, + E: PluginEvent, { type Material = Pool; @@ -727,7 +726,7 @@ where #[async_trait] impl Target for PostgresTarget where - E: Send + Sync + 'static + Clone + Serialize + DeserializeOwned, + E: PluginEvent, { fn id(&self) -> TargetID { self.id.clone() diff --git a/crates/targets/src/target/pulsar.rs b/crates/targets/src/target/pulsar.rs index 40c77ac58..2d6bea1ad 100644 --- a/crates/targets/src/target/pulsar.rs +++ b/crates/targets/src/target/pulsar.rs @@ -12,6 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. +use crate::plugin::PluginEvent; use crate::{ StoreError, Target, arn::TargetID, @@ -30,8 +31,6 @@ use crate::{ use async_trait::async_trait; use pulsar::{Authentication, Producer, Pulsar, TokioExecutor}; use rustfs_tls_runtime::load_cert_bundle_der_bytes; -use serde::Serialize; -use serde::de::DeserializeOwned; use std::fmt; use std::path::Path; use std::sync::atomic::{AtomicBool, Ordering}; @@ -186,7 +185,7 @@ pub async fn connect_pulsar(args: &PulsarArgs) -> Result, pub struct PulsarTarget where - E: Send + Sync + 'static + Clone + Serialize + DeserializeOwned, + E: PluginEvent, { id: TargetID, args: PulsarArgs, @@ -203,7 +202,7 @@ where impl PulsarTarget where - E: Send + Sync + 'static + Clone + Serialize + DeserializeOwned, + E: PluginEvent, { pub fn clone_box(&self) -> Box + Send + Sync> { Box::new(PulsarTarget:: { @@ -339,7 +338,7 @@ where #[async_trait] impl ReloadableTargetTls for PulsarTarget where - E: Send + Sync + 'static + Clone + Serialize + DeserializeOwned, + E: PluginEvent, { type Material = Pulsar; @@ -384,7 +383,7 @@ where #[async_trait] impl Target for PulsarTarget where - E: Send + Sync + 'static + Clone + Serialize + DeserializeOwned, + E: PluginEvent, { fn id(&self) -> TargetID { self.id.clone() diff --git a/crates/targets/src/target/redis.rs b/crates/targets/src/target/redis.rs index df8897bd7..94fa7a8eb 100644 --- a/crates/targets/src/target/redis.rs +++ b/crates/targets/src/target/redis.rs @@ -12,6 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. +use crate::plugin::PluginEvent; use crate::{ StoreError, Target, arn::TargetID, @@ -37,8 +38,6 @@ use redis::{ use rustfs_config::{REDIS_TLS_CA, REDIS_TLS_CLIENT_CERT, REDIS_TLS_CLIENT_KEY, REDIS_TLS_POLICY}; use rustls::pki_types::CertificateDer; use rustls::pki_types::pem::PemObject; -use serde::Serialize; -use serde::de::DeserializeOwned; use std::fmt; use std::io::BufReader; use std::path::Path; @@ -300,7 +299,7 @@ fn validate_redis_tls_config(url: &Url, tls: &RedisTlsConfig) -> Result<(), Targ pub struct RedisTarget where - E: Send + Sync + 'static + Clone + Serialize + DeserializeOwned, + E: PluginEvent, { id: TargetID, args: RedisArgs, @@ -326,7 +325,7 @@ where impl RedisTarget where - E: Send + Sync + 'static + Clone + Serialize + DeserializeOwned, + E: PluginEvent, { #[instrument(skip(args), fields(target_id_as_string = %id))] pub fn new(id: String, args: RedisArgs) -> Result { @@ -518,7 +517,7 @@ where #[async_trait] impl Target for RedisTarget where - E: Send + Sync + 'static + Clone + Serialize + DeserializeOwned, + E: PluginEvent, { fn id(&self) -> TargetID { self.id.clone() @@ -796,7 +795,7 @@ fn compute_retry_delay(attempt: usize, min_delay: Duration, max_delay: Duration) #[async_trait] impl ReloadableTargetTls for RedisTarget where - E: Send + Sync + 'static + Clone + Serialize + DeserializeOwned, + E: PluginEvent, { type Material = Client; diff --git a/crates/targets/src/target/webhook.rs b/crates/targets/src/target/webhook.rs index e6d1217ca..1993b2fcd 100644 --- a/crates/targets/src/target/webhook.rs +++ b/crates/targets/src/target/webhook.rs @@ -12,6 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. +use crate::plugin::PluginEvent; use crate::{ StoreError, Target, arn::TargetID, @@ -32,8 +33,6 @@ use parking_lot::Mutex; use reqwest::{Client, StatusCode, Url}; use rustfs_tls_runtime::load_cert_bundle_der_bytes; use rustfs_utils::egress::validate_outbound_url; -use serde::Serialize; -use serde::de::DeserializeOwned; use std::{ fmt, marker::PhantomData, @@ -132,7 +131,7 @@ impl WebhookArgs { /// A target that sends events to a webhook pub struct WebhookTarget where - E: Send + Sync + 'static + Clone + Serialize + DeserializeOwned, + E: PluginEvent, { id: TargetID, args: WebhookArgs, @@ -152,7 +151,7 @@ where impl WebhookTarget where - E: Send + Sync + 'static + Clone + Serialize + DeserializeOwned, + E: PluginEvent, { /// Clones the WebhookTarget, creating a new instance with the same configuration pub fn clone_box(&self) -> Box + Send + Sync> { @@ -476,7 +475,7 @@ where #[async_trait] impl Target for WebhookTarget where - E: Send + Sync + 'static + Clone + Serialize + DeserializeOwned, + E: PluginEvent, { fn id(&self) -> TargetID { self.id.clone() @@ -643,7 +642,7 @@ where #[async_trait] impl ReloadableTargetTls for WebhookTarget where - E: Send + Sync + 'static + Clone + Serialize + DeserializeOwned, + E: PluginEvent, { type Material = Client; diff --git a/rustfs/src/admin/handlers/extensions.rs b/rustfs/src/admin/handlers/extensions.rs index 09199594a..82f66ac34 100644 --- a/rustfs/src/admin/handlers/extensions.rs +++ b/rustfs/src/admin/handlers/extensions.rs @@ -33,10 +33,9 @@ use rustfs_extension_schema::{ }; use rustfs_policy::policy::action::{Action, AdminAction}; use rustfs_targets::{ - OpsDiagnosticsRegistry, OpsProfilerRegistry, TargetPluginExternalFlowGate, TargetPluginExternalFlowGateStatus, - builtin_extension_schemas, builtin_ops_diagnostics_contract, builtin_ops_diagnostics_extension_schema, - builtin_ops_profiler_contract, builtin_ops_profiler_extension_schema, catalog::example_external_webhook_plugin, - target_marketplace_extension_schema, + TargetPluginExternalFlowGate, TargetPluginExternalFlowGateStatus, builtin_extension_schemas, + builtin_ops_diagnostics_contract, builtin_ops_diagnostics_extension_schema, builtin_ops_profiler_contract, + builtin_ops_profiler_extension_schema, }; use s3s::header::CONTENT_TYPE; use s3s::{Body, S3Request, S3Response, S3Result, s3_error}; @@ -105,14 +104,14 @@ pub(crate) struct ExtensionInstanceEntry { pub config: HashMap, #[serde(skip_serializing_if = "Option::is_none")] pub operational_state: Option, - #[serde(default, skip_serializing_if = "Vec::is_empty")] + #[serde(skip_serializing_if = "Vec::is_empty")] pub diagnostic_codes: Vec, } #[derive(Debug, Clone, PartialEq, Eq, Serialize)] pub(crate) struct ExtensionInstancesResponse { pub instances: Vec, - #[serde(default, skip_serializing_if = "Vec::is_empty")] + #[serde(skip_serializing_if = "Vec::is_empty")] pub diagnostic_counts: Vec, pub truncated: bool, pub next_marker: Option, @@ -121,8 +120,6 @@ pub(crate) struct ExtensionInstancesResponse { async fn build_extension_catalog_response() -> Result { 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)); let runtime_capabilities = system::build_runtime_capabilities_response().await?; let cluster_snapshot = cluster_snapshot::build_cluster_snapshot_discovery_response().await; @@ -140,21 +137,8 @@ fn build_extension_runtime_capabilities_response( ) -> ExtensionRuntimeCapabilitiesResponse { let ops_diagnostics_schema = builtin_ops_diagnostics_extension_schema(); let ops_diagnostics_contract = builtin_ops_diagnostics_contract(); - let mut ops_diagnostics_registry = OpsDiagnosticsRegistry::new(); - debug_assert!( - ops_diagnostics_registry - .register_schema(&ops_diagnostics_schema, &ops_diagnostics_contract) - .is_ok() - ); - let ops_profiler_schema = builtin_ops_profiler_extension_schema(); let ops_profiler_contract = builtin_ops_profiler_contract(); - let mut ops_profiler_registry = OpsProfilerRegistry::new(); - debug_assert!( - ops_profiler_registry - .register_schema(&ops_profiler_schema, &ops_profiler_contract) - .is_ok() - ); let runtime_capability_path = Some(format!("{}{}", ADMIN_PREFIX, system::RUNTIME_CAPABILITIES_ROUTE_SUFFIX)); ExtensionRuntimeCapabilitiesResponse { @@ -336,6 +320,40 @@ mod tests { ); } + #[test] + fn builtin_ops_schemas_register_cleanly_in_runtime_registries() { + let mut diagnostics_registry = rustfs_targets::OpsDiagnosticsRegistry::new(); + diagnostics_registry + .register_schema( + &rustfs_targets::builtin_ops_diagnostics_extension_schema(), + &rustfs_targets::builtin_ops_diagnostics_contract(), + ) + .expect("builtin ops diagnostics schema should register"); + + let mut profiler_registry = rustfs_targets::OpsProfilerRegistry::new(); + profiler_registry + .register_schema( + &rustfs_targets::builtin_ops_profiler_extension_schema(), + &rustfs_targets::builtin_ops_profiler_contract(), + ) + .expect("builtin ops profiler schema should register"); + } + + #[tokio::test] + async fn extension_catalog_never_exposes_example_or_external_fixtures() { + let response = build_extension_catalog_response() + .await + .expect("extension catalog response should build"); + + assert!( + !response + .extensions + .iter() + .any(|schema| schema.extension_id.starts_with("external:")), + "example external plugin fixtures must not leak into the production extension catalog" + ); + } + #[tokio::test] async fn extension_catalog_exposes_valid_extension_schemas() { let response = build_extension_catalog_response() diff --git a/rustfs/src/admin/handlers/plugins_catalog.rs b/rustfs/src/admin/handlers/plugins_catalog.rs index 98c3a829e..fc25e9508 100644 --- a/rustfs/src/admin/handlers/plugins_catalog.rs +++ b/rustfs/src/admin/handlers/plugins_catalog.rs @@ -30,10 +30,7 @@ use rustfs_policy::policy::action::{Action, AdminAction}; use rustfs_targets::catalog::{ builtin::builtin_audit_target_admin_descriptors, builtin::builtin_notify_target_admin_descriptors, }; -use rustfs_targets::{ - BuiltinTargetAdminDescriptor, builtin_target_marketplace_manifest, builtin_target_plugin_installation, - catalog::example_external_webhook_plugin, -}; +use rustfs_targets::{BuiltinTargetAdminDescriptor, builtin_target_marketplace_manifest, builtin_target_plugin_installation}; use s3s::header::CONTENT_TYPE; use s3s::{Body, S3Request, S3Response, S3Result, s3_error}; use serde::Serialize; @@ -69,7 +66,6 @@ fn build_catalog_response() -> PluginCatalogResponse { } let mut plugins = plugins.into_values().collect::>(); - plugins.push(example_external_webhook_plugin_entry()); plugins.sort_by(|a, b| a.target_type.cmp(&b.target_type)); for plugin in &mut plugins { plugin.supported_domains.sort(); @@ -86,32 +82,6 @@ fn build_catalog_response() -> PluginCatalogResponse { } } -fn example_external_webhook_plugin_entry() -> PluginCatalogEntry { - let example = example_external_webhook_plugin(); - let manifest = example.manifest; - - PluginCatalogEntry { - plugin_id: manifest.plugin_id.to_string(), - target_type: manifest.target_type.to_string(), - display_name: manifest.display_name.to_string(), - provider: manifest.provider.to_string(), - version: manifest.version.to_string(), - packaging: PluginContractPackaging::from(manifest.packaging), - entrypoint_kind: PluginContractEntrypointKind::from(manifest.entrypoint_kind), - api_compatibility_version: manifest.api_compatibility_version.to_string(), - runtime_contract: PluginRuntimeContract::from(manifest.runtime_contract), - distribution: manifest.distribution.map(PluginDistributionContract::from), - supported_domains: manifest.supported_domains.iter().copied().map(Into::into).collect(), - secret_fields: manifest.secret_fields.iter().map(|field| (*field).to_string()).collect(), - domain_configs: vec![PluginCatalogDomainEntry { - domain: PluginContractDomain::Notify, - subsystem: "notify_webhook".to_string(), - valid_fields: example.valid_fields, - }], - installation: Some(example.installation.into()), - } -} - fn merge_catalog_descriptor(plugins: &mut HashMap<&'static str, PluginCatalogEntry>, descriptor: &BuiltinTargetAdminDescriptor) { let manifest = descriptor.manifest(); let marketplace = builtin_target_marketplace_manifest(manifest.target_type); @@ -251,6 +221,26 @@ mod tests { assert!(kafka.domain_configs.iter().any(|domain| domain.subsystem == "notify_kafka")); } + #[test] + fn plugin_catalog_never_exposes_example_or_external_fixtures() { + let response = build_catalog_response(); + + assert!( + response + .plugins + .iter() + .all(|plugin| plugin.packaging == PluginContractPackaging::Builtin), + "production plugin catalog must only expose builtin plugins until a real marketplace source exists" + ); + assert!( + !response + .plugins + .iter() + .any(|plugin| plugin.plugin_id.starts_with("external:")), + "example external plugin fixtures must not leak into the production catalog" + ); + } + #[test] fn plugin_catalog_exposes_secret_fields_only_as_metadata() { let response = build_catalog_response(); diff --git a/rustfs/src/admin/handlers/plugins_instances.rs b/rustfs/src/admin/handlers/plugins_instances.rs index 66008e067..3e3d9e478 100644 --- a/rustfs/src/admin/handlers/plugins_instances.rs +++ b/rustfs/src/admin/handlers/plugins_instances.rs @@ -43,7 +43,6 @@ use rustfs_config::server_config::{Config, KVS}; use rustfs_config::{AUDIT_DEFAULT_DIR, EVENT_DEFAULT_DIR, MAX_ADMIN_REQUEST_BODY_SIZE}; use rustfs_policy::policy::action::{Action, AdminAction}; use rustfs_targets::catalog::builtin::{builtin_audit_target_admin_descriptors, builtin_notify_target_admin_descriptors}; -use rustfs_targets::manifest::builtin_target_manifest; use rustfs_targets::{builtin_target_plugin_operational_state, runtime_state_from_status_label}; use s3s::{Body, S3Request, S3Response, S3Result, s3_error}; use std::collections::{BTreeMap, HashMap}; @@ -100,27 +99,18 @@ fn audit_target_specs() -> &'static [AdminTargetSpec] { const REDACTED_SECRET_VALUE: &str = "***redacted***"; -fn builtin_secret_fields_for_service(plugin_id: &str, service: &str) -> &'static [&'static str] { - if !plugin_id.starts_with("builtin:") { - return &[]; - } - - match service.to_ascii_lowercase().as_str() { - "webhook" => builtin_target_manifest("webhook").secret_fields, - "mqtt" => builtin_target_manifest("mqtt").secret_fields, - "kafka" => builtin_target_manifest("kafka").secret_fields, - "amqp" => builtin_target_manifest("amqp").secret_fields, - "nats" => builtin_target_manifest("nats").secret_fields, - "pulsar" => builtin_target_manifest("pulsar").secret_fields, - "mysql" => builtin_target_manifest("mysql").secret_fields, - "redis" => builtin_target_manifest("redis").secret_fields, - "postgres" => builtin_target_manifest("postgres").secret_fields, - _ => &[], - } +/// Secret fields for an instance, resolved from the plugin manifest carried by +/// the domain's admin target specs (single source of truth for redaction). +fn secret_fields_for_instance(domain: PluginContractDomain, subsystem: &str) -> &'static [&'static str] { + plugin_instance_domain_context(domain) + .specs + .iter() + .find(|spec| spec.subsystem == subsystem) + .map(|spec| spec.secret_fields) + .unwrap_or(&[]) } -fn map_instance_config(config: KVS, plugin_id: &str, service: &str) -> HashMap { - let secret_fields = builtin_secret_fields_for_service(plugin_id, service); +fn map_instance_config(config: KVS, secret_fields: &[&str]) -> HashMap { config .0 .into_iter() @@ -136,6 +126,46 @@ fn map_instance_config(config: KVS, plugin_id: &str, service: &str) -> HashMap, +) -> S3Result<()> { + for kv in key_values { + if kv.value != REDACTED_SECRET_VALUE { + continue; + } + if !secret_fields.iter().any(|field| field.eq_ignore_ascii_case(&kv.key)) { + continue; + } + let stored = stored_config + .and_then(|kvs| kvs.lookup(&kv.key)) + .filter(|value| !value.is_empty()) + .ok_or_else(|| { + s3_error!( + InvalidArgument, + "config key '{}' uses the redacted placeholder but there is no stored secret to preserve", + kv.key + ) + })?; + kv.value = stored; + } + Ok(()) +} + +/// Case-insensitive lookup of an instance's persisted config section. +fn stored_instance_config<'a>(config: &'a Config, subsystem: &str, target_name: &str) -> Option<&'a KVS> { + config.0.get(subsystem).and_then(|section| { + section + .iter() + .find(|(name, _)| name.eq_ignore_ascii_case(target_name)) + .map(|(_, kvs)| kvs) + }) +} + #[derive(Debug, Clone, Copy)] struct PluginInstanceDomainContext { domain: PluginContractDomain, @@ -163,17 +193,17 @@ fn plugin_instance_domain_context(domain: PluginContractDomain) -> PluginInstanc fn map_instance(instance: TargetInstanceReadModel) -> PluginInstanceEntry { let runtime_state = runtime_state_from_status_label(&instance.status); - let plugin_id = instance.plugin_id; - let service = instance.service; - let config = map_instance_config(instance.config, &plugin_id, &service); + let domain = PluginContractDomain::from(instance.domain); + let secret_fields = secret_fields_for_instance(domain, &instance.subsystem); + let config = map_instance_config(instance.config, secret_fields); PluginInstanceEntry { id: instance.canonical_id, - plugin_id, - domain: PluginContractDomain::from(instance.domain), + plugin_id: instance.plugin_id, + domain, subsystem: instance.subsystem, account_id: instance.account_id, - service, + service: instance.service, status: instance.status, source: map_instance_source(instance.source), enabled: instance.enabled, @@ -236,10 +266,12 @@ fn collect_instance_diagnostics( diagnostics } -async fn plugin_instance_detail(instance: TargetInstanceReadModel) -> PluginInstanceDetail { - let action = "reading plugin instance diagnostics"; - let context = plugin_instance_domain_context(PluginContractDomain::from(instance.domain)); - let diagnostics = collect_instance_diagnostics(&instance, plugin_instance_operation_block_reason(context, action).await); +/// Builds an instance detail view. Callers must run +/// [`refresh_persisted_module_switches`] once beforehand. +fn plugin_instance_detail(instance: TargetInstanceReadModel) -> PluginInstanceDetail { + let domain = PluginContractDomain::from(instance.domain); + let module_disabled_reason = module_disabled_block_reason(domain, "reading plugin instance diagnostics"); + let diagnostics = collect_instance_diagnostics(&instance, module_disabled_reason); let mut mapped = map_instance(instance); mapped.diagnostic_codes = diagnostics.iter().map(|item| item.code.clone()).collect(); @@ -577,15 +609,20 @@ fn plugin_instance_mutation_block_reason( shared_target_mutation_block_reason(context.specs, context.route_prefix, config, target_type, target_name, target_label) } -async fn plugin_instance_operation_block_reason(context: PluginInstanceDomainContext, action: &str) -> Option { +/// Reloads persisted module switches from the config store. Call once per +/// admin request before evaluating [`module_disabled_block_reason`] so that +/// list/detail paths do not hit the store once per domain or per instance. +async fn refresh_persisted_module_switches() { if let Err(err) = refresh_persisted_module_switches_from_store().await { warn!( error = %err, "failed to reload persisted module switches before checking plugin instance operation gating" ); } +} - match context.domain { +fn module_disabled_block_reason(domain: PluginContractDomain, action: &str) -> Option { + match domain { PluginContractDomain::Notify => { refresh_notify_module_enabled(); target_module_disabled_reason("notify", rustfs_config::ENV_NOTIFY_ENABLE, is_notify_module_enabled(), action) @@ -597,6 +634,11 @@ async fn plugin_instance_operation_block_reason(context: PluginInstanceDomainCon } } +async fn plugin_instance_operation_block_reason(context: PluginInstanceDomainContext, action: &str) -> Option { + refresh_persisted_module_switches().await; + module_disabled_block_reason(context.domain, action) +} + async fn plugin_instance_runtime_statuses(context: PluginInstanceDomainContext) -> S3Result> { match context.domain { PluginContractDomain::Notify => { @@ -623,7 +665,7 @@ async fn plugin_instance_config_snapshot(context: PluginInstanceDomainContext) - async fn collect_domain_instances(context: PluginInstanceDomainContext) -> S3Result> { let runtime_statuses = plugin_instance_runtime_statuses(context).await?; let config = plugin_instance_config_snapshot(context).await?; - let module_disabled_reason = plugin_instance_operation_block_reason(context, "listing plugin instances").await; + let module_disabled_reason = module_disabled_block_reason(context.domain, "listing plugin instances"); let mut entries = Vec::new(); for instance in collect_target_instances(context.specs, context.route_prefix, &config, runtime_statuses) { entries.push(plugin_instance_list_entry(instance, module_disabled_reason.clone())); @@ -632,6 +674,7 @@ async fn collect_domain_instances(context: PluginInstanceDomainContext) -> S3Res } pub(super) async fn collect_all_instances() -> S3Result> { + refresh_persisted_module_switches().await; let (mut notify_instances, audit_instances) = tokio::try_join!( collect_domain_instances(plugin_instance_domain_context(PluginContractDomain::Notify)), collect_domain_instances(plugin_instance_domain_context(PluginContractDomain::Audit)) @@ -714,11 +757,12 @@ impl Operation for GetPluginInstanceHandler { .get("id") .ok_or_else(|| s3_error!(InvalidArgument, "missing required parameter: 'id'"))?; + refresh_persisted_module_switches().await; let instance = find_plugin_instance(instance_id) .await? .ok_or_else(|| s3_error!(NoSuchKey, "plugin instance not found"))?; - let data = serde_json::to_vec(&plugin_instance_detail(instance).await) + let data = serde_json::to_vec(&plugin_instance_detail(instance)) .map_err(|e| s3_error!(InternalError, "failed to serialize response: {}", e))?; Ok(build_json_response(StatusCode::OK, Body::from(data), req.headers.get("x-request-id"))) } @@ -746,58 +790,40 @@ impl Operation for PutPluginInstanceHandler { .store_all_limited(MAX_ADMIN_REQUEST_BODY_SIZE) .await .map_err(|_| s3_error!(InvalidRequest, "failed to read request body"))?; - let body: PluginInstanceBody = serde_json::from_slice(&body_bytes) + let mut body: PluginInstanceBody = serde_json::from_slice(&body_bytes) .map_err(|e| s3_error!(InvalidArgument, "invalid json body for plugin instance config: {}", e))?; - match context.domain { - PluginContractDomain::Notify => { - let (_ns, config_snapshot) = load_notification_config_snapshot().await?; - if let Some(reason) = plugin_instance_mutation_block_reason( - context, - &config_snapshot, - resolved.target_spec.subsystem, - &resolved.target_name, - "plugin instance", - ) { - return Err(s3_error!(InvalidRequest, "{reason}")); - } - - let kvs = build_enabled_target_kvs( - context.specs, - body.key_values.iter().map(|kv| (kv.key.as_str(), kv.value.as_str())), - resolved.target_spec.subsystem, - context.default_queue_dir, - "plugin instance", - ) - .await?; - - set_plugin_instance_config(context, &resolved, kvs).await?; - } - PluginContractDomain::Audit => { - let config_snapshot = load_server_config_from_store().await?; - if let Some(reason) = plugin_instance_mutation_block_reason( - context, - &config_snapshot, - resolved.target_spec.subsystem, - &resolved.target_name, - "plugin instance", - ) { - return Err(s3_error!(InvalidRequest, "{reason}")); - } - - let kvs = build_enabled_target_kvs( - context.specs, - body.key_values.iter().map(|kv| (kv.key.as_str(), kv.value.as_str())), - resolved.target_spec.subsystem, - context.default_queue_dir, - "plugin instance", - ) - .await?; - - set_plugin_instance_config(context, &resolved, kvs).await?; - } + let config_snapshot = plugin_instance_config_snapshot(context).await?; + if let Some(reason) = plugin_instance_mutation_block_reason( + context, + &config_snapshot, + resolved.target_spec.subsystem, + &resolved.target_name, + "plugin instance", + ) { + return Err(s3_error!(InvalidRequest, "{reason}")); } + // GET responses redact secret values; a read-modify-write round trip + // sends the placeholder back, which must restore the stored secret + // instead of persisting the placeholder literal. + restore_redacted_secret_values( + &mut body.key_values, + resolved.target_spec.secret_fields, + stored_instance_config(&config_snapshot, resolved.target_spec.subsystem, &resolved.target_name), + )?; + + let kvs = build_enabled_target_kvs( + context.specs, + body.key_values.iter().map(|kv| (kv.key.as_str(), kv.value.as_str())), + resolved.target_spec.subsystem, + context.default_queue_dir, + "plugin instance", + ) + .await?; + + set_plugin_instance_config(context, &resolved, kvs).await?; + Ok(build_json_response(StatusCode::OK, Body::empty(), req.headers.get("x-request-id"))) } } @@ -819,37 +845,19 @@ impl Operation for DeletePluginInstanceHandler { return Err(s3_error!(InvalidRequest, "{reason}")); } - match context.domain { - PluginContractDomain::Notify => { - let (_ns, config_snapshot) = load_notification_config_snapshot().await?; - if let Some(reason) = plugin_instance_mutation_block_reason( - context, - &config_snapshot, - resolved.target_spec.subsystem, - &resolved.target_name, - "plugin instance", - ) { - return Err(s3_error!(InvalidRequest, "{reason}")); - } - - remove_plugin_instance_config(context, &resolved).await?; - } - PluginContractDomain::Audit => { - let config_snapshot = load_server_config_from_store().await?; - if let Some(reason) = plugin_instance_mutation_block_reason( - context, - &config_snapshot, - resolved.target_spec.subsystem, - &resolved.target_name, - "plugin instance", - ) { - return Err(s3_error!(InvalidRequest, "{reason}")); - } - - remove_plugin_instance_config(context, &resolved).await?; - } + let config_snapshot = plugin_instance_config_snapshot(context).await?; + if let Some(reason) = plugin_instance_mutation_block_reason( + context, + &config_snapshot, + resolved.target_spec.subsystem, + &resolved.target_name, + "plugin instance", + ) { + return Err(s3_error!(InvalidRequest, "{reason}")); } + remove_plugin_instance_config(context, &resolved).await?; + Ok(build_json_response(StatusCode::OK, Body::empty(), req.headers.get("x-request-id"))) } } @@ -1073,6 +1081,71 @@ mod tests { ); } + #[test] + fn put_round_trip_restores_stored_secret_for_redacted_placeholder() { + let mut key_values = vec![ + super::KeyValue { + key: WEBHOOK_ENDPOINT.to_string(), + value: "https://example.com/webhook".to_string(), + }, + super::KeyValue { + key: WEBHOOK_AUTH_TOKEN.to_string(), + value: super::REDACTED_SECRET_VALUE.to_string(), + }, + ]; + let mut stored = KVS::new(); + stored.insert(WEBHOOK_AUTH_TOKEN.to_string(), "stored-secret-token".to_string()); + + super::restore_redacted_secret_values(&mut key_values, &[WEBHOOK_AUTH_TOKEN], Some(&stored)) + .expect("redacted secret should be restored from stored config"); + + assert_eq!(key_values[0].value, "https://example.com/webhook"); + assert_eq!(key_values[1].value, "stored-secret-token"); + } + + #[test] + fn put_round_trip_rejects_redacted_placeholder_without_stored_secret() { + let mut key_values = vec![super::KeyValue { + key: WEBHOOK_AUTH_TOKEN.to_string(), + value: super::REDACTED_SECRET_VALUE.to_string(), + }]; + + let err = super::restore_redacted_secret_values(&mut key_values, &[WEBHOOK_AUTH_TOKEN], None) + .expect_err("placeholder without a stored secret must be rejected"); + assert!(err.to_string().contains("no stored secret")); + + let empty_stored = KVS::new(); + let err = super::restore_redacted_secret_values(&mut key_values, &[WEBHOOK_AUTH_TOKEN], Some(&empty_stored)) + .expect_err("placeholder with an empty stored secret must be rejected"); + assert!(err.to_string().contains("no stored secret")); + } + + #[test] + fn put_round_trip_leaves_placeholder_on_non_secret_fields_untouched() { + let mut key_values = vec![super::KeyValue { + key: WEBHOOK_ENDPOINT.to_string(), + value: super::REDACTED_SECRET_VALUE.to_string(), + }]; + + super::restore_redacted_secret_values(&mut key_values, &[WEBHOOK_AUTH_TOKEN], None) + .expect("non-secret fields are not subject to placeholder restoration"); + assert_eq!(key_values[0].value, super::REDACTED_SECRET_VALUE); + } + + #[test] + fn stored_instance_config_lookup_is_case_insensitive_on_target_name() { + let mut stored = KVS::new(); + stored.insert(WEBHOOK_AUTH_TOKEN.to_string(), "stored-secret-token".to_string()); + let config = Config(HashMap::from([( + NOTIFY_WEBHOOK_SUB_SYS.to_string(), + HashMap::from([("Primary".to_string(), stored)]), + )])); + + let found = super::stored_instance_config(&config, NOTIFY_WEBHOOK_SUB_SYS, "primary") + .expect("case-insensitive lookup should find the stored section"); + assert_eq!(found.lookup(WEBHOOK_AUTH_TOKEN).as_deref(), Some("stored-secret-token")); + } + #[test] fn plugin_instance_body_rejects_unknown_fields() { let err = serde_json::from_str::(r#"{"key_values":[],"unexpected_field":true}"#) diff --git a/rustfs/src/admin/handlers/target_descriptor.rs b/rustfs/src/admin/handlers/target_descriptor.rs index 98210fd48..2c115f61c 100644 --- a/rustfs/src/admin/handlers/target_descriptor.rs +++ b/rustfs/src/admin/handlers/target_descriptor.rs @@ -94,6 +94,9 @@ pub(crate) struct AdminTargetSpec { pub subsystem: &'static str, pub service: &'static str, pub valid_keys: &'static [&'static str], + /// Config keys whose values are secrets, sourced from the plugin manifest + /// so redaction never depends on hand-maintained per-service tables. + pub secret_fields: &'static [&'static str], validator: AdminRequestValidatorFn, } @@ -103,6 +106,7 @@ pub(crate) fn admin_target_spec_from_builtin(descriptor: &BuiltinTargetAdminDesc subsystem: admin.subsystem(), service: descriptor.manifest().target_type, valid_keys: descriptor.valid_fields(), + secret_fields: descriptor.manifest().secret_fields, validator: validator_from_metadata(admin), } } diff --git a/rustfs/src/admin/plugin_contract.rs b/rustfs/src/admin/plugin_contract.rs index fad9d2df1..880f167fb 100644 --- a/rustfs/src/admin/plugin_contract.rs +++ b/rustfs/src/admin/plugin_contract.rs @@ -163,24 +163,24 @@ pub(crate) struct PluginInstallationContract { pub validation_error: Option, } +impl From for PluginRevisionContract { + fn from(value: rustfs_targets::TargetPluginRevision) -> Self { + Self { + version: value.version, + digest_sha256: value.digest_sha256, + source: value.source, + installed_at: value.installed_at, + artifact_id: value.artifact_id, + } + } +} + impl From for PluginInstallationContract { fn from(value: TargetPluginInstallation) -> Self { Self { install_state: PluginInstallState::from(value.install_state), - current_revision: value.current_revision.map(|revision| PluginRevisionContract { - version: revision.version, - digest_sha256: revision.digest_sha256, - source: revision.source, - installed_at: revision.installed_at, - artifact_id: revision.artifact_id, - }), - previous_revision: value.previous_revision.map(|revision| PluginRevisionContract { - version: revision.version, - digest_sha256: revision.digest_sha256, - source: revision.source, - installed_at: revision.installed_at, - artifact_id: revision.artifact_id, - }), + current_revision: value.current_revision.map(PluginRevisionContract::from), + previous_revision: value.previous_revision.map(PluginRevisionContract::from), validation_error: value.validation_error, } } @@ -328,7 +328,7 @@ pub(crate) struct PluginInstanceEntry { pub config: HashMap, #[serde(skip_serializing_if = "Option::is_none")] pub operational_state: Option, - #[serde(default, skip_serializing_if = "Vec::is_empty")] + #[serde(skip_serializing_if = "Vec::is_empty")] pub diagnostic_codes: Vec, } @@ -361,14 +361,14 @@ pub(crate) struct PluginInstanceDiagnosticCount { pub(crate) struct PluginInstanceDetail { #[serde(flatten)] pub instance: PluginInstanceEntry, - #[serde(default, skip_serializing_if = "Vec::is_empty")] + #[serde(skip_serializing_if = "Vec::is_empty")] pub diagnostics: Vec, } #[derive(Debug, Clone, PartialEq, Eq, Serialize)] pub(crate) struct PluginInstancesResponse { pub instances: Vec, - #[serde(default, skip_serializing_if = "Vec::is_empty")] + #[serde(skip_serializing_if = "Vec::is_empty")] pub diagnostic_counts: Vec, pub truncated: bool, pub next_marker: Option,