diff --git a/rustfs/src/config/cli.rs b/rustfs/src/config/cli.rs index 8f9fabcfa..a969b0914 100644 --- a/rustfs/src/config/cli.rs +++ b/rustfs/src/config/cli.rs @@ -129,6 +129,86 @@ pub enum ConnectCommands { License(ConnectLicenseOpts), /// Capture a consent-bound local profile and write a signed export Profile(ConnectProfileOpts), + /// Capture allow-listed local log events and write a signed export + Logs(ConnectLogsOpts), +} + +/// `connect logs` options. +#[derive(Args, Clone)] +pub struct ConnectLogsOpts { + /// Directory containing an enrolled Connect device identity + #[arg(long = "state-dir")] + pub state_dir: PathBuf, + + /// New local archive path; an existing file is never replaced + #[arg(long)] + pub output: PathBuf, + + /// Capture the recent configured log window or tail new events + #[arg(long, value_enum, default_value = "batch")] + pub mode: ConnectLogsMode, + + /// Negotiated producer schema version + #[arg(long = "schema-version", default_value_t = 1)] + pub schema_version: u16, + + /// Negotiated producer capability + #[arg(long, default_value = "logs.capture@1")] + pub capability: String, + + /// Organization resource name bound to the export + #[arg(long, value_parser = NonEmptyStringValueParser::new())] + pub organization: String, + + /// Cluster resource name bound to the export + #[arg(long, value_parser = NonEmptyStringValueParser::new())] + pub cluster: String, + + /// Cluster-device resource name bound to the export + #[arg(long, value_parser = NonEmptyStringValueParser::new())] + pub device: String, + + /// UUIDv7 diagnostic run identifier issued by Connect + #[arg(long = "run-uid", value_parser = NonEmptyStringValueParser::new())] + pub run_uid: String, + + /// UUIDv7 artifact identifier issued by Connect + #[arg(long = "artifact-uid", value_parser = NonEmptyStringValueParser::new())] + pub artifact_uid: String, + + /// UUIDv7 consent identifier issued by Connect + #[arg(long = "consent-uid", value_parser = NonEmptyStringValueParser::new())] + pub consent_uid: String, + + /// Consent policy revision bound to this capture + #[arg(long = "policy-revision")] + pub policy_revision: u64, + + /// Consent expiry as UTC Unix seconds + #[arg(long = "consent-expires-at")] + pub consent_expires_at_unix: i64, + + /// Artifact expiry as UTC Unix seconds + #[arg(long = "expires-at")] + pub expires_at_unix: i64, + + /// Batch lookback or live capture duration in milliseconds + #[arg(long = "duration-millis")] + pub duration_millis: u64, + + /// Maximum exported event count + #[arg(long = "max-events", default_value_t = 1_024)] + pub max_events: usize, + + /// Confirm this explicit local L3 log capture + #[arg(long = "acknowledge-l3", required = true, action = clap::ArgAction::SetTrue)] + pub acknowledge_l3: bool, +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq, ValueEnum)] +pub enum ConnectLogsMode { + Batch, + Live, } /// `connect profile` options. @@ -608,6 +688,8 @@ pub enum CommandResult { ConnectLicense(ConnectLicenseCommands), /// Consent-bound local Connect profile export ConnectProfile(ConnectProfileOpts), + /// Consent-bound local Connect log export + ConnectLogs(ConnectLogsOpts), } /// Create default ServerOpts from environment variables @@ -829,6 +911,42 @@ mod tests { assert!(error.to_string().contains("--acknowledge-l3")); } + #[test] + fn connect_logs_requires_explicit_l3_acknowledgement() { + let arguments = [ + "rustfs", + "connect", + "logs", + "--state-dir", + "/var/lib/rustfs/connect", + "--output", + "/tmp/logs.zip", + "--organization", + "organizations/019e3ae0-0000-7000-8000-000000000001", + "--cluster", + "organizations/019e3ae0-0000-7000-8000-000000000001/clusters/019e3ae0-0000-7000-8000-000000000002", + "--device", + "organizations/019e3ae0-0000-7000-8000-000000000001/clusters/019e3ae0-0000-7000-8000-000000000002/clusterDevices/019e3ae0-0000-7000-8000-000000000003", + "--run-uid", + "019e3ae0-0000-7000-8000-000000000004", + "--artifact-uid", + "019e3ae0-0000-7000-8000-000000000005", + "--consent-uid", + "019e3ae0-0000-7000-8000-000000000006", + "--policy-revision", + "1", + "--consent-expires-at", + "4102444800", + "--expires-at", + "4102444700", + "--duration-millis", + "1000", + ]; + let error = Cli::try_parse_from(arguments).expect_err("unacknowledged log capture must fail"); + assert_eq!(error.kind(), ErrorKind::MissingRequiredArgument); + assert!(error.to_string().contains("--acknowledge-l3")); + } + #[test] fn server_help_lists_allocator_reclaim_environment() { let result = Cli::try_parse_from(["rustfs", "server", "--help"]); diff --git a/rustfs/src/config/mod.rs b/rustfs/src/config/mod.rs index 7fb451c36..e340642dd 100644 --- a/rustfs/src/config/mod.rs +++ b/rustfs/src/config/mod.rs @@ -52,6 +52,7 @@ mod config_test; // Re-export public types pub use cli::{CommandResult, InfoOpts, InfoType}; pub use cli::{ConnectLicenseArtifactOpts, ConnectLicenseCommands, ConnectLicenseScopeOpts}; +pub use cli::{ConnectLogsMode, ConnectLogsOpts}; pub use cli::{ConnectProfileOpts, ConnectProfileTool, ConnectThreadProfileScope}; pub use cli::{DiagnoseFormat, DiagnoseOpts}; pub use cli::{InspectBucketMetaOpts, InspectCommands, InspectOpts}; diff --git a/rustfs/src/config/opt.rs b/rustfs/src/config/opt.rs index 6e5fa050f..8387f7c43 100644 --- a/rustfs/src/config/opt.rs +++ b/rustfs/src/config/opt.rs @@ -141,6 +141,7 @@ impl Opt { ConnectCommands::Register(opts) => Ok(CommandResult::ConnectRegister(opts)), ConnectCommands::License(opts) => Ok(CommandResult::ConnectLicense(opts.command)), ConnectCommands::Profile(opts) => Ok(CommandResult::ConnectProfile(opts)), + ConnectCommands::Logs(opts) => Ok(CommandResult::ConnectLogs(opts)), }, Some(Commands::Server(opts)) => Self::server_command_result(Self::from_server_opts(*opts)), None => { diff --git a/rustfs/src/connect/diagnostics/logs.rs b/rustfs/src/connect/diagnostics/logs.rs new file mode 100644 index 000000000..01270fe34 --- /dev/null +++ b/rustfs/src/connect/diagnostics/logs.rs @@ -0,0 +1,879 @@ +// 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. + +//! Bounded capture of allow-listed RustFS structured log events. +//! +//! The collector reads only the active file configured for RustFS local JSON +//! logging. It exports timestamp offsets, severity, and one of three reviewed +//! event IDs. Raw messages, paths, headers, fields, and malformed lines are +//! counted as dropped and never enter the signed L3 artifact. + +use std::fs::{self, File, OpenOptions}; +use std::io::{Cursor, Read as _, Seek as _, SeekFrom, Write as _}; +use std::path::{Path, PathBuf}; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; + +use base64_simd::URL_SAFE_NO_PAD; +use p256::ecdsa::{Signature, SigningKey, signature::Signer as _}; +use p256::pkcs8::DecodePrivateKey as _; +use serde::Serialize; +use serde_json::Value; +use sha2::{Digest as _, Sha256}; +use thiserror::Error; +use time::{OffsetDateTime, format_description::well_known::Rfc3339}; +use tokio_util::sync::CancellationToken; +use uuid::{Uuid, Variant, Version}; +use zip::{CompressionMethod, ZipWriter, write::SimpleFileOptions}; + +use crate::connect::DeviceIdentity; + +pub const LOGS_SCHEMA_VERSION: u16 = 1; +pub const LOGS_CAPABILITY: &str = "logs.capture@1"; +pub const MAX_CAPTURE_DURATION: Duration = Duration::from_secs(30); +pub const MAX_EVENTS: usize = 1_024; +pub const MAX_RESULT_BYTES: usize = 262_144; +pub const MAX_SOURCE_BYTES: usize = 1_048_576; +pub const MAX_LINE_BYTES: usize = 4_096; +pub const MAX_BUILD_FEATURES: usize = 64; +pub const MAX_ARCHIVE_BYTES: usize = 524_288; +pub const MAX_DECOMPRESSED_BYTES: usize = 278_528; +pub const MAX_ENVELOPE_BYTES: usize = 16_384; + +const SIGNATURE_DOMAIN: &[u8] = b"rustfs-diagnostic-envelope-v1\0"; +const ENVELOPE_PATH: &str = "envelope.json"; +const SIGNATURE_PATH: &str = "envelope.sig"; +const RESULT_PATH: &str = "result.json"; +const OUTPUT_MODE: u32 = 0o600; +const MAX_VALIDITY_SECONDS: i64 = 2_592_000; +const MAX_FUTURE_SKEW_SECONDS: i64 = 300; +const POLL_INTERVAL: Duration = Duration::from_millis(50); + +static LOG_COLLECTOR_ACTIVE: AtomicBool = AtomicBool::new(false); + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum CaptureMode { + Batch, + Live, +} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct LocalLogConsent { + pub consent_uid: String, + pub policy_revision: u64, + pub expires_at_unix: i64, + pub confirmed: bool, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct LogProvenance { + repository: &'static str, + source_commit: String, + executable_sha256: String, + rustfs_version: String, + os_family: OsFamily, + architecture: Architecture, + build_features: Vec, +} + +impl LogProvenance { + pub fn new( + source_commit: impl Into, + executable_sha256: impl Into, + rustfs_version: impl Into, + build_features: Vec, + ) -> Self { + Self { + repository: "rustfs/rustfs", + source_commit: source_commit.into(), + executable_sha256: executable_sha256.into(), + rustfs_version: rustfs_version.into(), + os_family: OsFamily::current(), + architecture: Architecture::current(), + build_features, + } + } +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize)] +#[serde(rename_all = "SCREAMING_SNAKE_CASE")] +enum OsFamily { + Linux, + Darwin, + Windows, + Freebsd, + Other, +} + +impl OsFamily { + fn current() -> Self { + match std::env::consts::OS { + "linux" => Self::Linux, + "macos" => Self::Darwin, + "windows" => Self::Windows, + "freebsd" => Self::Freebsd, + _ => Self::Other, + } + } +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize)] +#[serde(rename_all = "lowercase")] +enum Architecture { + #[serde(rename = "x86_64")] + X86_64, + Aarch64, + Other, +} + +impl Architecture { + fn current() -> Self { + match std::env::consts::ARCH { + "x86_64" => Self::X86_64, + "aarch64" => Self::Aarch64, + _ => Self::Other, + } + } +} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct LogCaptureRequest { + pub organization_name: String, + pub cluster_name: String, + pub device_name: String, + pub run_uid: String, + pub artifact_uid: String, + pub schema_version: u16, + pub capability: String, + pub consent: LocalLogConsent, + pub produced_at_unix: i64, + pub expires_at_unix: i64, + pub nonce: [u8; 32], + pub mode: CaptureMode, + pub duration: Duration, + pub max_events: usize, + pub provenance: LogProvenance, +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize)] +#[serde(rename_all = "SCREAMING_SNAKE_CASE")] +pub enum LogOutcome { + Succeeded, +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize)] +#[serde(rename_all = "SCREAMING_SNAKE_CASE")] +pub enum LogReasonCode { + Complete, +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize)] +#[serde(rename_all = "SCREAMING_SNAKE_CASE")] +pub enum LogSeverity { + Info, + Warn, + Error, +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize)] +#[serde(rename_all = "SCREAMING_SNAKE_CASE")] +pub enum LogEventId { + DriveUnavailable, + RequestFailed, + ServiceStarted, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct CapturedLogEvent { + offset_millis: u64, + severity: LogSeverity, + event_id: LogEventId, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize)] +#[serde(rename_all = "camelCase")] +struct LogData { + events: Vec, + dropped_event_count: u64, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize)] +#[serde(rename_all = "camelCase")] +struct Coverage { + requested_units: u32, + completed_units: u32, + unit: &'static str, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize)] +#[serde(rename_all = "camelCase")] +struct LogResult { + schema_version: u16, + run_uid: String, + tool_id: &'static str, + capability: &'static str, + outcome: LogOutcome, + reason_code: LogReasonCode, + duration_millis: u64, + provenance: LogProvenance, + coverage: Coverage, + data: LogData, +} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct SignedLogExport { + pub artifact_uid: String, + pub archive_bytes: Vec, + pub archive_sha256: String, + pub event_count: usize, + pub dropped_event_count: u64, +} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct SavedLogExport { + pub artifact_uid: String, + pub archive_size_bytes: u64, + pub archive_sha256: String, +} + +#[derive(Debug, Error)] +pub enum LogCaptureError { + #[error("logs_local_consent_required")] + ConsentRequired, + #[error("logs_local_consent_expired")] + ConsentExpired, + #[error("logs_request_expired")] + Expired, + #[error("logs_invalid_request")] + InvalidRequest, + #[error("logs_unsupported_version")] + UnsupportedVersion, + #[error("logs_unsupported_capability")] + UnsupportedCapability, + #[error("logs_limit_exceeded")] + LimitExceeded, + #[error("logs_collection_cancelled")] + Cancelled, + #[error("logs_collection_already_running")] + Busy, + #[error("logs_file_source_unavailable")] + SourceUnavailable, + #[error("logs_export_signing_failed")] + Signing, + #[error("logs_export_exists")] + AlreadyExists, + #[error("logs_io_failed")] + Io(#[source] std::io::Error), + #[error("logs_export_encoding_failed")] + Encoding, + #[error("logs_export_durability_failed_after_commit")] + DurabilityAfterCommit(#[source] std::io::Error), +} + +#[derive(Clone, Debug)] +pub(crate) struct ConfiguredLogSource { + path: PathBuf, +} + +impl ConfiguredLogSource { + pub(crate) fn discover() -> Result { + let config = rustfs_obs::OtelConfig::new(); + let directory = config.log_directory.ok_or(LogCaptureError::SourceUnavailable)?; + let filename = config.log_filename.ok_or(LogCaptureError::SourceUnavailable)?; + Self::new(directory, filename) + } + + pub(crate) fn new(directory: impl AsRef, filename: impl AsRef) -> Result { + let filename = filename.as_ref(); + if filename.is_absolute() || filename.file_name() != Some(filename.as_os_str()) { + return Err(LogCaptureError::SourceUnavailable); + } + let directory = fs::canonicalize(directory).map_err(|_| LogCaptureError::SourceUnavailable)?; + let path = directory.join(filename); + let metadata = fs::symlink_metadata(&path).map_err(|_| LogCaptureError::SourceUnavailable)?; + if metadata.file_type().is_symlink() || !metadata.is_file() { + return Err(LogCaptureError::SourceUnavailable); + } + let canonical = fs::canonicalize(&path).map_err(|_| LogCaptureError::SourceUnavailable)?; + if canonical.parent() != Some(directory.as_path()) { + return Err(LogCaptureError::SourceUnavailable); + } + Ok(Self { path: canonical }) + } + + fn len(&self) -> Result { + fs::metadata(&self.path) + .map(|metadata| metadata.len()) + .map_err(LogCaptureError::Io) + } + + fn read(&self, start: u64, limit: usize) -> Result { + let mut file = open_read_only(&self.path)?; + let length = file.metadata().map_err(LogCaptureError::Io)?.len(); + let start = start.min(length); + file.seek(SeekFrom::Start(start)).map_err(LogCaptureError::Io)?; + let remaining = usize::try_from(length.saturating_sub(start)).unwrap_or(usize::MAX); + let mut bytes = Vec::with_capacity(limit.min(remaining)); + file.take(limit as u64).read_to_end(&mut bytes).map_err(LogCaptureError::Io)?; + let bytes_read = bytes.len() as u64; + Ok(SourceChunk { + bytes, + next_offset: start.saturating_add(bytes_read), + truncated_prefix: false, + }) + } +} + +struct SourceChunk { + bytes: Vec, + next_offset: u64, + truncated_prefix: bool, +} + +struct CollectorLease; + +impl CollectorLease { + fn acquire() -> Result { + LOG_COLLECTOR_ACTIVE + .compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire) + .map(|_| Self) + .map_err(|_| LogCaptureError::Busy) + } +} + +impl Drop for CollectorLease { + fn drop(&mut self) { + LOG_COLLECTOR_ACTIVE.store(false, Ordering::Release); + } +} + +pub async fn export_logs( + request: &LogCaptureRequest, + key: &DeviceIdentity, + cancel: &CancellationToken, +) -> Result { + request.validate(unix_now()?)?; + check_cancel(cancel)?; + let source = ConfiguredLogSource::discover()?; + export_logs_from(request, key, cancel, &source).await +} + +pub(crate) async fn export_logs_from( + request: &LogCaptureRequest, + key: &DeviceIdentity, + cancel: &CancellationToken, + source: &ConfiguredLogSource, +) -> Result { + request.validate(unix_now()?)?; + check_cancel(cancel)?; + let _lease = CollectorLease::acquire()?; + let started = Instant::now(); + let capture = match request.mode { + CaptureMode::Batch => capture_batch(source, cancel)?, + CaptureMode::Live => capture_live(source, request.duration, cancel).await?, + }; + check_cancel(cancel)?; + + let (events, dropped_event_count) = parse_events(capture.bytes, capture.truncated_prefix, request); + let result = LogResult { + schema_version: LOGS_SCHEMA_VERSION, + run_uid: request.run_uid.clone(), + tool_id: "logs.capture", + capability: LOGS_CAPABILITY, + outcome: LogOutcome::Succeeded, + reason_code: LogReasonCode::Complete, + duration_millis: u64::try_from(started.elapsed().as_millis()) + .unwrap_or(u64::MAX) + .max(1) + .min(30_000), + provenance: request.provenance.clone(), + coverage: Coverage { + requested_units: 1, + completed_units: 1, + unit: "WINDOW", + }, + data: LogData { + events, + dropped_event_count, + }, + }; + encode_signed_export(request, result, key, cancel) +} + +fn capture_batch(source: &ConfiguredLogSource, cancel: &CancellationToken) -> Result { + check_cancel(cancel)?; + let length = source.len()?; + let start = length.saturating_sub(MAX_SOURCE_BYTES as u64); + let mut chunk = source.read(start, MAX_SOURCE_BYTES)?; + chunk.truncated_prefix = start > 0; + Ok(chunk) +} + +async fn capture_live( + source: &ConfiguredLogSource, + duration: Duration, + cancel: &CancellationToken, +) -> Result { + let deadline = Instant::now() + duration; + let mut offset = source.len()?; + let mut bytes = Vec::new(); + let mut truncated_prefix = false; + loop { + check_cancel(cancel)?; + let length = source.len()?; + if length < offset { + offset = 0; + truncated_prefix = true; + } + if length > offset { + let remaining = MAX_SOURCE_BYTES.saturating_sub(bytes.len()); + if remaining == 0 { + truncated_prefix = true; + break; + } + let chunk = source.read(offset, remaining)?; + offset = chunk.next_offset; + bytes.extend_from_slice(&chunk.bytes); + truncated_prefix |= chunk.truncated_prefix; + } + if Instant::now() >= deadline { + break; + } + tokio::select! { + () = cancel.cancelled() => return Err(LogCaptureError::Cancelled), + () = tokio::time::sleep(POLL_INTERVAL.min(deadline.saturating_duration_since(Instant::now()))) => {} + } + } + Ok(SourceChunk { + bytes, + next_offset: offset, + truncated_prefix, + }) +} + +fn parse_events(bytes: Vec, truncated_prefix: bool, request: &LogCaptureRequest) -> (Vec, u64) { + let mut dropped = u64::from(truncated_prefix); + let mut parsed = Vec::new(); + let batch_end_millis = request.produced_at_unix.saturating_mul(1_000); + let batch_start_millis = + batch_end_millis.saturating_sub(i64::try_from(request.duration.as_millis()).unwrap_or(i64::MAX).max(1)); + for (index, line) in bytes.split(|byte| *byte == b'\n').enumerate() { + if line.is_empty() || (truncated_prefix && index == 0) { + continue; + } + if line.len() > MAX_LINE_BYTES { + dropped = dropped.saturating_add(1); + continue; + } + let Ok(value) = serde_json::from_slice::(line) else { + dropped = dropped.saturating_add(1); + continue; + }; + let Some(event) = typed_event(&value) else { + dropped = dropped.saturating_add(1); + continue; + }; + if request.mode == CaptureMode::Batch + && (event.offset_millis < u64::try_from(batch_start_millis).unwrap_or(0) + || event.offset_millis > u64::try_from(batch_end_millis).unwrap_or(0)) + { + dropped = dropped.saturating_add(1); + continue; + } + if parsed.len() == request.max_events { + dropped = dropped.saturating_add(1); + continue; + } + parsed.push(event); + } + let base = parsed.first().map_or(0, |event| event.offset_millis); + for event in &mut parsed { + event.offset_millis = event.offset_millis.saturating_sub(base); + } + (parsed, dropped) +} + +fn typed_event(value: &Value) -> Option { + let fields = value.as_object()?; + let timestamp = fields.get("timestamp")?.as_str()?; + let timestamp = OffsetDateTime::parse(timestamp, &Rfc3339).ok()?; + let timestamp_millis = u64::try_from(timestamp.unix_timestamp_nanos() / 1_000_000).ok()?; + let severity = match fields.get("level")?.as_str()? { + "INFO" => LogSeverity::Info, + "WARN" => LogSeverity::Warn, + "ERROR" => LogSeverity::Error, + _ => return None, + }; + let event_id = match fields.get("event")?.as_str()? { + "drive_unavailable" => LogEventId::DriveUnavailable, + "rpc_request_failed" | "admin_request_failed" | "http_request_failed" => LogEventId::RequestFailed, + "http_startup_endpoints" => LogEventId::ServiceStarted, + _ => return None, + }; + Some(CapturedLogEvent { + offset_millis: timestamp_millis, + severity, + event_id, + }) +} + +fn encode_signed_export( + request: &LogCaptureRequest, + result: LogResult, + key: &DeviceIdentity, + cancel: &CancellationToken, +) -> Result { + check_cancel(cancel)?; + if unix_now()? >= request.expires_at_unix { + return Err(LogCaptureError::Expired); + } + let event_count = result.data.events.len(); + let dropped_event_count = result.data.dropped_event_count; + let result_bytes = serde_json::to_vec(&result).map_err(|_| LogCaptureError::Encoding)?; + if result_bytes.is_empty() || result_bytes.len() > MAX_RESULT_BYTES { + return Err(LogCaptureError::LimitExceeded); + } + let device_key_id = hex_lower(&Sha256::digest(key.public_key_der())); + let envelope = Envelope { + format_version: "rustfs.connect.diagnosticEnvelope/1", + protocol_version: "v1", + organization_name: &request.organization_name, + cluster_name: &request.cluster_name, + device_name: &request.device_name, + run_uid: &request.run_uid, + artifact_uid: &request.artifact_uid, + tool_id: "logs.capture", + schema_version: LOGS_SCHEMA_VERSION, + classification: "L3", + consent_uid: &request.consent.consent_uid, + policy_revision: request.consent.policy_revision, + produced_at: timestamp(request.produced_at_unix)?, + expires_at: timestamp(request.expires_at_unix)?, + nonce: URL_SAFE_NO_PAD.encode_to_string(request.nonce), + device_key_id: &device_key_id, + payload: Payload { + path: RESULT_PATH, + media_type: "application/json", + size_bytes: result_bytes.len() as u64, + sha256: hex_lower(&Sha256::digest(&result_bytes)), + }, + }; + let envelope_bytes = serde_json::to_vec(&envelope).map_err(|_| LogCaptureError::Encoding)?; + if envelope_bytes.is_empty() || envelope_bytes.len() > MAX_ENVELOPE_BYTES { + return Err(LogCaptureError::LimitExceeded); + } + let signature_bytes = signature_document(key, &device_key_id, &envelope_bytes)?; + let decompressed = result_bytes + .len() + .checked_add(envelope_bytes.len()) + .and_then(|size| size.checked_add(signature_bytes.len())) + .ok_or(LogCaptureError::LimitExceeded)?; + if decompressed > MAX_DECOMPRESSED_BYTES { + return Err(LogCaptureError::LimitExceeded); + } + check_cancel(cancel)?; + let archive_bytes = archive(&envelope_bytes, &signature_bytes, &result_bytes)?; + if archive_bytes.len() > MAX_ARCHIVE_BYTES { + return Err(LogCaptureError::LimitExceeded); + } + Ok(SignedLogExport { + artifact_uid: request.artifact_uid.clone(), + archive_sha256: hex_lower(&Sha256::digest(&archive_bytes)), + archive_bytes, + event_count, + dropped_event_count, + }) +} + +pub fn save_signed_log_export( + output: &Path, + export: &SignedLogExport, + cancel: &CancellationToken, +) -> Result { + check_cancel(cancel)?; + if !uuid7(&export.artifact_uid) { + return Err(LogCaptureError::InvalidRequest); + } + let parent = output + .parent() + .filter(|path| !path.as_os_str().is_empty()) + .unwrap_or_else(|| Path::new(".")); + let filename = output.file_name().ok_or(LogCaptureError::InvalidRequest)?.to_string_lossy(); + let temporary = parent.join(format!(".{filename}.{}.partial", export.artifact_uid)); + let mut options = OpenOptions::new(); + options.write(true).create_new(true); + #[cfg(unix)] + { + use std::os::unix::fs::OpenOptionsExt as _; + options.mode(OUTPUT_MODE); + } + let mut file = options.open(&temporary).map_err(map_create_error)?; + let result = (|| { + file.write_all(&export.archive_bytes).map_err(LogCaptureError::Io)?; + check_cancel(cancel)?; + file.sync_all().map_err(LogCaptureError::Io)?; + fs::hard_link(&temporary, output).map_err(map_publish_error)?; + if let Err(error) = fs::remove_file(&temporary) { + return Err(LogCaptureError::DurabilityAfterCommit(error)); + } + #[cfg(unix)] + if let Err(error) = File::open(parent).and_then(|directory| directory.sync_all()) { + return Err(LogCaptureError::DurabilityAfterCommit(error)); + } + Ok(SavedLogExport { + artifact_uid: export.artifact_uid.clone(), + archive_size_bytes: export.archive_bytes.len() as u64, + archive_sha256: export.archive_sha256.clone(), + }) + })(); + if result.is_err() { + let _ = fs::remove_file(&temporary); + } + result +} + +impl LogCaptureRequest { + fn validate(&self, now_unix: i64) -> Result<(), LogCaptureError> { + if self.schema_version != LOGS_SCHEMA_VERSION { + return Err(LogCaptureError::UnsupportedVersion); + } + if self.capability != LOGS_CAPABILITY { + return Err(LogCaptureError::UnsupportedCapability); + } + if !self.consent.confirmed || self.consent.policy_revision == 0 { + return Err(LogCaptureError::ConsentRequired); + } + if self.consent.expires_at_unix <= now_unix || self.expires_at_unix > self.consent.expires_at_unix { + return Err(LogCaptureError::ConsentExpired); + } + let validity = self + .expires_at_unix + .checked_sub(self.produced_at_unix) + .ok_or(LogCaptureError::Expired)?; + if self.produced_at_unix > now_unix.saturating_add(MAX_FUTURE_SKEW_SECONDS) + || validity <= 0 + || self.expires_at_unix <= now_unix + || validity > MAX_VALIDITY_SECONDS + { + return Err(LogCaptureError::Expired); + } + if self.duration.is_zero() || self.duration > MAX_CAPTURE_DURATION || self.max_events == 0 || self.max_events > MAX_EVENTS + { + return Err(LogCaptureError::LimitExceeded); + } + if !uuid7(&self.run_uid) + || !uuid7(&self.artifact_uid) + || !uuid7(&self.consent.consent_uid) + || !resource_names_match(self) + || !lower_hex(&self.provenance.source_commit, 40) + || !lower_hex(&self.provenance.executable_sha256, 64) + || !version(&self.provenance.rustfs_version) + || self.provenance.build_features.len() > MAX_BUILD_FEATURES + || !self.provenance.build_features.iter().all(|feature| build_feature(feature)) + { + return Err(LogCaptureError::InvalidRequest); + } + Ok(()) + } +} + +#[derive(Serialize)] +#[serde(rename_all = "camelCase")] +struct Envelope<'a> { + format_version: &'static str, + protocol_version: &'static str, + organization_name: &'a str, + cluster_name: &'a str, + device_name: &'a str, + run_uid: &'a str, + artifact_uid: &'a str, + tool_id: &'static str, + schema_version: u16, + classification: &'static str, + consent_uid: &'a str, + policy_revision: u64, + produced_at: String, + expires_at: String, + nonce: String, + device_key_id: &'a str, + payload: Payload, +} + +#[derive(Serialize)] +#[serde(rename_all = "camelCase")] +struct Payload { + path: &'static str, + media_type: &'static str, + size_bytes: u64, + sha256: String, +} + +#[derive(Serialize)] +#[serde(rename_all = "camelCase")] +struct SignatureDocument<'a> { + algorithm: &'static str, + key_id: &'a str, + value: String, +} + +fn signature_document(key: &DeviceIdentity, key_id: &str, envelope: &[u8]) -> Result, LogCaptureError> { + let pkcs8 = key.to_pkcs8_der().map_err(|_| LogCaptureError::Signing)?; + let signing_key = SigningKey::from_pkcs8_der(pkcs8.as_slice()).map_err(|_| LogCaptureError::Signing)?; + let mut input = Vec::with_capacity(SIGNATURE_DOMAIN.len() + envelope.len()); + input.extend_from_slice(SIGNATURE_DOMAIN); + input.extend_from_slice(envelope); + let signature: Signature = signing_key.sign(&input); + let value = URL_SAFE_NO_PAD.encode_to_string(signature.normalize_s().to_bytes()); + serde_json::to_vec(&SignatureDocument { + algorithm: "ES256", + key_id, + value, + }) + .map_err(|_| LogCaptureError::Encoding) +} + +fn archive(envelope: &[u8], signature: &[u8], result: &[u8]) -> Result, LogCaptureError> { + let cursor = Cursor::new(Vec::with_capacity(envelope.len() + signature.len() + result.len() + 512)); + let mut writer = ZipWriter::new(cursor); + let options = SimpleFileOptions::DEFAULT + .compression_method(CompressionMethod::Stored) + .unix_permissions(OUTPUT_MODE); + for (name, bytes) in [(ENVELOPE_PATH, envelope), (SIGNATURE_PATH, signature), (RESULT_PATH, result)] { + writer.start_file(name, options).map_err(|_| LogCaptureError::Encoding)?; + writer.write_all(bytes).map_err(LogCaptureError::Io)?; + } + writer + .finish() + .map(|cursor| cursor.into_inner()) + .map_err(|_| LogCaptureError::Encoding) +} + +fn open_read_only(path: &Path) -> Result { + let mut options = OpenOptions::new(); + options.read(true); + #[cfg(unix)] + { + use std::os::unix::fs::OpenOptionsExt as _; + options.custom_flags(libc::O_NOFOLLOW); + } + options.open(path).map_err(LogCaptureError::Io) +} + +fn resource_names_match(request: &LogCaptureRequest) -> bool { + let Some(organization_uid) = request.organization_name.strip_prefix("organizations/") else { + return false; + }; + if !uuid7(organization_uid) { + return false; + } + let cluster_prefix = format!("{}/clusters/", request.organization_name); + let Some(cluster_uid) = request.cluster_name.strip_prefix(&cluster_prefix) else { + return false; + }; + if !uuid7(cluster_uid) { + return false; + } + let device_prefix = format!("{}/clusterDevices/", request.cluster_name); + request.device_name.strip_prefix(&device_prefix).is_some_and(uuid7) +} + +fn uuid7(value: &str) -> bool { + Uuid::parse_str(value).is_ok_and(|uuid| { + uuid.get_version() == Some(Version::SortRand) && uuid.get_variant() == Variant::RFC4122 && uuid.to_string() == value + }) +} + +fn lower_hex(value: &str, length: usize) -> bool { + value.len() == length + && value + .bytes() + .all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte)) +} + +fn version(value: &str) -> bool { + if value.is_empty() + || value.len() > 64 + || !value + .bytes() + .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'.' | b'-')) + { + return false; + } + let (core, suffix) = value + .split_once('-') + .map_or((value, None), |(core, suffix)| (core, Some(suffix))); + if suffix.is_some_and(str::is_empty) { + return false; + } + let mut parts = core.split('.'); + parts.clone().count() == 3 && parts.all(|part| !part.is_empty() && part.bytes().all(|byte| byte.is_ascii_digit())) +} + +fn build_feature(value: &str) -> bool { + !value.is_empty() + && value.len() <= 64 + && value.as_bytes()[0].is_ascii_lowercase() + && value + .bytes() + .all(|byte| byte.is_ascii_lowercase() || byte.is_ascii_digit() || matches!(byte, b'_' | b'-')) +} + +fn timestamp(unix: i64) -> Result { + OffsetDateTime::from_unix_timestamp(unix) + .map_err(|_| LogCaptureError::InvalidRequest)? + .format(&Rfc3339) + .map_err(|_| LogCaptureError::InvalidRequest) +} + +fn unix_now() -> Result { + let duration = SystemTime::now() + .duration_since(UNIX_EPOCH) + .map_err(|_| LogCaptureError::InvalidRequest)?; + i64::try_from(duration.as_secs()).map_err(|_| LogCaptureError::InvalidRequest) +} + +fn check_cancel(cancel: &CancellationToken) -> Result<(), LogCaptureError> { + if cancel.is_cancelled() { + Err(LogCaptureError::Cancelled) + } else { + Ok(()) + } +} + +fn hex_lower(bytes: &[u8]) -> String { + let mut value = String::with_capacity(bytes.len() * 2); + for byte in bytes { + use std::fmt::Write as _; + write!(&mut value, "{byte:02x}").expect("writing hexadecimal to a string cannot fail"); + } + value +} + +fn map_create_error(error: std::io::Error) -> LogCaptureError { + if error.kind() == std::io::ErrorKind::AlreadyExists { + LogCaptureError::AlreadyExists + } else { + LogCaptureError::Io(error) + } +} + +fn map_publish_error(error: std::io::Error) -> LogCaptureError { + if error.kind() == std::io::ErrorKind::AlreadyExists { + LogCaptureError::AlreadyExists + } else { + LogCaptureError::Io(error) + } +} diff --git a/rustfs/src/connect/diagnostics/mod.rs b/rustfs/src/connect/diagnostics/mod.rs index c60eafc67..278cef299 100644 --- a/rustfs/src/connect/diagnostics/mod.rs +++ b/rustfs/src/connect/diagnostics/mod.rs @@ -12,11 +12,16 @@ // See the License for the specific language governing permissions and // limitations under the License. +mod logs; mod profile_cpu; mod profile_memory; mod profile_threads; mod schedule; +pub use logs::{ + CaptureMode, LOGS_CAPABILITY, LOGS_SCHEMA_VERSION, LocalLogConsent, LogCaptureError, LogCaptureRequest, LogProvenance, + SavedLogExport, SignedLogExport, export_logs, save_signed_log_export, +}; pub use profile_cpu::{ CPU_PROFILE_CAPABILITY, LocalProfileConsent, MAX_PROFILE_DURATION, MEMORY_PROFILE_CAPABILITY, PROFILE_SCHEMA_VERSION, ProfileCaptureRequest, ProfileData, ProfileError, ProfileOutcome, ProfileProvenance, ProfileReasonCode, ProfileResult, diff --git a/rustfs/src/connect/mod.rs b/rustfs/src/connect/mod.rs index 2f8d07b06..1991d2b13 100644 --- a/rustfs/src/connect/mod.rs +++ b/rustfs/src/connect/mod.rs @@ -45,12 +45,14 @@ pub use client::{ClientError, ConnectClient, ConnectConfig}; pub use config::{HeartbeatConfig, HeartbeatConfigError, HeartbeatSchedule}; pub use credential_store::{CredentialStore, DeviceCredential}; pub use diagnostics::{ - CPU_PROFILE_CAPABILITY, DiagnosticCollectionPolicy, DiagnosticReceipt, DiagnosticScheduleError, DiagnosticScheduleRuntime, - DiagnosticScheduleStatus, LocalProfileConsent, MAX_PROFILE_DURATION, MEMORY_PROFILE_CAPABILITY, PROFILE_SCHEMA_VERSION, - ProfileCaptureRequest, ProfileData, ProfileError, ProfileOutcome, ProfileProvenance, ProfileReasonCode, ProfileResult, - ProfileTool, ReceiptOutcome, SavedProfileExport, SignedProfileExport, THREAD_PROFILE_CAPABILITY, ThreadProfileScope, - capture_cpu_profile, capture_thread_profile, encode_signed_profile_export, export_cpu_profile, export_memory_profile, - export_thread_profile, run_local_environment_once, save_signed_profile_export, spawn_environment_schedule, + CPU_PROFILE_CAPABILITY, CaptureMode, DiagnosticCollectionPolicy, DiagnosticReceipt, DiagnosticScheduleError, + DiagnosticScheduleRuntime, DiagnosticScheduleStatus, LOGS_CAPABILITY, LOGS_SCHEMA_VERSION, LocalLogConsent, + LocalProfileConsent, LogCaptureError, LogCaptureRequest, LogProvenance, MAX_PROFILE_DURATION, MEMORY_PROFILE_CAPABILITY, + PROFILE_SCHEMA_VERSION, ProfileCaptureRequest, ProfileData, ProfileError, ProfileOutcome, ProfileProvenance, + ProfileReasonCode, ProfileResult, ProfileTool, ReceiptOutcome, SavedLogExport, SavedProfileExport, SignedLogExport, + SignedProfileExport, THREAD_PROFILE_CAPABILITY, ThreadProfileScope, capture_cpu_profile, capture_thread_profile, + encode_signed_profile_export, export_cpu_profile, export_logs, export_memory_profile, export_thread_profile, + run_local_environment_once, save_signed_log_export, save_signed_profile_export, spawn_environment_schedule, }; pub use environment::{ ENVIRONMENT_CAPABILITY, ENVIRONMENT_SCHEMA_VERSION, EnvironmentCollectionRequest, EnvironmentError, diff --git a/rustfs/src/startup_entrypoint.rs b/rustfs/src/startup_entrypoint.rs index bff08dd12..83e1286c2 100644 --- a/rustfs/src/startup_entrypoint.rs +++ b/rustfs/src/startup_entrypoint.rs @@ -14,8 +14,8 @@ use crate::{ config::{ - CommandResult, Config, ConnectLicenseCommands, ConnectLicenseScopeOpts, ConnectProfileOpts, ConnectProfileTool, - ConnectThreadProfileScope, Opt, + CommandResult, Config, ConnectLicenseCommands, ConnectLicenseScopeOpts, ConnectLogsMode, ConnectLogsOpts, + ConnectProfileOpts, ConnectProfileTool, ConnectThreadProfileScope, Opt, }, startup_lifecycle::{StartupRuntimeLifecycle, run_startup_runtime_lifecycle}, startup_preflight::{StartupServerPreflightError, bootstrap_external_prefix_compat, init_startup_server_preflight}, @@ -135,6 +135,7 @@ async fn async_main() -> Result<()> { } CommandResult::ConnectLicense(command) => return execute_connect_license(command), CommandResult::ConnectProfile(options) => return execute_connect_profile(options).await, + CommandResult::ConnectLogs(options) => return execute_connect_logs(options).await, CommandResult::Server(config) => config, }; @@ -164,6 +165,88 @@ async fn async_main() -> Result<()> { } } +async fn execute_connect_logs(options: ConnectLogsOpts) -> Result<()> { + use crate::connect::{ + CaptureMode, IdentityStore, LocalLogConsent, LogCaptureRequest, LogProvenance, export_logs, save_signed_log_export, + }; + use rand::{TryRng as _, rngs::SysRng}; + + let key = IdentityStore::new(options.state_dir.join("identity")) + .load() + .map_err(Error::other)? + .ok_or_else(|| Error::other("connect logs requires an enrolled device identity"))?; + let executable_sha256 = hash_current_executable()?; + let produced_at_unix = SystemTime::now() + .duration_since(UNIX_EPOCH) + .map_err(Error::other) + .and_then(|duration| i64::try_from(duration.as_secs()).map_err(Error::other))?; + let mut nonce = [0_u8; 32]; + SysRng.try_fill_bytes(&mut nonce).map_err(Error::other)?; + let request = LogCaptureRequest { + organization_name: options.organization, + cluster_name: options.cluster, + device_name: options.device, + run_uid: options.run_uid, + artifact_uid: options.artifact_uid, + schema_version: options.schema_version, + capability: options.capability, + consent: LocalLogConsent { + consent_uid: options.consent_uid, + policy_revision: options.policy_revision, + expires_at_unix: options.consent_expires_at_unix, + confirmed: options.acknowledge_l3, + }, + produced_at_unix, + expires_at_unix: options.expires_at_unix, + nonce, + mode: match options.mode { + ConnectLogsMode::Batch => CaptureMode::Batch, + ConnectLogsMode::Live => CaptureMode::Live, + }, + duration: Duration::from_millis(options.duration_millis), + max_events: options.max_events, + provenance: LogProvenance::new( + crate::version::build::COMMIT_HASH, + executable_sha256, + env!("CARGO_PKG_VERSION"), + enabled_build_features(), + ), + }; + let cancel = tokio_util::sync::CancellationToken::new(); + let capture = export_logs(&request, &key, &cancel); + tokio::pin!(capture); + let export = tokio::select! { + biased; + signal = tokio::signal::ctrl_c() => { + signal.map_err(Error::other)?; + cancel.cancel(); + return Err(Error::other("log collection cancelled")); + } + result = capture.as_mut() => result.map_err(Error::other)?, + }; + drop(capture); + let output = options.output; + let writer_cancel = cancel.clone(); + let mut writer = tokio::task::spawn_blocking(move || save_signed_log_export(&output, &export, &writer_cancel)); + let receipt = tokio::select! { + biased; + signal = tokio::signal::ctrl_c() => { + signal.map_err(Error::other)?; + cancel.cancel(); + writer.await.map_err(Error::other)?.map_err(Error::other)? + } + result = &mut writer => result.map_err(Error::other)?.map_err(Error::other)?, + }; + + println!("tool=logs.capture outcome=SUCCEEDED reason=COMPLETE"); + println!( + "artifact={} bytes={} sha256={}", + receipt.artifact_uid, receipt.archive_size_bytes, receipt.archive_sha256 + ); + println!("upload=not-performed"); + Ok(()) +} + async fn execute_connect_profile(options: ConnectProfileOpts) -> Result<()> { use crate::connect::{ IdentityStore, LocalProfileConsent, ProfileCaptureRequest, ProfileProvenance, ThreadProfileScope, export_cpu_profile, diff --git a/rustfs/tests/connect_logs.rs b/rustfs/tests/connect_logs.rs new file mode 100644 index 000000000..98ffd6921 --- /dev/null +++ b/rustfs/tests/connect_logs.rs @@ -0,0 +1,318 @@ +// 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. + +mod connect { + pub use rustfs::connect::DeviceIdentity; +} + +#[allow(dead_code)] +#[path = "../src/connect/diagnostics/logs.rs"] +mod logs; + +use std::fs::{self, OpenOptions}; +use std::io::{Cursor, Read as _, Write as _}; +#[cfg(unix)] +use std::os::unix::fs::{PermissionsExt as _, symlink}; +use std::sync::Mutex; +use std::time::{Duration, SystemTime, UNIX_EPOCH}; + +use base64_simd::URL_SAFE_NO_PAD; +use logs::{ + CaptureMode, ConfiguredLogSource, LOGS_CAPABILITY, LocalLogConsent, LogCaptureError, LogCaptureRequest, LogProvenance, + export_logs_from, save_signed_log_export, +}; +use p256::ecdsa::signature::Verifier as _; +use p256::ecdsa::{Signature, VerifyingKey}; +use p256::pkcs8::DecodePublicKey as _; +use sha2::{Digest as _, Sha256}; +use time::{Duration as TimeDuration, OffsetDateTime, format_description::well_known::Rfc3339}; +use tokio_util::sync::CancellationToken; +use zip::ZipArchive; + +static TEST_LOCK: Mutex<()> = Mutex::new(()); + +fn now() -> i64 { + SystemTime::now().duration_since(UNIX_EPOCH).expect("current time").as_secs() as i64 +} + +fn timestamp(unix: i64, offset_millis: i64) -> String { + (OffsetDateTime::from_unix_timestamp(unix).expect("valid timestamp") + TimeDuration::milliseconds(offset_millis)) + .format(&Rfc3339) + .expect("RFC3339 timestamp") +} + +fn request(mode: CaptureMode) -> LogCaptureRequest { + let now = now(); + let organization = "organizations/019e3ae0-0000-7000-8000-000000000011"; + let cluster = format!("{organization}/clusters/019e3ae0-0000-7000-8000-000000000012"); + LogCaptureRequest { + organization_name: organization.to_string(), + cluster_name: cluster.clone(), + device_name: format!("{cluster}/clusterDevices/019e3ae0-0000-7000-8000-000000000013"), + run_uid: "019e3ae0-0000-7000-8000-000000000014".to_string(), + artifact_uid: "019e3ae0-0000-7000-8000-000000000015".to_string(), + schema_version: 1, + capability: LOGS_CAPABILITY.to_string(), + consent: LocalLogConsent { + consent_uid: "019e3ae0-0000-7000-8000-000000000016".to_string(), + policy_revision: 7, + expires_at_unix: now + 120, + confirmed: true, + }, + produced_at_unix: now, + expires_at_unix: now + 60, + nonce: [0x5a; 32], + mode, + duration: Duration::from_millis(100), + max_events: 1_024, + provenance: LogProvenance::new("c".repeat(40), "d".repeat(64), "1.0.0-rc.6", vec!["gcs".to_string()]), + } +} + +fn line(timestamp: &str, level: &str, event: &str, extra: &str) -> String { + format!(r#"{{"timestamp":"{timestamp}","level":"{level}","event":"{event}","message":"{extra}"}}"#) +} + +fn source(contents: &str) -> (tempfile::TempDir, ConfiguredLogSource) { + let directory = tempfile::tempdir().expect("temporary log directory"); + fs::write(directory.path().join("rustfs.log"), contents).expect("write active log"); + let source = ConfiguredLogSource::new(directory.path(), "rustfs.log").expect("configured log source"); + (directory, source) +} + +fn archive_entry(archive: &mut ZipArchive>>, name: &str) -> Vec { + let mut entry = archive.by_name(name).expect("archive entry"); + let mut bytes = Vec::new(); + entry.read_to_end(&mut bytes).expect("read archive entry"); + bytes +} + +#[tokio::test] +async fn batch_capture_exports_only_allow_listed_fields_in_a_signed_artifact() { + let _guard = TEST_LOCK.lock().expect("test lock"); + let mut request = request(CaptureMode::Batch); + request.duration = Duration::from_secs(2); + let first_timestamp = timestamp(request.produced_at_unix - 1, 0); + let second_timestamp = timestamp(request.produced_at_unix - 1, 125); + let logs = [ + line( + &first_timestamp, + "INFO", + "http_startup_endpoints", + "endpoint=/srv/customer-a Authorization: Bearer SYNTHETIC_TOKEN_123", + ), + line( + &second_timestamp, + "ERROR", + "rpc_request_failed", + "Cookie: session=SYNTHETIC_SESSION_123\\nprivate/path", + ), + String::new(), + ] + .join("\n"); + let (_directory, source) = source(&logs); + let key = connect::DeviceIdentity::generate(); + let export = export_logs_from(&request, &key, &CancellationToken::new(), &source) + .await + .expect("signed logs export"); + assert_eq!(export.event_count, 2); + assert_eq!(export.dropped_event_count, 0); + assert_eq!(export.archive_sha256, hex(&Sha256::digest(&export.archive_bytes))); + + let mut archive = ZipArchive::new(Cursor::new(export.archive_bytes.clone())).expect("logs archive"); + assert_eq!(archive.len(), 3); + let envelope_bytes = archive_entry(&mut archive, "envelope.json"); + let signature_bytes = archive_entry(&mut archive, "envelope.sig"); + let result_bytes = archive_entry(&mut archive, "result.json"); + let envelope: serde_json::Value = serde_json::from_slice(&envelope_bytes).expect("envelope JSON"); + let signature: serde_json::Value = serde_json::from_slice(&signature_bytes).expect("signature JSON"); + let result: serde_json::Value = serde_json::from_slice(&result_bytes).expect("result JSON"); + + assert_eq!(envelope["classification"], "L3"); + assert_eq!(envelope["toolId"], "logs.capture"); + assert_eq!(envelope["payload"]["sha256"], hex(&Sha256::digest(&result_bytes))); + assert_eq!(result["outcome"], "SUCCEEDED"); + assert_eq!(result["provenance"]["buildFeatures"], serde_json::json!(["gcs"])); + assert_eq!(result["data"]["events"][0]["eventId"], "SERVICE_STARTED"); + assert_eq!(result["data"]["events"][0]["offsetMillis"], 0); + assert_eq!(result["data"]["events"][1]["eventId"], "REQUEST_FAILED"); + assert_eq!(result["data"]["events"][1]["offsetMillis"], 125); + let encoded = serde_json::to_string(&result).expect("encoded result"); + for forbidden in [ + "message", + "endpoint", + "/srv/", + "Authorization", + "SYNTHETIC_TOKEN", + "Cookie", + "private/path", + ] { + assert!(!encoded.contains(forbidden), "result leaked forbidden material: {forbidden}"); + } + + let raw_signature = URL_SAFE_NO_PAD + .decode_to_vec(signature["value"].as_str().expect("signature value")) + .expect("base64url signature"); + let signature_value = Signature::from_slice(&raw_signature).expect("P-256 signature"); + assert_eq!(signature_value.normalize_s(), signature_value); + let public = VerifyingKey::from_public_key_der(&key.public_key_der()).expect("public key"); + let mut input = b"rustfs-diagnostic-envelope-v1\0".to_vec(); + input.extend_from_slice(&envelope_bytes); + public + .verify(&input, &signature_value) + .expect("signature over exact envelope bytes"); +} + +#[tokio::test] +async fn batch_capture_drops_unknown_malformed_oversized_and_excess_events() { + let _guard = TEST_LOCK.lock().expect("test lock"); + let mut request = request(CaptureMode::Batch); + request.duration = Duration::from_secs(2); + request.max_events = 1; + let first_timestamp = timestamp(request.produced_at_unix - 1, 0); + let second_timestamp = timestamp(request.produced_at_unix, 0); + let logs = [ + "not-json".to_string(), + line(&first_timestamp, "DEBUG", "rpc_request_failed", "debug"), + line(&first_timestamp, "ERROR", "future_event", "unknown"), + format!( + "{{\"timestamp\":\"{first_timestamp}\",\"level\":\"ERROR\",\"event\":\"rpc_request_failed\",\"message\":\"{}\"}}", + "x".repeat(4_097), + ), + line(&first_timestamp, "ERROR", "rpc_request_failed", "first"), + line(&second_timestamp, "ERROR", "rpc_request_failed", "second"), + String::new(), + ] + .join("\n"); + let (_directory, source) = source(&logs); + let key = connect::DeviceIdentity::generate(); + let export = export_logs_from(&request, &key, &CancellationToken::new(), &source) + .await + .expect("bounded export"); + assert_eq!(export.event_count, 1); + assert_eq!(export.dropped_event_count, 5); +} + +#[tokio::test] +async fn live_capture_tails_new_events_and_honors_cancellation() { + let _guard = TEST_LOCK.lock().expect("test lock"); + let (directory, source) = source(""); + let key = connect::DeviceIdentity::generate(); + let mut request = request(CaptureMode::Live); + request.duration = Duration::from_millis(120); + let writer = async { + tokio::time::sleep(Duration::from_millis(20)).await; + let mut file = OpenOptions::new() + .append(true) + .open(directory.path().join("rustfs.log")) + .expect("open active log"); + writeln!( + file, + "{}", + line("2026-09-12T12:00:00Z", "WARN", "drive_unavailable", "secret=/tmp/customer") + ) + .expect("append live log"); + file.flush().expect("flush live log"); + }; + let capture_cancel = CancellationToken::new(); + let capture = export_logs_from(&request, &key, &capture_cancel, &source); + let ((), export) = tokio::join!(writer, capture); + assert_eq!(export.expect("live export").event_count, 1); + + let cancelled = CancellationToken::new(); + cancelled.cancel(); + assert!(matches!( + export_logs_from(&request, &key, &cancelled, &source).await, + Err(LogCaptureError::Cancelled) + )); +} + +#[tokio::test] +async fn consent_limits_and_source_boundary_fail_closed() { + let _guard = TEST_LOCK.lock().expect("test lock"); + let (directory, source) = source(&line("2026-09-12T12:00:00Z", "INFO", "http_startup_endpoints", "safe")); + let key = connect::DeviceIdentity::generate(); + let mut denied = request(CaptureMode::Batch); + denied.consent.confirmed = false; + assert!(matches!( + export_logs_from(&denied, &key, &CancellationToken::new(), &source).await, + Err(LogCaptureError::ConsentRequired) + )); + + let mut over_limit = request(CaptureMode::Batch); + over_limit.max_events = logs::MAX_EVENTS + 1; + assert!(matches!( + export_logs_from(&over_limit, &key, &CancellationToken::new(), &source).await, + Err(LogCaptureError::LimitExceeded) + )); + + assert!(matches!( + ConfiguredLogSource::new(directory.path(), "../rustfs.log"), + Err(LogCaptureError::SourceUnavailable) + )); + #[cfg(unix)] + { + let outside = tempfile::NamedTempFile::new().expect("outside log"); + symlink(outside.path(), directory.path().join("linked.log")).expect("symlink log"); + assert!(matches!( + ConfiguredLogSource::new(directory.path(), "linked.log"), + Err(LogCaptureError::SourceUnavailable) + )); + } +} + +#[tokio::test] +async fn local_export_is_private_no_clobber_and_cancel_safe() { + let _guard = TEST_LOCK.lock().expect("test lock"); + let (_directory, source) = source(&line("2026-09-12T12:00:00Z", "INFO", "http_startup_endpoints", "safe")); + let key = connect::DeviceIdentity::generate(); + let export = export_logs_from(&request(CaptureMode::Batch), &key, &CancellationToken::new(), &source) + .await + .expect("signed export"); + let output_directory = tempfile::tempdir().expect("output directory"); + let output = output_directory.path().join("logs.zip"); + let receipt = save_signed_log_export(&output, &export, &CancellationToken::new()).expect("save export"); + assert_eq!(receipt.archive_sha256, export.archive_sha256); + assert_eq!(receipt.archive_size_bytes, export.archive_bytes.len() as u64); + assert_eq!(fs::read(&output).expect("saved artifact"), export.archive_bytes); + #[cfg(unix)] + assert_eq!(fs::metadata(&output).expect("metadata").permissions().mode() & 0o777, 0o600); + assert!(matches!( + save_signed_log_export(&output, &export, &CancellationToken::new()), + Err(LogCaptureError::AlreadyExists) + )); + + let mut forged = export.clone(); + forged.artifact_uid = "../escape".to_string(); + assert!(matches!( + save_signed_log_export(&output_directory.path().join("forged.zip"), &forged, &CancellationToken::new()), + Err(LogCaptureError::InvalidRequest) + )); + + let cancelled = CancellationToken::new(); + cancelled.cancel(); + assert!(matches!( + save_signed_log_export(&output_directory.path().join("cancelled.zip"), &export, &cancelled), + Err(LogCaptureError::Cancelled) + )); +} + +fn hex(bytes: &[u8]) -> String { + let mut encoded = String::with_capacity(bytes.len() * 2); + for byte in bytes { + use std::fmt::Write as _; + write!(&mut encoded, "{byte:02x}").expect("hex encoding"); + } + encoded +}