diff --git a/pipeline-engines/src/data_update_service.rs b/pipeline-engines/src/data_update_service.rs index 6c6e83d..b6b3916 100644 --- a/pipeline-engines/src/data_update_service.rs +++ b/pipeline-engines/src/data_update_service.rs @@ -64,6 +64,7 @@ use std::thread::JoinHandle; use std::time::{Duration, Instant}; use chrono::{DateTime, Utc}; +use fiftyone_pipeline_core::redact::redact_with; use md5::{Digest, Md5}; use crate::data_file::AspectEngineDataFile; @@ -568,7 +569,22 @@ impl ServiceHandle { } /// Poll the remote update URL, verify, decompress, write and refresh. + /// + /// Every failing exit comes through here, so no error leaves the service + /// carrying the data file's licence keys, however the poll failed. fn check_for_update_from_url(&self, registration: &Arc) -> DataUpdateResult { + let keys = ®istration + .data_file + .configuration() + .data_update_license_keys; + self.poll_update_url(registration).map_err(|mut error| { + error.message = redact_with(&error.message, keys).into_owned(); + error + }) + } + + /// Poll the remote update URL, verify, decompress, write and refresh. + fn poll_update_url(&self, registration: &Arc) -> DataUpdateResult { let data_file = ®istration.data_file; let config = data_file.configuration(); let url = match data_file.data_update_url() { @@ -597,7 +613,10 @@ impl ServiceHandle { let response = request.send().map_err(|e| { DataUpdateError::new( AutoUpdateStatus::HttpsError, - format!("error accessing data update service at '{url}': {e}"), + format!( + "error accessing data update service at '{url}': {}", + request_error_message(e) + ), ) })?; @@ -645,7 +664,10 @@ impl ServiceHandle { let compressed = response.bytes().map_err(|e| { DataUpdateError::new( AutoUpdateStatus::StreamError, - format!("error reading data update response from '{url}': {e}"), + format!( + "error reading data update response from '{url}': {}", + request_error_message(e) + ), ) })?; @@ -739,6 +761,21 @@ fn log_poll(registration: &Arc, result: DataUpdateResult) { } } +/// The message of a failed request and each of its causes, without the URL +/// the request error names. That URL is the formatted one, whose query +/// carries the licence keys. +fn request_error_message(error: reqwest::Error) -> String { + let error = error.without_url(); + let mut message = error.to_string(); + let mut cause = std::error::Error::source(&error); + while let Some(source) = cause { + message.push_str(": "); + message.push_str(&source.to_string()); + cause = source.source(); + } + message +} + /// Format a timestamp as an HTTP-date for the `If-Modified-Since` header. fn httpdate_format(when: DateTime) -> String { // RFC 7231 IMF-fixdate, for example "Sun, 06 Nov 1994 08:49:37 GMT". diff --git a/pipeline-engines/tests/remote_refresh.rs b/pipeline-engines/tests/remote_refresh.rs index df5478b..8d7df45 100644 --- a/pipeline-engines/tests/remote_refresh.rs +++ b/pipeline-engines/tests/remote_refresh.rs @@ -32,7 +32,8 @@ //! built, so neither is this test. #![cfg(feature = "data-update")] -use std::io::Write; +use std::io::{Read, Write}; +use std::net::TcpListener; use std::sync::atomic::{AtomicUsize, Ordering}; use std::sync::{Arc, Mutex}; @@ -42,7 +43,7 @@ use fiftyone_pipeline_core::{ }; use fiftyone_pipeline_engines::{ AspectDataBase, AspectEngine, AspectEngineDataFile, AspectPropertyMetaData, AutoUpdateStatus, - DataFileConfiguration, DataUpdateService, OnPremiseAspectEngine, + DataFileConfiguration, DataUpdateService, FiftyOneUrlFormatter, OnPremiseAspectEngine, }; /// A minimal on-premise engine that records every refresh and the bytes it was @@ -259,3 +260,74 @@ fn update_from_memory_goes_to_engine() { assert_eq!(status, AutoUpdateStatus::Success); assert_eq!(*disk.lock().unwrap(), bytes); } + +/// The licence key the Distributor formatter writes into the update URL. +const LICENCE_KEY: &str = "LICENCE-KEY-NOT-FOR-LOGS"; + +/// An engine whose data file polls `url` through the Distributor formatter, +/// so each request carries [`LICENCE_KEY`] in its query. +fn engine_polling(url: &str, dir: &std::path::Path) -> Arc { + let path = dir.join("device.dat"); + std::fs::write(&path, b"current data").unwrap(); + let config = DataFileConfiguration::builder(&path) + .data_update_url(url) + .data_update_license_keys([LICENCE_KEY]) + .url_formatter(Some(Arc::new(FiftyOneUrlFormatter))) + .automatic_updates_enabled(false) + .file_system_watcher_enabled(false) + .build(); + Arc::new(TestEngine::new(Arc::new(AspectEngineDataFile::new(config)))) +} + +/// A poll that cannot connect fails with a message that names the configured +/// URL and the cause, and not the licence key the request carried. +#[test] +fn a_failed_connection_keeps_the_licence_key_out_of_the_error() { + // Nothing listens on a port once its listener is dropped. + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let url = format!("http://{}/api/v2/download", listener.local_addr().unwrap()); + drop(listener); + let dir = std::env::temp_dir().join(format!("fiftyone-dut-key-{}", std::process::id())); + std::fs::create_dir_all(&dir).unwrap(); + let engine = engine_polling(&url, &dir); + + let error = DataUpdateService::new() + .check_for_update(&engine, None) + .expect_err("nothing is listening"); + let message = error.to_string(); + assert_eq!(error.status, AutoUpdateStatus::HttpsError, "{message}"); + assert!(!message.contains(LICENCE_KEY), "{message}"); + assert!(message.contains(&url), "{message}"); + assert!(message.contains("error sending request: "), "{message}"); + let _ = std::fs::remove_dir_all(&dir); +} + +/// A download whose body ends early fails with a message that names the +/// configured URL, and not the licence key the request carried. +#[test] +fn a_failed_download_keeps_the_licence_key_out_of_the_error() { + // A server that promises more bytes than it sends, then closes. + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let url = format!("http://{}/api/v2/download", listener.local_addr().unwrap()); + let server = std::thread::spawn(move || { + let (mut stream, _) = listener.accept().unwrap(); + let mut request = [0u8; 4096]; + let _ = stream.read(&mut request); + let _ = stream.write_all( + b"HTTP/1.1 200 OK\r\nContent-Length: 1000\r\nConnection: close\r\n\r\nshort", + ); + }); + let dir = std::env::temp_dir().join(format!("fiftyone-dut-body-{}", std::process::id())); + std::fs::create_dir_all(&dir).unwrap(); + let engine = engine_polling(&url, &dir); + + let error = DataUpdateService::new() + .check_for_update(&engine, None) + .expect_err("the body ends early"); + server.join().unwrap(); + let message = error.to_string(); + assert_eq!(error.status, AutoUpdateStatus::StreamError, "{message}"); + assert!(!message.contains(LICENCE_KEY), "{message}"); + assert!(message.contains(&url), "{message}"); + let _ = std::fs::remove_dir_all(&dir); +}