diff --git a/Cargo.lock b/Cargo.lock index cd8dee77a..957dc25ba 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -9080,6 +9080,7 @@ dependencies = [ "rustfs-policy", "rustfs-protocols", "rustfs-protos", + "rustfs-replication", "rustfs-rio", "rustfs-s3-ops", "rustfs-s3-types", diff --git a/crates/replication/src/config.rs b/crates/replication/src/config.rs index c5d46d61c..8e072f354 100644 --- a/crates/replication/src/config.rs +++ b/crates/replication/src/config.rs @@ -26,6 +26,53 @@ use serde::{Deserialize, Serialize}; use std::collections::{HashMap, HashSet}; use uuid::Uuid; +pub const REPLICATION_CAPABILITY_CONTRACT_VERSION: u32 = 1; + +pub const REPLICATION_WRITABLE_FIELDS: &[&str] = &[ + "Role", + "Rule.ID", + "Rule.Status", + "Rule.Priority", + "Rule.Filter.Prefix", + "Rule.Filter.Tag", + "Rule.Filter.And", + "Rule.Destination.Bucket", + "Rule.ExistingObjectReplication.Status", + "Rule.DeleteMarkerReplication.Status", + "Rule.DeleteReplication.Status", + "Rule.SourceSelectionCriteria.ReplicaModifications.Status", +]; + +pub const REPLICATION_READ_ONLY_HISTORICAL_FIELDS: &[&str] = &[ + "SourceSelectionCriteria.SseKmsEncryptedObjects", + "Destination.EncryptionConfiguration", + "Destination.Metrics", + "Destination.ReplicationTime", +]; + +pub const REMOTE_TARGET_CAPABILITY_CONTRACT_VERSION: u32 = 1; + +pub const REMOTE_TARGET_WRITABLE_FIELDS: &[&str] = &[ + "sourcebucket", + "endpoint", + "credentials.accessKey", + "credentials.secretKey", + "targetbucket", + "secure", + "path", + "api", + "arn", + "type", + "region", + "bandwidth", + "replicationSync", + "storage_class", + "skipTlsVerify", + "caCertPem", +]; + +pub const REMOTE_TARGET_UNSUPPORTED_FIELDS: &[&str] = &["disableProxy", "healthCheckDuration", "edge", "edgeSyncBeforeExpiry"]; + #[derive(Debug, Clone, Serialize, Deserialize, Default)] pub struct ObjectOpts { pub name: String, @@ -831,6 +878,34 @@ mod tests { assert_eq!(unsupported_replication_config_field(&config), Some("Destination.ReplicationTime")); } + #[test] + fn capability_fields_match_validator_rejections() { + let rejected_fields = [ + "SourceSelectionCriteria.SseKmsEncryptedObjects", + "Destination.EncryptionConfiguration", + "Destination.Metrics", + "Destination.ReplicationTime", + ]; + + for field in rejected_fields { + assert!( + REPLICATION_READ_ONLY_HISTORICAL_FIELDS.contains(&field), + "rejected field {field} must be advertised as readable historical data" + ); + assert!( + !REPLICATION_WRITABLE_FIELDS.contains(&field), + "rejected field {field} must not be advertised as writable" + ); + } + + for field in REPLICATION_WRITABLE_FIELDS { + assert!( + !REPLICATION_READ_ONLY_HISTORICAL_FIELDS.contains(field), + "field {field} cannot be both writable and historical-only" + ); + } + } + #[test] fn invalid_replication_status_fields_are_reported_before_persistence() { let arn = "arn:rustfs:replication:us-east-1:target:bucket"; diff --git a/crates/replication/src/lib.rs b/crates/replication/src/lib.rs index fda3da1fd..cf448d738 100644 --- a/crates/replication/src/lib.rs +++ b/crates/replication/src/lib.rs @@ -29,7 +29,9 @@ mod storage_api; pub mod tagging; pub use config::{ - ObjectOpts, ReplicationConfigurationExt, ReplicationTargetValidationError, active_replication_rule_destination_arns, + ObjectOpts, REMOTE_TARGET_CAPABILITY_CONTRACT_VERSION, REMOTE_TARGET_UNSUPPORTED_FIELDS, REMOTE_TARGET_WRITABLE_FIELDS, + REPLICATION_CAPABILITY_CONTRACT_VERSION, REPLICATION_READ_ONLY_HISTORICAL_FIELDS, REPLICATION_WRITABLE_FIELDS, + ReplicationConfigurationExt, ReplicationTargetValidationError, active_replication_rule_destination_arns, invalid_replication_config_status_field, replication_target_arns, should_remove_replication_target, unsupported_replication_config_field, validate_replication_config_target_arns, }; diff --git a/rustfs/Cargo.toml b/rustfs/Cargo.toml index 38e626b3d..4749a5175 100644 --- a/rustfs/Cargo.toml +++ b/rustfs/Cargo.toml @@ -217,6 +217,7 @@ rustfs-notify = { workspace = true } rustfs-obs = { workspace = true } rustfs-policy = { workspace = true } rustfs-protocols = { workspace = true } +rustfs-replication = { workspace = true } rustfs-protos = { workspace = true } rustfs-rio = { workspace = true } rustfs-s3-types = { workspace = true } diff --git a/rustfs/src/admin/handlers/replication.rs b/rustfs/src/admin/handlers/replication.rs index 8f71bdec6..6b73131f9 100644 --- a/rustfs/src/admin/handlers/replication.rs +++ b/rustfs/src/admin/handlers/replication.rs @@ -40,6 +40,7 @@ use matchit::Params; use rustfs_config::MAX_ADMIN_REQUEST_BODY_SIZE; use rustfs_credentials::Credentials; use rustfs_policy::policy::action::{Action, AdminAction}; +use rustfs_replication::{REMOTE_TARGET_UNSUPPORTED_FIELDS, REMOTE_TARGET_WRITABLE_FIELDS}; use s3s::header::CONTENT_TYPE; use s3s::{Body, S3Error, S3ErrorCode, S3Request, S3Response, S3Result, s3_error}; use serde::{Deserialize, Serialize}; @@ -179,12 +180,12 @@ impl RemoteTargetRequest { return Err(s3_error!(InvalidRequest, "credentials.secretKey is required")); } - for (unsupported, configured) in [ - ("disableProxy", self.disable_proxy), - ("healthCheckDuration", self.health_check_duration != 0), - ("edge", self.edge), - ("edgeSyncBeforeExpiry", self.edge_sync_before_expiry), - ] { + for (unsupported, configured) in REMOTE_TARGET_UNSUPPORTED_FIELDS.iter().copied().zip([ + self.disable_proxy, + self.health_check_duration != 0, + self.edge, + self.edge_sync_before_expiry, + ]) { if configured { return Err(s3_error!( InvalidRequest, @@ -1021,8 +1022,8 @@ impl Operation for ReplicationMrfHandler { #[cfg(test)] mod tests { use super::{ - RemoteTargetRequest, build_mrf_response, extract_query_params, unique_replication_peers, - validate_remote_target_tls_settings, + REMOTE_TARGET_UNSUPPORTED_FIELDS, REMOTE_TARGET_WRITABLE_FIELDS, RemoteTargetRequest, build_mrf_response, + extract_query_params, unique_replication_peers, validate_remote_target_tls_settings, }; use crate::admin::storage_api::bucket::target::BucketTarget; use crate::admin::storage_api::replication::{BucketStats, DurableMrfBacklog, MrfOpKind, MrfReplicateEntry}; @@ -1313,12 +1314,12 @@ mod tests { #[test] fn remote_target_request_rejects_unimplemented_fields() { - for (field, value) in [ - ("disableProxy", serde_json::json!(true)), - ("healthCheckDuration", serde_json::json!(5)), - ("edge", serde_json::json!(true)), - ("edgeSyncBeforeExpiry", serde_json::json!(true)), - ] { + for (field, value) in REMOTE_TARGET_UNSUPPORTED_FIELDS.iter().copied().zip([ + serde_json::json!(true), + serde_json::json!(5), + serde_json::json!(true), + serde_json::json!(true), + ]) { let mut request = valid_remote_target_request(); request[field] = value; let request: RemoteTargetRequest = @@ -1332,6 +1333,16 @@ mod tests { } } + #[test] + fn remote_target_capability_fields_do_not_overlap() { + for field in REMOTE_TARGET_UNSUPPORTED_FIELDS { + assert!( + !REMOTE_TARGET_WRITABLE_FIELDS.contains(field), + "remote target field {field} cannot be both writable and unsupported" + ); + } + } + #[test] fn remote_target_request_converts_to_bucket_target() { let target = serde_json::from_value::(valid_remote_target_request()) diff --git a/rustfs/src/admin/handlers/system.rs b/rustfs/src/admin/handlers/system.rs index 010bc5eb0..1a2088868 100644 --- a/rustfs/src/admin/handlers/system.rs +++ b/rustfs/src/admin/handlers/system.rs @@ -35,6 +35,10 @@ use matchit::Params; use rustfs_concurrency::WorkloadAdmissionRegistrySnapshot; use rustfs_madmin::{InfoMessage, StorageInfo}; use rustfs_policy::policy::action::{Action, AdminAction, S3Action}; +use rustfs_replication::{ + REMOTE_TARGET_CAPABILITY_CONTRACT_VERSION, REMOTE_TARGET_UNSUPPORTED_FIELDS, REMOTE_TARGET_WRITABLE_FIELDS, + REPLICATION_CAPABILITY_CONTRACT_VERSION, REPLICATION_READ_ONLY_HISTORICAL_FIELDS, REPLICATION_WRITABLE_FIELDS, +}; use rustfs_security_governance::{AdminRouteSpec, HttpMethod}; use s3s::header::CONTENT_TYPE; use s3s::{Body, S3Error, S3ErrorCode, S3Request, S3Response, S3Result, s3_error}; @@ -644,6 +648,7 @@ pub struct RuntimeCapabilitiesSummary { #[derive(Debug, Clone, PartialEq, Eq, Serialize)] pub struct RuntimeCapabilitiesResponse { pub summary: RuntimeCapabilitiesSummary, + pub replication: ReplicationCapabilities, pub manual_transition_jobs: ManualTransitionJobCapabilities, pub diagnostic_probes: DiagnosticProbeCapabilities, pub inspect_archive: super::inspect_archive::InspectArchiveCapability, @@ -657,6 +662,100 @@ pub struct RuntimeCapabilitiesResponse { pub topology_status: CapabilityStatus, } +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)] +#[serde(rename_all = "snake_case")] +pub enum ReplicationFieldState { + Supported, + ReadOnlyHistorical, + Unsupported, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)] +pub struct ReplicationFieldCapability { + pub name: &'static str, + pub state: ReplicationFieldState, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize)] +pub struct ReplicationFeatureCapabilities { + pub contract_version: u32, + pub status: CapabilityStatus, + pub fields: Vec, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize)] +pub struct ReplicationCapabilities { + pub contract_version: u32, + pub bucket_replication: ReplicationFeatureCapabilities, + pub remote_targets: ReplicationFeatureCapabilities, + pub mixed_version_policy: &'static str, +} + +impl ReplicationCapabilities { + fn current() -> Self { + let remote_target_routes = [ + admin_route_capability(HttpMethod::Get, "/rustfs/admin/v3/list-remote-targets"), + admin_route_capability(HttpMethod::Put, "/rustfs/admin/v3/set-remote-target"), + admin_route_capability(HttpMethod::Delete, "/rustfs/admin/v3/remove-remote-target"), + ]; + let remote_targets_supported = remote_target_routes + .iter() + .all(|status| status.state == CapabilityState::Supported); + + Self { + contract_version: REPLICATION_CAPABILITY_CONTRACT_VERSION, + bucket_replication: ReplicationFeatureCapabilities { + contract_version: REPLICATION_CAPABILITY_CONTRACT_VERSION, + status: CapabilityStatus::supported() + .with_reason("bucket replication validation and execution contracts are registered"), + fields: REPLICATION_WRITABLE_FIELDS + .iter() + .copied() + .map(|name| ReplicationFieldCapability { + name, + state: ReplicationFieldState::Supported, + }) + .chain( + REPLICATION_READ_ONLY_HISTORICAL_FIELDS + .iter() + .copied() + .map(|name| ReplicationFieldCapability { + name, + state: ReplicationFieldState::ReadOnlyHistorical, + }), + ) + .collect(), + }, + remote_targets: ReplicationFeatureCapabilities { + contract_version: REMOTE_TARGET_CAPABILITY_CONTRACT_VERSION, + status: if remote_targets_supported { + CapabilityStatus::supported().with_reason("remote target routes and field validation are registered") + } else { + CapabilityStatus::unsupported().with_reason("one or more remote target routes are unavailable") + }, + fields: REMOTE_TARGET_WRITABLE_FIELDS + .iter() + .copied() + .map(|name| ReplicationFieldCapability { + name, + state: ReplicationFieldState::Supported, + }) + .chain( + REMOTE_TARGET_UNSUPPORTED_FIELDS + .iter() + .copied() + .map(|name| ReplicationFieldCapability { + name, + state: ReplicationFieldState::Unsupported, + }), + ) + .collect(), + }, + mixed_version_policy: "fail_closed_when_capability_unknown_or_unsupported", + } + } +} + #[derive(Debug, Clone, PartialEq, Eq, Serialize)] pub struct ManualTransitionJobCapabilities { pub contract_version: u32, @@ -871,6 +970,7 @@ pub(crate) async fn build_runtime_capabilities_response() Ok(RuntimeCapabilitiesResponse { summary, + replication: ReplicationCapabilities::current(), manual_transition_jobs: ManualTransitionJobCapabilities::current(), diagnostic_probes: DiagnosticProbeCapabilities::current(), inspect_archive: super::inspect_archive::InspectArchiveCapability::current(local_drive_count), @@ -1144,6 +1244,41 @@ mod tests { assert_eq!(response.summary.site_replication_resync.state, CapabilityState::Supported); assert_eq!(response.summary.site_replication_repair.state, CapabilityState::Supported); assert_eq!(response.summary.manual_transition_jobs.state, CapabilityState::Supported); + assert_eq!(response.replication.contract_version, 1); + assert_eq!(response.replication.bucket_replication.contract_version, 1); + assert_eq!(response.replication.remote_targets.contract_version, 1); + assert_eq!(response.replication.bucket_replication.status.state, CapabilityState::Supported); + assert_eq!(response.replication.remote_targets.status.state, CapabilityState::Supported); + assert_eq!( + response.replication.mixed_version_policy, + "fail_closed_when_capability_unknown_or_unsupported" + ); + assert!( + response + .replication + .bucket_replication + .fields + .iter() + .any(|field| field.name == "Rule.DeleteReplication.Status" + && field.state == super::ReplicationFieldState::Supported) + ); + assert!( + response + .replication + .bucket_replication + .fields + .iter() + .any(|field| field.name == "Destination.EncryptionConfiguration" + && field.state == super::ReplicationFieldState::ReadOnlyHistorical) + ); + assert!( + response + .replication + .remote_targets + .fields + .iter() + .any(|field| field.name == "disableProxy" && field.state == super::ReplicationFieldState::Unsupported) + ); assert_eq!(response.manual_transition_jobs.contract_version, 1); assert_eq!(response.manual_transition_jobs.status.state, CapabilityState::Supported); assert_eq!(response.manual_transition_jobs.modes, ["enqueue_only", "async"]); @@ -1203,6 +1338,29 @@ mod tests { assert_eq!(value["summary"]["site_replication_resync"]["state"], "supported"); assert_eq!(value["summary"]["site_replication_repair"]["state"], "supported"); assert_eq!(value["summary"]["manual_transition_jobs"]["state"], "supported"); + assert_eq!(value["replication"]["contract_version"], 1); + assert_eq!(value["replication"]["bucket_replication"]["contract_version"], 1); + assert_eq!(value["replication"]["remote_targets"]["contract_version"], 1); + assert_eq!(value["replication"]["bucket_replication"]["status"]["state"], "supported"); + assert_eq!(value["replication"]["remote_targets"]["status"]["state"], "supported"); + assert_eq!( + value["replication"]["mixed_version_policy"], + "fail_closed_when_capability_unknown_or_unsupported" + ); + assert!( + value["replication"]["bucket_replication"]["fields"] + .as_array() + .expect("bucket replication fields should be an array") + .iter() + .any(|field| field["name"] == "Destination.EncryptionConfiguration" && field["state"] == "read_only_historical") + ); + assert!( + value["replication"]["remote_targets"]["fields"] + .as_array() + .expect("remote target fields should be an array") + .iter() + .any(|field| field["name"] == "disableProxy" && field["state"] == "unsupported") + ); assert_eq!(value["manual_transition_jobs"]["contract_version"], 1); assert_eq!(value["manual_transition_jobs"]["status"]["state"], "supported"); assert_eq!(value["manual_transition_jobs"]["modes"], json!(["enqueue_only", "async"]));