ipx/crates/feed/src/fetch.rs

//! Reading a feed over HTTP: conditional requests, redirects followed one by one, bounded bodies.

use crate::{privacy, wordpress};
use anyhow::{Context, Result, anyhow};
use reqwest::StatusCode;
use reqwest::header::HeaderName;
use reqwest::header::{ETAG, IF_MODIFIED_SINCE, IF_NONE_MATCH, LAST_MODIFIED};
use std::sync::atomic::AtomicU64;
use std::time::Duration;

/// What `fetch` got.
pub enum Fetched {
    /// Server said 304, or returned a body we already have.
    NotModified,
    /// A body to read, with the validators to send next time.
    Body { bytes: Vec<u8>, etag: Option<String>, last_modified: Option<String> },
}

/// The longest a feed may take, connecting to the last byte: a scan handles feeds in order, and
/// with no limit one hung server held every scan for as long as it did. Dreamwidth answered 504
/// after 60-67 s for a day, and each scan took 70-80 s instead of 15 (#108). A feed is small;
/// downloads, which are not, have no such limit.
pub const FEED_TIMEOUT: Duration = Duration::from_secs(30);

/// The longest a feed's server may send nothing, for the clients that fetch feeds to set: one
/// that took the connection and never answered ran the whole 30 s, every scan it was in (#230).
/// No feed that answered took over 8 s in a day of traces; one that keeps sending, however
/// slowly, still has until FEED_TIMEOUT.
pub const FEED_SILENCE: Duration = Duration::from_secs(15);

/// Feeds fetched and the bytes of their bodies, since the process started, for /metrics to set
/// beside memory (#224): a body is read whole, so a night of large feeds is a night of large
/// allocations.
pub static FETCHES: AtomicU64 = AtomicU64::new(0);

/// See `FETCHES`.
pub static FETCHED_BYTES: AtomicU64 = AtomicU64::new(0);

/// The most of a feed or page read. Without a limit an episode's MP3 given as a feed was read
/// whole into memory, at every check (#295). Real feeds can be large: a daily show's whole back
/// catalogue was 62 MB, and 50 MB here stopped it updating (#357).
pub(crate) const MAX_BODY: usize = 200 << 20;

/// A response's body, up to `MAX_BODY`: past it, an error, and the rest is never read.
pub(crate) async fn read_capped(mut resp: reqwest::Response) -> Result<Vec<u8>> {
    if resp.content_length().is_some_and(|n| n > MAX_BODY as u64) {
        return Err(anyhow!("larger than {} MB, too large to be a feed", MAX_BODY >> 20));
    }
    let mut out = Vec::new();
    while let Some(c) = resp.chunk().await? {
        out.extend_from_slice(&c);
        anyhow::ensure!(out.len() <= MAX_BODY, "larger than {} MB, too large to be a feed", MAX_BODY >> 20);
    }
    Ok(out)
}

/// Reads a feed, asking only for what changed since `etag` and `last_modified`. Redirects are
/// followed here, up to ten, so `auth` goes to the feed's own host alone. Also returns the new
/// address when every hop said the feed moved for good, for the caller to keep instead.
#[tracing::instrument(skip_all, fields(url = %privacy::redact(feed_url), http.response.body.size))]
pub async fn fetch(
    client: &reqwest::Client,
    feed_url: &str,
    auth: Option<(&str, Option<String>)>,
    etag: Option<&str>,
    last_modified: Option<&str>,
) -> Result<(Fetched, Option<String>)> {
    let start = reqwest::Url::parse(feed_url).context("the feed's address")?;
    let mut url = start.clone();
    let mut permanent = true;
    for _ in 0..10 {
        let mut req = client.get(url.clone()).timeout(FEED_TIMEOUT);
        if let Some(tag) = etag {
            req = req.header(IF_NONE_MATCH, tag);
        }
        if let Some(lm) = last_modified {
            req = req.header(IF_MODIFIED_SINCE, lm);
        }
        // The feed's own host only: a redirect elsewhere must not be handed the password.
        if let Some((user, password)) = &auth
            && url.host_str() == start.host_str()
        {
            req = req.basic_auth(user, password.as_deref());
        }

        let resp = req.send().await.context("connecting")?;
        let status = resp.status();
        if status.is_redirection() && status != StatusCode::NOT_MODIFIED {
            let to = resp
                .headers()
                .get(reqwest::header::LOCATION)
                .and_then(|v| v.to_str().ok())
                .ok_or_else(|| anyhow!("HTTP {status} without a Location to go to"))?;
            url = url.join(to).with_context(|| format!("redirected to {to:?}, which is not an address"))?;
            permanent &= matches!(status, StatusCode::MOVED_PERMANENTLY | StatusCode::PERMANENT_REDIRECT);
            continue;
        }
        let moved = (permanent && url != start).then(|| url.to_string());
        FETCHES.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
        if status == StatusCode::NOT_MODIFIED {
            return Ok((Fetched::NotModified, moved));
        }
        if !status.is_success() {
            // WordPress.com answers some readers' feed requests with a browser check that only
            // JavaScript passes, for hours at a time and whatever the User-Agent; its API has the
            // same posts and no check (#317).
            if status == StatusCode::FORBIDDEN
                && let Some(host) = url.host_str().map(str::to_owned)
                && read_capped(resp).await.is_ok_and(|b| wordpress::is_wpcom_challenge(&b))
            {
                return Ok((wordpress::wpcom_posts(client, &host).await?, moved));
            }
            // The original surfaced 401/407 specially; the code is enough for a UI to switch on.
            return Err(anyhow!("HTTP {status}"));
        }
        let header = |h: HeaderName| resp.headers().get(&h).and_then(|v| v.to_str().ok()).map(str::to_owned);
        let etag = header(ETAG);
        let last_modified = header(LAST_MODIFIED);
        let bytes = read_capped(resp).await.context("reading body")?;
        tracing::Span::current().record("http.response.body.size", bytes.len());
        FETCHED_BYTES.fetch_add(bytes.len() as u64, std::sync::atomic::Ordering::Relaxed);
        return Ok((Fetched::Body { bytes, etag, last_modified }, moved));
    }
    // Worded as reqwest worded it, which failure_kind reads as a redirect loop.
    Err(anyhow!("error following redirect for url ({url}): too many redirects"))
}

#[cfg(test)]
mod tests {

    use crate::failure;

    /// A server answering by path: /old moves for good to /new, /tmp for now, /chain for good
    /// to /tmp, /new is the feed.
    async fn redirecting_server() -> String {
        use tokio::io::{AsyncReadExt, AsyncWriteExt};
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
        let addr = listener.local_addr().unwrap();
        tokio::spawn(async move {
            loop {
                let Ok((mut sock, _)) = listener.accept().await else { return };
                tokio::spawn(async move {
                    let mut buf = [0u8; 2048];
                    let n = sock.read(&mut buf).await.unwrap_or(0);
                    let req = String::from_utf8_lossy(&buf[..n]);
                    let path = req.split_whitespace().nth(1).unwrap_or("/").to_owned();
                    let body = "<?xml version=\"1.0\"?><rss version=\"2.0\"><channel><title>T</title></channel></rss>";
                    let resp = match path.as_str() {
                        "/old" => {
                            "HTTP/1.1 301 Moved Permanently\r\nLocation: /new\r\nContent-Length: 0\r\n\r\n".to_owned()
                        }
                        "/tmp" => "HTTP/1.1 302 Found\r\nLocation: /new\r\nContent-Length: 0\r\n\r\n".to_owned(),
                        "/chain" => {
                            "HTTP/1.1 308 Permanent Redirect\r\nLocation: /tmp\r\nContent-Length: 0\r\n\r\n".to_owned()
                        }
                        // An episode's file given as a feed: its size said, its body never sent.
                        "/huge" => "HTTP/1.1 200 OK\r\nContent-Length: 600000000\r\n\r\n".to_owned(),
                        "/loop" => {
                            "HTTP/1.1 301 Moved Permanently\r\nLocation: /loop\r\nContent-Length: 0\r\n\r\n".to_owned()
                        }
                        _ => format!("HTTP/1.1 200 OK\r\nContent-Length: {}\r\n\r\n{body}", body.len()),
                    };
                    let _ = sock.write_all(resp.as_bytes()).await;
                });
            }
        });
        format!("http://{addr}")
    }

    #[tokio::test]
    async fn a_feed_that_moved_for_good_says_where_and_one_moved_for_now_does_not() {
        let base = redirecting_server().await;
        let client = reqwest::Client::builder().redirect(reqwest::redirect::Policy::none()).build().unwrap();
        let get = |path: &str| {
            let url = format!("{base}{path}");
            let client = client.clone();
            async move { super::fetch(&client, &url, None, None, None).await }
        };
        let (got, moved) = get("/old").await.unwrap();
        assert!(matches!(got, super::Fetched::Body { .. }));
        assert_eq!(moved, Some(format!("{base}/new")));
        assert_eq!(get("/tmp").await.unwrap().1, None); // 302: for now
        let huge = get("/huge").await.err().map(|e| format!("{e:#}")).unwrap_or_default();
        assert!(huge.contains("too large to be a feed"), "{huge}");
        assert_eq!(get("/chain").await.unwrap().1, None); // 308 then 302: not for good
        assert_eq!(get("/new").await.unwrap().1, None); // never moved
        let looped = get("/loop").await.err().unwrap().to_string();
        assert_eq!(failure::failure_kind(&looped).0, "redirect_loop");
    }
}