diff --git a/rustfs/src/config/cli.rs b/rustfs/src/config/cli.rs index 0efaceba9..6619b0fde 100644 --- a/rustfs/src/config/cli.rs +++ b/rustfs/src/config/cli.rs @@ -129,6 +129,8 @@ pub enum ConnectCommands { License(ConnectLicenseOpts), /// Deliver a reviewed signed artifact through the customer relay (Unix only) Relay(Box), + /// Upload an explicitly selected support report with the registered device identity + Report(ConnectReportOpts), /// Read the persisted deployment inventory and collect an approved environment summary Inventory(ConnectInventoryOpts), /// Run an explicitly approved, bounded local performance measurement @@ -198,6 +200,44 @@ pub struct ConnectRelayOpts { pub acknowledge_reviewed: bool, } +/// Support report operations. +#[derive(Args, Clone)] +pub struct ConnectReportOpts { + #[command(subcommand)] + pub command: ConnectReportCommands, +} + +/// Device-authenticated support report operations. +#[derive(Subcommand, Clone)] +pub enum ConnectReportCommands { + /// Upload one bounded archive through a short-lived object-store authorization + Upload(ConnectReportUploadOpts), +} + +/// `connect report upload` options. +#[derive(Args, Clone)] +pub struct ConnectReportUploadOpts { + /// Connect agent API HTTPS base URL + #[arg(long, value_parser = NonEmptyStringValueParser::new())] + pub endpoint: String, + + /// PEM root CA file used for Connect and its authorized object store + #[arg(long = "ca-file")] + pub ca_file: PathBuf, + + /// Directory containing the registered Connect device identity + #[arg(long = "state-dir")] + pub state_dir: PathBuf, + + /// Explicitly selected support report archive + #[arg(long)] + pub archive: PathBuf, + + /// Bounded timeout for each object upload request + #[arg(long = "upload-timeout-seconds", default_value_t = 600, value_parser = clap::value_parser!(u64).range(1..=900))] + pub upload_timeout_seconds: u64, +} + #[derive(Args, Clone)] pub struct ConnectInventoryOpts { #[command(subcommand)] @@ -1375,6 +1415,8 @@ pub enum CommandResult { ConnectLicense(ConnectLicenseCommands), /// Customer-operated relay of one reviewed signed artifact ConnectRelay(Box), + /// Device-authenticated upload of one support report archive + ConnectReportUpload(ConnectReportUploadOpts), /// Explicit local Connect environment inventory command ConnectEnvironmentInventory(ConnectEnvironmentInventoryOpts), /// Consent-bound local Connect drive performance export @@ -1434,7 +1476,7 @@ pub fn default_server_opts() -> ServerOpts { mod tests { use super::{ Cli, Commands, ConnectCommands, ConnectInventoryCommands, ConnectLicenseCommands, ConnectRelayMaterialKind, - InspectCommands, preprocess_args_for_legacy, + ConnectReportCommands, InspectCommands, preprocess_args_for_legacy, }; use crate::version; use clap::error::ErrorKind; @@ -1605,6 +1647,59 @@ mod tests { assert!(options.acknowledge_reviewed); } + #[test] + fn connect_report_upload_accepts_only_explicit_archive_and_transport_paths() { + let cli = Cli::try_parse_from([ + "rustfs", + "connect", + "report", + "upload", + "--endpoint", + "https://connect.example/agent/", + "--ca-file", + "/etc/rustfs/connect-ca.pem", + "--state-dir", + "/var/lib/rustfs/connect", + "--archive", + "/var/lib/rustfs/reports/support.tar.zst", + ]) + .expect("report upload arguments should parse"); + + let Some(Commands::Connect(connect)) = cli.command else { + panic!("connect command expected"); + }; + let ConnectCommands::Report(report) = connect.command else { + panic!("connect report command expected"); + }; + let ConnectReportCommands::Upload(upload) = report.command; + assert_eq!(upload.endpoint, "https://connect.example/agent/"); + assert_eq!(upload.archive, std::path::Path::new("/var/lib/rustfs/reports/support.tar.zst")); + assert_eq!(upload.upload_timeout_seconds, 600); + } + + #[test] + fn connect_report_upload_rejects_unbounded_timeout() { + let error = Cli::try_parse_from([ + "rustfs", + "connect", + "report", + "upload", + "--endpoint", + "https://connect.example/agent/", + "--ca-file", + "/etc/rustfs/connect-ca.pem", + "--state-dir", + "/var/lib/rustfs/connect", + "--archive", + "/var/lib/rustfs/reports/support.tar.zst", + "--upload-timeout-seconds", + "901", + ]) + .err() + .expect("unbounded upload timeout must fail"); + assert_eq!(error.kind(), ErrorKind::ValueValidation); + } + #[test] fn connect_register_has_no_token_value_or_environment_option() { for forbidden in ["--token", "--registration-token", "--token-env"] { diff --git a/rustfs/src/config/mod.rs b/rustfs/src/config/mod.rs index cae55f7f5..8f9ee3735 100644 --- a/rustfs/src/config/mod.rs +++ b/rustfs/src/config/mod.rs @@ -61,6 +61,7 @@ pub use cli::{ConnectLogsMode, ConnectLogsOpts}; pub use cli::{ConnectObjectPerformanceOperation, ConnectObjectPerformanceOpts}; pub use cli::{ConnectProfileOpts, ConnectProfileTool, ConnectThreadProfileScope}; pub use cli::{ConnectRelayMaterialKind, ConnectRelayOpts}; +pub use cli::{ConnectReportCommands, ConnectReportOpts, ConnectReportUploadOpts}; pub use cli::{ ConnectTelemetryArtifactOpts, ConnectTelemetryCommands, ConnectTelemetryOtlpOpts, ConnectTelemetryRecordOpts, ConnectTelemetryReplayOpts, diff --git a/rustfs/src/config/opt.rs b/rustfs/src/config/opt.rs index 09cf6728d..88f11ab69 100644 --- a/rustfs/src/config/opt.rs +++ b/rustfs/src/config/opt.rs @@ -19,8 +19,8 @@ use super::Config; use super::cli::{ - Cli, CommandResult, Commands, ConnectCommands, ConnectInventoryCommands, ConnectPerformanceCommands, ServerOpts, - default_server_opts, preprocess_args_for_legacy, + Cli, CommandResult, Commands, ConnectCommands, ConnectInventoryCommands, ConnectPerformanceCommands, ConnectReportCommands, + ServerOpts, default_server_opts, preprocess_args_for_legacy, }; use crate::apply_external_env_compat; use CommandResult::Server; @@ -144,6 +144,9 @@ impl Opt { ConnectCommands::Register(opts) => Ok(CommandResult::ConnectRegister(opts)), ConnectCommands::License(opts) => Ok(CommandResult::ConnectLicense(opts.command)), ConnectCommands::Relay(opts) => Ok(CommandResult::ConnectRelay(opts)), + ConnectCommands::Report(opts) => match opts.command { + ConnectReportCommands::Upload(opts) => Ok(CommandResult::ConnectReportUpload(opts)), + }, ConnectCommands::Inventory(opts) => match opts.command { ConnectInventoryCommands::Environment(opts) => Ok(CommandResult::ConnectEnvironmentInventory(opts)), }, diff --git a/rustfs/src/connect/mod.rs b/rustfs/src/connect/mod.rs index d58871d30..3f3ac85a5 100644 --- a/rustfs/src/connect/mod.rs +++ b/rustfs/src/connect/mod.rs @@ -40,6 +40,7 @@ pub mod offline; pub mod registration; pub mod registration_bootstrap; pub mod relay; +pub mod report_upload; pub mod runtime; mod telemetry; @@ -136,4 +137,5 @@ pub use relay::{ RelayReview, TrustedReceiptSigner, 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/report_upload.rs b/rustfs/src/connect/report_upload.rs new file mode 100644 index 000000000..71cf27f3c --- /dev/null +++ b/rustfs/src/connect/report_upload.rs @@ -0,0 +1,660 @@ +// 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. + +//! Device-authenticated upload of bounded support bundle archives. + +use std::collections::{HashMap, HashSet}; +use std::io::SeekFrom; +use std::path::Path; +use std::time::Duration; + +use chrono::{DateTime, Utc}; +use reqwest::header::{self, HeaderMap, HeaderName, HeaderValue}; +use reqwest::{Client, StatusCode, Url}; +use serde::{Deserialize, Serialize}; +use sha2::{Digest as _, Sha256}; +use tokio::fs::File; +use tokio::io::{AsyncReadExt as _, AsyncSeekExt as _}; +use tokio_util::io::ReaderStream; +use tokio_util::sync::CancellationToken; +use uuid::Uuid; + +use super::client::{TransportFailure, classify_transport_failure}; +use super::config::HeartbeatConfig; +use super::telemetry::{TelemetryDelivery, TelemetryError, TelemetryTransport, is_exact_utc_seconds}; + +const PROTOCOL_VERSION: &str = "v1"; +const CONTENT_TYPE: &str = "application/octet-stream"; +const MAX_ATTEMPTS: usize = 3; +const MAX_UPLOAD_TIMEOUT: Duration = Duration::from_secs(15 * 60); +const UPLOAD_BUFFER_BYTES: usize = 64 * 1024; + +/// Maximum archive size accepted by the Connect agent API. +pub const MAX_SUPPORT_BUNDLE_BYTES: u64 = 256 * 1024 * 1024; + +/// Verified outcome returned after Connect pins the uploaded object version. +#[derive(Clone, Debug, Eq, PartialEq, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct ReportUploadReceipt { + pub name: String, + pub uid: String, + pub state: String, + pub declared_size_bytes: u64, + pub declared_sha256: String, +} + +/// Uploads one local archive through the registered device identity. +pub struct ReportUploadClient { + transport: TelemetryTransport, + upload_client: Client, + initial_backoff: Duration, + max_backoff: Duration, + proxy_configured: bool, +} + +impl ReportUploadClient { + /// Reuses the Connect mTLS identity, root bundle, explicit proxy, and + /// disabled redirect/environment-proxy policy. + pub fn new(config: HeartbeatConfig, upload_timeout: Duration) -> Result { + if upload_timeout.is_zero() || upload_timeout > MAX_UPLOAD_TIMEOUT { + return Err(ReportUploadError::UploadTimeout); + } + let initial_backoff = config.schedule.initial_backoff; + let max_backoff = config.schedule.max_backoff; + let proxy_configured = config.proxy.is_some(); + let transport = TelemetryTransport::new(config).map_err(transport_error)?; + let upload_client = transport.presigned_client(upload_timeout).map_err(transport_error)?; + Ok(Self { + transport, + upload_client, + initial_backoff, + max_backoff, + proxy_configured, + }) + } + + /// Hashes the exact opened file, reserves one object, retries interrupted + /// single-object PUTs with fresh short-lived authorization, and completes + /// the reservation only after the object store accepts the archive. + pub async fn upload( + &self, + archive: &Path, + cancellation: &CancellationToken, + ) -> Result { + let prepared = prepare_archive(archive, cancellation).await?; + let request_id = Uuid::new_v4().to_string(); + let bundle_uid = Uuid::now_v7().to_string(); + let reserve = ReserveRequest { + protocol_version: PROTOCOL_VERSION, + request_id: &request_id, + bundle_uid: &bundle_uid, + content_type: CONTENT_TYPE, + declared_size_bytes: prepared.size, + declared_sha256: &prepared.sha256, + }; + + let mut backoff = self.initial_backoff; + let mut expected_name = None; + for attempt in 0..MAX_ATTEMPTS { + let (cluster_name, body) = self + .post_control("supportBundles", &reserve, StatusCode::CREATED, cancellation) + .await?; + let name = format!("{cluster_name}/supportBundles/{bundle_uid}"); + if expected_name.as_ref().is_some_and(|expected| expected != &name) { + return Err(ReportUploadError::Response); + } + expected_name = Some(name.clone()); + let reservation = decode_reservation(&body, &name, &bundle_uid, &prepared)?; + + match self.put(&prepared.file, &reservation.authorization, cancellation).await? { + UploadDelivery::Accepted | UploadDelivery::AlreadyPresent => break, + UploadDelivery::Retry { retry_after } if attempt + 1 < MAX_ATTEMPTS => { + let delay = retry_after.unwrap_or(backoff).clamp(self.initial_backoff, self.max_backoff); + sleep_or_cancel(cancellation, delay).await?; + backoff = backoff.saturating_mul(2).min(self.max_backoff); + } + UploadDelivery::Retry { .. } => return Err(ReportUploadError::RetryExhausted), + } + } + + let complete_request_id = Uuid::new_v4().to_string(); + let complete = CompleteRequest { + protocol_version: PROTOCOL_VERSION, + request_id: &complete_request_id, + }; + let path = format!("supportBundles/{bundle_uid}:completeUpload"); + let (cluster_name, body) = self.post_control(&path, &complete, StatusCode::OK, cancellation).await?; + let expected_name = expected_name.ok_or(ReportUploadError::Response)?; + if expected_name != format!("{cluster_name}/supportBundles/{bundle_uid}") { + return Err(ReportUploadError::Response); + } + decode_receipt(&body, &expected_name, &bundle_uid, &prepared) + } + + async fn post_control( + &self, + operation: &str, + body: &T, + expected_status: StatusCode, + cancellation: &CancellationToken, + ) -> Result<(String, Vec), ReportUploadError> { + let mut backoff = self.initial_backoff; + for attempt in 0..MAX_ATTEMPTS { + let delivery = tokio::select! { + biased; + () = cancellation.cancelled() => return Err(ReportUploadError::Cancelled), + result = self.transport.post_expect(operation, body, expected_status) => result.map_err(transport_error)?, + }; + match delivery { + TelemetryDelivery::Accepted { cluster_name, body } => return Ok((cluster_name, body)), + TelemetryDelivery::Retry { retry_after } if attempt + 1 < MAX_ATTEMPTS => { + let delay = retry_after.unwrap_or(backoff).clamp(self.initial_backoff, self.max_backoff); + sleep_or_cancel(cancellation, delay).await?; + backoff = backoff.saturating_mul(2).min(self.max_backoff); + } + TelemetryDelivery::Retry { .. } => return Err(ReportUploadError::RetryExhausted), + TelemetryDelivery::AuthenticationStopped { status, .. } => { + return Err(ReportUploadError::AuthenticationStopped { status }); + } + TelemetryDelivery::Rejected { status, .. } => { + return Err(ReportUploadError::ControlRejected { status }); + } + } + } + Err(ReportUploadError::RetryExhausted) + } + + async fn put( + &self, + file: &File, + authorization: &UploadAuthorization, + cancellation: &CancellationToken, + ) -> Result { + let mut source = file.try_clone().await.map_err(ReportUploadError::ArchiveRead)?; + source + .seek(SeekFrom::Start(0)) + .await + .map_err(ReportUploadError::ArchiveRead)?; + let body = reqwest::Body::wrap_stream(ReaderStream::with_capacity(source, UPLOAD_BUFFER_BYTES)); + let response = tokio::select! { + biased; + () = cancellation.cancelled() => return Err(ReportUploadError::Cancelled), + result = self.upload_client.put(authorization.url.clone()).headers(authorization.headers.clone()).body(body).send() => result, + }; + let response = match response { + Ok(response) => response, + Err(error) if error.is_body() => return Ok(UploadDelivery::Retry { retry_after: None }), + Err(error) if error.is_timeout() || error.is_connect() || error.is_request() => { + if let Some(failure) = classify_transport_failure(&error, self.proxy_configured) { + return Err(match failure { + TransportFailure::ProxyAuthentication => ReportUploadError::ProxyAuthentication, + TransportFailure::ProxyRejected => ReportUploadError::ProxyRejected, + TransportFailure::TlsPeer => ReportUploadError::TlsPeer, + }); + } + return Ok(UploadDelivery::Retry { retry_after: None }); + } + Err(_) => return Err(ReportUploadError::UploadTransport), + }; + let status = response.status(); + if status.is_success() { + return Ok(UploadDelivery::Accepted); + } + if status == StatusCode::PRECONDITION_FAILED { + // A lost successful PUT response can make the create-only replay + // fail. Connect still verifies size and digest before completion. + return Ok(UploadDelivery::AlreadyPresent); + } + if status == StatusCode::TOO_MANY_REQUESTS || status == StatusCode::REQUEST_TIMEOUT || status.is_server_error() { + return Ok(UploadDelivery::Retry { + retry_after: retry_after(response.headers(), Utc::now(), self.max_backoff), + }); + } + Err(ReportUploadError::UploadRejected { status: status.as_u16() }) + } +} + +struct PreparedArchive { + file: File, + size: u64, + sha256: String, + checksum_base64: String, +} + +async fn prepare_archive(path: &Path, cancellation: &CancellationToken) -> Result { + let mut file = File::open(path).await.map_err(ReportUploadError::ArchiveOpen)?; + let metadata = file.metadata().await.map_err(ReportUploadError::ArchiveRead)?; + if !metadata.is_file() { + return Err(ReportUploadError::ArchiveType); + } + if metadata.len() == 0 || metadata.len() > MAX_SUPPORT_BUNDLE_BYTES { + return Err(ReportUploadError::ArchiveSize); + } + + let mut hasher = Sha256::new(); + let mut size = 0u64; + let mut buffer = vec![0u8; UPLOAD_BUFFER_BYTES]; + loop { + let read = tokio::select! { + biased; + () = cancellation.cancelled() => return Err(ReportUploadError::Cancelled), + result = file.read(&mut buffer) => result.map_err(ReportUploadError::ArchiveRead)?, + }; + if read == 0 { + break; + } + size = size.checked_add(read as u64).ok_or(ReportUploadError::ArchiveSize)?; + if size > MAX_SUPPORT_BUNDLE_BYTES { + return Err(ReportUploadError::ArchiveSize); + } + hasher.update(&buffer[..read]); + } + if size != metadata.len() { + return Err(ReportUploadError::ArchiveChanged); + } + let current = file.metadata().await.map_err(ReportUploadError::ArchiveRead)?; + if current.len() != size { + return Err(ReportUploadError::ArchiveChanged); + } + let digest = hasher.finalize(); + Ok(PreparedArchive { + file, + size, + sha256: faster_hex::hex_string(&digest), + checksum_base64: base64_simd::STANDARD.encode_to_string(&digest), + }) +} + +#[derive(Serialize)] +#[serde(rename_all = "camelCase")] +struct ReserveRequest<'a> { + protocol_version: &'static str, + request_id: &'a str, + bundle_uid: &'a str, + content_type: &'static str, + declared_size_bytes: u64, + declared_sha256: &'a str, +} + +#[derive(Serialize)] +#[serde(rename_all = "camelCase")] +struct CompleteRequest<'a> { + protocol_version: &'static str, + request_id: &'a str, +} + +#[derive(Deserialize)] +#[serde(rename_all = "camelCase")] +struct ReservationResponse { + support_bundle: BundleResource, + upload_authorization: RawUploadAuthorization, +} + +#[derive(Deserialize)] +#[serde(rename_all = "camelCase")] +struct RawUploadAuthorization { + method: String, + url: String, + headers: HashMap, + expire_time: String, +} + +#[derive(Deserialize)] +#[serde(rename_all = "camelCase")] +struct BundleResource { + name: String, + uid: String, + state: String, + declared_size_bytes: u64, + declared_sha256: String, + expire_time: String, + create_time: String, + update_time: String, +} + +struct Reservation { + authorization: UploadAuthorization, +} + +struct UploadAuthorization { + url: Url, + headers: HeaderMap, +} + +enum UploadDelivery { + Accepted, + AlreadyPresent, + Retry { retry_after: Option }, +} + +fn decode_reservation( + body: &[u8], + expected_name: &str, + expected_uid: &str, + archive: &PreparedArchive, +) -> Result { + let response: ReservationResponse = serde_json::from_slice(body).map_err(|_| ReportUploadError::Response)?; + validate_resource(&response.support_bundle, "PENDING", expected_name, expected_uid, archive)?; + if response.upload_authorization.method != "PUT" || !is_short_lived_future_instant(&response.upload_authorization.expire_time) + { + return Err(ReportUploadError::UploadAuthorization); + } + let url = Url::parse(&response.upload_authorization.url).map_err(|_| ReportUploadError::UploadAuthorization)?; + if url.scheme() != "https" + || url.host_str().is_none() + || url.cannot_be_a_base() + || !url.username().is_empty() + || url.password().is_some() + || url.fragment().is_some() + { + return Err(ReportUploadError::UploadAuthorization); + } + let headers = validate_headers(response.upload_authorization.headers, archive)?; + Ok(Reservation { + authorization: UploadAuthorization { url, headers }, + }) +} + +fn decode_receipt( + body: &[u8], + expected_name: &str, + expected_uid: &str, + archive: &PreparedArchive, +) -> Result { + let response: BundleResource = serde_json::from_slice(body).map_err(|_| ReportUploadError::Response)?; + validate_resource(&response, "UPLOADED", expected_name, expected_uid, archive)?; + Ok(ReportUploadReceipt { + name: response.name, + uid: response.uid, + state: response.state, + declared_size_bytes: response.declared_size_bytes, + declared_sha256: response.declared_sha256, + }) +} + +fn validate_resource( + resource: &BundleResource, + expected_state: &str, + expected_name: &str, + expected_uid: &str, + archive: &PreparedArchive, +) -> Result<(), ReportUploadError> { + if resource.name != expected_name + || resource.uid != expected_uid + || resource.state != expected_state + || resource.declared_size_bytes != archive.size + || resource.declared_sha256 != archive.sha256 + || !is_exact_utc_seconds(&resource.expire_time) + || !is_exact_utc_seconds(&resource.create_time) + || !is_exact_utc_seconds(&resource.update_time) + { + return Err(ReportUploadError::Response); + } + Ok(()) +} + +fn validate_headers(raw: HashMap, archive: &PreparedArchive) -> Result { + let mut headers = HeaderMap::new(); + let mut names = HashSet::new(); + for (name, value) in raw { + let normalized = name.to_ascii_lowercase(); + if !names.insert(normalized.clone()) || forbidden_header(&normalized) { + return Err(ReportUploadError::UploadAuthorization); + } + let name = HeaderName::from_bytes(name.as_bytes()).map_err(|_| ReportUploadError::UploadAuthorization)?; + let value = HeaderValue::from_str(&value).map_err(|_| ReportUploadError::UploadAuthorization)?; + headers.insert(name, value); + } + let expected_length = archive.size.to_string(); + if headers.get(header::CONTENT_TYPE).and_then(|value| value.to_str().ok()) != Some(CONTENT_TYPE) + || headers.get(header::CONTENT_LENGTH).and_then(|value| value.to_str().ok()) != Some(expected_length.as_str()) + || headers.get(header::IF_NONE_MATCH).and_then(|value| value.to_str().ok()) != Some("*") + || headers.get("x-amz-checksum-sha256").and_then(|value| value.to_str().ok()) != Some(archive.checksum_base64.as_str()) + || headers + .get("x-amz-server-side-encryption") + .and_then(|value| value.to_str().ok()) + != Some("AES256") + { + return Err(ReportUploadError::UploadAuthorization); + } + Ok(headers) +} + +fn forbidden_header(name: &str) -> bool { + matches!( + name, + "authorization" + | "proxy-authorization" + | "cookie" + | "set-cookie" + | "host" + | "connection" + | "transfer-encoding" + | "upgrade" + | "te" + | "trailer" + ) +} + +fn is_short_lived_future_instant(value: &str) -> bool { + if !is_exact_utc_seconds(value) { + return false; + } + let now = Utc::now(); + DateTime::parse_from_rfc3339(value).is_ok_and(|instant| { + let instant = instant.with_timezone(&Utc); + instant > now && (instant - now).to_std().is_ok_and(|duration| duration <= MAX_UPLOAD_TIMEOUT) + }) +} + +fn retry_after(headers: &HeaderMap, now: DateTime, maximum: Duration) -> Option { + let value = headers.get(header::RETRY_AFTER)?.to_str().ok()?; + let delay = value.parse::().ok().map(Duration::from_secs).or_else(|| { + DateTime::parse_from_rfc2822(value) + .ok() + .and_then(|at| (at.with_timezone(&Utc) - now).to_std().ok()) + })?; + Some(delay.min(maximum)) +} + +async fn sleep_or_cancel(cancellation: &CancellationToken, delay: Duration) -> Result<(), ReportUploadError> { + tokio::select! { + biased; + () = cancellation.cancelled() => Err(ReportUploadError::Cancelled), + () = tokio::time::sleep(delay) => Ok(()), + } +} + +fn transport_error(error: TelemetryError) -> ReportUploadError { + match error { + TelemetryError::Endpoint => ReportUploadError::Endpoint, + TelemetryError::RootCertificate => ReportUploadError::RootCertificate, + TelemetryError::ProxyConfiguration => ReportUploadError::ProxyConfiguration, + TelemetryError::ProxyAuthentication => ReportUploadError::ProxyAuthentication, + TelemetryError::ProxyRejected => ReportUploadError::ProxyRejected, + TelemetryError::TlsPeer => ReportUploadError::TlsPeer, + TelemetryError::Schedule => ReportUploadError::Schedule, + TelemetryError::NotRegistered => ReportUploadError::NotRegistered, + TelemetryError::IdentityMissing => ReportUploadError::IdentityMissing, + TelemetryError::IdentityCertificate => ReportUploadError::IdentityCertificate, + TelemetryError::CredentialName => ReportUploadError::CredentialName, + TelemetryError::CredentialExpired => ReportUploadError::CredentialExpired, + TelemetryError::StateConflict => ReportUploadError::CredentialState, + TelemetryError::ResponseTooLarge => ReportUploadError::ResponseTooLarge, + TelemetryError::Url(_) + | TelemetryError::Transport(_) + | TelemetryError::Identity(_) + | TelemetryError::IdentityStore(_) + | TelemetryError::CredentialStore(_) + | TelemetryError::CredentialValidation(_) => ReportUploadError::UploadTransport, + } +} + +/// Safe, credential-redacted report upload failures. +#[derive(Debug, thiserror::Error)] +pub enum ReportUploadError { + #[error("Connect report upload timeout must be from one second through fifteen minutes")] + UploadTimeout, + #[error("Connect report upload endpoint must be an HTTPS base URL without credentials, query, or fragment")] + Endpoint, + #[error("Connect report upload root CA configuration is invalid")] + RootCertificate, + #[error("Connect report upload proxy configuration is invalid")] + ProxyConfiguration, + #[error("Connect proxy authentication failed; verify the configured proxy credential files")] + ProxyAuthentication, + #[error("Connect proxy connection failed; verify proxy availability, credentials, and the approved targets")] + ProxyRejected, + #[error("Connect TLS peer certificate validation failed; verify the endpoint and configured root CA")] + TlsPeer, + #[error("Connect report upload retry schedule is invalid")] + Schedule, + #[error("RustFS is not registered with Connect")] + NotRegistered, + #[error("the Connect device private key is missing")] + IdentityMissing, + #[error("the stored Connect certificate and device private key cannot form a TLS identity")] + IdentityCertificate, + #[error("the stored Connect credential name is invalid")] + CredentialName, + #[error("the stored Connect device certificate is not currently valid")] + CredentialExpired, + #[error("the persisted Connect credential transition is invalid")] + CredentialState, + #[error("Connect report upload response exceeded 64 KiB")] + ResponseTooLarge, + #[error("the report archive could not be opened")] + ArchiveOpen(#[source] std::io::Error), + #[error("the report archive could not be read")] + ArchiveRead(#[source] std::io::Error), + #[error("the report archive must be a regular file")] + ArchiveType, + #[error("the report archive must contain 1 byte through 256 MiB")] + ArchiveSize, + #[error("the report archive changed while its digest was calculated")] + ArchiveChanged, + #[error("Connect returned an invalid report upload response")] + Response, + #[error("Connect returned an invalid or expired report upload authorization")] + UploadAuthorization, + #[error("Connect authentication stopped report upload with HTTP {status}")] + AuthenticationStopped { status: u16 }, + #[error("Connect rejected report upload control request with HTTP {status}")] + ControlRejected { status: u16 }, + #[error("the report object upload was rejected with HTTP {status}")] + UploadRejected { status: u16 }, + #[error("the report object upload transport failed")] + UploadTransport, + #[error("Connect report upload exhausted its bounded retries")] + RetryExhausted, + #[error("Connect report upload was cancelled")] + Cancelled, +} + +#[cfg(test)] +mod tests { + use chrono::SecondsFormat; + use serde_json::json; + + use super::*; + + #[tokio::test] + async fn prepares_the_exact_bounded_archive() { + let temp = tempfile::tempdir().expect("tempdir"); + let path = temp.path().join("bundle.tar.zst"); + tokio::fs::write(&path, b"redacted support bundle") + .await + .expect("write bundle"); + + let archive = prepare_archive(&path, &CancellationToken::new()) + .await + .expect("prepare archive"); + + assert_eq!(archive.size, 23); + let digest = Sha256::digest(b"redacted support bundle"); + assert_eq!(archive.sha256, faster_hex::hex_string(&digest)); + assert_eq!(archive.checksum_base64, base64_simd::STANDARD.encode_to_string(&digest)); + } + + #[tokio::test] + async fn cancellation_stops_archive_preparation() { + let temp = tempfile::tempdir().expect("tempdir"); + let path = temp.path().join("bundle.tar.zst"); + tokio::fs::write(&path, b"bundle").await.expect("write bundle"); + let cancellation = CancellationToken::new(); + cancellation.cancel(); + + assert!(matches!(prepare_archive(&path, &cancellation).await, Err(ReportUploadError::Cancelled))); + } + + #[tokio::test] + async fn reservation_is_bound_to_archive_and_safe_headers() { + let temp = tempfile::tempdir().expect("tempdir"); + let path = temp.path().join("bundle.tar.zst"); + tokio::fs::write(&path, b"bundle").await.expect("write bundle"); + let archive = prepare_archive(&path, &CancellationToken::new()) + .await + .expect("prepare archive"); + let bundle_uid = Uuid::now_v7().to_string(); + let name = format!("organizations/o/clusters/c/supportBundles/{bundle_uid}"); + let now = Utc::now(); + let instant = |minutes| (now + chrono::Duration::minutes(minutes)).to_rfc3339_opts(SecondsFormat::Secs, true); + let response = json!({ + "supportBundle": { + "name": &name, + "uid": &bundle_uid, + "state": "PENDING", + "declaredSizeBytes": archive.size, + "declaredSha256": &archive.sha256, + "expireTime": instant(60), + "createTime": now.to_rfc3339_opts(SecondsFormat::Secs, true), + "updateTime": now.to_rfc3339_opts(SecondsFormat::Secs, true) + }, + "uploadAuthorization": { + "method": "PUT", + "url": "https://objects.example.test/upload?signature=hidden", + "headers": { + "Content-Type": CONTENT_TYPE, + "Content-Length": archive.size.to_string(), + "If-None-Match": "*", + "x-amz-checksum-sha256": &archive.checksum_base64, + "x-amz-server-side-encryption": "AES256" + }, + "expireTime": instant(5) + } + }); + + decode_reservation(&serde_json::to_vec(&response).expect("response JSON"), &name, &bundle_uid, &archive) + .expect("valid reservation"); + + let mut wrong_target = response.clone(); + wrong_target["supportBundle"]["name"] = json!(format!("organizations/o/clusters/other/supportBundles/{bundle_uid}")); + assert!(matches!( + decode_reservation(&serde_json::to_vec(&wrong_target).expect("response JSON"), &name, &bundle_uid, &archive,), + Err(ReportUploadError::Response) + )); + + let mut unsafe_response = response; + unsafe_response["uploadAuthorization"]["headers"]["Authorization"] = json!("secret"); + assert!(matches!( + decode_reservation( + &serde_json::to_vec(&unsafe_response).expect("response JSON"), + &name, + &bundle_uid, + &archive, + ), + Err(ReportUploadError::UploadAuthorization) + )); + } +} diff --git a/rustfs/src/connect/telemetry.rs b/rustfs/src/connect/telemetry.rs index d3bc82a27..828f07592 100644 --- a/rustfs/src/connect/telemetry.rs +++ b/rustfs/src/connect/telemetry.rs @@ -95,6 +95,15 @@ impl TelemetryTransport { } pub(crate) async fn post(&self, collection: &str, value: &T) -> Result { + self.post_expect(collection, value, StatusCode::OK).await + } + + pub(crate) async fn post_expect( + &self, + collection: &str, + value: &T, + expected_status: StatusCode, + ) -> Result { let mut authenticated = self.authenticated_client().await?; let mut refreshed = false; loop { @@ -138,7 +147,7 @@ impl TelemetryTransport { reason: response_reason(response).await, }); } - if status != StatusCode::OK { + if status != expected_status { return Ok(TelemetryDelivery::Rejected { status: status.as_u16(), reason: response_reason(response).await, @@ -151,6 +160,10 @@ impl TelemetryTransport { } } + pub(crate) fn presigned_client(&self, timeout: Duration) -> Result { + build_client(&self.roots, timeout, None, self.config.proxy.as_ref()).map_err(credential_recovery_error) + } + async fn authenticated_client(&self) -> Result { let lock = self.config.credential_store.lock().await?; let (credential, identity, _) = ConnectClient::recover_valid_credential_locked( diff --git a/rustfs/src/startup_entrypoint.rs b/rustfs/src/startup_entrypoint.rs index ecf7e2821..a764d8151 100644 --- a/rustfs/src/startup_entrypoint.rs +++ b/rustfs/src/startup_entrypoint.rs @@ -17,8 +17,8 @@ use crate::{ CommandResult, Config, ConnectClientPerformanceOperation, ConnectClientPerformanceOpts, ConnectDrivePerformanceOpts, ConnectEnvironmentInventoryOpts, ConnectLicenseCommands, ConnectLicenseScopeOpts, ConnectLogsMode, ConnectLogsOpts, ConnectObjectPerformanceOperation, ConnectObjectPerformanceOpts, ConnectProfileOpts, ConnectProfileTool, - ConnectRelayMaterialKind, ConnectRelayOpts, ConnectSiteReplicationPerformanceOpts, ConnectTelemetryArtifactOpts, - ConnectTelemetryCommands, ConnectThreadProfileScope, ConnectTopCommands, Opt, + ConnectRelayMaterialKind, ConnectRelayOpts, ConnectReportUploadOpts, ConnectSiteReplicationPerformanceOpts, + ConnectTelemetryArtifactOpts, ConnectTelemetryCommands, ConnectThreadProfileScope, ConnectTopCommands, Opt, }, startup_lifecycle::{StartupRuntimeLifecycle, run_startup_runtime_lifecycle}, startup_preflight::{StartupServerPreflightError, bootstrap_external_prefix_compat, init_startup_server_preflight}, @@ -139,6 +139,7 @@ async fn async_main() -> Result<()> { } CommandResult::ConnectLicense(command) => return execute_connect_license(command).await, CommandResult::ConnectRelay(options) => return execute_connect_relay(*options).await, + CommandResult::ConnectReportUpload(options) => return execute_connect_report_upload(options).await, CommandResult::ConnectEnvironmentInventory(options) => return execute_connect_environment_inventory(options).await, CommandResult::ConnectClientPerformance(options) => return execute_connect_client_performance(options).await, CommandResult::ConnectDrivePerformance(options) => return execute_connect_drive_performance(options).await, @@ -1412,6 +1413,35 @@ async fn execute_connect_license(command: ConnectLicenseCommands) -> Result<()> } } +async fn execute_connect_report_upload(options: ConnectReportUploadOpts) -> Result<()> { + use crate::connect::{CredentialStore, HeartbeatConfig, IdentityStore, ProxyConfig, ReportUploadClient}; + + let root_ca_pem = std::fs::read(&options.ca_file).map_err(Error::other)?; + let mut config = HeartbeatConfig::new( + &options.endpoint, + root_ca_pem, + IdentityStore::new(options.state_dir.join("identity")), + CredentialStore::new(options.state_dir.join("credential")), + options.state_dir.join("heartbeat/state.json"), + ); + config.proxy = ProxyConfig::from_env().map_err(Error::other)?; + let client = ReportUploadClient::new(config, Duration::from_secs(options.upload_timeout_seconds)).map_err(Error::other)?; + let cancellation = CancellationToken::new(); + let upload = client.upload(&options.archive, &cancellation); + tokio::pin!(upload); + let receipt = tokio::select! { + biased; + signal = tokio::signal::ctrl_c() => { + signal.map_err(Error::other)?; + cancellation.cancel(); + return Err(Error::other("connect report upload cancelled")); + } + result = upload.as_mut() => result.map_err(Error::other)?, + }; + println!("{}", serde_json::to_string(&receipt).map_err(Error::other)?); + Ok(()) +} + async fn execute_connect_relay(options: ConnectRelayOpts) -> Result<()> { use crate::connect::{ ProxyConfig, RelayHttpClient, RelayMaterialKind, RelayParty, TrustedReceiptSigner, prepare_approved_artifact, diff --git a/rustfs/tests/connect_report_upload.rs b/rustfs/tests/connect_report_upload.rs new file mode 100644 index 000000000..42851c9f0 --- /dev/null +++ b/rustfs/tests/connect_report_upload.rs @@ -0,0 +1,446 @@ +// 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; +use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; +use std::sync::{Arc, Mutex}; +use std::time::Duration; + +use bytes::Bytes; +use chrono::{SecondsFormat, Utc}; +use http_body_util::{BodyExt as _, Full}; +use hyper::service::service_fn; +use hyper::{Request, Response, StatusCode}; +use hyper_util::rt::TokioIo; +use rcgen::{ + BasicConstraints, CertificateParams, DistinguishedName, DnType, ExtendedKeyUsagePurpose, IsCa, Issuer, KeyPair, + KeyUsagePurpose, SanType, +}; +use rustfs::connect::{ + CredentialStore, DeviceCredential, HeartbeatConfig, HeartbeatSchedule, IdentityStore, ReportUploadClient, ReportUploadError, +}; +use rustls::RootCertStore; +use rustls::pki_types::{CertificateDer, PrivateKeyDer, PrivatePkcs8KeyDer}; +use rustls::server::WebPkiClientVerifier; +use serde_json::{Value, json}; +use sha2::{Digest as _, Sha256}; +use time::OffsetDateTime; +use tokio::net::TcpListener; +use tokio_rustls::TlsAcceptor; +use tokio_util::sync::CancellationToken; + +const ORGANIZATION_UID: &str = "0198f4b0-1a00-7c10-8d21-2e3f4a5b6c70"; +const CLUSTER_UID: &str = "0198f4b0-2b00-7d20-9e31-3f4a5b6c7d81"; +const DEVICE_UID: &str = "0198f4b0-3c00-7e30-8f41-4a5b6c7d8e92"; + +struct TestPki { + root_params: CertificateParams, + root_key: KeyPair, + root_der: CertificateDer<'static>, + root_pem: String, + server_der: CertificateDer<'static>, + server_key: PrivatePkcs8KeyDer<'static>, +} + +impl TestPki { + fn new() -> Self { + let now = OffsetDateTime::now_utc(); + let root_key = KeyPair::generate().expect("generate root key"); + let mut root_params = CertificateParams::default(); + root_params.is_ca = IsCa::Ca(BasicConstraints::Unconstrained); + root_params.not_before = now - time::Duration::days(30); + root_params.not_after = now + time::Duration::days(30); + root_params.key_usages = vec![KeyUsagePurpose::KeyCertSign, KeyUsagePurpose::DigitalSignature]; + let root = root_params.self_signed(&root_key).expect("sign root"); + + let server_key = KeyPair::generate().expect("generate server key"); + let mut server_params = CertificateParams::default(); + server_params.not_before = now - time::Duration::hours(1); + server_params.not_after = now + time::Duration::days(2); + server_params + .subject_alt_names + .push(SanType::DnsName("localhost".try_into().expect("valid DNS name"))); + server_params.extended_key_usages = vec![ExtendedKeyUsagePurpose::ServerAuth]; + let server = server_params + .signed_by(&server_key, &Issuer::from_params(&root_params, &root_key)) + .expect("sign server certificate"); + Self { + root_params, + root_key, + root_der: root.der().clone(), + root_pem: root.pem(), + server_der: server.der().clone(), + server_key: PrivatePkcs8KeyDer::from(server_key.serialize_der()), + } + } + + fn agent_server_config(&self) -> rustls::ServerConfig { + let mut roots = RootCertStore::empty(); + roots.add(self.root_der.clone()).expect("add client root"); + let verifier = WebPkiClientVerifier::builder(Arc::new(roots)) + .build() + .expect("client verifier"); + rustls::ServerConfig::builder() + .with_client_cert_verifier(verifier) + .with_single_cert(vec![self.server_der.clone()], PrivateKeyDer::Pkcs8(self.server_key.clone_key())) + .expect("agent TLS") + } + + fn object_server_config(&self) -> rustls::ServerConfig { + rustls::ServerConfig::builder() + .with_no_client_auth() + .with_single_cert(vec![self.server_der.clone()], PrivateKeyDer::Pkcs8(self.server_key.clone_key())) + .expect("object TLS") + } + + fn stores(&self, temp: &tempfile::TempDir) -> (IdentityStore, CredentialStore) { + let identity_store = IdentityStore::new(temp.path().join("identity")); + let identity = identity_store.load_or_create().expect("create identity"); + let private_key = PrivatePkcs8KeyDer::from(identity.to_pkcs8_der().expect("serialize key").to_vec()); + let device_key = KeyPair::from_pkcs8_der_and_sign_algo(&private_key, &rcgen::PKCS_ECDSA_P256_SHA256).expect("device key"); + let now = OffsetDateTime::now_utc(); + let not_before = now - time::Duration::hours(1); + let not_after = now + time::Duration::hours(23); + let mut params = CertificateParams::default(); + params.not_before = not_before; + params.not_after = not_after; + params.serial_number = Some(vec![1; 16].into()); + params.key_usages = vec![KeyUsagePurpose::DigitalSignature]; + params.extended_key_usages = vec![ExtendedKeyUsagePurpose::ClientAuth]; + params.distinguished_name = DistinguishedName::new(); + params.distinguished_name.push(DnType::CommonName, DEVICE_UID); + params.subject_alt_names.push(SanType::URI( + format!("urn:rustfs:connect:device:{DEVICE_UID}") + .try_into() + .expect("device URI"), + )); + let certificate = params + .signed_by(&device_key, &Issuer::from_params(&self.root_params, &self.root_key)) + .expect("device certificate"); + let credential = DeviceCredential { + name: format!("organizations/{ORGANIZATION_UID}/clusters/{CLUSTER_UID}/clusterDevices/{DEVICE_UID}"), + uid: DEVICE_UID.to_owned(), + protocol_version: "v1".to_owned(), + key_id: format!("x509-{}", "01".repeat(16)), + certificate_serial: "01".repeat(16), + certificate: certificate.pem(), + certificate_chain: certificate.pem(), + not_before_unix: not_before.unix_timestamp(), + not_after_unix: not_after.unix_timestamp(), + }; + let directory = temp.path().join("credential"); + fs::create_dir_all(&directory).expect("credential directory"); + let path = directory.join("device.crt.json"); + fs::write(&path, serde_json::to_vec(&credential).expect("credential JSON")).expect("write credential"); + private_mode(&path); + (identity_store, CredentialStore::new(directory)) + } +} + +struct ObjectServer { + endpoint: String, + attempts: Arc, + bodies: Arc>>>, + client_certificates: Arc>>, + task: tokio::task::JoinHandle<()>, +} + +impl Drop for ObjectServer { + fn drop(&mut self) { + self.task.abort(); + } +} + +async fn object_server(pki: &TestPki) -> ObjectServer { + let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind object server"); + let address = listener.local_addr().expect("object server address"); + let acceptor = TlsAcceptor::from(Arc::new(pki.object_server_config())); + let attempts = Arc::new(AtomicUsize::new(0)); + let observed_attempts = attempts.clone(); + let bodies = Arc::new(Mutex::new(Vec::new())); + let observed_bodies = bodies.clone(); + let client_certificates = Arc::new(Mutex::new(Vec::new())); + let observed_certificates = client_certificates.clone(); + let interrupt_first = Arc::new(AtomicBool::new(true)); + let task = tokio::spawn(async move { + while let Ok((stream, _)) = listener.accept().await { + let acceptor = acceptor.clone(); + let bodies = observed_bodies.clone(); + let certificates = observed_certificates.clone(); + let interrupt = interrupt_first.clone(); + observed_attempts.fetch_add(1, Ordering::SeqCst); + tokio::spawn(async move { + let Ok(stream) = acceptor.accept(stream).await else { return }; + certificates.lock().expect("certificate observations").push( + stream + .get_ref() + .1 + .peer_certificates() + .is_some_and(|certificates| !certificates.is_empty()), + ); + if interrupt.swap(false, Ordering::SeqCst) { + return; + } + let service = service_fn(move |request: Request| { + let bodies = bodies.clone(); + async move { + assert_eq!(request.method(), hyper::Method::PUT); + assert_eq!(request.uri().path(), "/upload"); + assert_eq!(request.headers().get("content-type").expect("content type"), "application/octet-stream"); + assert_eq!(request.headers().get("if-none-match").expect("create only"), "*"); + assert_eq!(request.headers().get("x-amz-server-side-encryption").expect("encryption"), "AES256"); + let declared_length = request + .headers() + .get("content-length") + .expect("content length") + .to_str() + .expect("content length text") + .parse::() + .expect("content length number"); + let checksum = request + .headers() + .get("x-amz-checksum-sha256") + .expect("checksum") + .to_str() + .expect("checksum text") + .to_owned(); + let body = request.into_body().collect().await.expect("upload body").to_bytes().to_vec(); + assert_eq!(declared_length, body.len()); + assert_eq!(checksum, base64_simd::STANDARD.encode_to_string(Sha256::digest(&body))); + bodies.lock().expect("uploaded bodies").push(body); + Ok::<_, hyper::Error>(Response::new(Full::new(Bytes::new()))) + } + }); + let _ = hyper::server::conn::http1::Builder::new() + .serve_connection(TokioIo::new(stream), service) + .await; + }); + } + }); + ObjectServer { + endpoint: format!("https://localhost:{}/upload?signature=hidden", address.port()), + attempts, + bodies, + client_certificates, + task, + } +} + +#[derive(Clone)] +struct BundleDeclaration { + uid: String, + size: u64, + sha256: String, +} + +struct AgentServer { + endpoint: String, + reserve_count: Arc, + complete_count: Arc, + task: tokio::task::JoinHandle<()>, +} + +impl Drop for AgentServer { + fn drop(&mut self) { + self.task.abort(); + } +} + +async fn agent_server(pki: &TestPki, upload_url: String) -> AgentServer { + let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind agent server"); + let address = listener.local_addr().expect("agent server address"); + let acceptor = TlsAcceptor::from(Arc::new(pki.agent_server_config())); + let declaration = Arc::new(Mutex::new(None::)); + let reserve_count = Arc::new(AtomicUsize::new(0)); + let observed_reserves = reserve_count.clone(); + let complete_count = Arc::new(AtomicUsize::new(0)); + let observed_completes = complete_count.clone(); + let task = tokio::spawn(async move { + while let Ok((stream, _)) = listener.accept().await { + let acceptor = acceptor.clone(); + let declaration = declaration.clone(); + let upload_url = upload_url.clone(); + let reserves = observed_reserves.clone(); + let completes = observed_completes.clone(); + tokio::spawn(async move { + let Ok(stream) = acceptor.accept(stream).await else { return }; + let service = service_fn(move |request: Request| { + let declaration = declaration.clone(); + let upload_url = upload_url.clone(); + let reserves = reserves.clone(); + let completes = completes.clone(); + async move { + let path = request.uri().path().to_owned(); + let request_body = request.into_body().collect().await.expect("control body").to_bytes(); + let request: Value = serde_json::from_slice(&request_body).expect("control JSON"); + let (status, body) = if path == format!("/agent/clusters/{CLUSTER_UID}/supportBundles") { + reserves.fetch_add(1, Ordering::SeqCst); + assert_eq!(request["protocolVersion"], "v1"); + assert_eq!(request["contentType"], "application/octet-stream"); + let current = BundleDeclaration { + uid: request["bundleUid"].as_str().expect("bundle uid").to_owned(), + size: request["declaredSizeBytes"].as_u64().expect("declared size"), + sha256: request["declaredSha256"].as_str().expect("declared digest").to_owned(), + }; + *declaration.lock().expect("declaration") = Some(current.clone()); + let resource = bundle_resource(¤t, "PENDING"); + let mut digest = [0u8; 32]; + faster_hex::hex_decode(current.sha256.as_bytes(), &mut digest).expect("digest hex"); + ( + StatusCode::CREATED, + json!({ + "supportBundle": resource, + "uploadAuthorization": { + "method": "PUT", + "url": upload_url, + "headers": { + "Content-Type": "application/octet-stream", + "Content-Length": current.size.to_string(), + "If-None-Match": "*", + "x-amz-checksum-sha256": base64_simd::STANDARD.encode_to_string(digest), + "x-amz-server-side-encryption": "AES256" + }, + "expireTime": instant(5) + } + }), + ) + } else { + let current = declaration.lock().expect("declaration").clone().expect("reserved bundle"); + assert_eq!( + path, + format!("/agent/clusters/{CLUSTER_UID}/supportBundles/{}:completeUpload", current.uid) + ); + assert_eq!(request["protocolVersion"], "v1"); + completes.fetch_add(1, Ordering::SeqCst); + (StatusCode::OK, bundle_resource(¤t, "UPLOADED")) + }; + Ok::<_, hyper::Error>( + Response::builder() + .status(status) + .header("content-type", "application/json") + .body(Full::new(Bytes::from(serde_json::to_vec(&body).expect("response JSON")))) + .expect("control response"), + ) + } + }); + let _ = hyper::server::conn::http1::Builder::new() + .serve_connection(TokioIo::new(stream), service) + .await; + }); + } + }); + AgentServer { + endpoint: format!("https://localhost:{}/agent/", address.port()), + reserve_count, + complete_count, + task, + } +} + +fn bundle_resource(declaration: &BundleDeclaration, state: &str) -> Value { + json!({ + "name": format!( + "organizations/{ORGANIZATION_UID}/clusters/{CLUSTER_UID}/supportBundles/{}", + declaration.uid + ), + "uid": &declaration.uid, + "state": state, + "declaredSizeBytes": declaration.size, + "declaredSha256": &declaration.sha256, + "expireTime": instant(60), + "createTime": instant(0), + "updateTime": instant(0) + }) +} + +fn instant(minutes: i64) -> String { + (Utc::now() + chrono::Duration::minutes(minutes)).to_rfc3339_opts(SecondsFormat::Secs, true) +} + +fn config(temp: &tempfile::TempDir, pki: &TestPki, endpoint: &str) -> HeartbeatConfig { + let (identity_store, credential_store) = pki.stores(temp); + HeartbeatConfig { + endpoint: endpoint.to_owned(), + root_ca_pem: pki.root_pem.as_bytes().to_vec(), + identity_store, + credential_store, + state_path: temp.path().join("heartbeat/state.json"), + schedule: HeartbeatSchedule { + cadence: Duration::from_secs(60), + jitter: Duration::ZERO, + timeout: Duration::from_secs(2), + initial_backoff: Duration::from_millis(10), + max_backoff: Duration::from_millis(20), + }, + proxy: None, + } +} + +#[tokio::test] +async fn interrupted_put_is_reauthorized_and_completed() { + let temp = tempfile::tempdir().expect("tempdir"); + let archive_path = temp.path().join("support-bundle.tar.zst"); + let archive = b"redacted support bundle evidence"; + fs::write(&archive_path, archive).expect("write archive"); + let pki = TestPki::new(); + let object = object_server(&pki).await; + let agent = agent_server(&pki, object.endpoint.clone()).await; + + let receipt = ReportUploadClient::new(config(&temp, &pki, &agent.endpoint), Duration::from_secs(5)) + .expect("report client") + .upload(&archive_path, &CancellationToken::new()) + .await + .expect("upload report"); + + assert_eq!(receipt.state, "UPLOADED"); + assert_eq!(receipt.declared_size_bytes, archive.len() as u64); + assert_eq!(receipt.declared_sha256, faster_hex::hex_string(&Sha256::digest(archive))); + assert_eq!(agent.reserve_count.load(Ordering::SeqCst), 2); + assert_eq!(agent.complete_count.load(Ordering::SeqCst), 1); + assert!(object.attempts.load(Ordering::SeqCst) >= 2); + assert_eq!(object.bodies.lock().expect("uploaded bodies").as_slice(), &[archive.to_vec()]); + let certificates = object.client_certificates.lock().expect("object certificates"); + assert!(certificates.len() >= 2); + assert!(certificates.iter().all(|presented| !presented)); +} + +#[tokio::test] +async fn untrusted_object_store_certificate_stops_before_completion() { + let temp = tempfile::tempdir().expect("tempdir"); + let archive_path = temp.path().join("support-bundle.tar.zst"); + fs::write(&archive_path, b"redacted support bundle evidence").expect("write archive"); + let agent_pki = TestPki::new(); + let object_pki = TestPki::new(); + let object = object_server(&object_pki).await; + let agent = agent_server(&agent_pki, object.endpoint.clone()).await; + + let error = ReportUploadClient::new(config(&temp, &agent_pki, &agent.endpoint), Duration::from_secs(5)) + .expect("report client") + .upload(&archive_path, &CancellationToken::new()) + .await + .expect_err("untrusted object store must fail"); + + assert!(matches!(error, ReportUploadError::TlsPeer)); + assert_eq!(agent.reserve_count.load(Ordering::SeqCst), 1); + assert_eq!(agent.complete_count.load(Ordering::SeqCst), 0); +} + +#[cfg(unix)] +fn private_mode(path: &std::path::Path) { + use std::os::unix::fs::PermissionsExt as _; + fs::set_permissions(path, fs::Permissions::from_mode(0o600)).expect("set private mode"); +} + +#[cfg(not(unix))] +fn private_mode(_path: &std::path::Path) {}