diff --git a/Dockerfile b/Dockerfile index a99b272..5deeed1 100644 --- a/Dockerfile +++ b/Dockerfile @@ -24,5 +24,18 @@ RUN apk add ca-certificates COPY --from=builder /app/target/release/opentubex-sync /app/opentubex-sync-server +# Run unprivileged. +# +# NOTE for SQLite deployments: a bind-mounted host directory keeps its host +# ownership and shadows the ownership set here, so the host directory must be +# owned by this uid or the database and its WAL sidecars are not writable: +# sudo chown -R 10001:10001 ./data +# See the "Running as non-root" section in README.md. +RUN addgroup -S -g 10001 opentubex \ + && adduser -S -u 10001 -G opentubex opentubex \ + && mkdir -p /app/data \ + && chown -R opentubex:opentubex /app +USER 10001:10001 + EXPOSE 8080 CMD ["./opentubex-sync-server"] diff --git a/README.md b/README.md index b0cf7e3..80e03fe 100644 --- a/README.md +++ b/README.md @@ -44,13 +44,89 @@ There are two ways to configure `sync-server` | Config option | Description | Default | Example | | ---------------------- | ---------------------------------------------------- | ------- | -------------------- | | `database_url` | Connection string for the database | None | sqlite://./db.sql | -| `secret_key` | Used to sign authentication tokens | None | SomeVeryLongString64 | +| `secret_key` | Used to sign authentication tokens. Required, min. 32 bytes | None | output of `openssl rand -hex 32` | +| `username_secret` | Used to derive account name hashes. Set it so that `secret_key` stays rotatable | falls back to `secret_key` | output of `openssl rand -hex 32` | +| `trust_forwarded_for` | Derive rate limiting client addresses from `X-Forwarded-For`. Required behind a reverse proxy | `false` | `true` | +| `trusted_proxy_hops` | Number of proxies in front of this server, used to pick the right `X-Forwarded-For` entry | `1` | `2` | | `allow_registration` | Whether to allow registering on this server | `true` | `false` | | `validate_submitted_metadata` | Whether to check incoming video data against YouTube | `true` | `false` | -| `migration_approval` | Exact comma-separated pending migration versions approved for an existing database after separately verifying a backup | None | `2026-07-21-180000-0000` | +| `migration_approval` | Exact comma-separated pending migration versions approved for an existing database after separately verifying a backup | None | `202607211800000000` | + +### Running behind a reverse proxy + +`/account/register` and `/account/login` are rate limited per client address. That +address is the immediate peer by default, which is only correct when the server is +directly reachable. + +The example compose files publish to `127.0.0.1`, so a reverse proxy is the normal +setup — and there every request arrives from the proxy's own address. Without +extra configuration all clients would then share a single bucket and legitimate +users would rate limit each other. Set `trust_forwarded_for = true` so the limit +is applied per real client: + +```yaml +environment: + - "TRUST_FORWARDED_FOR=true" +``` + +The address is taken from the *last* `X-Forwarded-For` entry, which is the one +your proxy appended, so a client cannot pick its own bucket by sending the header +itself. Only enable this when the server is not reachable directly, otherwise a +client can do exactly that. + +If you have more than one proxy in the chain — for example Cloudflare in front of +nginx — set `trusted_proxy_hops` to how many there are. Each proxy appends an +entry, so with two the last one is the inner proxy's address rather than the +client's, and leaving this at `1` would put every client in one bucket again. + +This limiter is per process and best-effort. Rate limiting at the proxy as well +is still recommended, and is required if you run multiple replicas. + +### Running as non-root + +The image runs as uid `10001`. A bind-mounted host directory keeps its host +ownership and shadows the ownership set in the image, so for SQLite deployments +prepare the data directory once before the first start: + +```sh +mkdir -p ./data +sudo chown -R 10001:10001 ./data +``` + +Without this the server cannot create `db.sqlite` or its WAL sidecar files. Note +that the failure surfaces as a database connection timeout rather than an obvious +permission error. If you are upgrading from an older image that ran as root, run +the same command against your existing `./data` directory — this changes +ownership only and does not touch the database contents. + +If you cannot use `sudo`, or your data directory is already owned by your own +user, override the container user instead of changing ownership: + +```yaml +services: + sync: + user: "1000:1000" # your own uid:gid, from `id -u`:`id -g` +``` + +The server only needs to write inside `/app/data`, so any uid that owns the data +directory works. This still avoids running as root. + +### Secrets + +The server refuses to start if `secret_key` is missing, shorter than 32 bytes, or +left at a placeholder such as `changeme`. Generate one with `openssl rand -hex 32`. + +Accounts are looked up by `HMAC(username)`, so changing the secret that derives +that hash makes every existing account unreachable. Set `username_secret` +explicitly (initially to the same value as `secret_key`) so that `secret_key` +itself can later be rotated — for example after a suspected leak — without +locking anyone out. Existing databases never apply pending migrations implicitly. After you create and verify -a separate backup, `migration_approval` must exactly match every pending version. SQLite +a separate backup, `migration_approval` must exactly match every pending version. +Versions are digits only, without the hyphens used in the migration directory +names — the startup error prints the exact value to use, so the simplest approach +is to start the server once and copy it from the log. SQLite also creates a consistent `*.pre-migration-` backup immediately before changing the schema. Remove the approval after deployment so it cannot authorize a later migration. diff --git a/config.toml b/config.toml index 8616ea2..bac2ba0 100644 --- a/config.toml +++ b/config.toml @@ -1,4 +1,13 @@ database_url = "./db.sqlite" -secret_key = "changeme" +# Generate with `openssl rand -hex 32`. The server refuses to start on a +# placeholder value or anything shorter than 32 bytes. +secret_key = "" +# Set this to the same value as secret_key above. Keeping it separate is what +# allows secret_key to be rotated later without locking out every account. +# Left unset it falls back to secret_key. +# username_secret = "" +# Enable only when the server is reachable exclusively through a reverse proxy, +# so that rate limiting sees real client addresses instead of the proxy's. +# trust_forwarded_for = true allow_registration = true validate_submitted_metadata = true diff --git a/docker-compose.postgres.yml b/docker-compose.postgres.yml index 2537ea6..aebe828 100644 --- a/docker-compose.postgres.yml +++ b/docker-compose.postgres.yml @@ -7,8 +7,18 @@ services: args: - "DATABASE_BACKEND=postgres" environment: - - "DATABASE_URL=postgres://postgres:password@db/opentubex-sync" - - "SECRET_KEY=changeme" + - "DATABASE_URL=postgres://postgres:${POSTGRES_PASSWORD:?generate one with: openssl rand -hex 32}@db/opentubex-sync" + # REQUIRED. Generate with `openssl rand -hex 32` and put it in a .env file + # next to this compose file. The server refuses to start on a placeholder + # value or anything shorter than 32 bytes. + - "SECRET_KEY=${SECRET_KEY:?generate one with: openssl rand -hex 32}" + # Set to the same value as SECRET_KEY. Keeping it separate is what allows + # SECRET_KEY to be rotated later without locking out every account. + - "USERNAME_SECRET=${USERNAME_SECRET:?set this to the same value as SECRET_KEY}" + # These files publish to 127.0.0.1, i.e. behind a reverse proxy, where + # every request arrives from the proxy and all clients would otherwise + # share one rate limit bucket. Remove this if exposing the port directly. + - "TRUST_FORWARDED_FOR=true" - "ALLOW_REGISTRATION=true" - "VALIDATE_SUBMITTED_METADATA=true" # Or use a configuration file: @@ -25,7 +35,7 @@ services: container_name: opentubex-sync-postgres environment: - "POSTGRES_USER=postgres" - - "POSTGRES_PASSWORD=password" + - "POSTGRES_PASSWORD=${POSTGRES_PASSWORD:?generate one with: openssl rand -hex 32}" - "POSTGRES_DB=opentubex-sync" volumes: - ./data:/var/lib/postgresql diff --git a/docker-compose.yml b/docker-compose.yml index ab76e3d..c1ea8ff 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -8,13 +8,26 @@ services: - "DATABASE_BACKEND=sqlite" environment: - "DATABASE_URL=./data/db.sqlite" - - "SECRET_KEY=changeme" + # REQUIRED. Generate with `openssl rand -hex 32` and put it in a .env file + # next to this compose file. The server refuses to start on a placeholder + # value or anything shorter than 32 bytes. + - "SECRET_KEY=${SECRET_KEY:?generate one with: openssl rand -hex 32}" + # Set to the same value as SECRET_KEY. Keeping it separate is what allows + # SECRET_KEY to be rotated later without locking out every account. + - "USERNAME_SECRET=${USERNAME_SECRET:?set this to the same value as SECRET_KEY}" + # These files publish to 127.0.0.1, i.e. behind a reverse proxy, where + # every request arrives from the proxy and all clients would otherwise + # share one rate limit bucket. Remove this if exposing the port directly. + - "TRUST_FORWARDED_FOR=true" - "ALLOW_REGISTRATION=true" - "VALIDATE_SUBMITTED_METADATA=true" # Existing databases refuse pending migrations unless this exactly lists # every pending migration version. Create and verify a backup before approval. - # - "MIGRATION_APPROVAL=2026-07-21-180000-0000" + # - "MIGRATION_APPROVAL=202607211800000000" volumes: + # The container runs as uid 10001 and this bind mount keeps host + # ownership, so create and chown the directory before the first start: + # mkdir -p ./data && sudo chown -R 10001:10001 ./data - ./data:/app/data ports: - "127.0.0.1:8080:8080" diff --git a/migrations/2026-07-28-150000-0000_watch_history_account_index/down.sql b/migrations/2026-07-28-150000-0000_watch_history_account_index/down.sql new file mode 100644 index 0000000..3bfd9b7 --- /dev/null +++ b/migrations/2026-07-28-150000-0000_watch_history_account_index/down.sql @@ -0,0 +1 @@ +DROP INDEX IF EXISTS watch_history_account_id; diff --git a/migrations/2026-07-28-150000-0000_watch_history_account_index/up.sql b/migrations/2026-07-28-150000-0000_watch_history_account_index/up.sql new file mode 100644 index 0000000..1dbc284 --- /dev/null +++ b/migrations/2026-07-28-150000-0000_watch_history_account_index/up.sql @@ -0,0 +1,5 @@ +-- watch_history is keyed by (video_id, account_id), so account_id is not the +-- leading column and per-account lookups cannot use the primary key index. +-- The row quota counts rows per account on every write, and clearing or +-- migrating an account's history also filters on it alone. +CREATE INDEX IF NOT EXISTS watch_history_account_id ON watch_history (account_id); diff --git a/src/auth.rs b/src/auth.rs index a8e7886..05d18e7 100644 --- a/src/auth.rs +++ b/src/auth.rs @@ -19,15 +19,23 @@ pub fn bytes_to_hex_string(bytes: &[u8]) -> String { .join("") } +/// How long a freshly minted token stays valid. Should be enough in most cases. +const TOKEN_TTL: Duration = Duration::from_hours(365 * 24); + +/// Allowance for clock skew between minting and verifying. +const EXPIRY_LEEWAY: Duration = Duration::from_secs(5 * 60); + +fn unix_now() -> u64 { + SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap_or_default() + .as_secs() +} + pub fn generate_jwt(account: &Account, secret_key: &[u8]) -> jsonwebtoken::errors::Result { let key = EncodingKey::from_secret(secret_key); - // tokens are valid for one year, should be enough in most cases - let expiration_date = SystemTime::now() - .checked_add(Duration::from_hours(365 * 24)) - .unwrap() - .duration_since(UNIX_EPOCH) - .unwrap() - .as_millis(); + // `exp` is defined in seconds since the epoch, not milliseconds. + let expiration_date = unix_now().saturating_add(TOKEN_TTL.as_secs()); let claims = JwtClaims { sub: account.id.clone(), @@ -36,10 +44,27 @@ pub fn generate_jwt(account: &Account, secret_key: &[u8]) -> jsonwebtoken::error encode(&Header::default(), &claims, &key) } +/// Whether `exp` is close enough to now that this server could have minted it. +/// +/// `Validation::default()` only rejects expiries in the past, with no upper +/// bound. Tokens minted before `exp` was corrected from milliseconds to seconds +/// carry values around the year 59000, so without this check they would stay +/// valid effectively forever and the expiry fix would only apply to new logins. +fn expiry_is_plausible(exp: u64, now: u64) -> bool { + exp <= now + .saturating_add(TOKEN_TTL.as_secs()) + .saturating_add(EXPIRY_LEEWAY.as_secs()) +} + /// Returns the User ID on success. pub fn verify_jwt(encoded_jwt: &str, secret_key: &[u8]) -> jsonwebtoken::errors::Result { let key = DecodingKey::from_secret(secret_key); let claims: JwtClaims = decode(encoded_jwt.as_bytes(), &key, &Validation::default())?.claims; + + if !expiry_is_plausible(claims.exp as u64, unix_now()) { + return Err(jsonwebtoken::errors::ErrorKind::InvalidToken.into()); + } + Ok(claims.sub) } @@ -72,3 +97,38 @@ pub fn hash_accountname(accountname: &str, secret_key: &[u8]) -> String { let result = &mac.finalize().into_bytes(); bytes_to_hex_string(result) } + +#[cfg(test)] +mod tests { + use super::{EXPIRY_LEEWAY, TOKEN_TTL, expiry_is_plausible}; + + const NOW: u64 = 1_800_000_000; + + #[test] + fn normal_expiries_are_accepted() { + assert!(expiry_is_plausible(NOW + 60, NOW)); + assert!(expiry_is_plausible(NOW + TOKEN_TTL.as_secs(), NOW)); + // a little clock skew is tolerated + assert!(expiry_is_plausible( + NOW + TOKEN_TTL.as_secs() + EXPIRY_LEEWAY.as_secs(), + NOW + )); + } + + /// Tokens minted before `exp` was corrected carry milliseconds, which + /// `Validation::default()` would happily accept for the next 57000 years. + #[test] + fn legacy_millisecond_expiries_are_rejected() { + let legacy_exp = (NOW + TOKEN_TTL.as_secs()) * 1000; + assert!(!expiry_is_plausible(legacy_exp, NOW)); + } + + #[test] + fn expiries_beyond_the_ttl_are_rejected() { + assert!(!expiry_is_plausible( + NOW + TOKEN_TTL.as_secs() + EXPIRY_LEEWAY.as_secs() + 1, + NOW + )); + assert!(!expiry_is_plausible(u64::MAX, NOW)); + } +} diff --git a/src/config.rs b/src/config.rs index 30488de..6f09a88 100644 --- a/src/config.rs +++ b/src/config.rs @@ -4,10 +4,52 @@ const fn default_true() -> bool { true } +/// One proxy is the common case when `trust_forwarded_for` is enabled. +const fn default_proxy_hops() -> usize { + 1 +} + +/// Minimum length for any configured secret. +const MIN_SECRET_LENGTH: usize = 32; + +/// Placeholder values shipped in the example configuration and documentation. +/// A server started with one of these has no authentication at all, because +/// anyone can forge tokens for any account. +const PLACEHOLDER_SECRETS: [&str; 5] = [ + "changeme", + "changeit", + "secret", + "someverylongstring64", + "supersecret", +]; + #[derive(serde::Deserialize, Clone)] pub struct Config { #[serde(rename = "secret_key")] pub secret: String, + /// Secret used to derive account name hashes. Kept separate from + /// `secret_key` so that the token signing secret can be rotated without + /// making every existing account unreachable. Falls back to `secret_key` + /// for deployments created before the two were split. + #[serde(default)] + username_secret: Option, + /// Whether to derive the rate limiting client address from + /// `X-Forwarded-For` instead of the immediate peer. + /// + /// Enable this only when the server is reachable exclusively through a + /// reverse proxy. Behind a proxy every request arrives from the proxy's own + /// address, so without this all clients share a single rate limit bucket. + /// If the server is directly reachable, leaving this off is what stops a + /// client from forging its own address. + #[serde(default)] + pub trust_forwarded_for: bool, + /// How many proxies sit in front of this server. + /// + /// Each appends to `X-Forwarded-For`, so with two hops the last entry is the + /// inner proxy's address rather than the client's. Only entries appended by + /// your own proxies can be trusted; anything further left is client-supplied. + #[serde(default = "default_proxy_hops")] + pub trusted_proxy_hops: usize, #[serde(default = "default_true")] pub allow_registration: bool, #[serde(default = "default_true")] @@ -17,10 +59,156 @@ pub struct Config { pub migration_approval: Option, } +impl Config { + /// Secret used to derive account name hashes. + /// + /// Note that changing this value makes all existing accounts unreachable, + /// since accounts are looked up by `HMAC(name)`. Only `secret_key` can be + /// rotated safely, and only once `username_secret` is set explicitly. + pub fn username_secret(&self) -> &str { + self.dedicated_username_secret().unwrap_or(&self.secret) + } + + /// The account name secret, if pinned independently of `secret_key`. + /// + /// A blank value counts as unset, so that an empty `username_secret=` in an + /// env file or a commented-out sample config falls back to `secret_key` + /// rather than failing validation. + fn dedicated_username_secret(&self) -> Option<&str> { + self.username_secret + .as_deref() + .map(str::trim) + .filter(|secret| !secret.is_empty()) + } + + /// Whether the account name secret is pinned independently of `secret_key`. + pub fn has_dedicated_username_secret(&self) -> bool { + self.dedicated_username_secret().is_some() + } +} + +fn validate_secret(name: &str, secret: &str) -> Result<(), ConfigError> { + if PLACEHOLDER_SECRETS.contains(&secret.to_ascii_lowercase().trim()) { + return Err(ConfigError::Message(format!( + "{name} is set to the placeholder value {secret:?}. Generate a unique secret, \ + e.g. with `openssl rand -hex 32`, before exposing this server." + ))); + } + + if secret.len() < MIN_SECRET_LENGTH { + return Err(ConfigError::Message(format!( + "{name} is too short ({} bytes); at least {MIN_SECRET_LENGTH} are required. \ + Generate one with `openssl rand -hex 32`.", + secret.len() + ))); + } + + Ok(()) +} + +fn validate_config(config: &Config) -> Result<(), ConfigError> { + validate_secret("secret_key", &config.secret)?; + if let Some(username_secret) = config.dedicated_username_secret() { + validate_secret("username_secret", username_secret)?; + } + + Ok(()) +} + pub fn build_config() -> Result { - config::Config::builder() + let config: Config = config::Config::builder() .add_source(config::File::with_name("config").required(false)) .add_source(config::Environment::with_convert_case(config::Case::Snake)) .build()? - .try_deserialize() + .try_deserialize()?; + + validate_config(&config)?; + + if !config.has_dedicated_username_secret() { + log::warn!( + "username_secret is not set, so account name hashes are derived from secret_key. \ + This means secret_key cannot be rotated without locking out every account. \ + Set username_secret to the current secret_key value to pin it." + ); + } + + Ok(config) +} + +#[cfg(test)] +mod tests { + use super::{Config, validate_config, validate_secret}; + + fn config_with(secret: &str, username_secret: Option<&str>) -> Config { + Config { + secret: secret.to_owned(), + username_secret: username_secret.map(str::to_owned), + trust_forwarded_for: false, + trusted_proxy_hops: 1, + allow_registration: true, + validate_submitted_metadata: true, + database_url: "./db.sqlite".to_owned(), + migration_approval: None, + } + } + + const STRONG: &str = "6f1c2f0e8a4b5d3c7e9f0a1b2c3d4e5f6a7b8c9d0e1f2a3b4c5d6e7f8a9b0c1d"; + + /// Asserts on the message, not just that validation failed: every + /// placeholder is also shorter than the minimum, so `is_err()` alone would + /// still pass if the placeholder matching were removed entirely. + #[test] + fn placeholder_secrets_are_rejected() { + for placeholder in [ + "changeme", + "ChAnGeMe", + "SomeVeryLongString64", + "SUPERSECRET", + ] { + let error = validate_secret("secret_key", placeholder) + .expect_err("placeholder must be rejected") + .to_string(); + assert!( + error.contains("placeholder"), + "{placeholder:?} was rejected as {error:?}, not as a placeholder" + ); + } + } + + #[test] + fn short_secrets_are_rejected() { + assert!(validate_secret("secret_key", "short").is_err()); + assert!(validate_secret("secret_key", &"a".repeat(31)).is_err()); + assert!(validate_secret("secret_key", &"a".repeat(32)).is_ok()); + } + + #[test] + fn username_secret_falls_back_to_secret_key() { + let config = config_with(STRONG, None); + assert_eq!(config.username_secret(), STRONG); + assert!(!config.has_dedicated_username_secret()); + } + + /// A blank value must fall back rather than fail validation, so that an + /// empty `USERNAME_SECRET=` in an env file is not a startup error. + #[test] + fn blank_username_secret_counts_as_unset() { + for blank in ["", " "] { + let config = config_with(STRONG, Some(blank)); + assert_eq!(config.username_secret(), STRONG); + assert!(!config.has_dedicated_username_secret()); + assert!(validate_config(&config).is_ok()); + } + } + + #[test] + fn username_secret_is_used_and_validated_when_set() { + let other = "0d1c0b9a8f7e6d5c4b3a2f1e0d9c8b7a6f5e4d3c2b1a0f9e8d7c6b5a4f3e2d1c"; + let config = config_with(STRONG, Some(other)); + assert_eq!(config.username_secret(), other); + assert!(config.has_dedicated_username_secret()); + assert!(validate_config(&config).is_ok()); + + assert!(validate_config(&config_with(STRONG, Some("changeme"))).is_err()); + } } diff --git a/src/database.rs b/src/database.rs index d37f2b1..707aa13 100644 --- a/src/database.rs +++ b/src/database.rs @@ -5,6 +5,7 @@ pub mod encrypted_sync; pub mod playlist; pub mod playlist_bookmark; pub mod public_playlist; +pub mod quota; pub mod subscription; pub mod subscription_groups; pub mod video; diff --git a/src/database/quota.rs b/src/database/quota.rs new file mode 100644 index 0000000..52a39e9 --- /dev/null +++ b/src/database/quota.rs @@ -0,0 +1,91 @@ +//! Per-account row quotas for the legacy plaintext tables. +//! +//! The encrypted sync path is bounded by `MAX_ENCRYPTED_SYNC_ACCOUNT_BYTES`, but +//! the plaintext tables had no limit at all, so a single account could grow the +//! database without bound (and sidestep the encrypted quota by using the legacy +//! endpoints instead). + +use diesel::prelude::*; +use diesel_async::RunQueryDsl; + +use crate::DbConnection; +use crate::database::DbError; + +/// Maximum rows one account may hold in any single plaintext table. +/// +/// Chosen to sit far above realistic library sizes while still bounding abuse. +pub const MAX_ROWS_PER_ACCOUNT: i64 = 50_000; + +pub async fn count_subscriptions(conn: &mut DbConnection, owner_id: &str) -> Result { + use crate::schema::subscription::dsl::*; + + subscription + .filter(account_id.eq(owner_id)) + .count() + .get_result(conn) + .await +} + +pub async fn count_watch_history(conn: &mut DbConnection, owner_id: &str) -> Result { + use crate::schema::watch_history::dsl::*; + + watch_history + .filter(account_id.eq(owner_id)) + .count() + .get_result(conn) + .await +} + +pub async fn count_playlist_videos( + conn: &mut DbConnection, + owner_id: &str, +) -> Result { + use crate::schema::playlist_video_member::dsl::*; + + playlist_video_member + .filter(account_id.eq(owner_id)) + .count() + .get_result(conn) + .await +} + +pub async fn count_playback_speeds( + conn: &mut DbConnection, + owner_id: &str, +) -> Result { + use crate::schema::channel_playback_speed::dsl::*; + + channel_playback_speed + .filter(account_id.eq(owner_id)) + .count() + .get_result(conn) + .await +} + +/// Whether storing `incoming` further rows would exceed the per-table quota. +/// +/// `incoming` is an upper bound: some of those rows may be updates to existing +/// ones, so this can reject slightly early at the very top of the range. +pub fn exceeds_row_quota(stored_rows: i64, incoming: usize) -> bool { + let incoming = i64::try_from(incoming).unwrap_or(i64::MAX); + + stored_rows.saturating_add(incoming) > MAX_ROWS_PER_ACCOUNT +} + +#[cfg(test)] +mod tests { + use super::{MAX_ROWS_PER_ACCOUNT, exceeds_row_quota}; + + #[test] + fn quota_allows_up_to_the_limit() { + assert!(!exceeds_row_quota(0, 1)); + assert!(!exceeds_row_quota(MAX_ROWS_PER_ACCOUNT - 1, 1)); + assert!(exceeds_row_quota(MAX_ROWS_PER_ACCOUNT, 1)); + } + + #[test] + fn quota_does_not_overflow_on_absurd_batches() { + assert!(exceeds_row_quota(i64::MAX, usize::MAX)); + assert!(exceeds_row_quota(0, usize::MAX)); + } +} diff --git a/src/handlers.rs b/src/handlers.rs index c46c115..5a24d30 100644 --- a/src/handlers.rs +++ b/src/handlers.rs @@ -62,6 +62,12 @@ pub enum HandlerError { EncryptedSyncConflict, #[error("encrypted sync collection is too large")] EncryptedSyncTooLarge, + #[error("too many items in one request (max {0})")] + BulkRequestTooLarge(usize), + #[error("too many requests; slow down and try again later")] + TooManyRequests, + #[error("account storage quota exceeded")] + StorageQuotaExceeded, #[error("encrypted sync account storage quota exceeded")] EncryptedSyncQuotaExceeded, #[error("this account requires the encrypted sync endpoint")] @@ -93,6 +99,9 @@ impl ResponseError for HandlerError { Self::ValidationErrorWithContext(_) => StatusCode::BAD_REQUEST, Self::EncryptedSyncConflict => StatusCode::CONFLICT, Self::EncryptedSyncTooLarge => StatusCode::PAYLOAD_TOO_LARGE, + Self::BulkRequestTooLarge(_) => StatusCode::PAYLOAD_TOO_LARGE, + Self::TooManyRequests => StatusCode::TOO_MANY_REQUESTS, + Self::StorageQuotaExceeded => StatusCode::PAYLOAD_TOO_LARGE, Self::EncryptedSyncQuotaExceeded => StatusCode::PAYLOAD_TOO_LARGE, Self::EncryptedSyncRequired => StatusCode::CONFLICT, Self::YouTubeConnectError => StatusCode::INTERNAL_SERVER_ERROR, @@ -102,6 +111,44 @@ impl ResponseError for HandlerError { pub type HandlerResult = Result; +/// Upper bound on how many items one bulk request may carry. +/// +/// Validation can issue a YouTube round-trip per distinct channel, so an +/// unbounded batch lets a single request occupy a worker for an unbounded time. +pub const MAX_BULK_ITEMS: usize = 1000; + +pub fn check_bulk_size(len: usize) -> HandlerResult<()> { + if len > MAX_BULK_ITEMS { + return Err(HandlerError::BulkRequestTooLarge(MAX_BULK_ITEMS)); + } + + Ok(()) +} + +/// Fail fast when an account is *already* over its quota. +/// +/// Deliberately does not charge for the incoming batch. All the bulk write paths +/// are upserts, so a client re-syncing data it already stored adds no rows; +/// charging the batch up front would reject those re-syncs even though they +/// change nothing, and would do so precisely when an account is near its limit. +/// [`check_stored_rows`] does the authoritative check after the writes. +pub fn check_already_over_quota(stored_rows: i64) -> HandlerResult<()> { + check_stored_rows(stored_rows) +} + +/// Authoritative quota check, for use after the writes inside a transaction. +/// +/// Counting the rows that actually exist is exact regardless of how many of the +/// incoming entries were updates rather than inserts. Returning an error from +/// inside the transaction rolls the writes back. +pub fn check_stored_rows(stored_rows: i64) -> HandlerResult<()> { + if crate::database::quota::exceeds_row_quota(stored_rows, 0) { + return Err(HandlerError::StorageQuotaExceeded); + } + + Ok(()) +} + impl From for HandlerError { fn from(error: diesel::result::Error) -> Self { Self::InternalDatabaseErrorWithContext(error.to_string()) diff --git a/src/handlers/channel_playback_speeds.rs b/src/handlers/channel_playback_speeds.rs index 6ea62b0..7348b00 100644 --- a/src/handlers/channel_playback_speeds.rs +++ b/src/handlers/channel_playback_speeds.rs @@ -1,14 +1,20 @@ use actix_web::{HttpResponse, Responder, delete, get, middleware::from_fn, put, web}; +use diesel_async::{AsyncConnection, scoped_futures::ScopedFutureExt}; use utoipa_actix_web::scope; use crate::{ - WebData, - database::channel_playback_speed::{ - get_channel_playback_speeds_by_account_id, remove_channel_playback_speed, - set_channel_playback_speed, + DbConnection, WebData, + database::{ + channel_playback_speed::{ + get_channel_playback_speeds_by_account_id, remove_channel_playback_speed, + set_channel_playback_speed, + }, + quota::count_playback_speeds, }, get_db_conn, - handlers::{HandlerError, HandlerResult, ScopedHandler, user::auth_middleware}, + handlers::{ + HandlerError, HandlerResult, ScopedHandler, check_stored_rows, user::auth_middleware, + }, models::{Account, ChannelPlaybackSpeed}, }; @@ -63,13 +69,40 @@ async fn put_channel_playback_speed( speed.account_id = account.id; let mut conn = get_db_conn!(pool); - set_channel_playback_speed(&mut conn, &speed) - .await - .map_err(|_| HandlerError::InternalDatabaseError)?; + store_playback_speed(&mut conn, &speed).await?; Ok(HttpResponse::Ok().json(speed)) } +/// Store a playback speed, enforcing the row quota in the same transaction. +/// +/// Checking and writing separately would let two concurrent requests both pass +/// the check and then both insert, pushing the account past the quota. +async fn store_playback_speed( + conn: &mut DbConnection, + speed: &ChannelPlaybackSpeed, +) -> HandlerResult<()> { + conn.transaction::<_, HandlerError, _>(|conn| { + async move { + set_channel_playback_speed(conn, speed) + .await + .map_err(|_| HandlerError::InternalDatabaseError)?; + + // Authoritative check on the rows that now exist. This is an upsert, + // so overwriting an existing speed must not be charged as a new row; + // an error here rolls the write back. + check_stored_rows( + count_playback_speeds(conn, &speed.account_id) + .await + .map_err(|_| HandlerError::InternalDatabaseError)?, + )?; + Ok(()) + } + .scope_boxed() + }) + .await +} + #[utoipa::path(responses((status = OK)), security(("api_jwt_token" = [])))] #[delete("/{channel_id}")] async fn delete_channel_playback_speed( diff --git a/src/handlers/encrypted_sync.rs b/src/handlers/encrypted_sync.rs index 6a7e4f4..89b4e5d 100644 --- a/src/handlers/encrypted_sync.rs +++ b/src/handlers/encrypted_sync.rs @@ -9,7 +9,7 @@ use crate::dto::{ EncryptedSyncCollectionResponse, EncryptedSyncCollectionRevision, EncryptedSyncManifest, PutEncryptedSync, SyncCapabilities, }; -use crate::handlers::user::auth_middleware; +use crate::handlers::user::{PlaintextSyncExempt, auth_middleware}; use crate::handlers::{HandlerError, HandlerResult, ScopedHandler}; use crate::models::Account; use crate::{WebData, get_db_conn}; @@ -41,6 +41,7 @@ impl ScopedHandler for EncryptedSyncHandler { scope::scope("").service( scope::scope("/encrypted_sync") .app_data(web::JsonConfig::default().limit(MAX_ENCRYPTED_SYNC_BYTES + 1024)) + .app_data(web::Data::new(PlaintextSyncExempt)) .wrap(actix_web::middleware::from_fn(auth_middleware)) .service(get_encrypted_sync_manifest) .service(get_legacy_encrypted_sync) diff --git a/src/handlers/playlists.rs b/src/handlers/playlists.rs index 5402cdf..56bf944 100644 --- a/src/handlers/playlists.rs +++ b/src/handlers/playlists.rs @@ -1,4 +1,5 @@ use actix_web::{HttpResponse, Responder, delete, get, middleware::from_fn, patch, post, web}; +use diesel_async::{AsyncConnection, scoped_futures::ScopedFutureExt}; use itertools::Itertools; use utoipa_actix_web::scope; @@ -11,12 +12,16 @@ use crate::{ get_playlist_by_id_with_videos, get_playlist_video_count, get_playlists_by_account_id, remove_video_from_playlist, update_existing_playlist, }, + quota::count_playlist_videos, }, dto::{CreatePlaylist, CreateVideo, ExtendedPlaylist, PlaylistResponse}, get_db_conn, - handlers::{HandlerError, HandlerResult, ScopedHandler, user::auth_middleware}, - models::{Account, Playlist}, - validation::validate_video_information_if_changed, + handlers::{ + HandlerError, HandlerResult, ScopedHandler, check_already_over_quota, check_bulk_size, + check_stored_rows, user::auth_middleware, + }, + models::{Account, Channel, Playlist}, + validation::{validate_videos_against_youtube, videos_requiring_validation}, }; pub struct PlaylistsHandler {} @@ -203,33 +208,69 @@ async fn add_to_playlist( playlist_id: web::Path, video_datas: web::Json>, ) -> HandlerResult { - let mut conn = get_db_conn!(pool); - - get_owned_playlist_or_error(&mut conn, &playlist_id, &account.id).await?; - let video_datas = video_datas.into_inner(); - let videos_grouped_by_uploader = video_datas - .iter() - .sorted_by(|a, b| Ord::cmp(&a.uploader.id, &b.uploader.id)) - .chunk_by(|video| video.uploader.clone()); - - for (mut channel, videos) in &videos_grouped_by_uploader { - let mut videos: Vec<_> = videos.cloned().collect(); - - validate_video_information_if_changed(&mut conn, &mut videos, &mut channel).await?; + check_bulk_size(video_datas.len())?; - // store channel information first before storing video to ensure data integrity - create_or_update_channel(&mut conn, &channel) - .await - .map_err(|_| HandlerError::InternalDatabaseError)?; + // one RSS fetch per distinct uploader rather than per video + let mut groups: Vec<(Channel, Vec)> = video_datas + .into_iter() + .sorted_by(|a, b| Ord::cmp(&a.uploader.id, &b.uploader.id)) + .chunk_by(|video| video.uploader.clone()) + .into_iter() + .map(|(channel, videos)| (channel, videos.collect())) + .collect(); - for video in videos { - add_video_to_playlist(&mut conn, &playlist_id, &account.id, &(&video).into()) + // Decide what needs validating, then release the connection before the + // network round-trips so a slow batch cannot hold one for minutes. + let mut needs_validation = Vec::with_capacity(groups.len()); + { + let mut conn = get_db_conn!(pool); + get_owned_playlist_or_error(&mut conn, &playlist_id, &account.id).await?; + check_already_over_quota( + count_playlist_videos(&mut conn, &account.id) .await - .map_err(|_| HandlerError::InternalDatabaseError)?; + .map_err(|_| HandlerError::InternalDatabaseError)?, + )?; + for (_, videos) in &groups { + needs_validation.push(videos_requiring_validation(&mut conn, videos).await); } } + for ((channel, videos), needs_validation) in groups.iter_mut().zip(&needs_validation) { + validate_videos_against_youtube(videos, needs_validation, channel).await?; + } + + let mut conn = get_db_conn!(pool); + conn.transaction::<_, HandlerError, _>(|conn| { + let (groups, playlist_id, account_id) = (&groups, &*playlist_id, &account.id); + async move { + for (channel, videos) in groups { + // store channel information first before storing video to ensure data integrity + create_or_update_channel(conn, channel) + .await + .map_err(|_| HandlerError::InternalDatabaseError)?; + + for video in videos { + add_video_to_playlist(conn, playlist_id, account_id, &video.into()) + .await + .map_err(|_| HandlerError::InternalDatabaseError)?; + } + } + + // Authoritative check on the rows that now exist. Playlist members are + // upserts, so counting afterwards is the only way to charge for what + // was actually added; an error here rolls the batch back. + check_stored_rows( + count_playlist_videos(conn, account_id) + .await + .map_err(|_| HandlerError::InternalDatabaseError)?, + )?; + Ok(()) + } + .scope_boxed() + }) + .await?; + Ok(HttpResponse::Created()) } diff --git a/src/handlers/subscriptions.rs b/src/handlers/subscriptions.rs index 9cab3f1..0fdd9b1 100644 --- a/src/handlers/subscriptions.rs +++ b/src/handlers/subscriptions.rs @@ -5,6 +5,7 @@ use utoipa_actix_web::scope; use crate::{ DbConnection, WebData, database::{ + quota::count_subscriptions, subscription::{ add_subscription_by_account_id, get_subscription_channel_by_account_id, get_subscriptions_by_account_id, remove_subscription_by_account_id, @@ -19,9 +20,12 @@ use crate::{ }, dto::ExtendedSubscriptionGroup, get_db_conn, - handlers::{HandlerError, HandlerResult, ScopedHandler, user::auth_middleware}, + handlers::{ + HandlerError, HandlerResult, ScopedHandler, check_already_over_quota, check_bulk_size, + check_stored_rows, user::auth_middleware, + }, models::{Account, Channel, SubscriptionGroup}, - validation::validate_channel_information_if_changed, + validation::{channels_requiring_validation, validate_channel_against_youtube}, }; pub struct SubscriptionsHandler {} @@ -74,18 +78,39 @@ async fn subscribe_bulk( pool: WebData, channels: web::Json>, ) -> HandlerResult { - let mut conn = get_db_conn!(pool); let mut channels = channels.into_inner(); - - for channel in &mut channels { - validate_channel_information_if_changed(&mut conn, channel).await?; + check_bulk_size(channels.len())?; + + // Work out what needs validating, then release the connection before the + // network round-trips so that a slow batch cannot hold one for minutes. + let pending = { + let mut conn = get_db_conn!(pool); + check_already_over_quota( + count_subscriptions(&mut conn, &account.id) + .await + .map_err(|_| HandlerError::InternalDatabaseError)?, + )?; + channels_requiring_validation(&mut conn, &channels).await + }; + for index in pending { + validate_channel_against_youtube(&mut channels[index]).await?; } + let mut conn = get_db_conn!(pool); conn.transaction::<_, HandlerError, _>(|conn| { async move { for channel in &channels { add_subscription_by_account_id(conn, channel, &account.id).await?; } + + // Authoritative check on the rows that now exist. Subscriptions are + // upserts, so counting afterwards is the only way to charge for what + // was actually added; an error here rolls the batch back. + check_stored_rows( + count_subscriptions(conn, &account.id) + .await + .map_err(|_| HandlerError::InternalDatabaseError)?, + )?; Ok(()) } .scope_boxed() @@ -122,12 +147,26 @@ async fn subscribe( pool: WebData, channel: web::Json, ) -> HandlerResult { - let mut conn = get_db_conn!(pool); - let mut channel = channel.into_inner(); - // verify that the provided information is valid - validate_channel_information_if_changed(&mut conn, &mut channel).await?; + // verify that the provided information is valid, without holding a + // connection across the YouTube round-trip + let needs_validation = { + let mut conn = get_db_conn!(pool); + check_already_over_quota( + count_subscriptions(&mut conn, &account.id) + .await + .map_err(|_| HandlerError::InternalDatabaseError)?, + )?; + !channels_requiring_validation(&mut conn, std::slice::from_ref(&channel)) + .await + .is_empty() + }; + if needs_validation { + validate_channel_against_youtube(&mut channel).await?; + } + + let mut conn = get_db_conn!(pool); match add_subscription_by_account_id(&mut conn, &channel, &account.id).await { Ok(_) => Ok(HttpResponse::Ok()), Err(err) => Err(HandlerError::InternalDatabaseErrorWithContext( diff --git a/src/handlers/user.rs b/src/handlers/user.rs index 63dc3f2..712283b 100644 --- a/src/handlers/user.rs +++ b/src/handlers/user.rs @@ -1,3 +1,5 @@ +use std::net::{IpAddr, SocketAddr}; + use actix_web::body::MessageBody; use actix_web::dev::{ServiceFactory, ServiceRequest, ServiceResponse}; use actix_web::middleware::Next; @@ -14,11 +16,20 @@ use crate::database::encrypted_sync; use crate::dto::LoginResponse; use crate::handlers::{HandlerError, HandlerResult, ScopedHandler}; use crate::models::Account; +use crate::rate_limit::RateLimiter; use crate::{CONFIG, WebData, dto, get_db_conn, models}; const AUTH_HEADER_KEY: &str = "Authorization"; const MIN_PASSWORD_LENGTH: usize = 8; +/// Marker registered on scopes that remain reachable after an account has +/// switched to encrypted sync. +/// +/// Everything else is refused for such accounts, so that an older client cannot +/// repopulate readable plaintext data. Add this only to scopes that either store +/// no plaintext sync data or are needed to manage the account itself. +pub struct PlaintextSyncExempt; + pub struct UserHandler {} impl ScopedHandler for UserHandler { fn get_service() -> scope::Scope< @@ -30,12 +41,21 @@ impl ScopedHandler for UserHandler { Error = actix_web::Error, >, > { + // register and login are reachable without credentials, so the whole + // scope is rate limited to blunt password guessing and registration spam. + // Note this must wrap the outer scope: an extra nested `scope("")` would + // match the prefix of, and therefore swallow, its sibling's routes. + // The limiter itself is registered once on the App, not here: this + // function runs per worker, so building it here would give every worker + // its own counters and multiply the effective limit by the worker count. scope::scope("/account") + .wrap(actix_web::middleware::from_fn(rate_limit_middleware)) .service(register_account) .service(login_account) // services that require auth start here .service( scope::scope("") + .app_data(web::Data::new(PlaintextSyncExempt)) .wrap(actix_web::middleware::from_fn(auth_middleware)) .service(delete_account), ) @@ -61,7 +81,7 @@ async fn register_account( let account = models::Account { id: Uuid::now_v7().to_string(), - name_hash: hash_accountname(&form.name, CONFIG.secret.as_bytes()), + name_hash: hash_accountname(&form.name, CONFIG.username_secret().as_bytes()), password_hash: hash_password(&form.password), }; @@ -93,7 +113,7 @@ async fn login_account( ) -> HandlerResult { let mut conn = get_db_conn!(pool); - let name = hash_accountname(&form.name, CONFIG.secret.as_bytes()); + let name = hash_accountname(&form.name, CONFIG.username_secret().as_bytes()); let Some(account) = find_account_by_name_hash(&mut conn, &name) .await .ok() @@ -138,6 +158,73 @@ async fn delete_account( } } +/// Middleware that rate limits unauthenticated endpoints per client address. +/// +/// Defaults to the peer address, since forwarded headers are client-controlled +/// when the server is directly reachable. Behind a reverse proxy the peer is the +/// proxy itself, which would put every client in one bucket, so +/// `trust_forwarded_for` switches to the forwarded address instead. +pub async fn rate_limit_middleware( + req: ServiceRequest, + next: Next, +) -> Result, actix_web::Error> { + let limiter: Option<&web::Data> = req.app_data(); + + if let Some(limiter) = limiter + && let Some(client) = rate_limit_client( + &req, + limiter.trusts_forwarded_for(), + limiter.trusted_proxy_hops(), + ) + && !limiter.check(client) + { + return Err(HandlerError::TooManyRequests.into()); + } + + next.call(req).await +} + +/// Address to rate limit a request against. +fn rate_limit_client( + req: &ServiceRequest, + trust_forwarded_for: bool, + trusted_proxy_hops: usize, +) -> Option { + let peer = req.peer_addr().map(|addr| addr.ip()); + + if !trust_forwarded_for { + return peer; + } + + req.headers() + .get("X-Forwarded-For") + .and_then(|value| value.to_str().ok()) + .and_then(|header| forwarded_client(header, trusted_proxy_hops)) + .or(peer) +} + +/// Client address from an `X-Forwarded-For` value. +/// +/// Counts back `trusted_proxy_hops` entries from the end, because each proxy +/// appends the address it observed. With one proxy that is the last entry; with +/// two it is the second to last, since the last is then the inner proxy's own +/// address. Entries further left are whatever the client sent, so counting from +/// the front would let a client pick its own bucket and bypass the limit. +fn forwarded_client(header: &str, trusted_proxy_hops: usize) -> Option { + header + .rsplit(',') + .map(str::trim) + .filter(|entry| !entry.is_empty()) + .nth(trusted_proxy_hops.saturating_sub(1)) + .and_then(|entry| { + entry + .parse::() + .ok() + // tolerate `host:port` forms + .or_else(|| entry.parse::().ok().map(|addr| addr.ip())) + }) +} + /// Middleware that ensures that the account is authenticated. pub async fn auth_middleware( req: ServiceRequest, @@ -170,9 +257,10 @@ pub async fn auth_middleware( return Err(HandlerError::AccountNotExists.into()); }; - let encrypted_endpoint = req.path().contains("/encrypted_sync"); - let account_deletion = req.path().ends_with("/account/delete"); - if !encrypted_endpoint && !account_deletion { + // Scopes opt out explicitly. Matching on the request path instead would let + // a route parameter such as `/playlists/encrypted_sync/videos` slip past. + let exempt = req.app_data::>().is_some(); + if !exempt { let encrypted_sync_enabled = encrypted_sync::exists(&mut conn, &account.id) .await .map_err(|_| HandlerError::InternalDatabaseError)?; @@ -195,3 +283,281 @@ pub async fn auth_middleware( next.call(req).await } + +#[cfg(test)] +mod tests { + use std::net::SocketAddr; + use std::time::Duration; + + use actix_web::{App, HttpResponse, Responder, get, http::StatusCode, test, web}; + + use super::{PlaintextSyncExempt, rate_limit_middleware}; + use crate::rate_limit::RateLimiter; + + // Nested so that `actix_web::test` imported above does not shadow the + // built-in `#[test]` attribute for these synchronous tests. + mod forwarded { + use crate::handlers::user::forwarded_client; + + fn resolved(header: &str) -> Option { + resolved_hops(header, 1) + } + + fn resolved_hops(header: &str, hops: usize) -> Option { + forwarded_client(header, hops).map(|ip| ip.to_string()) + } + + /// With two proxies the last entry is the inner proxy, not the client. + #[test] + fn forwarded_client_honours_the_hop_count() { + let header = "203.0.113.5, 10.0.0.7"; + assert_eq!(resolved_hops(header, 1).as_deref(), Some("10.0.0.7")); + assert_eq!(resolved_hops(header, 2).as_deref(), Some("203.0.113.5")); + // more hops than entries yields nothing, so the caller falls back + assert_eq!(resolved_hops(header, 3), None); + // zero is treated as one rather than panicking + assert_eq!(resolved_hops(header, 0).as_deref(), Some("10.0.0.7")); + } + + /// The last entry is the one a proxy appended; earlier entries are + /// whatever the client sent and must not be trusted. + #[test] + fn forwarded_client_uses_the_last_entry() { + assert_eq!(resolved("203.0.113.5").as_deref(), Some("203.0.113.5")); + // a client-supplied spoof followed by the address the proxy saw + assert_eq!( + resolved("1.1.1.1, 203.0.113.5").as_deref(), + Some("203.0.113.5") + ); + assert_eq!( + resolved("1.1.1.1, 203.0.113.5:44321").as_deref(), + Some("203.0.113.5") + ); + } + + #[test] + fn forwarded_client_ignores_unusable_values() { + assert_eq!(resolved(""), None); + assert_eq!(resolved("not-an-address"), None); + assert_eq!(resolved("1.1.1.1, "), Some("1.1.1.1".to_owned())); + } + } + + #[get("/ping")] + async fn ping() -> impl Responder { + HttpResponse::Ok() + } + + #[actix_web::post("/register")] + async fn stub_register() -> impl Responder { + HttpResponse::Created() + } + + #[actix_web::delete("/delete")] + async fn stub_delete() -> impl Responder { + HttpResponse::NoContent() + } + + /// Mirrors the real `/account` layout. An extra nested `scope("")` around + /// register/login would match the prefix of its sibling and swallow + /// `/account/delete`, so both routes must stay reachable. + #[actix_rt::test] + async fn both_account_routes_stay_routable() { + let app = test::init_service( + App::new().service( + web::scope("/account") + .app_data(web::Data::new(RateLimiter::default())) + .wrap(actix_web::middleware::from_fn(rate_limit_middleware)) + .service(stub_register) + .service(web::scope("").service(stub_delete)), + ), + ) + .await; + + let register = test::TestRequest::post() + .uri("/account/register") + .peer_addr("203.0.113.9:5000".parse().unwrap()) + .to_request(); + assert_eq!( + test::call_service(&app, register).await.status(), + StatusCode::CREATED + ); + + let delete = test::TestRequest::delete() + .uri("/account/delete") + .peer_addr("203.0.113.9:5000".parse().unwrap()) + .to_request(); + assert_eq!( + test::call_service(&app, delete).await.status(), + StatusCode::NO_CONTENT + ); + } + + /// Mirrors how `auth_middleware` decides whether a scope is exempt from the + /// encrypted-sync requirement, without needing a database. + async fn exemption_probe( + req: actix_web::dev::ServiceRequest, + next: actix_web::middleware::Next, + ) -> Result, actix_web::Error> + { + if req.app_data::>().is_none() { + return Err(crate::handlers::HandlerError::EncryptedSyncRequired.into()); + } + + next.call(req).await + } + + /// A playlist (or video) named `encrypted_sync` used to make the old + /// `path().contains("/encrypted_sync")` check exempt a plaintext route. + #[actix_rt::test] + async fn plaintext_routes_are_not_exempted_by_their_path() { + let app = test::init_service( + App::new() + .service( + web::scope("/encrypted_sync") + .app_data(web::Data::new(PlaintextSyncExempt)) + .wrap(actix_web::middleware::from_fn(exemption_probe)) + .service(ping), + ) + .service( + web::scope("/playlists") + .wrap(actix_web::middleware::from_fn(exemption_probe)) + .service(ping), + ), + ) + .await; + + let status = |uri: &'static str| async { + let req = test::TestRequest::get().uri(uri).to_request(); + match test::try_call_service(&app, req).await { + Ok(response) => response.status(), + Err(error) => error.as_response_error().status_code(), + } + }; + + // the genuinely encrypted scope stays reachable + assert_eq!(status("/encrypted_sync/ping").await, StatusCode::OK); + // a plaintext route is refused even though its path contains the marker word + assert_eq!( + status("/playlists/encrypted_sync/ping").await, + StatusCode::CONFLICT + ); + } + + /// Behind a proxy every request shares one peer address, so distinct + /// forwarded clients must get their own budgets once the header is trusted. + #[actix_rt::test] + async fn forwarded_clients_are_bucketed_separately() { + let app = test::init_service( + App::new().service( + web::scope("/t") + .app_data(web::Data::new( + RateLimiter::new(1, Duration::from_secs(3600)).trusting_forwarded_for(true), + )) + .wrap(actix_web::middleware::from_fn(rate_limit_middleware)) + .service(ping), + ), + ) + .await; + + // same proxy peer for every request, different real clients + let proxy: SocketAddr = "127.0.0.1:9000".parse().unwrap(); + let expected = [ + ("203.0.113.1", StatusCode::OK), + // a different client is not punished for the first one's request + ("203.0.113.2", StatusCode::OK), + // but the first client's own budget is spent + ("203.0.113.1", StatusCode::TOO_MANY_REQUESTS), + ]; + + for (client, want) in expected { + let req = test::TestRequest::get() + .uri("/t/ping") + .peer_addr(proxy) + .insert_header(("X-Forwarded-For", client)) + .to_request(); + let got = match test::try_call_service(&app, req).await { + Ok(response) => response.status(), + Err(error) => error.as_response_error().status_code(), + }; + assert_eq!(got, want, "client {client}"); + } + } + + /// Mirrors production wiring: the limiter is registered on the App, and the + /// middleware on a scope. A limiter built inside the per-worker closure + /// instead would give each worker its own counters. + #[actix_rt::test] + async fn app_level_limiter_is_visible_to_scope_middleware() { + let app = test::init_service( + App::new() + .app_data(web::Data::new(RateLimiter::new( + 1, + Duration::from_secs(3600), + ))) + .service( + web::scope("/t") + .wrap(actix_web::middleware::from_fn(rate_limit_middleware)) + .service(ping), + ), + ) + .await; + + let peer: SocketAddr = "203.0.113.20:5000".parse().unwrap(); + let call = |uri: &'static str| { + test::TestRequest::get() + .uri(uri) + .peer_addr(peer) + .to_request() + }; + + let first = match test::try_call_service(&app, call("/t/ping")).await { + Ok(response) => response.status(), + Err(error) => error.as_response_error().status_code(), + }; + let second = match test::try_call_service(&app, call("/t/ping")).await { + Ok(response) => response.status(), + Err(error) => error.as_response_error().status_code(), + }; + + assert_eq!(first, StatusCode::OK); + assert_eq!(second, StatusCode::TOO_MANY_REQUESTS); + } + + /// Guards against the middleware silently failing open, which would happen + /// if scope-level `app_data` were not visible to scope-level middleware. + #[actix_rt::test] + async fn rate_limit_middleware_rejects_a_burst() { + let app = test::init_service( + App::new().service( + web::scope("/t") + .app_data(web::Data::new(RateLimiter::new( + 2, + Duration::from_secs(3600), + ))) + .wrap(actix_web::middleware::from_fn(rate_limit_middleware)) + .service(ping), + ), + ) + .await; + + let peer: SocketAddr = "203.0.113.7:5000".parse().unwrap(); + let mut statuses = Vec::new(); + for _ in 0..3 { + let req = test::TestRequest::get() + .uri("/t/ping") + .peer_addr(peer) + .to_request(); + // the middleware signals rejection with an error, which actix renders + // through `ResponseError` in a real server + statuses.push(match test::try_call_service(&app, req).await { + Ok(response) => response.status(), + Err(error) => error.as_response_error().status_code(), + }); + } + + assert_eq!(statuses[0], StatusCode::OK); + assert_eq!(statuses[1], StatusCode::OK); + assert_eq!(statuses[2], StatusCode::TOO_MANY_REQUESTS); + } +} diff --git a/src/handlers/watch_history.rs b/src/handlers/watch_history.rs index b0a801e..599511b 100644 --- a/src/handlers/watch_history.rs +++ b/src/handlers/watch_history.rs @@ -7,6 +7,7 @@ use crate::{ WebData, database::{ channel::create_or_update_channel, + quota::count_watch_history, video::create_or_update_video, watch_history::{ DEFAULT_PAGE_SIZE, MAX_PAGE_SIZE, add_or_update_video_to_watch_history, @@ -16,9 +17,12 @@ use crate::{ }, dto::{CreateVideo, ExtendedWatchHistoryItem, WatchedState}, get_db_conn, - handlers::{HandlerError, HandlerResult, ScopedHandler, user::auth_middleware}, - models::{Account, WatchHistoryItem}, - validation::validate_video_information_if_changed_single, + handlers::{ + HandlerError, HandlerResult, ScopedHandler, check_already_over_quota, check_bulk_size, + check_stored_rows, user::auth_middleware, + }, + models::{Account, Channel, WatchHistoryItem}, + validation::{validate_videos_against_youtube, videos_requiring_validation}, }; pub struct WatchHistoryHandler; @@ -44,16 +48,67 @@ impl ScopedHandler for WatchHistoryHandler { } } -async fn prepare_watch_history_item( - conn: &mut crate::DbConnection, +/// Stamp ownership onto a batch of items and validate their videos. +/// +/// Validation is grouped by uploader, so a batch costs one YouTube round-trip +/// per distinct channel rather than one per item, and no pooled connection is +/// held while those round-trips happen. +async fn prepare_watch_history_items( + pool: &WebData, account_id: &str, - mut item: ExtendedWatchHistoryItem, -) -> HandlerResult { - item.metadata.account_id = account_id.to_string(); - item.metadata.video_id = item.video.id.clone(); + mut items: Vec, +) -> HandlerResult> { + for item in &mut items { + item.metadata.account_id = account_id.to_string(); + item.metadata.video_id = item.video.id.clone(); + } + + // group item indices by uploader, preserving the caller's ordering + let mut groups: Vec<(Channel, Vec)> = Vec::new(); + for (index, item) in items.iter().enumerate() { + match groups + .iter_mut() + .find(|(channel, _)| *channel == item.video.uploader) + { + Some((_, indices)) => indices.push(index), + None => groups.push((item.video.uploader.clone(), vec![index])), + } + } + + let mut grouped_videos: Vec> = groups + .iter() + .map(|(_, indices)| indices.iter().map(|i| items[*i].video.clone()).collect()) + .collect(); + + let mut needs_validation = Vec::with_capacity(groups.len()); + { + let mut conn = get_db_conn!(pool); + check_already_over_quota( + count_watch_history(&mut conn, account_id) + .await + .map_err(|_| HandlerError::InternalDatabaseError)?, + )?; + for videos in &grouped_videos { + needs_validation.push(videos_requiring_validation(&mut conn, videos).await); + } + } + + for (((channel, _), videos), needs_validation) in groups + .iter_mut() + .zip(grouped_videos.iter_mut()) + .zip(&needs_validation) + { + validate_videos_against_youtube(videos, needs_validation, channel).await?; + } + + // write the validated metadata back onto the items + for ((_, indices), videos) in groups.iter().zip(&grouped_videos) { + for (index, video) in indices.iter().zip(videos) { + items[*index].video = video.clone(); + } + } - validate_video_information_if_changed_single(conn, &mut item.video).await?; - Ok(item) + Ok(items) } async fn persist_watch_history_item( @@ -73,14 +128,43 @@ async fn persist_watch_history_item( Ok(()) } -async fn store_watch_history_item( +async fn persist_watch_history_items( conn: &mut crate::DbConnection, account_id: &str, - item: ExtendedWatchHistoryItem, -) -> HandlerResult { - let item = prepare_watch_history_item(conn, account_id, item).await?; - persist_watch_history_item(conn, &item).await?; - Ok(item) + items: &[ExtendedWatchHistoryItem], +) -> HandlerResult<()> { + conn.transaction::<_, HandlerError, _>(|conn| { + async move { + for item in items { + persist_watch_history_item(conn, item).await?; + } + + // Authoritative check on the rows that now exist. History entries are + // upserts, so counting afterwards is the only way to charge for what + // was actually added; an error here rolls the batch back. + check_stored_rows( + count_watch_history(conn, account_id) + .await + .map_err(|_| HandlerError::InternalDatabaseError)?, + )?; + Ok(()) + } + .scope_boxed() + }) + .await +} + +async fn store_watch_history_items( + pool: &WebData, + account_id: &str, + items: Vec, +) -> HandlerResult> { + let items = prepare_watch_history_items(pool, account_id, items).await?; + + let mut conn = get_db_conn!(pool); + persist_watch_history_items(&mut conn, account_id, &items).await?; + + Ok(items) } #[utoipa::path(responses((status = OK)), security(("api_jwt_token" = [])))] @@ -90,24 +174,10 @@ async fn add_to_watch_history_bulk( pool: WebData, items: web::Json>, ) -> HandlerResult { - let mut conn = get_db_conn!(pool); let items = items.into_inner(); - let mut prepared_items = Vec::with_capacity(items.len()); - - for item in items { - prepared_items.push(prepare_watch_history_item(&mut conn, &account.id, item).await?); - } + check_bulk_size(items.len())?; - conn.transaction::<_, HandlerError, _>(|conn| { - async move { - for item in &prepared_items { - persist_watch_history_item(conn, item).await?; - } - Ok(()) - } - .scope_boxed() - }) - .await?; + store_watch_history_items(&pool, &account.id, items).await?; Ok(HttpResponse::Ok()) } @@ -199,12 +269,11 @@ async fn add_to_watch_history( pool: WebData, watch_history_item: web::Json, ) -> HandlerResult { - let mut conn = get_db_conn!(pool); - - let watch_history_item = - store_watch_history_item(&mut conn, &account.id, watch_history_item.into_inner()).await?; + let mut stored = + store_watch_history_items(&pool, &account.id, vec![watch_history_item.into_inner()]) + .await?; - Ok(HttpResponse::Ok().json(watch_history_item)) + Ok(HttpResponse::Ok().json(stored.pop())) } #[utoipa::path(responses((status = OK)), security(("api_jwt_token" = [])))] diff --git a/src/main.rs b/src/main.rs index 9f6abeb..2d62931 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1,7 +1,11 @@ #[macro_use] extern crate diesel; -use std::{io, path::Path, sync::LazyLock}; +use std::{io, sync::LazyLock}; + +// only used by the sqlite pre-migration backup +#[cfg(feature = "sqlite")] +use std::path::Path; use actix_web::{App, HttpServer, middleware, web}; #[cfg(feature = "sqlite")] @@ -34,6 +38,7 @@ mod dto; mod handlers; mod models; mod openapi; +mod rate_limit; mod schema; mod validation; @@ -79,11 +84,21 @@ async fn main() -> io::Result<()> { log::info!("starting HTTP server at http://localhost:8080"); + // Built once and shared, because the closure below runs per worker. Building + // it inside would give each worker separate counters, multiplying the + // effective rate limit by the number of workers. + let rate_limiter = web::Data::new( + rate_limit::RateLimiter::default() + .trusting_forwarded_for(CONFIG.trust_forwarded_for) + .with_trusted_proxy_hops(CONFIG.trusted_proxy_hops), + ); + HttpServer::new(move || { let (app, generated_api) = App::new() .into_utoipa_app() // add DB pool handle to app data; enables use of `web::Data` extractor .app_data(web::Data::new(pool.clone())) + .app_data(rate_limiter.clone()) .service( utoipa_actix_web::scope("/v1") .service(UserHandler::get_service()) @@ -121,8 +136,11 @@ async fn initialize_db_pool(db_url: &str) -> Result { let url = url.to_owned(); Box::pin(async move { let mut conn = DbConnection::establish(&url).await?; + // `foreign_keys` defaults to OFF in SQLite and must be enabled per + // connection, otherwise none of the ON DELETE CASCADE constraints + // fire and deleting an account orphans all of its rows. conn.batch_execute( - "PRAGMA journal_mode = WAL; PRAGMA busy_timeout = 30000; PRAGMA synchronous = NORMAL;", + "PRAGMA journal_mode = WAL; PRAGMA busy_timeout = 30000; PRAGMA synchronous = NORMAL; PRAGMA foreign_keys = ON;", ) .await .map_err(|err| diesel::ConnectionError::BadConnection(err.to_string()))?; @@ -179,7 +197,16 @@ fn back_up_sqlite_before_migration( async fn run_migrations(pool: &DbPool, database_url: &str, approval: Option<&str>) { // https://github.com/diesel-rs/diesel_async/discussions/268 - let conn = pool.get_owned().await.unwrap(); + // + // An unwritable data directory shows up here as a pool timeout rather than a + // permission error, which is confusing enough to be worth calling out. + let conn = pool.get_owned().await.unwrap_or_else(|err| { + panic!( + "could not open the database at {database_url}: {err}. \ + If this is a timeout, check that the database and its directory are \ + writable by the user the server runs as (uid 10001 in the Docker image)." + ) + }); #[cfg(feature = "sqlite")] { diff --git a/src/models.rs b/src/models.rs index 090e22f..cbaef8b 100644 --- a/src/models.rs +++ b/src/models.rs @@ -19,7 +19,11 @@ use super::schema::*; #[diesel(table_name = account)] pub struct Account { pub id: String, + // Never serialize credential material into a response body. Deserialization + // is kept so the struct still round-trips through diesel. + #[serde(skip_serializing)] pub name_hash: String, + #[serde(skip_serializing)] pub password_hash: String, } diff --git a/src/rate_limit.rs b/src/rate_limit.rs new file mode 100644 index 0000000..1d9fc7f --- /dev/null +++ b/src/rate_limit.rs @@ -0,0 +1,247 @@ +//! Fixed-window rate limiting for unauthenticated endpoints. +//! +//! `/account/register` and `/account/login` are reachable without credentials, +//! so without a limit they allow unlimited password guessing and unlimited +//! account creation (each account can then consume storage quota). +//! +//! The counters are per process and in memory. That is enough to blunt abuse +//! from a single source; an operator running multiple replicas or needing +//! stronger guarantees should also rate limit at the reverse proxy. +//! +//! Register the limiter once on the `App` and share it. `HttpServer::new` runs +//! its factory once per worker, so a limiter built in there would give each +//! worker independent counters and multiply the effective limit by the worker +//! count. + +use std::collections::HashMap; +use std::net::IpAddr; +use std::sync::Mutex; +use std::time::{Duration, Instant}; + +/// Requests permitted per client bucket within one window. +pub const MAX_REQUESTS_PER_WINDOW: u32 = 10; +pub const WINDOW: Duration = Duration::from_secs(60); + +/// Hard bound on tracked buckets, so a flood of distinct addresses cannot grow +/// the map without limit. +const MAX_TRACKED_CLIENTS: usize = 100_000; + +/// Key a client by address, grouping IPv6 by its /64. +/// +/// A single IPv6 allocation is routinely a /64 or larger, so keying on the full +/// address would let one host rotate through addresses for unlimited login +/// attempts. IPv4 is keyed on the full address. +fn client_bucket(client: IpAddr) -> [u8; 8] { + match client { + IpAddr::V4(address) => { + let mut bucket = [0u8; 8]; + bucket[..4].copy_from_slice(&address.octets()); + // tag so that an IPv4 address cannot collide with a /64 prefix + bucket[4] = 1; + bucket + } + IpAddr::V6(address) => { + let mut bucket = [0u8; 8]; + bucket.copy_from_slice(&address.octets()[..8]); + bucket + } + } +} + +struct Window { + started_at: Instant, + count: u32, +} + +pub struct RateLimiter { + windows: Mutex>, + max_requests: u32, + window: Duration, + max_tracked_clients: usize, + trust_forwarded_for: bool, + trusted_proxy_hops: usize, +} + +impl RateLimiter { + pub fn new(max_requests: u32, window: Duration) -> Self { + Self::with_capacity(max_requests, window, MAX_TRACKED_CLIENTS) + } + + /// Same as [`RateLimiter::new`] with an explicit tracking cap, so that tests + /// can exercise the overflow path without inserting 100_000 entries. + pub fn with_capacity(max_requests: u32, window: Duration, max_tracked_clients: usize) -> Self { + Self { + windows: Mutex::new(HashMap::new()), + max_requests, + window, + max_tracked_clients, + trust_forwarded_for: false, + trusted_proxy_hops: 1, + } + } + + /// Resolve client addresses from `X-Forwarded-For` instead of the peer. + /// + /// Carried here rather than read from the global config on every request, so + /// that the middleware stays unit-testable. + pub fn trusting_forwarded_for(mut self, trust: bool) -> Self { + self.trust_forwarded_for = trust; + self + } + + /// Number of proxies in front of this server, at least one. + pub fn with_trusted_proxy_hops(mut self, hops: usize) -> Self { + self.trusted_proxy_hops = hops.max(1); + self + } + + pub fn trusts_forwarded_for(&self) -> bool { + self.trust_forwarded_for + } + + pub fn trusted_proxy_hops(&self) -> usize { + self.trusted_proxy_hops + } + + /// Record a request from `client` and report whether it is permitted. + pub fn check(&self, client: IpAddr) -> bool { + let now = Instant::now(); + let mut windows = self.windows.lock().expect("rate limiter mutex poisoned"); + + // Bound the map. Expired entries go first; if a flood of distinct + // addresses is still live beyond the cap, drop everything rather than + // refusing unknown addresses, which would let an attacker with a large + // address pool lock every legitimate user out of login. + if windows.len() >= self.max_tracked_clients { + windows.retain(|_, window| now.duration_since(window.started_at) < self.window); + if windows.len() >= self.max_tracked_clients { + log::warn!( + "rate limiter tracked more than {} live clients; resetting counters. \ + Rate limit at the reverse proxy as well.", + self.max_tracked_clients + ); + windows.clear(); + } + } + + let entry = windows.entry(client_bucket(client)).or_insert(Window { + started_at: now, + count: 0, + }); + + if now.duration_since(entry.started_at) >= self.window { + entry.started_at = now; + entry.count = 0; + } + + entry.count += 1; + entry.count <= self.max_requests + } +} + +impl Default for RateLimiter { + fn default() -> Self { + Self::new(MAX_REQUESTS_PER_WINDOW, WINDOW) + } +} + +#[cfg(test)] +mod tests { + use super::{RateLimiter, WINDOW}; + use std::net::{IpAddr, Ipv4Addr}; + use std::time::Duration; + + fn client(last_octet: u8) -> IpAddr { + IpAddr::V4(Ipv4Addr::new(127, 0, 0, last_octet)) + } + + #[test] + fn permits_up_to_the_limit_then_rejects() { + let limiter = RateLimiter::new(3, WINDOW); + + assert!(limiter.check(client(1))); + assert!(limiter.check(client(1))); + assert!(limiter.check(client(1))); + assert!(!limiter.check(client(1))); + assert!(!limiter.check(client(1))); + } + + #[test] + fn tracks_clients_independently() { + let limiter = RateLimiter::new(1, WINDOW); + + assert!(limiter.check(client(1))); + assert!(!limiter.check(client(1))); + // a different address still gets its own budget + assert!(limiter.check(client(2))); + } + + /// Rotating addresses inside one IPv6 /64 must not buy extra attempts. + #[test] + fn ipv6_addresses_share_a_64_bucket() { + let limiter = RateLimiter::new(2, WINDOW); + + let first: IpAddr = "2001:db8::1".parse().unwrap(); + let second: IpAddr = "2001:db8::dead:beef".parse().unwrap(); + let other_prefix: IpAddr = "2001:db8:0:1::1".parse().unwrap(); + + assert!(limiter.check(first)); + assert!(limiter.check(second)); + // budget for this /64 is now spent, whichever address is used + assert!(!limiter.check(first)); + assert!(!limiter.check(second)); + // a different /64 is a different bucket + assert!(limiter.check(other_prefix)); + } + + #[test] + fn ipv4_does_not_collide_with_an_ipv6_prefix() { + let limiter = RateLimiter::new(1, WINDOW); + + // an IPv6 /64 of all zeroes would otherwise share a key with 0.0.0.0 + let v4: IpAddr = "0.0.0.0".parse().unwrap(); + let v6: IpAddr = "::".parse().unwrap(); + + assert!(limiter.check(v4)); + assert!(limiter.check(v6)); + } + + /// Overflow must stay bounded and fail open rather than locking out unknown + /// clients. Uses a small cap so the eviction branch is genuinely reached. + #[test] + fn tracked_clients_stay_bounded_without_locking_clients_out() { + const CAP: usize = 64; + let limiter = RateLimiter::with_capacity(1, Duration::from_secs(3600), CAP); + + // every window is live, so eviction cannot reclaim anything and the + // limiter must fall back to clearing + let mut seen = 0usize; + for octet_a in 0..=3u8 { + for octet_b in 0..=255u8 { + limiter.check(IpAddr::V4(Ipv4Addr::new(10, 0, octet_a, octet_b))); + seen += 1; + } + } + assert!( + seen > CAP, + "inserted {seen} clients, which must exceed the cap {CAP} to be meaningful" + ); + + let tracked = limiter.windows.lock().unwrap().len(); + assert!(tracked <= CAP, "tracked {tracked} buckets, cap {CAP}"); + + // a fresh client is still served rather than rejected outright + assert!(limiter.check(IpAddr::V4(Ipv4Addr::new(198, 51, 100, 1)))); + } + + #[test] + fn budget_resets_after_the_window() { + let limiter = RateLimiter::new(1, Duration::from_millis(20)); + + assert!(limiter.check(client(1))); + assert!(!limiter.check(client(1))); + + std::thread::sleep(Duration::from_millis(30)); + assert!(limiter.check(client(1))); + } +} diff --git a/src/validation.rs b/src/validation.rs index f3bdb67..109c28b 100644 --- a/src/validation.rs +++ b/src/validation.rs @@ -1,6 +1,7 @@ //! Validates user-provided data to be valid (to some extent, as it only has limited info due to using YouTube's RSS feeds) use std::cmp::max; +use std::collections::HashSet; use itertools::Itertools; @@ -34,13 +35,14 @@ fn verify_image_url(image_url: &str) -> bool { return false; }; - for thumbnail_domain in ALLOWED_THUMBNAIL_DOMAINS { - if host.ends_with(thumbnail_domain) { - return true; - } - } - - false + // Match on label boundaries. A plain `ends_with` would also accept + // attacker-registrable domains such as `evil-youtube.com`. + ALLOWED_THUMBNAIL_DOMAINS.iter().any(|domain| { + host == *domain + || host + .strip_suffix(domain) + .is_some_and(|prefix| prefix.ends_with('.')) + }) } fn alphanumeric_words(s: &str) -> Vec { @@ -88,14 +90,55 @@ async fn is_channel_validation_required(conn: &mut DbConnection, channel: &Chann true } -pub async fn validate_channel_information_if_changed( +/// Decide which of `channels` still need to be checked against YouTube. +/// +/// This is the only phase that needs a database connection. Callers should run +/// it, release their connection, and only then call +/// [`validate_channel_against_youtube`], so that a batch of slow network +/// round-trips never occupies a pooled connection. +/// +/// Returned indices are deduplicated by the whole channel value. +/// +/// Deduplicating by id alone would be a validation bypass: two entries can share +/// an id but carry different names or avatars, and the caller persists every +/// entry. Skipping the second one would let it reach the database unvalidated. +/// Channels are shared between accounts, so that would also let one account +/// poison a channel's avatar for everyone. +pub async fn channels_requiring_validation( conn: &mut DbConnection, - channel: &mut Channel, -) -> HandlerResult<()> { - if !is_channel_validation_required(conn, channel).await { - return Ok(()); + channels: &[Channel], +) -> Vec { + let mut required = Vec::new(); + + for index in distinct_channel_indices(channels) { + if is_channel_validation_required(conn, &channels[index]).await { + required.push(index); + } } + required +} + +/// Indices of the first occurrence of each distinct channel value. +/// +/// Entries that are fully equal are interchangeable, so validating one covers +/// the rest. Entries that differ in any field are kept separately. +fn distinct_channel_indices(channels: &[Channel]) -> Vec { + let mut seen = HashSet::new(); + + channels + .iter() + .enumerate() + .filter(|(_, channel)| seen.insert(*channel)) + .map(|(index, _)| index) + .collect() +} + +/// Validate a channel against its YouTube RSS feed. +/// +/// Deliberately takes no database connection so that callers cannot hold one +/// across the network round-trip. +pub async fn validate_channel_against_youtube(channel: &mut Channel) -> HandlerResult<()> { let rss_channel = RssChannel::fetch_from_channel_id(&channel.id) .await .map_err(|_| HandlerError::YouTubeConnectError)?; @@ -124,22 +167,42 @@ fn validate_channel_information( Ok(channel) } -/// Requirement: all videos must be from the same channel! -pub async fn validate_video_information_if_changed_single( +/// Mark which videos still differ from the copy already stored, and therefore +/// need to be checked against YouTube. +/// +/// This is the only phase that needs a database connection. Run it, release the +/// connection, then call [`validate_videos_against_youtube`]. +pub async fn videos_requiring_validation( conn: &mut DbConnection, - video_data: &mut CreateVideo, -) -> HandlerResult<()> { - let mut video_datas = vec![video_data.clone()]; - validate_video_information_if_changed(conn, &mut video_datas, &mut video_data.uploader).await?; - (*video_data) = video_datas[0].clone(); + video_datas: &[CreateVideo], +) -> Vec { + if !CONFIG.validate_submitted_metadata { + return vec![false; video_datas.len()]; + } - Ok(()) + let mut required = Vec::with_capacity(video_datas.len()); + for video_data in video_datas { + // verification is only required if the video doesn't exist yet or has changed since then + let existing_video = get_video_by_id(conn, &video_data.id).await.ok().flatten(); + required.push( + !existing_video + .is_some_and(|existing| std::convert::Into::