-
+
{t.account.providerUntieNeedsPassword}
+ )} + ) : ( <> -{t.account.providerLead}
-+ {me.hasPassword ? t.account.providerLead : t.account.providerLeadNoPassword} +
+- {(rotate.error ?? revoke.error) instanceof ApiError ? (rotate.error ?? revoke.error)?.message : null} + {(rotate.error ?? revoke.error ?? toggle.error) instanceof ApiError + ? (rotate.error ?? revoke.error ?? toggle.error)?.message + : null}
) : null} diff --git a/migrations/postgres/0053_season_order_tvdb_index.sql b/migrations/postgres/0053_season_order_tvdb_index.sql new file mode 100644 index 0000000..bf00d27 --- /dev/null +++ b/migrations/postgres/0053_season_order_tvdb_index.sql @@ -0,0 +1,6 @@ +-- The TheTVDB relay finds the work that holds a season, or an episode an +-- order keeps, by TheTVDB's id of it, as it finds an episode (0026): a number +-- nobody holds read every season and every episode of every order, on each +-- request. Partial, as there: what is made by hand has no such id. +CREATE INDEX ix_media_season_tvdb ON media_season (tvdb_id) WHERE tvdb_id IS NOT NULL; +CREATE INDEX ix_order_episode_tvdb ON media_order_episode (tvdb_episode_id) WHERE tvdb_episode_id IS NOT NULL; diff --git a/migrations/sqlite/0053_season_order_tvdb_index.sql b/migrations/sqlite/0053_season_order_tvdb_index.sql new file mode 100644 index 0000000..bf00d27 --- /dev/null +++ b/migrations/sqlite/0053_season_order_tvdb_index.sql @@ -0,0 +1,6 @@ +-- The TheTVDB relay finds the work that holds a season, or an episode an +-- order keeps, by TheTVDB's id of it, as it finds an episode (0026): a number +-- nobody holds read every season and every episode of every order, on each +-- request. Partial, as there: what is made by hand has no such id. +CREATE INDEX ix_media_season_tvdb ON media_season (tvdb_id) WHERE tvdb_id IS NOT NULL; +CREATE INDEX ix_order_episode_tvdb ON media_order_episode (tvdb_episode_id) WHERE tvdb_episode_id IS NOT NULL; diff --git a/scripts/watch/common.py b/scripts/watch/common.py index ecdfb47..6df1c11 100644 --- a/scripts/watch/common.py +++ b/scripts/watch/common.py @@ -101,6 +101,7 @@ def _closed(url, data, headers, method, timeout, limit): """One request on a connection of its own, as urllib makes them.""" request = urllib.request.Request(url, data=data, headers=headers, method=method) try: + # nosemgrep: python.lang.security.audit.dynamic-urllib-use-detected.dynamic-urllib-use-detected — the URLs are the checks' own https constants with urllib.request.urlopen(request, timeout=timeout) as response: body = response.read(limit) if limit else response.read() return response.status, response.headers, _decoded(body, response.headers, limit) @@ -145,8 +146,10 @@ def _kept(url, data, headers, method, timeout, limit): def _connect(host, port, timeout): proxy = urllib.request.getproxies().get("https") if not proxy or urllib.request.proxy_bypass(host): + # nosemgrep: python.lang.security.audit.httpsconnection-detected.httpsconnection-detected — http.client verifies the certificate and the host name return http.client.HTTPSConnection(host, port, timeout=timeout) via = urllib.parse.urlsplit(proxy if "://" in proxy else f"http://{proxy}") + # nosemgrep: python.lang.security.audit.httpsconnection-detected.httpsconnection-detected — http.client verifies the certificate and the host name connection = http.client.HTTPSConnection(via.hostname, via.port or 80, timeout=timeout) tunnel = {} if via.username: diff --git a/src/api/anilist.rs b/src/api/anilist.rs index 638dd77..d0e09dd 100644 --- a/src/api/anilist.rs +++ b/src/api/anilist.rs @@ -86,6 +86,12 @@ const COLLECTIONS: &[&str] = &["GenreCollection", "MediaTagCollection"]; /// What a relayed request keeps of the caller's headers: what describes it. const ASKED_WITH: &[&str] = &["user-agent", "accept", "accept-language", "content-type"]; +/// Of those, the headers that may change AniList's answer: what a kept answer +/// is told apart by besides the query — the type a body is posted as says how +/// AniList reads it — as it travels asked with them, an answer filed without +/// them would be handed to a caller that asked otherwise. +const VARIES_BY: &[&str] = &["accept", "accept-language", "content-type"]; + /// How many entries of one answer are looked up here, in the order the /// answer lists them. A page is fifty at most; a user's whole list, in one /// answer, is not patched beyond this. @@ -258,7 +264,7 @@ pub async fn relay(State(state): State> =
LazyLock::new(|| Arc::new(tokio::sync::Semaphore::new(AT_ONCE)));
@@ -411,34 +437,156 @@ static DECODING: LazyLock> =
static AVIF: LazyLock> =
LazyLock::new(|| Arc::new(tokio::sync::Semaphore::new(1)));
-/// Run `work` — a decoding, a resizing, an encoding — off the server's
-/// threads, as many at once as [`AT_ONCE`] allows, and a decoder that
-/// panics on the bytes it was given as a picture refused rather than a
-/// thread brought down.
+/// What the pictures being decoded may hold in memory together
+/// ([`file::DECODE_BUDGET`]), however many the gate lets through and
+/// whatever they are: each takes the share its picture's worst case needs
+/// ([`file::decode_with`] counts it from the header) before a byte is
+/// allocated for it, and gives it back when what it decoded is dropped.
+/// Taken in turn, so that a large picture is not kept waiting for ever by a
+/// string of small ones; the thread that waits is the one the work runs on
+/// (see [`Turn::run`]), of which [`AT_ONCE`] and the night's one are all
+/// there are.
+struct Budget {
+ limit: u64,
+ shares: Mutex,
+ changed: Condvar,
+}
+
+#[derive(Default)]
+struct Shares {
+ /// What is held.
+ used: u64,
+ /// The next place in line to hand out, and the one whose turn it is.
+ next: u64,
+ serving: u64,
+}
+
+static BUDGET: LazyLock> = LazyLock::new(|| Arc::new(Budget::new(file::DECODE_BUDGET)));
+
+impl Budget {
+ fn new(limit: u64) -> Self {
+ Self {
+ limit,
+ shares: Mutex::new(Shares::default()),
+ changed: Condvar::new(),
+ }
+ }
+
+ /// A share of `bytes`, waited for until it fits beside those held and
+ /// every share asked for before it has been given. One larger than the
+ /// whole is the whole: the pictures that need more are refused before
+ /// they ask.
+ fn take(self: &Arc, bytes: u64) -> Held {
+ let bytes = bytes.min(self.limit);
+ let mut shares = self.shares.lock().unwrap_or_else(PoisonError::into_inner);
+ let place = shares.next;
+ shares.next += 1;
+ while shares.serving != place || shares.used + bytes > self.limit {
+ shares = self
+ .changed
+ .wait(shares)
+ .unwrap_or_else(PoisonError::into_inner);
+ }
+ shares.used += bytes;
+ shares.serving += 1;
+ // The one behind may fit beside this.
+ self.changed.notify_all();
+ Held {
+ budget: self.clone(),
+ bytes,
+ }
+ }
+}
+
+/// A share of the [`Budget`], given back when it is dropped.
+pub struct Held {
+ budget: Arc,
+ bytes: u64,
+}
+
+impl Drop for Held {
+ fn drop(&mut self) {
+ let mut shares = self
+ .budget
+ .shares
+ .lock()
+ .unwrap_or_else(PoisonError::into_inner);
+ shares.used = shares.used.saturating_sub(self.bytes);
+ self.budget.changed.notify_all();
+ }
+}
+
+/// A place among the pictures decoded at once ([`AT_ONCE`]), held until the
+/// work it was waited for is done. Taken before the picture is read from
+/// the store, so that no more pictures sit in memory waiting to be decoded
+/// than are being decoded.
+pub struct Turn(tokio::sync::OwnedSemaphorePermit);
+
+/// Wait for a place among the pictures decoded at once.
+pub async fn turn() -> Result {
+ let permit = DECODING
+ .clone()
+ .acquire_owned()
+ .await
+ .context("the pictures' gate is closed")?;
+ Ok(Turn(permit))
+}
+
+/// Every place at once, for a test that fills the gate: held by one test at a
+/// time, since two of them, each holding one of the two places, would wait
+/// for each other for ever.
+#[cfg(test)]
+pub async fn every_turn() -> (tokio::sync::MutexGuard<'static, ()>, [Turn; AT_ONCE]) {
+ static FILLING: tokio::sync::Mutex<()> = tokio::sync::Mutex::const_new(());
+ let filling = FILLING.lock().await;
+ let mut turns = Vec::new();
+ for _ in 0..AT_ONCE {
+ turns.push(turn().await.expect("a place"));
+ }
+ let turns = turns.try_into().ok().expect("as many places as there are");
+ (filling, turns)
+}
+
+impl Turn {
+ /// Run `work` — a decoding, a resizing, an encoding — off the server's
+ /// threads, holding this place until it is done, and a decoder that
+ /// panics on the bytes it was given as a picture refused rather than a
+ /// thread brought down.
+ pub async fn run(
+ self,
+ work: impl FnOnce() -> Result + Send + 'static,
+ ) -> Result {
+ run_holding(self.0, work).await
+ }
+}
+
+/// Run `work` off the server's threads, as many at once as [`AT_ONCE`]
+/// allows, and a decoder that panics on the bytes it was given as a picture
+/// refused rather than a thread brought down.
pub async fn off_thread(
work: impl FnOnce() -> Result + Send + 'static,
) -> Result {
- run_under(&DECODING, work).await
+ turn().await?.run(work).await
}
/// The same, for an AVIF copy: one at a time.
pub async fn off_thread_avif(
work: impl FnOnce() -> Result + Send + 'static,
) -> Result {
- run_under(&AVIF, work).await
+ let permit = AVIF
+ .clone()
+ .acquire_owned()
+ .await
+ .context("the pictures' gate is closed")?;
+ run_holding(permit, work).await
}
-async fn run_under(
- gate: &Arc,
+async fn run_holding(
+ permit: tokio::sync::OwnedSemaphorePermit,
work: impl FnOnce() -> Result + Send + 'static,
) -> Result {
// Held by the work itself, so that a request given up on meanwhile does
// not let another start beside it.
- let permit = gate
- .clone()
- .acquire_owned()
- .await
- .context("the pictures' gate is closed")?;
let joined = tokio::task::spawn_blocking(move || {
let _permit = permit;
std::panic::catch_unwind(AssertUnwindSafe(work))
@@ -506,6 +654,13 @@ mod tests {
"no width the interface never asks for"
);
assert_eq!(width_made(10_000, 20_000), None);
+ // Its own width is a copy only where it is no wider than the widest:
+ // past that it would be a request for the whole of a picture of any
+ // size, drawn and written again.
+ assert_eq!(width_made(1920, 1920), Some(1920));
+ assert_eq!(width_made(2000, 2000), None);
+ assert_eq!(width_made(9_000, 9_000), None);
+ assert_eq!(width_made(1920, 3000), Some(1920));
}
#[test]
@@ -570,12 +725,49 @@ mod tests {
assert_eq!(back.get_pixel(20, 70)[3], 0, "transparent where it was");
assert_eq!(back.get_pixel(380, 70)[3], 255, "opaque where it was");
- // An alpha channel opaque everywhere is no transparency.
+ // An alpha channel opaque everywhere is no transparency, and the
+ // colours are what they were, closed up where they lay.
let mut opaque = Vec::new();
- image::RgbaImage::from_pixel(50, 50, image::Rgba([1, 2, 3, 255]))
- .write_to(&mut Cursor::new(&mut opaque), image::ImageFormat::Png)
- .unwrap();
- assert!(!Decoded::read(&opaque).unwrap().alpha);
+ image::RgbaImage::from_fn(50, 40, |x, y| {
+ image::Rgba([(x * 5) as u8, (y * 6) as u8, 3, 255])
+ })
+ .write_to(&mut Cursor::new(&mut opaque), image::ImageFormat::Png)
+ .unwrap();
+ let decoded = Decoded::read(&opaque).unwrap();
+ assert!(!decoded.alpha);
+ let pixels = decoded.as_pixels();
+ assert_eq!((pixels.data.len(), pixels.channels()), (50 * 40 * 3, 3));
+ assert_eq!(&pixels.data[(7 * 50 + 9) * 3..][..3], [45, 42, 3]);
+ assert_eq!(&pixels.data[(39 * 50 + 49) * 3..][..3], [245, 234, 3]);
+ }
+
+ /// RGBA made RGB where it lies keeps every pixel's colours, in order.
+ #[test]
+ fn the_alpha_is_dropped_where_the_pixels_lie() {
+ let mut pixels = vec![1, 2, 3, 255, 4, 5, 6, 255, 7, 8, 9, 255];
+ drop_alpha(&mut pixels);
+ assert_eq!(pixels, [1, 2, 3, 4, 5, 6, 7, 8, 9]);
+ let mut none: Vec = Vec::new();
+ drop_alpha(&mut none);
+ assert!(none.is_empty());
+ }
+
+ /// A copy at the picture's own size draws the picture itself: no second
+ /// buffer of its pixels beside the first; any other size is resized.
+ #[test]
+ fn a_copy_at_the_pictures_own_size_copies_none_of_its_pixels() {
+ let decoded = Decoded::read(&png(300, 200)).unwrap();
+ let mut resized = None;
+ let same = drawn_at(&decoded, 300, 200, &mut resized).unwrap();
+ assert!(std::ptr::eq(same, &decoded));
+ assert!(resized.is_none());
+ let smaller = drawn_at(&decoded, 150, 100, &mut resized).unwrap();
+ assert!(!std::ptr::eq(smaller, &decoded));
+ assert_eq!((smaller.width, smaller.height), (150, 100));
+ // And the copy made is the same as before it was spared.
+ let copy = make(&decoded, Made::Width(300), Format::Webp, None).unwrap();
+ assert_eq!((copy.width, copy.height), (300, 200));
+ assert_eq!(size_of(©.bytes), (300, 200));
}
#[test]
@@ -637,6 +829,110 @@ mod tests {
assert!(prepare(b"not a picture", Some(CoverType::Poster), false).is_err());
}
+ /// Pictures decoding together hold no more than the budget: one that
+ /// does not fit waits for those that do to be done, and those behind it
+ /// wait their turn, though they would fit — so that a large picture is
+ /// not kept waiting by a string of small ones. A share is given back
+ /// when it is dropped.
+ #[test]
+ fn pictures_decoding_together_hold_no_more_than_the_budget_and_wait_in_turn() {
+ use std::{sync::mpsc, thread, time::Duration};
+
+ let budget = Arc::new(Budget::new(100));
+ let first = budget.take(60);
+ let held = |budget: &Budget| budget.shares.lock().unwrap().used;
+ assert_eq!(held(&budget), 60);
+
+ let (told, heard) = mpsc::channel();
+ let ask = |name: &'static str, bytes: u64| {
+ let (budget, told) = (budget.clone(), told.clone());
+ thread::spawn(move || {
+ let share = budget.take(bytes);
+ told.send(name).unwrap();
+ // Held a moment, as a picture is while it is decoded.
+ thread::sleep(Duration::from_millis(50));
+ drop(share);
+ })
+ };
+ // 60 does not fit beside 60; 30 would, and is behind it.
+ let large = ask("large", 60);
+ thread::sleep(Duration::from_millis(100));
+ let small = ask("small", 30);
+ assert!(
+ heard.recv_timeout(Duration::from_millis(300)).is_err(),
+ "nobody goes while the first holds its share, and the small one is in turn"
+ );
+ assert_eq!(held(&budget), 60);
+
+ drop(first);
+ let mut told_in_the_end = vec![
+ heard.recv_timeout(Duration::from_secs(5)).unwrap(),
+ heard.recv_timeout(Duration::from_secs(5)).unwrap(),
+ ];
+ told_in_the_end.sort_unstable();
+ assert_eq!(told_in_the_end, ["large", "small"]);
+ large.join().unwrap();
+ small.join().unwrap();
+ assert_eq!(held(&budget), 0, "every share is given back");
+
+ // A share larger than the whole is the whole, and never waits for
+ // more than there is.
+ let all = budget.take(500);
+ assert_eq!(all.bytes, 100);
+ assert_eq!(held(&budget), 100);
+ drop(all);
+ assert_eq!(held(&budget), 0);
+ }
+
+ /// What a picture reserves is its worst case from the header, taken
+ /// before it is decoded, held for as long as the picture is and given
+ /// back with it.
+ #[test]
+ fn what_a_picture_reserves_is_given_back_once_it_is_dropped() {
+ let budget = Arc::new(Budget::new(file::DECODE_BUDGET));
+ let used = || budget.shares.lock().unwrap().used;
+ let bytes = png(64, 48);
+ let (picture, held) =
+ file::decode_with(&bytes, true, |weight| budget.take(weight)).unwrap();
+ assert_eq!((picture.width(), picture.height()), (64, 48));
+ // 3 072 pixels of RGB, 9 bytes each with the copy to be made of
+ // them, the file's own bytes and the fixed ones.
+ assert_eq!(
+ held.bytes,
+ 64 * 48 * 9 + bytes.len() as u64 + 8 * 1024 * 1024
+ );
+ assert_eq!(used(), held.bytes);
+ drop(held);
+ assert_eq!(used(), 0);
+
+ // A picture read for the copies carries its share, a copy of it
+ // made smaller none, being within it.
+ let decoded = Decoded::read(&bytes).unwrap();
+ assert!(decoded._held.is_some());
+ assert!(decoded.resized(32, 24).unwrap()._held.is_none());
+ }
+
+ #[tokio::test]
+ async fn a_turn_is_waited_for_and_held_by_the_work() {
+ // Every place taken: another waits.
+ let (filling, [one, two]) = every_turn().await;
+ let mut third = Box::pin(turn());
+ assert!(
+ tokio::time::timeout(std::time::Duration::from_millis(100), &mut third)
+ .await
+ .is_err(),
+ "no more than {AT_ONCE} are decoded at once"
+ );
+ // The work holds its place until it is done, whoever stops waiting.
+ let done = one.run(|| Ok(1)).await.unwrap();
+ assert_eq!(done, 1);
+ let third = tokio::time::timeout(std::time::Duration::from_secs(5), third)
+ .await
+ .expect("a place freed by the work")
+ .unwrap();
+ drop((two, third, filling));
+ }
+
#[tokio::test]
async fn work_that_panics_is_a_picture_refused() {
let refused = off_thread(|| -> Result<()> { panic!("a decoder's bug") }).await;
diff --git a/src/memory.rs b/src/memory.rs
index 05cfef6..719773b 100644
--- a/src/memory.rs
+++ b/src/memory.rs
@@ -63,6 +63,7 @@ fn count(delta: isize) {
unsafe impl GlobalAlloc for Counting {
#[inline]
unsafe fn alloc(&self, layout: Layout) -> *mut u8 {
+ // nosemgrep: rust.lang.security.unsafe-usage.unsafe-usage — a GlobalAlloc has to call its allocator's unsafe methods, held by the SAFETY note
let at = unsafe { mimalloc::MiMalloc.alloc(layout) };
if !at.is_null() {
count(layout.size() as isize);
@@ -72,6 +73,7 @@ unsafe impl GlobalAlloc for Counting {
#[inline]
unsafe fn alloc_zeroed(&self, layout: Layout) -> *mut u8 {
+ // nosemgrep: rust.lang.security.unsafe-usage.unsafe-usage — a GlobalAlloc has to call its allocator's unsafe methods, held by the SAFETY note
let at = unsafe { mimalloc::MiMalloc.alloc_zeroed(layout) };
if !at.is_null() {
count(layout.size() as isize);
@@ -81,12 +83,14 @@ unsafe impl GlobalAlloc for Counting {
#[inline]
unsafe fn dealloc(&self, at: *mut u8, layout: Layout) {
+ // nosemgrep: rust.lang.security.unsafe-usage.unsafe-usage — a GlobalAlloc has to call its allocator's unsafe methods, held by the SAFETY note
unsafe { mimalloc::MiMalloc.dealloc(at, layout) };
count(-(layout.size() as isize));
}
#[inline]
unsafe fn realloc(&self, at: *mut u8, layout: Layout, new_size: usize) -> *mut u8 {
+ // nosemgrep: rust.lang.security.unsafe-usage.unsafe-usage — a GlobalAlloc has to call its allocator's unsafe methods, held by the SAFETY note
let moved = unsafe { mimalloc::MiMalloc.realloc(at, layout, new_size) };
// A failed realloc leaves the old block where it was, still held.
if !moved.is_null() {
diff --git a/src/outbound.rs b/src/outbound.rs
index edfc77d..007b9ab 100644
--- a/src/outbound.rs
+++ b/src/outbound.rs
@@ -397,6 +397,9 @@ fn guarded_builder(guard: Guard) -> reqwest::ClientBuilder {
.timeout(std::time::Duration::from_secs(60))
.connect_timeout(std::time::Duration::from_secs(10))
.redirect(guard.redirect_policy())
+ // A redirect is followed without telling the next host where it
+ // came from: the address left may carry a key in its query.
+ .referer(false)
}
#[cfg(test)]
diff --git a/src/providers/tmdb/mod.rs b/src/providers/tmdb/mod.rs
index 6f8a608..08de957 100644
--- a/src/providers/tmdb/mod.rs
+++ b/src/providers/tmdb/mod.rs
@@ -163,7 +163,13 @@ impl TmdbClient {
external_source: &str,
external_id: &str,
) -> Result {
- let url = format!("{}/find/{external_id}", self.base);
+ // One segment of the path, whatever the id holds: it is read from a
+ // request, a list, a feed, and is no address of anybody's choosing.
+ let url = format!(
+ "{}/find/{}",
+ self.base,
+ crate::providers::segment(external_id)
+ );
let params = [
("external_source", external_source.to_string()),
("language", self.language()),
@@ -754,6 +760,57 @@ mod tests {
assert_eq!(image_languages("ja"), "ja,en,null");
}
+ /// What an id written into the path of `/find` is asked as: the path
+ /// the stand-in was sent, as it was written on the wire.
+ async fn asked_to_find(id: &str) -> Vec {
+ use std::sync::{Arc, Mutex};
+
+ use axum::{Router, extract::Request};
+
+ let seen = Arc::new(Mutex::new(Vec::new()));
+ let kept = Arc::clone(&seen);
+ let app = Router::new().fallback(move |request: Request| {
+ let kept = Arc::clone(&kept);
+ async move {
+ kept.lock().unwrap().push(request.uri().path().to_string());
+ axum::Json(serde_json::json!({ "movie_results": [], "tv_results": [] }))
+ }
+ });
+ let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
+ .await
+ .expect("a port");
+ let at = listener.local_addr().expect("its address");
+ tokio::spawn(async move {
+ let _ = axum::serve(listener, app).await;
+ });
+
+ let client = TmdbClient::new(
+ reqwest::Client::new(),
+ &config::Tmdb {
+ api_key: Some("test".into()),
+ upstream: format!("http://{at}"),
+ language: "en-US".into(),
+ include_adult: false,
+ search_limit: 5,
+ passthrough: true,
+ },
+ );
+ client.find("imdb_id", id).await.expect("answered");
+ seen.lock().unwrap().clone()
+ }
+
+ #[tokio::test]
+ async fn an_id_asked_for_is_one_segment_of_the_path_whatever_it_holds() {
+ assert_eq!(asked_to_find("tt0903747").await, ["/3/find/tt0903747"]);
+ assert_eq!(asked_to_find("81189").await, ["/3/find/81189"]);
+ // Not a way to another path of TMDB's — nor to a query of its own.
+ assert_eq!(
+ asked_to_find("../account/1?session_id=s#x").await,
+ ["/3/find/..%2Faccount%2F1%3Fsession_id%3Ds%23x"]
+ );
+ assert_eq!(asked_to_find("tt1/../../3/account").await.len(), 1);
+ }
+
#[test]
fn backfill_ignores_episodes_absent_from_the_fallback() {
let mut target = season(vec![(1, 9, Some(""), None)]);
diff --git a/src/search/rank.rs b/src/search/rank.rs
index 7f132e9..8308d98 100644
--- a/src/search/rank.rs
+++ b/src/search/rank.rs
@@ -44,9 +44,26 @@ impl Form {
/// that it can only be a number in the title.
const YEARS: std::ops::RangeInclusive = 1870..=2100;
+/// The most characters a term is looked for by: what the list and the search
+/// of the native API take, and what the searches TMDB's, TheTVDB's, Plex's
+/// and Sonarr's clients make are given too, however long the term they send.
+/// A term longer is read by its first characters, folded or not.
+pub const TERM_LENGTH: usize = 200;
+
+/// `text`'s first [`TERM_LENGTH`] characters.
+fn clip(text: &str) -> &str {
+ match text.char_indices().nth(TERM_LENGTH) {
+ Some((at, _)) => &text[..at],
+ None => text,
+ }
+}
+
impl Term {
pub fn parse(text: &str) -> Self {
- let whole = norm(text);
+ // Cut before it is folded, which a text of megabytes is not to cost,
+ // and after, since a character may fold into a dozen.
+ let folded = norm(clip(text));
+ let whole = clip(&folded).trim_end().to_string();
let dated = whole.rsplit_once(' ').and_then(|(before, last)| {
let year: i32 = last.parse().ok().filter(|y| YEARS.contains(y))?;
(last.len() == 4 && !before.is_empty()).then(|| (Form::of(before.to_string()), year))
@@ -282,6 +299,38 @@ mod tests {
assert!(sequel > first, "{sequel} against {first}");
}
+ /// Whoever sends a term — the interface, TMDB's search, TheTVDB's, Plex,
+ /// Sonarr — it is read by its first two hundred characters, and a text
+ /// of megabytes costs what a long term does.
+ #[test]
+ fn a_term_is_read_by_its_first_two_hundred_characters() {
+ // Under the cap nothing changes.
+ assert_eq!(Term::parse("Dune 2021").whole.norm, "dune 2021");
+ let exact = "a".repeat(TERM_LENGTH);
+ assert_eq!(Term::parse(&exact).whole.norm, exact);
+
+ // One word and a year past the cap: the year is cut with the rest,
+ // and no title is read before it.
+ let long = format!("{exact} 2021");
+ let term = Term::parse(&long);
+ assert_eq!(term.whole.norm, exact);
+ assert!(term.dated.is_none());
+
+ // The cut never leaves a space at the end.
+ let spaced = format!("{} b", "a".repeat(TERM_LENGTH - 1));
+ assert_eq!(Term::parse(&spaced).whole.norm, "a".repeat(TERM_LENGTH - 1));
+
+ // A character that folds into many is cut after it is folded: the
+ // Arabic ligature of a whole invocation is eighteen of them.
+ let ligatures = "\u{FDFA}".repeat(TERM_LENGTH);
+ assert!(Term::parse(&ligatures).whole.norm.chars().count() <= TERM_LENGTH);
+
+ let huge = "ab ".repeat(1_000_000);
+ let term = Term::parse(&huge);
+ assert!(term.whole.norm.chars().count() <= TERM_LENGTH);
+ assert!(!term.is_empty());
+ }
+
#[test]
fn a_term_of_punctuation_alone_is_nothing_to_look_for() {
assert!(Term::parse("%").is_empty());
diff --git a/src/service/ids.rs b/src/service/ids.rs
index dfed97c..73989e3 100644
--- a/src/service/ids.rs
+++ b/src/service/ids.rs
@@ -308,6 +308,14 @@ pub fn own_episode(names: &str) -> i64 {
-(i64::from(hash % (i32::MAX as u32)) + 1)
}
+/// Whether `id` is a number [`own_episode`] can make: negative, and no
+/// further from nought than the 32-bit column a client keeps it in. Anything
+/// else — a positive number, nought, `-9999999999` — was never made here, so
+/// nothing is looked for under it.
+pub fn is_own_episode(id: i64) -> bool {
+ (-i64::from(i32::MAX)..0).contains(&id)
+}
+
/// A search term of the form `prefix:value`, which Sonarr and Radarr both use
/// to look a title up by a specific provider's id.
#[derive(Debug, PartialEq, Eq)]
@@ -650,6 +658,31 @@ mod tests {
assert!(own_episode("") < 0);
}
+ #[test]
+ fn only_a_number_an_episode_can_be_given_is_one_made_here() {
+ for names in [
+ "tmdb:170135",
+ "place:abc",
+ "season:tmdb:1",
+ "character:x",
+ "",
+ ] {
+ assert!(is_own_episode(own_episode(names)), "{names}");
+ }
+ // The ends of what it makes, and one past each.
+ assert!(is_own_episode(-1) && is_own_episode(-i64::from(i32::MAX)));
+ for never in [
+ 0,
+ 1,
+ 990_701,
+ -i64::from(i32::MAX) - 1,
+ -9_999_999_999,
+ i64::MIN,
+ ] {
+ assert!(!is_own_episode(never), "{never}");
+ }
+ }
+
#[test]
fn prefixed_terms_are_recognised() {
assert_eq!(classify("tvdb:81189"), TermLookup::Tvdb(81189));
diff --git a/src/service/series.rs b/src/service/series.rs
index de9e836..fe83fc8 100644
--- a/src/service/series.rs
+++ b/src/service/series.rs
@@ -214,6 +214,13 @@ pub async fn by_tmdb_id(state: &AppState, tmdb_id: i64) -> Result Result> {
+ // As a film's: an id that is none is no series', and never goes into a
+ // provider's address — it is written there, a path of TMDB's — nor into
+ // what is looked for.
+ let Some(imdb_id) = crate::domain::ids::normalize_imdb_id(imdb_id) else {
+ return Ok(None);
+ };
+ let imdb_id = imdb_id.as_str();
if let Some(item) = local(state, ExternalSource::Imdb, imdb_id).await?
&& item.kind == MediaKind::Series
{
@@ -1031,6 +1038,31 @@ mod tests {
item
}
+ /// An IMDb id comes from a request's body, a query, a list somebody
+ /// else keeps: one that is none is no series, and asks nobody.
+ #[tokio::test]
+ async fn an_imdb_id_that_is_none_is_no_series_and_goes_into_no_address() {
+ let (state, providers) = crate::service::testing::server().await;
+ for none in [
+ "../account/1?x=1",
+ "tt12345?language=fr",
+ "tt12345/../../3/account/tt",
+ "not an id",
+ "ttabc",
+ "tt",
+ "",
+ ] {
+ let found = by_imdb_id(&state, none).await.expect("answered");
+ assert!(found.is_none(), "{none:?}");
+ }
+ assert_eq!(providers.asked(), 0, "none of them was asked of anybody");
+
+ // What makes it a test: one that is an id, spelt as TMDB's find
+ // takes it, is asked for.
+ let _ = by_imdb_id(&state, "9907240").await;
+ assert!(providers.asked() > 0);
+ }
+
#[tokio::test]
async fn sonarrs_imdb_and_tmdb_terms_find_a_series_held_under_another_id_of_its_item() {
use crate::service::bridge::{Site, keep_for_tests};
diff --git a/src/service/webhook.rs b/src/service/webhook.rs
index b1b1ec1..2fcf4a8 100644
--- a/src/service/webhook.rs
+++ b/src/service/webhook.rs
@@ -229,6 +229,9 @@ impl Hooks {
// Webhooks asks, rather than followed: a 3xx is a failure, and
// the log says where to look.
.redirect(reqwest::redirect::Policy::none())
+ // Nothing is followed, so no address is ever named to another;
+ // said all the same, as it is for every client this server makes.
+ .referer(false)
.timeout(TIMEOUT)
.connect_timeout(CONNECT_TIMEOUT)
.build()
diff --git a/src/state.rs b/src/state.rs
index 7a79c89..e2a3d0f 100644
--- a/src/state.rs
+++ b/src/state.rs
@@ -175,6 +175,9 @@ pub struct Inner {
accounts: tokio::sync::Mutex<()>,
/// What each API answered since the start.
pub calls: Calls,
+ /// The numbers the TheTVDB relay made for what TheTVDB gave none, led back
+ /// to what they name: read once while the catalogue stays as it was.
+ pub made_ids: crate::api::thetvdb::MadeIds,
/// The media kept: the store, and the index of what is in it.
pub media: crate::media::Media,
/// The doors' certificates: the clients' listener and what it shows.
@@ -231,11 +234,13 @@ impl AppState {
let upstreams = std::sync::Arc::new(configured_hosts(&config));
let http = outbound_client(&upstreams, Redirects::Anywhere)?;
- // MyAnimeList's client id travels in a header of its own, which a
- // redirect to another origin — another host, but another port or
- // scheme of the same one too — would carry along: reqwest strips only
- // the ones it knows for credentials.
- let mal_http = outbound_client(&upstreams, Redirects::SameOrigin)?;
+ // The providers a credential travels to: TMDB's and Fanart's key in
+ // the address, TheTVDB's in the body of its sign-in, which a 307
+ // sends on again, MyAnimeList's client id in a header of its own,
+ // which reqwest does not know for a credential and keeps on a
+ // redirect to another port or scheme of the same host too. Each is
+ // followed only where it was first asked.
+ let keyed_http = outbound_client(&upstreams, Redirects::SameOrigin)?;
let slot = crate::cache::RedisSlot::default();
let caches = Caches::new(&config.cache, slot.clone());
@@ -270,20 +275,20 @@ impl AppState {
crate::config::Mode::Single => axum_extra::extract::cookie::Key::generate(),
crate::config::Mode::Multi => shared_cookie_key(&db).await?,
};
- let tmdb = TmdbClient::new(http.clone(), &config.tmdb);
+ let tmdb = TmdbClient::new(keyed_http.clone(), &config.tmdb);
let skyhook = SkyhookClient::new(http.clone(), &config.skyhook, instance.clone());
let radarr_metadata =
RadarrMetadataClient::new(http.clone(), &config.radarr_metadata, instance.clone());
- let fanart = FanartClient::new(http.clone(), &config.fanart, &config.tmdb.language);
+ let fanart = FanartClient::new(keyed_http.clone(), &config.fanart, &config.tmdb.language);
let tvdb = TvdbClient::new(
- http.clone(),
+ keyed_http.clone(),
&config.tvdb,
&config.tmdb.language,
instance.clone(),
);
let tvmaze = TvmazeClient::new(http.clone(), &config.tvmaze);
let anilist = AnilistClient::new(http.clone(), &config.anilist, instance.clone());
- let mal = MalClient::new(mal_http, &config.mal);
+ let mal = MalClient::new(keyed_http, &config.mal);
let fankai = FankaiClient::new(http.clone(), &config.fankai, &config.tmdb.language);
let fankai_wiki = FankaiWikiClient::new(http.clone(), &config.fankai_wiki);
@@ -346,6 +351,7 @@ impl AppState {
instance,
accounts: tokio::sync::Mutex::new(()),
calls: Calls::default(),
+ made_ids: crate::api::thetvdb::MadeIds::default(),
cookie_key,
tls,
coord,
@@ -1112,8 +1118,8 @@ enum Redirects {
/// To any host the guard does not refuse.
Anywhere,
/// Only at the origin first asked — its scheme, host and port: a
- /// credential in a header of the provider's own goes nowhere else, and
- /// not to another service of the same host.
+ /// credential of the provider's own goes nowhere else, and not to
+ /// another service of the same host.
SameOrigin,
}
@@ -1124,6 +1130,26 @@ fn leaves_origin(first: Option<&url::Url>, next: &url::Url) -> bool {
first.is_none_or(|first| !crate::outbound::same_origin(first.as_str(), next.as_str()))
}
+/// What every client of the providers is made from: this server's name, how
+/// long it waits, and no `Referer`. reqwest writes the address a redirect
+/// came from into one, query and all, and for a provider asked with its key
+/// in the address — TMDB's, Fanart's — that address carries the key to
+/// whoever the redirect names.
+fn provider_builder() -> reqwest::ClientBuilder {
+ reqwest::Client::builder()
+ .user_agent(concat!(
+ "arr-metadata-server/",
+ env!("CARGO_PKG_VERSION"),
+ " (+",
+ env!("CARGO_PKG_REPOSITORY"),
+ ")"
+ ))
+ .timeout(std::time::Duration::from_secs(30))
+ .connect_timeout(std::time::Duration::from_secs(10))
+ .pool_max_idle_per_host(16)
+ .referer(false)
+}
+
/// The client every provider, dataset, relay to TMDB and webhook is asked
/// with.
///
@@ -1138,17 +1164,7 @@ fn outbound_client(
redirects: Redirects,
) -> Result {
let guard = crate::outbound::Guard::default();
- reqwest::Client::builder()
- .user_agent(concat!(
- "arr-metadata-server/",
- env!("CARGO_PKG_VERSION"),
- " (+",
- env!("CARGO_PKG_REPOSITORY"),
- ")"
- ))
- .timeout(std::time::Duration::from_secs(30))
- .connect_timeout(std::time::Duration::from_secs(10))
- .pool_max_idle_per_host(16)
+ provider_builder()
.dns_resolver(std::sync::Arc::new(UpstreamResolver {
configured: std::sync::Arc::clone(upstreams),
}))
@@ -1358,6 +1374,110 @@ mod redirect_tests {
// With no first request on record, nothing is known to stay.
assert!(leaves_origin(None, &first));
}
+
+ /// A stand-in on the loopback: `/start` is sent on to `/next`, which says
+ /// the `Referer` it came with; `/away` is sent on to `elsewhere`.
+ async fn stand_in(elsewhere: &str) -> std::net::SocketAddr {
+ use axum::{Router, http::HeaderMap, response::Redirect, routing::get};
+
+ let away = elsewhere.to_string();
+ let app = Router::new()
+ .route("/start", get(|| async { Redirect::temporary("/next") }))
+ .route(
+ "/next",
+ get(|headers: HeaderMap| async move {
+ headers
+ .get("referer")
+ .and_then(|value| value.to_str().ok())
+ .unwrap_or("none")
+ .to_string()
+ }),
+ )
+ .route(
+ "/away",
+ get(move || async move { Redirect::temporary(&away) }),
+ );
+ let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
+ .await
+ .expect("a port");
+ let at = listener.local_addr().expect("its address");
+ tokio::spawn(async move {
+ let _ = axum::serve(listener, app).await;
+ });
+ at
+ }
+
+ #[tokio::test]
+ async fn a_redirect_is_followed_without_telling_where_it_came_from() {
+ let at = stand_in("/").await;
+ let asked = format!("http://{at}/start?api_key=s3cret");
+ let said = |builder: reqwest::ClientBuilder| {
+ let asked = asked.clone();
+ async move {
+ builder
+ .redirect(reqwest::redirect::Policy::limited(2))
+ .build()
+ .unwrap()
+ .get(asked)
+ .send()
+ .await
+ .expect("answered")
+ .text()
+ .await
+ .expect("read")
+ }
+ };
+ assert_eq!(said(super::provider_builder()).await, "none");
+ // What makes it a test: a client as reqwest builds one says the
+ // address it was sent to, key and all.
+ assert!(said(reqwest::Client::builder()).await.contains("s3cret"));
+ }
+
+ /// The server's own clients, as they are built: TMDB's, which carries its
+ /// key in the address, is handed a redirect to another origin as it came;
+ /// the client of a provider with no credential is judged by the guard
+ /// alone, which refuses a hop to the loopback with an error.
+ #[tokio::test]
+ async fn the_client_of_tmdb_follows_no_redirect_off_its_origin() {
+ use crate::api::relay::testing::{StandIn, server};
+
+ let elsewhere = stand_in("/").await;
+ let at = stand_in(&format!("http://{elsewhere}/next")).await;
+ let upstream =
+ StandIn::start(|_| (axum::http::StatusCode::OK, serde_json::json!({}))).await;
+ let state = server(&upstream).await;
+ let asked = format!("http://{at}/away?api_key=s3cret");
+
+ let answer = state
+ .tmdb
+ .request(reqwest::Method::GET, &asked)
+ .send()
+ .await
+ .expect("answered");
+ assert_eq!(answer.status(), reqwest::StatusCode::TEMPORARY_REDIRECT);
+ assert!(state.http.get(&asked).send().await.is_err());
+ }
+
+ #[tokio::test]
+ async fn a_provider_with_a_credential_is_not_followed_to_another_origin() {
+ let elsewhere = stand_in("/").await;
+ let at = stand_in(&format!("http://{elsewhere}/next")).await;
+ let upstreams = std::sync::Arc::new(std::collections::HashSet::new());
+ let keyed = super::outbound_client(&upstreams, super::Redirects::SameOrigin).unwrap();
+
+ // Another port of the same host is another service: the redirect is
+ // handed back as it came, and nothing is sent to it.
+ let answer = keyed
+ .get(format!("http://{at}/away?api_key=s3cret"))
+ .send()
+ .await
+ .expect("answered");
+ assert_eq!(answer.status(), reqwest::StatusCode::TEMPORARY_REDIRECT);
+ assert_eq!(
+ answer.headers()["location"],
+ format!("http://{elsewhere}/next")
+ );
+ }
}
#[cfg(test)]
diff --git a/src/telemetry.rs b/src/telemetry.rs
index 21a10d9..fb81865 100644
--- a/src/telemetry.rs
+++ b/src/telemetry.rs
@@ -397,7 +397,19 @@ pub fn redact(text: &str) -> String {
const MASK: &str = "***";
-/// The parameters whose values are secrets: keys, tokens, passwords.
+/// The parameters whose values are secrets: keys, tokens, passwords — and
+/// what an address carries a credential in besides, each by the name it
+/// is found under somewhere here: `auth`, where a list another server keeps
+/// private is read with a key (an import rule's address); `code`, `state`
+/// and `session_state`, what the identity provider sends a sign-in back with
+/// — a `code` is the one-time right to the person's tokens, and a `state`
+/// the proof the sign-in is the one begun; `session_id`, `guest_session_id`
+/// and `request_token`, the sessions TMDB's clients open, any of which acts
+/// as the account for as long as it lasts.
+///
+/// Every text that is logged, journaled, reported or traced is masked by
+/// this list: the request's span ([`redact_uri`]), the lines the journal
+/// keeps (`journal::masked`) and a server span's `url.query` ([`redact`]).
const SECRET_PARAMETERS: &[&str] = &[
"api_key",
"apikey",
@@ -415,6 +427,13 @@ const SECRET_PARAMETERS: &[&str] = &[
"signature",
"x-amz-signature",
"x-amz-credential",
+ "auth",
+ "code",
+ "state",
+ "session_state",
+ "session_id",
+ "guest_session_id",
+ "request_token",
];
/// `api_key=abc&language=fr` is `api_key=***&language=fr`, wherever such a
@@ -637,6 +656,39 @@ mod tests {
assert_eq!(mask_parameters("api_key="), "api_key=");
}
+ #[test]
+ fn what_a_sign_in_a_session_or_a_list_carries_in_an_address_is_masked() {
+ // The identity provider's way back to the callback: the code is
+ // the right to the person's tokens, the state is the sign-in's own.
+ assert_eq!(
+ redact(
+ "GET /api/v1/auth/oidc/callback?code=4/0AbCdEf&state=Xy12Zq&session_state=ab.cd&iss=https://idp.example"
+ ),
+ "GET /api/v1/auth/oidc/callback?code=***&state=***&session_state=***&iss=https://idp.example"
+ );
+ // TMDB's sessions, in any case and wherever the pair stands.
+ assert_eq!(
+ mask_parameters(
+ "language=fr&Session_ID=s3ss10n&guest_session_id=gu3st&request_token=r3qt0k&page=2"
+ ),
+ "language=fr&Session_ID=***&guest_session_id=***&request_token=***&page=2"
+ );
+ // An import rule's address, with the key its list is read with.
+ assert_eq!(
+ redact(
+ "error sending request for url (https://other.example/lists/top/ids.txt?kind=movie&auth=t0pS3cret)"
+ ),
+ "error sending request for url (https://other.example/lists/top/ids.txt?kind=movie&auth=***)"
+ );
+ // A name that only contains one is left alone, and a value that is
+ // none stays none.
+ assert_eq!(
+ mask_parameters("barcode=1&mistate=2&authors=3&postcode=4"),
+ "barcode=1&mistate=2&authors=3&postcode=4"
+ );
+ assert_eq!(mask_parameters("code=&state="), "code=&state=");
+ }
+
#[test]
fn credentials_after_their_scheme_are_masked() {
assert_eq!(
@@ -702,6 +754,26 @@ mod tests {
assert_eq!(redact_uri(&bare), "/health");
}
+ #[test]
+ fn the_request_span_never_holds_what_a_sign_in_or_a_session_carries() {
+ let callback: axum::http::Uri =
+ "/api/v1/auth/oidc/callback?code=authcode&state=csrf&session_state=sess"
+ .parse()
+ .unwrap();
+ assert_eq!(
+ redact_uri(&callback),
+ "/api/v1/auth/oidc/callback?code=***&state=***&session_state=***"
+ );
+ let tmdb: axum::http::Uri =
+ "/3/account/1/lists?session_id=s&guest_session_id=g&request_token=t&language=fr"
+ .parse()
+ .unwrap();
+ assert_eq!(
+ redact_uri(&tmdb),
+ "/3/account/1/lists?session_id=***&guest_session_id=***&request_token=***&language=fr"
+ );
+ }
+
#[test]
fn an_id_from_outside_is_taken_in_one_shape_only() {
let ok = |text: &str| acceptable_id(&HeaderValue::from_str(text).unwrap());
diff --git a/src/telemetry/journal.rs b/src/telemetry/journal.rs
index 601e91e..073a68b 100644
--- a/src/telemetry/journal.rs
+++ b/src/telemetry/journal.rs
@@ -503,6 +503,41 @@ mod tests {
assert!(text.contains("api_key=***"), "{text}");
}
+ #[test]
+ fn what_a_sign_in_or_a_session_carries_in_an_address_is_not_kept() {
+ let subscriber = tracing_subscriber::registry().with(layer());
+ tracing::subscriber::with_default(subscriber, || {
+ tracing::warn!(
+ callback = %"/api/v1/auth/oidc/callback?code=authcode123456&state=csrfstate7890&session_state=sessstate4321",
+ tmdb = %"/3/account/1/lists?session_id=tmdbsession5555&guest_session_id=guestsess6666&request_token=reqtoken7777",
+ list = %"https://other.example/ids.txt?auth=listkey8888&kind=movie",
+ "signin-line-marker"
+ );
+ });
+ let lines = kept("signin-line-marker");
+ let text = lines.first().expect("the line").as_text();
+ for planted in [
+ "authcode123456",
+ "csrfstate7890",
+ "sessstate4321",
+ "tmdbsession5555",
+ "guestsess6666",
+ "reqtoken7777",
+ "listkey8888",
+ ] {
+ assert!(!text.contains(planted), "{planted} in {text}");
+ }
+ for masked in [
+ "code=***",
+ "state=***",
+ "session_id=***",
+ "auth=***",
+ "kind=movie",
+ ] {
+ assert!(text.contains(masked), "{masked} in {text}");
+ }
+ }
+
#[test]
fn the_journal_is_read_by_level_by_words_and_after_a_line() {
let subscriber = tracing_subscriber::registry().with(layer());
diff --git a/src/telemetry/trace.rs b/src/telemetry/trace.rs
index 7ae00a6..ebd7e6f 100644
--- a/src/telemetry/trace.rs
+++ b/src/telemetry/trace.rs
@@ -744,6 +744,64 @@ pub(super) mod tests {
ON_HERE.with(|on| on.set(false));
}
+ /// The query a server span records is masked by the list the log is:
+ /// the key of a client, what a sign-in's address is sent back with, what
+ /// a TMDB session carries — and what is not a secret stays as it was.
+ #[tokio::test(flavor = "current_thread")]
+ async fn the_query_a_server_span_records_is_masked_as_the_log_masks_it() {
+ ON_HERE.with(|on| on.set(true));
+ let kept = Kept::default();
+ let provider_spans = SdkTracerProvider::builder()
+ .with_simple_exporter(kept.clone())
+ .with_id_generator(Chosen::default())
+ .build();
+ let subscriber = tracing_subscriber::registry().with(spans(provider_spans.tracer("test")));
+ let _default = tracing::subscriber::set_default(subscriber);
+
+ let (upstream, _) = provider().await;
+ let query = "code=authcode123456&state=csrfstate7890&session_state=sessstate4321\
+ &session_id=tmdbsession5555&guest_session_id=guestsess6666\
+ &request_token=reqtoken7777&auth=listkey8888&api_key=clientkey9999&language=fr";
+ let mut request = HttpRequest::get(format!("/v1/tvdb/shows/en/81189?{query}"))
+ .body(Body::empty())
+ .unwrap();
+ request.extensions_mut().insert(ConnectInfo::(
+ "203.0.113.5:4000".parse().unwrap(),
+ ));
+ let response = door(upstream, vec![]).oneshot(request).await.unwrap();
+ assert_eq!(response.status(), 200);
+ // The span ends with the answer's body, as the other test reads it.
+ let body = axum::body::to_bytes(response.into_body(), 1024)
+ .await
+ .unwrap();
+ assert_eq!(&body[..], b"answered");
+
+ let spans = std::mem::take(&mut *kept.0.lock().unwrap());
+ let server = spans
+ .iter()
+ .find(|s| s.span_kind == SpanKind::Server)
+ .expect("the server span");
+ let recorded = attribute(server, "url.query")
+ .expect("the query")
+ .to_string();
+ for planted in [
+ "authcode123456",
+ "csrfstate7890",
+ "sessstate4321",
+ "tmdbsession5555",
+ "guestsess6666",
+ "reqtoken7777",
+ "listkey8888",
+ "clientkey9999",
+ ] {
+ assert!(!recorded.contains(planted), "{planted} in {recorded}");
+ }
+ assert!(recorded.contains("language=fr"), "{recorded}");
+ assert!(recorded.contains("code=***&state=***"), "{recorded}");
+
+ ON_HERE.with(|on| on.set(false));
+ }
+
/// A call made outside any request — a background task's — is sent as
/// it is: no span, no `traceparent`.
#[tokio::test(flavor = "current_thread")]
diff --git a/src/ui.rs b/src/ui.rs
index 9bee92b..e3d7119 100644
--- a/src/ui.rs
+++ b/src/ui.rs
@@ -13,6 +13,11 @@
//! credential would be let in at all. A work's page carries what a search
//! engine reads of it too, as schema.org describes a film or a series — only
//! where the site is public, the one case a search engine is meant to read it.
+//!
+//! Finding what a preview says is the database's work, which anybody may ask
+//! for as often as they like: it is kept a few seconds, and what is not kept
+//! is counted against the caller's allowance as the API's requests are, a
+//! caller who has spent it getting the page without ([`shell`]).
use axum::{
Router,
@@ -84,13 +89,69 @@ async fn serve(State(state): State, request: Request) -> Response {
let Some(file) = Assets::get("index.html") else {
return not_found();
};
- let mut html = String::from_utf8_lossy(&file.data).into_owned();
- if let Some(preview) = preview(&state, path).await {
- html = preview.apply(&html, state.config.server.public_url.as_deref(), path);
- } else if let Some(public) = state.config.server.public_url.as_deref() {
- html = absolute_site_image(&html, public);
+ let html = String::from_utf8_lossy(&file.data).into_owned();
+ let peer = crate::auth::ratelimit::peer_of(&state, &request);
+ page(shell(&state, path, peer, &html).await)
+}
+
+/// The longest path a rendered preview is kept under: a path that may have
+/// one is a few dozen characters, and what follows a work's id is a stranger's
+/// to write.
+const KEPT_PATH: usize = 256;
+
+/// The page for a route: the app's own, with the preview of what the route
+/// shows in its head where there is one.
+///
+/// Looking a preview up is the database's work — a work, a list, a genre's
+/// count — which anybody may ask for as often as they like, one address after
+/// another. So a preview rendered lately is kept a few seconds, under the path
+/// and the catalogue's mark; one that is not kept is looked up on the
+/// caller's account, as the API's requests are counted
+/// ([`crate::auth::ratelimit::spend`]); and a caller whose allowance is spent
+/// is given the page without one, as a path that has none is. `peer` is the
+/// caller, as the limiter counts callers ([`crate::auth::ratelimit::peer_of`]).
+async fn shell(state: &AppState, path: &str, peer: std::net::IpAddr, html: &str) -> String {
+ let public = state.config.server.public_url.as_deref();
+ let plain = || match public {
+ Some(public) => absolute_site_image(html, public),
+ None => html.to_string(),
+ };
+
+ let Some(subject) = subject_of(path) else {
+ return plain();
+ };
+ if !open_to_readers(state) {
+ return plain();
+ }
+ let Ok(mark) = state.caches.catalogue_stamp(&state.db).await else {
+ return plain();
+ };
+ let key = format!("{}\n{mark}\n{path}", state.caches.stamp());
+ let kept = path.len() <= KEPT_PATH;
+ if kept && let Some(page) = state.caches.previews.get(&key).await {
+ return page.to_string();
+ }
+
+ if crate::auth::ratelimit::spend(state, peer, path)
+ .await
+ .is_err()
+ {
+ return plain();
}
- page(html)
+ let Some(preview) = preview(state, subject).await else {
+ return plain();
+ };
+ let page = preview.apply(html, public, path);
+ // Only what was found is kept: a path that names nothing is looked up
+ // again, on its caller's account, and is never the key of anything.
+ if kept {
+ state
+ .caches
+ .previews
+ .insert(key, std::sync::Arc::from(page.as_str()))
+ .await;
+ }
+ page
}
fn asset(path: &str) -> Option {
@@ -241,32 +302,48 @@ fn open_to_readers(state: &AppState) -> bool {
}
}
-/// The preview a path deserves: a work's, a list's, a page's, or none. The
-/// shape of the id or slug is checked before anything is looked up, so a
-/// stranger walking random addresses is answered from the page alone.
-async fn preview(state: &AppState, path: &str) -> Option {
- if !open_to_readers(state) {
- return None;
- }
+/// What a path's preview would be of, once the shape of its id or slug says
+/// it can have one at all.
+#[derive(Debug, PartialEq, Eq)]
+enum Subject<'a> {
+ /// A work, by its id.
+ Work(&'a str),
+ /// A curated list, by its slug or its id.
+ List(&'a str),
+ /// A genre's, a network's, a year's or a decade's page, and the key it is
+ /// asked for by, as written.
+ Page(&'a str, String),
+ /// A company's or a keyword's page, by this server's id for it.
+ Entity(&'a str, &'a str),
+ /// A character, by this server's id for it.
+ Character(&'a str),
+ /// A collection, by TMDB's id for it.
+ Collection(i64),
+}
+
+/// What a path asks a preview of, or nothing: the shape of the id or slug is
+/// checked before anything is looked up, so a stranger walking random
+/// addresses is answered from the page alone.
+fn subject_of(path: &str) -> Option> {
let mut parts = path.trim_start_matches('/').split('/');
match (parts.next()?, parts.next()?) {
- ("work", id) if uuid::Uuid::parse_str(id).is_ok() => work_preview(state, id).await,
- ("lists", key) if is_key(key) && parts.next().is_none() => list_preview(state, key).await,
+ ("work", id) if uuid::Uuid::parse_str(id).is_ok() => Some(Subject::Work(id)),
+ ("lists", key) if is_key(key) && parts.next().is_none() => Some(Subject::List(key)),
(page @ ("genre" | "network" | "year" | "decade"), key)
if !key.is_empty() && key.len() <= 128 && parts.next().is_none() =>
{
let key = urlencoding::decode(key).ok()?;
- page_preview(state, page, &key).await
+ Some(Subject::Page(page, key.into_owned()))
}
// A company's, a keyword's and a character's pages, by this
// server's id for them.
(page @ ("company" | "keyword"), id)
if uuid::Uuid::parse_str(id).is_ok() && parts.next().is_none() =>
{
- entity_preview(state, page, id).await
+ Some(Subject::Entity(page, id))
}
("characters", id) if uuid::Uuid::parse_str(id).is_ok() && parts.next().is_none() => {
- character_preview(state, id).await
+ Some(Subject::Character(id))
}
// A collection's, by TMDB's id for it.
("collections", id)
@@ -275,12 +352,25 @@ async fn preview(state: &AppState, path: &str) -> Option {
&& id.bytes().all(|b| b.is_ascii_digit())
&& parts.next().is_none() =>
{
- collection_preview(state, id.parse().ok()?).await
+ Some(Subject::Collection(id.parse().ok()?))
}
_ => None,
}
}
+/// The preview of what a path asks one of: a work's, a list's, a page's, or
+/// none where the catalogue holds nothing by that name.
+async fn preview(state: &AppState, subject: Subject<'_>) -> Option {
+ match subject {
+ Subject::Work(id) => work_preview(state, id).await,
+ Subject::List(key) => list_preview(state, key).await,
+ Subject::Page(page, key) => page_preview(state, page, &key).await,
+ Subject::Entity(page, id) => entity_preview(state, page, id).await,
+ Subject::Character(id) => character_preview(state, id).await,
+ Subject::Collection(tmdb_id) => collection_preview(state, tmdb_id).await,
+ }
+}
+
/// A list's slug or id: what `make_slug` produces, or a UUID.
fn is_key(key: &str) -> bool {
!key.is_empty()
@@ -1258,4 +1348,217 @@ mod tests {
assert!(!is_key("a/b"));
assert!(!is_key(&"a".repeat(129)));
}
+
+ #[test]
+ fn only_a_path_that_can_have_a_preview_is_asked_for_one() {
+ let id = "01a0d073-6ce8-760c-a2cb-113d4cf78503";
+ assert_eq!(subject_of(&format!("/work/{id}")), Some(Subject::Work(id)));
+ // What follows a work's id changes nothing of whose page it is.
+ assert_eq!(
+ subject_of(&format!("/work/{id}/seasons/2")),
+ Some(Subject::Work(id))
+ );
+ assert_eq!(
+ subject_of("/lists/soirees-d-automne"),
+ Some(Subject::List("soirees-d-automne"))
+ );
+ assert_eq!(
+ subject_of("/genre/Dr%C3%A2me"),
+ Some(Subject::Page("genre", "Drâme".into()))
+ );
+ assert_eq!(
+ subject_of("/network/hbo"),
+ Some(Subject::Page("network", "hbo".into()))
+ );
+ assert_eq!(
+ subject_of(&format!("/company/{id}")),
+ Some(Subject::Entity("company", id))
+ );
+ assert_eq!(
+ subject_of(&format!("/keyword/{id}")),
+ Some(Subject::Entity("keyword", id))
+ );
+ assert_eq!(
+ subject_of(&format!("/characters/{id}")),
+ Some(Subject::Character(id))
+ );
+ assert_eq!(subject_of("/collections/10"), Some(Subject::Collection(10)));
+
+ let long = format!("/genre/{}", "a".repeat(129));
+ for path in [
+ "/",
+ "/catalogue",
+ "/work",
+ "/work/not-an-id",
+ "/lists/Soirées",
+ "/lists/a/b",
+ "/genre/",
+ long.as_str(),
+ "/company/nobody",
+ "/characters/1",
+ "/collections/abc",
+ "/collections/1234567890123456789",
+ ] {
+ assert_eq!(subject_of(path), None, "{path}");
+ }
+ }
+
+ /// A server whose pages are offered a link preview whoever reads them,
+ /// and that lets an address make `per_minute` requests a minute.
+ async fn open_server(per_minute: u32) -> AppState {
+ let mut config = crate::config::Config::from_env().expect("a configuration");
+ config.mode = crate::config::Mode::Single;
+ config.database = crate::db::testing::database().await;
+ config.security.bootstrap_admin = None;
+ config.security.native_policy = SurfacePolicy::Open;
+ config.security.rate_limit_per_minute = per_minute;
+ config.cache.redis_url = None;
+ config.media.storage = crate::config::MediaStorage::Off;
+ config.clients = None;
+ config.tmdb.api_key = None;
+ config.tmdb.include_adult = false;
+ config.tvdb.api_key = None;
+ config.tvdb.enabled = false;
+ AppState::bootstrap(config).await.expect("a server")
+ }
+
+ /// Two dramas in the catalogue: the drama page's preview says so.
+ async fn two_dramas(state: &AppState) -> Vec {
+ let mut works = Vec::new();
+ for title in ["One", "Two"] {
+ let mut work = MediaItem::empty(MediaKind::Movie);
+ work.title = title.into();
+ work.slug = crate::domain::make_slug(title, None);
+ work.genres = vec!["Drama".to_string()];
+ repo::item::upsert(
+ &state.db,
+ repo::item::ItemWrite {
+ item: &work,
+ replace_children: true,
+ },
+ )
+ .await
+ .unwrap();
+ works.push(work);
+ }
+ crate::service::listing::relist_stale(&state.db, false, 100)
+ .await
+ .unwrap();
+ works
+ }
+
+ /// The page for a path, as a caller at one address is given it.
+ async fn previewed(state: &AppState, path: &str) -> String {
+ shell(state, path, std::net::IpAddr::from([203, 0, 113, 9]), PAGE).await
+ }
+
+ const TWO: &str = "2 works in the catalogue.";
+
+ /// What a link preview costs the database is asked of whoever asks: kept
+ /// a few seconds, counted against the caller's allowance when it is not
+ /// kept, and not looked up at all once the allowance is spent.
+ #[tokio::test]
+ async fn a_preview_is_looked_up_on_the_callers_account_and_kept_a_few_seconds() {
+ let state = open_server(2).await;
+ two_dramas(&state).await;
+
+ // A page that can have no preview is answered as it always was, and
+ // costs the caller nothing.
+ for _ in 0..5 {
+ let page = previewed(&state, "/catalogue").await;
+ assert!(page.contains("A catalogue."), "{page}");
+ }
+
+ // The first look-up is counted; the same path asked again is answered
+ // from what was kept, however often.
+ for _ in 0..5 {
+ let page = previewed(&state, "/genre/drama").await;
+ assert!(page.contains(TWO), "{page}");
+ }
+ // Another spelling of it is another path: a look-up, the second of
+ // the two the allowance holds.
+ let page = previewed(&state, "/genre/Drama").await;
+ assert!(page.contains(TWO), "{page}");
+
+ // Spent: a path that is not kept is given the page without one…
+ let spent = previewed(&state, "/genre/DRAMA").await;
+ assert!(!spent.contains(TWO), "{spent}");
+ assert!(spent.contains("A catalogue."), "{spent}");
+ // … and what is kept is still served.
+ let page = previewed(&state, "/genre/drama").await;
+ assert!(page.contains(TWO), "{page}");
+ }
+
+ /// A page is kept under the catalogue's mark: a write made here, which
+ /// moves it, is seen at once; a note on the genre, which does not, within
+ /// the seconds a page is kept.
+ #[tokio::test]
+ async fn a_kept_preview_is_looked_up_again_once_the_catalogue_moves() {
+ let state = open_server(600).await;
+ two_dramas(&state).await;
+
+ let first = previewed(&state, "/genre/drama").await;
+ assert!(first.contains(TWO), "{first}");
+
+ repo::editorial::put(
+ &state.db,
+ Scope::Genre,
+ "drama",
+ "en",
+ "The *quiet* ones.",
+ None,
+ )
+ .await
+ .unwrap();
+ let kept = previewed(&state, "/genre/drama").await;
+ assert_eq!(kept, first, "kept: nothing was asked again");
+
+ state.caches.bump_epoch().await;
+ let fresh = previewed(&state, "/genre/drama").await;
+ assert!(fresh.contains("The quiet ones."), "{fresh}");
+ assert!(!fresh.contains(TWO), "{fresh}");
+ }
+
+ /// A path that names nothing is looked up each time, on its caller's
+ /// account, and is never the key of anything kept.
+ #[tokio::test]
+ async fn a_path_that_names_nothing_is_never_kept() {
+ let state = open_server(600).await;
+ let works = two_dramas(&state).await;
+
+ for path in [
+ "/genre/isekai",
+ "/network/nobody",
+ "/lists/no-such-list",
+ "/year/nineteen",
+ "/work/01a0d073-6ce8-760c-a2cb-113d4cf78503",
+ ] {
+ let page = previewed(&state, path).await;
+ assert!(page.contains("A catalogue."), "{path}: {page}");
+ }
+ assert_eq!(state.caches.previews.entries().await, 0);
+
+ // A work found is kept; the same work asked for under a path too long
+ // to be a page's is found, and is not kept under it.
+ let id = &works[0].id;
+ let page = previewed(&state, &format!("/work/{id}")).await;
+ assert!(page.contains("One · Cinémathèque"), "{page}");
+ assert_eq!(state.caches.previews.entries().await, 1);
+ let long = format!("/work/{id}/{}", "x".repeat(KEPT_PATH));
+ let page = previewed(&state, &long).await;
+ assert!(page.contains("One · Cinémathèque"), "{page}");
+ assert_eq!(state.caches.previews.entries().await, 1);
+ }
+
+ /// Where no reader is let in without a credential, there is no preview:
+ /// nothing is looked up, and nothing is counted.
+ #[tokio::test]
+ async fn no_preview_is_looked_up_where_a_reader_is_not_let_in() {
+ let state = crate::api::native::testing::server().await;
+ two_dramas(&state).await;
+ assert!(!open_to_readers(&state));
+ let page = previewed(&state, "/genre/drama").await;
+ assert!(!page.contains(TWO), "{page}");
+ assert_eq!(state.caches.previews.entries().await, 0);
+ }
}
diff --git a/src/web.rs b/src/web.rs
index 36ca984..eeec052 100644
--- a/src/web.rs
+++ b/src/web.rs
@@ -38,10 +38,13 @@ const MAX_BODY_BYTES: usize = 1024 * 1024;
pub async fn serve(state: AppState) -> Result<()> {
let bind = state.config.server.bind;
- let limits = listen::Limits {
- connections: state.config.server.max_connections,
- head: state.config.server.header_read_timeout,
- };
+ let limits = listen::Limits::new(
+ state.config.server.max_connections,
+ state.config.server.header_read_timeout,
+ );
+ // How many connections each network holds at once, but for the proxies
+ // in front of the server, which hold those of everybody behind them.
+ let crowd = || listen::Crowd::new(&state.config.server.trusted_proxies);
// The loops that run for as long as the process does, kept: one that
// panics or returns when it should not is said so and started again.
@@ -138,7 +141,7 @@ pub async fn serve(state: AppState) -> Result<()> {
"listening for the clients (TLS)"
);
let served = listen::serve(
- listen::Door::new(listener, limits.connections),
+ listen::Door::new(listener, limits.connections, crowd()),
Some(clients.rustls.clone()),
clients_router,
handle.clone(),
@@ -208,7 +211,7 @@ pub async fn serve(state: AppState) -> Result<()> {
});
listen::serve(
- listen::Door::new(listener, limits.connections),
+ listen::Door::new(listener, limits.connections, crowd()),
tls,
router,
handle,
@@ -399,8 +402,9 @@ fn build_router(state: AppState) -> Router {
.route("/health", get(health))
.route("/ready", get(ready))
// The media kept, outside every guard: a key is a hash nobody
- // guesses, and Sonarr fetches a poster with no credential.
- .route("/media/{key}", get(crate::media::serve::get))
+ // guesses, and Sonarr fetches a poster with no credential. A copy
+ // made on demand counts against its caller, as the API does.
+ .route("/media/{key}", get(crate::media::serve::get_for_caller))
// The authority the clients are to trust, and the script that makes
// a container trust it: public, since a client fetches them before
// it can trust anything.
@@ -800,9 +804,11 @@ struct Ready {
tmdb_configured: bool,
}
-/// Liveness is `/health`; this is readiness, and it touches the database.
+/// Liveness is `/health`; this is readiness, and it touches the database —
+/// once in a few seconds, however often it is asked, since anybody may ask
+/// under any name and with no credential.
async fn ready(State(state): State) -> (StatusCode, Json) {
- let database = state.db.health().await.is_ok();
+ let database = state.caches.database_answers(&state.db).await;
let status = if database {
StatusCode::OK
@@ -972,6 +978,54 @@ mod policy_tests {
}
}
+#[cfg(test)]
+mod ready_tests {
+ use axum::{body::Body, http::Request};
+ use tower::ServiceExt as _;
+
+ use super::*;
+
+ /// What `/ready` says, as the status and whether it says the database
+ /// answers.
+ async fn readiness(router: &Router) -> (StatusCode, bool) {
+ let answer = router
+ .clone()
+ .oneshot(Request::get("/ready").body(Body::empty()).unwrap())
+ .await
+ .unwrap();
+ let status = answer.status();
+ let body = axum::body::to_bytes(answer.into_body(), usize::MAX)
+ .await
+ .unwrap();
+ let said: serde_json::Value = serde_json::from_slice(&body).unwrap();
+ (status, said["database"] == true)
+ }
+
+ /// The database is asked once in a few seconds, however often `/ready` is:
+ /// what was said is said again while the database goes away, and a server
+ /// that has not been asked yet finds out.
+ #[tokio::test]
+ async fn readiness_asks_the_database_once_in_a_few_seconds() {
+ let state = crate::api::native::testing::server().await;
+ let router = build_router(state.clone());
+ assert_eq!(readiness(&router).await, (StatusCode::OK, true));
+
+ state.db.close().await;
+ assert_eq!(
+ readiness(&router).await,
+ (StatusCode::OK, true),
+ "kept: the database was not asked again"
+ );
+
+ let down = crate::api::native::testing::server().await;
+ down.db.close().await;
+ assert_eq!(
+ readiness(&build_router(down)).await,
+ (StatusCode::SERVICE_UNAVAILABLE, false)
+ );
+ }
+}
+
#[cfg(test)]
mod well_known_tests {
use axum::{body::Body, http::Request};
diff --git a/src/web/listen.rs b/src/web/listen.rs
index 7d665d9..e39c7f3 100644
--- a/src/web/listen.rs
+++ b/src/web/listen.rs
@@ -11,37 +11,57 @@
//! * a semaphore in the accept loop holds each door to `AMS_MAX_CONNECTIONS`:
//! past it, a connection waits in the system's queue, not accepted, until
//! another one closes;
+//! * no one network holds more than [`PER_PEER`] of them: a connection past
+//! that is closed as it is accepted, so that one address cannot take every
+//! place of the door — a trusted proxy's, and this machine's own, are not
+//! counted, since everybody behind a proxy has its address;
//! * hyper is given a clock, and with it `AMS_HEADER_READ_TIMEOUT` for a
//! request's head — the first, and each one a connection kept alive waits
//! for — and a head no larger than a head needs to be;
//! * the bytes before that, which tell HTTP/2 from HTTP/1 and which hyper
//! does not time, are given the same;
-//! * an idle HTTP/2 connection is pinged, and closed when it stops answering.
+//! * an idle HTTP/2 connection is pinged, and closed when it stops answering —
+//! and closed when it has had no request in flight for [`IDLE`] whatever it
+//! answers, since a ping keeps a connection alive however little it is used;
+//! * a response that cannot be written because its peer does not read it is
+//! given up on after [`WRITE_STALL`], and the connection with it.
use std::{
+ collections::HashMap,
+ convert::Infallible,
future::Future,
io,
- net::SocketAddr,
+ net::{IpAddr, SocketAddr},
pin::Pin,
sync::{
- Arc, LazyLock,
- atomic::{AtomicU64, Ordering},
+ Arc, LazyLock, Mutex, PoisonError,
+ atomic::{AtomicU64, AtomicUsize, Ordering},
},
task::{Context, Poll},
time::{Duration, Instant},
};
-use axum::{Router, extract::Request};
+use axum::{
+ Router,
+ extract::ConnectInfo,
+ http::{Request, Response},
+};
use axum_server::{
AddrListener, Address, Handle, Server,
accept::{Accept, DefaultAcceptor},
tls_rustls::{RustlsAcceptor, RustlsConfig},
};
+use bytes::Bytes;
+use hyper::body::{Body as HttpBody, Frame, Incoming, SizeHint};
+use ipnet::IpNet;
use tokio::{
io::{AsyncRead, AsyncWrite, ReadBuf},
net::{TcpListener, TcpStream},
sync::{OwnedSemaphorePermit, Semaphore},
};
+use tower::Service;
+
+use crate::auth::ip;
/// What every door is held to.
#[derive(Clone, Copy, Debug)]
@@ -50,12 +70,37 @@ pub struct Limits {
pub connections: usize,
/// How long a connection has to say what it wants.
pub head: Duration,
+ /// How long a connection may have no request in flight.
+ pub idle: Duration,
+ /// How long a response may wait to be written to a peer that does not
+ /// read it.
+ pub write: Duration,
+}
+
+impl Limits {
+ /// The limits of a door holding `connections` at once, whose connections
+ /// have `head` to say what they want: idle no longer than that, nor than
+ /// [`IDLE`] — a head's time is what a connection kept alive was given to
+ /// sit idle in, and a longer one is the operator's word for it.
+ pub fn new(connections: usize, head: Duration) -> Self {
+ Self {
+ connections,
+ head,
+ idle: IDLE.max(head),
+ write: WRITE_STALL,
+ }
+ }
}
/// Connections a door bound by address alone holds; the configured doors
/// are given theirs.
const DEFAULT_CONNECTIONS: usize = 1024;
+/// The most connections one network holds at once — an address, or an IPv6
+/// /64, as the rate limit counts a caller. A browser holds a handful, a
+/// Sonarr or a Plex a few; a thousand and twenty-four is sixteen times this.
+const PER_PEER: usize = 64;
+
/// The most a request's head may take in a connection's buffer. A head is
/// a few kilobytes; hyper's default lets each connection hold four hundred
/// before refusing one, which a door full of slow connections turns into
@@ -67,20 +112,162 @@ const HEAD_BUFFER: usize = 64 * 1024;
const KEEP_ALIVE: Duration = Duration::from_secs(30);
const KEEP_ALIVE_TIMEOUT: Duration = Duration::from_secs(20);
-/// The address a connection came from, as the doors hand it on.
-#[derive(Clone, Copy, Debug)]
-pub struct Peer(pub SocketAddr);
+/// How long a connection may have no request in flight before it is closed.
+/// Longer than the idle time most clients keep a connection for — a browser's
+/// minute or two, a proxy's — so that a client reuses a connection that is
+/// still open, not one being closed as it sends.
+const IDLE: Duration = Duration::from_secs(120);
+
+/// How long a response may wait to be written to a peer that does not read
+/// it — the time a send buffer may stay full, as a web server's own send
+/// timeout is — before it is given up on.
+const WRITE_STALL: Duration = Duration::from_secs(60);
+
+/// How long bytes must have stopped coming before an idle connection is
+/// closed, so that a request on its way is not met by the connection closing:
+/// the bytes of a request read are not yet a request in flight.
+const SETTLE: Duration = Duration::from_secs(1);
+
+/// The address a connection came from, as the doors hand it on, and what the
+/// connection is doing.
+#[derive(Clone, Debug)]
+pub struct Peer {
+ addr: SocketAddr,
+ life: Arc,
+}
impl Address for Peer {
type Stream = Held;
type Listener = Door;
}
-/// What the handlers read as `ConnectInfo`: the peer's address,
-/// as before.
-impl axum::extract::connect_info::Connected for SocketAddr {
- fn connect_info(peer: Peer) -> Self {
- peer.0
+/// How many connections each network holds at once, and the networks nobody
+/// counts.
+pub struct Crowd {
+ most: usize,
+ exempt: Vec,
+ holding: Mutex>,
+}
+
+impl Crowd {
+ /// Every network held to [`PER_PEER`] connections but the `proxies` —
+ /// which hold those of everybody behind them — and this machine.
+ pub fn new(proxies: &[IpNet]) -> Self {
+ let machine = ["127.0.0.0/8", "::1/128"]
+ .into_iter()
+ .filter_map(|network| network.parse::().ok());
+ Self::of(PER_PEER, proxies.iter().copied().chain(machine).collect())
+ }
+
+ fn of(most: usize, exempt: Vec) -> Self {
+ Self {
+ most: most.max(1),
+ exempt,
+ holding: Mutex::new(HashMap::new()),
+ }
+ }
+
+ /// A share for a connection from `peer` — or none, when its network
+ /// holds as many as it may.
+ fn admit(self: &Arc, peer: SocketAddr) -> Option {
+ if ip::is_trusted_peer(Some(peer), &self.exempt) {
+ return Some(Share {
+ crowd: self.clone(),
+ network: None,
+ });
+ }
+ let network = ip::bucket(peer.ip());
+ let mut holding = self.holding.lock().unwrap_or_else(PoisonError::into_inner);
+ let held = holding.entry(network).or_default();
+ if *held >= self.most {
+ return None;
+ }
+ *held += 1;
+ Some(Share {
+ crowd: self.clone(),
+ network: Some(network),
+ })
+ }
+
+ /// How many networks hold a connection now.
+ #[cfg(test)]
+ fn networks(&self) -> usize {
+ self.holding
+ .lock()
+ .unwrap_or_else(PoisonError::into_inner)
+ .len()
+ }
+}
+
+/// A connection's place in its network's count, given back when it closes.
+struct Share {
+ crowd: Arc,
+ /// The network it is counted under; none for one nobody counts.
+ network: Option,
+}
+
+impl Drop for Share {
+ fn drop(&mut self) {
+ let Some(network) = self.network else {
+ return;
+ };
+ let mut holding = self
+ .crowd
+ .holding
+ .lock()
+ .unwrap_or_else(PoisonError::into_inner);
+ if let Some(held) = holding.get_mut(&network) {
+ *held = held.saturating_sub(1);
+ if *held == 0 {
+ holding.remove(&network);
+ }
+ }
+ }
+}
+
+/// What a connection is doing: shared by the stream it is read from and
+/// written to, and the service its requests are given to.
+#[derive(Debug)]
+struct Life {
+ /// The requests begun and not yet done with: a response whose last byte
+ /// was written, or whose caller has gone.
+ open: AtomicUsize,
+ /// When the connection last had none, or began: milliseconds after
+ /// [`EPOCH`].
+ idle_since: AtomicU64,
+}
+
+impl Life {
+ fn new() -> Self {
+ Self {
+ open: AtomicUsize::new(0),
+ idle_since: AtomicU64::new(now_ms()),
+ }
+ }
+
+ /// A request begins.
+ fn begin(self: &Arc) -> Open {
+ self.open.fetch_add(1, Ordering::SeqCst);
+ Open(self.clone())
+ }
+
+ fn open(&self) -> usize {
+ self.open.load(Ordering::SeqCst)
+ }
+
+ fn idle_since(&self) -> u64 {
+ self.idle_since.load(Ordering::SeqCst)
+ }
+}
+
+/// A request in flight, until it is dropped.
+struct Open(Arc);
+
+impl Drop for Open {
+ fn drop(&mut self) {
+ if self.0.open.fetch_sub(1, Ordering::SeqCst) == 1 {
+ self.0.idle_since.store(now_ms(), Ordering::SeqCst);
+ }
}
}
@@ -89,34 +276,53 @@ pub struct Door {
listener: TcpListener,
connections: usize,
places: Arc,
+ crowd: Arc,
/// When the door last said it was full, in seconds after [`EPOCH`] plus
/// one; nought for never.
said_full: AtomicU64,
+ /// The same, for a network that holds as many as it may.
+ said_crowded: AtomicU64,
}
static EPOCH: LazyLock = LazyLock::new(Instant::now);
+/// Milliseconds since [`EPOCH`].
+fn now_ms() -> u64 {
+ u64::try_from(EPOCH.elapsed().as_millis()).unwrap_or(u64::MAX)
+}
+
+fn millis(time: Duration) -> u64 {
+ u64::try_from(time.as_millis()).unwrap_or(u64::MAX)
+}
+
impl Door {
- pub fn new(listener: TcpListener, connections: usize) -> Self {
+ pub fn new(listener: TcpListener, connections: usize, crowd: Crowd) -> Self {
let connections = connections.max(1);
Self {
listener,
connections,
places: Arc::new(Semaphore::new(connections)),
+ crowd: Arc::new(crowd),
said_full: AtomicU64::new(0),
+ said_crowded: AtomicU64::new(0),
}
}
+}
+
+/// Whether a thing that stays so a while — a door full, a network holding its
+/// share — is to be said now: once a minute at most.
+fn say(last: &AtomicU64) -> bool {
+ let now = EPOCH.elapsed().as_secs() + 1;
+ let before = last.load(Ordering::Relaxed);
+ (before == 0 || now.saturating_sub(before) >= 60)
+ && last
+ .compare_exchange(before, now, Ordering::Relaxed, Ordering::Relaxed)
+ .is_ok()
+}
- /// Said once a minute at most: a door that is full stays full a while.
+impl Door {
fn say_full(&self) {
- let now = EPOCH.elapsed().as_secs() + 1;
- let last = self.said_full.load(Ordering::Relaxed);
- if (last == 0 || now.saturating_sub(last) >= 60)
- && self
- .said_full
- .compare_exchange(last, now, Ordering::Relaxed, Ordering::Relaxed)
- .is_ok()
- {
+ if say(&self.said_full) {
tracing::warn!(
connections = self.connections,
"a door holds as many connections as AMS_MAX_CONNECTIONS allows; new ones wait \
@@ -124,18 +330,31 @@ impl Door {
);
}
}
+
+ fn say_crowded(&self, peer: SocketAddr) {
+ if say(&self.said_crowded) {
+ tracing::warn!(
+ %peer,
+ most = self.crowd.most,
+ "a network holds as many connections as one may; its new ones are closed"
+ );
+ }
+ }
}
impl AddrListener for Door {
async fn bind_to(addr: Peer) -> io::Result {
Ok(Self::new(
- TcpListener::bind(addr.0).await?,
+ TcpListener::bind(addr.addr).await?,
DEFAULT_CONNECTIONS,
+ Crowd::new(&[]),
))
}
/// A place first, then the connection: past the limit nothing more is
- /// accepted, so nothing more holds a descriptor.
+ /// accepted, so nothing more holds a descriptor. A connection whose
+ /// network holds its share already is closed as it is accepted, and the
+ /// place goes to the next.
async fn accept_stream(&self) -> io::Result<(Held, Peer)> {
let place = match self.places.clone().try_acquire_owned() {
Ok(place) => place,
@@ -148,25 +367,41 @@ impl AddrListener for Door {
.map_err(io::Error::other)?
}
};
- let (stream, addr) = self.listener.accept().await?;
- Ok((
- Held {
- stream,
- _place: place,
- },
- Peer(addr),
- ))
+ loop {
+ let (stream, addr) = self.listener.accept().await?;
+ let Some(share) = self.crowd.admit(addr) else {
+ self.say_crowded(addr);
+ drop(stream);
+ continue;
+ };
+ let life = Arc::new(Life::new());
+ return Ok((
+ Held {
+ stream,
+ life: life.clone(),
+ _place: place,
+ _share: share,
+ },
+ Peer { addr, life },
+ ));
+ }
}
fn get_local_addr(&self) -> io::Result {
- self.listener.local_addr().map(Peer)
+ self.listener.local_addr().map(|addr| Peer {
+ addr,
+ life: Arc::new(Life::new()),
+ })
}
}
-/// An accepted connection, holding its place at the door until it closes.
+/// An accepted connection, holding its place at the door and in its network's
+/// count until it closes.
pub struct Held {
stream: TcpStream,
+ life: Arc,
_place: OwnedSemaphorePermit,
+ _share: Share,
}
impl AsyncRead for Held {
@@ -209,31 +444,44 @@ impl AsyncWrite for Held {
}
}
+/// How long a connection may be idle, and how long a write may stall.
+#[derive(Clone, Copy, Debug)]
+struct Tempo {
+ idle: Duration,
+ write: Duration,
+}
+
/// Gives a connection's first bytes — after TLS, where there is any — the
-/// time a request's head is given.
+/// time a request's head is given, and watches it from then on
+/// ([`Watched`]).
#[derive(Clone)]
pub struct Prompt {
inner: A,
within: Duration,
+ tempo: Tempo,
}
-impl Accept for Prompt
+impl Accept for Prompt
where
- A: Accept,
+ A: Accept,
A::Future: Send + 'static,
A::Stream: Send + 'static,
A::Service: Send + 'static,
{
- type Stream = Timed;
+ type Stream = Watched>;
type Service = A::Service;
type Future = Pin> + Send>>;
- fn accept(&self, stream: I, service: S) -> Self::Future {
+ fn accept(&self, stream: Held, service: S) -> Self::Future {
+ let life = stream.life.clone();
let accepted = self.inner.accept(stream, service);
- let within = self.within;
+ let (within, tempo) = (self.within, self.tempo);
Box::pin(async move {
let (stream, service) = accepted.await?;
- Ok((Timed::new(stream, within), service))
+ Ok((
+ Watched::new(Timed::new(stream, within), life, tempo),
+ service,
+ ))
})
}
}
@@ -341,6 +589,280 @@ impl AsyncWrite for Timed {
}
}
+/// A connection watched for what hyper's own timers do not see.
+///
+/// *Idle.* An HTTP/2 connection is pinged while it has nothing to say, and a
+/// peer that answers the pings is a peer that is there: kept for ever,
+/// whatever it asks, and so are those of a client that asks for something
+/// once a minute. One with no request in flight for [`Tempo::idle`] is
+/// closed — read to its end, as if its peer had gone — unless bytes are still
+/// coming in, which it is given a moment to finish.
+///
+/// *Stalled writes.* A response a peer does not read fills the send buffer,
+/// and what hyper writes waits for room for ever. A write that makes no
+/// headway for [`Tempo::write`] fails, and the connection with it.
+pub struct Watched {
+ inner: S,
+ life: Arc,
+ tempo: Tempo,
+ /// Wakes a read that has nothing to read when the connection may be due
+ /// to close.
+ idle: Pin>,
+ /// The moment `idle` is set to wake, in milliseconds after [`EPOCH`].
+ armed_for: u64,
+ /// When bytes were last read, in milliseconds after [`EPOCH`].
+ read_at: u64,
+ /// Armed while a write cannot go through.
+ stalled: Option>>,
+}
+
+/// When an idle connection is to be closed, in milliseconds after
+/// [`EPOCH`]: once it has had no request in flight for `idle` — from `since`
+/// — and bytes have stopped coming for [`SETTLE`] (a quarter of `idle`, where
+/// that is less) since `read_at`; but not later than twice `idle` from
+/// `since`, whatever arrives, since a peer that sends something every second
+/// is no peer that wants an answer.
+fn closes_at(since: u64, read_at: u64, idle: Duration) -> u64 {
+ let idle = millis(idle);
+ let settle = millis(SETTLE).min(idle / 4);
+ let earliest = since.saturating_add(idle);
+ let latest = since.saturating_add(idle.saturating_mul(2));
+ earliest.max(read_at.saturating_add(settle).min(latest))
+}
+
+impl Watched {
+ fn new(inner: S, life: Arc, tempo: Tempo) -> Self {
+ Self {
+ inner,
+ life,
+ tempo,
+ idle: Box::pin(tokio::time::sleep(tempo.idle)),
+ armed_for: now_ms().saturating_add(millis(tempo.idle)),
+ read_at: now_ms(),
+ stalled: None,
+ }
+ }
+
+ /// Whether the connection is to be closed now; where it is not, the timer
+ /// that wakes the read when it may be is set, and has the task's waker.
+ fn closes_idle(&mut self, cx: &mut Context<'_>) -> bool {
+ loop {
+ let now = now_ms();
+ let busy = self.life.open() > 0;
+ let at = if busy {
+ // Looked at again after an idle time: what is asked of it
+ // ends, and the connection is idle from then.
+ now.saturating_add(millis(self.tempo.idle))
+ } else {
+ closes_at(self.life.idle_since(), self.read_at, self.tempo.idle)
+ };
+ if !busy && now >= at {
+ return true;
+ }
+ // Set again only where it would wake the read too late, or has
+ // gone off: a busy connection finds nothing to read after every
+ // batch of bytes, and a timer moved each time is work for nothing.
+ // One that wakes it too soon only has it look again.
+ if self.armed_for <= now || self.armed_for > at {
+ let wake = tokio::time::Instant::now() + Duration::from_millis(at - now);
+ self.idle.as_mut().reset(wake);
+ self.armed_for = at;
+ }
+ if self.idle.as_mut().poll(cx).is_pending() {
+ return false;
+ }
+ // Gone off: looked at again, and set for what is then due.
+ self.armed_for = 0;
+ }
+ }
+
+ /// A write's answer, given up on if it has been pending for as long as
+ /// writes may stall.
+ fn watch(
+ &mut self,
+ cx: &mut Context<'_>,
+ polled: Poll>,
+ ) -> Poll> {
+ match polled {
+ Poll::Pending => {
+ let within = self.tempo.write;
+ let stalled = self
+ .stalled
+ .get_or_insert_with(|| Box::pin(tokio::time::sleep(within)));
+ if stalled.as_mut().poll(cx).is_ready() {
+ self.stalled = None;
+ return Poll::Ready(Err(io::Error::new(
+ io::ErrorKind::TimedOut,
+ "the peer did not read what it was sent in time",
+ )));
+ }
+ Poll::Pending
+ }
+ ready => {
+ self.stalled = None;
+ ready
+ }
+ }
+ }
+}
+
+impl AsyncRead for Watched {
+ fn poll_read(
+ self: Pin<&mut Self>,
+ cx: &mut Context<'_>,
+ buf: &mut ReadBuf<'_>,
+ ) -> Poll> {
+ let this = self.get_mut();
+ let before = buf.filled().len();
+ match Pin::new(&mut this.inner).poll_read(cx, buf) {
+ Poll::Ready(Ok(())) => {
+ if buf.filled().len() > before {
+ this.read_at = now_ms();
+ }
+ Poll::Ready(Ok(()))
+ }
+ Poll::Pending => {
+ if this.closes_idle(cx) {
+ // The end of what the peer sent: hyper takes the peer
+ // for gone, and closes the connection.
+ return Poll::Ready(Ok(()));
+ }
+ Poll::Pending
+ }
+ failed => failed,
+ }
+ }
+}
+
+impl AsyncWrite for Watched {
+ fn poll_write(
+ self: Pin<&mut Self>,
+ cx: &mut Context<'_>,
+ buf: &[u8],
+ ) -> Poll> {
+ let this = self.get_mut();
+ let polled = Pin::new(&mut this.inner).poll_write(cx, buf);
+ this.watch(cx, polled)
+ }
+
+ fn poll_write_vectored(
+ self: Pin<&mut Self>,
+ cx: &mut Context<'_>,
+ bufs: &[io::IoSlice<'_>],
+ ) -> Poll> {
+ let this = self.get_mut();
+ let polled = Pin::new(&mut this.inner).poll_write_vectored(cx, bufs);
+ this.watch(cx, polled)
+ }
+
+ fn is_write_vectored(&self) -> bool {
+ self.inner.is_write_vectored()
+ }
+
+ fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> {
+ let this = self.get_mut();
+ let polled = Pin::new(&mut this.inner).poll_flush(cx);
+ this.watch(cx, polled)
+ }
+
+ fn poll_shutdown(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> {
+ let this = self.get_mut();
+ let polled = Pin::new(&mut this.inner).poll_shutdown(cx);
+ this.watch(cx, polled)
+ }
+}
+
+/// What the requests of the connections of a door are served by: made for
+/// each connection, with the address it came from and what it is doing.
+#[derive(Clone)]
+struct Connections {
+ router: Router,
+}
+
+impl Service for Connections {
+ type Response = Tracked;
+ type Error = Infallible;
+ type Future = std::future::Ready>;
+
+ fn poll_ready(&mut self, _: &mut Context<'_>) -> Poll> {
+ Poll::Ready(Ok(()))
+ }
+
+ fn call(&mut self, peer: Peer) -> Self::Future {
+ std::future::ready(Ok(Tracked {
+ router: self.router.clone(),
+ addr: peer.addr,
+ life: peer.life,
+ }))
+ }
+}
+
+/// The router, serving one connection's requests: each told the address the
+/// connection came from — as `ConnectInfo`, which is what the
+/// handlers read — and counted among those in flight until its response's
+/// last byte is written.
+#[derive(Clone)]
+struct Tracked {
+ router: Router,
+ addr: SocketAddr,
+ life: Arc,
+}
+
+impl Service> for Tracked {
+ type Response = Response;
+ type Error = Infallible;
+ type Future = Pin> + Send>>;
+
+ fn poll_ready(&mut self, _: &mut Context<'_>) -> Poll> {
+ Poll::Ready(Ok(()))
+ }
+
+ fn call(&mut self, mut request: Request) -> Self::Future {
+ request.extensions_mut().insert(ConnectInfo(self.addr));
+ // Counted from here, and given to the response's body below: a
+ // request whose future is dropped — its caller gone — is no longer
+ // in flight either.
+ let open = self.life.begin();
+ let mut router = self.router.clone();
+ Box::pin(async move {
+ let response = match router.call(request).await {
+ Ok(response) => response,
+ Err(never) => match never {},
+ };
+ Ok(response.map(|body| Counted { body, _open: open }))
+ })
+ }
+}
+
+/// A response's body, keeping its request among those in flight until it is
+/// dropped — written to its end, or its caller gone. Everything else of the
+/// body is the body's: its frames, whether it is at its end, how large it is
+/// (hyper writes a `Content-Length` of it).
+struct Counted {
+ body: axum::body::Body,
+ _open: Open,
+}
+
+impl HttpBody for Counted {
+ type Data = Bytes;
+ type Error = axum::Error;
+
+ fn poll_frame(
+ mut self: Pin<&mut Self>,
+ cx: &mut Context<'_>,
+ ) -> Poll, axum::Error>>> {
+ Pin::new(&mut self.body).poll_frame(cx)
+ }
+
+ fn is_end_stream(&self) -> bool {
+ self.body.is_end_stream()
+ }
+
+ fn size_hint(&self) -> SizeHint {
+ self.body.size_hint()
+ }
+}
+
/// hyper's clock, kept by tokio: what lets hyper time a connection out.
#[derive(Clone, Copy, Debug)]
struct Clock;
@@ -376,14 +898,18 @@ pub async fn serve(
handle: Handle,
limits: Limits,
) -> io::Result<()> {
- let service =
- axum::ServiceExt::::into_make_service_with_connect_info::(router);
+ let service = Connections { router };
+ let tempo = Tempo {
+ idle: limits.idle,
+ write: limits.write,
+ };
match tls {
Some(config) => {
let mut server = Server::::from_listener(door)
.acceptor(Prompt {
inner: RustlsAcceptor::new(config),
within: limits.head,
+ tempo,
})
.handle(handle);
tune(&mut server, limits);
@@ -394,6 +920,7 @@ pub async fn serve(
.acceptor(Prompt {
inner: DefaultAcceptor::new(),
within: limits.head,
+ tempo,
})
.handle(handle);
tune(&mut server, limits);
@@ -423,14 +950,51 @@ mod tests {
use super::*;
use tokio::io::{AsyncReadExt, AsyncWriteExt};
- /// A door serving a single route, with the limits given; its address.
- async fn door(limits: Limits) -> (SocketAddr, Handle) {
+ impl Crowd {
+ /// Every network counted, this machine's too: what a test of the
+ /// count runs on, over loopback.
+ fn counting_all(most: usize) -> Self {
+ Self::of(most, Vec::new())
+ }
+ }
+
+ /// The limits a door of these tests has: generous, but for what a test
+ /// narrows.
+ fn limits() -> Limits {
+ Limits {
+ connections: 8,
+ head: Duration::from_secs(30),
+ idle: Duration::from_secs(60),
+ write: Duration::from_secs(60),
+ }
+ }
+
+ /// A door serving a few routes, with the limits given, its network
+ /// counted as the crowd says; its address.
+ async fn door_with(limits: Limits, crowd: Crowd) -> (SocketAddr, Handle) {
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
let addr = listener.local_addr().unwrap();
- let router = Router::new().route("/", axum::routing::get(|| async { "here" }));
+ let router = Router::new()
+ .route("/", axum::routing::get(|| async { "here" }))
+ .route(
+ "/slow",
+ axum::routing::get(|| async {
+ tokio::time::sleep(Duration::from_millis(900)).await;
+ "slowly"
+ }),
+ )
+ .route(
+ "/endless",
+ axum::routing::get(|| async {
+ let chunks = futures::stream::repeat_with(|| {
+ Ok::<_, io::Error>(Bytes::from(vec![0u8; 64 * 1024]))
+ });
+ axum::body::Body::from_stream(chunks)
+ }),
+ );
let handle = Handle::new();
tokio::spawn(serve(
- Door::new(listener, limits.connections),
+ Door::new(listener, limits.connections, crowd),
None,
router,
handle.clone(),
@@ -439,6 +1003,10 @@ mod tests {
(addr, handle)
}
+ async fn door(limits: Limits) -> (SocketAddr, Handle) {
+ door_with(limits, Crowd::new(&[])).await
+ }
+
async fn ask(addr: SocketAddr) -> String {
let mut stream = TcpStream::connect(addr).await.unwrap();
stream
@@ -456,11 +1024,31 @@ mod tests {
(head.lines().next().unwrap_or_default(), body)
}
+ /// Whether the door has closed a connection: its end is read within
+ /// `within`, and nothing else is.
+ async fn closed_within(stream: &mut TcpStream, within: Duration) -> bool {
+ let mut rest = Vec::new();
+ matches!(
+ tokio::time::timeout(within, stream.read_to_end(&mut rest)).await,
+ Ok(Ok(_) | Err(_))
+ )
+ }
+
+ /// Whether the door leaves a connection open for `for_`: nothing is read
+ /// of it, and it is not closed.
+ async fn open_for(stream: &mut TcpStream, for_: Duration) -> bool {
+ let mut byte = [0u8; 1];
+ tokio::time::timeout(for_, stream.read(&mut byte))
+ .await
+ .is_err()
+ }
+
#[tokio::test]
async fn a_door_answers_through_its_limits() {
let (addr, handle) = door(Limits {
connections: 4,
head: Duration::from_secs(5),
+ ..limits()
})
.await;
let answer = ask(addr).await;
@@ -473,6 +1061,7 @@ mod tests {
let (addr, handle) = door(Limits {
connections: 4,
head: Duration::from_millis(200),
+ ..limits()
})
.await;
@@ -516,6 +1105,7 @@ mod tests {
let (addr, handle) = door(Limits {
connections: 1,
head: Duration::from_secs(30),
+ ..limits()
})
.await;
@@ -557,4 +1147,366 @@ mod tests {
nothing.saw(b"");
assert!(nothing.deadline.is_some());
}
+
+ // ── the count of a network's connections ────────────────────────────
+
+ fn peer(address: &str) -> SocketAddr {
+ format!("{address}:40000").parse().unwrap()
+ }
+
+ #[test]
+ fn a_network_holds_so_many_connections_and_gives_a_share_back() {
+ let crowd = Arc::new(Crowd::counting_all(2));
+ let first = crowd.admit(peer("203.0.113.5")).expect("a share");
+ let _second = crowd.admit(peer("203.0.113.5")).expect("a share");
+ assert!(
+ crowd.admit(peer("203.0.113.5")).is_none(),
+ "a third is more than one network may hold"
+ );
+ // Another address has its own.
+ let _other = crowd.admit(peer("203.0.113.6")).expect("a share");
+ assert_eq!(crowd.networks(), 2);
+
+ // A share given back is the next connection's.
+ drop(first);
+ let _third = crowd
+ .admit(peer("203.0.113.5"))
+ .expect("the share given back");
+ assert!(crowd.admit(peer("203.0.113.5")).is_none());
+ }
+
+ #[test]
+ fn a_count_does_not_outlive_its_connections() {
+ let crowd = Arc::new(Crowd::counting_all(3));
+ let shares: Vec<_> = (0..3)
+ .map(|_| crowd.admit(peer("198.51.100.7")).expect("a share"))
+ .collect();
+ assert_eq!(crowd.networks(), 1);
+ drop(shares);
+ assert_eq!(crowd.networks(), 0, "nothing is kept of a network gone");
+ }
+
+ /// One machine has a whole /64 to pick its addresses from: they are one
+ /// network, as the rate limit counts them.
+ #[test]
+ fn an_ipv6_network_holds_one_count_between_its_addresses() {
+ let crowd = Arc::new(Crowd::counting_all(2));
+ let _a = crowd.admit(peer("[2001:db8:1:2::1]")).expect("a share");
+ let _b = crowd
+ .admit(peer("[2001:db8:1:2:ffff::9]"))
+ .expect("a share");
+ assert!(crowd.admit(peer("[2001:db8:1:2::3]")).is_none());
+ assert!(crowd.admit(peer("[2001:db8:1:3::1]")).is_some());
+
+ // An IPv4 client of a dual-stack socket is its IPv4 address.
+ let _c = crowd.admit(peer("[::ffff:192.0.2.1]")).expect("a share");
+ let _d = crowd.admit(peer("192.0.2.1")).expect("a share");
+ assert!(crowd.admit(peer("[::ffff:192.0.2.1]")).is_none());
+ }
+
+ /// Everybody behind a proxy has its address, and this machine's own are
+ /// the operator's: neither is counted.
+ #[test]
+ fn a_trusted_proxy_and_this_machine_are_not_counted() {
+ let proxies: Vec = vec!["10.0.0.0/8".parse().unwrap(), "fd00::/16".parse().unwrap()];
+ let mut crowd = Crowd::new(&proxies);
+ crowd.most = 1;
+ let crowd = Arc::new(crowd);
+ let mut held = Vec::new();
+ for address in [
+ "10.1.2.3",
+ "[fd00::7]",
+ "127.0.0.1",
+ "[::1]",
+ "[::ffff:127.0.0.1]",
+ ] {
+ for _ in 0..5 {
+ held.push(
+ crowd
+ .admit(peer(address))
+ .unwrap_or_else(|| panic!("{address} is counted")),
+ );
+ }
+ }
+ assert_eq!(crowd.networks(), 0);
+
+ // Any other is, and holds the one.
+ let _only = crowd.admit(peer("203.0.113.5")).expect("a share");
+ assert!(crowd.admit(peer("203.0.113.5")).is_none());
+ // The standard crowd leaves this machine out and counts the rest.
+ let standard = Arc::new(Crowd::new(&[]));
+ let counted: Vec<_> = (0..PER_PEER)
+ .map(|_| standard.admit(peer("203.0.113.5")).expect("a share"))
+ .collect();
+ assert!(standard.admit(peer("203.0.113.5")).is_none());
+ drop(counted);
+ }
+
+ #[tokio::test]
+ async fn a_network_past_its_share_is_closed_and_the_others_are_answered() {
+ let (addr, handle) = door_with(limits(), Crowd::counting_all(2)).await;
+
+ // Two connections, as many as this network may hold: idle, open.
+ let mut first = TcpStream::connect(addr).await.unwrap();
+ let mut second = TcpStream::connect(addr).await.unwrap();
+ tokio::time::sleep(Duration::from_millis(100)).await;
+
+ // The third is accepted, and closed with no word.
+ let mut third = TcpStream::connect(addr).await.unwrap();
+ assert!(
+ closed_within(&mut third, Duration::from_secs(5)).await,
+ "the door should have closed a third connection of one network"
+ );
+ // The two it holds are left as they were.
+ assert!(open_for(&mut first, Duration::from_millis(200)).await);
+ assert!(open_for(&mut second, Duration::from_millis(200)).await);
+
+ // One given back, the next is answered.
+ drop(first);
+ tokio::time::sleep(Duration::from_millis(100)).await;
+ let answer = tokio::time::timeout(Duration::from_secs(5), ask(addr))
+ .await
+ .expect("answered");
+ assert_eq!(parts(&answer).0, "HTTP/1.1 200 OK");
+ handle.shutdown();
+ }
+
+ // ── a connection with nothing in flight ─────────────────────────────
+
+ #[test]
+ fn an_idle_connection_is_closed_once_it_is_quiet_and_never_later_than_twice_its_time() {
+ let idle = Duration::from_secs(120);
+ // Quiet: closed when its idle time is up.
+ assert_eq!(closes_at(0, 0, idle), 120_000);
+ // Bytes read just before: a moment more, for what is on its way.
+ assert_eq!(closes_at(0, 119_900, idle), 120_900);
+ // Bytes read all the time: not later than twice the time.
+ assert_eq!(closes_at(0, 239_999, idle), 240_000);
+ assert_eq!(closes_at(1_000, 10_000_000, idle), 241_000);
+ // A short time settles in a quarter of it.
+ assert_eq!(closes_at(0, 100, Duration::from_millis(400)), 400);
+ assert_eq!(closes_at(0, 350, Duration::from_millis(400)), 450);
+ }
+
+ /// What HTTP/2 opens with and says next, as an idle client sends it: the
+ /// preface and an empty SETTINGS frame, and a PING.
+ fn h2_hello() -> Vec {
+ let mut bytes = PREFACE.to_vec();
+ bytes.extend_from_slice(&[0, 0, 0, 4, 0, 0, 0, 0, 0]);
+ bytes
+ }
+
+ fn h2_ping() -> [u8; 17] {
+ let mut ping = [0u8; 17];
+ ping[2] = 8;
+ ping[3] = 6;
+ ping
+ }
+
+ /// An idle HTTP/2 connection is closed whatever it answers — here, a
+ /// client that pings every fifty milliseconds — within twice the idle
+ /// time.
+ #[tokio::test]
+ async fn an_h2_connection_with_no_request_is_closed_though_it_pings() {
+ let (addr, handle) = door(Limits {
+ idle: Duration::from_millis(300),
+ ..limits()
+ })
+ .await;
+ let mut stream = TcpStream::connect(addr).await.unwrap();
+ stream.write_all(&h2_hello()).await.unwrap();
+
+ let started = Instant::now();
+ let mut buf = [0u8; 4096];
+ let mut closed_after = None;
+ while started.elapsed() < Duration::from_secs(10) {
+ if stream.write_all(&h2_ping()).await.is_err() {
+ closed_after = Some(started.elapsed());
+ break;
+ }
+ if let Ok(Ok(0) | Err(_)) =
+ tokio::time::timeout(Duration::from_millis(50), stream.read(&mut buf)).await
+ {
+ closed_after = Some(started.elapsed());
+ break;
+ }
+ }
+ let closed_after = closed_after.expect("the door should have closed the connection");
+ assert!(
+ closed_after >= Duration::from_millis(250),
+ "closed after {closed_after:?}, before its idle time"
+ );
+ assert!(
+ closed_after <= Duration::from_secs(5),
+ "closed after {closed_after:?}"
+ );
+ handle.shutdown();
+ }
+
+ /// An HTTP/1 connection kept alive between two requests is closed once
+ /// it has had none in flight for the idle time — here, shorter than the
+ /// head's.
+ #[tokio::test]
+ async fn an_h1_connection_with_no_request_is_closed_after_its_idle_time() {
+ let (addr, handle) = door(Limits {
+ idle: Duration::from_millis(800),
+ ..limits()
+ })
+ .await;
+ let mut stream = TcpStream::connect(addr).await.unwrap();
+ stream
+ .write_all(b"GET / HTTP/1.1\r\nHost: x\r\n\r\n")
+ .await
+ .unwrap();
+ let mut buf = [0u8; 1024];
+ let read = stream.read(&mut buf).await.unwrap();
+ assert!(String::from_utf8_lossy(&buf[..read]).starts_with("HTTP/1.1 200 OK"));
+
+ // Kept alive, and then closed.
+ assert!(open_for(&mut stream, Duration::from_millis(150)).await);
+ assert!(
+ closed_within(&mut stream, Duration::from_secs(3)).await,
+ "the door should have closed the idle connection"
+ );
+ handle.shutdown();
+ }
+
+ /// A connection with a request in flight is not idle, however long the
+ /// request takes: the answer comes, and the connection is closed after
+ /// it, once the idle time has passed.
+ #[tokio::test]
+ async fn a_request_in_flight_keeps_its_connection_open() {
+ let (addr, handle) = door(Limits {
+ idle: Duration::from_millis(300),
+ ..limits()
+ })
+ .await;
+ let mut stream = TcpStream::connect(addr).await.unwrap();
+ stream
+ .write_all(b"GET /slow HTTP/1.1\r\nHost: x\r\n\r\n")
+ .await
+ .unwrap();
+
+ // The handler takes three times the idle time to answer.
+ let mut answer = Vec::new();
+ let mut buf = [0u8; 1024];
+ while !String::from_utf8_lossy(&answer).contains("slowly") {
+ let read = tokio::time::timeout(Duration::from_secs(5), stream.read(&mut buf))
+ .await
+ .expect("an answer")
+ .expect("read");
+ assert!(read > 0, "closed before it was answered");
+ answer.extend_from_slice(&buf[..read]);
+ }
+ assert!(String::from_utf8_lossy(&answer).starts_with("HTTP/1.1 200 OK"));
+
+ // Nothing in flight now: closed once it has been idle.
+ assert!(
+ closed_within(&mut stream, Duration::from_secs(3)).await,
+ "the door should have closed the connection after its answer"
+ );
+ handle.shutdown();
+ }
+
+ // ── a response nobody reads ─────────────────────────────────────────
+
+ /// A response a peer does not read is given up on once a write has made
+ /// no headway for the write time, and the connection with it.
+ #[tokio::test]
+ async fn a_response_nobody_reads_is_given_up_on() {
+ let (addr, handle) = door(Limits {
+ write: Duration::from_millis(300),
+ ..limits()
+ })
+ .await;
+ let mut stream = TcpStream::connect(addr).await.unwrap();
+ stream
+ .write_all(b"GET /endless HTTP/1.1\r\nHost: x\r\n\r\n")
+ .await
+ .unwrap();
+
+ // Nothing is read while the send buffers fill and the write stalls.
+ tokio::time::sleep(Duration::from_millis(1200)).await;
+
+ // What was written before is there; and then the end, long before
+ // an endless body's.
+ let mut got = 0usize;
+ let mut buf = vec![0u8; 256 * 1024];
+ let ended = tokio::time::timeout(Duration::from_secs(10), async {
+ loop {
+ match stream.read(&mut buf).await {
+ Ok(0) | Err(_) => return,
+ Ok(read) => got += read,
+ }
+ }
+ })
+ .await;
+ assert!(
+ ended.is_ok(),
+ "the door should have given up on the response"
+ );
+ assert!(
+ got < 64 * 1024 * 1024,
+ "read {got} bytes of an endless body"
+ );
+ handle.shutdown();
+ }
+
+ /// A peer that reads what it is sent is not stalled, however slowly it
+ /// asks for the rest: the clock runs only while a write cannot go through.
+ #[tokio::test]
+ async fn a_response_that_is_read_is_not_given_up_on() {
+ let (addr, handle) = door(Limits {
+ write: Duration::from_secs(1),
+ ..limits()
+ })
+ .await;
+ let mut stream = TcpStream::connect(addr).await.unwrap();
+ stream
+ .write_all(b"GET /endless HTTP/1.1\r\nHost: x\r\n\r\n")
+ .await
+ .unwrap();
+ let mut buf = vec![0u8; 256 * 1024];
+ let mut got = 0usize;
+ let started = Instant::now();
+ while started.elapsed() < Duration::from_millis(2500) {
+ let read = tokio::time::timeout(Duration::from_secs(5), stream.read(&mut buf))
+ .await
+ .expect("bytes keep coming")
+ .expect("read");
+ assert!(read > 0, "the door closed a connection that was reading");
+ got += read;
+ }
+ assert!(got > 1024 * 1024, "read {got} bytes");
+ handle.shutdown();
+ }
+
+ /// The service's requests count as in flight until the response's last
+ /// byte, and the body is the router's own: its length is still told.
+ #[tokio::test]
+ async fn the_body_keeps_its_length_and_its_request_counts_until_it_is_dropped() {
+ let life = Arc::new(Life::new());
+ let open = life.begin();
+ assert_eq!(life.open(), 1);
+ let counted = Counted {
+ body: axum::body::Body::from("twelve bytes"),
+ _open: open,
+ };
+ assert_eq!(counted.size_hint().exact(), Some(12));
+ assert!(!counted.is_end_stream());
+ assert_eq!(life.open(), 1);
+ let before = life.idle_since();
+ tokio::time::sleep(Duration::from_millis(20)).await;
+ drop(counted);
+ assert_eq!(life.open(), 0);
+ assert!(life.idle_since() > before, "idle from the moment it ended");
+
+ let empty = Counted {
+ body: axum::body::Body::empty(),
+ _open: life.begin(),
+ };
+ assert!(empty.is_end_stream());
+ drop(empty);
+ assert_eq!(life.open(), 0);
+ }
}