From df6563e5c063cf7461aa71e0b180b7e99c76a20c Mon Sep 17 00:00:00 2001 From: Chris Date: Mon, 14 Sep 2026 03:17:57 +0800 Subject: [PATCH] feat(connect): add service license relay import (#7784) --- rustfs/src/config/cli.rs | 50 ++++ rustfs/src/config/mod.rs | 5 +- rustfs/src/connect/license.rs | 37 ++- rustfs/src/connect/license_relay.rs | 377 ++++++++++++++++++++++++++++ rustfs/src/connect/mod.rs | 11 +- rustfs/src/connect/relay.rs | 113 ++++++++- rustfs/src/startup_entrypoint.rs | 42 +++- rustfs/tests/connect_license.rs | 269 +++++++++++++++++++- 8 files changed, 887 insertions(+), 17 deletions(-) create mode 100644 rustfs/src/connect/license_relay.rs diff --git a/rustfs/src/config/cli.rs b/rustfs/src/config/cli.rs index dd195740d..f0c1fce8f 100644 --- a/rustfs/src/config/cli.rs +++ b/rustfs/src/config/cli.rs @@ -1051,6 +1051,10 @@ pub enum ConnectLicenseCommands { Show(ConnectLicenseScopeOpts), /// Check Connect and install an operator-approved replacement license Renew(ConnectLicenseRenewOpts), + /// Verify a license and write its reviewed relay envelope (Unix only) + RelayExport(ConnectLicenseRelayExportOpts), + /// Re-verify and install a reviewed relay envelope, then sign a destination receipt (Unix only) + RelayImport(ConnectLicenseRelayImportOpts), } /// Online renewal transport plus local trust and scope pins. @@ -1115,6 +1119,52 @@ pub struct ConnectLicenseArtifactOpts { pub scope: ConnectLicenseScopeOpts, } +/// A verified license artifact exported as an opaque relay envelope. +#[derive(Args, Clone)] +pub struct ConnectLicenseRelayExportOpts { + /// Downloaded signed license artifact to transfer unchanged + #[arg(long)] + pub artifact: PathBuf, + + /// New owner-only relay envelope file + #[arg(long)] + pub envelope: PathBuf, + + /// UUIDv7 identifying this exact relay attempt + #[arg(long = "transfer-uid", value_parser = NonEmptyStringValueParser::new())] + pub transfer_uid: String, + + #[command(flatten)] + pub scope: ConnectLicenseScopeOpts, + + /// Confirm the verified issuer, destination, digest, expiry, and license scope were reviewed + #[arg(long = "acknowledge-reviewed", required = true, action = clap::ArgAction::SetTrue)] + pub acknowledge_reviewed: bool, +} + +/// A cluster-side relay import with a pre-bound destination receipt key. +#[derive(Args, Clone)] +pub struct ConnectLicenseRelayImportOpts { + /// Owner-only service-license relay envelope + #[arg(long)] + pub envelope: PathBuf, + + /// Owner-only Ed25519 seed for the pre-bound cluster receipt key + #[arg(long = "receipt-signing-key-file")] + pub receipt_signing_key_file: PathBuf, + + /// SHA-256 key ID of the pre-bound cluster receipt key + #[arg(long = "receipt-key-id", value_parser = NonEmptyStringValueParser::new())] + pub receipt_key_id: String, + + #[command(flatten)] + pub scope: ConnectLicenseScopeOpts, + + /// Confirm the verified issuer, destination, digest, expiry, and license scope were reviewed + #[arg(long = "acknowledge-reviewed", required = true, action = clap::ArgAction::SetTrue)] + pub acknowledge_reviewed: bool, +} + /// Offline inspection subcommand options #[derive(Args, Clone)] pub struct InspectOpts { diff --git a/rustfs/src/config/mod.rs b/rustfs/src/config/mod.rs index 8f9ee3735..61c84bcbf 100644 --- a/rustfs/src/config/mod.rs +++ b/rustfs/src/config/mod.rs @@ -56,7 +56,10 @@ pub use cli::{ ConnectClientPerformanceOperation, ConnectClientPerformanceOpts, ConnectDrivePerformanceOpts, ConnectPerformanceCommands, }; pub use cli::{ConnectEnvironmentInventoryOpts, ConnectInventoryCommands}; -pub use cli::{ConnectLicenseArtifactOpts, ConnectLicenseCommands, ConnectLicenseRenewOpts, ConnectLicenseScopeOpts}; +pub use cli::{ + ConnectLicenseArtifactOpts, ConnectLicenseCommands, ConnectLicenseRelayExportOpts, ConnectLicenseRelayImportOpts, + ConnectLicenseRenewOpts, ConnectLicenseScopeOpts, +}; pub use cli::{ConnectLogsMode, ConnectLogsOpts}; pub use cli::{ConnectObjectPerformanceOperation, ConnectObjectPerformanceOpts}; pub use cli::{ConnectProfileOpts, ConnectProfileTool, ConnectThreadProfileScope}; diff --git a/rustfs/src/connect/license.rs b/rustfs/src/connect/license.rs index 7d994210f..a8c1df4cf 100644 --- a/rustfs/src/connect/license.rs +++ b/rustfs/src/connect/license.rs @@ -278,10 +278,19 @@ pub fn apply_license_artifact( artifact_path: &Path, state_directory: &Path, context: &LicenseVerificationContext, +) -> Result { + let artifact_bytes = read_license_artifact_file(artifact_path)?; + apply_license_bytes(&artifact_bytes, state_directory, context) +} + +pub(super) fn apply_license_bytes( + artifact_bytes: &[u8], + state_directory: &Path, + context: &LicenseVerificationContext, ) -> Result { fs::create_dir_all(state_directory).map_err(|source| state_io(state_directory, source))?; let _lock = lock_state(state_directory)?; - let candidate = validate_artifact(read_artifact(artifact_path)?, context, true)?; + let candidate = validate_artifact(parse_artifact(artifact_bytes)?, context, true)?; let state_path = state_path(state_directory, context); if let Some(current) = load_installed(&state_path, context)? { @@ -300,7 +309,16 @@ pub fn verify_license_artifact( state_directory: &Path, context: &LicenseVerificationContext, ) -> Result { - let candidate = validate_artifact(read_artifact(artifact_path)?, context, true)?; + let artifact_bytes = read_license_artifact_file(artifact_path)?; + verify_license_bytes(&artifact_bytes, state_directory, context) +} + +pub(super) fn verify_license_bytes( + artifact_bytes: &[u8], + state_directory: &Path, + context: &LicenseVerificationContext, +) -> Result { + let candidate = validate_artifact(parse_artifact(artifact_bytes)?, context, true)?; let state_path = state_path(state_directory, context); if let Some(current) = load_installed(&state_path, context)? && matches!(compare_sequence(&candidate, ¤t)?, SequenceDecision::Idempotent) @@ -468,9 +486,18 @@ fn parse_timestamp(value: &str) -> Result { .map_err(|_| failure(LicenseArtifactStatus::InvalidArtifact, "the license timestamp is invalid")) } -fn read_artifact(path: &Path) -> Result { - let bytes = read_bounded_regular_file(path, MAX_ARTIFACT_BYTES, LicenseArtifactStatus::InvalidArtifact, "license artifact")?; - serde_json::from_slice(&bytes).map_err(|_| failure(LicenseArtifactStatus::InvalidArtifact, "the license artifact is invalid")) +pub(super) fn read_license_artifact_file(path: &Path) -> Result, LicenseArtifactError> { + read_bounded_regular_file(path, MAX_ARTIFACT_BYTES, LicenseArtifactStatus::InvalidArtifact, "license artifact") +} + +fn parse_artifact(bytes: &[u8]) -> Result { + if bytes.is_empty() || bytes.len() as u64 > MAX_ARTIFACT_BYTES { + return Err(failure( + LicenseArtifactStatus::InvalidArtifact, + format!("the license artifact must be no larger than {MAX_ARTIFACT_BYTES} bytes"), + )); + } + serde_json::from_slice(bytes).map_err(|_| failure(LicenseArtifactStatus::InvalidArtifact, "the license artifact is invalid")) } fn read_bounded_regular_file( diff --git a/rustfs/src/connect/license_relay.rs b/rustfs/src/connect/license_relay.rs new file mode 100644 index 000000000..05f85a27d --- /dev/null +++ b/rustfs/src/connect/license_relay.rs @@ -0,0 +1,377 @@ +// Copyright 2024 RustFS Team +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//! Offline relay export and destination-side receipt for Connect service licenses. + +use std::collections::BTreeMap; +use std::fs; +use std::io::{self, Write as _}; +use std::path::{Path, PathBuf}; +use std::sync::atomic::{AtomicU64, Ordering}; + +use serde::{Deserialize, Serialize}; +use sha2::{Digest as _, Sha256}; +use time::OffsetDateTime; +use time::format_description::well_known::Rfc3339; + +use super::license::{ + LicenseArtifactError, LicenseClaims, LicenseReport, LicenseVerificationContext, apply_license_bytes, + read_license_artifact_file, verify_license_bytes, +}; +use super::relay::{ + DestinationReceiptSigner, RelayDirection, RelayEnvelope, RelayError, RelayMaterialKind, RelayParty, RelayReceiptOutcome, + RelayReview, decode_relay_envelope, prepare_approved_artifact, read_protected_relay_artifact, +}; + +const LEDGER_SCHEMA: &str = "rustfs.connect.serviceLicenseRelayLedger/1"; +const LEDGER_FILE: &str = "service-license-relay-ledger.json"; +const LOCK_FILE: &str = ".service-license-relay.lock"; +const MAX_LEDGER_BYTES: u64 = 4 * 1024 * 1024; +const MAX_LEDGER_ENTRIES: usize = 4096; + +#[cfg(unix)] +const STATE_FILE_MODE: u32 = 0o600; + +static STAGING_SEQUENCE: AtomicU64 = AtomicU64::new(0); + +#[derive(Debug, thiserror::Error)] +pub enum ServiceLicenseRelayError { + #[error(transparent)] + Relay(#[from] RelayError), + #[error(transparent)] + License(#[from] LicenseArtifactError), + #[error("the relay producer does not match the verified license issuer")] + ProducerMismatch, + #[error("the relay destination does not match the verified license deployment")] + DestinationMismatch, + #[error("the relay transfer identifier is already bound to different material")] + TransferConflict, + #[error("the service-license relay state is unavailable")] + StateUnavailable, +} + +#[derive(Debug, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct ServiceLicenseRelayExport { + pub review: RelayReview, + pub license: LicenseClaims, +} + +#[derive(Debug)] +pub struct ServiceLicenseRelayReceipt { + pub receipt_bytes: Vec, + pub outcome: RelayReceiptOutcome, + pub received_at: String, + pub license: LicenseClaims, +} + +#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)] +#[serde(deny_unknown_fields)] +#[serde(rename_all = "camelCase")] +struct TransferBinding { + material_kind: RelayMaterialKind, + direction: RelayDirection, + artifact_sha256: String, + producer: RelayParty, + destination: RelayParty, +} + +#[derive(Clone, Debug, Deserialize, Serialize)] +#[serde(deny_unknown_fields)] +#[serde(rename_all = "camelCase")] +struct ReplayRecord { + received_at: String, +} + +#[derive(Debug, Deserialize, Serialize)] +#[serde(deny_unknown_fields)] +#[serde(rename_all = "camelCase")] +struct RelayLedger { + schema: String, + transfers: BTreeMap, + replays: BTreeMap, +} + +impl Default for RelayLedger { + fn default() -> Self { + Self { + schema: LEDGER_SCHEMA.to_owned(), + transfers: BTreeMap::new(), + replays: BTreeMap::new(), + } + } +} + +pub fn export_service_license_relay( + artifact_path: &Path, + envelope_path: &Path, + transfer_uid: &str, + state_directory: &Path, + context: &LicenseVerificationContext, + confirmed: bool, +) -> Result { + let artifact_bytes = read_license_artifact_file(artifact_path)?; + let report = verify_license_bytes(&artifact_bytes, state_directory, context)?; + let license = verified_license(report)?; + let producer = producer(&license); + let destination = destination(&license); + let prepared = prepare_approved_artifact( + transfer_uid, + RelayMaterialKind::ServiceLicense, + &artifact_bytes, + producer, + destination, + |_| confirmed, + )?; + let envelope_bytes = serde_json::to_vec(&prepared.envelope).map_err(|_| RelayError::EnvelopeEncoding)?; + write_new_file(envelope_path, &envelope_bytes)?; + Ok(ServiceLicenseRelayExport { + review: prepared.review, + license, + }) +} + +pub fn receive_service_license_relay( + envelope_path: &Path, + state_directory: &Path, + context: &LicenseVerificationContext, + receipt_signer: &DestinationReceiptSigner, + confirmed: bool, +) -> Result { + if !confirmed { + return Err(RelayError::ApprovalRequired.into()); + } + let envelope_bytes = read_protected_relay_artifact(envelope_path)?; + let (envelope, artifact_bytes) = decode_relay_envelope(&envelope_bytes)?; + if envelope.material_kind != RelayMaterialKind::ServiceLicense || envelope.direction != RelayDirection::ConnectToCluster { + return Err(RelayError::UnsupportedMaterial.into()); + } + + let report = verify_license_bytes(&artifact_bytes, state_directory, context)?; + let license = verified_license(report)?; + let verified_producer = producer(&license); + let verified_destination = destination(&license); + if envelope.asserted_producer != verified_producer { + return Err(ServiceLicenseRelayError::ProducerMismatch); + } + if envelope.destination != verified_destination { + return Err(ServiceLicenseRelayError::DestinationMismatch); + } + + fs::create_dir_all(state_directory).map_err(|_| ServiceLicenseRelayError::StateUnavailable)?; + let _lock = lock_state(state_directory)?; + let ledger_path = state_directory.join(LEDGER_FILE); + let mut ledger = load_ledger(&ledger_path)?; + let binding = TransferBinding { + material_kind: envelope.material_kind, + direction: envelope.direction, + artifact_sha256: envelope.artifact.sha256.clone(), + producer: verified_producer.clone(), + destination: verified_destination, + }; + if ledger + .transfers + .get(&envelope.transfer_uid) + .is_some_and(|existing| existing != &binding) + { + return Err(ServiceLicenseRelayError::TransferConflict); + } + if !ledger.transfers.contains_key(&envelope.transfer_uid) { + if ledger.transfers.len() >= MAX_LEDGER_ENTRIES { + return Err(ServiceLicenseRelayError::StateUnavailable); + } + ledger.transfers.insert(envelope.transfer_uid.clone(), binding); + persist_ledger(&ledger_path, &ledger)?; + } + + let replay_id = replay_id(&envelope); + let (outcome, received_at) = if let Some(existing) = ledger.replays.get(&replay_id) { + (RelayReceiptOutcome::Duplicate, existing.received_at.clone()) + } else { + if ledger.replays.len() >= MAX_LEDGER_ENTRIES { + return Err(ServiceLicenseRelayError::StateUnavailable); + } + let applied = apply_license_bytes(&artifact_bytes, state_directory, context)?; + let received_at = receipt_time(context.now_unix)?; + let outcome = if applied.idempotent { + RelayReceiptOutcome::Duplicate + } else { + RelayReceiptOutcome::Applied + }; + ledger.replays.insert( + replay_id, + ReplayRecord { + received_at: received_at.clone(), + }, + ); + persist_ledger(&ledger_path, &ledger)?; + (outcome, received_at) + }; + let receipt_bytes = receipt_signer.sign(&envelope, verified_producer, outcome, received_at.clone())?; + Ok(ServiceLicenseRelayReceipt { + receipt_bytes, + outcome, + received_at, + license, + }) +} + +fn verified_license(report: LicenseReport) -> Result { + report.license.ok_or(ServiceLicenseRelayError::StateUnavailable) +} + +fn producer(license: &LicenseClaims) -> RelayParty { + RelayParty { + party_type: "CONNECT_LICENSE_ISSUER".to_owned(), + name: license.issuer.clone(), + key_id: Some(license.key_id.clone()), + } +} + +fn destination(license: &LicenseClaims) -> RelayParty { + RelayParty { + party_type: "CLUSTER".to_owned(), + name: license.deployment.clone(), + key_id: None, + } +} + +fn replay_id(envelope: &RelayEnvelope) -> String { + let mut digest = Sha256::new(); + digest.update(b"SERVICE_LICENSE\0"); + digest.update(envelope.artifact.sha256.as_bytes()); + digest.update([0]); + digest.update(envelope.destination.party_type.as_bytes()); + digest.update([0]); + digest.update(envelope.destination.name.as_bytes()); + hex_simd::encode_to_string(digest.finalize(), hex_simd::AsciiCase::Lower) +} + +fn receipt_time(now_unix: i64) -> Result { + let time = OffsetDateTime::from_unix_timestamp(now_unix).map_err(|_| ServiceLicenseRelayError::StateUnavailable)?; + time.format(&Rfc3339).map_err(|_| ServiceLicenseRelayError::StateUnavailable) +} + +fn load_ledger(path: &Path) -> Result { + let bytes = match fs::read(path) { + Ok(bytes) => bytes, + Err(error) if error.kind() == io::ErrorKind::NotFound => return Ok(RelayLedger::default()), + Err(_) => return Err(ServiceLicenseRelayError::StateUnavailable), + }; + check_mode(path)?; + if bytes.len() as u64 > MAX_LEDGER_BYTES { + return Err(ServiceLicenseRelayError::StateUnavailable); + } + let ledger: RelayLedger = serde_json::from_slice(&bytes).map_err(|_| ServiceLicenseRelayError::StateUnavailable)?; + if ledger.schema != LEDGER_SCHEMA || ledger.transfers.len() > MAX_LEDGER_ENTRIES || ledger.replays.len() > MAX_LEDGER_ENTRIES + { + return Err(ServiceLicenseRelayError::StateUnavailable); + } + Ok(ledger) +} + +fn persist_ledger(path: &Path, ledger: &RelayLedger) -> Result<(), ServiceLicenseRelayError> { + let bytes = serde_json::to_vec(ledger).map_err(|_| ServiceLicenseRelayError::StateUnavailable)?; + if bytes.len() as u64 > MAX_LEDGER_BYTES { + return Err(ServiceLicenseRelayError::StateUnavailable); + } + let parent = path.parent().ok_or(ServiceLicenseRelayError::StateUnavailable)?; + let temporary = temporary_path(path); + let mut options = fs::OpenOptions::new(); + options.write(true).create_new(true); + #[cfg(unix)] + { + use std::os::unix::fs::OpenOptionsExt as _; + options.mode(STATE_FILE_MODE); + } + let mut file = options + .open(&temporary) + .map_err(|_| ServiceLicenseRelayError::StateUnavailable)?; + let result = file.write_all(&bytes).and_then(|()| file.sync_all()); + drop(file); + if result.is_err() || fs::rename(&temporary, path).is_err() || sync_directory(parent).is_err() { + let _ = fs::remove_file(&temporary); + return Err(ServiceLicenseRelayError::StateUnavailable); + } + Ok(()) +} + +fn write_new_file(path: &Path, bytes: &[u8]) -> Result<(), ServiceLicenseRelayError> { + let mut options = fs::OpenOptions::new(); + options.write(true).create_new(true); + #[cfg(unix)] + { + use std::os::unix::fs::OpenOptionsExt as _; + options.mode(STATE_FILE_MODE); + } + let mut file = options.open(path).map_err(|_| ServiceLicenseRelayError::StateUnavailable)?; + if file.write_all(bytes).and_then(|()| file.sync_all()).is_err() { + drop(file); + let _ = fs::remove_file(path); + return Err(ServiceLicenseRelayError::StateUnavailable); + } + Ok(()) +} + +fn lock_state(directory: &Path) -> Result { + let path = directory.join(LOCK_FILE); + let mut options = fs::OpenOptions::new(); + options.create(true).truncate(false).read(true).write(true); + #[cfg(unix)] + { + use std::os::unix::fs::OpenOptionsExt as _; + options.mode(STATE_FILE_MODE).custom_flags(libc::O_NOFOLLOW); + } + let file = options.open(path).map_err(|_| ServiceLicenseRelayError::StateUnavailable)?; + #[cfg(unix)] + { + use std::os::unix::fs::PermissionsExt as _; + file.set_permissions(fs::Permissions::from_mode(STATE_FILE_MODE)) + .map_err(|_| ServiceLicenseRelayError::StateUnavailable)?; + } + file.lock().map_err(|_| ServiceLicenseRelayError::StateUnavailable)?; + Ok(file) +} + +fn temporary_path(path: &Path) -> PathBuf { + let file_name = path.file_name().and_then(|name| name.to_str()).unwrap_or(LEDGER_FILE); + path.with_file_name(format!( + ".{file_name}.{}.{}.tmp", + std::process::id(), + STAGING_SEQUENCE.fetch_add(1, Ordering::Relaxed) + )) +} + +#[cfg(unix)] +fn check_mode(path: &Path) -> Result<(), ServiceLicenseRelayError> { + use std::os::unix::fs::PermissionsExt as _; + let metadata = fs::symlink_metadata(path).map_err(|_| ServiceLicenseRelayError::StateUnavailable)?; + if !metadata.is_file() || metadata.permissions().mode() & 0o7777 != STATE_FILE_MODE { + return Err(ServiceLicenseRelayError::StateUnavailable); + } + Ok(()) +} + +#[cfg(not(unix))] +fn check_mode(_path: &Path) -> Result<(), ServiceLicenseRelayError> { + Ok(()) +} + +fn sync_directory(directory: &Path) -> io::Result<()> { + #[cfg(unix)] + fs::File::open(directory)?.sync_all()?; + #[cfg(not(unix))] + let _ = directory; + Ok(()) +} diff --git a/rustfs/src/connect/mod.rs b/rustfs/src/connect/mod.rs index 3f3ac85a5..c8abdcdfe 100644 --- a/rustfs/src/connect/mod.rs +++ b/rustfs/src/connect/mod.rs @@ -35,6 +35,7 @@ pub mod identity; pub mod identity_store; pub mod inventory; pub mod license; +pub mod license_relay; pub mod license_renewal; pub mod offline; pub mod registration; @@ -128,14 +129,18 @@ pub use license::{ LICENSE_DOMAIN_SEPARATION_TAG, LicenseArtifactError, LicenseArtifactStatus, LicenseClaims, LicenseReport, LicenseVerificationContext, apply_license_artifact, inspect_installed_license, verify_license_artifact, }; +pub use license_relay::{ + ServiceLicenseRelayError, ServiceLicenseRelayExport, ServiceLicenseRelayReceipt, export_service_license_relay, + receive_service_license_relay, +}; pub use license_renewal::{LicenseRenewalClient, LicenseRenewalError, LicenseRenewalOutcome}; pub use offline::{EnrollmentError, OfflineEnrollment, OfflineKeyStore, VerifiedChallenge}; pub use registration::{RegistrationToken, TokenError}; pub use registration_bootstrap::{RegistrationBootstrapError, RegistrationBootstrapResult, register_from_protected_input}; pub use relay::{ - RelayDirection, RelayError, RelayHttpClient, RelayMaterialKind, RelayParty, RelayReceiptOutcome, RelayReceiptPayload, - RelayReview, TrustedReceiptSigner, prepare_approved_artifact, read_protected_relay_artifact, - read_protected_relay_authentication, + DestinationReceiptSigner, RelayDirection, RelayError, RelayHttpClient, RelayMaterialKind, RelayParty, RelayReceiptOutcome, + RelayReceiptPayload, RelayReview, TrustedReceiptSigner, decode_relay_envelope, prepare_approved_artifact, + read_protected_relay_artifact, read_protected_relay_authentication, }; pub use report_upload::{MAX_SUPPORT_BUNDLE_BYTES, ReportUploadClient, ReportUploadError, ReportUploadReceipt}; pub use runtime::{HeartbeatRuntime, InventoryRuntime, spawn_heartbeat_runtime, spawn_inventory_runtime}; diff --git a/rustfs/src/connect/relay.rs b/rustfs/src/connect/relay.rs index f39641b99..78c224a5e 100644 --- a/rustfs/src/connect/relay.rs +++ b/rustfs/src/connect/relay.rs @@ -18,7 +18,7 @@ //! accepts success only from a receipt signed by the preconfigured destination. use base64_simd::{STANDARD as BASE64_STANDARD, URL_SAFE_NO_PAD}; -use ed25519_dalek::{Signature, VerifyingKey}; +use ed25519_dalek::{Signature, Signer as _, SigningKey, VerifyingKey}; use reqwest::{Client, StatusCode, Url, header}; use rustls::pki_types::{CertificateDer, pem::PemObject as _}; use serde::{Deserialize, Serialize}; @@ -44,6 +44,8 @@ const MAX_RECEIPT_BYTES: usize = 64 * 1024; const RETRY_DELAY: Duration = Duration::from_millis(250); const MAX_AUTHENTICATION_BYTES: u64 = 8 * 1024; const MAX_PUBLIC_KEY_BYTES: u64 = 256; +const MAX_PRIVATE_KEY_BYTES: u64 = 256; +const MAX_RELAY_ENVELOPE_BYTES: usize = MAX_RELAY_ARTIFACT_BYTES * 2 + 64 * 1024; #[derive(Clone, Copy, Debug, Deserialize, Eq, Hash, PartialEq, Serialize)] #[serde(rename_all = "SCREAMING_SNAKE_CASE")] @@ -129,7 +131,7 @@ pub struct RelayReceiptPayload { pub received_at: String, } -#[derive(Debug, Deserialize)] +#[derive(Debug, Deserialize, Serialize)] #[serde(deny_unknown_fields)] #[serde(rename_all = "camelCase")] struct RelayReceiptSignature { @@ -138,7 +140,7 @@ struct RelayReceiptSignature { value: String, } -#[derive(Debug, Deserialize)] +#[derive(Debug, Deserialize, Serialize)] #[serde(deny_unknown_fields)] struct SignedRelayReceipt { payload: String, @@ -151,6 +153,69 @@ pub struct TrustedReceiptSigner { verifying_key: VerifyingKey, } +#[derive(Clone)] +pub struct DestinationReceiptSigner { + key_id: String, + signing_key: SigningKey, +} + +impl DestinationReceiptSigner { + pub fn new(key_id: String, seed: [u8; 32]) -> Result { + let signing_key = SigningKey::from_bytes(&seed); + let actual_key_id = hex_lower(&Sha256::digest(signing_key.verifying_key().as_bytes())); + if !is_sha256(&key_id) || key_id != actual_key_id { + return Err(RelayError::ReceiptTrustInvalid); + } + Ok(Self { key_id, signing_key }) + } + + pub fn from_private_key_file(path: &Path, key_id: String) -> Result { + let encoded = read_protected_bytes(path, MAX_PRIVATE_KEY_BYTES)?; + let encoded = std::str::from_utf8(&encoded) + .map_err(|_| RelayError::ReceiptTrustInvalid)? + .trim(); + let seed = decode_canonical_base64url(encoded).ok_or(RelayError::ReceiptTrustInvalid)?; + let seed: [u8; 32] = seed.try_into().map_err(|_| RelayError::ReceiptTrustInvalid)?; + Self::new(key_id, seed) + } + + pub(crate) fn sign( + &self, + envelope: &RelayEnvelope, + producer: RelayParty, + outcome: RelayReceiptOutcome, + received_at: String, + ) -> Result, RelayError> { + let payload = RelayReceiptPayload { + format_version: RELAY_RECEIPT_FORMAT.to_owned(), + protocol_version: envelope.protocol_version.clone(), + transfer_uid: envelope.transfer_uid.clone(), + material_kind: envelope.material_kind, + direction: envelope.direction, + artifact_sha256: envelope.artifact.sha256.clone(), + producer, + destination: envelope.destination.clone(), + outcome, + received_at, + }; + let payload = serde_json::to_vec(&payload).map_err(|_| RelayError::ReceiptEncoding)?; + let mut signed = Vec::with_capacity(RELAY_RECEIPT_DOMAIN_SEPARATION_TAG.len() + 1 + payload.len()); + signed.extend_from_slice(RELAY_RECEIPT_DOMAIN_SEPARATION_TAG.as_bytes()); + signed.push(0); + signed.extend_from_slice(&payload); + let signature = self.signing_key.sign(&signed).to_bytes(); + serde_json::to_vec(&SignedRelayReceipt { + payload: URL_SAFE_NO_PAD.encode_to_string(&payload), + signature: RelayReceiptSignature { + algorithm: "Ed25519".to_owned(), + key_id: self.key_id.clone(), + value: URL_SAFE_NO_PAD.encode_to_string(signature), + }, + }) + .map_err(|_| RelayError::ReceiptEncoding) + } +} + impl TrustedReceiptSigner { pub fn new(key_id: String, public_key: [u8; 32]) -> Result { if !is_sha256(&key_id) || hex_lower(&Sha256::digest(public_key)) != key_id { @@ -207,6 +272,10 @@ pub enum RelayError { ApprovalRequired, #[error("the relay envelope could not be encoded")] EnvelopeEncoding, + #[error("the relay envelope is malformed")] + EnvelopeInvalid, + #[error("the relay artifact digest does not match the exact decoded bytes")] + ArtifactDigestMismatch, #[error("the destination receipt trust configuration is invalid")] ReceiptTrustInvalid, #[error("the destination receipt is malformed")] @@ -217,6 +286,8 @@ pub enum RelayError { ReceiptSignatureInvalid, #[error("the destination receipt does not bind this transfer")] ReceiptMismatch, + #[error("the destination receipt could not be encoded")] + ReceiptEncoding, #[error("delivery produced no verified receipt after three attempts")] DeliveryUnknown, #[error("the relay control API rejected delivery with HTTP {0}")] @@ -451,6 +522,36 @@ where Ok(PreparedRelay { review, envelope }) } +pub fn decode_relay_envelope(envelope_bytes: &[u8]) -> Result<(RelayEnvelope, Vec), RelayError> { + if envelope_bytes.is_empty() || envelope_bytes.len() > MAX_RELAY_ENVELOPE_BYTES { + return Err(RelayError::EnvelopeInvalid); + } + let envelope: RelayEnvelope = serde_json::from_slice(envelope_bytes).map_err(|_| RelayError::EnvelopeInvalid)?; + if envelope.format_version != RELAY_ENVELOPE_FORMAT + || envelope.protocol_version != "v1" + || validate_transfer_uid(&envelope.transfer_uid).is_err() + || validate_route(envelope.material_kind, &envelope.asserted_producer, &envelope.destination).is_err() + || envelope.direction != direction_for(envelope.material_kind) + || envelope.artifact.encoding != "base64" + || !is_sha256(&envelope.artifact.sha256) + { + return Err(RelayError::EnvelopeInvalid); + } + let artifact_bytes = BASE64_STANDARD + .decode_to_vec(envelope.artifact.bytes.as_bytes()) + .map_err(|_| RelayError::EnvelopeInvalid)?; + if artifact_bytes.is_empty() + || artifact_bytes.len() > MAX_RELAY_ARTIFACT_BYTES + || BASE64_STANDARD.encode_to_string(&artifact_bytes) != envelope.artifact.bytes + { + return Err(RelayError::InvalidArtifact); + } + if hex_lower(&Sha256::digest(&artifact_bytes)) != envelope.artifact.sha256 { + return Err(RelayError::ArtifactDigestMismatch); + } + Ok((envelope, artifact_bytes)) +} + pub fn verify_receipt( receipt_bytes: &[u8], envelope: &RelayEnvelope, @@ -498,12 +599,16 @@ fn validate_route( } let valid = match material_kind { RelayMaterialKind::OfflineEnrollmentResponse | RelayMaterialKind::DiagnosticBundleManifest => { - asserted_producer.party_type == "DEVICE" && asserted_producer.key_id.is_some() && destination.party_type == "CONNECT" + asserted_producer.party_type == "DEVICE" + && asserted_producer.key_id.is_some() + && destination.party_type == "CONNECT" + && destination.key_id.is_none() } RelayMaterialKind::ServiceLicense => { asserted_producer.party_type == "CONNECT_LICENSE_ISSUER" && asserted_producer.key_id.is_some() && destination.party_type == "CLUSTER" + && destination.key_id.is_none() } }; valid.then_some(()).ok_or(RelayError::UnsupportedMaterial) diff --git a/rustfs/src/startup_entrypoint.rs b/rustfs/src/startup_entrypoint.rs index 4028f56bc..b8ffa6743 100644 --- a/rustfs/src/startup_entrypoint.rs +++ b/rustfs/src/startup_entrypoint.rs @@ -1361,10 +1361,11 @@ async fn execute_connect_license(command: ConnectLicenseCommands) -> Result<()> ConnectLicenseCommands::Import(options) | ConnectLicenseCommands::Verify(options) => &options.scope, ConnectLicenseCommands::Show(options) => options, ConnectLicenseCommands::Renew(options) => &options.scope, + ConnectLicenseCommands::RelayExport(options) => &options.scope, + ConnectLicenseCommands::RelayImport(options) => &options.scope, }; - let context = license_context(scope); if let ConnectLicenseCommands::Renew(options) = &command { - let context = context.map_err(Error::other)?; + let context = license_context(scope).map_err(Error::other)?; let root_ca_pem = std::fs::read(&options.ca_file).map_err(Error::other)?; let mut config = HeartbeatConfig::new( &options.endpoint, @@ -1392,12 +1393,49 @@ async fn execute_connect_license(command: ConnectLicenseCommands) -> Result<()> } return Ok(()); } + if let ConnectLicenseCommands::RelayExport(options) = &command { + let context = license_context(scope).map_err(Error::other)?; + let exported = crate::connect::export_service_license_relay( + &options.artifact, + &options.envelope, + &options.transfer_uid, + &scope.state_dir, + &context, + options.acknowledge_reviewed, + ) + .map_err(Error::other)?; + println!("{}", serde_json::to_string(&exported).map_err(Error::other)?); + return Ok(()); + } + if let ConnectLicenseCommands::RelayImport(options) = &command { + let context = license_context(scope).map_err(Error::other)?; + let signer = crate::connect::DestinationReceiptSigner::from_private_key_file( + &options.receipt_signing_key_file, + options.receipt_key_id.clone(), + ) + .map_err(Error::other)?; + let received = crate::connect::receive_service_license_relay( + &options.envelope, + &scope.state_dir, + &context, + &signer, + options.acknowledge_reviewed, + ) + .map_err(Error::other)?; + std::io::stdout().write_all(&received.receipt_bytes)?; + std::io::stdout().write_all(b"\n")?; + return Ok(()); + } + let context = license_context(scope); let report = match context { Ok(context) => match &command { ConnectLicenseCommands::Import(options) => apply_license_artifact(&options.artifact, &scope.state_dir, &context), ConnectLicenseCommands::Verify(options) => verify_license_artifact(&options.artifact, &scope.state_dir, &context), ConnectLicenseCommands::Show(_) => inspect_installed_license(&scope.state_dir, &context), ConnectLicenseCommands::Renew(_) => unreachable!("renewal is handled before local license commands"), + ConnectLicenseCommands::RelayExport(_) | ConnectLicenseCommands::RelayImport(_) => { + unreachable!("relay commands are handled before local license commands") + } } .unwrap_or_else(|error| { let installed = matches!(&command, ConnectLicenseCommands::Show(_)) diff --git a/rustfs/tests/connect_license.rs b/rustfs/tests/connect_license.rs index 76e1f097d..d5ad32bed 100644 --- a/rustfs/tests/connect_license.rs +++ b/rustfs/tests/connect_license.rs @@ -19,8 +19,10 @@ use std::process::Command; use base64_simd::URL_SAFE_NO_PAD; use ed25519_dalek::{Signer as _, SigningKey}; use rustfs::connect::{ - LICENSE_DOMAIN_SEPARATION_TAG, LicenseArtifactStatus, LicenseClaims, LicenseVerificationContext, apply_license_artifact, - inspect_installed_license, verify_license_artifact, + DestinationReceiptSigner, LICENSE_DOMAIN_SEPARATION_TAG, LicenseArtifactStatus, LicenseClaims, LicenseVerificationContext, + RelayMaterialKind, RelayReceiptOutcome, ServiceLicenseRelayError, TrustedReceiptSigner, apply_license_artifact, + decode_relay_envelope, export_service_license_relay, inspect_installed_license, receive_service_license_relay, + verify_license_artifact, }; use serde::Deserialize; use serde_json::{Value, json}; @@ -36,6 +38,7 @@ const OTHER_ORGANIZATION_DEPLOYMENT: &str = const ISSUER: &str = "test-connect-issuer"; const AUDIENCE: &str = "test-rustfs-cluster"; const SERVICE: &str = "SUPPORT"; +const RELAY_TRANSFER_UID: &str = "0198f3a1-a300-7c30-8c33-223344556677"; #[derive(Deserialize)] #[serde(rename_all = "camelCase")] @@ -313,6 +316,197 @@ fn rustfs_cli_runs_verify_import_and_show_with_structured_json() { assert_eq!(foreign_json["installed"], false); } +#[cfg(unix)] +#[test] +fn rustfs_cli_relay_preserves_signed_bytes_and_returns_a_bound_destination_receipt() { + use std::os::unix::fs::PermissionsExt as _; + + let license_key = SigningKey::from_bytes(&[42; 32]); + let license_public_key = license_key.verifying_key().to_bytes(); + let license_key_id = hex_simd::encode_to_string(Sha256::digest(license_public_key), hex_simd::AsciiCase::Lower); + let receipt_key = SigningKey::from_bytes(&[43; 32]); + let receipt_public_key = receipt_key.verifying_key().to_bytes(); + let receipt_key_id = hex_simd::encode_to_string(Sha256::digest(receipt_public_key), hex_simd::AsciiCase::Lower); + let temporary = TempDir::new().expect("create temp directory"); + let artifact_bytes = signed_artifact( + &license_key, + claims( + &license_key_id, + 2, + "018cc251-f400-7000-8000-000000000002", + "018cc251-f400-7000-8000-000000000001", + ), + ); + let artifact = write_artifact(temporary.path(), "license.json", &artifact_bytes); + let public_key = write_artifact(temporary.path(), "license.pub", &URL_SAFE_NO_PAD.encode_to_string(license_public_key)); + let receipt_seed = write_artifact( + temporary.path(), + "receipt.seed", + &URL_SAFE_NO_PAD.encode_to_string(receipt_key.to_bytes()), + ); + fs::set_permissions(&receipt_seed, fs::Permissions::from_mode(0o600)).expect("protect receipt seed"); + let envelope_path = temporary.path().join("service-license.relay.json"); + let source_state = temporary.path().join("source-state"); + let destination_state = temporary.path().join("destination-state"); + + let export = relay_export_cli(&artifact, &envelope_path, &public_key, &source_state, &license_key_id, true); + assert!( + export.status.success(), + "relay export failed: {}", + String::from_utf8_lossy(&export.stderr) + ); + let review: Value = serde_json::from_slice(&export.stdout).expect("relay review must be JSON"); + assert_eq!(review["review"]["materialKind"], "SERVICE_LICENSE"); + assert_eq!(review["review"]["destination"]["name"], DEPLOYMENT); + assert_eq!(review["license"]["expireTime"], "2099-01-01T00:00:00Z"); + assert_eq!(fs::metadata(&envelope_path).unwrap().permissions().mode() & 0o777, 0o600); + let envelope_bytes = fs::read(&envelope_path).expect("read relay envelope"); + let (envelope, relayed_artifact) = decode_relay_envelope(&envelope_bytes).expect("decode relay envelope"); + assert_eq!(relayed_artifact, artifact_bytes.as_bytes()); + assert_eq!(envelope.material_kind, RelayMaterialKind::ServiceLicense); + assert_eq!(envelope.asserted_producer.name, ISSUER); + + let rejected = relay_import_cli( + &envelope_path, + &receipt_seed, + &public_key, + &destination_state, + &license_key_id, + &receipt_key_id, + false, + ); + assert!(!rejected.status.success(), "relay import without review acknowledgement must fail"); + + let imported = relay_import_cli( + &envelope_path, + &receipt_seed, + &public_key, + &destination_state, + &license_key_id, + &receipt_key_id, + true, + ); + assert!( + imported.status.success(), + "relay import failed: {}", + String::from_utf8_lossy(&imported.stderr) + ); + let receipt_trust = TrustedReceiptSigner::new(receipt_key_id.clone(), receipt_public_key).expect("trust destination key"); + let receipt = rustfs::connect::relay::verify_receipt(&imported.stdout, &envelope, &receipt_trust) + .expect("destination receipt must bind the exact transfer"); + assert_eq!(receipt.outcome, RelayReceiptOutcome::Applied); + assert_eq!(receipt.transfer_uid, RELAY_TRANSFER_UID); + assert_eq!(Some(receipt.artifact_sha256.as_str()), review["review"]["artifactSha256"].as_str()); + assert_eq!(receipt.producer, envelope.asserted_producer); + assert_eq!(receipt.destination, envelope.destination); + + let repeated = relay_import_cli( + &envelope_path, + &receipt_seed, + &public_key, + &destination_state, + &license_key_id, + &receipt_key_id, + true, + ); + assert!( + repeated.status.success(), + "repeated relay import failed: {}", + String::from_utf8_lossy(&repeated.stderr) + ); + let duplicate = rustfs::connect::relay::verify_receipt(&repeated.stdout, &envelope, &receipt_trust) + .expect("duplicate receipt must remain valid"); + assert_eq!(duplicate.outcome, RelayReceiptOutcome::Duplicate); + assert_eq!(duplicate.received_at, receipt.received_at); + assert_eq!( + inspect_installed_license( + &destination_state, + &LicenseVerificationContext::new( + license_public_key, + license_key_id, + ISSUER.to_owned(), + AUDIENCE.to_owned(), + ORGANIZATION.to_owned(), + DEPLOYMENT.to_owned(), + SERVICE.to_owned(), + 1_800_000_000, + ) + .expect("test context"), + ) + .expect("relayed license must be installed") + .license + .map(|license| license.sequence), + Some(2) + ); +} + +#[cfg(unix)] +#[test] +fn service_license_relay_rejects_transfer_uid_reuse_before_replacing_the_license() { + let license_key = SigningKey::from_bytes(&[44; 32]); + let public_key = license_key.verifying_key().to_bytes(); + let key_id = hex_simd::encode_to_string(Sha256::digest(public_key), hex_simd::AsciiCase::Lower); + let context = LicenseVerificationContext::new( + public_key, + key_id.clone(), + ISSUER.to_owned(), + AUDIENCE.to_owned(), + ORGANIZATION.to_owned(), + DEPLOYMENT.to_owned(), + SERVICE.to_owned(), + 1_800_000_000, + ) + .expect("test context"); + let temporary = TempDir::new().expect("create temp directory"); + let source_state = temporary.path().join("source-state"); + let destination_state = temporary.path().join("destination-state"); + let first_artifact = write_artifact( + temporary.path(), + "first.json", + &signed_artifact( + &license_key, + claims(&key_id, 2, "018cc251-f400-7000-8000-000000000002", "018cc251-f400-7000-8000-000000000001"), + ), + ); + let replacement_artifact = write_artifact( + temporary.path(), + "replacement.json", + &signed_artifact( + &license_key, + claims(&key_id, 3, "018cc251-f400-7000-8000-000000000005", "018cc251-f400-7000-8000-000000000001"), + ), + ); + let first_envelope = temporary.path().join("first.relay.json"); + let replacement_envelope = temporary.path().join("replacement.relay.json"); + export_service_license_relay(&first_artifact, &first_envelope, RELAY_TRANSFER_UID, &source_state, &context, true) + .expect("export first license"); + export_service_license_relay( + &replacement_artifact, + &replacement_envelope, + RELAY_TRANSFER_UID, + &source_state, + &context, + true, + ) + .expect("export replacement license"); + let receipt_key = SigningKey::from_bytes(&[45; 32]); + let receipt_key_id = + hex_simd::encode_to_string(Sha256::digest(receipt_key.verifying_key().to_bytes()), hex_simd::AsciiCase::Lower); + let receipt_signer = DestinationReceiptSigner::new(receipt_key_id, receipt_key.to_bytes()).expect("destination signer"); + receive_service_license_relay(&first_envelope, &destination_state, &context, &receipt_signer, true) + .expect("install first license"); + let conflict = receive_service_license_relay(&replacement_envelope, &destination_state, &context, &receipt_signer, true) + .expect_err("one transfer UID cannot identify different bytes"); + assert!(matches!(conflict, ServiceLicenseRelayError::TransferConflict)); + assert_eq!( + inspect_installed_license(&destination_state, &context) + .expect("first license must remain installed") + .license + .map(|license| license.sequence), + Some(2) + ); +} + fn claims(key_id: &str, sequence: u64, license_uid: &str, grant_uid: &str) -> LicenseClaims { LicenseClaims { purpose: "RUSTFS_CONNECT_SERVICE_LICENSE".to_owned(), @@ -385,3 +579,74 @@ fn cli_for_deployment( .output() .expect("run rustfs-cli") } + +#[cfg(unix)] +fn relay_export_cli( + artifact: &Path, + envelope: &Path, + public_key: &Path, + state: &Path, + key_id: &str, + acknowledge_reviewed: bool, +) -> std::process::Output { + let mut command = Command::new(env!("CARGO_BIN_EXE_rustfs-cli")); + command + .args(["connect", "license", "relay-export", "--artifact"]) + .arg(artifact) + .arg("--envelope") + .arg(envelope) + .args(["--transfer-uid", RELAY_TRANSFER_UID]); + license_scope_args(&mut command, public_key, state, key_id); + if acknowledge_reviewed { + command.arg("--acknowledge-reviewed"); + } + command.output().expect("run relay export") +} + +#[cfg(unix)] +#[allow(clippy::too_many_arguments)] +fn relay_import_cli( + envelope: &Path, + receipt_seed: &Path, + public_key: &Path, + state: &Path, + key_id: &str, + receipt_key_id: &str, + acknowledge_reviewed: bool, +) -> std::process::Output { + let mut command = Command::new(env!("CARGO_BIN_EXE_rustfs-cli")); + command + .args(["connect", "license", "relay-import", "--envelope"]) + .arg(envelope) + .arg("--receipt-signing-key-file") + .arg(receipt_seed) + .args(["--receipt-key-id", receipt_key_id]); + license_scope_args(&mut command, public_key, state, key_id); + if acknowledge_reviewed { + command.arg("--acknowledge-reviewed"); + } + command.output().expect("run relay import") +} + +#[cfg(unix)] +fn license_scope_args(command: &mut Command, public_key: &Path, state: &Path, key_id: &str) { + command + .arg("--state-dir") + .arg(state) + .arg("--public-key-file") + .arg(public_key) + .args([ + "--key-id", + key_id, + "--issuer", + ISSUER, + "--audience", + AUDIENCE, + "--organization", + ORGANIZATION, + "--deployment", + DEPLOYMENT, + "--service-code", + SERVICE, + ]); +}