From 71a8939d9b72decbca19ec69b0ad8e43c816592f Mon Sep 17 00:00:00 2001 From: Chris Date: Sun, 13 Sep 2026 11:04:37 +0800 Subject: [PATCH] Add client-to-deployment performance diagnostics (#7726) feat: add client performance diagnostics --- crates/signer/src/lib.rs | 1 + crates/signer/src/request_signature_v4.rs | 13 + rustfs/src/admin/handlers/diagnostics.rs | 60 + rustfs/src/admin/route_policy.rs | 6 + rustfs/src/admin/route_registration_test.rs | 1 + rustfs/src/config/cli.rs | 111 +- rustfs/src/config/mod.rs | 4 +- rustfs/src/config/opt.rs | 3 +- rustfs/src/connect/diagnostics/mod.rs | 8 + rustfs/src/connect/diagnostics/perf_client.rs | 1255 +++++++++++++++++ rustfs/src/connect/mod.rs | 33 +- rustfs/src/startup_entrypoint.rs | 149 +- rustfs/tests/connect_perf_client.rs | 584 ++++++++ 13 files changed, 2203 insertions(+), 25 deletions(-) create mode 100644 rustfs/src/connect/diagnostics/perf_client.rs create mode 100644 rustfs/tests/connect_perf_client.rs diff --git a/crates/signer/src/lib.rs b/crates/signer/src/lib.rs index d802b0f32..8daa0366f 100644 --- a/crates/signer/src/lib.rs +++ b/crates/signer/src/lib.rs @@ -32,4 +32,5 @@ pub use request_signature_v4::sign_v4; pub use request_signature_v4::sign_v4_trailer; pub use request_signature_v4::try_pre_sign_v4; pub use request_signature_v4::try_sign_v4; +pub use request_signature_v4::try_sign_v4_headers; pub use request_signature_v4::try_sign_v4_trailer; diff --git a/crates/signer/src/request_signature_v4.rs b/crates/signer/src/request_signature_v4.rs index 1c82b9662..cd21f2008 100644 --- a/crates/signer/src/request_signature_v4.rs +++ b/crates/signer/src/request_signature_v4.rs @@ -679,6 +679,19 @@ pub fn try_sign_v4( .map_err(|failure| failure.error) } +pub fn try_sign_v4_headers( + parts: request::Parts, + content_len: i64, + access_key_id: &str, + secret_access_key: &str, + session_token: &str, + location: &str, +) -> SignResult { + let request = request::Request::from_parts(parts, Body::empty()); + try_sign_v4(request, content_len, access_key_id, secret_access_key, session_token, location) + .map(|request| request.into_parts().0.headers) +} + pub fn sign_v4_trailer( req: request::Request, access_key_id: &str, diff --git a/rustfs/src/admin/handlers/diagnostics.rs b/rustfs/src/admin/handlers/diagnostics.rs index 18758b2fc..7908e856c 100644 --- a/rustfs/src/admin/handlers/diagnostics.rs +++ b/rustfs/src/admin/handlers/diagnostics.rs @@ -53,6 +53,7 @@ const CONTENT_TYPE_NDJSON: &str = "application/x-ndjson"; pub(crate) const CLIENT_DEVNULL_MAX_BYTES: u64 = 1024 * 1024 * 1024; pub(crate) const CLIENT_DEVNULL_MAX_DURATION: Duration = Duration::from_secs(30); pub(crate) const CLIENT_DEVNULL_MAX_CONCURRENCY: usize = 4; +pub(crate) const CLIENT_DEVNULL_SOURCE_MAX_BYTES: u64 = 1024 * 1024; static CLIENT_DEVNULL_ADMISSION: Semaphore = Semaphore::const_new(CLIENT_DEVNULL_MAX_CONCURRENCY); /// Cap on how many locks a single `top/locks` response enumerates, matching the @@ -124,6 +125,11 @@ pub fn register_diagnostics_route(r: &mut S3Router) -> std::io:: format!("{ADMIN_PREFIX}/v3/speedtest/client/devnull").as_str(), AdminOperation(&SpeedtestClientDevnullHandler {}), )?; + r.insert( + Method::GET, + format!("{ADMIN_PREFIX}/v3/speedtest/client/devnull").as_str(), + AdminOperation(&SpeedtestClientSourceHandler {}), + )?; Ok(()) } @@ -940,6 +946,23 @@ impl Operation for SpeedtestHandler { /// number (mirrors MinIO's `ClientDevNull`). pub struct SpeedtestClientDevnullHandler {} +/// `GET /v3/speedtest/client/devnull?bytes=N` — bounded generated download. +pub struct SpeedtestClientSourceHandler {} + +fn client_source_bytes(uri: &Uri) -> S3Result { + let bytes = query_value(uri, "bytes") + .and_then(|value| value.parse::().ok()) + .ok_or_else(|| s3_error!(InvalidRequest, "client speedtest requires a positive bytes parameter"))?; + if bytes == 0 || bytes > CLIENT_DEVNULL_SOURCE_MAX_BYTES { + return Err(s3_error!( + EntityTooLarge, + "client speedtest download exceeds the {}-byte limit", + CLIENT_DEVNULL_SOURCE_MAX_BYTES + )); + } + usize::try_from(bytes).map_err(|_| s3_error!(EntityTooLarge, "client speedtest download exceeds platform limits")) +} + fn validate_client_devnull_content_length(headers: &HeaderMap) -> S3Result<()> { let Some(content_length) = headers.get(CONTENT_LENGTH) else { return Ok(()); @@ -1029,6 +1052,17 @@ impl Operation for SpeedtestClientDevnullHandler { } } +#[async_trait::async_trait] +impl Operation for SpeedtestClientSourceHandler { + async fn call(&self, req: S3Request, _params: Params<'_, '_>) -> S3Result> { + authorize(&req, AdminAction::HealthInfoAdminAction).await?; + let bytes = client_source_bytes(&req.uri)?; + let _permit = acquire_client_devnull_permit()?; + + Ok(S3Response::new((StatusCode::OK, Body::from(vec![0_u8; bytes])))) + } +} + // --------------------------------------------------------------------------- // Query helpers // --------------------------------------------------------------------------- @@ -1179,6 +1213,32 @@ mod tests { assert_eq!(total, 4); } + #[test] + fn client_source_requires_a_bounded_positive_size() { + let valid = format!("/rustfs/admin/v3/speedtest/client/devnull?bytes={CLIENT_DEVNULL_SOURCE_MAX_BYTES}") + .parse::() + .expect("valid URI"); + assert_eq!( + client_source_bytes(&valid).expect("size at the limit should succeed"), + usize::try_from(CLIENT_DEVNULL_SOURCE_MAX_BYTES).expect("source limit fits usize") + ); + + for invalid in [ + "/rustfs/admin/v3/speedtest/client/devnull", + "/rustfs/admin/v3/speedtest/client/devnull?bytes=0", + "/rustfs/admin/v3/speedtest/client/devnull?bytes=invalid", + ] { + let uri = invalid.parse::().expect("valid URI"); + assert!(client_source_bytes(&uri).is_err(), "{invalid} must fail"); + } + + let oversized = format!("/rustfs/admin/v3/speedtest/client/devnull?bytes={}", CLIENT_DEVNULL_SOURCE_MAX_BYTES + 1) + .parse::() + .expect("valid URI"); + let err = client_source_bytes(&oversized).expect_err("oversized source request must fail"); + assert_eq!(err.code(), &S3ErrorCode::EntityTooLarge); + } + #[test] fn client_devnull_rejects_invalid_content_length() { let mut headers = HeaderMap::new(); diff --git a/rustfs/src/admin/route_policy.rs b/rustfs/src/admin/route_policy.rs index 7a6137c21..9fbcd531a 100644 --- a/rustfs/src/admin/route_policy.rs +++ b/rustfs/src/admin/route_policy.rs @@ -812,6 +812,12 @@ pub const ADMIN_ROUTE_POLICY_SPECS: &[AdminRouteSpec] = &[ HEALTH_INFO, RouteRiskLevel::High, ), + admin( + HttpMethod::Get, + "/rustfs/admin/v3/speedtest/client/devnull", + HEALTH_INFO, + RouteRiskLevel::High, + ), admin(HttpMethod::Post, "/rustfs/admin/v4/inspect/archive", INSPECT_DATA, RouteRiskLevel::High), // MinIO-compatible profiling / trace endpoints. admin(HttpMethod::Post, "/rustfs/admin/v3/profiling/start", PROFILING, RouteRiskLevel::High), diff --git a/rustfs/src/admin/route_registration_test.rs b/rustfs/src/admin/route_registration_test.rs index 1673a00a9..6b5394823 100644 --- a/rustfs/src/admin/route_registration_test.rs +++ b/rustfs/src/admin/route_registration_test.rs @@ -375,6 +375,7 @@ fn expected_admin_route_matrix() -> Vec { admin_route(Method::POST, "/v3/speedtest/net"), admin_route(Method::POST, "/v3/speedtest/site"), admin_route(Method::POST, "/v3/speedtest/client/devnull"), + admin_route(Method::GET, "/v3/speedtest/client/devnull"), admin_route(Method::GET, "/debug/tls/status"), admin_route(Method::POST, "/v3/kms/create-key"), admin_route(Method::POST, "/v3/kms/key/create"), diff --git a/rustfs/src/config/cli.rs b/rustfs/src/config/cli.rs index cab9154cd..93f5bb738 100644 --- a/rustfs/src/config/cli.rs +++ b/rustfs/src/config/cli.rs @@ -213,8 +213,115 @@ pub struct ConnectPerformanceOpts { #[derive(Subcommand, Clone)] pub enum ConnectPerformanceCommands { + /// Measure bounded client-to-deployment transfer performance + Client(Box), /// Measure generated-file write and warm page-cache read performance - Drive(ConnectDrivePerformanceOpts), + Drive(Box), +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq, ValueEnum)] +pub enum ConnectClientPerformanceOperation { + Get, + Put, +} + +#[derive(Args, Clone)] +pub struct ConnectClientPerformanceOpts { + /// Directory containing an enrolled Connect device identity + #[arg(long = "state-dir")] + pub state_dir: PathBuf, + + /// RustFS deployment endpoint + #[arg(long, value_parser = NonEmptyStringValueParser::new())] + pub endpoint: String, + + /// Optional PEM root certificate for the deployment endpoint + #[arg(long = "ca-file")] + pub ca_file: Option, + + /// Optional explicit HTTP(S) proxy without embedded credentials + #[arg(long, value_parser = NonEmptyStringValueParser::new())] + pub proxy: Option, + + /// Owner-readable file containing the S3 access key + #[arg(long = "access-key-file")] + pub access_key_file: PathBuf, + + /// Owner-readable file containing the S3 secret key + #[arg(long = "secret-key-file")] + pub secret_key_file: PathBuf, + + /// Optional owner-readable file containing an S3 session token + #[arg(long = "session-token-file")] + pub session_token_file: Option, + + /// New local archive path; an existing file is never replaced + #[arg(long)] + pub output: PathBuf, + + /// Negotiated producer schema version + #[arg(long = "schema-version", default_value_t = 1)] + pub schema_version: u16, + + /// Negotiated producer capability + #[arg(long, default_value = "performance.client@1", value_parser = NonEmptyStringValueParser::new())] + pub capability: String, + + /// Organization resource name bound to the export + #[arg(long, value_parser = NonEmptyStringValueParser::new())] + pub organization: String, + + /// Cluster resource name bound to the export + #[arg(long, value_parser = NonEmptyStringValueParser::new())] + pub cluster: String, + + /// Cluster-device resource name bound to the export + #[arg(long, value_parser = NonEmptyStringValueParser::new())] + pub device: String, + + /// UUIDv7 diagnostic run identifier issued by Connect + #[arg(long = "run-uid", value_parser = NonEmptyStringValueParser::new())] + pub run_uid: String, + + /// UUIDv7 artifact identifier issued by Connect + #[arg(long = "artifact-uid", value_parser = NonEmptyStringValueParser::new())] + pub artifact_uid: String, + + /// UUIDv7 consent identifier issued by Connect + #[arg(long = "consent-uid", value_parser = NonEmptyStringValueParser::new())] + pub consent_uid: String, + + /// Consent policy revision bound to this measurement + #[arg(long = "policy-revision")] + pub policy_revision: u64, + + /// Consent expiry as UTC Unix seconds + #[arg(long = "consent-expires-at")] + pub consent_expires_at_unix: i64, + + /// Artifact expiry as UTC Unix seconds + #[arg(long = "expires-at")] + pub expires_at_unix: i64, + + /// Client transfer operation + #[arg(long, value_enum)] + pub operation: ConnectClientPerformanceOperation, + + /// Generated transfer size in bytes + #[arg(long = "traffic-bytes", default_value_t = 65_536)] + pub traffic_bytes: u64, + + /// Maximum wall-clock duration in milliseconds + #[arg(long = "duration-millis", default_value_t = 1_000)] + pub duration_millis: u64, + + /// Stable opaque alias for the deployment target + #[arg(long = "target-alias", default_value = "deployment-1", value_parser = NonEmptyStringValueParser::new())] + pub target_alias: String, + + /// Confirm this explicit local L1 diagnostic operation + #[arg(long = "acknowledge-l1", required = true, action = clap::ArgAction::SetTrue)] + pub acknowledge_l1: bool, } #[derive(Args, Clone)] @@ -952,6 +1059,8 @@ pub enum CommandResult { ConnectLicense(ConnectLicenseCommands), /// Consent-bound local Connect drive performance export ConnectDrivePerformance(ConnectDrivePerformanceOpts), + /// Consent-bound client-to-deployment performance export + ConnectClientPerformance(ConnectClientPerformanceOpts), /// Consent-bound local Connect profile export ConnectProfile(ConnectProfileOpts), /// Consent-bound local Connect log export diff --git a/rustfs/src/config/mod.rs b/rustfs/src/config/mod.rs index 2f62c89d9..c8b472c15 100644 --- a/rustfs/src/config/mod.rs +++ b/rustfs/src/config/mod.rs @@ -51,7 +51,9 @@ mod config_test; // Re-export public types pub use cli::{CommandResult, InfoOpts, InfoType}; -pub use cli::{ConnectDrivePerformanceOpts, ConnectPerformanceCommands}; +pub use cli::{ + ConnectClientPerformanceOperation, ConnectClientPerformanceOpts, ConnectDrivePerformanceOpts, ConnectPerformanceCommands, +}; pub use cli::{ConnectLicenseArtifactOpts, ConnectLicenseCommands, ConnectLicenseScopeOpts}; pub use cli::{ConnectLogsMode, ConnectLogsOpts}; pub use cli::{ConnectProfileOpts, ConnectProfileTool, ConnectThreadProfileScope}; diff --git a/rustfs/src/config/opt.rs b/rustfs/src/config/opt.rs index ed5cdab01..acc0e86f4 100644 --- a/rustfs/src/config/opt.rs +++ b/rustfs/src/config/opt.rs @@ -144,7 +144,8 @@ impl Opt { ConnectCommands::Register(opts) => Ok(CommandResult::ConnectRegister(opts)), ConnectCommands::License(opts) => Ok(CommandResult::ConnectLicense(opts.command)), ConnectCommands::Performance(opts) => match opts.command { - ConnectPerformanceCommands::Drive(opts) => Ok(CommandResult::ConnectDrivePerformance(opts)), + ConnectPerformanceCommands::Client(opts) => Ok(CommandResult::ConnectClientPerformance(*opts)), + ConnectPerformanceCommands::Drive(opts) => Ok(CommandResult::ConnectDrivePerformance(*opts)), }, ConnectCommands::Profile(opts) => Ok(CommandResult::ConnectProfile(opts)), ConnectCommands::Logs(opts) => Ok(CommandResult::ConnectLogs(opts)), diff --git a/rustfs/src/connect/diagnostics/mod.rs b/rustfs/src/connect/diagnostics/mod.rs index dd9ed07c2..e0737add6 100644 --- a/rustfs/src/connect/diagnostics/mod.rs +++ b/rustfs/src/connect/diagnostics/mod.rs @@ -13,6 +13,7 @@ // limitations under the License. mod logs; +mod perf_client; mod perf_drive; mod profile_cpu; mod profile_memory; @@ -32,6 +33,13 @@ pub use logs::{ CaptureMode, LOGS_CAPABILITY, LOGS_SCHEMA_VERSION, LocalLogConsent, LogCaptureError, LogCaptureRequest, LogProvenance, SavedLogExport, SignedLogExport, export_logs, save_signed_log_export, }; +pub use perf_client::{ + CLIENT_CAPABILITY, CLIENT_SCHEMA_VERSION, ClientDiagnosticResult, ClientMeasurement, ClientOperation, ClientOutcome, + ClientPerformanceData, ClientPerformanceError, ClientPerformanceRequest, ClientProbe, ClientProbeError, ClientProbeFuture, + ClientProbeMeasurement, ClientProvenance, ClientReasonCode, ClientTargetParameters, ClientTargetReasonCode, + ClientTargetResult, ClientTargetUnits, HttpClientProbe, LocalClientConsent, SavedClientExport, SignedClientExport, + measure_client, read_protected_client_credential, save_signed_client_export, sign_client_export, validate_client_limits, +}; pub use perf_drive::{ DRIVE_CAPABILITY, DRIVE_SCHEMA_VERSION, DriveDiagnosticResult, DriveMeasurement, DriveOutcome, DrivePerformanceData, DrivePerformanceError, DrivePerformanceRequest, DriveProvenance, DriveReadMode, DriveReasonCode, DriveTargetParameters, diff --git a/rustfs/src/connect/diagnostics/perf_client.rs b/rustfs/src/connect/diagnostics/perf_client.rs new file mode 100644 index 000000000..00c27c5fc --- /dev/null +++ b/rustfs/src/connect/diagnostics/perf_client.rs @@ -0,0 +1,1255 @@ +// 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. + +//! Consent-bound client-to-deployment performance measurement. +//! +//! The producer sends generated bytes to, or reads generated bytes from, the +//! bounded RustFS admin speedtest endpoint. It never reads or writes customer +//! objects and keeps deployment credentials local to the invoking process. + +use std::fs::{self, File, OpenOptions}; +use std::future::Future; +use std::io::{Cursor, Read, Write as _}; +use std::path::Path; +use std::pin::Pin; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; + +use base64_simd::URL_SAFE_NO_PAD; +use bytes::Bytes; +use futures::StreamExt as _; +use p256::ecdsa::{Signature, SigningKey, signature::Signer as _}; +use p256::pkcs8::DecodePrivateKey as _; +use reqwest::{Client, Method, Response, StatusCode, Url}; +use serde::{Deserialize, Serialize}; +use sha2::{Digest as _, Sha256}; +use thiserror::Error; +use time::{OffsetDateTime, format_description::well_known::Rfc3339}; +use tokio_util::sync::CancellationToken; +use uuid::{Uuid, Variant, Version}; +use zeroize::Zeroizing; +use zip::{CompressionMethod, ZipWriter, write::SimpleFileOptions}; + +use crate::connect::DeviceIdentity; + +pub const CLIENT_SCHEMA_VERSION: u16 = 1; +pub const CLIENT_TOOL_ID: &str = "performance.client"; +pub const CLIENT_CAPABILITY: &str = "performance.client@1"; +pub const MAX_CLIENT_DURATION: Duration = Duration::from_secs(30); +pub const MAX_CLIENT_TRAFFIC_BYTES: u64 = 1_048_576; +pub const MAX_CLIENT_BANDWIDTH_BYTES_PER_SECOND: u64 = 1_048_576; +pub const MAX_CLIENT_RESULT_BYTES: usize = 262_144; + +const MAX_SAFE_INTEGER: u64 = 9_007_199_254_740_991; +const MAX_BUILD_FEATURES: usize = 64; +const MAX_VALIDITY_SECONDS: i64 = 2_592_000; +const MAX_FUTURE_SKEW_SECONDS: i64 = 300; +const MAX_ENVELOPE_BYTES: usize = 16_384; +const MAX_ARCHIVE_BYTES: usize = 524_288; +const MAX_DECOMPRESSED_BYTES: usize = 278_528; +const SIGNATURE_DOMAIN: &[u8] = b"rustfs-diagnostic-envelope-v1\0"; +const ENVELOPE_PATH: &str = "envelope.json"; +const SIGNATURE_PATH: &str = "envelope.sig"; +const RESULT_PATH: &str = "result.json"; +const CLIENT_SPEEDTEST_PATH: &str = "/rustfs/admin/v3/speedtest/client/devnull"; +const MAX_CLIENT_RESPONSE_BYTES: usize = 16_384; +const OUTPUT_MODE: u32 = 0o600; + +static CLIENT_COLLECTOR_ACTIVE: AtomicBool = AtomicBool::new(false); + +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize)] +#[serde(rename_all = "SCREAMING_SNAKE_CASE")] +pub enum ClientOperation { + GetObject, + PutObject, +} + +impl ClientOperation { + pub const fn as_str(self) -> &'static str { + match self { + Self::GetObject => "GET_OBJECT", + Self::PutObject => "PUT_OBJECT", + } + } +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize)] +#[serde(rename_all = "SCREAMING_SNAKE_CASE")] +pub enum ClientOutcome { + Succeeded, + Failed, + Cancelled, +} + +impl ClientOutcome { + pub const fn as_str(self) -> &'static str { + match self { + Self::Succeeded => "SUCCEEDED", + Self::Failed => "FAILED", + Self::Cancelled => "CANCELLED", + } + } +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize)] +#[serde(rename_all = "SCREAMING_SNAKE_CASE")] +pub enum ClientReasonCode { + Complete, + SourceUnavailable, + PermissionDenied, + Cancelled, + CollectionFailed, +} + +impl ClientReasonCode { + pub const fn as_str(self) -> &'static str { + match self { + Self::Complete => "COMPLETE", + Self::SourceUnavailable => "SOURCE_UNAVAILABLE", + Self::PermissionDenied => "PERMISSION_DENIED", + Self::Cancelled => "CANCELLED", + Self::CollectionFailed => "COLLECTION_FAILED", + } + } +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize)] +#[serde(rename_all = "SCREAMING_SNAKE_CASE")] +pub enum ClientTargetReasonCode { + Complete, + EndpointUnavailable, + ProxyFailure, + PermissionDenied, + TimedOut, + Cancelled, + ProtocolFailure, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct ClientTargetParameters { + pub operation: ClientOperation, + pub requested_bytes: u64, + pub duration_millis: u64, + pub concurrency: u8, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct ClientTargetUnits { + pub bytes: &'static str, + pub duration: &'static str, + pub latency: &'static str, + pub operation_count: &'static str, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct ClientProvenance { + repository: &'static str, + source_commit: String, + executable_sha256: String, + rustfs_version: String, + os_family: ClientOsFamily, + architecture: ClientArchitecture, + build_features: Vec, +} + +impl ClientProvenance { + pub fn new( + source_commit: impl Into, + executable_sha256: impl Into, + rustfs_version: impl Into, + build_features: Vec, + ) -> Self { + Self { + repository: "rustfs/rustfs", + source_commit: source_commit.into(), + executable_sha256: executable_sha256.into(), + rustfs_version: rustfs_version.into(), + os_family: ClientOsFamily::current(), + architecture: ClientArchitecture::current(), + build_features, + } + } +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize)] +#[serde(rename_all = "SCREAMING_SNAKE_CASE")] +enum ClientOsFamily { + Linux, + Darwin, + Windows, + Freebsd, + Other, +} + +impl ClientOsFamily { + fn current() -> Self { + match std::env::consts::OS { + "linux" => Self::Linux, + "macos" => Self::Darwin, + "windows" => Self::Windows, + "freebsd" => Self::Freebsd, + _ => Self::Other, + } + } +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize)] +#[serde(rename_all = "lowercase")] +enum ClientArchitecture { + #[serde(rename = "x86_64")] + X86_64, + Aarch64, + Other, +} + +impl ClientArchitecture { + fn current() -> Self { + match std::env::consts::ARCH { + "x86_64" => Self::X86_64, + "aarch64" => Self::Aarch64, + _ => Self::Other, + } + } +} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct LocalClientConsent { + pub consent_uid: String, + pub policy_revision: u64, + pub expires_at_unix: i64, + pub confirmed: bool, +} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct ClientPerformanceRequest { + pub organization_name: String, + pub cluster_name: String, + pub device_name: String, + pub run_uid: String, + pub artifact_uid: String, + pub schema_version: u16, + pub capability: String, + pub consent: LocalClientConsent, + pub produced_at_unix: i64, + pub expires_at_unix: i64, + pub nonce: [u8; 32], + pub duration: Duration, + pub operation: ClientOperation, + pub traffic_bytes: u64, + pub target_alias: String, + pub provenance: ClientProvenance, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct ClientPerformanceData { + pub operation: ClientOperation, + pub transferred_bytes: u64, + pub completed_operations: u64, + pub duration_millis: u64, + pub error_count: u64, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize)] +#[serde(rename_all = "camelCase")] +struct ClientCoverage { + requested_units: u32, + completed_units: u32, + unit: &'static str, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct ClientDiagnosticResult { + schema_version: u16, + run_uid: String, + tool_id: &'static str, + capability: &'static str, + outcome: ClientOutcome, + reason_code: ClientReasonCode, + duration_millis: u64, + provenance: ClientProvenance, + coverage: ClientCoverage, + data: Option, +} + +impl ClientDiagnosticResult { + pub fn outcome(&self) -> ClientOutcome { + self.outcome + } + + pub fn reason_code(&self) -> ClientReasonCode { + self.reason_code + } + + pub fn data(&self) -> Option<&ClientPerformanceData> { + self.data.as_ref() + } +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct ClientTargetResult { + pub target_alias: String, + pub outcome: ClientOutcome, + pub reason_code: ClientTargetReasonCode, + pub parameters: ClientTargetParameters, + pub units: ClientTargetUnits, + pub transferred_bytes: u64, + pub completed_operations: u64, + pub latency_micros: Option, + pub duration_millis: u64, +} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct ClientMeasurement { + pub result: ClientDiagnosticResult, + pub target: ClientTargetResult, +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub struct ClientProbeMeasurement { + pub transferred_bytes: u64, + pub duration: Duration, + pub latency: Duration, +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum ClientProbeError { + EndpointUnavailable, + ProxyFailure, + PermissionDenied, + TimedOut, + Cancelled, + ProtocolFailure, +} + +pub type ClientProbeFuture<'a> = Pin> + Send + 'a>>; + +pub trait ClientProbe: Send + Sync { + fn probe<'a>(&'a self, request: &'a ClientPerformanceRequest, cancel: &'a CancellationToken) -> ClientProbeFuture<'a>; +} + +pub struct HttpClientProbe { + endpoint: Url, + client: Client, + access_key: Zeroizing, + secret_key: Zeroizing, + session_token: Zeroizing, + proxy_configured: bool, +} + +impl HttpClientProbe { + pub fn new( + endpoint: &str, + root_ca_pem: Option<&[u8]>, + proxy: Option<&str>, + access_key: Zeroizing, + secret_key: Zeroizing, + session_token: Zeroizing, + timeout: Duration, + ) -> Result { + let endpoint = deployment_endpoint(endpoint)?; + if access_key.is_empty() || secret_key.is_empty() { + return Err(ClientPerformanceError::InvalidCredential); + } + let mut builder = Client::builder() + .no_proxy() + .redirect(reqwest::redirect::Policy::none()) + .timeout(timeout); + if let Some(root_ca_pem) = root_ca_pem { + let certificate = + reqwest::Certificate::from_pem(root_ca_pem).map_err(|_| ClientPerformanceError::InvalidRootCertificate)?; + builder = builder.add_root_certificate(certificate); + } + if let Some(proxy) = proxy { + let proxy_url = proxy_url(proxy)?; + builder = builder.proxy(reqwest::Proxy::all(proxy_url).map_err(|_| ClientPerformanceError::InvalidProxy)?); + } + let client = builder.build().map_err(|_| ClientPerformanceError::TransportConfiguration)?; + Ok(Self { + endpoint, + client, + access_key, + secret_key, + session_token, + proxy_configured: proxy.is_some(), + }) + } + + async fn execute( + &self, + request: &ClientPerformanceRequest, + cancel: &CancellationToken, + ) -> Result { + let started = Instant::now(); + let traffic_bytes = usize::try_from(request.traffic_bytes).map_err(|_| ClientProbeError::ProtocolFailure)?; + let payload = match request.operation { + ClientOperation::PutObject => Bytes::from(vec![0xa5; traffic_bytes]), + ClientOperation::GetObject => Bytes::new(), + }; + let mut url = self + .endpoint + .join(CLIENT_SPEEDTEST_PATH) + .map_err(|_| ClientProbeError::ProtocolFailure)?; + let method = match request.operation { + ClientOperation::PutObject => Method::POST, + ClientOperation::GetObject => { + url.query_pairs_mut().append_pair("bytes", &request.traffic_bytes.to_string()); + Method::GET + } + }; + let payload_hash = hex_lower(&Sha256::digest(&payload)); + let unsigned = http::Request::builder() + .method(method.clone()) + .uri(url.as_str()) + .header("x-amz-content-sha256", payload_hash) + .body(()) + .map_err(|_| ClientProbeError::ProtocolFailure)?; + let signed_headers = rustfs_signer::try_sign_v4_headers( + unsigned.into_parts().0, + i64::try_from(payload.len()).map_err(|_| ClientProbeError::ProtocolFailure)?, + &self.access_key, + &self.secret_key, + &self.session_token, + "us-east-1", + ) + .map_err(|_| ClientProbeError::ProtocolFailure)?; + let send = self.client.request(method, url).headers(signed_headers).body(payload).send(); + let response = tokio::select! { + () = cancel.cancelled() => return Err(ClientProbeError::Cancelled), + response = send => response.map_err(|error| self.transport_error(&error))?, + }; + let status = response.status(); + if status == StatusCode::UNAUTHORIZED || status == StatusCode::FORBIDDEN { + return Err(ClientProbeError::PermissionDenied); + } + if !status.is_success() { + return Err(ClientProbeError::ProtocolFailure); + } + let transferred_bytes = match request.operation { + ClientOperation::PutObject => { + let body = self.read_response_body(response, MAX_CLIENT_RESPONSE_BYTES, cancel).await?; + let response: ClientDevnullResponse = + serde_json::from_slice(&body).map_err(|_| ClientProbeError::ProtocolFailure)?; + if !response.measured || response.kind != "client-devnull" || response.rx_bytes != request.traffic_bytes { + return Err(ClientProbeError::ProtocolFailure); + } + request.traffic_bytes + } + ClientOperation::GetObject => { + let body = self.read_response_body(response, traffic_bytes, cancel).await?; + let body_len = u64::try_from(body.len()).map_err(|_| ClientProbeError::ProtocolFailure)?; + if body_len != request.traffic_bytes || body.iter().any(|byte| *byte != 0) { + return Err(ClientProbeError::ProtocolFailure); + } + body_len + } + }; + let latency = started.elapsed(); + Ok(ClientProbeMeasurement { + transferred_bytes, + duration: latency, + latency, + }) + } + + async fn read_response_body( + &self, + response: Response, + max_bytes: usize, + cancel: &CancellationToken, + ) -> Result, ClientProbeError> { + let max_bytes_u64 = u64::try_from(max_bytes).map_err(|_| ClientProbeError::ProtocolFailure)?; + if response.content_length().is_some_and(|length| length > max_bytes_u64) { + return Err(ClientProbeError::ProtocolFailure); + } + let mut body = Vec::with_capacity(max_bytes.min(MAX_CLIENT_RESPONSE_BYTES)); + let mut stream = response.bytes_stream(); + loop { + let chunk = tokio::select! { + () = cancel.cancelled() => return Err(ClientProbeError::Cancelled), + chunk = stream.next() => chunk, + }; + let Some(chunk) = chunk else { + break; + }; + let chunk = chunk.map_err(|error| self.transport_error(&error))?; + let new_length = body.len().checked_add(chunk.len()).ok_or(ClientProbeError::ProtocolFailure)?; + if new_length > max_bytes { + return Err(ClientProbeError::ProtocolFailure); + } + body.extend_from_slice(&chunk); + } + Ok(body) + } + + fn transport_error(&self, error: &reqwest::Error) -> ClientProbeError { + if error.is_timeout() { + ClientProbeError::TimedOut + } else if self.proxy_configured { + ClientProbeError::ProxyFailure + } else if error.is_connect() { + ClientProbeError::EndpointUnavailable + } else { + ClientProbeError::ProtocolFailure + } + } +} + +impl ClientProbe for HttpClientProbe { + fn probe<'a>(&'a self, request: &'a ClientPerformanceRequest, cancel: &'a CancellationToken) -> ClientProbeFuture<'a> { + Box::pin(self.execute(request, cancel)) + } +} + +#[derive(Deserialize)] +struct ClientDevnullResponse { + kind: String, + measured: bool, + rx_bytes: u64, +} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct SignedClientExport { + pub artifact_uid: String, + pub outcome: ClientOutcome, + pub reason_code: ClientReasonCode, + pub envelope_json: Vec, + pub envelope_signature: Vec, + pub result_json: Vec, + pub archive_bytes: Vec, + pub archive_sha256: String, +} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct SavedClientExport { + pub artifact_uid: String, + pub archive_size_bytes: u64, + pub archive_sha256: String, +} + +#[derive(Debug, Error)] +pub enum ClientPerformanceError { + #[error("client_performance_local_consent_required")] + ConsentRequired, + #[error("client_performance_local_consent_expired")] + ConsentExpired, + #[error("client_performance_request_expired")] + Expired, + #[error("client_performance_invalid_request")] + InvalidRequest, + #[error("client_performance_unsupported_version")] + UnsupportedVersion, + #[error("client_performance_unsupported_capability")] + UnsupportedCapability, + #[error("client_performance_limit_exceeded")] + LimitExceeded, + #[error("client_performance_collection_cancelled")] + Cancelled, + #[error("client_performance_busy")] + Busy, + #[error("client_performance_invalid_endpoint")] + InvalidEndpoint, + #[error("client_performance_invalid_proxy")] + InvalidProxy, + #[error("client_performance_invalid_root_certificate")] + InvalidRootCertificate, + #[error("client_performance_invalid_credential")] + InvalidCredential, + #[error("client_performance_transport_configuration")] + TransportConfiguration, + #[error("client_performance_signing_failed")] + Signing, + #[error("client_performance_encoding_failed")] + Encoding, + #[error("client_performance_output_exists")] + AlreadyExists, + #[error("client_performance_io_failed")] + Io(#[source] std::io::Error), + #[error("client_performance_output_durability_failed")] + DurabilityAfterCommit(#[source] std::io::Error), +} + +pub async fn measure_client( + request: &ClientPerformanceRequest, + probe: &impl ClientProbe, + cancel: &CancellationToken, +) -> Result { + request.validate(unix_now()?)?; + if CLIENT_COLLECTOR_ACTIVE + .compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire) + .is_err() + { + return Err(ClientPerformanceError::Busy); + } + let _guard = ActiveGuard; + if cancel.is_cancelled() { + return Ok(terminal_measurement( + request, + ClientOutcome::Cancelled, + ClientReasonCode::Cancelled, + ClientTargetReasonCode::Cancelled, + Duration::ZERO, + )); + } + let started = Instant::now(); + let deadline = tokio::time::sleep(request.duration); + tokio::pin!(deadline); + let probe = probe.probe(request, cancel); + tokio::pin!(probe); + let result = tokio::select! { + () = cancel.cancelled() => Err(ClientProbeError::Cancelled), + () = &mut deadline => Err(ClientProbeError::TimedOut), + result = &mut probe => result, + }; + Ok(match result { + Ok(sample) if sample.transferred_bytes == request.traffic_bytes => success_measurement(request, sample), + Ok(_) => failed_measurement(request, started.elapsed(), ClientTargetReasonCode::ProtocolFailure), + Err(ClientProbeError::Cancelled) => terminal_measurement( + request, + ClientOutcome::Cancelled, + ClientReasonCode::Cancelled, + ClientTargetReasonCode::Cancelled, + started.elapsed(), + ), + Err(error) => failed_measurement(request, started.elapsed(), target_reason(error)), + }) +} + +pub fn sign_client_export( + request: &ClientPerformanceRequest, + measurement: &ClientMeasurement, + key: &DeviceIdentity, + cancel: &CancellationToken, +) -> Result { + request.validate(unix_now()?)?; + check_cancel(cancel)?; + let result = &measurement.result; + if result.outcome != ClientOutcome::Succeeded + || result.data.is_none() + || result.run_uid != request.run_uid + || result.schema_version != CLIENT_SCHEMA_VERSION + || result.tool_id != CLIENT_TOOL_ID + || result.capability != CLIENT_CAPABILITY + { + return Err(ClientPerformanceError::InvalidRequest); + } + let result_json = serde_json::to_vec(result).map_err(|_| ClientPerformanceError::Encoding)?; + if result_json.is_empty() || result_json.len() > MAX_CLIENT_RESULT_BYTES { + return Err(ClientPerformanceError::LimitExceeded); + } + let device_key_id = hex_lower(&Sha256::digest(key.public_key_der())); + let result_sha256 = hex_lower(&Sha256::digest(&result_json)); + let envelope = ClientEnvelope { + format_version: "rustfs.connect.diagnosticEnvelope/1", + protocol_version: "v1", + organization_name: &request.organization_name, + cluster_name: &request.cluster_name, + device_name: &request.device_name, + run_uid: &request.run_uid, + artifact_uid: &request.artifact_uid, + tool_id: CLIENT_TOOL_ID, + schema_version: CLIENT_SCHEMA_VERSION, + classification: "L1", + consent_uid: &request.consent.consent_uid, + policy_revision: request.consent.policy_revision, + produced_at: timestamp(request.produced_at_unix)?, + expires_at: timestamp(request.expires_at_unix)?, + nonce: URL_SAFE_NO_PAD.encode_to_string(request.nonce), + device_key_id: &device_key_id, + payload: ClientPayload { + path: RESULT_PATH, + media_type: "application/json", + size_bytes: u64::try_from(result_json.len()).map_err(|_| ClientPerformanceError::LimitExceeded)?, + sha256: &result_sha256, + }, + }; + let envelope_json = serde_json::to_vec(&envelope).map_err(|_| ClientPerformanceError::Encoding)?; + if envelope_json.is_empty() || envelope_json.len() > MAX_ENVELOPE_BYTES { + return Err(ClientPerformanceError::LimitExceeded); + } + let envelope_signature = signature_document(key, &device_key_id, &envelope_json)?; + let decompressed = result_json + .len() + .checked_add(envelope_json.len()) + .and_then(|size| size.checked_add(envelope_signature.len())) + .ok_or(ClientPerformanceError::LimitExceeded)?; + if decompressed > MAX_DECOMPRESSED_BYTES { + return Err(ClientPerformanceError::LimitExceeded); + } + check_cancel(cancel)?; + if unix_now()? >= request.expires_at_unix { + return Err(ClientPerformanceError::Expired); + } + let archive_bytes = archive(&envelope_json, &envelope_signature, &result_json)?; + if archive_bytes.len() > MAX_ARCHIVE_BYTES { + return Err(ClientPerformanceError::LimitExceeded); + } + let archive_sha256 = hex_lower(&Sha256::digest(&archive_bytes)); + Ok(SignedClientExport { + artifact_uid: request.artifact_uid.clone(), + outcome: result.outcome, + reason_code: result.reason_code, + envelope_json, + envelope_signature, + result_json, + archive_bytes, + archive_sha256, + }) +} + +pub fn save_signed_client_export( + output: &Path, + export: &SignedClientExport, + cancel: &CancellationToken, +) -> Result { + check_cancel(cancel)?; + if !uuid7(&export.artifact_uid) || export.archive_bytes.is_empty() { + return Err(ClientPerformanceError::InvalidRequest); + } + if export.archive_bytes.len() > MAX_ARCHIVE_BYTES { + return Err(ClientPerformanceError::LimitExceeded); + } + if hex_lower(&Sha256::digest(&export.archive_bytes)) != export.archive_sha256 { + return Err(ClientPerformanceError::InvalidRequest); + } + let parent = output + .parent() + .filter(|path| !path.as_os_str().is_empty()) + .unwrap_or_else(|| Path::new(".")); + let filename = output + .file_name() + .ok_or(ClientPerformanceError::InvalidRequest)? + .to_string_lossy(); + let temporary = parent.join(format!(".{filename}.{}.partial", export.artifact_uid)); + let mut options = OpenOptions::new(); + options.write(true).create_new(true); + #[cfg(unix)] + { + use std::os::unix::fs::OpenOptionsExt as _; + options.mode(OUTPUT_MODE); + } + let mut file = options.open(&temporary).map_err(map_create_error)?; + let saved = (|| { + file.write_all(&export.archive_bytes).map_err(ClientPerformanceError::Io)?; + check_cancel(cancel)?; + file.sync_all().map_err(ClientPerformanceError::Io)?; + check_cancel(cancel)?; + fs::hard_link(&temporary, output).map_err(map_publish_error)?; + fs::remove_file(&temporary).map_err(ClientPerformanceError::DurabilityAfterCommit)?; + #[cfg(unix)] + File::open(parent) + .and_then(|directory| directory.sync_all()) + .map_err(ClientPerformanceError::DurabilityAfterCommit)?; + Ok(SavedClientExport { + artifact_uid: export.artifact_uid.clone(), + archive_size_bytes: u64::try_from(export.archive_bytes.len()).map_err(|_| ClientPerformanceError::LimitExceeded)?, + archive_sha256: export.archive_sha256.clone(), + }) + })(); + if saved.is_err() { + let _ = fs::remove_file(&temporary); + } + saved +} + +pub fn validate_client_limits(duration: Duration, traffic_bytes: u64) -> Result<(), ClientPerformanceError> { + if duration.is_zero() + || duration.as_millis() == 0 + || duration > MAX_CLIENT_DURATION + || traffic_bytes == 0 + || traffic_bytes > MAX_CLIENT_TRAFFIC_BYTES + { + return Err(ClientPerformanceError::LimitExceeded); + } + let bandwidth_budget = u64::try_from( + u128::from(MAX_CLIENT_BANDWIDTH_BYTES_PER_SECOND) + .checked_mul(duration.as_millis()) + .ok_or(ClientPerformanceError::LimitExceeded)? + / 1_000, + ) + .map_err(|_| ClientPerformanceError::LimitExceeded)?; + if traffic_bytes > bandwidth_budget.max(1) { + return Err(ClientPerformanceError::LimitExceeded); + } + Ok(()) +} + +pub fn read_protected_client_credential(path: &Path) -> Result, ClientPerformanceError> { + let metadata = fs::symlink_metadata(path).map_err(ClientPerformanceError::Io)?; + if metadata.file_type().is_symlink() || !metadata.is_file() { + return Err(ClientPerformanceError::InvalidCredential); + } + #[cfg(unix)] + { + use std::os::unix::fs::{MetadataExt as _, OpenOptionsExt as _, PermissionsExt as _}; + if metadata.uid() != process_uid() || metadata.permissions().mode() & 0o077 != 0 { + return Err(ClientPerformanceError::InvalidCredential); + } + let mut options = OpenOptions::new(); + options.read(true).custom_flags(libc::O_CLOEXEC | libc::O_NOFOLLOW); + let mut file = options.open(path).map_err(ClientPerformanceError::Io)?; + let opened = file.metadata().map_err(ClientPerformanceError::Io)?; + if opened.dev() != metadata.dev() || opened.ino() != metadata.ino() { + return Err(ClientPerformanceError::InvalidCredential); + } + read_credential_value(&mut file) + } + #[cfg(not(unix))] + { + let mut file = File::open(path).map_err(ClientPerformanceError::Io)?; + read_credential_value(&mut file) + } +} + +fn read_credential_value(reader: &mut impl Read) -> Result, ClientPerformanceError> { + let mut bytes = Vec::with_capacity(256); + reader + .take(4_097) + .read_to_end(&mut bytes) + .map_err(ClientPerformanceError::Io)?; + if bytes.is_empty() || bytes.len() > 4_096 || bytes.contains(&0) { + return Err(ClientPerformanceError::InvalidCredential); + } + while matches!(bytes.last(), Some(b'\n' | b'\r')) { + bytes.pop(); + } + let value = String::from_utf8(bytes).map_err(|_| ClientPerformanceError::InvalidCredential)?; + if value.is_empty() || value.len() > 4_096 || value.trim() != value { + return Err(ClientPerformanceError::InvalidCredential); + } + Ok(Zeroizing::new(value)) +} + +#[cfg(unix)] +#[allow(unsafe_code)] +fn process_uid() -> u32 { + // SAFETY: geteuid has no pointer arguments or caller preconditions. + unsafe { libc::geteuid() } +} + +impl ClientPerformanceRequest { + fn validate(&self, now_unix: i64) -> Result<(), ClientPerformanceError> { + if self.schema_version != CLIENT_SCHEMA_VERSION { + return Err(ClientPerformanceError::UnsupportedVersion); + } + if self.capability != CLIENT_CAPABILITY { + return Err(ClientPerformanceError::UnsupportedCapability); + } + if !self.consent.confirmed || self.consent.policy_revision == 0 { + return Err(ClientPerformanceError::ConsentRequired); + } + if self.consent.expires_at_unix <= now_unix || self.expires_at_unix > self.consent.expires_at_unix { + return Err(ClientPerformanceError::ConsentExpired); + } + let validity = self + .expires_at_unix + .checked_sub(self.produced_at_unix) + .ok_or(ClientPerformanceError::Expired)?; + if self.produced_at_unix > now_unix.saturating_add(MAX_FUTURE_SKEW_SECONDS) + || validity <= 0 + || validity > MAX_VALIDITY_SECONDS + || self.expires_at_unix <= now_unix + { + return Err(ClientPerformanceError::Expired); + } + validate_client_limits(self.duration, self.traffic_bytes)?; + if !uuid7(&self.run_uid) + || !uuid7(&self.artifact_uid) + || !uuid7(&self.consent.consent_uid) + || !resource_names_match(self) + || self.target_alias.is_empty() + || self.target_alias.len() > 128 + || !self + .target_alias + .bytes() + .all(|byte| byte.is_ascii_alphanumeric() || b"._:-".contains(&byte)) + || !lower_hex_string(&self.provenance.source_commit, 40) + || !lower_hex_string(&self.provenance.executable_sha256, 64) + || !version(&self.provenance.rustfs_version) + || self.provenance.build_features.len() > MAX_BUILD_FEATURES + || !self.provenance.build_features.iter().all(|value| build_feature(value)) + { + return Err(ClientPerformanceError::InvalidRequest); + } + Ok(()) + } +} + +struct ActiveGuard; + +impl Drop for ActiveGuard { + fn drop(&mut self) { + CLIENT_COLLECTOR_ACTIVE.store(false, Ordering::Release); + } +} + +fn success_measurement(request: &ClientPerformanceRequest, sample: ClientProbeMeasurement) -> ClientMeasurement { + let duration_millis = elapsed_millis(sample.duration); + let data = ClientPerformanceData { + operation: request.operation, + transferred_bytes: sample.transferred_bytes, + completed_operations: 1, + duration_millis, + error_count: 0, + }; + ClientMeasurement { + result: result( + request, + ClientOutcome::Succeeded, + ClientReasonCode::Complete, + duration_millis, + 1, + Some(data), + ), + target: ClientTargetResult { + target_alias: request.target_alias.clone(), + outcome: ClientOutcome::Succeeded, + reason_code: ClientTargetReasonCode::Complete, + parameters: target_parameters(request), + units: target_units(), + transferred_bytes: sample.transferred_bytes, + completed_operations: 1, + latency_micros: Some(elapsed_micros(sample.latency)), + duration_millis, + }, + } +} + +fn failed_measurement( + request: &ClientPerformanceRequest, + elapsed: Duration, + target_reason: ClientTargetReasonCode, +) -> ClientMeasurement { + let reason = match target_reason { + ClientTargetReasonCode::EndpointUnavailable => ClientReasonCode::SourceUnavailable, + ClientTargetReasonCode::PermissionDenied => ClientReasonCode::PermissionDenied, + _ => ClientReasonCode::CollectionFailed, + }; + terminal_measurement(request, ClientOutcome::Failed, reason, target_reason, elapsed) +} + +fn terminal_measurement( + request: &ClientPerformanceRequest, + outcome: ClientOutcome, + reason_code: ClientReasonCode, + target_reason: ClientTargetReasonCode, + elapsed: Duration, +) -> ClientMeasurement { + let duration_millis = elapsed_millis_allow_zero(elapsed); + ClientMeasurement { + result: result(request, outcome, reason_code, duration_millis, 0, None), + target: ClientTargetResult { + target_alias: request.target_alias.clone(), + outcome, + reason_code: target_reason, + parameters: target_parameters(request), + units: target_units(), + transferred_bytes: 0, + completed_operations: 0, + latency_micros: None, + duration_millis, + }, + } +} + +fn target_reason(error: ClientProbeError) -> ClientTargetReasonCode { + match error { + ClientProbeError::EndpointUnavailable => ClientTargetReasonCode::EndpointUnavailable, + ClientProbeError::ProxyFailure => ClientTargetReasonCode::ProxyFailure, + ClientProbeError::PermissionDenied => ClientTargetReasonCode::PermissionDenied, + ClientProbeError::TimedOut => ClientTargetReasonCode::TimedOut, + ClientProbeError::Cancelled => ClientTargetReasonCode::Cancelled, + ClientProbeError::ProtocolFailure => ClientTargetReasonCode::ProtocolFailure, + } +} + +fn target_parameters(request: &ClientPerformanceRequest) -> ClientTargetParameters { + ClientTargetParameters { + operation: request.operation, + requested_bytes: request.traffic_bytes, + duration_millis: elapsed_millis_allow_zero(request.duration), + concurrency: 1, + } +} + +fn target_units() -> ClientTargetUnits { + ClientTargetUnits { + bytes: "BYTE", + duration: "MILLISECOND", + latency: "MICROSECOND", + operation_count: "OPERATION", + } +} + +fn result( + request: &ClientPerformanceRequest, + outcome: ClientOutcome, + reason_code: ClientReasonCode, + duration_millis: u64, + completed_units: u32, + data: Option, +) -> ClientDiagnosticResult { + ClientDiagnosticResult { + schema_version: CLIENT_SCHEMA_VERSION, + run_uid: request.run_uid.clone(), + tool_id: CLIENT_TOOL_ID, + capability: CLIENT_CAPABILITY, + outcome, + reason_code, + duration_millis: duration_millis.min(30_000), + provenance: request.provenance.clone(), + coverage: ClientCoverage { + requested_units: 1, + completed_units, + unit: "WINDOW", + }, + data, + } +} + +#[derive(Serialize)] +#[serde(rename_all = "camelCase")] +struct ClientEnvelope<'a> { + format_version: &'static str, + protocol_version: &'static str, + organization_name: &'a str, + cluster_name: &'a str, + device_name: &'a str, + run_uid: &'a str, + artifact_uid: &'a str, + tool_id: &'static str, + schema_version: u16, + classification: &'static str, + consent_uid: &'a str, + policy_revision: u64, + produced_at: String, + expires_at: String, + nonce: String, + device_key_id: &'a str, + payload: ClientPayload<'a>, +} + +#[derive(Serialize)] +#[serde(rename_all = "camelCase")] +struct ClientPayload<'a> { + path: &'static str, + media_type: &'static str, + size_bytes: u64, + sha256: &'a str, +} + +#[derive(Serialize)] +#[serde(rename_all = "camelCase")] +struct ClientSignature<'a> { + algorithm: &'static str, + key_id: &'a str, + value: String, +} + +fn signature_document(key: &DeviceIdentity, key_id: &str, envelope: &[u8]) -> Result, ClientPerformanceError> { + let pkcs8 = key.to_pkcs8_der().map_err(|_| ClientPerformanceError::Signing)?; + let signing_key = SigningKey::from_pkcs8_der(pkcs8.as_slice()).map_err(|_| ClientPerformanceError::Signing)?; + let mut input = Vec::with_capacity(SIGNATURE_DOMAIN.len() + envelope.len()); + input.extend_from_slice(SIGNATURE_DOMAIN); + input.extend_from_slice(envelope); + let signature: Signature = signing_key.sign(&input); + serde_json::to_vec(&ClientSignature { + algorithm: "ES256", + key_id, + value: URL_SAFE_NO_PAD.encode_to_string(signature.normalize_s().to_bytes()), + }) + .map_err(|_| ClientPerformanceError::Encoding) +} + +fn archive(envelope: &[u8], signature: &[u8], result: &[u8]) -> Result, ClientPerformanceError> { + let cursor = Cursor::new(Vec::with_capacity(envelope.len() + signature.len() + result.len() + 512)); + let mut writer = ZipWriter::new(cursor); + let options = SimpleFileOptions::DEFAULT + .compression_method(CompressionMethod::Stored) + .unix_permissions(OUTPUT_MODE); + for (name, bytes) in [(ENVELOPE_PATH, envelope), (SIGNATURE_PATH, signature), (RESULT_PATH, result)] { + writer + .start_file(name, options) + .map_err(|_| ClientPerformanceError::Encoding)?; + writer.write_all(bytes).map_err(ClientPerformanceError::Io)?; + } + writer + .finish() + .map(|cursor| cursor.into_inner()) + .map_err(|_| ClientPerformanceError::Encoding) +} + +fn deployment_endpoint(value: &str) -> Result { + let mut url = Url::parse(value).map_err(|_| ClientPerformanceError::InvalidEndpoint)?; + let local_http = url.scheme() == "http" + && url + .host_str() + .is_some_and(|host| host == "localhost" || host.parse::().is_ok_and(|ip| ip.is_loopback())); + if (url.scheme() != "https" && !local_http) + || url.cannot_be_a_base() + || !url.username().is_empty() + || url.password().is_some() + || url.query().is_some() + || url.fragment().is_some() + { + return Err(ClientPerformanceError::InvalidEndpoint); + } + url.set_query(None); + url.set_fragment(None); + if !url.path().ends_with('/') { + url.set_path(&format!("{}/", url.path())); + } + Ok(url) +} + +fn proxy_url(value: &str) -> Result { + let url = Url::parse(value).map_err(|_| ClientPerformanceError::InvalidProxy)?; + if !matches!(url.scheme(), "http" | "https") + || url.cannot_be_a_base() + || !url.username().is_empty() + || url.password().is_some() + || url.query().is_some() + || url.fragment().is_some() + { + return Err(ClientPerformanceError::InvalidProxy); + } + Ok(url) +} + +fn resource_names_match(request: &ClientPerformanceRequest) -> bool { + let Some(organization_uid) = request.organization_name.strip_prefix("organizations/") else { + return false; + }; + if !uuid7(organization_uid) { + return false; + } + let cluster_prefix = format!("{}/clusters/", request.organization_name); + let Some(cluster_uid) = request.cluster_name.strip_prefix(&cluster_prefix) else { + return false; + }; + if !uuid7(cluster_uid) { + return false; + } + let device_prefix = format!("{}/clusterDevices/", request.cluster_name); + request.device_name.strip_prefix(&device_prefix).is_some_and(uuid7) +} + +fn uuid7(value: &str) -> bool { + Uuid::parse_str(value).is_ok_and(|uuid| { + uuid.get_version() == Some(Version::SortRand) && uuid.get_variant() == Variant::RFC4122 && uuid.to_string() == value + }) +} + +fn lower_hex_string(value: &str, length: usize) -> bool { + value.len() == length + && value + .bytes() + .all(|byte| byte.is_ascii_hexdigit() && !byte.is_ascii_uppercase()) +} + +fn version(value: &str) -> bool { + if value.is_empty() || value.len() > 64 { + return false; + } + let (core, prerelease) = value + .split_once('-') + .map_or((value, None), |(core, prerelease)| (core, Some(prerelease))); + let mut parts = core.split('.'); + let valid_core = (0..3).all(|_| { + parts + .next() + .is_some_and(|part| !part.is_empty() && part.bytes().all(|byte| byte.is_ascii_digit())) + }) && parts.next().is_none(); + valid_core + && prerelease.is_none_or(|part| { + !part.is_empty() + && part + .bytes() + .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'.' | b'-')) + }) +} + +fn build_feature(value: &str) -> bool { + value.len() <= 64 + && value.as_bytes().split_first().is_some_and(|(first, rest)| { + first.is_ascii_lowercase() + && rest + .iter() + .all(|byte| byte.is_ascii_lowercase() || byte.is_ascii_digit() || matches!(byte, b'-' | b'_')) + }) +} + +fn elapsed_millis(duration: Duration) -> u64 { + elapsed_millis_allow_zero(duration).max(1) +} + +fn elapsed_millis_allow_zero(duration: Duration) -> u64 { + u64::try_from(duration.as_millis()).unwrap_or(u64::MAX).min(MAX_SAFE_INTEGER) +} + +fn elapsed_micros(duration: Duration) -> u64 { + u64::try_from(duration.as_micros()).unwrap_or(u64::MAX).min(MAX_SAFE_INTEGER) +} + +fn unix_now() -> Result { + SystemTime::now() + .duration_since(UNIX_EPOCH) + .map_err(|_| ClientPerformanceError::InvalidRequest) + .and_then(|duration| i64::try_from(duration.as_secs()).map_err(|_| ClientPerformanceError::InvalidRequest)) +} + +fn timestamp(unix: i64) -> Result { + OffsetDateTime::from_unix_timestamp(unix) + .map_err(|_| ClientPerformanceError::InvalidRequest)? + .format(&Rfc3339) + .map_err(|_| ClientPerformanceError::Encoding) +} + +fn check_cancel(cancel: &CancellationToken) -> Result<(), ClientPerformanceError> { + if cancel.is_cancelled() { + Err(ClientPerformanceError::Cancelled) + } else { + Ok(()) + } +} + +fn hex_lower(bytes: &[u8]) -> String { + let mut encoded = String::with_capacity(bytes.len() * 2); + for byte in bytes { + use std::fmt::Write as _; + let _ = write!(encoded, "{byte:02x}"); + } + encoded +} + +fn map_create_error(error: std::io::Error) -> ClientPerformanceError { + if error.kind() == std::io::ErrorKind::AlreadyExists { + ClientPerformanceError::AlreadyExists + } else { + ClientPerformanceError::Io(error) + } +} + +fn map_publish_error(error: std::io::Error) -> ClientPerformanceError { + if error.kind() == std::io::ErrorKind::AlreadyExists { + ClientPerformanceError::AlreadyExists + } else { + ClientPerformanceError::Io(error) + } +} diff --git a/rustfs/src/connect/mod.rs b/rustfs/src/connect/mod.rs index 830502e20..73f4d3466 100644 --- a/rustfs/src/connect/mod.rs +++ b/rustfs/src/connect/mod.rs @@ -45,17 +45,21 @@ pub use client::{ClientError, ConnectClient, ConnectConfig}; pub use config::{HeartbeatConfig, HeartbeatConfigError, HeartbeatSchedule}; pub use credential_store::{CredentialStore, DeviceCredential}; pub use diagnostics::{ - CPU_PROFILE_CAPABILITY, CaptureMode, DRIVE_CAPABILITY, DRIVE_SCHEMA_VERSION, DiagnosticCollectionPolicy, DiagnosticReceipt, - DiagnosticScheduleError, DiagnosticScheduleRuntime, DiagnosticScheduleStatus, DriveDiagnosticResult, DriveMeasurement, - DriveOutcome, DrivePerformanceData, DrivePerformanceError, DrivePerformanceRequest, DriveProvenance, DriveReadMode, - DriveReasonCode, DriveTargetParameters, DriveTargetReasonCode, DriveTargetResult, DriveTargetUnits, LOGS_CAPABILITY, - LOGS_SCHEMA_VERSION, LocalDriveConsent, LocalLogConsent, LocalOtlpHeaders, LocalProfileConsent, LocalTelemetryConsent, - LocallyReviewedTraceArtifact, LogCaptureError, LogCaptureRequest, LogProvenance, MAX_OTLP_BODY_BYTES, MAX_PROFILE_DURATION, - MAX_SAFE_INTEGER, MAX_TELEMETRY_DURATION, MAX_TELEMETRY_RESULT_BYTES, MAX_TELEMETRY_SPANS, MEMORY_PROFILE_CAPABILITY, - ObservedTelemetrySpan, OperationSummary, OtlpBatch, OtlpForwardError, OtlpReceipt, PROFILE_SCHEMA_VERSION, - ProfileCaptureRequest, ProfileData, ProfileError, ProfileOutcome, ProfileProvenance, ProfileReasonCode, ProfileResult, - ProfileTool, ReceiptOutcome, RecordedTrace, ReplayedTrace, SavedDriveExport, SavedLogExport, SavedProfileExport, - SavedTelemetryExport, SignedDriveExport, SignedLogExport, SignedProfileExport, SignedTelemetryExport, + CLIENT_CAPABILITY, CLIENT_SCHEMA_VERSION, CPU_PROFILE_CAPABILITY, CaptureMode, ClientDiagnosticResult, ClientMeasurement, + ClientOperation, ClientOutcome, ClientPerformanceData, ClientPerformanceError, ClientPerformanceRequest, ClientProbe, + ClientProbeError, ClientProbeFuture, ClientProbeMeasurement, ClientProvenance, ClientReasonCode, ClientTargetParameters, + ClientTargetReasonCode, ClientTargetResult, ClientTargetUnits, DRIVE_CAPABILITY, DRIVE_SCHEMA_VERSION, + DiagnosticCollectionPolicy, DiagnosticReceipt, DiagnosticScheduleError, DiagnosticScheduleRuntime, DiagnosticScheduleStatus, + DriveDiagnosticResult, DriveMeasurement, DriveOutcome, DrivePerformanceData, DrivePerformanceError, DrivePerformanceRequest, + DriveProvenance, DriveReadMode, DriveReasonCode, DriveTargetParameters, DriveTargetReasonCode, DriveTargetResult, + DriveTargetUnits, HttpClientProbe, LOGS_CAPABILITY, LOGS_SCHEMA_VERSION, LocalClientConsent, LocalDriveConsent, + LocalLogConsent, LocalOtlpHeaders, LocalProfileConsent, LocalTelemetryConsent, LocallyReviewedTraceArtifact, LogCaptureError, + LogCaptureRequest, LogProvenance, MAX_OTLP_BODY_BYTES, MAX_PROFILE_DURATION, MAX_SAFE_INTEGER, MAX_TELEMETRY_DURATION, + MAX_TELEMETRY_RESULT_BYTES, MAX_TELEMETRY_SPANS, MEMORY_PROFILE_CAPABILITY, ObservedTelemetrySpan, OperationSummary, + OtlpBatch, OtlpForwardError, OtlpReceipt, PROFILE_SCHEMA_VERSION, ProfileCaptureRequest, ProfileData, ProfileError, + ProfileOutcome, ProfileProvenance, ProfileReasonCode, ProfileResult, ProfileTool, ReceiptOutcome, RecordedTrace, + ReplayedTrace, SavedClientExport, SavedDriveExport, SavedLogExport, SavedProfileExport, SavedTelemetryExport, + SignedClientExport, SignedDriveExport, SignedLogExport, SignedProfileExport, SignedTelemetryExport, TELEMETRY_OTLP_CAPABILITY, TELEMETRY_RECORD_CAPABILITY, TELEMETRY_REPLAY_CAPABILITY, TELEMETRY_SCHEMA_VERSION, THREAD_PROFILE_CAPABILITY, TelemetryArtifactConsent, TelemetryArtifactError, TelemetryArtifactRequest, TelemetryCoverage, TelemetryDiagnosticResult, TelemetryOperation, TelemetryOutcome, TelemetryProducerError, TelemetryProvenance, @@ -63,9 +67,10 @@ pub use diagnostics::{ TraceAnalysisError, TraceRecordCapture, TraceRecordCompletion, TraceRecordLimits, TraceReplayError, analyze_trace, capture_cpu_profile, capture_thread_profile, encode_signed_profile_export, encode_signed_telemetry_export, export_cpu_profile, export_logs, export_memory_profile, export_thread_profile, export_trace_otlp, export_trace_otlp_result, - measure_drive, record_diagnostic_result, record_trace, record_trace_bus, replay_trace, replay_trace_result, - run_local_environment_once, save_signed_drive_export, save_signed_log_export, save_signed_profile_export, - save_signed_telemetry_export, sign_drive_export, spawn_environment_schedule, validate_drive_limits, + measure_client, measure_drive, read_protected_client_credential, record_diagnostic_result, record_trace, record_trace_bus, + replay_trace, replay_trace_result, run_local_environment_once, save_signed_client_export, save_signed_drive_export, + save_signed_log_export, save_signed_profile_export, save_signed_telemetry_export, sign_client_export, sign_drive_export, + spawn_environment_schedule, validate_client_limits, validate_drive_limits, }; pub use diagnostics::{ LocalTopConsent, MAX_TOP_DURATION, MAX_TOP_EXPORT_VALIDITY, NetworkCounterSnapshot, SavedTopExport, SignedTopExport, diff --git a/rustfs/src/startup_entrypoint.rs b/rustfs/src/startup_entrypoint.rs index 78def041d..89ebda2eb 100644 --- a/rustfs/src/startup_entrypoint.rs +++ b/rustfs/src/startup_entrypoint.rs @@ -14,9 +14,10 @@ use crate::{ config::{ - CommandResult, Config, ConnectDrivePerformanceOpts, ConnectLicenseCommands, ConnectLicenseScopeOpts, ConnectLogsMode, - ConnectLogsOpts, ConnectProfileOpts, ConnectProfileTool, ConnectTelemetryArtifactOpts, ConnectTelemetryCommands, - ConnectThreadProfileScope, ConnectTopCommands, Opt, + CommandResult, Config, ConnectClientPerformanceOperation, ConnectClientPerformanceOpts, ConnectDrivePerformanceOpts, + ConnectLicenseCommands, ConnectLicenseScopeOpts, ConnectLogsMode, ConnectLogsOpts, ConnectProfileOpts, + ConnectProfileTool, ConnectTelemetryArtifactOpts, ConnectTelemetryCommands, ConnectThreadProfileScope, + ConnectTopCommands, Opt, }, startup_lifecycle::{StartupRuntimeLifecycle, run_startup_runtime_lifecycle}, startup_preflight::{StartupServerPreflightError, bootstrap_external_prefix_compat, init_startup_server_preflight}, @@ -136,6 +137,7 @@ async fn async_main() -> Result<()> { return Ok(()); } CommandResult::ConnectLicense(command) => return execute_connect_license(command), + CommandResult::ConnectClientPerformance(options) => return execute_connect_client_performance(options).await, CommandResult::ConnectDrivePerformance(options) => return execute_connect_drive_performance(options).await, CommandResult::ConnectProfile(options) => return execute_connect_profile(options).await, CommandResult::ConnectLogs(options) => return execute_connect_logs(options).await, @@ -606,6 +608,137 @@ async fn finish_top_capture( Ok(()) } +async fn execute_connect_client_performance(options: ConnectClientPerformanceOpts) -> Result<()> { + use crate::connect::{ + ClientOperation, ClientOutcome, ClientPerformanceRequest, ClientProvenance, HttpClientProbe, IdentityStore, + LocalClientConsent, measure_client, read_protected_client_credential, save_signed_client_export, sign_client_export, + validate_client_limits, + }; + use rand::{TryRng as _, rngs::SysRng}; + use zeroize::Zeroizing; + + let duration = Duration::from_millis(options.duration_millis); + validate_client_limits(duration, options.traffic_bytes).map_err(Error::other)?; + let key = IdentityStore::new(options.state_dir.join("identity")) + .load() + .map_err(Error::other)? + .ok_or_else(|| Error::other("connect client performance requires an enrolled device identity"))?; + let access_key = read_protected_client_credential(&options.access_key_file).map_err(Error::other)?; + let secret_key = read_protected_client_credential(&options.secret_key_file).map_err(Error::other)?; + let session_token = options + .session_token_file + .as_deref() + .map(read_protected_client_credential) + .transpose() + .map_err(Error::other)? + .unwrap_or_else(|| Zeroizing::new(String::new())); + let root_ca = if let Some(path) = options.ca_file.as_deref() { + const MAX_ROOT_CA_BYTES: u64 = 1_048_576; + let mut bytes = Vec::with_capacity(16 * 1024); + std::fs::File::open(path)? + .take(MAX_ROOT_CA_BYTES + 1) + .read_to_end(&mut bytes)?; + let max_bytes = usize::try_from(MAX_ROOT_CA_BYTES).map_err(Error::other)?; + if bytes.len() > max_bytes { + return Err(Error::other("connect client root CA exceeds the 1048576-byte limit")); + } + Some(bytes) + } else { + None + }; + let probe = HttpClientProbe::new( + &options.endpoint, + root_ca.as_deref(), + options.proxy.as_deref(), + access_key, + secret_key, + session_token, + duration, + ) + .map_err(Error::other)?; + let executable_sha256 = hash_current_executable()?; + let produced_at_unix = unix_now()?; + let mut nonce = [0_u8; 32]; + SysRng.try_fill_bytes(&mut nonce).map_err(Error::other)?; + let request = ClientPerformanceRequest { + organization_name: options.organization, + cluster_name: options.cluster, + device_name: options.device, + run_uid: options.run_uid, + artifact_uid: options.artifact_uid, + schema_version: options.schema_version, + capability: options.capability, + consent: LocalClientConsent { + consent_uid: options.consent_uid, + policy_revision: options.policy_revision, + expires_at_unix: options.consent_expires_at_unix, + confirmed: options.acknowledge_l1, + }, + produced_at_unix, + expires_at_unix: options.expires_at_unix, + nonce, + duration, + operation: match options.operation { + ConnectClientPerformanceOperation::Get => ClientOperation::GetObject, + ConnectClientPerformanceOperation::Put => ClientOperation::PutObject, + }, + traffic_bytes: options.traffic_bytes, + target_alias: options.target_alias, + provenance: ClientProvenance::new( + crate::version::build::COMMIT_HASH, + executable_sha256, + env!("CARGO_PKG_VERSION"), + enabled_build_features(), + ), + }; + let cancel = CancellationToken::new(); + let measurement = measure_client(&request, &probe, &cancel); + tokio::pin!(measurement); + let measurement = tokio::select! { + biased; + signal = tokio::signal::ctrl_c() => { + signal.map_err(Error::other)?; + cancel.cancel(); + measurement.await.map_err(Error::other)? + } + result = measurement.as_mut() => result.map_err(Error::other)?, + }; + let target_json = serde_json::to_string(&measurement.target).map_err(Error::other)?; + println!( + "tool=performance.client outcome={} reason={}", + measurement.result.outcome().as_str(), + measurement.result.reason_code().as_str() + ); + println!("target={target_json}"); + std::io::stdout().flush()?; + if measurement.result.outcome() != ClientOutcome::Succeeded { + return Err(Error::other(format!( + "client performance collection ended with {}", + measurement.result.outcome().as_str() + ))); + } + + let export = sign_client_export(&request, &measurement, &key, &cancel).map_err(Error::other)?; + let output = options.output; + let writer_cancel = cancel.clone(); + let mut writer = tokio::task::spawn_blocking(move || save_signed_client_export(&output, &export, &writer_cancel)); + let receipt = tokio::select! { + biased; + signal = tokio::signal::ctrl_c() => { + signal.map_err(Error::other)?; + cancel.cancel(); + writer.await.map_err(Error::other)?.map_err(Error::other)? + } + result = &mut writer => result.map_err(Error::other)?.map_err(Error::other)?, + }; + println!( + "artifact={} bytes={} sha256={}", + receipt.artifact_uid, receipt.archive_size_bytes, receipt.archive_sha256 + ); + println!("upload=not-performed"); + Ok(()) +} + async fn execute_connect_drive_performance(options: ConnectDrivePerformanceOpts) -> Result<()> { use crate::connect::{ DriveOutcome, DrivePerformanceRequest, DriveProvenance, IdentityStore, LocalDriveConsent, measure_drive, @@ -808,12 +941,12 @@ async fn execute_connect_profile(options: ConnectProfileOpts) -> Result<()> { fn hash_current_executable() -> Result { use sha2::{Digest as _, Sha256}; - const MAX_EXECUTABLE_BYTES: u64 = 1_073_741_824; + const MAX_EXECUTABLE_BYTES: u64 = 2_147_483_648; let path = std::env::current_exe().map_err(Error::other)?; let mut file = std::fs::File::open(path).map_err(Error::other)?; let metadata = file.metadata().map_err(Error::other)?; if !metadata.is_file() || metadata.len() == 0 || metadata.len() > MAX_EXECUTABLE_BYTES { - return Err(Error::other("current executable is outside the profile provenance limit")); + return Err(Error::other("current executable is outside the diagnostic provenance limit")); } let mut hasher = Sha256::new(); let mut buffer = [0_u8; 64 * 1024]; @@ -825,14 +958,14 @@ fn hash_current_executable() -> Result { } read_bytes = read_bytes .checked_add(u64::try_from(count).map_err(Error::other)?) - .ok_or_else(|| Error::other("current executable is outside the profile provenance limit"))?; + .ok_or_else(|| Error::other("current executable is outside the diagnostic provenance limit"))?; if read_bytes > MAX_EXECUTABLE_BYTES { - return Err(Error::other("current executable is outside the profile provenance limit")); + return Err(Error::other("current executable is outside the diagnostic provenance limit")); } hasher.update(&buffer[..count]); } if read_bytes != metadata.len() { - return Err(Error::other("current executable changed while hashing profile provenance")); + return Err(Error::other("current executable changed while hashing diagnostic provenance")); } Ok(hex_simd::encode_to_string(hasher.finalize(), hex_simd::AsciiCase::Lower)) } diff --git a/rustfs/tests/connect_perf_client.rs b/rustfs/tests/connect_perf_client.rs new file mode 100644 index 000000000..2c2152062 --- /dev/null +++ b/rustfs/tests/connect_perf_client.rs @@ -0,0 +1,584 @@ +// Copyright 2024 RustFS Team +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +use std::fs::{self, File}; +use std::io::{Cursor, Read as _}; +#[cfg(unix)] +use std::os::unix::fs::PermissionsExt as _; +use std::path::Path; +use std::process::Command; +use std::sync::atomic::{AtomicUsize, Ordering}; +use std::time::{Duration, SystemTime, UNIX_EPOCH}; + +use base64_simd::URL_SAFE_NO_PAD; +use p256::ecdsa::{Signature, VerifyingKey, signature::Verifier as _}; +use p256::pkcs8::DecodePublicKey as _; +use rustfs::connect::{ + CLIENT_CAPABILITY, ClientOperation, ClientOutcome, ClientPerformanceError, ClientPerformanceRequest, ClientProbe, + ClientProbeError, ClientProbeFuture, ClientProbeMeasurement, ClientProvenance, ClientReasonCode, ClientTargetReasonCode, + DeviceIdentity, HttpClientProbe, LocalClientConsent, measure_client, save_signed_client_export, sign_client_export, + validate_client_limits, +}; +use rustfs::embedded::{RustFSServerBuilder, find_available_port}; +use sha2::{Digest as _, Sha256}; +use tokio_util::sync::CancellationToken; +use zeroize::Zeroizing; + +static TEST_LOCK: tokio::sync::Mutex<()> = tokio::sync::Mutex::const_new(()); + +fn now() -> i64 { + SystemTime::now().duration_since(UNIX_EPOCH).expect("current time").as_secs() as i64 +} + +fn request(operation: ClientOperation) -> ClientPerformanceRequest { + let now = now(); + let organization = "organizations/019e3ae0-0000-7000-8000-000000000010"; + let cluster = format!("{organization}/clusters/019e3ae0-0000-7000-8000-000000000011"); + ClientPerformanceRequest { + organization_name: organization.to_owned(), + cluster_name: cluster.clone(), + device_name: format!("{cluster}/clusterDevices/019e3ae0-0000-7000-8000-000000000012"), + run_uid: "019e3ae0-0000-7000-8000-000000000013".to_owned(), + artifact_uid: "019e3ae0-0000-7000-8000-000000000014".to_owned(), + schema_version: 1, + capability: CLIENT_CAPABILITY.to_owned(), + consent: LocalClientConsent { + consent_uid: "019e3ae0-0000-7000-8000-000000000015".to_owned(), + policy_revision: 7, + expires_at_unix: now + 120, + confirmed: true, + }, + produced_at_unix: now, + expires_at_unix: now + 60, + nonce: [0x6b; 32], + duration: Duration::from_secs(1), + operation, + traffic_bytes: 65_536, + target_alias: "deployment-1".to_owned(), + provenance: ClientProvenance::new("c".repeat(40), "d".repeat(64), "1.0.0-rc.6", vec!["default".to_owned()]), + } +} + +struct SuccessfulProbe; + +impl ClientProbe for SuccessfulProbe { + fn probe<'a>(&'a self, request: &'a ClientPerformanceRequest, _cancel: &'a CancellationToken) -> ClientProbeFuture<'a> { + Box::pin(async move { + Ok(ClientProbeMeasurement { + transferred_bytes: request.traffic_bytes, + duration: Duration::from_millis(100), + latency: Duration::from_millis(7), + }) + }) + } +} + +struct ErrorProbe(ClientProbeError); + +impl ClientProbe for ErrorProbe { + fn probe<'a>(&'a self, _request: &'a ClientPerformanceRequest, _cancel: &'a CancellationToken) -> ClientProbeFuture<'a> { + Box::pin(async move { Err(self.0) }) + } +} + +struct PendingProbe; + +impl ClientProbe for PendingProbe { + fn probe<'a>(&'a self, _request: &'a ClientPerformanceRequest, _cancel: &'a CancellationToken) -> ClientProbeFuture<'a> { + Box::pin(std::future::pending()) + } +} + +struct CountingProbe(AtomicUsize); + +impl ClientProbe for CountingProbe { + fn probe<'a>(&'a self, request: &'a ClientPerformanceRequest, _cancel: &'a CancellationToken) -> ClientProbeFuture<'a> { + self.0.fetch_add(1, Ordering::Relaxed); + Box::pin(async move { + Ok(ClientProbeMeasurement { + transferred_bytes: request.traffic_bytes, + duration: Duration::from_millis(100), + latency: Duration::from_millis(1), + }) + }) + } +} + +#[tokio::test] +async fn typed_get_and_put_results_match_the_frozen_schema() { + let _guard = TEST_LOCK.lock().await; + for operation in [ClientOperation::GetObject, ClientOperation::PutObject] { + let request = request(operation); + let measurement = measure_client(&request, &SuccessfulProbe, &CancellationToken::new()) + .await + .expect("client measurement"); + assert_eq!(measurement.result.outcome(), ClientOutcome::Succeeded); + assert_eq!(measurement.result.reason_code(), ClientReasonCode::Complete); + let data = measurement.result.data().expect("aggregate data"); + assert_eq!(data.operation, operation); + assert_eq!(data.transferred_bytes, 65_536); + assert_eq!(data.completed_operations, 1); + assert_eq!(data.duration_millis, 100); + assert_eq!(data.error_count, 0); + assert_eq!(measurement.target.target_alias, "deployment-1"); + assert_eq!(measurement.target.parameters.operation, operation); + assert_eq!(measurement.target.parameters.requested_bytes, 65_536); + assert_eq!(measurement.target.parameters.concurrency, 1); + assert_eq!(measurement.target.latency_micros, Some(7_000)); + + let value = serde_json::to_value(&measurement.result).expect("result JSON"); + assert_eq!(value["toolId"], "performance.client"); + assert_eq!(value["capability"], "performance.client@1"); + assert_eq!(value["coverage"]["unit"], "WINDOW"); + assert_eq!(value["data"]["operation"], operation.as_str()); + assert!(value["data"].get("latencyMicros").is_none()); + } +} + +#[tokio::test] +async fn consent_and_budget_fail_before_transport() { + let _guard = TEST_LOCK.lock().await; + let probe = CountingProbe(AtomicUsize::new(0)); + + let mut unsupported_version = request(ClientOperation::PutObject); + unsupported_version.schema_version = 2; + assert!(matches!( + measure_client(&unsupported_version, &probe, &CancellationToken::new()).await, + Err(ClientPerformanceError::UnsupportedVersion) + )); + + let mut unsupported_capability = request(ClientOperation::PutObject); + unsupported_capability.capability = "performance.client@2".to_owned(); + assert!(matches!( + measure_client(&unsupported_capability, &probe, &CancellationToken::new()).await, + Err(ClientPerformanceError::UnsupportedCapability) + )); + + let mut no_consent = request(ClientOperation::PutObject); + no_consent.consent.confirmed = false; + assert!(matches!( + measure_client(&no_consent, &probe, &CancellationToken::new()).await, + Err(ClientPerformanceError::ConsentRequired) + )); + + let mut over_traffic = request(ClientOperation::PutObject); + over_traffic.traffic_bytes = 1_048_577; + assert!(matches!( + measure_client(&over_traffic, &probe, &CancellationToken::new()).await, + Err(ClientPerformanceError::LimitExceeded) + )); + + let mut invalid_version = request(ClientOperation::PutObject); + invalid_version.provenance = + ClientProvenance::new("a".repeat(40), "b".repeat(64), "release_candidate", vec!["default".to_owned()]); + assert!(matches!( + measure_client(&invalid_version, &probe, &CancellationToken::new()).await, + Err(ClientPerformanceError::InvalidRequest) + )); + + let mut invalid_feature = request(ClientOperation::PutObject); + invalid_feature.provenance = ClientProvenance::new("a".repeat(40), "b".repeat(64), "1.0.0-rc.6", vec!["Default".to_owned()]); + assert!(matches!( + measure_client(&invalid_feature, &probe, &CancellationToken::new()).await, + Err(ClientPerformanceError::InvalidRequest) + )); + assert_eq!(probe.0.load(Ordering::Relaxed), 0); + + assert!(validate_client_limits(Duration::from_secs(1), 1_048_576).is_ok()); + assert!(matches!( + validate_client_limits(Duration::from_secs(1), 1_048_577), + Err(ClientPerformanceError::LimitExceeded) + )); + assert!(matches!( + validate_client_limits(Duration::from_nanos(1), 1), + Err(ClientPerformanceError::LimitExceeded) + )); + assert!(matches!( + validate_client_limits(Duration::from_secs(30) + Duration::from_nanos(1), 1), + Err(ClientPerformanceError::LimitExceeded) + )); +} + +#[tokio::test] +async fn endpoint_proxy_permission_timeout_and_cancel_are_explicit() { + let _guard = TEST_LOCK.lock().await; + for (error, expected_reason, expected_result_reason) in [ + ( + ClientProbeError::EndpointUnavailable, + ClientTargetReasonCode::EndpointUnavailable, + ClientReasonCode::SourceUnavailable, + ), + ( + ClientProbeError::ProxyFailure, + ClientTargetReasonCode::ProxyFailure, + ClientReasonCode::CollectionFailed, + ), + ( + ClientProbeError::PermissionDenied, + ClientTargetReasonCode::PermissionDenied, + ClientReasonCode::PermissionDenied, + ), + ( + ClientProbeError::TimedOut, + ClientTargetReasonCode::TimedOut, + ClientReasonCode::CollectionFailed, + ), + ] { + let measurement = measure_client(&request(ClientOperation::PutObject), &ErrorProbe(error), &CancellationToken::new()) + .await + .expect("typed failure"); + assert_eq!(measurement.result.outcome(), ClientOutcome::Failed); + assert_eq!(measurement.result.reason_code(), expected_result_reason); + assert_eq!(measurement.target.reason_code, expected_reason); + assert!(measurement.result.data().is_none()); + } + + let cancel = CancellationToken::new(); + cancel.cancel(); + let measurement = measure_client(&request(ClientOperation::PutObject), &SuccessfulProbe, &cancel) + .await + .expect("typed cancellation"); + assert_eq!(measurement.result.outcome(), ClientOutcome::Cancelled); + assert_eq!(measurement.target.reason_code, ClientTargetReasonCode::Cancelled); +} + +#[tokio::test] +async fn deadline_and_in_flight_cancellation_stop_a_stalled_probe() { + let _guard = TEST_LOCK.lock().await; + let mut timed = request(ClientOperation::GetObject); + timed.duration = Duration::from_millis(20); + timed.traffic_bytes = 1_024; + let measurement = measure_client(&timed, &PendingProbe, &CancellationToken::new()) + .await + .expect("typed timeout"); + assert_eq!(measurement.result.outcome(), ClientOutcome::Failed); + assert_eq!(measurement.target.reason_code, ClientTargetReasonCode::TimedOut); + + let cancel = CancellationToken::new(); + let cancellation = cancel.clone(); + tokio::spawn(async move { + tokio::task::yield_now().await; + cancellation.cancel(); + }); + let measurement = measure_client(&request(ClientOperation::PutObject), &PendingProbe, &cancel) + .await + .expect("typed in-flight cancellation"); + assert_eq!(measurement.result.outcome(), ClientOutcome::Cancelled); + assert_eq!(measurement.target.reason_code, ClientTargetReasonCode::Cancelled); +} + +#[tokio::test] +async fn only_one_client_collector_can_run_at_a_time() { + let _guard = TEST_LOCK.lock().await; + let first_cancel = CancellationToken::new(); + let second_cancel = CancellationToken::new(); + let first_request = request(ClientOperation::GetObject); + let second_request = request(ClientOperation::PutObject); + let first = measure_client(&first_request, &PendingProbe, &first_cancel); + let second = async { + tokio::task::yield_now().await; + measure_client(&second_request, &PendingProbe, &second_cancel).await + }; + let cancellation = async { + tokio::time::sleep(Duration::from_millis(20)).await; + first_cancel.cancel(); + }; + let (first, second, ()) = tokio::join!(first, second, cancellation); + assert_eq!(first.expect("typed cancellation").result.outcome(), ClientOutcome::Cancelled); + assert!(matches!(second, Err(ClientPerformanceError::Busy))); +} + +#[tokio::test] +async fn configured_proxy_and_direct_endpoint_failures_are_distinct() { + let _guard = TEST_LOCK.lock().await; + let proxy_probe = HttpClientProbe::new( + "http://127.0.0.1:1", + None, + Some("http://127.0.0.1:9"), + Zeroizing::new("access".to_owned()), + Zeroizing::new("secret".to_owned()), + Zeroizing::new(String::new()), + Duration::from_millis(100), + ) + .expect("proxy probe"); + let measurement = measure_client(&request(ClientOperation::GetObject), &proxy_probe, &CancellationToken::new()) + .await + .expect("typed proxy failure"); + assert_eq!(measurement.target.reason_code, ClientTargetReasonCode::ProxyFailure); + + let direct_probe = HttpClientProbe::new( + "http://127.0.0.1:9", + None, + None, + Zeroizing::new("access".to_owned()), + Zeroizing::new("secret".to_owned()), + Zeroizing::new(String::new()), + Duration::from_millis(100), + ) + .expect("direct probe"); + let measurement = measure_client(&request(ClientOperation::GetObject), &direct_probe, &CancellationToken::new()) + .await + .expect("typed endpoint failure"); + assert_eq!(measurement.target.reason_code, ClientTargetReasonCode::EndpointUnavailable); +} + +#[tokio::test] +async fn oversized_response_is_rejected_without_buffering_it() { + use tokio::io::{AsyncReadExt as _, AsyncWriteExt as _}; + + let _guard = TEST_LOCK.lock().await; + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.expect("response listener"); + let address = listener.local_addr().expect("listener address"); + let server = tokio::spawn(async move { + let (mut socket, _) = listener.accept().await.expect("client connection"); + let mut request = vec![0_u8; 16 * 1024]; + let _ = socket.read(&mut request).await.expect("request headers"); + socket + .write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 65537\r\nConnection: close\r\n\r\n") + .await + .expect("response headers"); + }); + let probe = HttpClientProbe::new( + &format!("http://{address}"), + None, + None, + Zeroizing::new("access".to_owned()), + Zeroizing::new("secret".to_owned()), + Zeroizing::new(String::new()), + Duration::from_secs(1), + ) + .expect("client probe"); + let measurement = measure_client(&request(ClientOperation::GetObject), &probe, &CancellationToken::new()) + .await + .expect("typed protocol failure"); + assert_eq!(measurement.target.reason_code, ClientTargetReasonCode::ProtocolFailure); + server.await.expect("response server"); +} + +#[tokio::test] +async fn signed_result_is_saved_without_overwrite() { + let _guard = TEST_LOCK.lock().await; + let request = request(ClientOperation::PutObject); + let measurement = measure_client(&request, &SuccessfulProbe, &CancellationToken::new()) + .await + .expect("client measurement"); + let export = sign_client_export(&request, &measurement, &DeviceIdentity::generate(), &CancellationToken::new()) + .expect("signed export"); + let directory = tempfile::tempdir().expect("output directory"); + let output = directory.path().join("client.zip"); + let saved = save_signed_client_export(&output, &export, &CancellationToken::new()).expect("save export"); + assert_eq!(saved.archive_sha256, export.archive_sha256); + assert!(matches!( + save_signed_client_export(&output, &export, &CancellationToken::new()), + Err(ClientPerformanceError::AlreadyExists) + )); + + let mut tampered = export.clone(); + tampered.archive_bytes[0] ^= 0xff; + assert!(matches!( + save_signed_client_export(&directory.path().join("tampered.zip"), &tampered, &CancellationToken::new()), + Err(ClientPerformanceError::InvalidRequest) + )); + + let mut oversized = export.clone(); + oversized.archive_bytes = vec![0; 524_289]; + oversized.archive_sha256 = hex_lower(&Sha256::digest(&oversized.archive_bytes)); + assert!(matches!( + save_signed_client_export(&directory.path().join("oversized.zip"), &oversized, &CancellationToken::new()), + Err(ClientPerformanceError::LimitExceeded) + )); + + let cancelled = CancellationToken::new(); + cancelled.cancel(); + assert!(matches!( + save_signed_client_export(&directory.path().join("cancelled.zip"), &export, &cancelled), + Err(ClientPerformanceError::Cancelled) + )); +} + +#[tokio::test] +async fn real_rustfs_endpoint_supports_bounded_get_and_put() { + let _guard = TEST_LOCK.lock().await; + let port = match find_available_port() { + Ok(port) => port, + Err(err) if err.kind() == std::io::ErrorKind::PermissionDenied => return, + Err(err) => panic!("find free port: {err}"), + }; + let server = RustFSServerBuilder::new() + .address(format!("127.0.0.1:{port}")) + .access_key("client-perf-access") + .secret_key("client-perf-secret") + .build() + .await + .expect("start embedded server"); + let probe = HttpClientProbe::new( + &server.endpoint(), + None, + None, + Zeroizing::new(server.access_key().to_owned()), + Zeroizing::new(server.secret_key().to_owned()), + Zeroizing::new(String::new()), + Duration::from_secs(2), + ) + .expect("client probe"); + + for operation in [ClientOperation::GetObject, ClientOperation::PutObject] { + let mut request = request(operation); + request.duration = Duration::from_secs(2); + let measurement = measure_client(&request, &probe, &CancellationToken::new()) + .await + .expect("real client measurement"); + assert_eq!(measurement.result.outcome(), ClientOutcome::Succeeded); + assert_eq!(measurement.result.data().expect("data").transferred_bytes, 65_536); + assert_eq!(measurement.target.completed_operations, 1); + } +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn production_cli_writes_a_verifiable_export_with_exact_binary_provenance() { + let _guard = TEST_LOCK.lock().await; + let port = match find_available_port() { + Ok(port) => port, + Err(err) if err.kind() == std::io::ErrorKind::PermissionDenied => return, + Err(err) => panic!("find free port: {err}"), + }; + let server = RustFSServerBuilder::new() + .address(format!("127.0.0.1:{port}")) + .access_key("client-perf-access") + .secret_key("client-perf-secret") + .build() + .await + .expect("start embedded server"); + let temp = tempfile::tempdir().expect("CLI tempdir"); + let state = temp.path().join("state"); + let output = temp.path().join("client.zip"); + let access_key_file = temp.path().join("access-key"); + let secret_key_file = temp.path().join("secret-key"); + write_credential(&access_key_file, server.access_key()); + write_credential(&secret_key_file, server.secret_key()); + let identity = rustfs::connect::IdentityStore::new(state.join("identity")) + .load_or_create() + .expect("enrolled identity"); + + let current = now(); + let organization = "organizations/019e3ae0-0000-7000-8000-000000000010"; + let cluster = format!("{organization}/clusters/019e3ae0-0000-7000-8000-000000000011"); + let mut command = Command::new(env!("CARGO_BIN_EXE_rustfs")); + command + .args(["connect", "performance", "client", "--state-dir"]) + .arg(&state) + .args(["--endpoint", &server.endpoint(), "--access-key-file"]) + .arg(&access_key_file) + .arg("--secret-key-file") + .arg(&secret_key_file) + .arg("--output") + .arg(&output) + .args(["--organization", organization, "--cluster", &cluster, "--device"]) + .arg(format!("{cluster}/clusterDevices/019e3ae0-0000-7000-8000-000000000012")) + .args([ + "--run-uid", + "019e3ae0-0000-7000-8000-000000000013", + "--artifact-uid", + "019e3ae0-0000-7000-8000-000000000014", + "--consent-uid", + "019e3ae0-0000-7000-8000-000000000015", + "--policy-revision", + "7", + "--consent-expires-at", + &(current + 120).to_string(), + "--expires-at", + &(current + 60).to_string(), + "--operation", + "get", + "--traffic-bytes", + "65536", + "--duration-millis", + "1000", + "--acknowledge-l1", + ]); + let result = tokio::task::spawn_blocking(move || command.output()) + .await + .expect("CLI task") + .expect("run production rustfs binary"); + + assert!(result.status.success(), "stderr: {}", String::from_utf8_lossy(&result.stderr)); + let stdout = String::from_utf8(result.stdout).expect("UTF-8 stdout"); + assert!(stdout.contains("tool=performance.client outcome=SUCCEEDED reason=COMPLETE\n")); + assert!(stdout.contains("upload=not-performed\n")); + #[cfg(unix)] + assert_eq!(fs::metadata(&output).expect("output metadata").permissions().mode(), 0o100600); + + let archive_bytes = fs::read(&output).expect("saved archive"); + let mut archive = zip::ZipArchive::new(Cursor::new(archive_bytes.as_slice())).expect("signed archive"); + let envelope_bytes = read_archive_member(&mut archive, "envelope.json"); + let signature_bytes = read_archive_member(&mut archive, "envelope.sig"); + let result_bytes = read_archive_member(&mut archive, "result.json"); + let envelope: serde_json::Value = serde_json::from_slice(&envelope_bytes).expect("envelope JSON"); + let signed_result: serde_json::Value = serde_json::from_slice(&result_bytes).expect("result JSON"); + assert_eq!(envelope["classification"], "L1"); + assert_eq!(envelope["payload"]["sha256"], hex_lower(&Sha256::digest(&result_bytes))); + assert_eq!(signed_result["data"]["operation"], "GET_OBJECT"); + assert_eq!(signed_result["data"]["transferredBytes"], 65_536); + assert_eq!(signed_result["provenance"]["sourceCommit"], rustfs::version::build::COMMIT_HASH); + assert_eq!( + signed_result["provenance"]["executableSha256"], + sha256_file(Path::new(env!("CARGO_BIN_EXE_rustfs"))) + ); + + let signature_document: serde_json::Value = serde_json::from_slice(&signature_bytes).expect("signature JSON"); + let raw = URL_SAFE_NO_PAD + .decode_to_vec(signature_document["value"].as_str().expect("signature value")) + .expect("base64url signature"); + let signature = Signature::from_slice(&raw).expect("P-256 signature"); + let mut signed = b"rustfs-diagnostic-envelope-v1\0".to_vec(); + signed.extend_from_slice(&envelope_bytes); + VerifyingKey::from_public_key_der(&identity.public_key_der()) + .expect("public key") + .verify(&signed, &signature) + .expect("valid ES256 signature"); +} + +fn write_credential(path: &Path, value: &str) { + fs::write(path, value).expect("write credential"); + #[cfg(unix)] + fs::set_permissions(path, fs::Permissions::from_mode(0o600)).expect("protect credential"); +} + +fn read_archive_member(archive: &mut zip::ZipArchive>, name: &str) -> Vec { + let mut bytes = Vec::new(); + archive + .by_name(name) + .expect("archive member") + .read_to_end(&mut bytes) + .expect("read archive member"); + bytes +} + +fn sha256_file(path: &Path) -> String { + let mut file = File::open(path).expect("open exact binary"); + let mut digest = Sha256::new(); + let mut buffer = [0_u8; 64 * 1024]; + loop { + let read = file.read(&mut buffer).expect("hash exact binary"); + if read == 0 { + break; + } + digest.update(&buffer[..read]); + } + hex_lower(&digest.finalize()) +} + +fn hex_lower(bytes: &[u8]) -> String { + hex_simd::encode_to_string(bytes, hex_simd::AsciiCase::Lower) +}