From a310c8feefdd0d52330b0ce0f03836d823be1112 Mon Sep 17 00:00:00 2001 From: D3SOX Date: Tue, 28 Jul 2026 15:16:14 +0200 Subject: [PATCH 1/7] fix(security): harden auth, storage limits and deployment defaults Security review of the fork ahead of its first release. Findings are grouped below by what an attacker or a careless operator could achieve. Critical: - The example compose files and config shipped `secret_key = "changeme"` and the server started happily with it. Since that key signs JWTs, anyone knowing the default could forge a token for any account. Placeholder and short secrets are now rejected at startup, and the compose files require the value to be supplied. - Account name hashes were derived from `secret_key`, so rotating it after a leak made every existing account unreachable. A separate `username_secret` now derives them; it falls back to `secret_key` with a warning so existing deployments keep working. High: - JWT `exp` was written with `as_millis()`, but the claim is defined in seconds, so tokens were valid until roughly the year 59400 instead of one year. - SQLite defaults `foreign_keys` to OFF per connection and diesel never enables it, so every ON DELETE CASCADE was inert: deleting an account removed one row and orphaned all of its subscriptions, history, playlists and ciphertext. This contradicted the deletion guarantees in README and PRIVACY. Note that the pragma does not clean up rows already orphaned by earlier deletions. - Metadata validation issued one unbounded, untimed YouTube request per item while holding a pooled connection, so a single bulk request could occupy a connection indefinitely and a handful could exhaust the pool. Requests now time out, batches are capped, fetches are deduplicated per channel, and validation is split so no connection is held across network I/O. - Registration and login were unauthenticated and unlimited, and the legacy plaintext tables had no quota at all, so the encrypted-sync quota could be sidestepped by using the older endpoints. Adds per-address rate limiting and per-account row quotas. Medium and low: - The encrypted-sync requirement was decided with `path().contains("/encrypted_sync")`, so a playlist named `encrypted_sync` made plaintext routes exempt. Scopes now opt out explicitly. - Thumbnail domains were matched with `ends_with`, accepting attacker registrable domains such as `evil-youtube.com`. - Channel and playlist ids were interpolated into feed URLs unencoded. - `Account` serialized `password_hash`, one careless `.json()` from a leak. - The container ran as root. Two notes on the implementation itself. The YouTube client is thread-local rather than a global: a `reqwest::Client` binds its connection pool to the runtime that first uses it, and actix runs one current-thread runtime per worker, so a shared client fails once a second worker serves a request. And the rate limiter wraps the `/account` scope directly, because an additional nested `scope("")` matches the prefix of its sibling and would have made `DELETE /account/delete` return 404. BREAKING CHANGE: `secret_key` is now validated at startup. Deployments using a placeholder or a secret shorter than 32 bytes will refuse to start until it is replaced. Replacing it invalidates existing tokens and, unless `username_secret` is set to the previous `secret_key`, makes existing accounts unreachable. --- Dockerfile | 8 + README.md | 14 +- config.toml | 7 +- docker-compose.postgres.yml | 12 +- docker-compose.yml | 8 +- src/auth.rs | 5 +- src/config.rs | 130 +++++++++++++++- src/database.rs | 1 + src/database/quota.rs | 91 +++++++++++ src/handlers.rs | 33 ++++ src/handlers/channel_playback_speeds.rs | 19 ++- src/handlers/encrypted_sync.rs | 3 +- src/handlers/playlists.rs | 54 +++++-- src/handlers/subscriptions.rs | 52 +++++-- src/handlers/user.rs | 196 +++++++++++++++++++++++- src/handlers/watch_history.rs | 136 +++++++++++----- src/main.rs | 6 +- src/models.rs | 4 + src/rate_limit.rs | 116 ++++++++++++++ src/validation.rs | 105 +++++++++---- ytrss/src/channel.rs | 13 +- ytrss/src/lib.rs | 44 ++++++ ytrss/src/playlist.rs | 16 +- 23 files changed, 935 insertions(+), 138 deletions(-) create mode 100644 src/database/quota.rs create mode 100644 src/rate_limit.rs diff --git a/Dockerfile b/Dockerfile index a99b272..28a5d00 100644 --- a/Dockerfile +++ b/Dockerfile @@ -24,5 +24,13 @@ RUN apk add ca-certificates COPY --from=builder /app/target/release/opentubex-sync /app/opentubex-sync-server +# Run unprivileged. /app/data is owned by this user so the SQLite database and +# its WAL sidecar files stay writable. +RUN addgroup -S -g 10001 opentubex \ + && adduser -S -u 10001 -G opentubex opentubex \ + && mkdir -p /app/data \ + && chown -R opentubex:opentubex /app +USER opentubex + EXPOSE 8080 CMD ["./opentubex-sync-server"] diff --git a/README.md b/README.md index b0cf7e3..ec08eeb 100644 --- a/README.md +++ b/README.md @@ -44,11 +44,23 @@ 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` | | `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` | +### 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 also creates a consistent `*.pre-migration-` backup immediately before changing diff --git a/config.toml b/config.toml index 8616ea2..024e244 100644 --- a/config.toml +++ b/config.toml @@ -1,4 +1,9 @@ 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. +username_secret = "" allow_registration = true validate_submitted_metadata = true diff --git a/docker-compose.postgres.yml b/docker-compose.postgres.yml index 2537ea6..b6bd037 100644 --- a/docker-compose.postgres.yml +++ b/docker-compose.postgres.yml @@ -7,8 +7,14 @@ 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}" - "ALLOW_REGISTRATION=true" - "VALIDATE_SUBMITTED_METADATA=true" # Or use a configuration file: @@ -25,7 +31,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..cb557ae 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -8,7 +8,13 @@ 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}" - "ALLOW_REGISTRATION=true" - "VALIDATE_SUBMITTED_METADATA=true" # Existing databases refuse pending migrations unless this exactly lists diff --git a/src/auth.rs b/src/auth.rs index a8e7886..e34af75 100644 --- a/src/auth.rs +++ b/src/auth.rs @@ -21,13 +21,14 @@ pub fn bytes_to_hex_string(bytes: &[u8]) -> String { 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 + // tokens are valid for one year, should be enough in most cases. + // `exp` is defined in seconds since the epoch, not milliseconds. let expiration_date = SystemTime::now() .checked_add(Duration::from_hours(365 * 24)) .unwrap() .duration_since(UNIX_EPOCH) .unwrap() - .as_millis(); + .as_secs(); let claims = JwtClaims { sub: account.id.clone(), diff --git a/src/config.rs b/src/config.rs index 30488de..76580ee 100644 --- a/src/config.rs +++ b/src/config.rs @@ -4,10 +4,30 @@ const fn default_true() -> bool { true } +/// 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, #[serde(default = "default_true")] pub allow_registration: bool, #[serde(default = "default_true")] @@ -17,10 +37,116 @@ 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.username_secret.as_deref().unwrap_or(&self.secret) + } + + /// Whether the account name secret is pinned independently of `secret_key`. + pub fn has_dedicated_username_secret(&self) -> bool { + self.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.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), + allow_registration: true, + validate_submitted_metadata: true, + database_url: "./db.sqlite".to_owned(), + migration_approval: None, + } + } + + const STRONG: &str = "6f1c2f0e8a4b5d3c7e9f0a1b2c3d4e5f6a7b8c9d0e1f2a3b4c5d6e7f8a9b0c1d"; + + #[test] + fn placeholder_secrets_are_rejected() { + assert!(validate_secret("secret_key", "changeme").is_err()); + // rejected regardless of casing, and even though it is long enough + assert!(validate_secret("secret_key", "SomeVeryLongString64").is_err()); + } + + #[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()); + } + + #[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..04ae9fd 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,30 @@ 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(()) +} + +/// Reject the request if storing `incoming` more rows would exceed the +/// per-account quota for a plaintext table. +pub fn check_row_quota(stored_rows: i64, incoming: usize) -> HandlerResult<()> { + if crate::database::quota::exceeds_row_quota(stored_rows, incoming) { + 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..39e03d6 100644 --- a/src/handlers/channel_playback_speeds.rs +++ b/src/handlers/channel_playback_speeds.rs @@ -3,12 +3,17 @@ 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, + 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_row_quota, user::auth_middleware, + }, models::{Account, ChannelPlaybackSpeed}, }; @@ -63,6 +68,12 @@ async fn put_channel_playback_speed( speed.account_id = account.id; let mut conn = get_db_conn!(pool); + check_row_quota( + count_playback_speeds(&mut conn, &speed.account_id) + .await + .map_err(|_| HandlerError::InternalDatabaseError)?, + 1, + )?; set_channel_playback_speed(&mut conn, &speed) .await .map_err(|_| HandlerError::InternalDatabaseError)?; 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..bf7abdc 100644 --- a/src/handlers/playlists.rs +++ b/src/handlers/playlists.rs @@ -11,12 +11,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_bulk_size, check_row_quota, + user::auth_middleware, + }, + models::{Account, Channel, Playlist}, + validation::{validate_videos_against_youtube, videos_requiring_validation}, }; pub struct PlaylistsHandler {} @@ -203,28 +207,48 @@ 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() + check_bulk_size(video_datas.len())?; + + // 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()); + .chunk_by(|video| video.uploader.clone()) + .into_iter() + .map(|(channel, videos)| (channel, videos.collect())) + .collect(); - for (mut channel, videos) in &videos_grouped_by_uploader { - let mut videos: Vec<_> = videos.cloned().collect(); + // 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_row_quota( + count_playlist_videos(&mut conn, &account.id) + .await + .map_err(|_| HandlerError::InternalDatabaseError)?, + groups.iter().map(|(_, videos)| videos.len()).sum(), + )?; + for (_, videos) in &groups { + needs_validation.push(videos_requiring_validation(&mut conn, videos).await); + } + } - validate_video_information_if_changed(&mut conn, &mut videos, &mut channel).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); + for (channel, videos) in &groups { // store channel information first before storing video to ensure data integrity - create_or_update_channel(&mut conn, &channel) + create_or_update_channel(&mut conn, channel) .await .map_err(|_| HandlerError::InternalDatabaseError)?; for video in videos { - add_video_to_playlist(&mut conn, &playlist_id, &account.id, &(&video).into()) + add_video_to_playlist(&mut conn, &playlist_id, &account.id, &video.into()) .await .map_err(|_| HandlerError::InternalDatabaseError)?; } diff --git a/src/handlers/subscriptions.rs b/src/handlers/subscriptions.rs index 9cab3f1..1709c69 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_bulk_size, check_row_quota, + 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,13 +78,26 @@ 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_row_quota( + count_subscriptions(&mut conn, &account.id) + .await + .map_err(|_| HandlerError::InternalDatabaseError)?, + channels.len(), + )?; + 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 { @@ -122,12 +139,27 @@ 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_row_quota( + count_subscriptions(&mut conn, &account.id) + .await + .map_err(|_| HandlerError::InternalDatabaseError)?, + 1, + )?; + !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..fcc43b5 100644 --- a/src/handlers/user.rs +++ b/src/handlers/user.rs @@ -14,11 +14,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 +39,19 @@ 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. scope::scope("/account") + .app_data(web::Data::new(RateLimiter::default())) + .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 +77,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 +109,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 +154,26 @@ async fn delete_account( } } +/// Middleware that rate limits unauthenticated endpoints per client address. +/// +/// Uses the peer address rather than any forwarded header, since those are +/// attacker-controlled unless a trusted proxy rewrites them. +pub async fn rate_limit_middleware( + req: ServiceRequest, + next: Next, +) -> Result, actix_web::Error> { + let limiter: Option<&web::Data> = req.app_data(); + let client = req.peer_addr().map(|addr| addr.ip()); + + if let (Some(limiter), Some(client)) = (limiter, client) + && !limiter.check(client) + { + return Err(HandlerError::TooManyRequests.into()); + } + + next.call(req).await +} + /// Middleware that ensures that the account is authenticated. pub async fn auth_middleware( req: ServiceRequest, @@ -170,9 +206,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 +232,152 @@ 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; + + #[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 + ); + } + + /// 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..6e43722 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_bulk_size, check_row_quota, + user::auth_middleware, + }, + models::{Account, Channel, WatchHistoryItem}, + validation::{validate_videos_against_youtube, videos_requiring_validation}, }; pub struct WatchHistoryHandler; @@ -44,16 +48,68 @@ 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])), + } + } - validate_video_information_if_changed_single(conn, &mut item.video).await?; - Ok(item) + 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_row_quota( + count_watch_history(&mut conn, account_id) + .await + .map_err(|_| HandlerError::InternalDatabaseError)?, + items.len(), + )?; + 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(); + } + } + + Ok(items) } async fn persist_watch_history_item( @@ -73,14 +129,33 @@ async fn persist_watch_history_item( Ok(()) } -async fn store_watch_history_item( +async fn persist_watch_history_items( conn: &mut crate::DbConnection, + items: &[ExtendedWatchHistoryItem], +) -> HandlerResult<()> { + conn.transaction::<_, HandlerError, _>(|conn| { + async move { + for item in items { + persist_watch_history_item(conn, item).await?; + } + Ok(()) + } + .scope_boxed() + }) + .await +} + +async fn store_watch_history_items( + pool: &WebData, 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: 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, &items).await?; + + Ok(items) } #[utoipa::path(responses((status = OK)), security(("api_jwt_token" = [])))] @@ -90,24 +165,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()); + check_bulk_size(items.len())?; - for item in items { - prepared_items.push(prepare_watch_history_item(&mut conn, &account.id, item).await?); - } - - 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 +260,11 @@ async fn add_to_watch_history( pool: WebData, watch_history_item: web::Json, ) -> HandlerResult { - let mut conn = get_db_conn!(pool); + let mut stored = + store_watch_history_items(&pool, &account.id, vec![watch_history_item.into_inner()]) + .await?; - let watch_history_item = - store_watch_history_item(&mut conn, &account.id, 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..300ec94 100644 --- a/src/main.rs +++ b/src/main.rs @@ -34,6 +34,7 @@ mod dto; mod handlers; mod models; mod openapi; +mod rate_limit; mod schema; mod validation; @@ -121,8 +122,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()))?; 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..3178734 --- /dev/null +++ b/src/rate_limit.rs @@ -0,0 +1,116 @@ +//! 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. + +use std::collections::HashMap; +use std::net::IpAddr; +use std::sync::Mutex; +use std::time::{Duration, Instant}; + +/// Requests permitted per client address within one window. +pub const MAX_REQUESTS_PER_WINDOW: u32 = 10; +pub const WINDOW: Duration = Duration::from_secs(60); + +/// Entries are dropped once a window elapses, but a burst of distinct addresses +/// would still grow the map, so cap how many we track at once. +const MAX_TRACKED_CLIENTS: usize = 100_000; + +struct Window { + started_at: Instant, + count: u32, +} + +pub struct RateLimiter { + windows: Mutex>, + max_requests: u32, + window: Duration, +} + +impl RateLimiter { + pub fn new(max_requests: u32, window: Duration) -> Self { + Self { + windows: Mutex::new(HashMap::new()), + max_requests, + window, + } + } + + /// 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"); + + // Drop expired entries so the map does not grow without bound. + if windows.len() >= MAX_TRACKED_CLIENTS { + windows.retain(|_, window| now.duration_since(window.started_at) < self.window); + } + + let entry = windows.entry(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))); + } + + #[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..913d631 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,38 @@ 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 channel id. +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 seen = HashSet::new(); + let mut required = Vec::new(); + + for (index, channel) in channels.iter().enumerate() { + if !seen.insert(&channel.id) { + continue; + } + if is_channel_validation_required(conn, channel).await { + required.push(index); + } } + required +} + +/// 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 +150,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::