perf(nas): process listings, artwork caches, and upload recovery natively

Move WebDAV/HTML parsing, content-addressed artwork caching, cover promotion, and streamed upload verification to native workers. Preserve proxy authentication, cancellation, cache limits, and deep XML fallback behavior.
This commit is contained in:
zarzet committed 2026-10-06 23:27:07 +07:00
1 parent 867371cb67
commit 927fd033cd
15 files changed
+1349 -98

No files matched your search

+21 -1
View File
@@ -53,13 +53,33 @@ fn run(
fn execute(
request: &Value,
_bytes: &[u8],
bytes: &[u8],
check: &(dyn Fn() -> Result<(), String> + Sync),
_manager: Option<&ExtensionManager>,
) -> Result<Value, String> {
let operation = string(request, "operation")?;
match operation {
"hash_file" => Ok(json!({"sha256": hash_file(string(request, "path")?, check)?})),
"matches_network_upload" => crate::network_tags::matches_upload(request, check),
"network_metadata" => {
let metadata = crate::network_tags::read_to_cache(
string(request, "url")?,
string(request, "display_name")?,
string(request, "cache_directory")?,
check,
)?;
let cacheable = serde_json::to_vec(&metadata)
.map_err(|error| error.to_string())?
.len()
<= 128 * 1024;
Ok(json!({"metadata": metadata, "cacheable": cacheable}))
}
"promote_network_cover" => crate::network_tags::promote_cover(
string(request, "path")?,
string(request, "directory")?,
check,
),
"parse_network_listing" => crate::native_listing::execute(request, Some(bytes)),
_ => Err(format!("Unknown native data operation: {operation}")),
}
}
+2
View File
@@ -11,6 +11,8 @@ mod logging;
mod lyrics;
mod manager;
mod metadata;
mod native_listing;
mod native_xml;
mod network_tags;
mod progress;
mod repository;
@@ -0,0 +1,235 @@
//! Pure directory-listing parser. Transport and credentials remain owned by
//! the platform; only a bounded response body enters this worker operation.
use scraper::{Html, Selector};
use serde_json::{Value, json};
use std::collections::HashMap;
use url::Url;
pub(crate) fn execute(request: &Value, bytes: Option<&[u8]>) -> Result<Value, String> {
if bytes.map_or_else(
|| request["body"].as_str().unwrap_or_default().len(),
<[u8]>::len,
) > 4 * 1024 * 1024
{
return Err("Directory listing is too large".into());
}
let body = match bytes {
Some(bytes) => String::from_utf8_lossy(bytes).into_owned(),
None => request["body"].as_str().unwrap_or_default().to_owned(),
};
let target = Url::parse(required(request, "base_url")?).map_err(|e| e.to_string())?;
let path = required(request, "path")?;
let mut base = target;
if !base.path().ends_with('/') {
base.set_path(&format!("{}/", base.path()));
}
let mut entries = Vec::<Value>::new();
let mut positions = HashMap::<String, usize>::new();
let mut add = |href: &str, directory: bool, size: Option<i64>| -> Result<(), String> {
let uri = base.join(&dart_href(href)).map_err(|e| e.to_string())?;
if uri.origin() != base.origin()
|| uri.query().is_some()
|| uri.fragment().is_some()
|| !uri.username().is_empty()
|| uri.password().is_some()
|| !uri.path().starts_with(base.path())
{
return Ok(());
}
let tail = decode_component(&uri.path()[base.path().len()..])?;
let name = tail.strip_suffix('/').unwrap_or(&tail);
if name.is_empty()
|| name.contains('/')
|| matches!(name, "." | "..")
|| name.contains('\\')
{
return Ok(());
}
let relative = format!("{path}{name}{}", if directory { "/" } else { "" });
let entry = json!({"path": relative, "name": name, "directory": directory, "size": size});
if let Some(index) = positions.get(&relative) {
entries[*index] = entry;
} else {
positions.insert(relative, entries.len());
entries.push(entry);
}
Ok(())
};
if request["protocol"] == "webdav" {
crate::native_xml::check_depth(&body)?;
let document = roxmltree::Document::parse(&body).map_err(|e| e.to_string())?;
for response in document
.descendants()
.filter(|node| named(*node, "response"))
{
let href = response.descendants().find(|node| named(*node, "href"));
let props: Vec<_> = response
.descendants()
.filter(|node| {
named(*node, "propstat")
&& node.children().any(|child| {
named(child, "status") && inner_text(child).contains(" 200 ")
})
})
.collect();
if let Some(href) = href.filter(|_| !props.is_empty()) {
let values: Vec<_> = props
.iter()
.flat_map(|node| node.descendants().skip(1))
.collect();
let directory = values.iter().any(|node| named(*node, "collection"));
let size = values
.iter()
.find(|node| named(**node, "getcontentlength"))
.and_then(|node| inner_text(*node).trim().parse::<i64>().ok());
add(&inner_text(href), directory, size)?;
}
}
} else {
let document = Html::parse_document(&body);
let selector = Selector::parse("a[href]").map_err(|e| e.to_string())?;
for anchor in document.select(&selector) {
let href = anchor.value().attr("href").unwrap_or_default();
let path = href.split(['?', '#']).next().unwrap_or_default();
let directory = path.ends_with('/') || path.ends_with('\\');
add(href, directory, None)?;
}
}
entries.sort_by(|a, b| {
b["directory"]
.as_bool()
.cmp(&a["directory"].as_bool())
.then_with(|| {
// Dart String.compareTo orders UTF-16 code units, rather than UTF-8.
dart_lower(a["name"].as_str().unwrap_or_default())
.encode_utf16()
.cmp(dart_lower(b["name"].as_str().unwrap_or_default()).encode_utf16())
})
});
Ok(json!({"entries": entries}))
}
fn dart_lower(value: &str) -> String {
// String.toLowerCase uses simple character mappings (İ -> i), without
// Rust's multi-character expansions or contextual final sigma.
value
.chars()
.map(|ch| ch.to_lowercase().next().unwrap_or(ch))
.collect()
}
fn required<'a>(request: &'a Value, key: &str) -> Result<&'a str, String> {
request[key]
.as_str()
.ok_or_else(|| format!("Missing listing {key}"))
}
fn dart_href(href: &str) -> String {
// Uri.resolve preserves edge spaces/controls and quotes invalid '%' as
// a literal. WHATWG URL otherwise trims controls and keeps broken escapes.
let mut result = String::with_capacity(href.len());
let bytes = href.as_bytes();
for (index, character) in href.char_indices() {
if character == '%'
&& !(bytes.get(index + 1).is_some_and(u8::is_ascii_hexdigit)
&& bytes.get(index + 2).is_some_and(u8::is_ascii_hexdigit))
{
result.push_str("%25");
} else if character <= ' ' || character == '\u{007f}' {
result.push_str(&format!("%{:02X}", character as u32));
} else {
result.push(character);
}
}
result
}
fn named(node: roxmltree::Node<'_, '_>, name: &str) -> bool {
node.is_element() && node.tag_name().name() == name
}
fn inner_text(node: roxmltree::Node<'_, '_>) -> String {
node.descendants()
.filter(|node| node.is_text())
.filter_map(|node| node.text())
.collect()
}
fn decode_component(value: &str) -> Result<String, String> {
let mut decoded = Vec::with_capacity(value.len());
let mut bytes = value.bytes();
while let Some(byte) = bytes.next() {
if byte == b'%' {
let first = bytes.next().and_then(hex).ok_or("Invalid percent escape")?;
let second = bytes.next().and_then(hex).ok_or("Invalid percent escape")?;
decoded.push(first * 16 + second);
} else {
decoded.push(byte);
}
}
String::from_utf8(decoded).map_err(|e| e.to_string())
}
fn hex(byte: u8) -> Option<u8> {
match byte {
b'0'..=b'9' => Some(byte - b'0'),
b'A'..=b'F' => Some(byte - b'A' + 10),
b'a'..=b'f' => Some(byte - b'a' + 10),
_ => None,
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn shared_dart_listing_fixtures_match_native_output() {
let cases: Value = serde_json::from_str(include_str!(
"../../../../test/fixtures/native_metadata_parsers.json"
))
.unwrap();
for case in cases["listings"].as_array().unwrap() {
let result = execute(case, None).unwrap();
assert_eq!(result["entries"], case["entries"]);
}
}
#[test]
fn html_children_entities_unicode_and_origin_are_preserved() {
let result = execute(&json!({"base_url":"https://nas.test/music/", "path":"", "protocol":"http", "body":
"<a href='../'>parent</a><a href='Folder/'>folder</a><a href='A%20%2B%23%25.flac'>audio</a><a href='音.flac'>music</a><a href='nested/song.flac'>nested</a><a href='https://other.test/music/b.flac'>other</a><a href='bad.flac?q=1'>query</a><a href='A%20%2B%23%25.flac'>duplicate</a>"}), None).unwrap();
assert_eq!(result["entries"].as_array().unwrap().len(), 3);
assert_eq!(result["entries"][0]["path"], "Folder/");
assert_eq!(result["entries"][1]["name"], "A +#%.flac");
assert_eq!(result["entries"][2]["name"], "音.flac");
}
#[test]
fn dav_only_accepts_successful_properties_and_immediate_children() {
let result = execute(&json!({"base_url":"https://nas.test/dav/Album/", "path":"Album/", "protocol":"webdav", "body":
"<d:multistatus xmlns:d='DAV:'><d:response><d:href>/dav/Album/A%20B.flac</d:href><d:propstat><d:prop><d:getcontentlength>123</d:getcontentlength></d:prop><d:status>HTTP/1.1 200 OK</d:status></d:propstat></d:response><d:response><d:href>/dav/Album/Missing.flac</d:href><d:propstat><d:prop/><d:status>HTTP/1.1 404 Not Found</d:status></d:propstat></d:response></d:multistatus>"}), None).unwrap();
assert_eq!(
result["entries"],
json!([{"path":"Album/A B.flac", "name":"A B.flac", "directory":false, "size":123}])
);
}
#[test]
fn encoded_slash_and_credentials_do_not_escape_the_root() {
let result = execute(&json!({"base_url":"https://nas.test/music/", "path":"", "protocol":"http", "body":
"<a href='a%2fb.flac'>nested</a><a href='https://user@nas.test/music/private.flac'>credentials</a><a href='ok.flac#part'>fragment</a>"}), None).unwrap();
assert_eq!(result["entries"], json!([]));
}
#[test]
fn deeply_nested_dav_response_is_rejected_before_xml_tokenization() {
let body = format!(
"<multistatus>{}ignored{}</multistatus>",
"<wrapper>".repeat(1_000),
"</wrapper>".repeat(1_000),
);
let error = execute(&json!({"base_url":"https://nas.test/dav/", "path":"", "protocol":"webdav", "body":body}), None).unwrap_err();
assert_eq!(error, "XML nesting exceeds 16 elements");
}
}
@@ -0,0 +1,109 @@
//! Bounds for the recursive tokenizer used by our native XML readers.
pub(crate) fn check_depth(input: &str) -> Result<(), String> {
// roxmltree's tokenizer recurses through nested elements. Guard its input
// before parsing so pathological metadata returns an error, not a native
// stack overflow. Ordinary TTML is only a handful of elements deep.
const MAX_DEPTH: usize = 16;
let bytes = input.as_bytes();
let (mut index, mut depth) = (0, 0_usize);
while index < bytes.len() {
if bytes[index] != b'<' {
index += 1;
continue;
}
let tail = &input[index..];
let ending = if tail.starts_with("<!--") {
Some("-->")
} else if tail.starts_with("<![CDATA[") {
Some("]]>")
} else if tail.starts_with("<?") {
Some("?>")
} else {
None
};
if let Some(ending) = ending {
let Some(end) = tail.find(ending) else {
return Ok(()); // The XML parser reports unterminated markup.
};
index += end + ending.len();
continue;
}
let Some(end) = markup_end(tail, tail.starts_with("<!DOCTYPE")) else {
return Ok(());
};
if tail.starts_with("</") {
depth = depth.saturating_sub(1);
} else if !tail.starts_with("<!")
&& !tail[..end - 1]
.trim_end_matches([' ', '\t', '\r', '\n'])
.ends_with('/')
{
depth += 1;
if depth > MAX_DEPTH {
return Err(format!("XML nesting exceeds {MAX_DEPTH} elements"));
}
}
index += end;
}
Ok(())
}
pub(crate) fn markup_end(input: &str, doctype: bool) -> Option<usize> {
let bytes = input.as_bytes();
let (mut quote, mut brackets, mut position) = (None, 0_usize, 0);
while position < bytes.len() {
let byte = bytes[position];
if let Some(current) = quote {
if byte == current {
quote = None;
}
} else {
// A DOCTYPE's internal subset may contain comments or processing
// instructions whose brackets and tag-like text have no meaning.
let ending = if doctype && bytes[position..].starts_with(b"<!--") {
Some(b"-->".as_slice())
} else if doctype && bytes[position..].starts_with(b"<?") {
Some(b"?>".as_slice())
} else {
None
};
if let Some(ending) = ending {
let end = bytes[position..]
.windows(ending.len())
.position(|window| window == ending)?;
position += end + ending.len();
continue;
}
match byte {
b'\'' | b'"' => quote = Some(byte),
b'[' if doctype => brackets += 1,
b']' if doctype => brackets = brackets.saturating_sub(1),
b'>' if brackets == 0 => return Some(position + 1),
_ => {}
}
}
position += 1;
}
None
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn bounds_actual_element_nesting_and_ignores_self_closing_nodes() {
let maximum = format!("{}{}", "<a>".repeat(16), "</a>".repeat(16));
check_depth(&maximum).unwrap();
assert!(check_depth(&format!("<a>{maximum}</a>")).is_err());
check_depth(&format!("<a>{}</a>", "<b/>".repeat(10_000))).unwrap();
}
#[test]
fn quoted_attributes_and_opaque_sections_are_not_element_depth() {
let markup = "<span>".repeat(1_000);
check_depth(&format!(
"<?xml version='1.0'?><!DOCTYPE a [<!-- ]> {markup} --> <?note ]>?> <!ENTITY b '{markup}'>]><a value='>'><!--{markup}--><![CDATA[{markup}]]><b value=\">\"/></a>"
)).unwrap();
}
}
+368 -37
View File
@@ -1,14 +1,17 @@
//! Bounded, seekable metadata reads through the app's authenticated loopback
//! proxy. The native reader never receives NAS credentials or external URLs.
use base64::Engine;
use serde_json::Value;
use serde_json::{Value, json};
use sha2::{Digest, Sha256};
use spotiflac_core::tags;
use spotiflac_network::{
HttpRequest, NetworkService, NetworkSession, policy::NetworkPermissions, url::UrlParts,
};
use std::{
collections::{BTreeMap, VecDeque},
io::{self, Read, Seek, SeekFrom},
fs::{self, File},
io::{self, Read, Seek, SeekFrom, Write},
path::Path,
sync::{Arc, OnceLock},
time::{Duration, Instant},
};
@@ -18,10 +21,142 @@ const MAX_BYTES: usize = 16 * 1024 * 1024;
const MAX_REQUESTS: usize = 80;
const DEADLINE: Duration = Duration::from_secs(20);
fn proxy_session(url: &str, timeout: Duration) -> Result<NetworkSession, String> {
let parsed = UrlParts::parse(url).ok_or("invalid network proxy URL")?;
let token = parsed.path.strip_prefix(b"/").unwrap_or_default();
if parsed.scheme != "http"
|| parsed.hostname != "127.0.0.1"
|| parsed.port.is_none()
|| parsed.has_credentials
|| !parsed.raw_query.is_empty()
|| !parsed.fragment.is_empty()
|| token.len() != 48
|| !token.iter().all(u8::is_ascii_hexdigit)
{
return Err("URL must reference the local network proxy".into());
}
static NETWORK: OnceLock<Result<Arc<NetworkService>, String>> = OnceLock::new();
let service = NETWORK
.get_or_init(|| {
let service = NetworkService::new().map_err(|e| e.to_string())?;
service.set_allow_private_network(true);
Ok(service)
})
.as_ref()
.map_err(Clone::clone)?;
Ok(service
.session(
NetworkPermissions {
domains: vec!["127.0.0.1".into()],
allow_http: true,
},
timeout,
)
.direct_media())
}
/// Reconcile a lost NAS rename response with streamed bytes. Credentials and
/// pins remain inside the app's existing proxy; neither file crosses Flutter.
pub(crate) fn matches_upload(
request: &Value,
check: &dyn Fn() -> Result<(), String>,
) -> Result<Value, String> {
check()?;
let url = crate::data_jobs::string(request, "url")?;
let path = crate::data_jobs::string(request, "path")?;
let length = request["length"].as_u64().ok_or("Missing upload length")?;
let idle = request["idle_timeout_ms"]
.as_u64()
.filter(|ms| *ms > 0 && *ms <= 300_000)
.ok_or("Invalid upload idle timeout")?;
let metadata = fs::metadata(path).map_err(|error| error.to_string())?;
if !metadata.is_file() || metadata.len() != length {
return Ok(json!({"matches": false}));
}
let idle = Duration::from_millis(idle);
let session = proxy_session(url, idle)?;
let mut stream = session.open_stream(
HttpRequest {
url: url.into(),
method: "GET".into(),
body: String::new(),
headers: BTreeMap::from([("Accept-Encoding".into(), "identity".into())]),
default_json: false,
user_agent: "SpotiFLAC/NetworkUploadRecovery".into(),
},
Duration::from_secs(u32::MAX.into()),
idle,
check,
)?;
let header = |name: &str| {
stream
.response
.headers
.iter()
.find(|(key, _)| key.eq_ignore_ascii_case(name))
.and_then(|(_, values)| values.first())
.map(String::as_str)
};
if stream.response.status != 200
|| header("content-length")
.and_then(|value| value.parse::<u64>().ok())
.is_some_and(|value| value != length)
|| header("content-encoding").is_some_and(|value| !value.eq_ignore_ascii_case("identity"))
{
return Ok(json!({"matches": false}));
}
let mut hash = Sha256::new();
let mut buffer = [0; 64 * 1024];
let mut received = 0_u64;
loop {
check()?;
let count = stream.read(&mut buffer, check)?;
if count == 0 {
break;
}
received = received
.checked_add(count as u64)
.ok_or("Upload length overflow")?;
if received > length {
return Err("Network destination changed".into());
}
hash.update(&buffer[..count]);
}
if received != length {
return Ok(json!({"matches": false}));
}
let remote: String = hash
.finalize()
.iter()
.map(|byte| format!("{byte:02x}"))
.collect();
let local = crate::data_jobs::hash_file(path, check)?;
check()?;
Ok(json!({"matches": remote == local}))
}
pub(crate) fn read(
url: &str,
hint: &str,
check: &dyn Fn() -> Result<(), String>,
) -> Result<Value, String> {
read_internal(url, hint, None, check)
}
pub(crate) fn read_to_cache(
url: &str,
hint: &str,
cache_directory: &str,
check: &dyn Fn() -> Result<(), String>,
) -> Result<Value, String> {
read_internal(url, hint, Some(Path::new(cache_directory)), check)
}
fn read_internal(
url: &str,
hint: &str,
cache_directory: Option<&Path>,
check: &dyn Fn() -> Result<(), String>,
) -> Result<Value, String> {
let mut reader = RangeReader::new(url, check)?;
let mut metadata = tags::read_file_metadata(&mut reader, hint, hint, check)?;
@@ -30,13 +165,150 @@ pub(crate) fn read(
&& !cover.data.is_empty()
&& cover.data.len() <= 4 * 1024 * 1024
{
metadata["cover_base64"] = base64::engine::general_purpose::STANDARD
.encode(cover.data)
.into();
if let Some(directory) = cache_directory {
// Artwork I/O failures preserve successfully parsed tags/lyrics.
if let Ok((path, digest)) = cache_cover(&cover.data, directory, check) {
metadata["cover_path"] = path.into();
metadata["cover_sha256"] = digest.into();
}
} else {
metadata["cover_base64"] = base64::engine::general_purpose::STANDARD
.encode(cover.data)
.into();
}
}
check()?;
Ok(metadata)
}
fn cache_cover(
bytes: &[u8],
directory: &Path,
check: &dyn Fn() -> Result<(), String>,
) -> Result<(String, String), String> {
check()?;
if !directory.is_absolute() {
return Err("Artwork cache must be absolute".into());
}
fs::create_dir_all(directory).map_err(|error| error.to_string())?;
let digest: String = Sha256::digest(bytes)
.iter()
.map(|byte| format!("{byte:02x}"))
.collect();
let target = directory.join(format!("network_cover_{digest}.image"));
let mut staged =
tempfile::NamedTempFile::new_in(directory).map_err(|error| error.to_string())?;
staged.write_all(bytes).map_err(|error| error.to_string())?;
staged
.as_file()
.sync_all()
.map_err(|error| error.to_string())?;
check()?;
staged.persist(&target).map_err(|error| error.to_string())?;
let mut files = fs::read_dir(directory)
.map_err(|error| error.to_string())?
.filter_map(Result::ok)
.filter_map(|entry| {
let name = entry.file_name();
let name = name.to_string_lossy();
let legacy = name.strip_suffix(".image").is_some_and(|digest| {
digest.len() == 64 && digest.bytes().all(|byte| byte.is_ascii_hexdigit())
});
if cover_digest(&name).is_none() && !legacy {
return None;
}
let metadata = entry.metadata().ok()?;
if !metadata.is_file() {
return None;
}
Some((entry.path(), metadata.modified().ok()?))
})
.collect::<Vec<_>>();
files.sort_by_key(|(_, time)| std::cmp::Reverse(*time));
for (old, _) in files.into_iter().skip(32) {
if old != target {
let _ = fs::remove_file(old);
}
}
Ok((target.to_string_lossy().into_owned(), digest))
}
fn cover_digest(name: &str) -> Option<&str> {
let digest = name
.strip_prefix("network_cover_")?
.strip_suffix(".image")?;
(digest.len() == 64 && digest.bytes().all(|byte| byte.is_ascii_hexdigit())).then_some(digest)
}
pub(crate) fn promote_cover(
path: &str,
directory: &str,
check: &dyn Fn() -> Result<(), String>,
) -> Result<Value, String> {
check()?;
let source = Path::new(path);
let destination = Path::new(directory);
if !source.is_absolute() || !destination.is_absolute() {
return Err("Artwork paths must be absolute".into());
}
if !source
.metadata()
.is_ok_and(|metadata| metadata.is_file() && metadata.len() > 0)
{
return Err("Artwork input must be a nonempty regular file".into());
}
// Only our immutable content-addressed cache skips a second hash. Older
// cache filenames use the same streamed hash as the previous Dart caller.
let digest = source
.parent()
.filter(|parent| {
parent
.file_name()
.is_some_and(|name| name == "network_metadata")
})
.and_then(|_| source.file_name())
.and_then(|name| name.to_str())
.and_then(cover_digest)
.map(str::to_owned)
.map(Ok)
.unwrap_or_else(|| crate::data_jobs::hash_file(path, check))?;
fs::create_dir_all(destination).map_err(|error| error.to_string())?;
let target = destination.join(format!("network_cover_{digest}.image"));
if !target
.metadata()
.is_ok_and(|metadata| metadata.is_file() && metadata.len() > 0)
{
let mut input = File::open(source).map_err(|error| error.to_string())?;
if !input
.metadata()
.map_err(|error| error.to_string())?
.is_file()
{
return Err("Artwork input must be a regular file".into());
}
let mut staged =
tempfile::NamedTempFile::new_in(destination).map_err(|error| error.to_string())?;
let mut buffer = [0; 64 * 1024];
loop {
check()?;
let count = input.read(&mut buffer).map_err(|error| error.to_string())?;
if count == 0 {
break;
}
staged
.write_all(&buffer[..count])
.map_err(|error| error.to_string())?;
}
staged
.as_file()
.sync_all()
.map_err(|error| error.to_string())?;
check()?;
staged.persist(&target).map_err(|error| error.to_string())?;
}
Ok(json!({"cover_path": target.to_string_lossy(), "sha256": digest}))
}
struct RangeReader<'a> {
url: String,
session: NetworkSession,
@@ -51,37 +323,7 @@ struct RangeReader<'a> {
impl<'a> RangeReader<'a> {
fn new(url: &str, check: &'a dyn Fn() -> Result<(), String>) -> Result<Self, String> {
let parsed = UrlParts::parse(url).ok_or("invalid metadata URL")?;
let token = parsed.path.strip_prefix(b"/").unwrap_or_default();
if parsed.scheme != "http"
|| parsed.hostname != "127.0.0.1"
|| parsed.port.is_none()
|| parsed.has_credentials
|| !parsed.raw_query.is_empty()
|| !parsed.fragment.is_empty()
|| token.len() != 48
|| !token.iter().all(u8::is_ascii_hexdigit)
{
return Err("metadata URL must reference the local network proxy".into());
}
static NETWORK: OnceLock<Result<Arc<NetworkService>, String>> = OnceLock::new();
let service = NETWORK
.get_or_init(|| {
let service = NetworkService::new().map_err(|e| e.to_string())?;
service.set_allow_private_network(true);
Ok(service)
})
.as_ref()
.map_err(Clone::clone)?;
let session = service
.session(
NetworkPermissions {
domains: vec!["127.0.0.1".into()],
allow_http: true,
},
DEADLINE,
)
.direct_media();
let session = proxy_session(url, DEADLINE)?;
let mut reader = Self {
url: url.into(),
session,
@@ -279,9 +521,14 @@ mod tests {
let mut reader = BufReader::new(socket.try_clone().unwrap());
let mut line = String::new();
let mut range = (0, data.len() - 1);
let mut headers_complete = false;
loop {
line.clear();
if reader.read_line(&mut line).unwrap_or(0) == 0 || line == "\r\n" {
if reader.read_line(&mut line).unwrap_or(0) == 0 {
break;
}
if line == "\r\n" {
headers_complete = true;
break;
}
if let Some(value) = line.to_lowercase().strip_prefix("range: bytes=") {
@@ -292,6 +539,12 @@ mod tests {
);
}
}
// A connector may abandon an idle socket before sending a
// request. Sending a response after EOF/timeout produces an
// unsolicited HTTP message and a spurious client failure.
if !headers_complete {
continue;
}
if !ranges {
range = (0, data.len() - 1);
}
@@ -339,6 +592,84 @@ mod tests {
.concat()
}
#[test]
fn fixture_does_not_respond_to_abandoned_connections() {
let fixture = Fixture::new(b"abc".to_vec(), false);
let address = fixture
.url
.strip_prefix("http://")
.unwrap()
.split('/')
.next()
.unwrap();
let mut socket = std::net::TcpStream::connect(address).unwrap();
socket
.set_read_timeout(Some(Duration::from_secs(3)))
.unwrap();
assert_eq!(socket.read(&mut [0; 64]).unwrap(), 0);
assert_eq!(fixture.bytes.load(Ordering::Relaxed), 0);
let mut local = tempfile::NamedTempFile::new().unwrap();
local.write_all(b"abc").unwrap();
assert_eq!(
matches_upload(
&json!({"url": fixture.url,
"path": local.path(), "length": 3, "idle_timeout_ms": 20000}),
&|| Ok(())
)
.unwrap()["matches"],
true
);
}
#[test]
fn cache_artwork_is_atomic_content_addressed_and_bounded() {
let root = tempfile::tempdir().unwrap();
let cache = root.path().join("network_metadata");
fs::create_dir(&cache).unwrap();
// Legacy source-addressed entries participate in the same 32-file cap.
for value in 0..35 {
fs::write(cache.join(format!("{value:064x}.image")), [value]).unwrap();
}
fs::write(cache.join("unrelated.txt"), b"keep").unwrap();
let (path, digest) = cache_cover(b"cover bytes", &cache, &|| Ok(())).unwrap();
assert_eq!(fs::read(&path).unwrap(), b"cover bytes");
assert_eq!(
digest,
crate::data_jobs::hash_file(&path, &|| Ok(())).unwrap()
);
let again = cache_cover(b"cover bytes", &cache, &|| Ok(())).unwrap();
assert_eq!(again.0, path);
assert_eq!(fs::read_dir(&cache).unwrap().count(), 33);
let covers = root.path().join("library_covers");
let promoted = promote_cover(&path, covers.to_str().unwrap(), &|| Ok(())).unwrap();
let target = promoted["cover_path"].as_str().unwrap();
assert_eq!(fs::read(target).unwrap(), b"cover bytes");
fs::remove_file(&path).unwrap();
assert_eq!(fs::read(target).unwrap(), b"cover bytes");
assert!(cache_cover(b"cancelled", &cache, &|| Err("cancelled".into())).is_err());
assert_eq!(fs::read_dir(&covers).unwrap().count(), 1);
}
#[test]
fn streamed_upload_recovery_compares_content_and_checks_cancellation() {
let fixture = Fixture::new(b"abc".to_vec(), false);
let mut local = tempfile::NamedTempFile::new().unwrap();
local.write_all(b"abc").unwrap();
let request =
json!({"url":fixture.url,"path":local.path(),"length":3,"idle_timeout_ms":20000});
assert_eq!(
matches_upload(&request, &|| Ok(())).unwrap()["matches"],
true
);
fs::write(local.path(), b"abd").unwrap();
assert_eq!(
matches_upload(&request, &|| Ok(())).unwrap()["matches"],
false
);
assert!(matches_upload(&request, &|| Err("cancelled".into())).is_err());
let mut invalid = request.clone();
invalid["url"] = "https://example.test/private".into();
assert!(matches_upload(&invalid, &|| Ok(())).is_err());
}
#[test]
fn reads_flac_tags_lyrics_cover_without_audio_download() {
let input = [
b"fLaC".as_slice(),