diff --git a/SECURITY.md b/SECURITY.md index 9bcc217..01e9a4d 100644 --- a/SECURITY.md +++ b/SECURITY.md @@ -16,9 +16,13 @@ how much a report matters: controlled. `..`, absolute paths, or a symlink that redirects a later write should all fail the install, not be skipped. - **Content accepted without verification.** Every chunk's SHA-1 is checked - against its content-addressed id before it reaches the disk. Anything that - writes bytes that were not checked is a hole, and it is what makes fetching - over a plain-HTTP lancache safe. + against its content-addressed id before it reaches the disk, and a manifest + must name the depot and manifest id that were asked for. Anything that writes + bytes that were not checked is a hole. Manifest signatures are not verified: + manifests, like everything else, are only ever fetched over TLS, and tapline + does not speak plain HTTP, so it cannot use a plain-HTTP lancache. Legacy + (pre-SteamPipe) Workshop files come with no hash; their declared size is + checked instead. - **A credential leaving where it should not.** Passwords are encrypted with Steam's RSA key and never written to disk; only the refresh token persists. A password reaching a log, an argument list, an error message or the token @@ -26,7 +30,8 @@ how much a report matters: - **Anything executed from downloaded content.** Depots ship `installscript.vdf`. tapline parses and reports it and must never run it — installing a game server is not a remote-code-execution primitive. -- **Decompression that is not bounded** by the size the manifest declares. +- **Decompression that is not bounded** by the size the manifest or archive + declares, or that holds a whole entry in memory to do it. ## What is not a vulnerability diff --git a/crates/tapline-auth/src/local.rs b/crates/tapline-auth/src/local.rs index d570f3e..209433b 100644 --- a/crates/tapline-auth/src/local.rs +++ b/crates/tapline-auth/src/local.rs @@ -33,9 +33,7 @@ pub fn home_relative_roots() -> &'static [&'static str] { /// Where the account's home directory is. /// /// Windows does not set `HOME`; it sets `USERPROFILE`. Anything that looks for -/// a Steam install has to ask for both, which is what this is for -- looking -/// for only one of them is how `libraries()` used to come back empty on every -/// Windows machine. +/// a Steam install has to ask for both, or it finds nothing on Windows. fn home() -> Option { std::env::var_os("HOME") .or_else(|| std::env::var_os("USERPROFILE")) @@ -206,8 +204,8 @@ mod tests { #[test] fn no_home_still_leaves_the_install_directories_to_look_in() { - // `libraries()` used to give up here, which on Windows was every time: - // it asked only for HOME, which Windows does not set. + // No home directory is the ordinary case on Windows when only HOME is + // consulted, so this must not give up. let roots = roots(None); assert_eq!( roots.len(), diff --git a/crates/tapline-auth/src/sddl.rs b/crates/tapline-auth/src/sddl.rs index 70a9301..045b289 100644 --- a/crates/tapline-auth/src/sddl.rs +++ b/crates/tapline-auth/src/sddl.rs @@ -119,9 +119,8 @@ mod tests { const OWNER: &str = "S-1-5-21-1111111111-2222222222-3333333333-1001"; - /// What the comparison used to be, before Windows' alias substitution made - /// a real SID comparison necessary. It is still what every case below but - /// the alias one needs. + /// A plain string comparison of SIDs, without the alias handling the real + /// one needs. Every case below but the alias one only needs this. fn owned_by(owner: &str) -> impl Fn(&str) -> bool + '_ { move |holder| holder.eq_ignore_ascii_case(owner) } diff --git a/crates/tapline-auth/src/store.rs b/crates/tapline-auth/src/store.rs index ceb74f1..f22dde5 100644 --- a/crates/tapline-auth/src/store.rs +++ b/crates/tapline-auth/src/store.rs @@ -193,23 +193,28 @@ impl TokenStore { } fn read_all(path: &Path) -> Result, TokenStoreError> { - let text = match std::fs::read_to_string(path) { + let mut text = match std::fs::read_to_string(path) { Ok(text) => text, Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(Vec::new()), Err(error) => return Err(TokenStoreError::Backend(error.to_string())), }; - check_permissions(path)?; - - let mut out = Vec::new(); - for line in text.lines() { - if line.trim().is_empty() { - continue; - } - let (account, token) = line.split_once('\t').ok_or(TokenStoreError::Malformed)?; - out.push((account.to_owned(), token.to_owned())); + if let Err(error) = check_permissions(path) { + text.zeroize(); + return Err(error); } - Ok(out) + + let entries = text + .lines() + .filter(|line| !line.trim().is_empty()) + .map(|line| { + line.split_once('\t') + .map(|(account, token)| (account.to_owned(), token.to_owned())) + .ok_or(TokenStoreError::Malformed) + }) + .collect(); + text.zeroize(); + entries } fn write_all(path: &Path, entries: &[(String, String)]) -> Result<(), TokenStoreError> { @@ -364,6 +369,14 @@ fn check_permissions(path: &Path) -> Result<(), TokenStoreError> { } fn write_file(path: &Path, token: &StoredToken) -> Result<(), TokenStoreError> { + // The file is one `accounttoken` line per entry; either separator in a + // field would corrupt every other entry on the next read. + if [&token.account, &token.refresh_token] + .iter() + .any(|field| field.contains(['\t', '\n', '\r'])) + { + return Err(TokenStoreError::Malformed); + } let mut entries = read_all(path)?; entries.retain(|(account, _)| account != &token.account); entries.push((token.account.clone(), token.refresh_token.clone())); diff --git a/crates/tapline-cdn/src/lib.rs b/crates/tapline-cdn/src/lib.rs index 4b271a0..d051248 100644 --- a/crates/tapline-cdn/src/lib.rs +++ b/crates/tapline-cdn/src/lib.rs @@ -70,6 +70,16 @@ pub async fn fetch_chunk_bytes( depot: DepotId, chunk: &Chunk, ) -> Result, CdnError> { + // The stored form is the compressed container plus AES padding and IV; a + // manifest claiming more than that of the largest chunk would otherwise + // set how much the HTTP client is willing to buffer. + if chunk.compressed_size as usize > tapline_chunk::MAX_CHUNK + 4096 { + return Err(CdnError::Container(format!( + "the manifest claims a {}-byte chunk", + chunk.compressed_size + ))); + } + let id = chunk.id_hex(); let url = format!("https://{host}/depot/{depot}/chunk/{id}"); @@ -125,10 +135,19 @@ pub fn decode_chunk_into( let container = tapline_crypto::decrypt_content_owned(depot_key, stored).map_err(|_| CdnError::Decrypt)?; - tapline_chunk::decode_into(&container, tapline_chunk::MAX_CHUNK, plaintext) + // Bounded by what the manifest declares, not only by the format's ceiling. + let limit = (chunk.uncompressed_size as usize).min(tapline_chunk::MAX_CHUNK); + tapline_chunk::decode_into(&container, limit, plaintext) .map_err(|e| CdnError::Container(e.to_string()))?; let plaintext = &*plaintext; + if plaintext.len() != chunk.uncompressed_size as usize { + return Err(CdnError::WrongLength { + expected: chunk.uncompressed_size, + actual: plaintext.len(), + }); + } + let digest = tapline_crypto::sha1(plaintext); if digest != chunk.id { return Err(CdnError::IntegrityFailure { @@ -138,13 +157,6 @@ pub fn decode_chunk_into( }); } - if plaintext.len() != chunk.uncompressed_size as usize { - return Err(CdnError::WrongLength { - expected: chunk.uncompressed_size, - actual: plaintext.len(), - }); - } - Ok(()) } @@ -173,6 +185,12 @@ pub async fn fetch_manifest( let manifest = Manifest::parse(&response.body, depot_key) .map_err(|e| CdnError::Container(e.to_string()))?; + if manifest.depot.get() != 0 && manifest.depot != depot { + return Err(CdnError::Container(format!( + "asked for a manifest of depot {depot}, received one of {}", + manifest.depot + ))); + } if manifest.id.get() != manifest_id { return Err(CdnError::Container(format!( "asked for manifest {manifest_id}, received {}", diff --git a/crates/tapline-chunk/src/lib.rs b/crates/tapline-chunk/src/lib.rs index 1db11c7..f30aa28 100644 --- a/crates/tapline-chunk/src/lib.rs +++ b/crates/tapline-chunk/src/lib.rs @@ -79,9 +79,7 @@ pub fn decode_into(input: &[u8], max_output: usize, out: &mut Vec) -> Result return vz::decode_into(input, max_output, out); } if zip::matches(input) { - let decoded = zip::decode(input, max_output)?; - out.clear(); - out.extend_from_slice(&decoded); + *out = zip::decode(input, max_output)?; return Ok(()); } diff --git a/crates/tapline-ext/src/lib.rs b/crates/tapline-ext/src/lib.rs index 20ae185..f31b032 100644 --- a/crates/tapline-ext/src/lib.rs +++ b/crates/tapline-ext/src/lib.rs @@ -126,6 +126,12 @@ pub trait EntrySink { fn end(&mut self) -> Result<(), ExtensionError>; + /// Called instead of `end` when an entry could not be decoded, so a sink + /// can throw away what it wrote of it. + fn abort(&mut self) -> Result<(), ExtensionError> { + Ok(()) + } + fn finish(&mut self) -> Result<(), ExtensionError> { Ok(()) } @@ -144,6 +150,9 @@ impl EntrySink for &mut S { fn end(&mut self) -> Result<(), ExtensionError> { (**self).end() } + fn abort(&mut self) -> Result<(), ExtensionError> { + (**self).abort() + } fn finish(&mut self) -> Result<(), ExtensionError> { (**self).finish() } @@ -162,6 +171,9 @@ impl EntrySink for Box { fn end(&mut self) -> Result<(), ExtensionError> { (**self).end() } + fn abort(&mut self) -> Result<(), ExtensionError> { + (**self).abort() + } fn finish(&mut self) -> Result<(), ExtensionError> { (**self).finish() } diff --git a/crates/tapline-ffi/src/lib.rs b/crates/tapline-ffi/src/lib.rs index 8209704..d1157df 100644 --- a/crates/tapline-ffi/src/lib.rs +++ b/crates/tapline-ffi/src/lib.rs @@ -62,10 +62,10 @@ fn pool() -> Option<&'static std::sync::Arc> { static POOL: OnceLock> = OnceLock::new(); Some(POOL.get_or_init(|| { // This is the moment the budget stops being changeable, so it is where - // the latch belongs. Setting it in `spawn_job` instead meant that - // asking for the current budget -- which builds the pool -- left the - // latch down, and the next `tapline_set_total_concurrency` answered - // TAPLINE_OK while the pool went on running at the old size. + // the latch belongs. Set any later, asking for the current budget -- + // which builds the pool -- would leave the latch down, and the next + // `tapline_set_total_concurrency` would answer TAPLINE_OK while the + // pool went on running at the old size. STARTED.store(true, std::sync::atomic::Ordering::Relaxed); let configured = TOTAL_CONCURRENCY.load(std::sync::atomic::Ordering::Relaxed) as usize; let budget = if configured == 0 { @@ -787,12 +787,12 @@ pub unsafe extern "C" fn tapline_job_next( return TAPLINE_BAD_ARGUMENT; }; - // One lock for the whole function. Taking the held slot, reading the queue - // and putting the message back used to be three separate critical - // sections, so two threads calling this on one job could both find the slot - // empty, both take a message, and both write the slot -- losing one of - // them. The header says every function here is safe to call from any - // thread, so this has to serialise rather than interleave. + // One lock for the whole function. With separate critical sections for + // taking the held slot, reading the queue and putting the message back, two + // threads calling this on one job could both find the slot empty, both take + // a message, and both write the slot -- losing one of them. The header says + // every function here is safe to call from any thread, so this has to + // serialise rather than interleave. let Ok(mut slot) = job.held.lock() else { set_error("the job's event queue was poisoned"); return TAPLINE_BAD_ARGUMENT; @@ -883,9 +883,8 @@ pub unsafe extern "C" fn tapline_last_error(buf: *mut u8, cap: usize, out_len: * return TAPLINE_BUFFER_TOO_SMALL; } if message.is_empty() { - // Nothing to copy, and no error set is the ordinary case -- asking - // for the length with a null buffer used to answer - // TAPLINE_BUFFER_TOO_SMALL for the zero bytes it needed. + // Nothing to copy. No error set is the ordinary case, and asking + // for its length with a null buffer is not a buffer too small. return TAPLINE_OK; } // SAFETY: `buf` is non-null and writable for `cap` bytes per the diff --git a/crates/tapline-fs/src/file.rs b/crates/tapline-fs/src/file.rs index 7705bc7..3c90346 100644 --- a/crates/tapline-fs/src/file.rs +++ b/crates/tapline-fs/src/file.rs @@ -128,6 +128,19 @@ pub fn remove_existing(path: &Path) -> io::Result<()> { } } +/// Removes `path` if it is a symlink, leaving what it points at alone. +/// +/// For a path the manifest now names as a file or directory where an earlier +/// install left a link: writing through the link would land at its target. +pub fn remove_symlink(path: &Path) -> io::Result<()> { + match std::fs::symlink_metadata(path) { + Ok(metadata) if metadata.file_type().is_symlink() => remove_existing(path), + Ok(_) => Ok(()), + Err(e) if e.kind() == io::ErrorKind::NotFound => Ok(()), + Err(e) => Err(e), + } +} + #[cfg(unix)] fn names_a_directory(file_type: &std::fs::FileType) -> bool { file_type.is_dir() @@ -281,6 +294,25 @@ mod tests { ); } + #[test] + fn remove_symlink_takes_only_links() { + let scratch = Scratch::new("file-remove-symlink"); + let target = scratch.join("real.txt"); + std::fs::write(&target, b"kept").expect("seed"); + let link = scratch.join("link.txt"); + symlink(Path::new("real.txt"), &link).expect("symlink"); + + remove_symlink(&target).expect("a regular file is left alone"); + remove_symlink(&link).expect("remove"); + remove_symlink(&scratch.join("never-existed")).expect("nothing to remove"); + + assert!( + std::fs::symlink_metadata(&link).is_err(), + "the link is still there" + ); + assert_eq!(std::fs::read(&target).expect("read"), b"kept"); + } + #[test] fn removing_a_path_that_is_not_there_is_not_an_error() { let scratch = Scratch::new("file-remove-missing"); diff --git a/crates/tapline-fs/src/lib.rs b/crates/tapline-fs/src/lib.rs index 6386d57..eca5746 100644 --- a/crates/tapline-fs/src/lib.rs +++ b/crates/tapline-fs/src/lib.rs @@ -1,5 +1,5 @@ mod file; mod path; -pub use file::{read_exact_at, remove_existing, set_mode, symlink, write_all_at}; +pub use file::{read_exact_at, remove_existing, remove_symlink, set_mode, symlink, write_all_at}; pub use path::{PathError, SafePath, validate_path, validate_symlink}; diff --git a/crates/tapline-fs/src/path.rs b/crates/tapline-fs/src/path.rs index ea39595..4f61906 100644 --- a/crates/tapline-fs/src/path.rs +++ b/crates/tapline-fs/src/path.rs @@ -1,4 +1,5 @@ use std::fmt; +use std::io; use std::path::{Component, Path, PathBuf}; #[derive(Debug, Clone, PartialEq, Eq)] @@ -9,6 +10,7 @@ pub enum PathError { Prefix, InteriorNul, SymlinkEscapes { link: String, target: String }, + ThroughSymlink { at: String }, } impl fmt::Display for PathError { @@ -25,6 +27,9 @@ impl fmt::Display for PathError { "the symlink {link} points outside the install root, at {target}" ) } + Self::ThroughSymlink { at } => { + write!(f, "the path passes through the symlink {at}") + } } } } @@ -45,6 +50,37 @@ impl SafePath { root.join(&self.0) } + /// Resolves under `root` like [`Self::resolve`], but refuses a path whose + /// directories below `root` include a symlink. + /// + /// Symlinks come from the manifest too, so a write that follows one lands + /// wherever an earlier install pointed it. A directory that does not exist + /// yet ends the check: nothing below it can be a link. + pub fn resolve_without_links(&self, root: &Path) -> io::Result { + let mut at = root.to_path_buf(); + let mut components = self.0.components().peekable(); + while let Some(component) = components.next() { + if components.peek().is_none() { + break; + } + at.push(component); + match std::fs::symlink_metadata(&at) { + Ok(metadata) if metadata.file_type().is_symlink() => { + return Err(io::Error::new( + io::ErrorKind::InvalidInput, + PathError::ThroughSymlink { + at: at.to_string_lossy().into_owned(), + }, + )); + } + Ok(_) => {} + Err(e) if e.kind() == io::ErrorKind::NotFound => break, + Err(e) => return Err(e), + } + } + Ok(root.join(&self.0)) + } + #[must_use] pub fn as_str(&self) -> String { self.0.to_string_lossy().replace('\\', "/") @@ -109,9 +145,18 @@ pub fn validate_symlink(link: &SafePath, target: &str) -> Result = Vec::new(); for component in target_path.components() { match component { - Component::Normal(_) => depth += 1, + Component::Normal(part) => { + depth += 1; + downs.push(part); + } Component::CurDir => {} Component::ParentDir => { depth -= 1; @@ -121,6 +166,9 @@ pub fn validate_symlink(link: &SafePath, target: &str) -> Result { return Err(PathError::SymlinkEscapes { @@ -131,7 +179,15 @@ pub fn validate_symlink(link: &SafePath, target: &str) -> Result) -> bool { @@ -254,6 +310,56 @@ mod tests { )); } + #[test] + fn a_symlink_target_is_collapsed_so_no_later_link_can_redirect_its_climb() { + let link = validate_path("l2").expect("valid link path"); + assert_eq!( + validate_symlink(&link, "a/b/l1/..").expect("stays inside"), + Path::new("a/b") + ); + let nested = validate_path("bin/linux64/lib.so").expect("valid link path"); + assert_eq!( + validate_symlink(&nested, "../x/../../lib/./real.so").expect("stays inside"), + Path::new("../../lib/real.so") + ); + assert_eq!( + validate_symlink(&nested, "a/..").expect("stays inside"), + Path::new(".") + ); + } + + #[test] + fn a_path_under_a_symlinked_directory_is_refused() { + let root = std::env::temp_dir().join(format!("tapline-fs-links-{}", std::process::id())); + let _ = std::fs::remove_dir_all(&root); + std::fs::create_dir_all(root.join("real/dir")).expect("scratch"); + crate::symlink(Path::new("real"), &root.join("link")).expect("symlink"); + + let through = validate_path("link/dir/file").expect("lexically fine"); + let error = through + .resolve_without_links(&root) + .expect_err("the write would follow a link"); + assert_eq!(error.kind(), io::ErrorKind::InvalidInput); + + let direct = validate_path("real/dir/file").expect("lexically fine"); + assert_eq!( + direct + .resolve_without_links(&root) + .expect("no link on the way"), + root.join("real/dir/file") + ); + let fresh = validate_path("new/deeper/file").expect("lexically fine"); + fresh + .resolve_without_links(&root) + .expect("nothing exists yet"); + let final_link = validate_path("link").expect("lexically fine"); + final_link + .resolve_without_links(&root) + .expect("only directories on the way are checked"); + + let _ = std::fs::remove_dir_all(&root); + } + #[test] fn a_safe_path_resolves_under_the_root_it_is_given() { let path = validate_path("tf/cfg/server.cfg").expect("must accept"); diff --git a/crates/tapline-gmad/src/lib.rs b/crates/tapline-gmad/src/lib.rs index 5215bff..7a81ea8 100644 --- a/crates/tapline-gmad/src/lib.rs +++ b/crates/tapline-gmad/src/lib.rs @@ -36,8 +36,12 @@ pub fn read_index(path: &Path) -> Result<(Addon, u64), ExtensionError> { filled += read; let slice = buffer.get(..filled).unwrap_or_default(); - match parse(slice) { + // The index alone: the content it describes runs far past the probe, + // and checking entries against the probe instead of the file refused + // every addon whose content ends beyond INDEX_LIMIT. + match parse_index(slice) { Ok(addon) => { + check_fits(&addon, file.metadata()?.len())?; let offset = addon .entries .first() @@ -50,7 +54,6 @@ pub fn read_index(path: &Path) -> Result<(Addon, u64), ExtensionError> { ExtensionError::Malformed { reason, .. } if reason.contains("ends in the middle") || reason.contains("not terminated") - || reason.contains("past the archive") ); if !truncated || read == 0 && filled < buffer.len() { return Err(error); @@ -74,6 +77,22 @@ pub fn read_index(path: &Path) -> Result<(Addon, u64), ExtensionError> { } } +fn check_fits(addon: &Addon, total: u64) -> Result<(), ExtensionError> { + for entry in &addon.entries { + let end = (entry.offset as u64).saturating_add(entry.size); + if end > total { + return Err(ExtensionError::Malformed { + extension: "gmad", + reason: format!( + "{:?} claims {} bytes at offset {}, past the archive's {total}", + entry.path, entry.size, entry.offset + ), + }); + } + } + Ok(()) +} + fn read_upto(file: &mut std::fs::File, buf: &mut [u8]) -> Result { let mut total = 0; while total < buf.len() { @@ -441,6 +460,39 @@ mod tests { assert!(ToZip::new().claims(&file("addons/x.gma"))); } + #[test] + fn the_index_of_an_addon_larger_than_the_probe_limit_is_read_from_its_head() { + let dir = std::env::temp_dir().join(format!("tapline-gmad-big-{}", std::process::id())); + std::fs::create_dir_all(&dir).expect("scratch"); + let path = dir.join("big.gma"); + + let content = (INDEX_LIMIT as u64) * 2; + let mut head = Vec::new(); + head.extend_from_slice(MAGIC); + head.push(3); + head.extend_from_slice(&0_u64.to_le_bytes()); + head.extend_from_slice(&1_u64.to_le_bytes()); + head.push(0); + head.extend_from_slice(b"Big\0desc\0author\0"); + head.extend_from_slice(&1_i32.to_le_bytes()); + head.extend_from_slice(&1_u32.to_le_bytes()); + head.extend_from_slice(b"models/huge.mdl\0"); + head.extend_from_slice(&(content as i64).to_le_bytes()); + head.extend_from_slice(&0_u32.to_le_bytes()); + head.extend_from_slice(&0_u32.to_le_bytes()); + let file = std::fs::File::create(&path).expect("create"); + std::io::Write::write_all(&mut &file, &head).expect("head"); + // Sparse: the content is never read, only its length matters. + file.set_len(head.len() as u64 + content).expect("extend"); + drop(file); + + let result = read_index(&path); + let _ = std::fs::remove_dir_all(&dir); + let (addon, offset) = result.expect("the index is complete in the first few bytes"); + assert_eq!(addon.entries.len(), 1); + assert_eq!(offset, head.len() as u64); + } + #[test] fn the_names_are_stable() { assert_eq!(Extract::new().name(), "gmad"); diff --git a/crates/tapline-gmad/src/sink.rs b/crates/tapline-gmad/src/sink.rs index 9b1491e..77b1967 100644 --- a/crates/tapline-gmad/src/sink.rs +++ b/crates/tapline-gmad/src/sink.rs @@ -86,6 +86,16 @@ impl EntrySink for ToDirectory { } Ok(()) } + + fn abort(&mut self) -> Result<(), ExtensionError> { + if self.current.take().is_none() { + return Ok(()); + } + match self.targets.get(self.at) { + Some(target) => Ok(std::fs::remove_file(target)?), + None => Ok(()), + } + } } const BATCH_BYTES: usize = 8 << 20; @@ -169,7 +179,8 @@ impl EntrySink for ToZip { fn begin(&mut self, entry: &ArchiveEntry, index: usize) -> Result<(), ExtensionError> { self.at = index; self.buffer.clear(); - self.buffer.reserve(entry.size as usize); + self.buffer + .reserve(entry.size.min(BATCH_BYTES as u64) as usize); Ok(()) } @@ -245,6 +256,26 @@ mod tests { Scratch(path) } + #[test] + fn an_aborted_entry_leaves_no_partial_file() { + let dir = scratch("abort"); + let entries = [ArchiveEntry { + path: "lua/a.lua".into(), + size: 8, + offset: 0, + stored_size: 8, + compression: tapline_ext::Compression::Stored, + }]; + let mut sink = ToDirectory::new(&dir.0); + sink.index(&entries).expect("index"); + sink.begin(&entries[0], 0).expect("begin"); + sink.data(b"prin").expect("data"); + sink.abort().expect("abort"); + + assert!(!dir.0.join("lua/a.lua").exists()); + assert!(sink.produced().is_empty()); + } + #[test] fn a_directory_sink_writes_the_files() { let raw = build(&[("lua/a.lua", b"print(1)"), ("b.txt", b"two")]); @@ -402,6 +433,13 @@ impl EntrySink for Filtered { Ok(()) } + fn abort(&mut self) -> Result<(), ExtensionError> { + if std::mem::take(&mut self.passing) { + self.inner.abort()?; + } + Ok(()) + } + fn finish(&mut self) -> Result<(), ExtensionError> { self.inner.finish() } @@ -464,6 +502,17 @@ impl EntrySink for Fanout { Ok(()) } + fn abort(&mut self) -> Result<(), ExtensionError> { + let mut first = Ok(()); + for sink in &mut self.sinks { + let result = sink.abort(); + if first.is_ok() { + first = result; + } + } + first + } + fn finish(&mut self) -> Result<(), ExtensionError> { let mut first = Ok(()); for sink in &mut self.sinks { diff --git a/crates/tapline-manifest/src/container.rs b/crates/tapline-manifest/src/container.rs index 599871c..6310e2a 100644 --- a/crates/tapline-manifest/src/container.rs +++ b/crates/tapline-manifest/src/container.rs @@ -143,7 +143,7 @@ fn unzip(bytes: &[u8]) -> Result, ManifestError> { let out = match method { 0 => data.to_vec(), - 8 => miniz_oxide::inflate::decompress_to_vec_with_limit(data, MAX_MANIFEST) + 8 => miniz_oxide::inflate::decompress_to_vec_with_limit(data, uncompressed_size) .map_err(|e| ManifestError::Decompress(format!("{:?}", e.status)))?, other => return Err(ManifestError::UnsupportedCompression(other)), }; diff --git a/crates/tapline-net/src/session.rs b/crates/tapline-net/src/session.rs index 9629e92..1a5dc41 100644 --- a/crates/tapline-net/src/session.rs +++ b/crates/tapline-net/src/session.rs @@ -80,9 +80,6 @@ impl Session { pub async fn recv(&mut self) -> Result { loop { if !self.pending.is_empty() { - if self.pending.is_empty() { - continue; - } return Ok(self.pending.remove(0)); } let bytes = self.transport.recv().await?; @@ -105,14 +102,25 @@ impl Session { let frame = self.recv().await?; if frame.reply_to() == Some(job_id) { - self.pending.splice(0..0, unrelated); + self.hold_back(unrelated); return Ok(frame); } if frame.emsg == EMsg::CLIENT_LOGGED_OFF { let eresult = frame.header.eresult.unwrap_or(0); return Err(NetError::Steam { eresult }); } - unrelated.push(frame); + push_bounded(&mut unrelated, frame); + } + } + + /// Puts frames that arrived while waiting for something else back in + /// front of the queue, keeping only the newest [`MAX_HELD`]: nothing in + /// the library drains unsolicited frames, so over a long session they + /// would otherwise pile up and be rescanned by every later wait. + fn hold_back(&mut self, unrelated: Vec) { + self.pending.splice(0..0, unrelated); + if let Some(excess) = self.pending.len().checked_sub(MAX_HELD) { + self.pending.drain(..excess); } } @@ -215,7 +223,7 @@ impl Session { let frame = self.recv().await?; match frame.emsg { EMsg::CLIENT_LOGON_RESPONSE => { - self.pending.splice(0..0, unrelated); + self.hold_back(unrelated); return Ok(frame); } EMsg::CLIENT_LOGGED_OFF => { @@ -223,7 +231,7 @@ impl Session { eresult: frame.header.eresult.unwrap_or(0), }); } - _ => unrelated.push(frame), + _ => push_bounded(&mut unrelated, frame), } } } @@ -272,6 +280,16 @@ impl Session { const OS_LINUX: u32 = 16; +/// How many unsolicited frames a session keeps for [`Session::take_unsolicited`]. +const MAX_HELD: usize = 256; + +fn push_bounded(frames: &mut Vec, frame: Frame) { + if frames.len() >= MAX_HELD { + frames.remove(0); + } + frames.push(frame); +} + #[cfg(test)] mod tests { use super::*; diff --git a/crates/tapline-pics/src/app.rs b/crates/tapline-pics/src/app.rs index 40f30c8..57d35ed 100644 --- a/crates/tapline-pics/src/app.rs +++ b/crates/tapline-pics/src/app.rs @@ -1,5 +1,5 @@ use tapline_ids::{AppId, DepotId, ManifestId}; -use tapline_vdf::{Object, Value}; +use tapline_vdf::Object; #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] pub enum Os { @@ -225,16 +225,6 @@ impl AppInfo { } } -#[allow(dead_code, reason = "kept next to the accessors it mirrors")] -fn nested<'a>(object: &'a Object, path: &[&str]) -> Option<&'a Value> { - let mut current = object; - let (last, parents) = path.split_last()?; - for step in parents { - current = current.get_object(step)?; - } - current.get(last) -} - #[cfg(test)] mod tests { use super::*; diff --git a/crates/tapline-pipe/src/lib.rs b/crates/tapline-pipe/src/lib.rs index 3d9f2eb..12dfead 100644 --- a/crates/tapline-pipe/src/lib.rs +++ b/crates/tapline-pipe/src/lib.rs @@ -304,14 +304,21 @@ async fn run_selective( let mut streamed = 0_u64; for ((entry, stored), index) in selected.iter().zip(pieces.iter()).zip(chosen.iter()) { - let bytes = decode_entry(&pipeline.format, entry, stored)?; - let bytes = &bytes; sink.begin(entry, *index)?; - if !bytes.is_empty() { - sink.data(bytes)?; + let decoded = decode_entry(&pipeline.format, entry, stored, &mut |piece| { + if piece.is_empty() { + return Ok(()); + } + streamed += piece.len() as u64; + sink.data(piece) + }); + if let Err(error) = decoded { + // The decode error is the one worth reporting; a failed cleanup + // only leaves the partial file the error already explains. + let _ = sink.abort(); + return Err(error); } sink.end()?; - streamed += bytes.len() as u64; observe(tapline::Event::Progress { bytes_done: streamed, bytes_total: selected.iter().map(|entry| entry.size).sum(), @@ -380,13 +387,21 @@ fn decode_entry( format: &str, entry: &tapline_ext::ArchiveEntry, stored: &[u8], -) -> Result, PipeError> { - match entry.compression { - tapline_ext::Compression::Stored => Ok(stored.to_vec()), - tapline_ext::Compression::Deflate => match format { - "zip" => Ok(tapline_zip::decode(entry, stored)?), - other => Err(SpecError::UnknownFormat(other.to_owned()).into()), - }, + emit: &mut dyn FnMut(&[u8]) -> Result<(), tapline_ext::ExtensionError>, +) -> Result<(), PipeError> { + match (format, entry.compression) { + ("zip", _) => Ok(tapline_zip::decode_into(entry, stored, emit)?), + ("gma", tapline_ext::Compression::Stored) => { + if stored.len() as u64 > entry.size { + return Err(tapline_ext::ExtensionError::Malformed { + extension: "gma", + reason: format!("{:?} holds more than its {} bytes", entry.path, entry.size), + } + .into()); + } + Ok(emit(stored)?) + } + (other, _) => Err(SpecError::UnknownFormat(other.to_owned()).into()), } } diff --git a/crates/tapline-rt-tokio/src/directory.rs b/crates/tapline-rt-tokio/src/directory.rs index 1ad3a71..9201715 100644 --- a/crates/tapline-rt-tokio/src/directory.rs +++ b/crates/tapline-rt-tokio/src/directory.rs @@ -4,6 +4,9 @@ use tokio::io::{AsyncReadExt, AsyncWriteExt}; const DIRECTORY_HOST: &str = "api.steampowered.com"; +/// A list of 64 servers is around 10 KiB. +const MAX_RESPONSE: u64 = 1024 * 1024; + #[derive(Debug, Clone, PartialEq, Eq)] pub struct CmServer { pub endpoint: String, @@ -75,7 +78,16 @@ async fn get(host: &str, path: &str) -> io::Result { stream.flush().await?; let mut raw = Vec::new(); - stream.read_to_end(&mut raw).await?; + (&mut stream) + .take(MAX_RESPONSE + 1) + .read_to_end(&mut raw) + .await?; + if raw.len() as u64 > MAX_RESPONSE { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + "the CM directory response is implausibly large", + )); + } let text = String::from_utf8_lossy(&raw).into_owned(); let (head, body) = text diff --git a/crates/tapline-rt-tokio/src/http.rs b/crates/tapline-rt-tokio/src/http.rs index 0b659e3..3d28fb8 100644 --- a/crates/tapline-rt-tokio/src/http.rs +++ b/crates/tapline-rt-tokio/src/http.rs @@ -85,15 +85,24 @@ impl HttpClient { } fn split_url(url: &str) -> Result<(String, u16, String), FetchError> { + // The URL is written into the request line verbatim, and some URLs come + // from Steam's replies: whitespace or a line break would split the request. + if url + .bytes() + .any(|byte| byte.is_ascii_whitespace() || byte.is_ascii_control()) + { + return Err(FetchError::InvalidUrl(url.to_owned())); + } let (scheme, rest) = url .split_once("://") .ok_or_else(|| FetchError::InvalidUrl(url.to_owned()))?; - let default_port = match scheme { - "https" => 443, - "http" => 80, - _ => return Err(FetchError::InvalidUrl(url.to_owned())), - }; + // Every connection is TLS. Accepting `http://` would only dial TLS on + // port 80, so it is refused rather than failing in a confusing way. + if scheme != "https" { + return Err(FetchError::InvalidUrl(url.to_owned())); + } + let default_port = 443; let (authority, path) = match rest.find('/') { Some(index) => ( @@ -323,6 +332,10 @@ async fn read_to_end( } } +/// A chunk-size line is a hex number and optional extensions; anything longer +/// than this is not one, and waiting for its end would buffer without bound. +const MAX_CHUNK_LINE: usize = 4096; + async fn read_chunked( connection: &mut Connection, start: Vec, @@ -330,17 +343,21 @@ async fn read_chunked( ) -> Result, FetchError> { let mut raw = start; let mut out = Vec::new(); - let mut cursor = 0_usize; loop { let line_end = loop { - if let Some(index) = find_crlf(&raw, cursor) { + if let Some(index) = find_crlf(&raw, 0) { break index; } + if raw.len() > MAX_CHUNK_LINE { + return Err(FetchError::MalformedResponse( + "chunk size line too long".into(), + )); + } fill(connection, &mut raw).await?; }; - let size_text = String::from_utf8_lossy(raw.get(cursor..line_end).unwrap_or_default()); + let size_text = String::from_utf8_lossy(raw.get(..line_end).unwrap_or_default()); let size_text = size_text.split(';').next().unwrap_or("").trim(); let size = usize::from_str_radix(size_text, 16) .map_err(|_| FetchError::MalformedResponse(format!("bad chunk size: {size_text}")))?; @@ -355,6 +372,9 @@ async fn read_chunked( return Ok(out); } + if (out.len() as u64).saturating_add(size as u64) > limit { + return Err(FetchError::BodyTooLarge { limit }); + } let data_end = data_start .checked_add(size) .ok_or_else(|| FetchError::MalformedResponse("chunk size overflow".into()))?; @@ -363,10 +383,7 @@ async fn read_chunked( } out.extend_from_slice(raw.get(data_start..data_end).unwrap_or_default()); - if out.len() as u64 > limit { - return Err(FetchError::BodyTooLarge { limit }); - } - cursor = data_end + 2; + raw.drain(..data_end + 2); } } @@ -410,12 +427,8 @@ mod tests { ) ); assert_eq!( - split_url("http://lancache.lan:8080/depot/1/chunk/a").expect("must split"), - ( - "lancache.lan".to_owned(), - 8080, - "/depot/1/chunk/a".to_owned() - ) + split_url("https://cache.lan:8443/depot/1/chunk/a").expect("must split"), + ("cache.lan".to_owned(), 8443, "/depot/1/chunk/a".to_owned()) ); assert_eq!( split_url("https://example.invalid").expect("must split"), @@ -426,9 +439,12 @@ mod tests { #[test] fn unusable_urls_are_refused() { assert!(split_url("ftp://example.invalid/x").is_err()); + assert!(split_url("http://example.invalid/x").is_err()); assert!(split_url("no-scheme.invalid/x").is_err()); assert!(split_url("https:///path").is_err()); assert!(split_url("https://host:notaport/x").is_err()); + assert!(split_url("https://host/x HTTP/1.1\r\nX-Injected: 1").is_err()); + assert!(split_url("https://host/a b").is_err()); } #[test] diff --git a/crates/tapline-rt-tokio/src/sink.rs b/crates/tapline-rt-tokio/src/sink.rs index ba225f8..65fc9b3 100644 --- a/crates/tapline-rt-tokio/src/sink.rs +++ b/crates/tapline-rt-tokio/src/sink.rs @@ -25,6 +25,10 @@ impl FileSink { self.file.sync_all() } + pub fn write_at_blocking(&self, offset: u64, data: &[u8]) -> io::Result<()> { + tapline_fs::write_all_at(&self.file, data, offset) + } + pub fn read_at(&self, offset: u64, len: usize) -> io::Result> { let mut buffer = vec![0_u8; len]; tapline_fs::read_exact_at(&self.file, &mut buffer, offset)?; @@ -34,7 +38,7 @@ impl FileSink { impl Sink for FileSink { async fn write_at(&self, offset: u64, data: &[u8]) -> io::Result<()> { - tapline_fs::write_all_at(&self.file, data, offset) + self.write_at_blocking(offset, data) } async fn allocate(&self, len: u64) -> io::Result<()> { diff --git a/crates/tapline-rt-tokio/src/tls.rs b/crates/tapline-rt-tokio/src/tls.rs index b2d84b5..7fdcf36 100644 --- a/crates/tapline-rt-tokio/src/tls.rs +++ b/crates/tapline-rt-tokio/src/tls.rs @@ -2,7 +2,7 @@ use crate::ws::WebSocket; use rustls::pki_types::ServerName; use std::io; use std::sync::Arc; -use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; +use tokio::io::{AsyncBufReadExt, AsyncReadExt, AsyncWriteExt, BufReader}; use tokio::net::TcpStream; use tokio_rustls::TlsConnector; @@ -12,17 +12,26 @@ const CM_PATH: &str = "/cmsocket/"; const WEBSOCKET_GUID: &str = "258EAFA5-E914-47DA-95CA-C5AB0DC85B11"; +/// Built once: the root store is ~150 certificates to copy, and the config +/// holds the session cache, so a fresh one per connection meant no CDN +/// reconnect could ever resume a TLS session. fn client_config() -> Arc { - let roots = rustls::RootCertStore { - roots: webpki_roots::TLS_SERVER_ROOTS.to_vec(), - }; - Arc::new( - rustls::ClientConfig::builder() - .with_root_certificates(roots) - .with_no_client_auth(), - ) + static CONFIG: std::sync::OnceLock> = std::sync::OnceLock::new(); + Arc::clone(CONFIG.get_or_init(|| { + let roots = rustls::RootCertStore { + roots: webpki_roots::TLS_SERVER_ROOTS.to_vec(), + }; + Arc::new( + rustls::ClientConfig::builder() + .with_root_certificates(roots) + .with_no_client_auth(), + ) + })) } +/// The upgrade response is a status line and a handful of headers. +const MAX_UPGRADE_RESPONSE: u64 = 16 * 1024; + pub async fn connect_tls(host: &str, port: u16) -> io::Result { let tcp = TcpStream::connect((host, port)).await?; tcp.set_nodelay(true)?; @@ -73,8 +82,9 @@ async fn upgrade(stream: TlsStream, host: &str, port: u16) -> io::Result io::Result io::Result<()> { - let mut header = Vec::with_capacity(14); + let mut header = Vec::with_capacity(14 + payload.len()); header.push(0x80 | opcode); let mask_bit = 0x80_u8; @@ -129,15 +129,14 @@ impl WebSocket { let mask = tapline_crypto::random_bytes::<4>(); header.extend_from_slice(&mask); - let mut masked = payload.to_vec(); - for (index, byte) in masked.iter_mut().enumerate() { - if let Some(key) = mask.get(index % 4) { - *byte ^= *key; - } - } + header.extend( + payload + .iter() + .zip(mask.iter().cycle()) + .map(|(byte, key)| byte ^ key), + ); self.stream.write_all(&header).await?; - self.stream.write_all(&masked).await?; self.stream.flush().await } diff --git a/crates/tapline-zip/src/lib.rs b/crates/tapline-zip/src/lib.rs index 3ddd2e7..75abb4e 100644 --- a/crates/tapline-zip/src/lib.rs +++ b/crates/tapline-zip/src/lib.rs @@ -196,22 +196,57 @@ pub fn finalize( } pub fn decode(entry: &ArchiveEntry, stored: &[u8]) -> Result, ExtensionError> { - match entry.compression { - Compression::Stored => Ok(stored.to_vec()), - Compression::Deflate => { - let limit = usize::try_from(entry.size).map_err(|_| ExtensionError::Malformed { - extension: "zip", - reason: format!( - "{:?} claims {} bytes, which will not fit", - entry.path, entry.size - ), - })?; - miniz_oxide::inflate::decompress_to_vec_with_limit(stored, limit).map_err(|error| { - ExtensionError::Malformed { - extension: "zip", - reason: format!("{:?} would not inflate: {:?}", entry.path, error.status), - } - }) + let mut out = Vec::new(); + decode_into(entry, stored, &mut |piece| { + out.extend_from_slice(piece); + Ok(()) + })?; + Ok(out) +} + +/// Inflates `stored` a piece at a time into `emit`, so an entry never has to +/// sit in memory whole: a few kilobytes of deflate can claim gigabytes. +/// Fails if the output runs past the entry's declared size. +pub fn decode_into( + entry: &ArchiveEntry, + stored: &[u8], + emit: &mut dyn FnMut(&[u8]) -> Result<(), ExtensionError>, +) -> Result<(), ExtensionError> { + use miniz_oxide::inflate::stream::{InflateState, inflate}; + use miniz_oxide::{DataFormat, MZError, MZFlush, MZStatus}; + + let malformed = |reason: String| ExtensionError::Malformed { + extension: "zip", + reason: format!("{:?} {reason}", entry.path), + }; + + if entry.compression == Compression::Stored { + if stored.len() as u64 > entry.size { + return Err(malformed(format!( + "holds more than its {} bytes", + entry.size + ))); + } + return emit(stored); + } + + let mut state = InflateState::new_boxed(DataFormat::Raw); + let mut buffer = vec![0_u8; 64 * 1024]; + let mut input = stored; + let mut written = 0_u64; + loop { + let result = inflate(&mut state, input, &mut buffer, MZFlush::None); + input = input.get(result.bytes_consumed..).unwrap_or_default(); + written = written.saturating_add(result.bytes_written as u64); + if written > entry.size { + return Err(malformed(format!("inflates past its {} bytes", entry.size))); + } + emit(buffer.get(..result.bytes_written).unwrap_or_default())?; + match result.status { + Ok(MZStatus::StreamEnd) => return Ok(()), + Ok(_) if result.bytes_consumed > 0 || result.bytes_written > 0 => {} + Ok(_) | Err(MZError::Buf) => return Err(malformed("ended mid-stream".to_owned())), + Err(error) => return Err(malformed(format!("would not inflate: {error:?}"))), } } } @@ -441,4 +476,49 @@ mod tests { let _ = plan(prefix, 0); } } + + fn deflated(body: &[u8], claimed: u64) -> (ArchiveEntry, Vec) { + let stored = miniz_oxide::deflate::compress_to_vec(body, 6); + let entry = ArchiveEntry { + path: "big.bin".to_owned(), + size: claimed, + offset: 0, + stored_size: stored.len() as u64, + compression: Compression::Deflate, + }; + (entry, stored) + } + + #[test] + fn an_entry_inflates_in_pieces_without_holding_it_whole() { + let body = vec![7_u8; 1024 * 1024]; + let (entry, stored) = deflated(&body, body.len() as u64); + let mut pieces = 0; + let mut total = 0; + decode_into(&entry, &stored, &mut |piece| { + assert!(piece.len() <= 64 * 1024); + pieces += 1; + total += piece.len(); + Ok(()) + }) + .expect("decode"); + assert_eq!(total, body.len()); + assert!(pieces > 1); + } + + #[test] + fn an_entry_inflating_past_its_size_is_refused() { + let body = vec![0_u8; 1024 * 1024]; + let (entry, stored) = deflated(&body, 1000); + let error = decode(&entry, &stored).expect_err("must refuse"); + assert!(error.to_string().contains("past its 1000 bytes"), "{error}"); + } + + #[test] + fn a_cut_short_deflate_stream_is_an_error() { + let body: Vec = (0..100_000_u32).map(|n| (n * 7 % 251) as u8).collect(); + let (entry, stored) = deflated(&body, body.len() as u64); + let cut = stored.get(..stored.len() / 2).expect("half"); + assert!(decode(&entry, cut).is_err()); + } } diff --git a/crates/tapline/src/session.rs b/crates/tapline/src/session.rs index 0958e65..a38f6d4 100644 --- a/crates/tapline/src/session.rs +++ b/crates/tapline/src/session.rs @@ -323,7 +323,7 @@ impl Session { )) .await?; - let reply = self.cm.wait_for_job(job).await?; + let reply = answered(self.cm.wait_for_job(job)).await?; let response: CMsgClientGetDepotDecryptionKeyResponse = reply.decode_body()?; let eresult = response.eresult.unwrap_or(0); @@ -346,7 +346,7 @@ impl Session { app: AppId, options: &InstallOptions, ) -> Result, InstallError> { - let info = tapline_pics::product_info(&mut self.cm, app).await?; + let info = answered(tapline_pics::product_info(&mut self.cm, app)).await?; self.app_name = info.name().map(str::to_owned); self.build_id = info.build_id(&options.branch); let depots = info.depots(&options.filter()); @@ -1021,7 +1021,7 @@ impl Session { } pub async fn app_info(&mut self, app: AppId) -> Result { - Ok(tapline_pics::product_info(&mut self.cm, app).await?) + Ok(answered(tapline_pics::product_info(&mut self.cm, app)).await?) } pub async fn workshop_details( @@ -1050,7 +1050,7 @@ impl Session { let resolved = if app.get() == 0 { None } else { - tapline_pics::product_info(&mut self.cm, app) + answered(tapline_pics::product_info(&mut self.cm, app)) .await .ok() .and_then(|info| info.workshop_depot()) @@ -1086,7 +1086,7 @@ impl Session { let workshop_depot = if query.app.get() == 0 { None } else { - tapline_pics::product_info(&mut self.cm, query.app) + answered(tapline_pics::product_info(&mut self.cm, query.app)) .await .ok() .and_then(|info| info.workshop_depot()) @@ -1259,8 +1259,19 @@ impl Session { response.status ))); } + // A legacy item comes with no hash to check it against, so its + // declared size is the one thing that can catch a cut-short or + // padded download before it is written. + if item.size > 0 && response.body.len() as u64 != item.size { + return Err(InstallError::Io(format!( + "the Workshop CDN sent {} bytes for {url}, the item is {}", + response.body.len(), + item.size + ))); + } - let path = safe.resolve(&target.install_dir); + let path = safe.resolve_without_links(&target.install_dir)?; + tapline_fs::remove_symlink(&path)?; let sink = FileSink::create(&path)?; sink.write_at(0, &response.body).await?; sink.sync().await?; @@ -1346,7 +1357,8 @@ impl Session { path: file.path.clone(), reason, })?; - let target = safe.resolve(&options.install_dir); + let target = safe.resolve_without_links(&options.install_dir)?; + tapline_fs::remove_symlink(&target)?; let existing = if options.resume { std::fs::metadata(&target) @@ -1493,14 +1505,27 @@ struct ChunkContext { async fn fetch_decode_write_chunk( ctx: &ChunkContext, chunk: &tapline_manifest::Chunk, - sink: &FileSink, + sink: &Arc, resuming: bool, ) -> Result { - if resuming - && let Ok(bytes) = sink.read_at(chunk.offset, chunk.uncompressed_size as usize) - && tapline_crypto::sha1(&bytes) == chunk.id - { - return Ok(ChunkOutcome::reused()); + // Reading back and hashing a chunk, decoding one, and writing one are all + // blocking work, so each runs on the blocking pool rather than stalling a + // runtime worker that other chunks' network I/O is waiting on. + // + // The size is the manifest's claim, and a claim past what any chunk can be + // would be allocated by the read-back before a byte is read. + if resuming && chunk.uncompressed_size as usize <= tapline_chunk::MAX_CHUNK { + let sink = Arc::clone(sink); + let chunk = chunk.clone(); + let intact = tokio::task::spawn_blocking(move || { + sink.read_at(chunk.offset, chunk.uncompressed_size as usize) + .is_ok_and(|bytes| tapline_crypto::sha1(&bytes) == chunk.id) + }) + .await + .unwrap_or(false); + if intact { + return Ok(ChunkOutcome::reused()); + } } let mut last_error = None; @@ -1526,19 +1551,31 @@ async fn fetch_decode_write_chunk( let for_decode = chunk.clone(); let host_for_decode = host.clone(); let key = ctx.key; - let decoded = tokio::task::spawn_blocking(move || { - tapline_cdn::decode_chunk_owned(stored, &for_decode, &key, &host_for_decode) + let sink = Arc::clone(sink); + // Outer error: the write failed, which another host cannot fix. + // Inner error: this host's bytes were bad, so the next one is tried. + let landed = tokio::task::spawn_blocking(move || { + let plaintext = match tapline_cdn::decode_chunk_owned( + stored, + &for_decode, + &key, + &host_for_decode, + ) { + Ok(plaintext) => plaintext, + Err(error) => return Ok(Err(error)), + }; + sink.write_at_blocking(for_decode.offset, &plaintext)?; + Ok::<_, std::io::Error>(Ok(plaintext.len())) }) .await - .map_err(|e| InstallError::Io(e.to_string()))?; + .map_err(|e| InstallError::Io(e.to_string()))??; - match decoded { - Ok(plaintext) => { - sink.write_at(chunk.offset, &plaintext).await?; + match landed { + Ok(written) => { return Ok(ChunkOutcome::fetched( &host, u64::from(chunk.compressed_size), - plaintext.len() as u64, + written as u64, )); } Err(error) => last_error = Some(error), @@ -1563,7 +1600,9 @@ fn create_directories( path: file.path.clone(), reason, })?; - std::fs::create_dir_all(safe.resolve(install_dir))?; + let path = safe.resolve_without_links(install_dir)?; + tapline_fs::remove_symlink(&path)?; + std::fs::create_dir_all(path)?; } Ok(()) } @@ -1596,7 +1635,7 @@ fn create_symlinks( } })?; - let link_path = safe.resolve(install_dir); + let link_path = safe.resolve_without_links(install_dir)?; if let Some(parent) = link_path.parent() { std::fs::create_dir_all(parent)?; } @@ -1853,6 +1892,25 @@ fn cdn_hosts(servers: &[CContentServerDirectory_ServerInfo]) -> Vec { .collect() } +/// How long a request to the CM server may go unanswered. `wait_for_job` +/// itself waits forever, and a server that drops one request while keeping +/// the connection open would otherwise hang the whole install. +const REPLY_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(60); + +async fn answered>( + request: impl std::future::Future>, +) -> Result { + tokio::time::timeout(REPLY_TIMEOUT, request) + .await + .unwrap_or_else(|_| { + Err(tapline_net::NetError::Io(format!( + "Steam did not answer within {} seconds", + REPLY_TIMEOUT.as_secs() + )) + .into()) + }) +} + #[cfg(test)] mod tests { use super::accounts_to_try; diff --git a/crates/tapline/src/validate.rs b/crates/tapline/src/validate.rs index cb7fc46..39f6a41 100644 --- a/crates/tapline/src/validate.rs +++ b/crates/tapline/src/validate.rs @@ -80,6 +80,16 @@ where let mut corrupt = Vec::new(); for chunk in &file.chunks { + // Checked before reading, because `read_chunk` allocates the + // manifest's claimed length up front. + let fits = chunk + .offset + .checked_add(u64::from(chunk.uncompressed_size)) + .is_some_and(|end| end <= file.size); + if !fits || chunk.uncompressed_size as usize > tapline_chunk::MAX_CHUNK { + corrupt.push(chunk.offset); + continue; + } let bytes = match read_chunk(&target, chunk.offset, chunk.uncompressed_size as usize) { Ok(bytes) => bytes, Err(error) => { diff --git a/crates/tapline/src/workshop.rs b/crates/tapline/src/workshop.rs index 4ec2266..4c1a09d 100644 --- a/crates/tapline/src/workshop.rs +++ b/crates/tapline/src/workshop.rs @@ -75,7 +75,7 @@ pub fn classify( .file_url .as_deref() .filter(|url| !url.is_empty()) - .map(str::to_owned); + .map(https); let content = match (manifest, url) { (Some(handle), _) => { @@ -137,6 +137,15 @@ impl From for InstallError { } } +/// Older items still list `http://` download links. Steam's UGC hosts serve +/// the same path over TLS, and tapline only speaks TLS. +fn https(url: &str) -> String { + match url.strip_prefix("http://") { + Some(rest) => format!("https://{rest}"), + None => url.to_owned(), + } +} + #[cfg(test)] mod tests { use super::*; @@ -318,4 +327,13 @@ mod tests { crate::WorkshopLayout::SteamCmd ); } + + #[test] + fn an_old_http_download_link_is_fetched_over_tls() { + assert_eq!( + super::https("http://cloud-3.steamusercontent.com/ugc/1/2/"), + "https://cloud-3.steamusercontent.com/ugc/1/2/" + ); + assert_eq!(super::https("https://x.invalid/a"), "https://x.invalid/a"); + } } diff --git a/crates/tapline/tests/streamed.rs b/crates/tapline/tests/streamed.rs index f427275..37ff9d6 100644 --- a/crates/tapline/tests/streamed.rs +++ b/crates/tapline/tests/streamed.rs @@ -2,11 +2,9 @@ use std::path::PathBuf; use tapline::{Session, Window}; -use tapline_ids::{AppId, PublishedFileId}; +use tapline_ids::PublishedFileId; const ITEM: PublishedFileId = PublishedFileId(104_691_717); -#[allow(dead_code)] -const APP: AppId = AppId(4000); fn scratch_root() -> PathBuf { let base = std::env::var("TAPLINE_TEST_DIR").unwrap_or_else(|_| {