feat(connect): add device report upload client (#7774)

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<ConnectRelayOpts>),
+    /// 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<ConnectRelayOpts>),
+    /// 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"] {

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,

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)),
                 },

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};

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<Self, ReportUploadError> {
+        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<ReportUploadReceipt, ReportUploadError> {
+        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<T: Serialize>(
+        &self,
+        operation: &str,
+        body: &T,
+        expected_status: StatusCode,
+        cancellation: &CancellationToken,
+    ) -> Result<(String, Vec<u8>), 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<UploadDelivery, ReportUploadError> {
+        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<PreparedArchive, ReportUploadError> {
+    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<String, String>,
+    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<Duration> },
+}
+
+fn decode_reservation(
+    body: &[u8],
+    expected_name: &str,
+    expected_uid: &str,
+    archive: &PreparedArchive,
+) -> Result<Reservation, ReportUploadError> {
+    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<ReportUploadReceipt, ReportUploadError> {
+    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<String, String>, archive: &PreparedArchive) -> Result<HeaderMap, ReportUploadError> {
+    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<Utc>, maximum: Duration) -> Option<Duration> {
+    let value = headers.get(header::RETRY_AFTER)?.to_str().ok()?;
+    let delay = value.parse::<u64>().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)
+        ));
+    }
+}

rustfs/src/connect/telemetry.rs

@@ -95,6 +95,15 @@ impl TelemetryTransport {
     }
 
     pub(crate) async fn post<T: Serialize>(&self, collection: &str, value: &T) -> Result<TelemetryDelivery, TelemetryError> {
+        self.post_expect(collection, value, StatusCode::OK).await
+    }
+
+    pub(crate) async fn post_expect<T: Serialize>(
+        &self,
+        collection: &str,
+        value: &T,
+        expected_status: StatusCode,
+    ) -> Result<TelemetryDelivery, TelemetryError> {
         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<Client, TelemetryError> {
+        build_client(&self.roots, timeout, None, self.config.proxy.as_ref()).map_err(credential_recovery_error)
+    }
+
     async fn authenticated_client(&self) -> Result<AuthenticatedClient, TelemetryError> {
         let lock = self.config.credential_store.lock().await?;
         let (credential, identity, _) = ConnectClient::recover_valid_credential_locked(

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,

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<AtomicUsize>,
+    bodies: Arc<Mutex<Vec<Vec<u8>>>>,
+    client_certificates: Arc<Mutex<Vec<bool>>>,
+    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<hyper::body::Incoming>| {
+                    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::<usize>()
+                            .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<AtomicUsize>,
+    complete_count: Arc<AtomicUsize>,
+    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::<BundleDeclaration>));
+    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<hyper::body::Incoming>| {
+                    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(&current, "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(&current, "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) {}