feat(server): stats globais e rate limit por peer
This commit is contained in:
@@ -9,13 +9,19 @@ pub async fn announce(
|
|||||||
Json(req): Json<AnnounceRequest>,
|
Json(req): Json<AnnounceRequest>,
|
||||||
) -> AppResult<Json<serde_json::Value>> {
|
) -> AppResult<Json<serde_json::Value>> {
|
||||||
validate::validate_announce(&req)?;
|
validate::validate_announce(&req)?;
|
||||||
|
|
||||||
|
if !state.rate.lock().unwrap().check(&req.peer_id) {
|
||||||
|
return Err(AppError::BadRequest("rate limit exceeded".into()));
|
||||||
|
}
|
||||||
|
|
||||||
let db = state.db.lock().unwrap();
|
let db = state.db.lock().unwrap();
|
||||||
let id = crate::store::insert_file(&db, &req)
|
let id = crate::store::insert_file(&db, &req).map_err(|e| match e {
|
||||||
.map_err(|e| match e {
|
rusqlite::Error::SqliteFailure(err, _)
|
||||||
rusqlite::Error::SqliteFailure(err, _) if err.code == rusqlite::ErrorCode::ConstraintViolation => {
|
if err.code == rusqlite::ErrorCode::ConstraintViolation =>
|
||||||
AppError::BadRequest("duplicate file (hash already announced)".into())
|
{
|
||||||
}
|
AppError::BadRequest("duplicate file (hash already announced)".into())
|
||||||
other => AppError::Db(other),
|
}
|
||||||
})?;
|
other => AppError::Db(other),
|
||||||
|
})?;
|
||||||
Ok(Json(serde_json::json!({ "file_id": id, "ok": true })))
|
Ok(Json(serde_json::json!({ "file_id": id, "ok": true })))
|
||||||
}
|
}
|
||||||
@@ -3,6 +3,7 @@ pub mod files;
|
|||||||
pub mod health;
|
pub mod health;
|
||||||
pub mod peers;
|
pub mod peers;
|
||||||
pub mod search;
|
pub mod search;
|
||||||
|
pub mod stats;
|
||||||
pub mod ws;
|
pub mod ws;
|
||||||
|
|
||||||
use axum::{routing::{get, post}, Router};
|
use axum::{routing::{get, post}, Router};
|
||||||
@@ -16,6 +17,7 @@ pub fn router(state: SharedState) -> Router {
|
|||||||
.route("/files/{file_id}", get(files::get_file))
|
.route("/files/{file_id}", get(files::get_file))
|
||||||
.route("/announce", post(announce::announce))
|
.route("/announce", post(announce::announce))
|
||||||
.route("/peers/{file_id}", get(peers::peers))
|
.route("/peers/{file_id}", get(peers::peers))
|
||||||
|
.route("/stats", get(stats::stats))
|
||||||
.route("/tracker/ws", get(ws::ws_handler))
|
.route("/tracker/ws", get(ws::ws_handler))
|
||||||
.with_state(state);
|
.with_state(state);
|
||||||
|
|
||||||
|
|||||||
@@ -0,0 +1,18 @@
|
|||||||
|
use axum::extract::State;
|
||||||
|
use axum::Json;
|
||||||
|
use crate::error::AppResult;
|
||||||
|
use crate::state::SharedState;
|
||||||
|
|
||||||
|
pub async fn stats(State(state): State<SharedState>) -> AppResult<Json<serde_json::Value>> {
|
||||||
|
let db = state.db.lock().unwrap();
|
||||||
|
let files: i64 = db.query_row("SELECT count(*) FROM files", [], |r| r.get(0))?;
|
||||||
|
let bytes: i64 = db.query_row("SELECT COALESCE(SUM(size_bytes),0) FROM files", [], |r| r.get(0))?;
|
||||||
|
let t = state.tracker.lock().unwrap();
|
||||||
|
let online_peers = t.online_total();
|
||||||
|
Ok(Json(serde_json::json!({
|
||||||
|
"files": files,
|
||||||
|
"total_bytes": bytes,
|
||||||
|
"online_peers": online_peers,
|
||||||
|
"version": env!("CARGO_PKG_VERSION"),
|
||||||
|
})))
|
||||||
|
}
|
||||||
@@ -2,6 +2,7 @@ pub mod api;
|
|||||||
pub mod config;
|
pub mod config;
|
||||||
pub mod error;
|
pub mod error;
|
||||||
pub mod models;
|
pub mod models;
|
||||||
|
pub mod rate;
|
||||||
pub mod search;
|
pub mod search;
|
||||||
pub mod state;
|
pub mod state;
|
||||||
pub mod store;
|
pub mod store;
|
||||||
|
|||||||
@@ -0,0 +1,58 @@
|
|||||||
|
use std::collections::HashMap;
|
||||||
|
use std::time::{Duration, Instant};
|
||||||
|
|
||||||
|
pub struct RateLimiter {
|
||||||
|
window: Duration,
|
||||||
|
max: u32,
|
||||||
|
hits: HashMap<String, (Instant, u32)>,
|
||||||
|
}
|
||||||
|
|
||||||
|
impl RateLimiter {
|
||||||
|
pub fn new(per_second: u32) -> Self {
|
||||||
|
Self {
|
||||||
|
window: Duration::from_secs(1),
|
||||||
|
max: per_second,
|
||||||
|
hits: HashMap::new(),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
pub fn check(&mut self, key: &str) -> bool {
|
||||||
|
let now = Instant::now();
|
||||||
|
let entry = self.hits.entry(key.to_string()).or_insert((now, 0));
|
||||||
|
if now.duration_since(entry.0) > self.window {
|
||||||
|
*entry = (now, 0);
|
||||||
|
}
|
||||||
|
entry.1 += 1;
|
||||||
|
entry.1 <= self.max
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
#[cfg(test)]
|
||||||
|
mod tests {
|
||||||
|
use super::*;
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn allows_until_limit() {
|
||||||
|
let mut rl = RateLimiter::new(2);
|
||||||
|
assert!(rl.check("k"));
|
||||||
|
assert!(rl.check("k"));
|
||||||
|
assert!(!rl.check("k"));
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn keys_are_isolated() {
|
||||||
|
let mut rl = RateLimiter::new(1);
|
||||||
|
assert!(rl.check("a"));
|
||||||
|
assert!(rl.check("b"));
|
||||||
|
assert!(!rl.check("a"));
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn resets_after_window() {
|
||||||
|
let mut rl = RateLimiter::new(1);
|
||||||
|
assert!(rl.check("k"));
|
||||||
|
assert!(!rl.check("k"));
|
||||||
|
rl.hits.get_mut("k").unwrap().0 = Instant::now() - Duration::from_secs(2);
|
||||||
|
assert!(rl.check("k"));
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -1,6 +1,7 @@
|
|||||||
use std::sync::{Arc, Mutex};
|
use std::sync::{Arc, Mutex};
|
||||||
use rusqlite::Connection;
|
use rusqlite::Connection;
|
||||||
use crate::config::Config;
|
use crate::config::Config;
|
||||||
|
use crate::rate::RateLimiter;
|
||||||
use crate::ws::tracker::PeerTracker;
|
use crate::ws::tracker::PeerTracker;
|
||||||
|
|
||||||
pub type DbConn = Mutex<Connection>;
|
pub type DbConn = Mutex<Connection>;
|
||||||
@@ -9,6 +10,7 @@ pub struct AppState {
|
|||||||
pub config: Config,
|
pub config: Config,
|
||||||
pub db: DbConn,
|
pub db: DbConn,
|
||||||
pub tracker: Mutex<PeerTracker>,
|
pub tracker: Mutex<PeerTracker>,
|
||||||
|
pub rate: Mutex<RateLimiter>,
|
||||||
}
|
}
|
||||||
|
|
||||||
impl AppState {
|
impl AppState {
|
||||||
@@ -17,6 +19,7 @@ impl AppState {
|
|||||||
config,
|
config,
|
||||||
db: Mutex::new(db),
|
db: Mutex::new(db),
|
||||||
tracker: Mutex::new(PeerTracker::default()),
|
tracker: Mutex::new(PeerTracker::default()),
|
||||||
|
rate: Mutex::new(RateLimiter::new(10)),
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
+8
-10
@@ -87,22 +87,20 @@ pub fn search_files(
|
|||||||
FROM files f",
|
FROM files f",
|
||||||
);
|
);
|
||||||
let mut params: Vec<Box<dyn rusqlite::ToSql>> = Vec::new();
|
let mut params: Vec<Box<dyn rusqlite::ToSql>> = Vec::new();
|
||||||
let mut where_added = false;
|
let mut clauses: Vec<String> = Vec::new();
|
||||||
|
|
||||||
if let Some(query) = crate::search::fts_query(q) {
|
if let Some(query) = crate::search::fts_query(q) {
|
||||||
sql.push_str(" JOIN files_fts ON files_fts.rowid = f.rowid WHERE files_fts MATCH ?1");
|
sql.push_str(" JOIN files_fts ON files_fts.rowid = f.rowid");
|
||||||
|
clauses.push("files_fts MATCH ?".to_string());
|
||||||
params.push(Box::new(query));
|
params.push(Box::new(query));
|
||||||
where_added = true;
|
|
||||||
}
|
}
|
||||||
if let Some(ft) = ftype {
|
if let Some(ft) = ftype {
|
||||||
if !where_added {
|
|
||||||
sql.push_str(" WHERE");
|
|
||||||
where_added = true;
|
|
||||||
} else {
|
|
||||||
sql.push_str(" AND");
|
|
||||||
}
|
|
||||||
params.push(Box::new(ft.to_string()));
|
params.push(Box::new(ft.to_string()));
|
||||||
sql.push_str(&format!(" f.ftype = ?{}", params.len()));
|
clauses.push(format!("f.ftype = ?{}", params.len()));
|
||||||
|
}
|
||||||
|
if !clauses.is_empty() {
|
||||||
|
sql.push_str(" WHERE ");
|
||||||
|
sql.push_str(&clauses.join(" AND "));
|
||||||
}
|
}
|
||||||
params.push(Box::new(limit));
|
params.push(Box::new(limit));
|
||||||
sql.push_str(&format!(" ORDER BY f.created_at DESC LIMIT ?{}", params.len()));
|
sql.push_str(&format!(" ORDER BY f.created_at DESC LIMIT ?{}", params.len()));
|
||||||
|
|||||||
@@ -0,0 +1,39 @@
|
|||||||
|
mod common;
|
||||||
|
|
||||||
|
use common::{announce_test_file, test_server};
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn stats_reflects_files() {
|
||||||
|
let server = test_server();
|
||||||
|
announce_test_file(&server, "a.ogg", "audio", &"a1".repeat(32)).await;
|
||||||
|
announce_test_file(&server, "b.ogg", "audio", &"a2".repeat(32)).await;
|
||||||
|
announce_test_file(&server, "c.pdf", "document", &"a3".repeat(32)).await;
|
||||||
|
let res = server.get("/api/v1/stats").await;
|
||||||
|
assert_eq!(res.status_code(), 200);
|
||||||
|
let body = res.json::<serde_json::Value>();
|
||||||
|
assert_eq!(body["files"], 3);
|
||||||
|
assert_eq!(body["total_bytes"], 300);
|
||||||
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn announce_rate_limited() {
|
||||||
|
let server = test_server();
|
||||||
|
// 10 por segundo no RateLimiter padrão; dispara 12
|
||||||
|
let mut last_status = axum::http::StatusCode::OK;
|
||||||
|
for i in 0..12 {
|
||||||
|
let res = server.post("/api/v1/announce")
|
||||||
|
.json(&serde_json::json!({
|
||||||
|
"peer_id": "ral",
|
||||||
|
"name": format!("f{i}.ogg"),
|
||||||
|
"description": "",
|
||||||
|
"type": "audio",
|
||||||
|
"extension": "ogg",
|
||||||
|
"size_bytes": 10,
|
||||||
|
"hash": format!("{:02x}", i).repeat(32),
|
||||||
|
"license": "authored"
|
||||||
|
}))
|
||||||
|
.await;
|
||||||
|
last_status = res.status_code();
|
||||||
|
}
|
||||||
|
assert_eq!(last_status, axum::http::StatusCode::BAD_REQUEST);
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user