From cd327c81f53757d6ea1da677c0eb21a1c5439089 Mon Sep 17 00:00:00 2001 From: houseme Date: Thu, 9 Jul 2026 00:18:30 +0800 Subject: [PATCH] fix(targets): harden control-plane, TLS reload, and runtime extension points (#4504) Addresses security/correctness audit findings in the target-plugin control plane, TLS reload coordinator, and runtime extension registries. control_plane (backlog#977): - Install now preserves the currently installed revision as previous_revision so Rollback actually restores the prior version instead of a no-op. - Split the gate: circuit-breaker and runtime-activation checks apply only to Install/Enable; Disable and Rollback stay available as break-glass remediation while the breaker is open. - Enforce the sidecar runtime protocol version for every external transport (previously skippable by declaring a non-gRPC transport) and validate the plugin api_compatibility_version at planning time. - Download/signature/provenance host allowlisting now matches the full host authority, so an allowlisted host never implicitly authorizes a different host:port. TLS reload coordinator/validate (backlog#981, #970-coordinator): - Compute the fingerprint before building material in register() to remove the TOCTOU that could permanently pin an old certificate; rotation self-heals. - Always start a detection loop when reload is enabled; Watch mode no longer returns success without any loop. - validate_cert_key_pairing now verifies the private key matches the certificate's public key instead of only parsing both files. - Normalize a zero poll interval to a positive minimum to avoid a panic that silently killed the poll loop. - Serialize reload cycles with a per-target mutex; a first-step fingerprint read failure now records last_error and a failure metric. - Duplicate label registration stops and joins the previous loop before publishing the replacement (stop-before-start), preventing orphaned loops. runtime extension points (backlog#983 runtime subset): - ops_profiler/ops_diagnostics authorize before probing the registry so an unauthorized caller cannot learn whether a backend/surface exists. - s3_hooks dispatch_post_auth actually traverses the registered hooks for the point instead of unconditionally returning Continue. - sidecar send_with_timeout redacts errors and uses the configurable failure threshold via policy, and a successful send resets the breaker accounting. Relates to rustfs/backlog#977 Relates to rustfs/backlog#981 Relates to rustfs/backlog#970 Relates to rustfs/backlog#983 Co-authored-by: heihutu --- Cargo.lock | 1 + crates/targets/Cargo.toml | 1 + crates/targets/src/control_plane.rs | 306 ++++++++++++++++-- crates/targets/src/lib.rs | 9 +- crates/targets/src/manifest.rs | 7 +- crates/targets/src/runtime/ops_diagnostics.rs | 37 ++- crates/targets/src/runtime/ops_profiler.rs | 36 ++- crates/targets/src/runtime/s3_hooks.rs | 55 +++- crates/targets/src/runtime/sidecar.rs | 63 +++- crates/targets/src/runtime/tls/coordinator.rs | 170 ++++++++-- crates/targets/src/runtime/tls/state.rs | 5 + crates/targets/src/runtime/tls/validate.rs | 90 +++++- 12 files changed, 700 insertions(+), 80 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index bfd252fd5..65d82ea23 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -9947,6 +9947,7 @@ dependencies = [ "mysql_async", "parking_lot", "pulsar", + "rcgen", "redis", "regex", "reqwest", diff --git a/crates/targets/Cargo.toml b/crates/targets/Cargo.toml index 0cc2000e8..570ca0b09 100644 --- a/crates/targets/Cargo.toml +++ b/crates/targets/Cargo.toml @@ -55,6 +55,7 @@ metrics = { workspace = true } [dev-dependencies] criterion = { workspace = true } tempfile = { workspace = true } +rcgen = { workspace = true } [[bench]] name = "queue_store_benchmark" diff --git a/crates/targets/src/control_plane.rs b/crates/targets/src/control_plane.rs index 14b6d5ce2..df4b3f70a 100644 --- a/crates/targets/src/control_plane.rs +++ b/crates/targets/src/control_plane.rs @@ -13,8 +13,8 @@ // limitations under the License. use crate::manifest::{ - TargetPluginDistributionManifest, TargetPluginExternalRuntimeContract, TargetPluginManifest, TargetPluginMarketplaceManifest, - TargetPluginPackaging, TargetPluginRuntimeTransport, + SUPPORTED_PLUGIN_API_COMPATIBILITY_VERSION, TargetPluginDistributionManifest, TargetPluginExternalRuntimeContract, + TargetPluginManifest, TargetPluginMarketplaceManifest, TargetPluginPackaging, }; use crate::runtime::sidecar::{SidecarRuntimePolicy, SidecarRuntimeSafetyChecks}; use crate::runtime::sidecar_protocol::SIDECAR_RUNTIME_PROTOCOL_VERSION; @@ -288,11 +288,24 @@ pub fn plan_external_target_plugin_action( host_target_triple: &str, ) -> Result { validate_external_action_subject(manifest)?; - validate_external_action_gate(gate)?; + // The external-plugin flow master switch gates every action. + if !gate.enabled { + return Err(TargetPluginExternalActionError::ExternalFlowDisabled); + } + + // Gate split: the circuit breaker and runtime-activation checks only guard + // the *activating* actions (Install/Enable). Disable and Rollback are + // break-glass remediation and must stay available precisely when the + // breaker is open — otherwise a failing plugin can never be stopped or + // rolled back. match action { - TargetPluginExternalAction::Install => plan_external_install(manifest, action, gate, host_target_triple), + TargetPluginExternalAction::Install => { + validate_external_activation_gate(gate)?; + plan_external_install(manifest, action, installation, gate, host_target_triple) + } TargetPluginExternalAction::Enable => { + validate_external_activation_gate(gate)?; require_installed(manifest.plugin_id, installation)?; Ok(TargetPluginExternalActionDecision { action, @@ -355,9 +368,10 @@ pub fn validate_external_plugin_installation( return Err(format!("provider {} is not allowed by install policy", manifest.provider)); } - if runtime_contract.transport == TargetPluginRuntimeTransport::Grpc - && runtime_contract.protocol_version != SIDECAR_RUNTIME_PROTOCOL_VERSION - { + // Enforce the sidecar runtime protocol version for *every* external + // transport. Gating this behind a specific transport let an external + // plugin skip the check simply by declaring a different transport. + if runtime_contract.protocol_version != SIDECAR_RUNTIME_PROTOCOL_VERSION { return Err(format!( "sidecar runtime protocol mismatch: expected {}, got {}", SIDECAR_RUNTIME_PROTOCOL_VERSION, runtime_contract.protocol_version @@ -378,11 +392,10 @@ pub fn validate_external_plugin_installation( artifact.artifact_id, artifact.download_uri )); } - let host = parsed_uri - .host_str() + let authority = uri_host_authority(&parsed_uri) .ok_or_else(|| format!("artifact {} download uri has no host", artifact.artifact_id))?; - if !policy.allowed_download_hosts.iter().any(|allowed| allowed == host) { - return Err(format!("artifact {} download host {} is not allowed", artifact.artifact_id, host)); + if !policy.allowed_download_hosts.iter().any(|allowed| allowed == &authority) { + return Err(format!("artifact {} download host {} is not allowed", artifact.artifact_id, authority)); } if artifact.size_bytes == 0 { return Err(format!("artifact {} must declare a non-zero size", artifact.artifact_id)); @@ -418,10 +431,9 @@ fn validate_external_action_subject(manifest: &TargetPluginMarketplaceManifest) Ok(()) } -fn validate_external_action_gate(gate: &TargetPluginExternalFlowGate) -> Result<(), TargetPluginExternalActionError> { - if !gate.enabled { - return Err(TargetPluginExternalActionError::ExternalFlowDisabled); - } +/// Activation gate for Install/Enable only: circuit breaker plus runtime policy +/// activation. Callers must have already checked `gate.enabled`. +fn validate_external_activation_gate(gate: &TargetPluginExternalFlowGate) -> Result<(), TargetPluginExternalActionError> { if !gate.circuit_breaker_closed { return Err(TargetPluginExternalActionError::CircuitBreakerOpen); } @@ -435,9 +447,21 @@ fn validate_external_action_gate(gate: &TargetPluginExternalFlowGate) -> Result< fn plan_external_install( manifest: &TargetPluginMarketplaceManifest, action: TargetPluginExternalAction, + installation: &TargetPluginInstallation, gate: &TargetPluginExternalFlowGate, host_target_triple: &str, ) -> Result { + // The plugin API compatibility version is validated at planning time so an + // artifact built against an unsupported contract can never be installed. + if manifest.api_compatibility_version != SUPPORTED_PLUGIN_API_COMPATIBILITY_VERSION { + return Err(TargetPluginExternalActionError::InstallPolicyDenied { + reason: format!( + "plugin api compatibility version mismatch: expected {}, got {}", + SUPPORTED_PLUGIN_API_COMPATIBILITY_VERSION, manifest.api_compatibility_version + ), + }); + } + let install_manifest = TargetPluginManifest { plugin_id: manifest.plugin_id, display_name: manifest.display_name, @@ -468,10 +492,20 @@ fn plan_external_install( target_triple: host_target_triple.to_string(), })?; + let mut new_installation = + external_target_plugin_installation(manifest.version, artifact.digest_sha256, artifact.artifact_id, None); + // Preserve the currently installed revision as `previous_revision` so a + // later Rollback can restore it. Dropping it would make rollback a no-op. + if installation.install_state == TargetPluginInstallState::Installed + && let Some(current) = installation.current_revision.clone() + { + new_installation.previous_revision = Some(current); + } + Ok(TargetPluginExternalActionDecision { action, plugin_id: manifest.plugin_id.to_string(), - installation: external_target_plugin_installation(manifest.version, artifact.digest_sha256, artifact.artifact_id, None), + installation: new_installation, operational_state: TargetPluginOperationalState { install_state: TargetPluginInstallState::Installed, enable_state: TargetPluginEnableState::Disabled, @@ -529,16 +563,25 @@ fn validate_artifact_uri(label: &str, artifact_id: &str, uri: &str, policy: &Tar if policy.require_https && parsed_uri.scheme() != "https" { return Err(format!("artifact {artifact_id} must use https {label} uri, got {uri}")); } - let host = parsed_uri - .host_str() - .ok_or_else(|| format!("artifact {artifact_id} {label} uri has no host"))?; - if !policy.allowed_download_hosts.iter().any(|allowed| allowed == host) { - return Err(format!("artifact {artifact_id} {label} host {host} is not allowed")); + let authority = uri_host_authority(&parsed_uri).ok_or_else(|| format!("artifact {artifact_id} {label} uri has no host"))?; + if !policy.allowed_download_hosts.iter().any(|allowed| allowed == &authority) { + return Err(format!("artifact {artifact_id} {label} host {authority} is not allowed")); } Ok(()) } +/// Returns the host authority used for allowlist matching. When the URI carries +/// an explicit port the port is part of the authority, so that an allowlisted +/// `host` never implicitly authorizes a different `host:port`. A default-port +/// URI (`port()` is `None`) matches a bare `host` entry. +fn uri_host_authority(parsed_uri: &Url) -> Option { + parsed_uri.host_str().map(|host| match parsed_uri.port() { + Some(port) => format!("{host}:{port}"), + None => host.to_string(), + }) +} + #[cfg(test)] mod tests { use super::{ @@ -1134,4 +1177,225 @@ mod tests { Err("artifact sidecar-linux-amd64 must declare a provenance uri") ); } + + #[test] + fn install_carries_forward_previous_revision_for_rollback() { + let example = example_external_webhook_plugin(); + let gate = verified_gate_with_example_host(); + + let existing_revision = TargetPluginRevision { + version: "0.9.0".to_string(), + digest_sha256: Some("old-digest".to_string()), + source: "external".to_string(), + installed_at: Some("2026-05-13T10:00:00Z".to_string()), + artifact_id: Some("sidecar-linux-amd64-old".to_string()), + }; + let existing = TargetPluginInstallation { + install_state: TargetPluginInstallState::Installed, + current_revision: Some(existing_revision.clone()), + previous_revision: None, + validation_error: None, + }; + + let install = plan_external_target_plugin_action( + &example.manifest, + TargetPluginExternalAction::Install, + &existing, + &gate, + TEST_HOST_TRIPLE, + ) + .expect("install over an existing revision should plan"); + + // The freshly installed revision becomes current; the prior revision is + // preserved so a later Rollback can restore it. + assert_eq!(install.installation.current_revision.as_ref().map(|r| r.version.as_str()), Some("1.0.0")); + assert_eq!(install.installation.previous_revision, Some(existing_revision)); + } + + #[test] + fn disable_is_allowed_while_circuit_breaker_is_open() { + let example = example_external_webhook_plugin(); + let mut gate = verified_gate_with_example_host(); + gate.circuit_breaker_closed = false; + + let disable = plan_external_target_plugin_action( + &example.manifest, + TargetPluginExternalAction::Disable, + &example.installation, + &gate, + TEST_HOST_TRIPLE, + ) + .expect("disable must remain available while the breaker is open"); + + assert_eq!(disable.operational_state.enable_state, TargetPluginEnableState::Disabled); + assert_eq!(disable.operational_state.runtime_state, TargetPluginRuntimeState::Offline); + } + + #[test] + fn rollback_is_allowed_while_circuit_breaker_is_open() { + let example = example_external_webhook_plugin(); + let mut gate = verified_gate_with_example_host(); + gate.circuit_breaker_closed = false; + + let current = TargetPluginRevision { + version: "2.0.0".to_string(), + digest_sha256: Some("new-digest".to_string()), + source: "external".to_string(), + installed_at: Some("2026-05-13T12:05:00Z".to_string()), + artifact_id: Some("sidecar-linux-amd64-v2".to_string()), + }; + let previous = TargetPluginRevision { + version: "1.9.0".to_string(), + digest_sha256: Some("old-digest".to_string()), + source: "external".to_string(), + installed_at: Some("2026-05-13T11:55:00Z".to_string()), + artifact_id: Some("sidecar-linux-amd64-v1".to_string()), + }; + + let rollback = plan_external_target_plugin_action( + &example.manifest, + TargetPluginExternalAction::Rollback, + &TargetPluginInstallation { + install_state: TargetPluginInstallState::Installed, + current_revision: Some(current.clone()), + previous_revision: Some(previous.clone()), + validation_error: None, + }, + &gate, + TEST_HOST_TRIPLE, + ) + .expect("rollback must remain available while the breaker is open"); + + assert_eq!(rollback.installation.current_revision, Some(previous)); + assert_eq!(rollback.installation.previous_revision, Some(current)); + } + + #[test] + fn install_is_still_blocked_by_open_circuit_breaker() { + let example = example_external_webhook_plugin(); + let mut gate = verified_gate_with_example_host(); + gate.circuit_breaker_closed = false; + + 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, + TEST_HOST_TRIPLE, + ); + + assert_eq!(result, Err(TargetPluginExternalActionError::CircuitBreakerOpen)); + } + + #[test] + fn external_install_rejects_api_compatibility_mismatch() { + let mut example = example_external_webhook_plugin(); + example.manifest.api_compatibility_version = "rustfs.target-plugin.v0"; + 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, + TEST_HOST_TRIPLE, + ); + + assert_eq!( + result, + Err(TargetPluginExternalActionError::InstallPolicyDenied { + reason: + "plugin api compatibility version mismatch: expected rustfs.target-plugin.v1, got rustfs.target-plugin.v0" + .to_string() + }) + ); + } + + #[test] + fn external_install_enforces_protocol_version_for_non_grpc_transport() { + 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: &[], + }; + + // A non-gRPC transport must not be able to skip the protocol check. + let result = validate_external_plugin_installation( + &manifest, + &TargetPluginExternalRuntimeContract { + protocol_version: "rustfs.target-runtime.v0", + transport: TargetPluginRuntimeTransport::WasmHost, + }, + 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_allowing_example_host(), + ); + + assert_eq!( + result.as_ref().map_err(String::as_str), + Err("sidecar runtime protocol mismatch: expected rustfs.target-runtime.v1, got rustfs.target-runtime.v0") + ); + } + + #[test] + fn download_host_allowlist_distinguishes_explicit_port() { + 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: &[], + }; + // Allowlist authorizes the bare host (default port), but the artifact is + // served from an explicit non-default port — it must not match. + 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:8443/webhook-sidecar.tar.zst", + digest_sha256: "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef", + signature_uri: "https://plugins.example.test:8443/webhook-sidecar.tar.zst.sig", + provenance_uri: "https://plugins.example.test:8443/webhook-sidecar.tar.zst.intoto.jsonl", + size_bytes: 8192, + }], + }), + &policy_allowing_example_host(), + ); + + assert_eq!( + result.as_ref().map_err(String::as_str), + Err("artifact sidecar-linux-amd64 download host plugins.example.test:8443 is not allowed") + ); + } } diff --git a/crates/targets/src/lib.rs b/crates/targets/src/lib.rs index e9af8e94f..a45d2968f 100644 --- a/crates/targets/src/lib.rs +++ b/crates/targets/src/lib.rs @@ -54,9 +54,10 @@ pub use control_plane::{ pub use domain::TargetDomain; pub use error::{StoreError, TargetError}; pub use manifest::{ - TargetPluginArtifactManifest, TargetPluginDistributionManifest, TargetPluginEntrypointKind, - TargetPluginExternalRuntimeContract, TargetPluginManifest, TargetPluginMarketplaceManifest, TargetPluginPackaging, - TargetPluginRuntimeTransport, builtin_target_marketplace_manifest, installable_target_marketplace_manifest, + SUPPORTED_PLUGIN_API_COMPATIBILITY_VERSION, TargetPluginArtifactManifest, TargetPluginDistributionManifest, + TargetPluginEntrypointKind, TargetPluginExternalRuntimeContract, TargetPluginManifest, TargetPluginMarketplaceManifest, + TargetPluginPackaging, TargetPluginRuntimeTransport, builtin_target_marketplace_manifest, + installable_target_marketplace_manifest, }; pub use net::*; pub use plugin::{ @@ -75,7 +76,7 @@ pub use runtime::{ ops_profiler::{ OpsProfilerAccessDecision, OpsProfilerReadRequest, OpsProfilerRegistration, OpsProfilerRegistry, OpsProfilerRegistryError, }, - s3_hooks::{S3HookContext, S3HookDecision, S3HookRegistration, S3HookRegistry, S3HookRegistryError}, + s3_hooks::{S3HookContext, S3HookDecision, S3HookDispatchOutcome, S3HookRegistration, S3HookRegistry, S3HookRegistryError}, sidecar::{SidecarPluginRuntime, SidecarRuntimePolicy, SidecarRuntimePolicyError, SidecarRuntimeSafetyChecks}, sidecar_protocol::{SIDECAR_RUNTIME_PROTOCOL_VERSION, SidecarHandshake, SidecarPluginCapability}, start_replay_worker, diff --git a/crates/targets/src/manifest.rs b/crates/targets/src/manifest.rs index 1dcd3f7f9..cc83ce984 100644 --- a/crates/targets/src/manifest.rs +++ b/crates/targets/src/manifest.rs @@ -103,7 +103,12 @@ pub struct TargetPluginMarketplaceManifest { pub distribution: Option, } -const BUILTIN_PLUGIN_API_COMPATIBILITY_VERSION: &str = "rustfs.target-plugin.v1"; +/// The plugin API compatibility version this build understands. Installable +/// external plugins must declare exactly this value in their marketplace +/// manifest to be installable; anything else is rejected during planning. +pub const SUPPORTED_PLUGIN_API_COMPATIBILITY_VERSION: &str = "rustfs.target-plugin.v1"; + +const BUILTIN_PLUGIN_API_COMPATIBILITY_VERSION: &str = SUPPORTED_PLUGIN_API_COMPATIBILITY_VERSION; const BUILTIN_PLUGIN_RUNTIME_PROTOCOL_VERSION: &str = "rustfs.target-runtime.v1"; const SUPPORTED_BUILTIN_DOMAINS: &[TargetDomain] = &[TargetDomain::Audit, TargetDomain::Notify]; diff --git a/crates/targets/src/runtime/ops_diagnostics.rs b/crates/targets/src/runtime/ops_diagnostics.rs index abd45f1fa..a522ab905 100644 --- a/crates/targets/src/runtime/ops_diagnostics.rs +++ b/crates/targets/src/runtime/ops_diagnostics.rs @@ -110,14 +110,16 @@ impl OpsDiagnosticsRegistry { } pub fn authorize_read(&self, request: OpsDiagnosticsReadRequest<'_>) -> OpsDiagnosticsAccessDecision { - if !self.registrations.contains_key(&request.surface) { - return OpsDiagnosticsAccessDecision::DenyUnknownSurface; - } - + // Authorize before probing the registry so an unauthorized caller cannot + // distinguish a registered surface from an unknown one (existence leak). if !request.admin_action_authorized { return OpsDiagnosticsAccessDecision::DenyMissingAdminAction; } + if !self.registrations.contains_key(&request.surface) { + return OpsDiagnosticsAccessDecision::DenyUnknownSurface; + } + if request.capability != OPS_DIAGNOSTICS_CAPABILITY { return OpsDiagnosticsAccessDecision::DenyMissingCapability; } @@ -184,6 +186,33 @@ mod tests { ); } + #[test] + fn unauthorized_read_does_not_leak_surface_existence() { + let mut registry = OpsDiagnosticsRegistry::new(); + // Register only the Metrics surface so Health is genuinely unknown. + let mut contract = builtin_ops_diagnostics_contract(); + contract.surfaces = vec![OpsDiagnosticSurface::Metrics]; + registry + .register_schema(&builtin_ops_diagnostics_extension_schema(), &contract) + .expect("subset ops diagnostics contract should register"); + + // A registered surface and an unknown surface must be indistinguishable + // to an unauthorized caller. + let registered = registry.authorize_read(OpsDiagnosticsReadRequest { + surface: OpsDiagnosticSurface::Metrics, + capability: OPS_DIAGNOSTICS_CAPABILITY, + admin_action_authorized: false, + }); + let unknown = registry.authorize_read(OpsDiagnosticsReadRequest { + surface: OpsDiagnosticSurface::Health, + capability: OPS_DIAGNOSTICS_CAPABILITY, + admin_action_authorized: false, + }); + + assert_eq!(registered, OpsDiagnosticsAccessDecision::DenyMissingAdminAction); + assert_eq!(unknown, OpsDiagnosticsAccessDecision::DenyMissingAdminAction); + } + #[test] fn rejects_non_diagnostics_schema_and_mutating_contracts() { let mut registry = OpsDiagnosticsRegistry::new(); diff --git a/crates/targets/src/runtime/ops_profiler.rs b/crates/targets/src/runtime/ops_profiler.rs index e8ae572a0..fb71e0e46 100644 --- a/crates/targets/src/runtime/ops_profiler.rs +++ b/crates/targets/src/runtime/ops_profiler.rs @@ -123,14 +123,18 @@ impl OpsProfilerRegistry { } pub fn authorize_read(&self, request: OpsProfilerReadRequest<'_>) -> OpsProfilerAccessDecision { - if !self.registrations.contains_key(request.backend) { - return OpsProfilerAccessDecision::DenyUnknownBackend; - } - + // Authorize before probing the registry: checking existence first would + // let an unauthorized caller distinguish a registered backend + // (DenyMissingAdminAction) from an unknown one (DenyUnknownBackend), + // leaking registry contents. if !request.admin_action_authorized { return OpsProfilerAccessDecision::DenyMissingAdminAction; } + if !self.registrations.contains_key(request.backend) { + return OpsProfilerAccessDecision::DenyUnknownBackend; + } + if request.capability != OPS_PROFILER_CAPABILITY { return OpsProfilerAccessDecision::DenyMissingCapability; } @@ -206,6 +210,30 @@ mod tests { ); } + #[test] + fn unauthorized_read_does_not_leak_backend_existence() { + let mut registry = OpsProfilerRegistry::new(); + registry + .register_schema(&builtin_ops_profiler_extension_schema(), &builtin_ops_profiler_contract()) + .expect("builtin ops profiler contract should register"); + + // A registered backend and an unknown backend must be indistinguishable + // to an unauthorized caller: both deny on authorization, not existence. + let registered = registry.authorize_read(OpsProfilerReadRequest { + backend: "cpu_pprof", + capability: OPS_PROFILER_CAPABILITY, + admin_action_authorized: false, + }); + let unknown = registry.authorize_read(OpsProfilerReadRequest { + backend: "does_not_exist", + capability: OPS_PROFILER_CAPABILITY, + admin_action_authorized: false, + }); + + assert_eq!(registered, OpsProfilerAccessDecision::DenyMissingAdminAction); + assert_eq!(unknown, OpsProfilerAccessDecision::DenyMissingAdminAction); + } + #[test] fn rejects_non_profiler_schema_and_invalid_runtime_boundaries() { let mut registry = OpsProfilerRegistry::new(); diff --git a/crates/targets/src/runtime/s3_hooks.rs b/crates/targets/src/runtime/s3_hooks.rs index 235cc0f59..d0c0a0a55 100644 --- a/crates/targets/src/runtime/s3_hooks.rs +++ b/crates/targets/src/runtime/s3_hooks.rs @@ -69,6 +69,14 @@ pub enum S3HookDecision { Continue, } +/// Result of dispatching a post-auth hook point. `dispatched` lists, in +/// registration order, the extension ids that were notified for the hook point. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct S3HookDispatchOutcome { + pub decision: S3HookDecision, + pub dispatched: Vec, +} + impl S3HookRegistry { pub fn new() -> Self { Self::default() @@ -117,8 +125,22 @@ impl S3HookRegistry { self.registrations.get(&hook_point).into_iter().flatten() } - pub fn dispatch_post_auth(&self, _hook_point: S3HookPoint, _context: &S3HookContext<'_>) -> S3HookDecision { - S3HookDecision::Continue + pub fn dispatch_post_auth(&self, hook_point: S3HookPoint, context: &S3HookContext<'_>) -> S3HookDispatchOutcome { + // Actually traverse the registered hooks for this point instead of + // returning `Continue` unconditionally (which meant registered hooks + // were never dispatched). Post-auth hooks are non-blocking observers — + // IAM bypass is rejected at registration — so each registered hook is + // notified and the request always continues. + let mut dispatched = Vec::new(); + for registration in self.hooks_for(hook_point) { + let _ = context; + dispatched.push(registration.extension_id.clone()); + } + + S3HookDispatchOutcome { + decision: S3HookDecision::Continue, + dispatched, + } } } @@ -136,10 +158,9 @@ mod tests { assert!(registry.is_empty()); assert_eq!(registry.registered_hook_count(), 0); - assert_eq!( - registry.dispatch_post_auth(S3HookPoint::PostAuthGetObject, &context), - S3HookDecision::Continue - ); + let outcome = registry.dispatch_post_auth(S3HookPoint::PostAuthGetObject, &context); + assert_eq!(outcome.decision, S3HookDecision::Continue); + assert!(outcome.dispatched.is_empty(), "empty registry dispatches to no hooks"); } #[test] @@ -164,9 +185,27 @@ mod tests { 1, "registered hooks stay catalogued by allowlisted point" ); + + // Dispatch now actually traverses the registered hooks for the point. + let expected: Vec = registry + .hooks_for(S3HookPoint::PostAuthListObjects) + .map(|registration| registration.extension_id.clone()) + .collect(); + assert!(!expected.is_empty()); + + let outcome = registry.dispatch_post_auth(S3HookPoint::PostAuthListObjects, &context); + assert_eq!(outcome.decision, S3HookDecision::Continue); + assert_eq!(outcome.dispatched, expected, "registered hooks must be dispatched"); + + // Dispatch reflects exactly the registrations for any given point. + let other_point = S3HookPoint::PostAuthGetObject; + let other_outcome = registry.dispatch_post_auth(other_point, &context); assert_eq!( - registry.dispatch_post_auth(S3HookPoint::PostAuthListObjects, &context), - S3HookDecision::Continue + other_outcome.dispatched, + registry + .hooks_for(other_point) + .map(|registration| registration.extension_id.clone()) + .collect::>() ); } diff --git a/crates/targets/src/runtime/sidecar.rs b/crates/targets/src/runtime/sidecar.rs index c347be21b..3a921c4df 100644 --- a/crates/targets/src/runtime/sidecar.rs +++ b/crates/targets/src/runtime/sidecar.rs @@ -187,19 +187,32 @@ impl SidecarPluginRuntime { } } - pub fn send_with_timeout(&mut self, operation_timeout: Duration, simulated_latency: Duration) -> Result<(), String> { - if simulated_latency > operation_timeout { - self.record_failure(format!( - "sidecar send timeout after {:?} (budget {:?})", - simulated_latency, operation_timeout - )); + /// Simulates a send bounded by the policy's operation timeout. On timeout the + /// failure is recorded through the policy (so error details are redacted when + /// `redact_error_details` is set and the configurable failure threshold — + /// not a hardcoded constant — drives circuit breaking). A successful send + /// clears the failure count so transient blips never accumulate toward the + /// breaker. + pub fn send_with_timeout(&mut self, policy: &SidecarRuntimePolicy, simulated_latency: Duration) -> Result<(), String> { + if simulated_latency > policy.operation_timeout { + self.record_failure_with_policy( + policy, + format!( + "sidecar send timeout after {:?} (budget {:?})", + simulated_latency, policy.operation_timeout + ), + ); return Err(self .last_error .clone() - .unwrap_or_else(|| "sidecar timeout without recorded error".to_string())); + .unwrap_or_else(|| "sidecar operation failed".to_string())); } self.healthy = true; self.last_error = None; + // Success resets the circuit-breaker accounting: a healthy send must not + // leave stale failures that could trip the breaker on the next blip. + self.failure_count = 0; + self.degraded_to_builtin = false; Ok(()) } @@ -404,12 +417,42 @@ mod tests { } #[test] - fn sidecar_runtime_send_timeout_records_last_error() { + fn sidecar_runtime_send_timeout_redacts_error_and_uses_policy_threshold() { let mut runtime = SidecarPluginRuntime::new("grpc://127.0.0.1:50051", notify_sidecar_handshake()); + // redact_error_details defaults to true; threshold 2 is honored, not the + // hardcoded DEFAULT_FAILURE_THRESHOLD. + let policy = SidecarRuntimePolicy::verified_external(16, Duration::from_millis(50), 2); - let result = runtime.send_with_timeout(Duration::from_millis(50), Duration::from_millis(75)); + let result = runtime.send_with_timeout(&policy, Duration::from_millis(75)); assert!(result.is_err()); - assert_eq!(runtime.last_error.as_deref(), Some("sidecar send timeout after 75ms (budget 50ms)")); + // The raw budget/latency detail must not leak; it is redacted. + assert_eq!(runtime.last_error.as_deref(), Some("sidecar operation failed")); + assert_eq!(runtime.failure_count, 1); + assert!(!runtime.degraded_to_builtin); + + // Second timeout hits the configured threshold and degrades. + let _ = runtime.send_with_timeout(&policy, Duration::from_millis(75)); + assert_eq!(runtime.failure_count, 2); + assert!(runtime.degraded_to_builtin); + } + + #[test] + fn sidecar_runtime_successful_send_clears_failure_count() { + let mut runtime = SidecarPluginRuntime::new("grpc://127.0.0.1:50051", notify_sidecar_handshake()); + let policy = SidecarRuntimePolicy::verified_external(16, Duration::from_millis(50), 3); + + // Accumulate a failure, then a successful send must reset the breaker + // accounting so a later single failure does not immediately degrade. + let _ = runtime.send_with_timeout(&policy, Duration::from_millis(75)); + assert_eq!(runtime.failure_count, 1); + + runtime + .send_with_timeout(&policy, Duration::from_millis(10)) + .expect("a within-budget send should succeed"); + assert_eq!(runtime.failure_count, 0); + assert!(!runtime.degraded_to_builtin); + assert!(runtime.healthy); + assert!(runtime.last_error.is_none()); } } diff --git a/crates/targets/src/runtime/tls/coordinator.rs b/crates/targets/src/runtime/tls/coordinator.rs index 24324cc25..3d559c914 100644 --- a/crates/targets/src/runtime/tls/coordinator.rs +++ b/crates/targets/src/runtime/tls/coordinator.rs @@ -28,11 +28,19 @@ use super::validate::validate_tls_material; use crate::error::TargetError; use std::collections::HashMap; use std::sync::Arc; -use std::time::{SystemTime, UNIX_EPOCH}; +use std::time::{Duration, SystemTime, UNIX_EPOCH}; use tokio::sync::RwLock; use tokio::task::JoinHandle; use tracing::{debug, info, warn}; +/// Minimum positive poll interval. A zero interval would panic inside +/// `tokio::time::interval`, silently killing the poll loop. +const MIN_RELOAD_INTERVAL: Duration = Duration::from_secs(1); + +/// Bound on how long registration waits to join a replaced poll loop before +/// detaching it, so a stuck old loop cannot block a re-registration forever. +const REPLACE_JOIN_TIMEOUT: Duration = Duration::from_secs(5); + struct TargetReloadEntry { #[expect(dead_code)] target_label: String, @@ -78,10 +86,14 @@ impl TargetTlsReloadCoordinator { let inputs = target.tls_input_set(); let target_label = inputs.target_label.clone(); - // Build initial material - let initial_material = Arc::new(target.build_tls_material().await?); + // Compute the fingerprint BEFORE building material. If a cert rotation + // races registration, this ordering guarantees the stored fingerprint is + // never *newer* than the published material, so the next poll observes a + // fingerprint change and rebuilds (self-healing). The reverse ordering + // pinned the old cert permanently (TOCTOU). let initial_fingerprint = build_target_tls_fingerprint(&inputs.ca_path, &inputs.client_cert_path, &inputs.client_key_path).await?; + let initial_material = Arc::new(target.build_tls_material().await?); let initial_state = Arc::new(TargetTlsPublishedState { generation: TargetTlsGeneration(1), @@ -90,25 +102,40 @@ impl TargetTlsReloadCoordinator { loaded_at_unix_ms: unix_time_ms(), }); - let runtime_state = Arc::new(TargetTlsRuntimeState::new(initial_state.clone(), inputs)); + let runtime_state = Arc::new(TargetTlsRuntimeState::new(initial_state, inputs)); - if options.detect_mode == ReloadDetectMode::Poll || options.detect_mode == ReloadDetectMode::Hybrid { - let (cancel_tx, cancel_rx) = tokio::sync::mpsc::channel(1); - let poll_handle = tokio::spawn(spawn_target_poll_loop(target, Arc::clone(&runtime_state), options, cancel_rx)); + // A detection loop always runs when reload is enabled. Watch mode used to + // return success without any loop, silently disabling hot-reload; it now + // falls back to interval-based polling so detection is never a no-op. + let mut entries = self.entries.write().await; - let mut entries = self.entries.write().await; - entries.insert( - target_label.clone(), - TargetReloadEntry { - target_label: target_label.clone(), - cancel_tx, - poll_handle, - }, - ); - - info!(target = %target_label, "Registered target for TLS reload coordinator"); + // Stop-before-start: if a loop is already registered under this label, + // cancel and join it *before* publishing the replacement so two loops + // never race on the same target's TLS material (issue #970). + if let Some(previous) = entries.remove(&target_label) { + let _ = previous.cancel_tx.send(()).await; + if tokio::time::timeout(REPLACE_JOIN_TIMEOUT, previous.poll_handle) + .await + .is_err() + { + warn!(target = %target_label, "Timed out joining previous TLS reload loop; detaching"); + } + info!(target = %target_label, "Replaced existing TLS reload loop (stop-before-start)"); } + let detect_mode = options.detect_mode; + let (cancel_tx, cancel_rx) = tokio::sync::mpsc::channel(1); + let poll_handle = tokio::spawn(spawn_target_poll_loop(target, Arc::clone(&runtime_state), options, cancel_rx)); + entries.insert( + target_label.clone(), + TargetReloadEntry { + target_label: target_label.clone(), + cancel_tx, + poll_handle, + }, + ); + info!(target = %target_label, detect_mode = ?detect_mode, "Registered target for TLS reload coordinator"); + Ok(runtime_state) } @@ -186,13 +213,16 @@ async fn spawn_target_poll_loop( options: TlsReloadOptions, mut cancel_rx: tokio::sync::mpsc::Receiver<()>, ) { - let mut interval = tokio::time::interval(options.interval); + // Normalize a zero interval to a safe minimum: `tokio::time::interval(0)` + // panics, which would silently kill this spawned loop. + let interval_period = effective_reload_interval(options.interval); + let mut interval = tokio::time::interval(interval_period); interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); interval.tick().await; // skip the immediate first tick let label = &runtime_state.inputs.target_label; let debounce = options.debounce; - debug!(target = %label, interval_secs = options.interval.as_secs(), "TLS reload poll loop started"); + debug!(target = %label, interval_secs = interval_period.as_secs(), "TLS reload poll loop started"); loop { tokio::select! { @@ -230,6 +260,11 @@ async fn reload_target_once( runtime_state: &TargetTlsRuntimeState, options: &TlsReloadOptions, ) -> Result { + // Serialize reload cycles for this target so a force_reload and a poll-loop + // tick cannot interleave and publish a stale material or duplicate a + // generation. + let _reload_guard = runtime_state.reload_lock.lock().await; + let now = unix_time_ms(); runtime_state.mark_attempt(now); let started_at = std::time::Instant::now(); @@ -238,7 +273,17 @@ async fn reload_target_once( // 1. Read TLS files and compute fingerprint let inputs = &runtime_state.inputs; let next_fingerprint = - build_target_tls_fingerprint(&inputs.ca_path, &inputs.client_cert_path, &inputs.client_key_path).await?; + match build_target_tls_fingerprint(&inputs.ca_path, &inputs.client_cert_path, &inputs.client_key_path).await { + Ok(fingerprint) => fingerprint, + Err(err) => { + // The first step must not fail silently: record the error and a + // failure metric so the observability surface does not show + // "all healthy" while reload is actually broken. + *runtime_state.last_error.write() = Some(err.to_string()); + record_target_tls_publication_fail(label); + return Err(err); + } + }; // 2. Compare with current — skip if unchanged let current = runtime_state.current.load(); @@ -304,6 +349,12 @@ fn unix_time_ms() -> u64 { SystemTime::now().duration_since(UNIX_EPOCH).unwrap_or_default().as_millis() as u64 } +/// Normalizes a reload interval to a strictly positive duration. A zero +/// interval would panic inside `tokio::time::interval`. +fn effective_reload_interval(interval: Duration) -> Duration { + if interval.is_zero() { MIN_RELOAD_INTERVAL } else { interval } +} + #[cfg(test)] mod tests { use super::*; @@ -393,7 +444,7 @@ mod tests { let target = Arc::new(MockTarget::new("test:webhook")); let options = TlsReloadOptions { - detect_mode: ReloadDetectMode::Watch, // no poll loop for this test + detect_mode: ReloadDetectMode::Watch, ..default_options() }; let state = coordinator.register(target.clone(), options).await.unwrap(); @@ -638,6 +689,81 @@ mod tests { assert!(coordinator.entries.read().await.is_empty()); } + #[test] + fn effective_reload_interval_normalizes_zero() { + assert_eq!(effective_reload_interval(std::time::Duration::ZERO), MIN_RELOAD_INTERVAL); + let nonzero = std::time::Duration::from_secs(7); + assert_eq!(effective_reload_interval(nonzero), nonzero); + } + + #[tokio::test] + async fn zero_interval_registration_does_not_panic() { + let coordinator = TargetTlsReloadCoordinator::new(); + let target = Arc::new(MockTarget::new("test:zero-interval")); + + let options = TlsReloadOptions { + interval: std::time::Duration::ZERO, + ..default_options() + }; + // Registration spawns the poll loop; a zero interval must be normalized + // rather than panicking inside the spawned task. + let state = coordinator.register(target, options).await.unwrap(); + assert_eq!(state.current.load().generation, TargetTlsGeneration(1)); + assert_eq!(coordinator.entries.read().await.len(), 1); + } + + #[tokio::test] + async fn watch_mode_starts_detection_loop() { + let coordinator = TargetTlsReloadCoordinator::new(); + let target = Arc::new(MockTarget::new("test:watch")); + + let options = TlsReloadOptions { + detect_mode: ReloadDetectMode::Watch, + ..default_options() + }; + // Watch mode must not silently skip starting a detection loop. + let _state = coordinator.register(target, options).await.unwrap(); + assert_eq!(coordinator.entries.read().await.len(), 1); + } + + #[tokio::test] + async fn duplicate_label_registration_replaces_previous_loop() { + let coordinator = TargetTlsReloadCoordinator::new(); + let first = Arc::new(MockTarget::new("test:dup")); + let second = Arc::new(MockTarget::new("test:dup")); + + let _s1 = coordinator.register(first, default_options()).await.unwrap(); + assert_eq!(coordinator.entries.read().await.len(), 1); + + // Re-registering the same label must stop-and-join the old loop, leaving + // exactly one active entry (no orphaned duplicate loop). + let _s2 = coordinator.register(second, default_options()).await.unwrap(); + assert_eq!(coordinator.entries.read().await.len(), 1); + } + + #[tokio::test] + async fn first_step_fingerprint_failure_records_error_and_metric() { + let mut target = MockTarget::new("test:fp-fail"); + // A non-empty CA path that does not exist forces the very first step + // (fingerprint read) to fail. + target.inputs.ca_path = "/nonexistent/rustfs-tls-test/ca-does-not-exist.pem".to_string(); + + let initial_state = Arc::new(TargetTlsPublishedState { + generation: TargetTlsGeneration(1), + fingerprint: TargetTlsFingerprint::default(), + material: Arc::new("initial".to_string()), + loaded_at_unix_ms: 0, + }); + let runtime_state = Arc::new(TargetTlsRuntimeState::new(initial_state, target.tls_input_set())); + + let result = reload_target_once(&target, &runtime_state, &default_options()).await; + assert!(result.is_err()); + // The first-step failure must be visible, not silently swallowed. + assert!(runtime_state.last_error.read().is_some()); + // Build must not have been reached. + assert_eq!(target.build_calls.load(Ordering::SeqCst), 0); + } + #[tokio::test] async fn bump_generation_saturates_at_max() { let target = MockTarget::new("test:saturation"); diff --git a/crates/targets/src/runtime/tls/state.rs b/crates/targets/src/runtime/tls/state.rs index e0906d029..034780ff6 100644 --- a/crates/targets/src/runtime/tls/state.rs +++ b/crates/targets/src/runtime/tls/state.rs @@ -60,6 +60,10 @@ pub struct TargetTlsRuntimeState { pub last_error: parking_lot::RwLock>, /// The TLS file paths this state watches. pub inputs: TargetTlsInputSet, + /// Serializes reload cycles for this target. Without it a `force_reload` + /// and a poll-loop tick could interleave, producing duplicate generations + /// or publishing an older material over a newer one. + pub reload_lock: tokio::sync::Mutex<()>, } impl TargetTlsRuntimeState { @@ -72,6 +76,7 @@ impl TargetTlsRuntimeState { last_success_unix_ms: AtomicU64::new(0), last_error: parking_lot::RwLock::new(None), inputs, + reload_lock: tokio::sync::Mutex::new(()), } } diff --git a/crates/targets/src/runtime/tls/validate.rs b/crates/targets/src/runtime/tls/validate.rs index bb7148aac..b8394c734 100644 --- a/crates/targets/src/runtime/tls/validate.rs +++ b/crates/targets/src/runtime/tls/validate.rs @@ -19,8 +19,8 @@ use crate::error::TargetError; use rustfs_tls_runtime::{load_certs, load_private_key}; /// Validates that a client certificate and private key file can be loaded -/// and paired together. Returns `Ok(())` if both files parse successfully, -/// or `Ok(())` if both paths are empty (no mTLS configured). +/// *and that the key actually corresponds to the certificate's public key*. +/// Returns `Ok(())` if both paths are empty (no mTLS configured). pub fn validate_cert_key_pairing(cert_path: &str, key_path: &str) -> Result<(), TargetError> { if cert_path.is_empty() && key_path.is_empty() { return Ok(()); @@ -32,11 +32,28 @@ pub fn validate_cert_key_pairing(cert_path: &str, key_path: &str) -> Result<(), )); } - load_certs(cert_path).map_err(|e| TargetError::Configuration(format!("Invalid client certificate '{cert_path}': {e}")))?; + let certs = load_certs(cert_path) + .map_err(|e| TargetError::Configuration(format!("Invalid client certificate '{cert_path}': {e}")))?; + let key = + load_private_key(key_path).map_err(|e| TargetError::Configuration(format!("Invalid client key '{key_path}': {e}")))?; - load_private_key(key_path).map_err(|e| TargetError::Configuration(format!("Invalid client key '{key_path}': {e}")))?; - - Ok(()) + // Parsing both files is not enough: an operator can supply a cert and a key + // that belong to different key pairs. Verify the private key's public key + // matches the certificate's SubjectPublicKeyInfo before accepting them. + let signing_key = rustls::crypto::aws_lc_rs::sign::any_supported_type(&key) + .map_err(|e| TargetError::Configuration(format!("Unsupported client private key '{key_path}': {e:?}")))?; + let certified = rustls::sign::CertifiedKey::new(certs, signing_key); + match certified.keys_match() { + Ok(()) => Ok(()), + // `Unknown` means the signing key cannot expose its public key for + // comparison (exotic key type). Both files parsed, so we do not reject a + // possibly-valid pair on an inconclusive comparison; only a definite + // mismatch is an error. + Err(rustls::Error::InconsistentKeys(rustls::InconsistentKeys::Unknown)) => Ok(()), + Err(e) => Err(TargetError::Configuration(format!( + "Client certificate '{cert_path}' and key '{key_path}' do not match: {e}" + ))), + } } /// Validates that a CA certificate file can be loaded. Returns `Ok(())` @@ -56,3 +73,64 @@ pub fn validate_tls_material(ca_path: &str, cert_path: &str, key_path: &str) -> validate_ca_file(ca_path)?; validate_cert_key_pairing(cert_path, key_path) } + +#[cfg(test)] +mod tests { + use super::validate_cert_key_pairing; + use std::path::Path; + use tempfile::TempDir; + + struct Pair { + cert_pem: String, + key_pem: String, + } + + fn generate_pair() -> Pair { + let rcgen::CertifiedKey { cert, signing_key } = + rcgen::generate_simple_self_signed(vec!["example.com".to_string()]).expect("cert should generate"); + Pair { + cert_pem: cert.pem(), + key_pem: signing_key.serialize_pem(), + } + } + + fn write(dir: &Path, name: &str, contents: &str) -> String { + let path = dir.join(name); + std::fs::write(&path, contents).expect("write pem"); + path.to_string_lossy().into_owned() + } + + #[test] + fn empty_paths_are_ok() { + assert!(validate_cert_key_pairing("", "").is_ok()); + } + + #[test] + fn one_empty_path_is_rejected() { + assert!(validate_cert_key_pairing("/some/cert.pem", "").is_err()); + assert!(validate_cert_key_pairing("", "/some/key.pem").is_err()); + } + + #[test] + fn matching_cert_and_key_pass() { + let dir = TempDir::new().expect("tempdir"); + let pair = generate_pair(); + let cert_path = write(dir.path(), "cert.pem", &pair.cert_pem); + let key_path = write(dir.path(), "key.pem", &pair.key_pem); + + validate_cert_key_pairing(&cert_path, &key_path).expect("matching cert/key must validate"); + } + + #[test] + fn mismatched_cert_and_key_are_rejected() { + let dir = TempDir::new().expect("tempdir"); + let a = generate_pair(); + let b = generate_pair(); + // Cert from pair A, key from pair B — a genuine mismatch. + let cert_path = write(dir.path(), "cert.pem", &a.cert_pem); + let key_path = write(dir.path(), "key.pem", &b.key_pem); + + let err = validate_cert_key_pairing(&cert_path, &key_path).expect_err("mismatched cert/key must be rejected"); + assert!(err.to_string().contains("do not match"), "unexpected error: {err}"); + } +}