fix(podcasts): bound the episodes a channel lists and refuse a downgrading redirect

docs/privacy.md

@@ -202,7 +202,10 @@ is checked before it is contacted:
 - the name is resolved first, and every address must be public, by the same
   rules as the portrait fetcher above; the connection is pinned to the address
   that was checked, and a redirect is checked the same way before it is
-  followed;
+  followed — and refused if it would leave https for http, whatever
+  `WAVEFLOW_PODCAST_ALLOW_HTTP` says, so a token sent over https is not
+  handed on in clear by a redirect (an address typed or listed as http is
+  sent in clear from the start, which is what the switch admits);
 - a feed is at most `WAVEFLOW_PODCAST_MAX_FEED_BYTES` (16 MiB) within
   `WAVEFLOW_PODCAST_FEED_TIMEOUT_SECS` (20); an episode at most
   `WAVEFLOW_PODCAST_MAX_EPISODE_BYTES` (1 GiB) within

src/podcast.rs

@@ -14,7 +14,10 @@
 //! - **Redirects are followed by hand, at most [`MAX_REDIRECTS`]**, and each
 //!   hop is put through every rule above before it is contacted. The shared
 //!   client follows none: a `302` to `http://169.254.169.254/` is refused as an
-//!   address, never fetched.
+//!   address, never fetched. **A hop never leaves https for http**, whatever
+//!   `WAVEFLOW_PODCAST_ALLOW_HTTP` says: the switch admits feeds and
+//!   enclosures served in clear, not a secure address handing its request —
+//!   and the token a private feed carries in it — to a clear one.
 //! - **Bodies are bounded while they are read**, and the whole request by a
 //!   deadline.
 //! - **Nothing about a request is logged** but a verdict from a fixed list: a
@@ -146,6 +149,17 @@ async fn resolve(url: &url::Url, policy: AddressPolicy) -> Result<SocketAddr, Fe
     Ok(addresses[0])
 }
 
+/// Where a redirect from `current` to `location` leads, refused when it would
+/// leave https for http. Every other rule is applied to the hop when it is
+/// opened.
+fn next_hop(current: &url::Url, location: &str) -> Result<url::Url, FetchError> {
+    let next = current.join(location).map_err(|_| FetchError::Redirects)?;
+    if current.scheme() == "https" && next.scheme() == "http" {
+        return Err(FetchError::Destination);
+    }
+    Ok(next)
+}
+
 /// Opens `url` under every rule above, following redirects by hand, and
 /// answers the response that carries the body.
 ///
@@ -185,7 +199,7 @@ pub async fn open(url: &url::Url, policy: FetchPolicy) -> Result<reqwest::Respon
                 .get(reqwest::header::LOCATION)
                 .and_then(|value| value.to_str().ok())
                 .ok_or(FetchError::Redirects)?;
-            current = current.join(location).map_err(|_| FetchError::Redirects)?;
+            current = next_hop(&current, location)?;
             continue;
         }
         if !status.is_success() {
@@ -698,6 +712,57 @@ mod tests {
         assert!(feed.episodes.is_empty());
     }
 
+    /// A secure address may move anywhere secure, a clear one anywhere; a
+    /// secure one may not move into the clear. A scheme-relative target
+    /// (`//host/path`) keeps the scheme it came from, so it never downgrades.
+    #[test]
+    fn a_redirect_never_leaves_https_for_http() {
+        let secure = url::Url::parse("https://feeds.example.org/show.xml?auth=t").unwrap();
+        let clear = url::Url::parse("http://feeds.example.org/show.xml").unwrap();
+        assert_eq!(
+            next_hop(&secure, "https://cdn.example.org/show.xml")
+                .unwrap()
+                .as_str(),
+            "https://cdn.example.org/show.xml"
+        );
+        assert_eq!(
+            next_hop(&secure, "/moved.xml").unwrap().as_str(),
+            "https://feeds.example.org/moved.xml"
+        );
+        assert_eq!(
+            next_hop(&secure, "//cdn.example.org/show.xml")
+                .unwrap()
+                .as_str(),
+            "https://cdn.example.org/show.xml"
+        );
+        assert_eq!(
+            next_hop(&clear, "//cdn.example.org/show.xml")
+                .unwrap()
+                .as_str(),
+            "http://cdn.example.org/show.xml"
+        );
+        assert_eq!(
+            next_hop(&clear, "https://cdn.example.org/show.xml")
+                .unwrap()
+                .as_str(),
+            "https://cdn.example.org/show.xml"
+        );
+        assert_eq!(
+            next_hop(&clear, "http://cdn.example.org/show.xml")
+                .unwrap()
+                .as_str(),
+            "http://cdn.example.org/show.xml"
+        );
+        assert_eq!(
+            next_hop(&secure, "http://cdn.example.org/show.xml"),
+            Err(FetchError::Destination)
+        );
+        assert_eq!(
+            next_hop(&secure, "HTTP://cdn.example.org/show.xml"),
+            Err(FetchError::Destination)
+        );
+    }
+
     #[test]
     fn a_feed_address_is_absolute_http_and_carries_no_credentials() {
         let public = AddressPolicy::Public;

src/services/mod.rs

@@ -1627,7 +1627,7 @@ pub use lyrics_uploads::{LyricsFile, MAX_UPLOAD_BYTES as MAX_LYRICS_UPLOAD_BYTES
 pub use mixes::{Mix, MixQuery, MixSummary, DEFAULT_MIX_SIZE, MAX_MIX_SIZE};
 pub use podcasts::{
     EpisodeBookmark, PodcastChannelItem, PodcastEpisodeItem, PodcastFailure, PodcastOverview,
-    NEWEST_EPISODES_DEFAULT,
+    MAX_EPISODES_LISTED, NEWEST_EPISODES_DEFAULT,
 };
 pub use quality::{
     AnalysisState, QualityAnalysis, QualitySummary, QualityTrack, QualityVerdict,

src/services/podcasts.rs

@@ -36,6 +36,15 @@ const NEWEST_EPISODES_MAX: i64 = 500;
 /// How many channels one instance may hold. A refresh fetches every one.
 const MAX_PODCAST_CHANNELS: i64 = 500;
 
+/// How many episodes a channel lists: the newest, as many as one feed may
+/// carry. Episodes are never purged — a refresh adds what a feed newly names
+/// and keeps the rest — so a daily show would otherwise grow every answer by a
+/// year's worth of rows a year, and `getPodcasts` answers every channel with
+/// its episodes by default. An episode past the bound that is stored, or on
+/// its way, is listed all the same: it holds a share of the quota, and the
+/// listing is where an administrator deletes it.
+pub const MAX_EPISODES_LISTED: i64 = podcast::MAX_EPISODES_PER_FEED as i64;
+
 /// A channel, as every surface reads it.
 #[derive(Debug, Clone, Serialize, ToSchema)]
 pub struct PodcastChannelItem {
@@ -69,6 +78,8 @@ pub struct PodcastChannelItem {
     pub error: Option<String>,
     pub created_at: i64,
     pub refreshed_at: Option<i64>,
+    /// Every episode the channel holds, which may be more than `episodes`
+    /// lists: see [`MAX_EPISODES_LISTED`].
     pub episode_count: i64,
     #[serde(skip_serializing_if = "Option::is_none")]
     pub episodes: Option<Vec<PodcastEpisodeItem>>,
@@ -258,21 +269,35 @@ impl DomainServices {
             .map(|row| channel_from_row(row, can_manage))
             .collect::<Result<Vec<_>, _>>()?;
         if with_episodes {
+            // One statement for every channel, ranked within each, rather than
+            // one per channel.
+            let rows = sqlx::query(concat!(
+                episode_columns!(),
+                "JOIN (SELECT id AS ranked_id, ROW_NUMBER() OVER (PARTITION BY channel_id \
+                         ORDER BY published_at DESC NULLS LAST, title COLLATE NOCASE, id) \
+                         AS listed_rank \
+                       FROM podcast_episode \
+                       WHERE status<>'deleted' AND (? IS NULL OR channel_id=?)) r \
+                   ON r.ranked_id=e.id \
+                 WHERE r.listed_rank<=? OR e.status IN ('completed', 'downloading') \
+                 ORDER BY e.channel_id, r.listed_rank"
+            ))
+            .bind(user_id.to_string())
+            .bind(only.as_deref())
+            .bind(only.as_deref())
+            .bind(MAX_EPISODES_LISTED)
+            .fetch_all(&mut *connection)
+            .await?;
+            let mut by_channel: HashMap<Uuid, Vec<PodcastEpisodeItem>> = HashMap::new();
+            for row in rows {
+                let episode = episode_from_row(row)?;
+                by_channel
+                    .entry(episode.channel_id)
+                    .or_default()
+                    .push(episode);
+            }
             for channel in &mut channels {
-                let rows = sqlx::query(concat!(
-                    episode_columns!(),
-                    "WHERE e.channel_id=? AND e.status<>'deleted' \
-                     ORDER BY e.published_at DESC NULLS LAST, e.title COLLATE NOCASE, e.id"
-                ))
-                .bind(user_id.to_string())
-                .bind(channel.id.to_string())
-                .fetch_all(&mut *connection)
-                .await?;
-                channel.episodes = Some(
-                    rows.into_iter()
-                        .map(episode_from_row)
-                        .collect::<Result<Vec<_>, _>>()?,
-                );
+                channel.episodes = Some(by_channel.remove(&channel.id).unwrap_or_default());
             }
         }
         Ok(channels)

tests/podcasts.rs

@@ -889,6 +889,107 @@ async fn the_subsonic_podcast_methods_answer_the_schema_shapes() {
         .is_none_or(|channels| channels.as_array().is_some_and(Vec::is_empty)));
 }
 
+/// Episodes are never purged, so a channel lists only its newest
+/// [`MAX_EPISODES_LISTED`] — and, past them, any it stores, which an
+/// administrator must still be able to reach to delete. Every channel's
+/// episodes are read in one statement and handed to their own channel.
+#[tokio::test]
+async fn a_channel_lists_its_newest_episodes_and_every_stored_one() {
+    use waveflow_server::services::MAX_EPISODES_LISTED;
+
+    let host = fake_host().await;
+    let (_temp, config, state) = podcast_app(|_| {}).await;
+    let (_, admin_password, _) = account(&state, "pod-cap-admin", AccountRole::Admin).await;
+    let (_, _, reader_key) = account(&state, "pod-cap-reader", AccountRole::User).await;
+    let router = waveflow_server::app(&config, state.clone());
+    let admin_token = login_token(&router, "pod-cap-admin", &admin_password).await;
+
+    let (status, long) = subscribe(&router, &admin_token, &format!("{}/feed.xml", host.base)).await;
+    assert_eq!(status, StatusCode::CREATED, "{long}");
+    let long_id = uuid_of(&long);
+    let short_url = format!(
+        "{}/private/p{}/podcast.xml",
+        host.base,
+        Uuid::new_v4().simple()
+    );
+    let (status, short) = subscribe(&router, &admin_token, &short_url).await;
+    assert_eq!(status, StatusCode::CREATED, "{short}");
+    let short_id = uuid_of(&short);
+
+    // Years of a daily show the feed no longer names, all older than the four
+    // it does: the oldest of them stored, the rest never downloaded.
+    let backlog = MAX_EPISODES_LISTED + 10;
+    let now = now_ms();
+    let mut tx = state.db.pool().begin().await.unwrap();
+    for n in 0..backlog {
+        let stored = n == 0;
+        sqlx::query(
+            "INSERT INTO podcast_episode \
+               (id, channel_id, guid, title, published_at, enclosure_url, status, file_name, \
+                created_at, updated_at) \
+             VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
+        )
+        .bind(Uuid::new_v4().to_string())
+        .bind(long_id.to_string())
+        .bind(format!("old-{n}"))
+        .bind(format!("Old {n}"))
+        .bind(1_000_000 + n * 1_000)
+        .bind(format!("{}/audio.mp3", host.base))
+        .bind(if stored { "completed" } else { "new" })
+        .bind(stored.then(|| format!("{}.mp3", Uuid::new_v4())))
+        .bind(now)
+        .bind(now)
+        .execute(&mut *tx)
+        .await
+        .unwrap();
+    }
+    tx.commit().await.unwrap();
+    let listed = usize::try_from(MAX_EPISODES_LISTED).unwrap();
+
+    let read = json_body(
+        send_json(
+            &router,
+            "GET",
+            &format!("/api/v2/podcasts/{long_id}"),
+            &admin_token,
+            None,
+        )
+        .await,
+    )
+    .await;
+    assert_eq!(read["episode_count"], 4 + backlog, "the count is every one");
+    let episodes = read["episodes"].as_array().unwrap();
+    assert_eq!(episodes.len(), listed + 1, "the newest, and the stored one");
+    assert_eq!(episodes[0]["title"], "Through a redirect");
+    assert_eq!(
+        episodes[listed - 1]["title"],
+        format!("Old {}", backlog - (MAX_EPISODES_LISTED - 4)),
+        "the oldest of the newest"
+    );
+    assert_eq!(episodes[listed]["title"], "Old 0");
+    assert_eq!(episodes[listed]["status"], "completed");
+
+    // Every channel at once, as `getPodcasts` answers by default: each keeps
+    // its own episodes, however they were read.
+    let podcasts = subsonic_json(&router, "getPodcasts", &reader_key, "").await;
+    let channels = podcasts["subsonic-response"]["podcasts"]["channel"]
+        .as_array()
+        .unwrap();
+    assert_eq!(channels.len(), 2);
+    for (id, count) in [(long_id, listed + 1), (short_id, 4)] {
+        let channel = channels
+            .iter()
+            .find(|channel| channel["id"] == id.to_string())
+            .unwrap();
+        let episodes = channel["episode"].as_array().unwrap();
+        assert_eq!(episodes.len(), count, "{id}");
+        assert!(episodes
+            .iter()
+            .all(|episode| episode["channelId"] == id.to_string()));
+        assert_eq!(episodes[0]["title"], "Through a redirect");
+    }
+}
+
 /// Decision 8: a private feed's token — in its path, in its query and on every
 /// enclosure — reaches the account that may manage podcasts and nobody else,
 /// on any surface. A reader is shown the feed's scheme and host in the

webapp/e2e/smart-playlists-and-podcasts.spec.ts

@@ -577,6 +577,10 @@ test("subscribes to a podcast, downloads, plays and keeps the place", async ({
   await page.goto("/podcasts");
   await expect(page.getByRole("heading", { name: "Podcasts" })).toBeVisible();
   await expect(page.getByText("No podcasts yet.")).toBeVisible();
+  // Whoever subscribes is told, at the field, that the show is not theirs alone.
+  await expect(page.getByLabel("Feed address")).toHaveAccessibleDescription(
+    /^Visible to every account/,
+  );
   await page.getByLabel("Feed address").fill("https://feeds.example.org/night.xml");
   await page.getByRole("button", { name: "Subscribe" }).click();
   expect(podcastWrites).toEqual(["subscribe https://feeds.example.org/night.xml"]);

webapp/src/pages/podcasts-page.tsx

@@ -112,23 +112,31 @@ export function PodcastsPage() {
       </PageHeader>
       {overview.enabled ? null : <p className="muted">{t("podcasts.off")}</p>}
       {overview.can_manage ? (
-        <form
-          className="inline-form"
-          onSubmit={(event) => void subscribe(event)}
-        >
-          <input
-            type="url"
-            value={url}
-            onChange={(event) => setUrl(event.target.value)}
-            placeholder="https://example.org/feed.xml"
-            aria-label={t("podcasts.url")}
-            required
-            disabled={busy}
-          />
-          <button type="submit" disabled={busy || !url.trim()}>
-            {t("podcasts.subscribe")}
-          </button>
-        </form>
+        <>
+          <form
+            className="inline-form"
+            onSubmit={(event) => void subscribe(event)}
+          >
+            <input
+              type="url"
+              value={url}
+              onChange={(event) => setUrl(event.target.value)}
+              placeholder="https://example.org/feed.xml"
+              aria-label={t("podcasts.url")}
+              aria-describedby="podcast-shared"
+              required
+              disabled={busy}
+            />
+            <button type="submit" disabled={busy || !url.trim()}>
+              {t("podcasts.subscribe")}
+            </button>
+          </form>
+          {/* One list per instance: a private feed's episodes reach every
+              account once it is added, though its address does not. */}
+          <p id="podcast-shared" className="muted">
+            {t("podcasts.shared")}
+          </p>
+        </>
       ) : null}
       {message ? (
         <p className="muted" role="status">

webapp/src/state/i18n.tsx

@@ -1110,6 +1110,8 @@ const en = {
   "podcasts.off":
     "Podcasts are switched off on this server. What was kept before stays readable; nothing new is fetched.",
   "podcasts.url": "Feed address",
+  "podcasts.shared":
+    "Visible to every account: anyone on this server sees the show and can play the episodes it downloads. Only administrators see the feed address.",
   "podcasts.subscribe": "Subscribe",
   "podcasts.refresh": "Refresh every feed",
   "podcasts.refreshing": "The feeds are being fetched again.",
@@ -2266,6 +2268,8 @@ const fr: Record<TranslationKey, TranslationValue> = {
   "podcasts.off":
     "Les podcasts sont désactivés sur ce serveur. Ce qui a été conservé reste lisible ; rien de nouveau n’est récupéré.",
   "podcasts.url": "Adresse du flux",
+  "podcasts.shared":
+    "Visible par tous les comptes : chacun sur ce serveur voit l’émission et peut écouter les épisodes téléchargés. Seuls les administrateurs voient l’adresse du flux.",
   "podcasts.subscribe": "S’abonner",
   "podcasts.refresh": "Actualiser tous les flux",
   "podcasts.refreshing": "Les flux sont en cours de récupération.",