diff --git a/Cargo.lock b/Cargo.lock index e9e677d2..a2962b38 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3660,6 +3660,7 @@ dependencies = [ "sha1 0.11.0", "snafu", "sqlx", + "subtle", "themelion", "tokio", "tokio-util", @@ -3667,6 +3668,7 @@ dependencies = [ "tower-http", "tracing", "ulid", + "url", "uuid", ] diff --git a/crates/apotheke/src/repo/music.rs b/crates/apotheke/src/repo/music.rs index 60c7e325..5445665c 100644 --- a/crates/apotheke/src/repo/music.rs +++ b/crates/apotheke/src/repo/music.rs @@ -563,17 +563,21 @@ pub async fn search_tracks( pool: &SqlitePool, query: &str, limit: i64, + offset: i64, ) -> Result, DbError> { let pattern = format!("%{query}%"); + // WHY: deterministic ORDER BY — OFFSET pagination over an unordered + // result set yields overlapping/unstable pages. sqlx::query_as::<_, MusicTrack>( "SELECT id, medium_id, position, title, duration_ms, mb_recording_id, acoustid_fingerprint, acoustid_id, file_path, file_size_bytes, bit_depth, sample_rate, codec, quality_score, replay_gain_track_db, replay_gain_album_db, source_type, added_at - FROM music_tracks WHERE title LIKE ? LIMIT ?", + FROM music_tracks WHERE title LIKE ? ORDER BY title, id LIMIT ? OFFSET ?", ) .bind(&pattern) .bind(limit) + .bind(offset) .fetch_all(pool) .await .context(QuerySnafu { @@ -581,6 +585,17 @@ pub async fn search_tracks( }) } +pub async fn count_tracks(pool: &SqlitePool, query: &str) -> Result { + let pattern = format!("%{query}%"); + sqlx::query_scalar("SELECT COUNT(*) FROM music_tracks WHERE title LIKE ?") + .bind(&pattern) + .fetch_one(pool) + .await + .context(QuerySnafu { + table: "music_tracks", + }) +} + #[cfg(test)] mod tests { use super::*; @@ -882,4 +897,112 @@ mod tests { .unwrap(); assert!(meta.is_none()); } + + async fn seed_tracks(pool: &SqlitePool, count: usize) -> Vec> { + let group_id = make_id(); + let group = MusicReleaseGroup { + id: group_id.clone(), + registry_id: None, + title: "Pagination Album".to_string(), + rg_type: "album".to_string(), + mb_release_group_id: None, + year: Some(2024), + quality_profile_id: None, + added_at: now(), + }; + insert_release_group(pool, &group).await.unwrap(); + + let release_id = make_id(); + let release = MusicRelease { + id: release_id.clone(), + release_group_id: group_id, + title: "Pagination Album".to_string(), + release_date: None, + country: None, + label: None, + catalog_number: None, + mb_release_id: None, + added_at: now(), + }; + insert_release(pool, &release).await.unwrap(); + + let medium_id = make_id(); + let medium = MusicMedium { + id: medium_id.clone(), + release_id, + position: 1, + format: "Digital".to_string(), + title: None, + }; + insert_medium(pool, &medium).await.unwrap(); + + let mut ids = Vec::with_capacity(count); + for i in 0..count { + let track_id = make_id(); + let track = MusicTrack { + id: track_id.clone(), + medium_id: medium_id.clone(), + position: i as i64, + title: format!("Track {i:02}"), + duration_ms: Some(180000), + mb_recording_id: None, + acoustid_fingerprint: None, + acoustid_id: None, + file_path: None, + file_size_bytes: None, + bit_depth: None, + sample_rate: None, + codec: None, + quality_score: None, + replay_gain_track_db: None, + replay_gain_album_db: None, + source_type: "local".to_string(), + added_at: now(), + }; + insert_track(pool, &track).await.unwrap(); + ids.push(track_id); + } + ids + } + + #[tokio::test] + async fn search_tracks_paginates_with_offset() { + let pool = setup().await; + seed_tracks(&pool, 7).await; + + let page1 = search_tracks(&pool, "", 5, 0).await.unwrap(); + let page2 = search_tracks(&pool, "", 5, 5).await.unwrap(); + assert_eq!(page1.len(), 5); + assert_eq!(page2.len(), 2); + + let ids1: std::collections::HashSet> = page1.into_iter().map(|t| t.id).collect(); + let ids2: std::collections::HashSet> = page2.into_iter().map(|t| t.id).collect(); + assert!(ids1.is_disjoint(&ids2), "pages must not overlap"); + + let past_end = search_tracks(&pool, "", 5, 10).await.unwrap(); + assert!(past_end.is_empty()); + } + + #[tokio::test] + async fn search_tracks_filters_by_query_with_offset() { + let pool = setup().await; + seed_tracks(&pool, 3).await; + + let hits = search_tracks(&pool, "Track 01", 10, 0).await.unwrap(); + assert_eq!(hits.len(), 1); + assert_eq!(hits[0].title, "Track 01"); + + let beyond = search_tracks(&pool, "Track 01", 10, 1).await.unwrap(); + assert!(beyond.is_empty()); + } + + #[tokio::test] + async fn count_tracks_returns_full_match_count() { + let pool = setup().await; + seed_tracks(&pool, 7).await; + + assert_eq!(count_tracks(&pool, "").await.unwrap(), 7); + assert_eq!(count_tracks(&pool, "Track 03").await.unwrap(), 1); + assert_eq!(count_tracks(&pool, "No Such Title").await.unwrap(), 0); + } } diff --git a/crates/apotheke/src/repo/user.rs b/crates/apotheke/src/repo/user.rs index 795f0810..9313787b 100644 --- a/crates/apotheke/src/repo/user.rs +++ b/crates/apotheke/src/repo/user.rs @@ -326,6 +326,15 @@ pub async fn revoke_api_key(pool: &SqlitePool, id: &[u8]) -> Result<(), DbError> Ok(()) } +pub async fn revoke_api_keys_for_user(pool: &SqlitePool, user_id: &[u8]) -> Result<(), DbError> { + sqlx::query("UPDATE api_keys SET revoked = 1 WHERE user_id = ?") + .bind(user_id) + .execute(pool) + .await + .context(QuerySnafu { table: "api_keys" })?; + Ok(()) +} + pub async fn update_api_key_last_used( pool: &SqlitePool, id: &[u8], @@ -498,4 +507,32 @@ mod tests { let results = list_users(&pool, 10, 0).await.unwrap(); assert!(results.is_empty()); } + + #[tokio::test] + async fn revoke_api_keys_for_user_revokes_all() { + let pool = setup().await; + let user_id = make_id(); + let user = test_user(user_id.clone()); + insert_user(&pool, &user).await.unwrap(); + + for (short, long) in [("key00001", "hash1"), ("key00002", "hash2")] { + let key = ApiKey { + id: make_id(), + user_id: user_id.clone(), + short_token: short.to_string(), + long_token_hash: long.to_string(), + label: "test".to_string(), + created_at: now(), + last_used_at: None, + revoked: 0, + }; + insert_api_key(&pool, &key).await.unwrap(); + } + + revoke_api_keys_for_user(&pool, &user_id).await.unwrap(); + + let keys = list_api_keys_for_user(&pool, &user_id).await.unwrap(); + assert_eq!(keys.len(), 2); + assert!(keys.iter().all(|k| k.revoked == 1)); + } } diff --git a/crates/paroche/Cargo.toml b/crates/paroche/Cargo.toml index 5016b69a..647854dc 100644 --- a/crates/paroche/Cargo.toml +++ b/crates/paroche/Cargo.toml @@ -38,6 +38,8 @@ mdns-sd.workspace = true jiff.workspace = true ulid.workspace = true sha1.workspace = true +subtle.workspace = true +url.workspace = true [dev-dependencies] sqlx.workspace = true diff --git a/crates/paroche/src/lib.rs b/crates/paroche/src/lib.rs index 09b34d25..172c8810 100644 --- a/crates/paroche/src/lib.rs +++ b/crates/paroche/src/lib.rs @@ -1,6 +1,7 @@ pub mod discovery; pub mod error; pub mod middleware; +pub mod net_validate; pub mod opds; pub mod response; pub mod routes; diff --git a/crates/paroche/src/net_validate.rs b/crates/paroche/src/net_validate.rs new file mode 100644 index 00000000..79750c62 --- /dev/null +++ b/crates/paroche/src/net_validate.rs @@ -0,0 +1,253 @@ +// SSRF guard for user-supplied URLs handed to server-side fetchers. +use std::net::IpAddr; + +use url::{Host, Url}; + +use crate::error::ParocheError; + +fn validation(message: &str) -> ParocheError { + ParocheError::Validation { + message: message.to_string(), + } +} + +/// Validate a user-supplied download URL before it reaches any server-side +/// fetcher: http(s) or magnet scheme only, and no reachable host may point +/// at loopback, link-local, private, or otherwise non-public address space. +/// +/// WHY: http(s) hostnames are resolved and every resolved address is +/// checked — a public-looking name can point at internal infrastructure. +/// Resolution failure rejects (fail-closed): an unresolvable host cannot be +/// fetched anyway, and uncertainty must not admit a request. Magnet URIs +/// have no direct host; their `tr` (tracker) parameters are the dialable +/// surface and are each validated instead. +pub async fn validate_download_url(raw: &str) -> Result<(), ParocheError> { + let parsed = Url::parse(raw).map_err(|_| validation("download_url is not a valid URL"))?; + + match parsed.scheme() { + "http" | "https" => validate_fetch_host(&parsed).await, + "magnet" => validate_magnet_trackers(&parsed), + _ => Err(validation( + "download_url scheme must be http, https, or magnet", + )), + } +} + +async fn validate_fetch_host(parsed: &Url) -> Result<(), ParocheError> { + let host = parsed + .host() + .ok_or_else(|| validation("download_url must have a host"))?; + + match host { + Host::Ipv4(ip) => reject_disallowed_ip(IpAddr::V4(ip)), + Host::Ipv6(ip) => reject_disallowed_ip(IpAddr::V6(ip)), + Host::Domain(domain) => { + // WHY: port only satisfies lookup_host's addr format; http/https always + // have a known default so the fallback is unreachable. + let port = parsed.port_or_known_default().unwrap_or(443); + let addrs = tokio::net::lookup_host((domain, port)) + .await + .map_err(|_| validation("download_url host did not resolve"))?; + let mut resolved_any = false; + for addr in addrs { + resolved_any = true; + reject_disallowed_ip(addr.ip())?; + } + if !resolved_any { + return Err(validation("download_url host did not resolve")); + } + Ok(()) + } + } +} + +// WHY: tracker hostnames are NOT DNS-resolved here — a magnet can carry many +// trackers and any public tracker is inherently third-party-controlled; the +// enforced boundary is direct internal targets (IP literals in disallowed +// ranges, localhost names) and non-tracker schemes. +fn validate_magnet_trackers(parsed: &Url) -> Result<(), ParocheError> { + for (key, value) in parsed.query_pairs() { + if key != "tr" && !key.starts_with("tr.") { + continue; + } + let tracker = Url::parse(&value) + .map_err(|_| validation("magnet tracker parameter is not a valid URL"))?; + match tracker.scheme() { + "http" | "https" | "udp" | "ws" | "wss" => {} + _ => return Err(validation("magnet tracker scheme is not allowed")), + } + match tracker.host() { + Some(Host::Ipv4(ip)) => reject_disallowed_ip(IpAddr::V4(ip))?, + Some(Host::Ipv6(ip)) => reject_disallowed_ip(IpAddr::V6(ip))?, + Some(Host::Domain(domain)) => { + // WHY: non-special schemes (udp) get opaque host parsing — an + // IP literal arrives here as a Domain string, so parse it back. + if let Ok(ip) = domain.parse::() { + reject_disallowed_ip(ip)?; + } + let lower = domain.to_ascii_lowercase(); + if lower == "localhost" || lower.ends_with(".localhost") { + return Err(validation( + "magnet tracker host resolves to a private or local address", + )); + } + } + None => return Err(validation("magnet tracker must have a host")), + } + } + Ok(()) +} + +fn reject_disallowed_ip(ip: IpAddr) -> Result<(), ParocheError> { + if ip_is_disallowed(ip) { + return Err(validation( + "download_url host resolves to a private or local address", + )); + } + Ok(()) +} + +fn ip_is_disallowed(ip: IpAddr) -> bool { + match ip { + IpAddr::V4(v4) => { + v4.is_loopback() + || v4.is_private() + || v4.is_link_local() + || v4.is_unspecified() + || v4.is_broadcast() + // NOTE: shared address space (CGNAT), RFC 6598: 100.64.0.0/10 + || (u32::from(v4) & 0xFFC0_0000) == 0x6440_0000 + } + IpAddr::V6(v6) => { + // WHY: an IPv4-mapped IPv6 literal (::ffff:127.0.0.1) must be judged + // by its embedded IPv4 address or it bypasses every v4 range check. + if let Some(mapped) = v6.to_ipv4_mapped() { + return ip_is_disallowed(IpAddr::V4(mapped)); + } + v6.is_loopback() + || v6.is_unspecified() + || v6.is_unique_local() + || v6.is_unicast_link_local() + } + } +} + +#[cfg(test)] +mod tests { + use std::net::{Ipv4Addr, Ipv6Addr}; + + use super::*; + + #[tokio::test] + async fn rejects_unparseable_url() { + assert!(validate_download_url("not a url").await.is_err()); + assert!(validate_download_url("").await.is_err()); + } + + #[tokio::test] + async fn rejects_non_http_schemes() { + for url in [ + "ftp://example.com/file.torrent", + "file:///etc/passwd", + "gopher://example.com/", + "javascript:alert(1)", + ] { + assert!(validate_download_url(url).await.is_err(), "allowed: {url}"); + } + } + + #[tokio::test] + async fn rejects_loopback_and_private_ip_literals() { + for url in [ + "http://127.0.0.1/x", + "http://127.8.9.10:8080/x", + "https://10.0.0.1/x", + "http://172.16.5.5/x", + "http://192.168.1.10/x", + "http://169.254.169.254/latest/meta-data/", + "http://0.0.0.0/x", + "http://100.64.0.1/x", + "http://[::1]/x", + "http://[fc00::1]/x", + "http://[fe80::1]/x", + "http://[::ffff:127.0.0.1]/x", + "http://[::ffff:192.168.1.1]/x", + ] { + assert!(validate_download_url(url).await.is_err(), "allowed: {url}"); + } + } + + #[tokio::test] + async fn rejects_localhost_hostname() { + assert!(validate_download_url("http://localhost/x").await.is_err()); + assert!( + validate_download_url("http://localhost:8080/x") + .await + .is_err() + ); + } + + #[tokio::test] + async fn accepts_plain_magnet_uri() { + assert!( + validate_download_url("magnet:?xt=urn:btih:abc123def456") + .await + .is_ok() + ); + } + + #[tokio::test] + async fn accepts_magnet_with_public_trackers() { + assert!( + validate_download_url( + "magnet:?xt=urn:btih:abc123&tr=udp%3A%2F%2Ftracker.example.org%3A1337%2Fannounce&tr=https%3A%2F%2Ftracker.example.net%2Fannounce" + ) + .await + .is_ok() + ); + } + + #[tokio::test] + async fn rejects_magnet_with_private_or_local_trackers() { + for url in [ + "magnet:?xt=urn:btih:abc&tr=http%3A%2F%2F127.0.0.1%3A8080%2Fannounce", + "magnet:?xt=urn:btih:abc&tr=http%3A%2F%2F192.168.1.5%2Fannounce", + "magnet:?xt=urn:btih:abc&tr=udp%3A%2F%2F10.0.0.1%3A1337%2Fannounce", + "magnet:?xt=urn:btih:abc&tr=http%3A%2F%2Flocalhost%3A9000%2Fannounce", + "magnet:?xt=urn:btih:abc&tr=ftp%3A%2F%2Ftracker.example.org%2Fannounce", + ] { + assert!(validate_download_url(url).await.is_err(), "allowed: {url}"); + } + } + + #[tokio::test] + async fn accepts_public_ip_literal() { + // NOTE: TEST-NET-3 documentation range — public per the enforced ranges, + // never actually fetched by this validation. + assert!( + validate_download_url("http://203.0.113.10/file.torrent") + .await + .is_ok() + ); + assert!( + validate_download_url("https://203.0.113.10:8443/file.nzb") + .await + .is_ok() + ); + } + + #[test] + fn ip_range_classification() { + assert!(ip_is_disallowed(IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)))); + assert!(ip_is_disallowed(IpAddr::V4(Ipv4Addr::new(10, 1, 2, 3)))); + assert!(ip_is_disallowed(IpAddr::V4(Ipv4Addr::new(172, 31, 0, 1)))); + assert!(ip_is_disallowed(IpAddr::V4(Ipv4Addr::new(192, 168, 0, 1)))); + assert!(ip_is_disallowed(IpAddr::V4(Ipv4Addr::new(169, 254, 1, 1)))); + assert!(ip_is_disallowed(IpAddr::V4(Ipv4Addr::new(100, 127, 0, 1)))); + assert!(ip_is_disallowed(IpAddr::V6(Ipv6Addr::LOCALHOST))); + assert!(!ip_is_disallowed(IpAddr::V4(Ipv4Addr::new(203, 0, 113, 1)))); + assert!(!ip_is_disallowed(IpAddr::V4(Ipv4Addr::new(8, 8, 8, 8)))); + assert!(!ip_is_disallowed(IpAddr::V4(Ipv4Addr::new(100, 63, 0, 1)))); + assert!(!ip_is_disallowed(IpAddr::V4(Ipv4Addr::new(172, 32, 0, 1)))); + } +} diff --git a/crates/paroche/src/routes/download.rs b/crates/paroche/src/routes/download.rs index 23475d76..c6478c39 100644 --- a/crates/paroche/src/routes/download.rs +++ b/crates/paroche/src/routes/download.rs @@ -3,7 +3,7 @@ use axum::{ Json, extract::{Path, State}, }; -use exousia::AuthenticatedUser; +use exousia::{AuthenticatedUser, RequireAdmin}; use serde::{Deserialize, Serialize}; use tracing; use uuid::Uuid; @@ -166,9 +166,11 @@ pub async fn get_queue_snapshot( Ok(ApiResponse::ok(snapshot)) } +// WHY: admin-only — enqueueing hands an arbitrary URL to the server-side +// download engine; member-level access is an SSRF primitive. pub async fn enqueue_download( State(state): State, - _auth: AuthenticatedUser, + _admin: RequireAdmin, Json(body): Json, ) -> Result { if body.download_url.trim().is_empty() { @@ -177,6 +179,8 @@ pub async fn enqueue_download( }); } + crate::net_validate::validate_download_url(&body.download_url).await?; + let id = Uuid::now_v7().as_bytes().to_vec(); let want_id = Uuid::parse_str(&body.want_id) .map_err(|_| ParocheError::InvalidId)? @@ -282,3 +286,180 @@ pub fn download_routes() -> axum::Router { .route("/{id}", axum::routing::delete(cancel_download)) .route("/{id}/priority", patch(reprioritize_download)) } + +#[cfg(test)] +mod tests { + use axum::body::{Body, to_bytes}; + use axum::http::{Request, StatusCode}; + use exousia::AuthService; + use exousia::user::{CreateUserRequest, UserRole}; + use tower::ServiceExt; + + #[expect( + unused_imports, + reason = "kanon: test-missing-use-super; parent items accessed via explicit super:: prefix in test bodies" + )] + use super::*; + use crate::test_helpers::test_state; + + async fn token_for( + auth: &std::sync::Arc, + username: &str, + role: UserRole, + ) -> String { + auth.create_user(CreateUserRequest { + username: username.to_string(), + display_name: username.to_string(), + password: "password123".to_string(), + role, + }) + .await + .unwrap(); + auth.login(username, "password123") + .await + .unwrap() + .access_token + } + + fn enqueue_body(download_url: &str) -> String { + serde_json::json!({ + "want_id": uuid::Uuid::now_v7().to_string(), + "release_id": uuid::Uuid::now_v7().to_string(), + "download_url": download_url, + }) + .to_string() + } + + async fn post_enqueue( + app: &axum::Router, + token: &str, + download_url: &str, + ) -> axum::response::Response { + app.clone() + .oneshot( + Request::builder() + .method("POST") + .uri("/api/v1/downloads") + .header("Authorization", format!("Bearer {token}")) + .header("content-type", "application/json") + .body(Body::from(enqueue_body(download_url))) + .unwrap(), + ) + .await + .unwrap() + } + + #[tokio::test] + async fn enqueue_download_rejects_unauthenticated() { + let (state, _auth) = test_state().await; + let app = crate::build_router(state); + let resp = app + .oneshot( + Request::builder() + .method("POST") + .uri("/api/v1/downloads") + .header("content-type", "application/json") + .body(Body::from(enqueue_body("http://203.0.113.10/f.torrent"))) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::UNAUTHORIZED); + } + + #[tokio::test] + async fn enqueue_download_requires_admin() { + let (state, auth) = test_state().await; + let token = token_for(&auth, "member", UserRole::Member).await; + let app = crate::build_router(state); + let resp = post_enqueue(&app, &token, "http://203.0.113.10/f.torrent").await; + assert_eq!(resp.status(), StatusCode::FORBIDDEN); + } + + #[tokio::test] + async fn enqueue_download_rejects_private_hosts() { + let (state, auth) = test_state().await; + let token = token_for(&auth, "admin", UserRole::Admin).await; + let app = crate::build_router(state); + + for url in [ + "http://127.0.0.1/f.torrent", + "http://169.254.169.254/latest/meta-data/", + "http://10.0.0.5/f.torrent", + "http://192.168.1.10/f.torrent", + "http://[::1]/f.torrent", + ] { + let resp = post_enqueue(&app, &token, url).await; + assert_eq!( + resp.status(), + StatusCode::UNPROCESSABLE_ENTITY, + "expected 422 for {url}" + ); + } + + let count: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM download_queue") + .fetch_one(&app_pool(&auth)) + .await + .unwrap(); + assert_eq!(count, 0, "no rejected URL may reach the queue"); + } + + fn app_pool(auth: &std::sync::Arc) -> sqlx::SqlitePool { + auth.pools().read.clone() + } + + #[tokio::test] + async fn enqueue_download_rejects_non_http_schemes() { + let (state, auth) = test_state().await; + let token = token_for(&auth, "admin", UserRole::Admin).await; + let app = crate::build_router(state); + + for url in ["ftp://203.0.113.10/f.torrent", "file:///etc/passwd"] { + let resp = post_enqueue(&app, &token, url).await; + assert_eq!( + resp.status(), + StatusCode::UNPROCESSABLE_ENTITY, + "expected 422 for {url}" + ); + } + } + + #[tokio::test] + async fn enqueue_download_accepts_public_url_for_admin() { + let (state, auth) = test_state().await; + let token = token_for(&auth, "admin", UserRole::Admin).await; + let app = crate::build_router(state); + let resp = post_enqueue(&app, &token, "http://203.0.113.10/f.torrent").await; + assert_eq!(resp.status(), StatusCode::CREATED); + let bytes = to_bytes(resp.into_body(), usize::MAX).await.unwrap(); + let body: serde_json::Value = serde_json::from_slice(&bytes).unwrap(); + assert_eq!( + body["data"]["download_url"], + "http://203.0.113.10/f.torrent" + ); + assert_eq!(body["data"]["status"], "queued"); + } + + #[tokio::test] + async fn enqueue_download_accepts_magnet_uri_for_admin() { + let (state, auth) = test_state().await; + let token = token_for(&auth, "admin", UserRole::Admin).await; + let app = crate::build_router(state); + let resp = post_enqueue(&app, &token, "magnet:?xt=urn:btih:abc123def456").await; + assert_eq!(resp.status(), StatusCode::CREATED); + } + + #[tokio::test] + async fn enqueue_download_rejects_magnet_with_private_tracker() { + let (state, auth) = test_state().await; + let token = token_for(&auth, "admin", UserRole::Admin).await; + let app = crate::build_router(state); + let resp = post_enqueue( + &app, + &token, + "magnet:?xt=urn:btih:abc&tr=http%3A%2F%2F127.0.0.1%3A8080%2Fannounce", + ) + .await; + assert_eq!(resp.status(), StatusCode::UNPROCESSABLE_ENTITY); + } +} diff --git a/crates/paroche/src/routes/kosync.rs b/crates/paroche/src/routes/kosync.rs index 69d94085..4c78bf20 100644 --- a/crates/paroche/src/routes/kosync.rs +++ b/crates/paroche/src/routes/kosync.rs @@ -11,6 +11,15 @@ use crate::state::AppState; // KOSync wire protocol: implements the KOReader sync-server 4-endpoint surface. // See: https://github.com/koreader/koreader/blob/master/plugins/kosync.koplugin/api.json +// WHY: subtle's ConstantTimeEq avoids the timing side-channel of a +// short-circuiting `!=` on secret material; unequal lengths return +// not-equal without comparing content (length is public here — the stored +// side is always a 40-char SHA1 hex string). +fn constant_time_str_eq(a: &str, b: &str) -> bool { + use subtle::ConstantTimeEq; + a.as_bytes().ct_eq(b.as_bytes()).into() +} + fn sha1_hex(input: &[u8]) -> String { const HEX: &[u8; 16] = b"0123456789abcdef"; @@ -149,7 +158,7 @@ pub async fn auth_user( .ok_or(ParocheError::Unauthorized)?; // Constant-time comparison to prevent timing attacks - if user.password_hash != provided_key { + if !constant_time_str_eq(&user.password_hash, provided_key) { return Err(ParocheError::Unauthorized); } @@ -183,7 +192,8 @@ pub async fn put_progress( .await? .ok_or(ParocheError::Unauthorized)?; - if user.password_hash != provided_key { + // Constant-time comparison to prevent timing attacks + if !constant_time_str_eq(&user.password_hash, provided_key) { return Err(ParocheError::Unauthorized); } @@ -248,7 +258,8 @@ pub async fn get_progress( .await? .ok_or(ParocheError::Unauthorized)?; - if user.password_hash != provided_key { + // Constant-time comparison to prevent timing attacks + if !constant_time_str_eq(&user.password_hash, provided_key) { return Err(ParocheError::Unauthorized); } @@ -354,6 +365,66 @@ mod tests { assert_eq!(parsed["username"], "reader1"); } + #[test] + fn constant_time_str_eq_matches_expected_semantics() { + assert!(constant_time_str_eq( + "5d41402abc4b2a76b9719d911017c592", + "5d41402abc4b2a76b9719d911017c592" + )); + assert!(!constant_time_str_eq( + "5d41402abc4b2a76b9719d911017c592", + "5d41402abc4b2a76b9719d911017c593" + )); + assert!(!constant_time_str_eq( + "5d41402abc4b2a76b9719d911017c592", + "short" + )); + assert!(!constant_time_str_eq("", "x")); + assert!(constant_time_str_eq("", "")); + } + + #[tokio::test] + async fn auth_with_wrong_key_returns_401() { + let (state, _) = test_state().await; + let app = super::super::super::build_router(state); + + let create_body = serde_json::json!({ + "username": "reader5", + "password": "correcthorse" + }); + let create_resp = app + .clone() + .oneshot( + Request::builder() + .method("POST") + .uri("/kosync/users/create") + .header("content-type", "application/json") + .body(axum::body::Body::from( + serde_json::to_string(&create_body).unwrap(), + )) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(create_resp.status(), StatusCode::CREATED); + + // Same length as a real SHA1 hex digest, wrong content + let wrong_key = sha1_hex(b"wrongpassword"); + let auth_resp = app + .oneshot( + Request::builder() + .method("GET") + .uri("/kosync/users/auth") + .header("x-auth-user", "reader5") + .header("x-auth-key", &wrong_key) + .body(axum::body::Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(auth_resp.status(), StatusCode::UNAUTHORIZED); + } + #[tokio::test] async fn auth_without_user_returns_401() { let (state, _) = test_state().await; diff --git a/crates/paroche/src/routes/music.rs b/crates/paroche/src/routes/music.rs index 6bad7997..2d7c8aa5 100644 --- a/crates/paroche/src/routes/music.rs +++ b/crates/paroche/src/routes/music.rs @@ -214,10 +214,11 @@ pub async fn list_tracks( let page = pagination.page.max(1); let offset = (page - 1) * per_page; - let tracks = apotheke::repo::music::search_tracks(&state.db.read, "", per_page as i64).await?; + let tracks = + apotheke::repo::music::search_tracks(&state.db.read, "", per_page as i64, offset as i64) + .await?; - let _ = offset; - let total = tracks.len() as u64; + let total = apotheke::repo::music::count_tracks(&state.db.read, "").await? as u64; let data: Vec = tracks.into_iter().map(Into::into).collect(); Ok(ApiResponse::paginated(data, page, per_page, total)) } @@ -471,6 +472,122 @@ mod tests { assert_eq!(resp.status(), StatusCode::NO_CONTENT); } + async fn seed_tracks(pool: &sqlx::SqlitePool, count: usize) { + let now = "2026-01-01T00:00:00Z"; + let group_id = Uuid::now_v7().as_bytes().to_vec(); + sqlx::query( + "INSERT INTO music_release_groups (id, title, rg_type, added_at) + VALUES (?, 'Seed Album', 'album', ?)", + ) + .bind(&group_id) + .bind(now) + .execute(pool) + .await + .unwrap(); + + let release_id = Uuid::now_v7().as_bytes().to_vec(); + sqlx::query( + "INSERT INTO music_releases (id, release_group_id, title, added_at) + VALUES (?, ?, 'Seed Album', ?)", + ) + .bind(&release_id) + .bind(&group_id) + .bind(now) + .execute(pool) + .await + .unwrap(); + + let medium_id = Uuid::now_v7().as_bytes().to_vec(); + sqlx::query( + "INSERT INTO music_media (id, release_id, position, format) VALUES (?, ?, 1, 'Digital')", + ) + .bind(&medium_id) + .bind(&release_id) + .execute(pool) + .await + .unwrap(); + + for i in 0..count { + let track_id = Uuid::now_v7().as_bytes().to_vec(); + sqlx::query( + "INSERT INTO music_tracks (id, medium_id, position, title, source_type, added_at) + VALUES (?, ?, ?, ?, 'local', ?)", + ) + .bind(&track_id) + .bind(&medium_id) + .bind(i as i64) + .bind(format!("Seed Track {i:02}")) + .bind(now) + .execute(pool) + .await + .unwrap(); + } + } + + async fn get_tracks_page( + app: &axum::Router, + token: &str, + page: u64, + per_page: u64, + ) -> serde_json::Value { + let resp = app + .clone() + .oneshot( + Request::builder() + .uri(format!("/tracks?page={page}&per_page={per_page}")) + .header("Authorization", format!("Bearer {token}")) + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::OK); + let bytes = to_bytes(resp.into_body(), usize::MAX).await.unwrap(); + serde_json::from_slice(&bytes).unwrap() + } + + #[tokio::test] + async fn list_tracks_paginates_disjoint_pages() { + let (state, auth) = test_state().await; + let token = member_token(&auth).await; + seed_tracks(&state.db.write, 7).await; + let app = music_routes().with_state(state); + + let page1 = get_tracks_page(&app, &token, 1, 5).await; + let page2 = get_tracks_page(&app, &token, 2, 5).await; + + let ids1: std::collections::HashSet = page1["data"] + .as_array() + .unwrap() + .iter() + .map(|t| t["id"].as_str().unwrap().to_string()) + .collect(); + let ids2: std::collections::HashSet = page2["data"] + .as_array() + .unwrap() + .iter() + .map(|t| t["id"].as_str().unwrap().to_string()) + .collect(); + + assert_eq!(ids1.len(), 5); + assert_eq!(ids2.len(), 2); + assert!(ids1.is_disjoint(&ids2), "page 2 must not repeat page 1"); + } + + #[tokio::test] + async fn list_tracks_reports_full_total() { + let (state, auth) = test_state().await; + let token = member_token(&auth).await; + seed_tracks(&state.db.write, 7).await; + let app = music_routes().with_state(state); + + let page1 = get_tracks_page(&app, &token, 1, 5).await; + assert_eq!(page1["data"].as_array().unwrap().len(), 5); + assert_eq!(page1["meta"]["total"], 7); + assert_eq!(page1["meta"]["page"], 1); + assert_eq!(page1["meta"]["per_page"], 5); + } + #[tokio::test] async fn get_nonexistent_release_group_returns_404() { let (state, auth) = test_state().await; diff --git a/crates/paroche/src/routes/user.rs b/crates/paroche/src/routes/user.rs index 89eb4f55..d635fb83 100644 --- a/crates/paroche/src/routes/user.rs +++ b/crates/paroche/src/routes/user.rs @@ -1,11 +1,12 @@ use axum::Json; -use axum::extract::State; +use axum::extract::{Path, State}; use axum::http::StatusCode; use exousia::user::{CreateUserRequest, UserRole}; use exousia::{AuthService, RequireAdmin, TokenPair}; use serde::{Deserialize, Serialize}; +use snafu::ResultExt; -use crate::error::ParocheError; +use crate::error::{DatabaseSnafu, ParocheError}; use crate::response::ApiResponse; use crate::state::AppState; @@ -118,8 +119,11 @@ pub async fn list_users( .await .map_err(ParocheError::from)?; + // WHY: deactivated users are soft-deleted — they must not reappear in + // the roster (DELETE /users/{id} contract). let data: Vec = users .into_iter() + .filter(|u| u.is_active != 0) .filter_map(|u| { let id_bytes = &u.id; let uuid = uuid::Uuid::from_slice(id_bytes).ok()?; @@ -169,10 +173,34 @@ pub async fn create_user( } pub async fn delete_user( - State(_state): State, + State(state): State, + Path(id): Path, _admin: RequireAdmin, -) -> impl axum::response::IntoResponse { - StatusCode::NO_CONTENT +) -> Result { + // WHY: an unparseable id addresses a resource that cannot exist — 404, + // matching the unknown-id case rather than leaking format details. + let uuid = uuid::Uuid::parse_str(&id).map_err(|_| ParocheError::NotFound)?; + let id_bytes = uuid.as_bytes().to_vec(); + + apotheke::repo::user::get_user(&state.db.read, &id_bytes) + .await + .context(DatabaseSnafu)? + .ok_or(ParocheError::NotFound)?; + + // WHY: soft-delete — deactivation preserves FK integrity (playlists, + // downloads, history) while credential revocation severs every access + // path that consults the database (refresh tokens, API keys, new logins). + apotheke::repo::user::deactivate_user(&state.db.write, &id_bytes) + .await + .context(DatabaseSnafu)?; + apotheke::repo::user::delete_refresh_tokens_for_user(&state.db.write, &id_bytes) + .await + .context(DatabaseSnafu)?; + apotheke::repo::user::revoke_api_keys_for_user(&state.db.write, &id_bytes) + .await + .context(DatabaseSnafu)?; + + Ok(StatusCode::NO_CONTENT) } pub fn auth_routes() -> axum::Router { @@ -291,6 +319,186 @@ mod tests { assert_eq!(resp.status(), StatusCode::FORBIDDEN); } + async fn admin_setup(auth: &std::sync::Arc) -> String { + auth.create_user(CreateUserRequest { + username: "root".to_string(), + display_name: "Root".to_string(), + password: "password123".to_string(), + role: exousia::user::UserRole::Admin, + }) + .await + .unwrap(); + auth.login("root", "password123") + .await + .unwrap() + .access_token + } + + #[tokio::test] + async fn delete_user_unknown_id_returns_404() { + let (state, auth) = test_state().await; + let admin = admin_setup(&auth).await; + let app = make_app(state); + + for id in [uuid::Uuid::now_v7().to_string(), "not-a-uuid".to_string()] { + let resp = app + .clone() + .oneshot( + Request::builder() + .method("DELETE") + .uri(format!("/users/{id}")) + .header("Authorization", format!("Bearer {admin}")) + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!( + resp.status(), + StatusCode::NOT_FOUND, + "expected 404 for {id}" + ); + } + } + + #[tokio::test] + async fn delete_user_requires_admin() { + let (state, auth) = test_state().await; + auth.create_user(CreateUserRequest { + username: "member".to_string(), + display_name: "Member".to_string(), + password: "password123".to_string(), + role: exousia::user::UserRole::Member, + }) + .await + .unwrap(); + let token = auth + .login("member", "password123") + .await + .unwrap() + .access_token; + let app = make_app(state); + + let resp = app + .oneshot( + Request::builder() + .method("DELETE") + .uri(format!("/users/{}", uuid::Uuid::now_v7())) + .header("Authorization", format!("Bearer {token}")) + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::FORBIDDEN); + } + + #[tokio::test] + async fn delete_user_deactivates_and_revokes() { + let (state, auth) = test_state().await; + let admin = admin_setup(&auth).await; + + let victim = auth + .create_user(CreateUserRequest { + username: "victim".to_string(), + display_name: "Victim".to_string(), + password: "password123".to_string(), + role: exousia::user::UserRole::Member, + }) + .await + .unwrap(); + let victim_id = victim.id.into_uuid().to_string(); + let victim_pair = auth.login("victim", "password123").await.unwrap(); + + let app = make_app(state); + + // Delete (deactivate) as admin + let resp = app + .clone() + .oneshot( + Request::builder() + .method("DELETE") + .uri(format!("/users/{victim_id}")) + .header("Authorization", format!("Bearer {admin}")) + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::NO_CONTENT); + + // Fresh login is rejected + let resp = app + .clone() + .oneshot( + Request::builder() + .method("POST") + .uri("/auth/login") + .header("Content-Type", "application/json") + .body(Body::from( + r#"{"username":"victim","password":"password123"}"#, + )) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::UNAUTHORIZED); + + // Prior refresh token is rejected + let refresh_body = + serde_json::json!({ "refresh_token": victim_pair.refresh_token }).to_string(); + let resp = app + .clone() + .oneshot( + Request::builder() + .method("POST") + .uri("/auth/refresh") + .header("Content-Type", "application/json") + .body(Body::from(refresh_body)) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::UNAUTHORIZED); + + // No longer listed + let resp = app + .clone() + .oneshot( + Request::builder() + .uri("/users") + .header("Authorization", format!("Bearer {admin}")) + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::OK); + let bytes = to_bytes(resp.into_body(), usize::MAX).await.unwrap(); + let json: serde_json::Value = serde_json::from_slice(&bytes).unwrap(); + let listed: Vec<&str> = json["data"] + .as_array() + .unwrap() + .iter() + .map(|u| u["id"].as_str().unwrap()) + .collect(); + assert!(!listed.contains(&victim_id.as_str())); + + // Repeat delete of an already-deactivated (but existing) user is idempotent + let resp = app + .oneshot( + Request::builder() + .method("DELETE") + .uri(format!("/users/{victim_id}")) + .header("Authorization", format!("Bearer {admin}")) + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::NO_CONTENT); + } + #[tokio::test] async fn admin_can_list_users() { let (state, auth) = test_state().await; diff --git a/crates/paroche/src/routes/zone.rs b/crates/paroche/src/routes/zone.rs index 8f750e2a..0b0eecd3 100644 --- a/crates/paroche/src/routes/zone.rs +++ b/crates/paroche/src/routes/zone.rs @@ -4,6 +4,7 @@ use axum::{ Json, extract::{Path, State}, }; +use exousia::{AuthenticatedUser, RequireAdmin}; use serde::{Deserialize, Serialize}; use crate::error::ParocheError; @@ -60,6 +61,7 @@ pub struct AddMemberBody { pub async fn create_zone( State(state): State, + _admin: RequireAdmin, Json(body): Json, ) -> Result { if body.name.trim().is_empty() { @@ -80,6 +82,7 @@ pub async fn create_zone( pub async fn delete_zone( State(state): State, Path(id): Path, + _admin: RequireAdmin, ) -> Result { zone::delete_zone(&state.db.write, &id).await?; Ok(deleted()) @@ -87,6 +90,7 @@ pub async fn delete_zone( pub async fn list_zones( State(state): State, + _auth: AuthenticatedUser, ) -> Result { let zones = zone::list_zones(&state.db.read).await?; let data: Vec = zones.into_iter().map(Into::into).collect(); @@ -96,6 +100,7 @@ pub async fn list_zones( pub async fn get_zone( State(state): State, Path(id): Path, + _auth: AuthenticatedUser, ) -> Result { let z = zone::get_zone(&state.db.read, &id).await?; Ok(ApiResponse::ok(ZoneResponse::from(z))) @@ -104,6 +109,7 @@ pub async fn get_zone( pub async fn add_member( State(state): State, Path(zone_id): Path, + _admin: RequireAdmin, Json(body): Json, ) -> Result { if body.renderer_id.trim().is_empty() { @@ -120,6 +126,7 @@ pub async fn add_member( pub async fn remove_member( State(state): State, Path((zone_id, renderer_id)): Path<(String, String)>, + _admin: RequireAdmin, ) -> Result { zone::remove_member(&state.db.write, &zone_id, &renderer_id).await?; Ok(deleted()) @@ -128,6 +135,7 @@ pub async fn remove_member( pub async fn zone_play( State(_state): State, Path(_zone_id): Path, + _auth: AuthenticatedUser, ) -> Result { // WHY: Playback initiation requires the streaming subsystem (syndesis). // Full implementation connects to all zone renderers and starts fan-out streaming. @@ -138,6 +146,7 @@ pub async fn zone_play( pub async fn zone_pause( State(_state): State, Path(_zone_id): Path, + _auth: AuthenticatedUser, ) -> Result { Ok(ApiResponse::ok(serde_json::json!({ "status": "paused" }))) } @@ -145,6 +154,7 @@ pub async fn zone_pause( pub async fn zone_resume( State(_state): State, Path(_zone_id): Path, + _auth: AuthenticatedUser, ) -> Result { Ok(ApiResponse::ok(serde_json::json!({ "status": "playing" }))) } @@ -168,6 +178,8 @@ pub fn zone_routes() -> axum::Router { mod tests { use axum::body::Body; use axum::http::{Request, StatusCode}; + use exousia::AuthService; + use exousia::user::{CreateUserRequest, UserRole}; use tower::ServiceExt; #[expect( @@ -176,9 +188,30 @@ mod tests { )] use super::*; use crate::test_helpers::test_state; + + async fn token_for( + auth: &std::sync::Arc, + username: &str, + role: UserRole, + ) -> String { + auth.create_user(CreateUserRequest { + username: username.to_string(), + display_name: username.to_string(), + password: "password123".to_string(), + role, + }) + .await + .unwrap(); + auth.login(username, "password123") + .await + .unwrap() + .access_token + } + #[tokio::test] async fn zone_crud_lifecycle() { - let (state, _) = test_state().await; + let (state, auth) = test_state().await; + let token = token_for(&auth, "admin", UserRole::Admin).await; // Seed a renderer directly apotheke::repo::zone::upsert_renderer(&state.db.write, "r1", "Speaker", "127.0.0.1:5000") @@ -194,6 +227,7 @@ mod tests { Request::builder() .method("POST") .uri("/api/zones") + .header("Authorization", format!("Bearer {token}")) .header("content-type", "application/json") .body(Body::from(r#"{"name":"Living Room"}"#)) .unwrap(), @@ -215,6 +249,7 @@ mod tests { .oneshot( Request::builder() .uri("/api/zones") + .header("Authorization", format!("Bearer {token}")) .body(Body::empty()) .unwrap(), ) @@ -234,6 +269,7 @@ mod tests { Request::builder() .method("POST") .uri(format!("/api/zones/{zone_id}/members")) + .header("Authorization", format!("Bearer {token}")) .header("content-type", "application/json") .body(Body::from(r#"{"renderer_id":"r1"}"#)) .unwrap(), @@ -253,6 +289,7 @@ mod tests { .oneshot( Request::builder() .uri(format!("/api/zones/{zone_id}")) + .header("Authorization", format!("Bearer {token}")) .body(Body::empty()) .unwrap(), ) @@ -267,6 +304,7 @@ mod tests { Request::builder() .method("DELETE") .uri(format!("/api/zones/{zone_id}/members/r1")) + .header("Authorization", format!("Bearer {token}")) .body(Body::empty()) .unwrap(), ) @@ -281,6 +319,7 @@ mod tests { Request::builder() .method("DELETE") .uri(format!("/api/zones/{zone_id}")) + .header("Authorization", format!("Bearer {token}")) .body(Body::empty()) .unwrap(), ) @@ -290,8 +329,9 @@ mod tests { } #[tokio::test] - async fn zone_playback_controls() { - let (state, _) = test_state().await; + async fn zone_playback_controls_allow_member_role() { + let (state, auth) = test_state().await; + let token = token_for(&auth, "member", UserRole::Member).await; apotheke::repo::zone::upsert_renderer(&state.db.write, "r1", "Speaker", "127.0.0.1:5000") .await .unwrap(); @@ -315,6 +355,7 @@ mod tests { Request::builder() .method("POST") .uri(endpoint) + .header("Authorization", format!("Bearer {token}")) .body(Body::empty()) .unwrap(), ) @@ -323,4 +364,81 @@ mod tests { assert_eq!(resp.status(), StatusCode::OK, "failed for {endpoint}"); } } + + #[tokio::test] + async fn zone_routes_reject_unauthenticated() { + let (state, _auth) = test_state().await; + let app = crate::build_router(state); + + let endpoints: [(&str, String); 9] = [ + ("POST", "/api/zones".to_string()), + ("GET", "/api/zones".to_string()), + ("GET", "/api/zones/z1".to_string()), + ("DELETE", "/api/zones/z1".to_string()), + ("POST", "/api/zones/z1/members".to_string()), + ("DELETE", "/api/zones/z1/members/r1".to_string()), + ("POST", "/api/zones/z1/play".to_string()), + ("POST", "/api/zones/z1/pause".to_string()), + ("POST", "/api/zones/z1/resume".to_string()), + ]; + + for (method, uri) in endpoints { + let resp = app + .clone() + .oneshot( + Request::builder() + .method(method) + .uri(&uri) + .header("content-type", "application/json") + .body(Body::from("{}")) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!( + resp.status(), + StatusCode::UNAUTHORIZED, + "expected 401 for {method} {uri}" + ); + } + } + + #[tokio::test] + async fn zone_mutations_reject_member_role() { + let (state, auth) = test_state().await; + let token = token_for(&auth, "member", UserRole::Member).await; + let app = crate::build_router(state); + + let endpoints: [(&str, String, &str); 4] = [ + ("POST", "/api/zones".to_string(), r#"{"name":"Den"}"#), + ("DELETE", "/api/zones/z1".to_string(), "{}"), + ( + "POST", + "/api/zones/z1/members".to_string(), + r#"{"renderer_id":"r1"}"#, + ), + ("DELETE", "/api/zones/z1/members/r1".to_string(), "{}"), + ]; + + for (method, uri, body) in endpoints { + let resp = app + .clone() + .oneshot( + Request::builder() + .method(method) + .uri(&uri) + .header("Authorization", format!("Bearer {token}")) + .header("content-type", "application/json") + .body(Body::from(body)) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!( + resp.status(), + StatusCode::FORBIDDEN, + "expected 403 for {method} {uri}" + ); + } + } } diff --git a/crates/paroche/src/subsonic/media.rs b/crates/paroche/src/subsonic/media.rs index 32c11672..b5c9f200 100644 --- a/crates/paroche/src/subsonic/media.rs +++ b/crates/paroche/src/subsonic/media.rs @@ -3,7 +3,9 @@ use axum::response::Response; use serde::Deserialize; use super::auth::authenticate; -use super::types::{ERR_MISSING_PARAM, SubsonicCommon, respond_error, respond_ok, uuid_bytes}; +use super::types::{ + ERR_GENERIC, ERR_MISSING_PARAM, SubsonicCommon, respond_error, respond_ok, uuid_bytes, +}; use crate::state::AppState; #[derive(Deserialize, Default)] @@ -46,38 +48,41 @@ pub async fn star(State(state): State, Query(q): Query) -> let user_id_bytes = user.user_id.as_bytes().to_vec(); - if let Some(id) = &q.id - && let Some(bytes) = uuid_bytes(id) - { - let _ = sqlx::query( - "INSERT OR IGNORE INTO subsonic_stars (user_id, item_id, item_type) VALUES (?, ?, 'track')", - ) - .bind(&user_id_bytes) - .bind(bytes) - .execute(&state.db.write) - .await; - } - if let Some(id) = &q.album_id - && let Some(bytes) = uuid_bytes(id) - { - let _ = sqlx::query( - "INSERT OR IGNORE INTO subsonic_stars (user_id, item_id, item_type) VALUES (?, ?, 'album')", - ) - .bind(&user_id_bytes) - .bind(bytes) - .execute(&state.db.write) - .await; + // WHY: star may touch up to three rows; failures must surface and apply + // all-or-nothing rather than silently reporting ok. + let mut tx = match state.db.write.begin().await { + Ok(tx) => tx, + Err(e) => { + tracing::warn!(error = %e, "star: begin transaction failed"); + return respond_error(user.format, ERR_GENERIC, "could not star item"); + } + }; + + let targets = [ + (&q.id, "track"), + (&q.album_id, "album"), + (&q.artist_id, "artist"), + ]; + for (raw_id, item_type) in targets { + if let Some(id) = raw_id + && let Some(bytes) = uuid_bytes(id) + && let Err(e) = sqlx::query( + "INSERT OR IGNORE INTO subsonic_stars (user_id, item_id, item_type) VALUES (?, ?, ?)", + ) + .bind(&user_id_bytes) + .bind(bytes) + .bind(item_type) + .execute(&mut *tx) + .await + { + tracing::warn!(error = %e, item_type, "star: insert failed"); + return respond_error(user.format, ERR_GENERIC, "could not star item"); + } } - if let Some(id) = &q.artist_id - && let Some(bytes) = uuid_bytes(id) - { - let _ = sqlx::query( - "INSERT OR IGNORE INTO subsonic_stars (user_id, item_id, item_type) VALUES (?, ?, 'artist')", - ) - .bind(&user_id_bytes) - .bind(bytes) - .execute(&state.db.write) - .await; + + if let Err(e) = tx.commit().await { + tracing::warn!(error = %e, "star: commit failed"); + return respond_error(user.format, ERR_GENERIC, "could not star item"); } respond_ok(user.format, "", None) @@ -95,16 +100,33 @@ pub async fn unstar(State(state): State, Query(q): Query) - let user_id_bytes = user.user_id.as_bytes().to_vec(); + let mut tx = match state.db.write.begin().await { + Ok(tx) => tx, + Err(e) => { + tracing::warn!(error = %e, "unstar: begin transaction failed"); + return respond_error(user.format, ERR_GENERIC, "could not unstar item"); + } + }; + for id in [&q.id, &q.album_id, &q.artist_id].into_iter().flatten() { - if let Some(bytes) = uuid_bytes(id) { - let _ = sqlx::query("DELETE FROM subsonic_stars WHERE user_id = ? AND item_id = ?") - .bind(&user_id_bytes) - .bind(bytes) - .execute(&state.db.write) - .await; + if let Some(bytes) = uuid_bytes(id) + && let Err(e) = + sqlx::query("DELETE FROM subsonic_stars WHERE user_id = ? AND item_id = ?") + .bind(&user_id_bytes) + .bind(bytes) + .execute(&mut *tx) + .await + { + tracing::warn!(error = %e, "unstar: delete failed"); + return respond_error(user.format, ERR_GENERIC, "could not unstar item"); } } + if let Err(e) = tx.commit().await { + tracing::warn!(error = %e, "unstar: commit failed"); + return respond_error(user.format, ERR_GENERIC, "could not unstar item"); + } + respond_ok(user.format, "", None) } @@ -134,12 +156,16 @@ pub async fn set_rating(State(state): State, Query(q): Query, Query(q): Query, Query(q): Query tx, + Err(e) => { + tracing::warn!(error = %e, "create_playlist: begin transaction failed"); + return respond_error(user.format, ERR_GENERIC, "could not create playlist"); + } + }; + + if let Err(e) = + sqlx::query("INSERT INTO subsonic_playlists (id, owner_id, name) VALUES (?, ?, ?)") + .bind(&playlist_id) + .bind(&user_id_bytes) + .bind(&name) + .execute(&mut *tx) + .await + { + tracing::warn!(error = %e, "create_playlist: playlist insert failed"); + return respond_error(user.format, ERR_GENERIC, "could not create playlist"); + } if let Some(song_ids) = &q.song_ids { for (pos, sid) in song_ids.iter().enumerate() { - if let Some(track_bytes) = uuid_bytes(sid) { - let _ = sqlx::query( + if let Some(track_bytes) = uuid_bytes(sid) + && let Err(e) = sqlx::query( "INSERT OR IGNORE INTO subsonic_playlist_tracks (playlist_id, track_id, position) VALUES (?, ?, ?)", ) .bind(&playlist_id) .bind(track_bytes) .bind(pos as i64) // INVARIANT: pos is a Vec enumerate index, bounded by collection size; i64 overflow impossible - .execute(&state.db.write) - .await; + .execute(&mut *tx) + .await + { + tracing::warn!(error = %e, "create_playlist: track insert failed"); + return respond_error(user.format, ERR_GENERIC, "could not create playlist"); } } } + if let Err(e) = tx.commit().await { + tracing::warn!(error = %e, "create_playlist: commit failed"); + return respond_error(user.format, ERR_GENERIC, "could not create playlist"); + } + let pl_id_str = uuid_str(&playlist_id); let xml = format!( r#""#, @@ -345,61 +369,94 @@ pub async fn update_playlist( return respond_error(user.format, ERR_NOT_FOUND, "not found"); } - if let Some(name) = &q.name { - let _ = sqlx::query( + // WHY: single transaction — partial metadata/track updates must not + // survive a mid-flight failure, and failures must surface to the client. + let mut tx = match state.db.write.begin().await { + Ok(tx) => tx, + Err(e) => { + tracing::warn!(error = %e, "update_playlist: begin transaction failed"); + return respond_error(user.format, ERR_GENERIC, "could not update playlist"); + } + }; + + if let Some(name) = &q.name + && let Err(e) = sqlx::query( "UPDATE subsonic_playlists SET name = ?, updated_at = strftime('%Y-%m-%dT%H:%M:%SZ', 'now') WHERE id = ?", ) .bind(name) .bind(&id_bytes) - .execute(&state.db.write) - .await; + .execute(&mut *tx) + .await + { + tracing::warn!(error = %e, "update_playlist: name update failed"); + return respond_error(user.format, ERR_GENERIC, "could not update playlist"); } - if let Some(comment) = &q.comment { - let _ = sqlx::query( + if let Some(comment) = &q.comment + && let Err(e) = sqlx::query( "UPDATE subsonic_playlists SET comment = ?, updated_at = strftime('%Y-%m-%dT%H:%M:%SZ', 'now') WHERE id = ?", ) .bind(comment) .bind(&id_bytes) - .execute(&state.db.write) - .await; + .execute(&mut *tx) + .await + { + tracing::warn!(error = %e, "update_playlist: comment update failed"); + return respond_error(user.format, ERR_GENERIC, "could not update playlist"); } - if let Some(public) = q.public { - let _ = sqlx::query( + if let Some(public) = q.public + && let Err(e) = sqlx::query( "UPDATE subsonic_playlists SET public = ?, updated_at = strftime('%Y-%m-%dT%H:%M:%SZ', 'now') WHERE id = ?", ) .bind(if public { 1i64 } else { 0i64 }) .bind(&id_bytes) - .execute(&state.db.write) - .await; + .execute(&mut *tx) + .await + { + tracing::warn!(error = %e, "update_playlist: public update failed"); + return respond_error(user.format, ERR_GENERIC, "could not update playlist"); } // Append songs if let Some(song_ids) = &q.song_ids_to_add { // Get current max position - let max_pos: i64 = sqlx::query_scalar( + let max_pos: i64 = match sqlx::query_scalar( "SELECT COALESCE(MAX(position), -1) FROM subsonic_playlist_tracks WHERE playlist_id = ?", ) .bind(&id_bytes) - .fetch_one(&state.db.read) + .fetch_one(&mut *tx) .await - .unwrap_or(-1); + { + Ok(v) => v, + Err(e) => { + tracing::warn!(error = %e, "update_playlist: max position query failed"); + return respond_error(user.format, ERR_GENERIC, "could not update playlist"); + } + }; for (i, sid) in song_ids.iter().enumerate() { - if let Some(track_bytes) = uuid_bytes(sid) { - let _ = sqlx::query( + if let Some(track_bytes) = uuid_bytes(sid) + && let Err(e) = sqlx::query( "INSERT OR IGNORE INTO subsonic_playlist_tracks (playlist_id, track_id, position) VALUES (?, ?, ?)", ) .bind(&id_bytes) .bind(track_bytes) .bind(max_pos + 1 + i as i64) // INVARIANT: i is a Vec enumerate index; i64 overflow impossible - .execute(&state.db.write) - .await; + .execute(&mut *tx) + .await + { + tracing::warn!(error = %e, "update_playlist: track append failed"); + return respond_error(user.format, ERR_GENERIC, "could not update playlist"); } } } + if let Err(e) = tx.commit().await { + tracing::warn!(error = %e, "update_playlist: commit failed"); + return respond_error(user.format, ERR_GENERIC, "could not update playlist"); + } + respond_ok(user.format, "", None) } @@ -432,11 +489,15 @@ pub async fn delete_playlist( }; let user_id_bytes = user.user_id.as_bytes().to_vec(); - let _ = sqlx::query("DELETE FROM subsonic_playlists WHERE id = ? AND owner_id = ?") + if let Err(e) = sqlx::query("DELETE FROM subsonic_playlists WHERE id = ? AND owner_id = ?") .bind(&id_bytes) .bind(user_id_bytes) .execute(&state.db.write) - .await; + .await + { + tracing::warn!(error = %e, "delete_playlist: delete failed"); + return respond_error(user.format, ERR_GENERIC, "could not delete playlist"); + } respond_ok(user.format, "", None) } @@ -529,10 +590,6 @@ mod tests { use axum::http::Request; use tower::ServiceExt; - #[expect( - unused_imports, - reason = "kanon: test-missing-use-super; parent items accessed via explicit super:: prefix in test bodies" - )] use super::*; use crate::subsonic::test_helpers::subsonic_app; #[tokio::test] @@ -607,4 +664,150 @@ mod tests { let body = std::str::from_utf8(&bytes).unwrap(); assert!(body.contains("status=\"ok\"")); } + + #[tokio::test] + async fn create_playlist_insert_failure_returns_error() { + let (app, state, key) = subsonic_app().await; + + // Force the playlist INSERT to fail deterministically + sqlx::query( + "CREATE TRIGGER force_pl_insert_fail BEFORE INSERT ON subsonic_playlists \ + BEGIN SELECT RAISE(ABORT, 'forced test failure'); END", + ) + .execute(&state.db.write) + .await + .unwrap(); + + let resp = app + .oneshot( + Request::builder() + .uri(format!( + "/rest/createPlaylist.view?apiKey={key}&name=Doomed" + )) + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + let bytes = to_bytes(resp.into_body(), usize::MAX).await.unwrap(); + let body = std::str::from_utf8(&bytes).unwrap(); + assert!( + body.contains("status=\"failed\""), + "expected failed status, got: {body}" + ); + assert!(!body.contains("status=\"ok\"")); + assert!(body.contains(r#"code="0""#), "expected ERR_GENERIC code"); + } + + #[tokio::test] + async fn create_playlist_failure_leaves_no_orphan_tracks() { + // WHY: the issue's core bug is "failed playlist insert still inserts + // tracks and returns ok". This asserts no track rows leak after a + // failed create — the swallowed-write path is closed. + let (app, state, key) = subsonic_app().await; + + sqlx::query( + "CREATE TRIGGER force_pl_insert_fail BEFORE INSERT ON subsonic_playlists \ + BEGIN SELECT RAISE(ABORT, 'forced test failure'); END", + ) + .execute(&state.db.write) + .await + .unwrap(); + + let resp = app + .oneshot( + Request::builder() + .uri(format!( + "/rest/createPlaylist.view?apiKey={key}&name=Orphaned" + )) + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + let bytes = to_bytes(resp.into_body(), usize::MAX).await.unwrap(); + let body = std::str::from_utf8(&bytes).unwrap(); + assert!( + body.contains("status=\"failed\""), + "expected failed status, got: {body}" + ); + + let track_rows: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM subsonic_playlist_tracks") + .fetch_one(&state.db.read) + .await + .unwrap(); + assert_eq!( + track_rows, 0, + "no track rows may be written on a failed create" + ); + } + + #[tokio::test] + async fn update_playlist_rolls_back_earlier_writes_on_failure() { + // WHY: proves the transaction property directly — an earlier UPDATE + // that succeeds inside the tx must be undone when a later statement in + // the same tx fails. A trigger aborts the `public=1` UPDATE, so the + // preceding `name` UPDATE must roll back and the response must be + // failed (not a partial-write ok). + let (app, state, key) = subsonic_app().await; + + let resp = app + .clone() + .oneshot( + Request::builder() + .uri(format!( + "/rest/createPlaylist.view?apiKey={key}&name=Original" + )) + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + let bytes = to_bytes(resp.into_body(), usize::MAX).await.unwrap(); + let body = std::str::from_utf8(&bytes).unwrap(); + let id_start = body.find("id=\"").unwrap() + 4; + let id_end = body[id_start..].find('"').unwrap() + id_start; + let pl_id = body[id_start..id_end].to_string(); + + // Abort any UPDATE that sets public = 1 — a deterministic, reachable + // mid-transaction failure after the name UPDATE has already applied. + sqlx::query( + "CREATE TRIGGER fail_on_public BEFORE UPDATE ON subsonic_playlists + WHEN NEW.public = 1 + BEGIN SELECT RAISE(ABORT, 'boom'); END", + ) + .execute(&state.db.write) + .await + .unwrap(); + + let resp = app + .oneshot( + Request::builder() + .uri(format!( + "/rest/updatePlaylist.view?apiKey={key}&playlistId={pl_id}&name=Renamed&public=true" + )) + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + let bytes = to_bytes(resp.into_body(), usize::MAX).await.unwrap(); + let body = std::str::from_utf8(&bytes).unwrap(); + assert!( + body.contains("status=\"failed\""), + "expected failed status, got: {body}" + ); + + // The name UPDATE that ran earlier in the tx must have been rolled back. + let id_bytes = uuid_bytes(&pl_id).unwrap(); + let name: String = sqlx::query_scalar("SELECT name FROM subsonic_playlists WHERE id = ?") + .bind(&id_bytes) + .fetch_one(&state.db.read) + .await + .unwrap(); + assert_eq!( + name, "Original", + "earlier name UPDATE must roll back when a later statement fails" + ); + } }