From c26ca03233809835c9fbb61f15d646bf378f9eb2 Mon Sep 17 00:00:00 2001 From: James Rosewell Date: Mon, 26 Oct 2026 23:28:47 +0000 Subject: [PATCH 1/3] Test that a failed poll keeps the licence key out of its error --- pipeline-engines/tests/remote_refresh.rs | 76 +++++++++++++++++++++++- 1 file changed, 74 insertions(+), 2 deletions(-) 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); +} From d1d94495c826163e2375c9a2d81c32372d0e470d Mon Sep 17 00:00:00 2001 From: James Rosewell Date: Mon, 26 Oct 2026 23:28:48 +0000 Subject: [PATCH 2/3] Keep the licence keys out of the data update errors reqwest names the URL it requested in its errors, and that URL is the formatted one, whose query carries the licence keys. The service put the error into its own message, so each key reached the error a caller gets and the warning the scheduler logs. The message now takes the error without its URL, with its causes, and names the configured URL. --- pipeline-engines/src/data_update_service.rs | 25 +++++++++++++++++++-- 1 file changed, 23 insertions(+), 2 deletions(-) diff --git a/pipeline-engines/src/data_update_service.rs b/pipeline-engines/src/data_update_service.rs index 6c6e83d..9fd8d4a 100644 --- a/pipeline-engines/src/data_update_service.rs +++ b/pipeline-engines/src/data_update_service.rs @@ -597,7 +597,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 +648,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 +745,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". From 8716d71e9a3ddaf424c481831a1f777616dbb965 Mon Sep 17 00:00:00 2001 From: James Rosewell Date: Tue, 27 Oct 2026 00:02:41 +0000 Subject: [PATCH 3/3] Take the licence keys out of every data update error by value, as the cloud request engine does --- pipeline-engines/src/data_update_service.rs | 16 ++++++++++++++++ 1 file changed, 16 insertions(+) diff --git a/pipeline-engines/src/data_update_service.rs b/pipeline-engines/src/data_update_service.rs index 9fd8d4a..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() {