fix(dos): bound uploads, offload bcrypt, cap image decode, split queues
H3: re-enable a request-body limit (550MB, sized to the largest allowed video + multipart overhead) instead of DefaultBodyLimit::disable(), and drop the second full-file copy by keeping the upload as Bytes. H4: image decode now goes through ImageReader with explicit dimension (12000x12000) and allocation caps, rejecting decompression bombs before the expensive resize, and the whole spawn_blocking is wrapped in a 120s timeout mirroring the ffmpeg guard. H8: every bcrypt hash/verify (join, recover, admin_login, host PIN reset) now runs via spawn_blocking through a new services::password helper, so concurrent auths can't pin the async workers. M7: images and videos get independent compression permit pools, so slow videos can no longer starve image-preview generation. H5: add a generous per-IP daily account-creation cap (NAT-safe at ~100 guests) on top of the burst cap, and wire up the previously-dead upload_count_quota_* config as an event-wide file-count cap that a re-join cannot reset. Also bumps generated recovery PINs from 4 to 6 digits (part of M4). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -15,6 +15,7 @@ use crate::models::event::Event;
|
|||||||
use crate::models::session::Session;
|
use crate::models::session::Session;
|
||||||
use crate::models::user::{User, UserRole};
|
use crate::models::user::{User, UserRole};
|
||||||
use crate::services::config;
|
use crate::services::config;
|
||||||
|
use crate::services::password;
|
||||||
use crate::services::rate_limiter::client_ip;
|
use crate::services::rate_limiter::client_ip;
|
||||||
use crate::state::AppState;
|
use crate::state::AppState;
|
||||||
|
|
||||||
@@ -39,13 +40,23 @@ pub async fn join(
|
|||||||
let ip = client_ip(&headers, "unknown");
|
let ip = client_ip(&headers, "unknown");
|
||||||
let rate_limits_on = config::get_bool(&state.pool, "rate_limits_enabled", true).await;
|
let rate_limits_on = config::get_bool(&state.pool, "rate_limits_enabled", true).await;
|
||||||
let join_rate_on = config::get_bool(&state.pool, "join_rate_enabled", true).await;
|
let join_rate_on = config::get_bool(&state.pool, "join_rate_enabled", true).await;
|
||||||
if rate_limits_on && join_rate_on
|
if rate_limits_on && join_rate_on {
|
||||||
&& !state.rate_limiter.check(format!("join:{ip}"), 5, Duration::from_secs(60))
|
// Short burst cap (typos / double-taps) AND a generous per-IP daily cap
|
||||||
{
|
// on account creation. The daily cap bounds mass account-minting (which
|
||||||
return Err(AppError::TooManyRequests(
|
// otherwise resets the per-user upload budget, see the event-wide count
|
||||||
"Zu viele Anfragen. Bitte warte kurz und versuche es erneut.".into(),
|
// quota in the upload handler) while staying well above a real ~100-guest
|
||||||
None,
|
// venue sharing one NAT'd IP.
|
||||||
));
|
let burst_ok =
|
||||||
|
state.rate_limiter.check(format!("join:{ip}"), 5, Duration::from_secs(60));
|
||||||
|
let daily_ok = state
|
||||||
|
.rate_limiter
|
||||||
|
.check(format!("join_day:{ip}"), 200, Duration::from_secs(86_400));
|
||||||
|
if !burst_ok || !daily_ok {
|
||||||
|
return Err(AppError::TooManyRequests(
|
||||||
|
"Zu viele Anfragen. Bitte warte kurz und versuche es erneut.".into(),
|
||||||
|
None,
|
||||||
|
));
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
let display_name = body.display_name.trim();
|
let display_name = body.display_name.trim();
|
||||||
@@ -78,10 +89,10 @@ pub async fn join(
|
|||||||
)));
|
)));
|
||||||
}
|
}
|
||||||
|
|
||||||
// Generate a 4-digit PIN
|
// Generate a 6-digit PIN (≈20 bits vs the old ~13.3, so brute-forcing it
|
||||||
let pin: String = format!("{:04}", rand::rng().random_range(0..10000u32));
|
// against the 3-strike lockout is far harder).
|
||||||
let pin_hash =
|
let pin: String = format!("{:06}", rand::rng().random_range(0..1_000_000u32));
|
||||||
bcrypt::hash(&pin, 12).map_err(|e| AppError::Internal(anyhow::anyhow!(e)))?;
|
let pin_hash = password::hash(pin.clone(), 12).await?;
|
||||||
|
|
||||||
let user = User::create(&state.pool, event.id, display_name, &pin_hash).await?;
|
let user = User::create(&state.pool, event.id, display_name, &pin_hash).await?;
|
||||||
|
|
||||||
@@ -180,8 +191,8 @@ pub async fn recover(
|
|||||||
User::reset_pin_attempts(&state.pool, user.id).await?;
|
User::reset_pin_attempts(&state.pool, user.id).await?;
|
||||||
}
|
}
|
||||||
|
|
||||||
let pin_matches = bcrypt::verify(&body.pin, &user.recovery_pin_hash)
|
let pin_matches =
|
||||||
.unwrap_or(false);
|
password::verify(body.pin.clone(), user.recovery_pin_hash.clone()).await;
|
||||||
|
|
||||||
if pin_matches {
|
if pin_matches {
|
||||||
// Reset failed attempts on success
|
// Reset failed attempts on success
|
||||||
@@ -271,8 +282,11 @@ pub async fn admin_login(
|
|||||||
));
|
));
|
||||||
}
|
}
|
||||||
|
|
||||||
let valid = bcrypt::verify(&body.password, &state.config.admin_password_hash)
|
let valid = password::verify(
|
||||||
.unwrap_or(false);
|
body.password.clone(),
|
||||||
|
state.config.admin_password_hash.clone(),
|
||||||
|
)
|
||||||
|
.await;
|
||||||
|
|
||||||
if !valid {
|
if !valid {
|
||||||
tracing::warn!(ip = %ip, "admin_login: wrong password");
|
tracing::warn!(ip = %ip, "admin_login: wrong password");
|
||||||
@@ -298,8 +312,7 @@ pub async fn admin_login(
|
|||||||
let dummy_pin: String = (0..32)
|
let dummy_pin: String = (0..32)
|
||||||
.map(|_| rand::rng().random_range(b'a'..=b'z') as char)
|
.map(|_| rand::rng().random_range(b'a'..=b'z') as char)
|
||||||
.collect();
|
.collect();
|
||||||
let dummy_hash = bcrypt::hash(&dummy_pin, 4)
|
let dummy_hash = password::hash(dummy_pin, 4).await?;
|
||||||
.map_err(|e| AppError::Internal(anyhow::anyhow!(e)))?;
|
|
||||||
let user = User::create(&state.pool, event.id, admin_name, &dummy_hash).await?;
|
let user = User::create(&state.pool, event.id, admin_name, &dummy_hash).await?;
|
||||||
sqlx::query("UPDATE \"user\" SET role = 'admin' WHERE id = $1")
|
sqlx::query("UPDATE \"user\" SET role = 'admin' WHERE id = $1")
|
||||||
.bind(user.id)
|
.bind(user.id)
|
||||||
|
|||||||
@@ -282,9 +282,8 @@ pub async fn reset_user_pin(
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
let pin: String = format!("{:04}", rand::rng().random_range(0..10000u32));
|
let pin: String = format!("{:06}", rand::rng().random_range(0..1_000_000u32));
|
||||||
let pin_hash =
|
let pin_hash = crate::services::password::hash(pin.clone(), 12).await?;
|
||||||
bcrypt::hash(&pin, 12).map_err(|e| AppError::Internal(anyhow::anyhow!(e)))?;
|
|
||||||
|
|
||||||
sqlx::query(
|
sqlx::query(
|
||||||
"UPDATE \"user\"
|
"UPDATE \"user\"
|
||||||
|
|||||||
@@ -90,7 +90,7 @@ pub async fn upload(
|
|||||||
let max_image_mb: i64 = config::get_i64(&state.pool, "max_image_size_mb", 20).await;
|
let max_image_mb: i64 = config::get_i64(&state.pool, "max_image_size_mb", 20).await;
|
||||||
let max_video_mb: i64 = config::get_i64(&state.pool, "max_video_size_mb", 500).await;
|
let max_video_mb: i64 = config::get_i64(&state.pool, "max_video_size_mb", 500).await;
|
||||||
|
|
||||||
let mut file_data: Option<Vec<u8>> = None;
|
let mut file_data: Option<axum::body::Bytes> = None;
|
||||||
let mut caption: Option<String> = None;
|
let mut caption: Option<String> = None;
|
||||||
let mut hashtags_csv: Option<String> = None;
|
let mut hashtags_csv: Option<String> = None;
|
||||||
|
|
||||||
@@ -100,11 +100,11 @@ pub async fn upload(
|
|||||||
"file" => {
|
"file" => {
|
||||||
// The client-declared filename and Content-Type are deliberately
|
// The client-declared filename and Content-Type are deliberately
|
||||||
// ignored — the stored MIME and extension are derived from the
|
// ignored — the stored MIME and extension are derived from the
|
||||||
// file's magic bytes (see `classify`).
|
// file's magic bytes (see `classify`). Kept as `Bytes` (one copy
|
||||||
|
// off the wire) rather than an extra `.to_vec()`.
|
||||||
file_data = Some(
|
file_data = Some(
|
||||||
field.bytes().await
|
field.bytes().await
|
||||||
.map_err(|e| AppError::BadRequest(format!("Datei konnte nicht gelesen werden: {e}")))?
|
.map_err(|e| AppError::BadRequest(format!("Datei konnte nicht gelesen werden: {e}")))?,
|
||||||
.to_vec(),
|
|
||||||
);
|
);
|
||||||
}
|
}
|
||||||
"caption" => {
|
"caption" => {
|
||||||
@@ -172,6 +172,26 @@ pub async fn upload(
|
|||||||
None
|
None
|
||||||
};
|
};
|
||||||
|
|
||||||
|
// Event-wide file-count cap. Unlike the per-user budget, this can't be reset
|
||||||
|
// by minting a fresh account via /join, so it bounds total uploads no matter
|
||||||
|
// how many accounts an abuser creates. Wires up the previously-dead
|
||||||
|
// `upload_count_quota_*` config (default off to preserve existing behavior).
|
||||||
|
if quota_on && config::get_bool(&state.pool, "upload_count_quota_enabled", false).await {
|
||||||
|
let max_count = config::get_i64(&state.pool, "upload_count_quota_max", 10_000).await;
|
||||||
|
let (count,): (i64,) = sqlx::query_as(
|
||||||
|
"SELECT COUNT(*) FROM upload WHERE event_id = $1 AND deleted_at IS NULL",
|
||||||
|
)
|
||||||
|
.bind(auth.event_id)
|
||||||
|
.fetch_one(&state.pool)
|
||||||
|
.await?;
|
||||||
|
if count >= max_count {
|
||||||
|
return Err(AppError::TooManyRequests(
|
||||||
|
"Das Upload-Limit für dieses Event ist erreicht.".into(),
|
||||||
|
None,
|
||||||
|
));
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
// Reserve quota and insert the row in a single transaction so the byte
|
// Reserve quota and insert the row in a single transaction so the byte
|
||||||
// counter and the upload row are atomic (M6: no quota leak if we crash
|
// counter and the upload row are atomic (M6: no quota leak if we crash
|
||||||
// between them) and the reservation is a conditional UPDATE that row-locks
|
// between them) and the reservation is a conditional UPDATE that row-locks
|
||||||
@@ -222,7 +242,7 @@ pub async fn upload(
|
|||||||
return Err(AppError::Internal(e.into()));
|
return Err(AppError::Internal(e.into()));
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
if let Err(e) = tokio::fs::write(&absolute_path, &data).await {
|
if let Err(e) = tokio::fs::write(&absolute_path, data.as_ref()).await {
|
||||||
tx.rollback().await.ok();
|
tx.rollback().await.ok();
|
||||||
return Err(AppError::Internal(e.into()));
|
return Err(AppError::Internal(e.into()));
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -57,9 +57,11 @@ async fn main() -> Result<()> {
|
|||||||
.route("/api/v1/recover", post(auth::handlers::recover))
|
.route("/api/v1/recover", post(auth::handlers::recover))
|
||||||
.route("/api/v1/admin/login", post(auth::handlers::admin_login))
|
.route("/api/v1/admin/login", post(auth::handlers::admin_login))
|
||||||
.route("/api/v1/session", delete(auth::handlers::logout))
|
.route("/api/v1/session", delete(auth::handlers::logout))
|
||||||
// Upload — body limit disabled; size validation is done inside the handler
|
// Upload — cap the request body so a single request can't buffer the box
|
||||||
|
// into OOM. Sized to the largest allowed video (500 MB) plus multipart
|
||||||
|
// overhead; the per-category limit is still enforced inside the handler.
|
||||||
.route("/api/v1/upload", post(handlers::upload::upload)
|
.route("/api/v1/upload", post(handlers::upload::upload)
|
||||||
.route_layer(DefaultBodyLimit::disable()))
|
.route_layer(DefaultBodyLimit::max(550 * 1024 * 1024)))
|
||||||
.route(
|
.route(
|
||||||
"/api/v1/upload/{id}",
|
"/api/v1/upload/{id}",
|
||||||
patch(handlers::upload::edit_upload).delete(handlers::upload::delete_upload),
|
patch(handlers::upload::edit_upload).delete(handlers::upload::delete_upload),
|
||||||
|
|||||||
@@ -11,16 +11,27 @@ use crate::state::SseEvent;
|
|||||||
|
|
||||||
#[derive(Clone)]
|
#[derive(Clone)]
|
||||||
pub struct CompressionWorker {
|
pub struct CompressionWorker {
|
||||||
semaphore: Arc<Semaphore>,
|
/// Separate permit pools for images and videos so a couple of slow/large
|
||||||
|
/// videos (each holding a permit across the full ffmpeg wait) can never
|
||||||
|
/// starve image-preview generation, and vice versa.
|
||||||
|
image_sem: Arc<Semaphore>,
|
||||||
|
video_sem: Arc<Semaphore>,
|
||||||
pool: PgPool,
|
pool: PgPool,
|
||||||
media_path: PathBuf,
|
media_path: PathBuf,
|
||||||
sse_tx: broadcast::Sender<SseEvent>,
|
sse_tx: broadcast::Sender<SseEvent>,
|
||||||
}
|
}
|
||||||
|
|
||||||
impl CompressionWorker {
|
impl CompressionWorker {
|
||||||
pub fn new(pool: PgPool, media_path: PathBuf, concurrency: usize, sse_tx: broadcast::Sender<SseEvent>) -> Self {
|
pub fn new(
|
||||||
|
pool: PgPool,
|
||||||
|
media_path: PathBuf,
|
||||||
|
image_concurrency: usize,
|
||||||
|
video_concurrency: usize,
|
||||||
|
sse_tx: broadcast::Sender<SseEvent>,
|
||||||
|
) -> Self {
|
||||||
Self {
|
Self {
|
||||||
semaphore: Arc::new(Semaphore::new(concurrency)),
|
image_sem: Arc::new(Semaphore::new(image_concurrency)),
|
||||||
|
video_sem: Arc::new(Semaphore::new(video_concurrency)),
|
||||||
pool,
|
pool,
|
||||||
media_path,
|
media_path,
|
||||||
sse_tx,
|
sse_tx,
|
||||||
@@ -31,7 +42,9 @@ impl CompressionWorker {
|
|||||||
pub fn process(&self, upload_id: Uuid, original_path: String, mime_type: String) {
|
pub fn process(&self, upload_id: Uuid, original_path: String, mime_type: String) {
|
||||||
let worker = self.clone();
|
let worker = self.clone();
|
||||||
tokio::spawn(async move {
|
tokio::spawn(async move {
|
||||||
let _permit = worker.semaphore.acquire().await;
|
let is_video = mime_type.starts_with("video/");
|
||||||
|
let sem = if is_video { &worker.video_sem } else { &worker.image_sem };
|
||||||
|
let _permit = sem.acquire().await;
|
||||||
match worker.do_process(upload_id, &original_path, &mime_type).await {
|
match worker.do_process(upload_id, &original_path, &mime_type).await {
|
||||||
Ok(_) => {
|
Ok(_) => {
|
||||||
tracing::info!("compression completed for upload {upload_id}");
|
tracing::info!("compression completed for upload {upload_id}");
|
||||||
@@ -91,10 +104,25 @@ impl CompressionWorker {
|
|||||||
let preview_path_clone = preview_path.clone();
|
let preview_path_clone = preview_path.clone();
|
||||||
let mime_owned = mime_type.to_string();
|
let mime_owned = mime_type.to_string();
|
||||||
|
|
||||||
// Run blocking image operations in a spawn_blocking task
|
// Run blocking image operations in a spawn_blocking task, bounded by a
|
||||||
tokio::task::spawn_blocking(move || -> Result<()> {
|
// hard timeout (mirrors the ffmpeg guard) so a pathological decode can't
|
||||||
let img = image::open(&original)
|
// hold the permit forever.
|
||||||
.context("failed to open image")?;
|
let handle = tokio::task::spawn_blocking(move || -> Result<()> {
|
||||||
|
use image::ImageReader;
|
||||||
|
|
||||||
|
// Reject decompression bombs *before* fully decoding: a small file can
|
||||||
|
// otherwise expand to enormous dimensions and a very expensive resize.
|
||||||
|
// 12000×12000 covers any real phone photo; max_alloc caps memory.
|
||||||
|
let mut reader = ImageReader::open(&original)
|
||||||
|
.context("failed to open image")?
|
||||||
|
.with_guessed_format()
|
||||||
|
.context("failed to read image header")?;
|
||||||
|
let mut limits = image::Limits::default();
|
||||||
|
limits.max_image_width = Some(12_000);
|
||||||
|
limits.max_image_height = Some(12_000);
|
||||||
|
limits.max_alloc = Some(256 * 1024 * 1024);
|
||||||
|
reader.limits(limits);
|
||||||
|
let img = reader.decode().context("failed to decode image")?;
|
||||||
|
|
||||||
// Resize to max 800px wide, preserving aspect ratio
|
// Resize to max 800px wide, preserving aspect ratio
|
||||||
let preview = img.resize(800, 800, image::imageops::FilterType::Lanczos3);
|
let preview = img.resize(800, 800, image::imageops::FilterType::Lanczos3);
|
||||||
@@ -115,8 +143,12 @@ impl CompressionWorker {
|
|||||||
}
|
}
|
||||||
|
|
||||||
Ok(())
|
Ok(())
|
||||||
})
|
});
|
||||||
.await??;
|
|
||||||
|
match tokio::time::timeout(std::time::Duration::from_secs(120), handle).await {
|
||||||
|
Ok(join) => join.context("image task panicked")??,
|
||||||
|
Err(_) => anyhow::bail!("image processing timeout after 120s"),
|
||||||
|
}
|
||||||
|
|
||||||
Ok(format!("previews/{preview_filename}"))
|
Ok(format!("previews/{preview_filename}"))
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -5,5 +5,6 @@ pub mod jobs;
|
|||||||
pub mod maintenance;
|
pub mod maintenance;
|
||||||
pub mod media_fs;
|
pub mod media_fs;
|
||||||
pub mod media_token;
|
pub mod media_token;
|
||||||
|
pub mod password;
|
||||||
pub mod rate_limiter;
|
pub mod rate_limiter;
|
||||||
pub mod sse_tickets;
|
pub mod sse_tickets;
|
||||||
|
|||||||
25
backend/src/services/password.rs
Normal file
25
backend/src/services/password.rs
Normal file
@@ -0,0 +1,25 @@
|
|||||||
|
//! bcrypt hashing/verification offloaded to the blocking pool.
|
||||||
|
//!
|
||||||
|
//! bcrypt cost-12 is ~250–400ms of synchronous CPU. Called directly in an async
|
||||||
|
//! handler it pins a Tokio worker for that whole time, so a handful of
|
||||||
|
//! concurrent joins/recovers/admin-logins can starve every other request
|
||||||
|
//! (feed, SSE, upload). Routing the work through `spawn_blocking` keeps the
|
||||||
|
//! async reactor responsive.
|
||||||
|
|
||||||
|
use crate::error::AppError;
|
||||||
|
|
||||||
|
/// Hash a secret with the given bcrypt cost, off the async reactor.
|
||||||
|
pub async fn hash(plain: String, cost: u32) -> Result<String, AppError> {
|
||||||
|
tokio::task::spawn_blocking(move || bcrypt::hash(&plain, cost))
|
||||||
|
.await
|
||||||
|
.map_err(|e| AppError::Internal(anyhow::anyhow!(e)))?
|
||||||
|
.map_err(|e| AppError::Internal(anyhow::anyhow!(e)))
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Verify a secret against a bcrypt hash, off the async reactor. Returns `false`
|
||||||
|
/// on any error (mirrors the previous `unwrap_or(false)` fail-closed behavior).
|
||||||
|
pub async fn verify(plain: String, hash: String) -> bool {
|
||||||
|
tokio::task::spawn_blocking(move || bcrypt::verify(&plain, &hash).unwrap_or(false))
|
||||||
|
.await
|
||||||
|
.unwrap_or(false)
|
||||||
|
}
|
||||||
@@ -36,8 +36,14 @@ pub struct AppState {
|
|||||||
impl AppState {
|
impl AppState {
|
||||||
pub fn new(pool: PgPool, config: AppConfig) -> Self {
|
pub fn new(pool: PgPool, config: AppConfig) -> Self {
|
||||||
let (sse_tx, _) = broadcast::channel(256);
|
let (sse_tx, _) = broadcast::channel(256);
|
||||||
let compression =
|
// Independent image (2) and video (2) permit pools — see CompressionWorker.
|
||||||
CompressionWorker::new(pool.clone(), config.media_path.clone(), 2, sse_tx.clone());
|
let compression = CompressionWorker::new(
|
||||||
|
pool.clone(),
|
||||||
|
config.media_path.clone(),
|
||||||
|
2,
|
||||||
|
2,
|
||||||
|
sse_tx.clone(),
|
||||||
|
);
|
||||||
Self {
|
Self {
|
||||||
pool,
|
pool,
|
||||||
config,
|
config,
|
||||||
|
|||||||
Reference in New Issue
Block a user