From 056fdb9eaaf6a08cef8ee62357bf8a46a4e3ef37 Mon Sep 17 00:00:00 2001 From: Power2All Date: Sat, 25 Jul 2026 22:06:30 +0200 Subject: [PATCH 1/4] * Full code auditing, and found some possible security issues in the future. * Applied a README fix about library installation on Linux (Thanks https://github.com/diederikdehaas). * Some small version bumps, nothing special. --- README.md | 14 +- src/api/api.rs | 9 +- src/common/common.rs | 16 +- src/config/config.rs | 4 + src/config/impls/configuration.rs | 67 ++++++- src/config/structs/api_trackers_config.rs | 8 + src/config/structs/http_trackers_config.rs | 8 + src/config/structs/tracker_config.rs | 9 + src/config/structs/udp_trackers_config.rs | 8 + src/database/impls/database_connector.rs | 8 +- .../impls/database_connector_mysql.rs | 6 +- .../impls/database_connector_pgsql.rs | 6 +- .../impls/database_connector_sqlite.rs | 6 +- src/database/impls/query_builder.rs | 19 +- src/database/traits/database_backend.rs | 2 +- src/http/http.rs | 33 ++-- src/main.rs | 1 + src/security/security.rs | 36 ++-- src/security/tests.rs | 31 +++- src/tracker/impls/announce_entry.rs | 12 +- src/tracker/impls/torrent_counts.rs | 2 + src/tracker/impls/torrent_sharding.rs | 7 +- src/tracker/impls/torrent_tracker.rs | 2 +- src/tracker/impls/torrent_tracker_handlers.rs | 2 +- src/tracker/impls/torrent_tracker_keys.rs | 17 +- src/tracker/impls/torrent_tracker_peers.rs | 92 ++++++---- .../impls/torrent_tracker_rtctorrent.rs | 12 ++ src/tracker/impls/torrent_tracker_torrents.rs | 2 +- .../impls/torrent_tracker_torrents_updates.rs | 93 +++------- src/tracker/impls/torrent_tracker_users.rs | 12 +- src/tracker/impls/torrent_update_data.rs | 5 +- src/tracker/structs/torrent_counts.rs | 5 + src/tracker/types/torrents_updates.rs | 122 ++++++++++++- src/udp/enums/server_error.rs | 2 + src/udp/impls/parse_pool.rs | 32 +++- src/udp/impls/udp_server.rs | 135 +++++++++------ src/udp/structs/udp_server.rs | 8 +- src/udp/udp.rs | 4 +- src/websocket/websocket.rs | 14 +- tests/common/mod.rs | 12 +- tests/config_tests.rs | 93 +++++++++- tests/tracker_tests.rs | 163 +++++++++++++++++- tests/udp_tests.rs | 123 ++++++++++++- 43 files changed, 1001 insertions(+), 261 deletions(-) diff --git a/README.md b/README.md index 66da1acf..0b830298 100644 --- a/README.md +++ b/README.md @@ -65,7 +65,7 @@ It is a workspace member but is **not compiled by default** (`default-members` e ```bash # Debian / Ubuntu -sudo apt-get install libfontconfig1-dev +sudo apt-get install libfontconfig-dev # Fedora / RHEL sudo dnf install fontconfig-devel @@ -610,6 +610,18 @@ echo "WebRTC seeds: {$data['rtc_seeds']}"; ### ChangeLog +#### v4.2.17 +* Full code auditing, and found some possible security issues in the future. +* Applied a README fix about library installation on Linux (Thanks https://github.com/diederikdehaas). +* Some small version bumps, nothing special. + +#### v4.2.16 +* Added clearer error reporting for invalid hex-based identifiers. +* Tightened UUID and database identifier validation rules. +* Improved BEP 15 UDP request/response encoding and decoding for consistent big-endian behavior. +* Updated application/desktop/container version to 4.2.16 (including Docker image tags) and refreshed bundled dependency versions. +* Updated UDP parsing tests to match the revised packet encoding. + #### v4.2.15 * Did some further performance tweaking * Bumped versions diff --git a/src/api/api.rs b/src/api/api.rs index eba35e00..0bf79ab7 100644 --- a/src/api/api.rs +++ b/src/api/api.rs @@ -344,7 +344,8 @@ pub async fn api_service_token(request: &HttpRequest, config: Arc } /// Determines the client IP, honouring the configured `real_ip` header when trusted proxies -/// are enabled; falls back to the socket peer address. +/// are enabled and the request came from a configured proxy; falls back to the socket peer +/// address. /// /// # Errors /// @@ -355,11 +356,15 @@ pub async fn api_service_retrieve_remote_ip(request: &HttpRequest, data: Arc) -> Result, CustomError> { let mut queries: HashMap = HashMap::with_capacity(12); if let Some(result) = query { + validate_query_string_length(result)?; for query_item in result.split('&') { if query_item.is_empty() { continue; @@ -129,12 +134,17 @@ pub(crate) fn bin2hex(data: &[u8; 20], f: &mut Formatter) -> fmt::Result { write!(f, "{}", std::str::from_utf8(&chars).unwrap()) } -pub struct Hex20(pub [u8; 40]); +/// A 20-byte hash rendered as 40 lowercase hex characters in a stack buffer. +/// +/// The field is private: [`bin20_to_hex`] must remain the only constructor for +/// [`Hex20::as_str`] to stay sound. +pub struct Hex20([u8; 40]); impl Hex20 { /// Returns the hex string as `&str` (always 40 valid ASCII characters). #[inline] pub fn as_str(&self) -> &str { + // SAFETY: `bin20_to_hex` is the only constructor and writes only ASCII hex digits. unsafe { std::str::from_utf8_unchecked(&self.0) } } } diff --git a/src/config/config.rs b/src/config/config.rs index 483dd30a..d63ed281 100644 --- a/src/config/config.rs +++ b/src/config/config.rs @@ -12,6 +12,10 @@ pub(crate) fn default_rtc_interval() -> u64 { 30 } pub(crate) fn default_rtc_peers_timeout() -> u64 { 120 } /// Serde default interval in seconds between key-expiry sweeps: `60`. pub(crate) fn default_keys_cleanup_interval() -> u64 { 60 } +/// Serde default cap on peers kept per torrent per peer map: `10000`. +pub(crate) fn default_max_peers_per_torrent() -> u64 { 10_000 } +/// Serde default cap on SDP answers queued for one RtcTorrent peer: `32`. +pub(crate) fn default_max_rtc_pending_answers() -> u64 { 32 } /// Serde default Prometheus metric namespace: `torrust_actix`. pub(crate) fn default_prometheus_id() -> String { String::from("torrust_actix") } /// Serde default cluster mode: standalone (no clustering). diff --git a/src/config/impls/configuration.rs b/src/config/impls/configuration.rs index 4806858d..f355f4f1 100644 --- a/src/config/impls/configuration.rs +++ b/src/config/impls/configuration.rs @@ -48,6 +48,8 @@ impl Configuration { peers_timeout: 2700, peers_cleanup_interval: 900, peers_cleanup_threads: 256, + max_peers_per_torrent: 10_000, + max_rtc_pending_answers: 32, total_downloads: 0, swagger: false, prometheus_id: String::from("torrust_actix"), @@ -144,6 +146,8 @@ impl Configuration { bind_address: String::from("0.0.0.0:6969"), real_ip: String::from("X-Real-IP"), trusted_proxies: false, + trusted_proxy_ips: Vec::new(), + trusted_proxy_addrs: Vec::new(), keep_alive: 60, request_timeout: 15, disconnect_timeout: 15, @@ -167,6 +171,8 @@ impl Configuration { reuse_address: true, use_payload_ip: false, simple_proxy_protocol: false, + proxy_addresses: Vec::new(), + proxy_addrs: Vec::new(), receive_method: UdpReceiveMethod::recvmmsg, } ), @@ -176,6 +182,8 @@ impl Configuration { bind_address: String::from("0.0.0.0:8080"), real_ip: String::from("X-Real-IP"), trusted_proxies: false, + trusted_proxy_ips: Vec::new(), + trusted_proxy_addrs: Vec::new(), keep_alive: 60, request_timeout: 30, disconnect_timeout: 30, @@ -230,6 +238,12 @@ impl Configuration { if let Ok(value) = env::var("TRACKER__PEERS_CLEANUP_THREADS") { config.tracker_config.peers_cleanup_threads = parse_env_num::("TRACKER__PEERS_CLEANUP_THREADS", &value, 256); } + if let Ok(value) = env::var("TRACKER__MAX_PEERS_PER_TORRENT") { + config.tracker_config.max_peers_per_torrent = parse_env_num::("TRACKER__MAX_PEERS_PER_TORRENT", &value, 10_000); + } + if let Ok(value) = env::var("TRACKER__MAX_RTC_PENDING_ANSWERS") { + config.tracker_config.max_rtc_pending_answers = parse_env_num::("TRACKER__MAX_RTC_PENDING_ANSWERS", &value, 32); + } if let Ok(value) = env::var("TRACKER__PROMETHEUS_ID") { config.tracker_config.prometheus_id = value; } @@ -766,11 +780,39 @@ impl Configuration { } } Self::env_overrides(&mut config); + Self::resolve_proxy_lists(&mut config); println!("[VALIDATE] Validating configuration..."); Self::validate(config.clone()); Ok(config) } + /// Parses the configured proxy address lists into [`IpAddr`]s once, so the request path + /// only has to compare already-parsed addresses. + /// + /// # Panics + /// + /// Panics when an entry is not a valid IP address. + pub fn resolve_proxy_lists(config: &mut Configuration) { + fn parse_list(label: &str, raw: &[String]) -> Vec { + raw.iter() + .map(|entry| { + entry.trim().parse::().unwrap_or_else(|_| { + panic!("[VALIDATE CONFIG] {label} contains an invalid IP address: \"{entry}\"") + }) + }) + .collect() + } + for (index, server) in config.http_server.iter_mut().enumerate() { + server.trusted_proxy_addrs = parse_list(&format!("http_server[{index}].trusted_proxy_ips"), &server.trusted_proxy_ips); + } + for (index, server) in config.api_server.iter_mut().enumerate() { + server.trusted_proxy_addrs = parse_list(&format!("api_server[{index}].trusted_proxy_ips"), &server.trusted_proxy_ips); + } + for (index, server) in config.udp_server.iter_mut().enumerate() { + server.proxy_addrs = parse_list(&format!("udp_server[{index}].proxy_addresses"), &server.proxy_addresses); + } + } + /// Validates the complete configuration (API key strength, identifier patterns, tracker, /// Sentry, cluster, cache and compression sections). /// @@ -826,6 +868,9 @@ impl Configuration { assert!(!api_server.ssl_key.is_empty(), "[VALIDATE CONFIG] api_server[{index}] ssl=true but ssl_key is not set"); assert!(!api_server.ssl_cert.is_empty(), "[VALIDATE CONFIG] api_server[{index}] ssl=true but ssl_cert is not set"); } + if api_server.trusted_proxies && api_server.trusted_proxy_addrs.is_empty() { + eprintln!("[SECURITY WARNING] api_server[{index}] trusted_proxies=true with an empty trusted_proxy_ips list: the '{}' header is honoured from any source, so a client reaching this listener directly can spoof its IP. Set trusted_proxy_ips to your proxy addresses.", api_server.real_ip); + } } } for (index, http_server) in config.http_server.iter().enumerate() { @@ -843,6 +888,9 @@ impl Configuration { assert!(!http_server.ssl_key.is_empty(), "[VALIDATE CONFIG] http_server[{index}] ssl=true but ssl_key is not set"); assert!(!http_server.ssl_cert.is_empty(), "[VALIDATE CONFIG] http_server[{index}] ssl=true but ssl_cert is not set"); } + if http_server.trusted_proxies && http_server.trusted_proxy_addrs.is_empty() { + eprintln!("[SECURITY WARNING] http_server[{index}] trusted_proxies=true with an empty trusted_proxy_ips list: the '{}' header is honoured from any source, so a client reaching this listener directly can spoof its IP. Set trusted_proxy_ips to your proxy addresses.", http_server.real_ip); + } } } for (index, udp_server) in config.udp_server.iter().enumerate() { @@ -853,6 +901,12 @@ impl Configuration { ); assert!(udp_server.udp_threads > 0, "[VALIDATE CONFIG] udp_server[{index}] udp_threads must be > 0"); assert!(udp_server.worker_threads > 0, "[VALIDATE CONFIG] udp_server[{index}] worker_threads must be > 0"); + if udp_server.simple_proxy_protocol && udp_server.proxy_addrs.is_empty() { + eprintln!("[SECURITY WARNING] udp_server[{index}] simple_proxy_protocol=true with an empty proxy_addresses list: the SPP header is trusted from any sender, so anyone can choose the client address the tracker records. Set proxy_addresses to your load balancer addresses."); + } + if udp_server.use_payload_ip { + eprintln!("[SECURITY WARNING] udp_server[{index}] use_payload_ip=true lets a UDP client choose the IP address recorded for it, which can be used to point swarms at a third party. Only enable this behind a trusted proxy."); + } } } Self::validate_tracker(&config); @@ -881,6 +935,12 @@ impl Configuration { assert!(tc.rtc_peers_timeout > 0, "[VALIDATE CONFIG] rtc_peers_timeout must be > 0"); println!("[VALIDATE] request_interval: {}s (min: {}s)", tc.request_interval, tc.request_interval_minimum); println!("[VALIDATE] peers_timeout: {}s, cleanup_interval: {}s, cleanup_threads: {}", tc.peers_timeout, tc.peers_cleanup_interval, tc.peers_cleanup_threads); + if tc.max_peers_per_torrent == 0 { + eprintln!("[SECURITY WARNING] max_peers_per_torrent=0 disables the per-torrent peer limit: a client announcing with made-up peer ids can grow memory until peers_timeout expires them."); + } else { + println!("[VALIDATE] max_peers_per_torrent: {} per peer map", tc.max_peers_per_torrent); + } + assert!(tc.max_rtc_pending_answers > 0, "[VALIDATE CONFIG] max_rtc_pending_answers must be > 0"); println!("[VALIDATE] rtc_interval: {}s, rtc_peers_timeout: {}s", tc.rtc_interval, tc.rtc_peers_timeout); } @@ -1026,6 +1086,8 @@ impl Configuration { remarks.insert(("tracker_config", "keys_enabled"), "# Optional: defaults to false -- require announce keys"); remarks.insert(("tracker_config", "keys_cleanup_interval"), "# Optional: defaults to 60 -- expired-key cleanup interval (seconds)"); remarks.insert(("tracker_config", "users_enabled"), "# Optional: defaults to false -- enable per-user statistics"); + remarks.insert(("tracker_config", "max_peers_per_torrent"), "# Optional: defaults to 10000 -- max peers kept per torrent per peer map (0 = unlimited, not recommended)"); + remarks.insert(("tracker_config", "max_rtc_pending_answers"), "# Optional: defaults to 32 -- max SDP answers queued for one RtcTorrent peer"); remarks.insert(("tracker_config", "swagger"), "# Optional: defaults to false -- expose Swagger UI at /swagger-ui/"); remarks.insert(("tracker_config", "prometheus_id"), "# Optional: defaults to \"torrust_actix\" -- Prometheus metric label"); remarks.insert(("tracker_config", "cluster"), "# Optional: defaults to \"standalone\" -- cluster mode: standalone | master | slave"); @@ -1083,6 +1145,9 @@ impl Configuration { remarks.insert(("database_structure.users", "column_completed"), "# Optional: defaults to \"completed\""); remarks.insert(("database_structure.users", "column_updated"), "# Optional: defaults to \"updated\""); remarks.insert(("database_structure.users", "column_active"), "# Optional: defaults to \"active\""); + remarks.insert(("http_server", "trusted_proxy_ips"), "# Optional: defaults to [] -- source IPs allowed to set real_ip; empty trusts any sender, which allows IP spoofing"); + remarks.insert(("api_server", "trusted_proxy_ips"), "# Optional: defaults to [] -- source IPs allowed to set real_ip; empty trusts any sender, which allows IP spoofing"); + remarks.insert(("udp_server", "proxy_addresses"), "# Optional: defaults to [] -- source IPs allowed to send a Simple Proxy Protocol header; empty trusts any sender"); let mut section_remarks: std::collections::HashMap<&str, &str> = std::collections::HashMap::new(); section_remarks.insert("sentry_config", "# Optional section: the entire [sentry_config] block can be omitted (defaults to disabled)"); section_remarks.insert("database_structure.torrents", "# Optional section: omit to use default table/column names for torrents"); @@ -1094,7 +1159,7 @@ impl Configuration { let mut current_section = String::new(); for line in toml.lines() { let trimmed = line.trim(); - if trimmed.starts_with('[') && !trimmed.starts_with("[[") { + if trimmed.starts_with('[') { let inner = trimmed.trim_start_matches('[').trim_end_matches(']'); current_section = inner.to_string(); if let Some(remark) = section_remarks.get(inner) { diff --git a/src/config/structs/api_trackers_config.rs b/src/config/structs/api_trackers_config.rs index b6afd23e..7dd4402a 100644 --- a/src/config/structs/api_trackers_config.rs +++ b/src/config/structs/api_trackers_config.rs @@ -2,6 +2,7 @@ use serde::{ Deserialize, Serialize }; +use std::net::IpAddr; #[derive(Serialize, Deserialize, Debug, Clone)] pub struct ApiTrackersConfig { @@ -10,6 +11,13 @@ pub struct ApiTrackersConfig { pub real_ip: String, #[serde(default)] pub trusted_proxies: bool, + /// Source addresses allowed to set the `real_ip` header. An empty list honours the header + /// from any source, which lets a client reaching this listener directly spoof its own IP. + #[serde(default)] + pub trusted_proxy_ips: Vec, + /// Parsed form of [`Self::trusted_proxy_ips`], filled in once at startup. + #[serde(skip)] + pub trusted_proxy_addrs: Vec, pub keep_alive: u64, pub request_timeout: u64, pub disconnect_timeout: u64, diff --git a/src/config/structs/http_trackers_config.rs b/src/config/structs/http_trackers_config.rs index 56dc64a6..5fb683b7 100644 --- a/src/config/structs/http_trackers_config.rs +++ b/src/config/structs/http_trackers_config.rs @@ -2,6 +2,7 @@ use serde::{ Deserialize, Serialize }; +use std::net::IpAddr; #[derive(Serialize, Deserialize, Debug, Clone)] pub struct HttpTrackersConfig { @@ -10,6 +11,13 @@ pub struct HttpTrackersConfig { pub real_ip: String, #[serde(default)] pub trusted_proxies: bool, + /// Source addresses allowed to set the `real_ip` header. An empty list honours the header + /// from any source, which lets a client reaching this listener directly spoof its own IP. + #[serde(default)] + pub trusted_proxy_ips: Vec, + /// Parsed form of [`Self::trusted_proxy_ips`], filled in once at startup. + #[serde(skip)] + pub trusted_proxy_addrs: Vec, pub keep_alive: u64, pub request_timeout: u64, pub disconnect_timeout: u64, diff --git a/src/config/structs/tracker_config.rs b/src/config/structs/tracker_config.rs index 383ec4c3..a1009d7b 100644 --- a/src/config/structs/tracker_config.rs +++ b/src/config/structs/tracker_config.rs @@ -36,6 +36,15 @@ pub struct TrackerConfig { pub peers_cleanup_interval: u64, /// Number of parallel threads used for peer cleanup. pub peers_cleanup_threads: u64, + /// Maximum peers kept per torrent in each peer map (IPv4 seeds, IPv6 seeds, IPv4 + /// leechers, IPv6 leechers, RTC seeds, RTC leechers). When a map is full a stale peer + /// is evicted to make room. `0` disables the limit, leaving memory bounded only by + /// announce rate times `peers_timeout`. + #[serde(default = "crate::config::config::default_max_peers_per_torrent")] + pub max_peers_per_torrent: u64, + /// Maximum SDP answers queued for a single RtcTorrent peer before the oldest is dropped. + #[serde(default = "crate::config::config::default_max_rtc_pending_answers")] + pub max_rtc_pending_answers: u64, /// Cumulative download count loaded from (or persisted to) the database. pub total_downloads: u64, /// Enable the built-in Swagger UI at `/swagger-ui/`. diff --git a/src/config/structs/udp_trackers_config.rs b/src/config/structs/udp_trackers_config.rs index 81d34525..464a9830 100644 --- a/src/config/structs/udp_trackers_config.rs +++ b/src/config/structs/udp_trackers_config.rs @@ -3,6 +3,7 @@ use serde::{ Deserialize, Serialize }; +use std::net::IpAddr; #[derive(Serialize, Deserialize, Debug, Clone)] pub struct UdpTrackersConfig { @@ -17,6 +18,13 @@ pub struct UdpTrackersConfig { pub use_payload_ip: bool, #[serde(default)] pub simple_proxy_protocol: bool, + /// Source addresses allowed to prepend a Simple Proxy Protocol header. An empty list trusts + /// the header from any sender, letting anyone choose the client address the tracker records. + #[serde(default)] + pub proxy_addresses: Vec, + /// Parsed form of [`Self::proxy_addresses`], filled in once at startup. + #[serde(skip)] + pub proxy_addrs: Vec, #[serde(default)] pub receive_method: UdpReceiveMethod, } \ No newline at end of file diff --git a/src/database/impls/database_connector.rs b/src/database/impls/database_connector.rs index ca426a5b..9910a3b9 100644 --- a/src/database/impls/database_connector.rs +++ b/src/database/impls/database_connector.rs @@ -325,27 +325,27 @@ impl DatabaseConnector { /// /// Returns the underlying `sqlx` error when the database operation fails, or /// `Error::RowNotFound` when no backend is initialised for the configured engine. - pub async fn save_torrents(&self, tracker: Arc, torrents: BTreeMap) -> Result<(), Error> + pub async fn save_torrents(&self, tracker: Arc, torrents: &BTreeMap) -> Result<(), Error> { let transaction = crate::utils::sentry_tracing::start_trace_transaction("db_save_torrents", "database"); let result: Result<(), Error> = match self.engine.as_ref() { Some(DatabaseDrivers::sqlite3) => { if let Some(ref sqlite) = self.sqlite { - sqlite.save_torrents(tracker, torrents.clone()).await + sqlite.save_torrents(tracker, torrents).await } else { Err(Error::RowNotFound) } } Some(DatabaseDrivers::mysql) => { if let Some(ref mysql) = self.mysql { - mysql.save_torrents(tracker, torrents.clone()).await + mysql.save_torrents(tracker, torrents).await } else { Err(Error::RowNotFound) } } Some(DatabaseDrivers::pgsql) => { if let Some(ref pgsql) = self.pgsql { - pgsql.save_torrents(tracker, torrents.clone()).await + pgsql.save_torrents(tracker, torrents).await } else { Err(Error::RowNotFound) } diff --git a/src/database/impls/database_connector_mysql.rs b/src/database/impls/database_connector_mysql.rs index 362b76aa..09916cc3 100644 --- a/src/database/impls/database_connector_mysql.rs +++ b/src/database/impls/database_connector_mysql.rs @@ -237,7 +237,7 @@ impl DatabaseConnectorMySQL { pub async fn save_torrents( &self, tracker: Arc, - torrents: BTreeMap, + torrents: &BTreeMap, ) -> Result<(), Error> { let mut transaction = self.pool.begin().await?; let mut handled = 0u64; @@ -246,7 +246,7 @@ impl DatabaseConnectorMySQL { let is_binary = structure.bin_type_infohash; let chunk_size = db_config.chunk_size; let mut in_chunk = 0u64; - for (info_hash, (counts, updates_action)) in &torrents { + for (info_hash, (counts, updates_action)) in torrents { handled += 1; let hash_str = info_hash.to_string(); match updates_action { @@ -929,7 +929,7 @@ impl DatabaseBackend for DatabaseConnectorMySQL { async fn save_torrents( &self, tracker: Arc, - torrents: BTreeMap, + torrents: &BTreeMap, ) -> Result<(), Error> { DatabaseConnectorMySQL::save_torrents(self, tracker, torrents).await } diff --git a/src/database/impls/database_connector_pgsql.rs b/src/database/impls/database_connector_pgsql.rs index 84be8c71..3389fd7f 100644 --- a/src/database/impls/database_connector_pgsql.rs +++ b/src/database/impls/database_connector_pgsql.rs @@ -235,7 +235,7 @@ impl DatabaseConnectorPgSQL { pub async fn save_torrents( &self, tracker: Arc, - torrents: BTreeMap, + torrents: &BTreeMap, ) -> Result<(), Error> { let mut transaction = self.pool.begin().await?; let mut handled = 0u64; @@ -244,7 +244,7 @@ impl DatabaseConnectorPgSQL { let is_binary = structure.bin_type_infohash; let chunk_size = db_config.chunk_size; let mut in_chunk = 0u64; - for (info_hash, (counts, updates_action)) in &torrents { + for (info_hash, (counts, updates_action)) in torrents { handled += 1; let hash_str = info_hash.to_string(); match updates_action { @@ -937,7 +937,7 @@ impl DatabaseBackend for DatabaseConnectorPgSQL { async fn save_torrents( &self, tracker: Arc, - torrents: BTreeMap, + torrents: &BTreeMap, ) -> Result<(), Error> { DatabaseConnectorPgSQL::save_torrents(self, tracker, torrents).await } diff --git a/src/database/impls/database_connector_sqlite.rs b/src/database/impls/database_connector_sqlite.rs index 04cb40d7..2589beb0 100644 --- a/src/database/impls/database_connector_sqlite.rs +++ b/src/database/impls/database_connector_sqlite.rs @@ -257,7 +257,7 @@ impl DatabaseConnectorSQLite { pub async fn save_torrents( &self, tracker: Arc, - torrents: BTreeMap, + torrents: &BTreeMap, ) -> Result<(), Error> { let transaction = crate::utils::sentry_tracing::start_trace_transaction("save_torrents", "database"); let mut transaction_db = self.pool.begin().await?; @@ -267,7 +267,7 @@ impl DatabaseConnectorSQLite { let is_binary = structure.bin_type_infohash; let chunk_size = db_config.chunk_size; let mut in_chunk = 0u64; - for (info_hash, (counts, updates_action)) in &torrents { + for (info_hash, (counts, updates_action)) in torrents { handled += 1; let hash_str = info_hash.to_string(); match updates_action { @@ -954,7 +954,7 @@ impl DatabaseBackend for DatabaseConnectorSQLite { async fn save_torrents( &self, tracker: Arc, - torrents: BTreeMap, + torrents: &BTreeMap, ) -> Result<(), Error> { DatabaseConnectorSQLite::save_torrents(self, tracker, torrents).await } diff --git a/src/database/impls/query_builder.rs b/src/database/impls/query_builder.rs index 54a05f25..0ea8d42a 100644 --- a/src/database/impls/query_builder.rs +++ b/src/database/impls/query_builder.rs @@ -8,10 +8,13 @@ impl QueryBuilder { } /// Quotes a table or column identifier for the bound engine. + /// + /// Quote characters are stripped so an identifier can never break out of its quoting. pub fn quote_identifier(&self, identifier: &str) -> String { + let safe: String = identifier.chars().filter(|c| *c != '`' && *c != '"' && *c != '\0').collect(); match self.engine { - DatabaseDrivers::sqlite3 | DatabaseDrivers::mysql => format!("`{identifier}`"), - DatabaseDrivers::pgsql => identifier.to_string(), + DatabaseDrivers::sqlite3 | DatabaseDrivers::mysql => format!("`{safe}`"), + DatabaseDrivers::pgsql => safe, } } @@ -24,9 +27,17 @@ impl QueryBuilder { } } - /// Formats a value as a single-quoted SQL string literal. + /// Formats a value as a single-quoted SQL string literal, escaping embedded quotes (and + /// backslashes on MySQL, which treats them as escapes by default). + /// + /// Statements here are assembled with `format!` and passed to sqlx via `AssertSqlSafe`, so + /// nothing downstream escapes anything: literals must be self-contained. pub fn text_literal(&self, value: &str) -> String { - format!("'{value}'") + let escaped = match self.engine { + DatabaseDrivers::mysql => value.replace('\\', "\\\\").replace('\'', "''"), + DatabaseDrivers::sqlite3 | DatabaseDrivers::pgsql => value.replace('\'', "''"), + }; + format!("'{escaped}'") } /// Builds the engine-specific upsert conflict clause updating the given columns. diff --git a/src/database/traits/database_backend.rs b/src/database/traits/database_backend.rs index f2c2d62d..1f3aa5a4 100644 --- a/src/database/traits/database_backend.rs +++ b/src/database/traits/database_backend.rs @@ -24,7 +24,7 @@ pub trait DatabaseBackend: Send + Sync { async fn save_torrents( &self, tracker: Arc, - torrents: BTreeMap, + torrents: &BTreeMap, ) -> Result<(), Error>; async fn save_whitelist( diff --git a/src/http/http.rs b/src/http/http.rs index e2994d91..67d4b29a 100644 --- a/src/http/http.rs +++ b/src/http/http.rs @@ -96,6 +96,10 @@ pub fn http_service_routes(data: Arc) -> Box ben_bytes!(b"rtctorrent not enabled" as &[u8]) + }.encode()); + } let tracker_config = &data.config.tracker_config; if tracker_config.whitelist_enabled && !data.check_whitelist(announce_unwrapped.info_hash) { return HttpResponse::Ok().content_type(ContentType::plaintext()).body(ERR_UNKNOWN_INFO_HASH.clone()); @@ -410,19 +421,14 @@ pub async fn http_service_announce_handler(request: HttpRequest, ip: IpAddr, dat let request_interval = tracker_config.request_interval as i64; let request_interval_minimum = tracker_config.request_interval_minimum as i64; let rtc_interval = tracker_config.rtc_interval as i64; - let is_rtc_request = announce_unwrapped.rtctorrent.unwrap_or(false); - if is_rtc_request && !rtctorrent_enabled { - return HttpResponse::Ok().content_type(ContentType::plaintext()).body(ben_map! { - "failure reason" => ben_bytes!(b"rtctorrent not enabled" as &[u8]) - }.encode()); - } + // Counts come from `counts`, never from the peer maps: those are bounded copies. let seeds_count = if is_rtc_request { - torrent_entry.rtc_seeds.len() as i64 + torrent_entry.counts.rtc_seeds as i64 } else { torrent_entry.counts.total_seeds() as i64 }; let peers_count = if is_rtc_request { - torrent_entry.rtc_peers.len() as i64 + torrent_entry.counts.rtc_peers as i64 } else { torrent_entry.counts.total_peers() as i64 }; @@ -845,7 +851,8 @@ pub async fn http_service_decode_hex_user_id(hash: String) -> Result std::io::Result<()> udp_server_object.reuse_address, udp_server_object.use_payload_ip, udp_server_object.simple_proxy_protocol, + Arc::new(udp_server_object.proxy_addrs.clone()), udp_server_object.receive_method, tracker.clone(), udp_rx.clone(), diff --git a/src/security/security.rs b/src/security/security.rs index 53b210d8..54aff3b5 100644 --- a/src/security/security.rs +++ b/src/security/security.rs @@ -40,15 +40,14 @@ pub fn validate_api_key_strength(api_key: &str) -> bool { } /// Compares two strings in constant time to prevent timing attacks on token checks. +/// +/// A length mismatch is folded into the result rather than short-circuiting, so an early return +/// cannot reveal the expected length. pub fn constant_time_eq(a: &str, b: &str) -> bool { - if a.len() != b.len() { - return false; - } - let a_bytes = a.as_bytes(); - let b_bytes = b.as_bytes(); - let mut result = 0u8; - for (x, y) in a_bytes.iter().zip(b_bytes.iter()) { - result |= x ^ y; + let (a, b) = (a.as_bytes(), b.as_bytes()); + let mut result = u8::from(a.len() != b.len()); + for i in 0..a.len().max(b.len()) { + result |= a.get(i).copied().unwrap_or(0) ^ b.get(i).copied().unwrap_or(0); } result == 0 } @@ -82,13 +81,6 @@ pub fn validate_peer_message(message: &str) -> Result<(), CustomError> { "Peer message exceeds maximum size of {MAX_PEER_MESSAGE_SIZE} bytes" ))); } - let suspicious_patterns = [" Result<(), CustomError> { /// /// Returns a [`CustomError`] when the format is invalid. pub fn validate_info_hash_hex(info_hash: &str) -> Result<(), CustomError> { - if info_hash.len() == 40 && info_hash.chars().all(|c| c.is_ascii_hexdigit()) { + if info_hash.len() == MAX_INFO_HASH_HEX_LENGTH && info_hash.bytes().all(|b| b.is_ascii_hexdigit()) { return Ok(()); } - if info_hash.len() >= 15 && info_hash.len() <= 60 { - return Ok(()); - } - Err(CustomError::new("info_hash has invalid format")) } @@ -114,10 +102,7 @@ pub fn validate_info_hash_hex(info_hash: &str) -> Result<(), CustomError> { /// /// Returns a [`CustomError`] when the format is invalid. pub fn validate_peer_id_hex(peer_id: &str) -> Result<(), CustomError> { - if peer_id.len() == 40 && peer_id.chars().all(|c| c.is_ascii_hexdigit()) { - return Ok(()); - } - if peer_id.len() >= 15 && peer_id.len() <= 60 { + if peer_id.len() == MAX_PEER_ID_HEX_LENGTH && peer_id.bytes().all(|b| b.is_ascii_hexdigit()) { return Ok(()); } Err(CustomError::new("peer_id has invalid format")) @@ -139,6 +124,9 @@ pub fn validate_query_string_length(query: &str) -> Result<(), CustomError> { /// Validates a client-supplied IP string (from a proxy header) before parsing it. /// +/// With `trusted_proxies_enabled = false`, loopback/private/unspecified values are rejected so +/// an untrusted sender cannot claim an internal address. +/// /// # Errors /// /// Returns a [`CustomError`] when the value is not a plausible IP or proxies are not trusted. diff --git a/src/security/tests.rs b/src/security/tests.rs index b682a0fd..38c2f13c 100644 --- a/src/security/tests.rs +++ b/src/security/tests.rs @@ -35,6 +35,10 @@ mod security_tests { #[test] fn test_constant_time_eq_different_length() { assert!(!constant_time_eq("test", "test_key")); + assert!(!constant_time_eq("test_key", "test")); + assert!(!constant_time_eq("", "test")); + assert!(!constant_time_eq("test", "")); + assert!(constant_time_eq("", "")); } #[test] @@ -65,19 +69,32 @@ mod security_tests { #[test] fn test_validate_peer_message_content() { assert!(validate_peer_message("normal message").is_ok()); - assert!(validate_peer_message("v=0\r\no=- 0 0 IN IP4 127.0.0.1\r\n").is_ok()); - } - - #[test] - fn test_validate_peer_message_suspicious() { - assert!(validate_peer_message("").is_err()); - assert!(validate_peer_message("javascript:alert(1)").is_err()); + // An SDP body is opaque to the tracker: only its size is bounded. + assert!(validate_peer_message("v=0\r\no=- 0 0 IN IP4 127.0.0.1\r\na=candidate:1 1 UDP 1 10.0.0.1 9 typ host\r\n").is_ok()); } #[test] fn test_validate_info_hash() { assert!(validate_info_hash_hex("3b245504cf5f11bb3ee84da598e4e5b78e5c2dde").is_ok()); + assert!(validate_info_hash_hex("3B245504CF5F11BB3EE84DA598E4E5B78E5C2DDE").is_ok()); assert!(validate_info_hash_hex("invalid!hash").is_err()); + // Only exactly 40 hex characters may pass. + assert!(validate_info_hash_hex("3b245504cf5f11bb3ee84da598e4e5b78e5c2dd").is_err()); + assert!(validate_info_hash_hex("3b245504cf5f11bb3ee84da598e4e5b78e5c2ddez").is_err()); + assert!(validate_info_hash_hex("not hex but twenty five ch").is_err()); + } + + #[test] + fn test_validate_peer_id_hex() { + assert!(validate_peer_id_hex("2d7142343235302d6b6568786f6272736e397a").is_err()); + assert!(validate_peer_id_hex("2d7142343235302d6b6568786f6272736e397a30").is_ok()); + assert!(validate_peer_id_hex("this is not hex at all but is long enough").is_err()); + } + + #[test] + fn test_validate_query_string_length() { + assert!(validate_query_string_length(&"a".repeat(MAX_QUERY_STRING_LENGTH)).is_ok()); + assert!(validate_query_string_length(&"a".repeat(MAX_QUERY_STRING_LENGTH + 1)).is_err()); } #[test] diff --git a/src/tracker/impls/announce_entry.rs b/src/tracker/impls/announce_entry.rs index ab48ea53..a405426a 100644 --- a/src/tracker/impls/announce_entry.rs +++ b/src/tracker/impls/announce_entry.rs @@ -12,17 +12,17 @@ use std::time::Instant; impl AnnounceEntry { /// Builds an announce snapshot from a live [`TorrentEntry`]. /// - /// Classic peer maps are capped at `SNAPSHOT_PEER_CAP` entries (enough to build a 72-peer - /// response); `counts` carries the exact full-swarm totals captured under the same lock. - /// RTC maps are copied in full, as RTC responses iterate all of them. + /// Every peer map, RTC included, is capped at `SNAPSHOT_PEER_CAP` entries (enough to build + /// a 72-peer response) so the per-announce cost does not grow with swarm size. `counts` + /// carries the exact full-swarm totals, captured under the same lock. pub fn from_entry(entry: &TorrentEntry) -> Self { AnnounceEntry { seeds: bounded_clone(&entry.seeds), seeds_ipv6: bounded_clone(&entry.seeds_ipv6), peers: bounded_clone(&entry.peers), peers_ipv6: bounded_clone(&entry.peers_ipv6), - rtc_seeds: entry.rtc_seeds.clone(), - rtc_peers: entry.rtc_peers.clone(), + rtc_seeds: bounded_clone(&entry.rtc_seeds), + rtc_peers: bounded_clone(&entry.rtc_peers), completed: entry.completed, updated: entry.updated, counts: TorrentCounts::from_entry(entry), @@ -46,6 +46,8 @@ impl Default for AnnounceEntry { seeds_ipv6: 0, peers_ipv4: 0, peers_ipv6: 0, + rtc_seeds: 0, + rtc_peers: 0, completed: 0, }, } diff --git a/src/tracker/impls/torrent_counts.rs b/src/tracker/impls/torrent_counts.rs index 61957723..ffa6a36f 100644 --- a/src/tracker/impls/torrent_counts.rs +++ b/src/tracker/impls/torrent_counts.rs @@ -9,6 +9,8 @@ impl TorrentCounts { seeds_ipv6: entry.seeds_ipv6.len(), peers_ipv4: entry.peers.len(), peers_ipv6: entry.peers_ipv6.len(), + rtc_seeds: entry.rtc_seeds.len(), + rtc_peers: entry.rtc_peers.len(), completed: entry.completed, } } diff --git a/src/tracker/impls/torrent_sharding.rs b/src/tracker/impls/torrent_sharding.rs index 994109de..b7b67b34 100644 --- a/src/tracker/impls/torrent_sharding.rs +++ b/src/tracker/impls/torrent_sharding.rs @@ -115,8 +115,11 @@ impl TorrentSharding { let (mut torrents_removed, mut seeds_removed, mut peers_removed) = (0u64, 0u64, 0u64); if let Some(shard_arc) = torrent_tracker.torrents_sharding.shards.get(shard as usize) { let now = std::time::Instant::now(); - let cutoff = now.checked_sub(peer_timeout).unwrap(); - let rtc_cutoff = now.checked_sub(rtc_peer_timeout).unwrap(); + // `Instant`'s origin is boot time on Linux, so this underflows while uptime is below + // the timeout. Nothing can be expired yet then; unwrapping here would abort. + let (Some(cutoff), Some(rtc_cutoff)) = (now.checked_sub(peer_timeout), now.checked_sub(rtc_peer_timeout)) else { + return; + }; let mut expired_full: Vec = Vec::new(); #[allow(clippy::type_complexity)] let mut expired_partial: Vec<(InfoHash, Vec, Vec, Vec, Vec, Vec, Vec)> = Vec::new(); diff --git a/src/tracker/impls/torrent_tracker.rs b/src/tracker/impls/torrent_tracker.rs index 68fcf17d..b664f83e 100644 --- a/src/tracker/impls/torrent_tracker.rs +++ b/src/tracker/impls/torrent_tracker.rs @@ -56,7 +56,7 @@ impl TorrentTracker { cache, certificate_store: Arc::new(CertificateStore::new()), torrents_sharding: Arc::new(Default::default()), - torrents_updates: Arc::new(RwLock::new(HashMap::new())), + torrents_updates: Arc::new(Default::default()), torrents_whitelist: Arc::new(RwLock::new(HashSet::new())), torrents_whitelist_updates: Arc::new(RwLock::new(HashMap::new())), torrents_blacklist: Arc::new(RwLock::new(HashSet::new())), diff --git a/src/tracker/impls/torrent_tracker_handlers.rs b/src/tracker/impls/torrent_tracker_handlers.rs index 49cf7bd6..8e46cda4 100644 --- a/src/tracker/impls/torrent_tracker_handlers.rs +++ b/src/tracker/impls/torrent_tracker_handlers.rs @@ -382,7 +382,7 @@ impl TorrentTracker { let result = scrape_query.info_hash.iter() .map(|&info_hash| { let counts = data.get_torrent_counts(info_hash).unwrap_or(crate::tracker::structs::torrent_counts::TorrentCounts { - seeds_ipv4: 0, seeds_ipv6: 0, peers_ipv4: 0, peers_ipv6: 0, completed: 0, + seeds_ipv4: 0, seeds_ipv6: 0, peers_ipv4: 0, peers_ipv6: 0, rtc_seeds: 0, rtc_peers: 0, completed: 0, }); (info_hash, counts) }) diff --git a/src/tracker/impls/torrent_tracker_keys.rs b/src/tracker/impls/torrent_tracker_keys.rs index 68b691b8..64921305 100644 --- a/src/tracker/impls/torrent_tracker_keys.rs +++ b/src/tracker/impls/torrent_tracker_keys.rs @@ -43,8 +43,10 @@ impl TorrentTracker { } } - /// Adds an announce key whose expiry is set to now + `timeout` seconds. A `timeout` of 0 - /// therefore expires immediately; there is no permanent-key value. + /// Adds an announce key whose expiry is set to now + `timeout` seconds. + /// + /// This stores an absolute timestamp, so `timeout = 0` means "expires now". Only a *stored* + /// expiry of `0`, as loaded from a database row, means permanent. /// /// Returns `true` when the key was newly inserted, `false` when it was refreshed. pub fn add_key(&self, hash: InfoHash, timeout: i64) -> bool @@ -91,11 +93,15 @@ impl TorrentTracker { } } - /// Returns `true` when the announce key exists (used on every keyed announce/scrape). + /// Returns `true` when the announce key exists and has not expired (used on every keyed + /// announce/scrape). An expiry of `0` means the key never expires. pub fn check_key(&self, hash: InfoHash) -> bool { let lock = self.keys.read_recursive(); lock.get(&hash).is_some_and(|&key| { + if key == 0 { + return true; + } let key_time = Utc.timestamp_opt(key, 0) .single() .map_or(UNIX_EPOCH, SystemTime::from); @@ -113,6 +119,8 @@ impl TorrentTracker { /// Removes every announce key whose expiry timestamp has passed. /// + /// Keys with an expiry of `0` are permanent and are left alone. + /// /// Runs periodically from the key-cleanup task. pub fn clean_keys(&self) { @@ -121,6 +129,9 @@ impl TorrentTracker { { let lock = self.keys.read_recursive(); for (&hash, &key_time) in lock.iter() { + if key_time == 0 { + continue; + } let time = Utc.timestamp_opt(key_time, 0) .single() .map_or(UNIX_EPOCH, SystemTime::from); diff --git a/src/tracker/impls/torrent_tracker_peers.rs b/src/tracker/impls/torrent_tracker_peers.rs index 2b3be852..92d816be 100644 --- a/src/tracker/impls/torrent_tracker_peers.rs +++ b/src/tracker/impls/torrent_tracker_peers.rs @@ -13,7 +13,47 @@ use log::info; use std::collections::hash_map::Entry; use std::net::SocketAddr; +/// How many entries to sample when choosing an eviction victim. +const EVICTION_SAMPLE: usize = 8; + +/// Makes room in a peer map before inserting `peer_id`, evicting an entry when the map is at +/// `cap`. Peer ids are client-chosen, so an uncapped map grows once per id announced. +/// +/// Returns the number of entries evicted, so the caller can keep the global counters exact. +/// +/// ponytail: samples `EVICTION_SAMPLE` entries and evicts the stalest of them rather than +/// scanning for the true oldest, which would be O(map len) under the shard write lock on +/// exactly the path a flood of made-up peer ids drives. Expiry is the cleanup task's job; this +/// is only a safety valve. If eviction accuracy matters, add an updated-ordered index. +#[inline] +fn enforce_peer_cap(map: &mut AHashMap, peer_id: PeerId, cap: usize) -> i64 { + if cap == 0 || map.len() < cap || map.contains_key(&peer_id) { + return 0; + } + let mut evicted = 0i64; + // Normally evicts exactly one entry; loops only if `cap` was lowered at runtime. + while map.len() >= cap { + let victim = map + .iter() + .take(EVICTION_SAMPLE) + .min_by_key(|(_, peer)| peer.updated) + .map(|(id, _)| *id); + let Some(victim) = victim else { + break; + }; + map.remove(&victim); + evicted += 1; + } + evicted +} + impl TorrentTracker { + /// Returns the configured per-map peer cap (`0` means unlimited). + #[inline] + fn peer_cap(&self) -> usize { + self.config.tracker_config.max_peers_per_torrent as usize + } + /// Returns up to `amount` seeds and peers of a torrent, filtered by IP family and excluding /// `self_peer_id`. Returns `None` when the torrent is unknown. pub fn get_torrent_peers(&self, info_hash: InfoHash, amount: usize, ip_type: TorrentPeersType, self_peer_id: Option) -> Option @@ -185,38 +225,32 @@ impl TorrentTracker { self.update_stats(StatsEvent::Completed, 1); entry.completed += 1; } + let cap = self.peer_cap(); if torrent_peer.is_rtctorrent() { - if torrent_peer.left == NumberOfBytes(0) { - self.update_stats(StatsEvent::Seeds, 1); - let mut new_peer = torrent_peer; - if !old_rtc_pending_answers.is_empty() - && let Some(ref mut rtc) = new_peer.rtc_data { - rtc.pending_answers = old_rtc_pending_answers; - } - entry.rtc_seeds.insert(peer_id, new_peer); - } else { - self.update_stats(StatsEvent::Peers, 1); - let mut new_peer = torrent_peer; - if !old_rtc_pending_answers.is_empty() - && let Some(ref mut rtc) = new_peer.rtc_data { - rtc.pending_answers = old_rtc_pending_answers; - } - entry.rtc_peers.insert(peer_id, new_peer); - } - } else if torrent_peer.left == NumberOfBytes(0) { - self.update_stats(StatsEvent::Seeds, 1); - if torrent_peer.peer_addr.is_ipv4() { - entry.seeds.insert(peer_id, torrent_peer); - } else { - entry.seeds_ipv6.insert(peer_id, torrent_peer); + let is_seed = torrent_peer.left == NumberOfBytes(0); + let mut new_peer = torrent_peer; + if !old_rtc_pending_answers.is_empty() + && let Some(ref mut rtc) = new_peer.rtc_data { + rtc.pending_answers = old_rtc_pending_answers; } + let target = if is_seed { &mut entry.rtc_seeds } else { &mut entry.rtc_peers }; + let evicted = enforce_peer_cap(target, peer_id, cap); + target.insert(peer_id, new_peer); + let event = if is_seed { StatsEvent::Seeds } else { StatsEvent::Peers }; + self.update_stats(event, 1 - evicted); } else { - self.update_stats(StatsEvent::Peers, 1); - if torrent_peer.peer_addr.is_ipv4() { - entry.peers.insert(peer_id, torrent_peer); - } else { - entry.peers_ipv6.insert(peer_id, torrent_peer); - } + let is_seed = torrent_peer.left == NumberOfBytes(0); + let is_ipv4 = torrent_peer.peer_addr.is_ipv4(); + let target = match (is_seed, is_ipv4) { + (true, true) => &mut entry.seeds, + (true, false) => &mut entry.seeds_ipv6, + (false, true) => &mut entry.peers, + (false, false) => &mut entry.peers_ipv6, + }; + let evicted = enforce_peer_cap(target, peer_id, cap); + target.insert(peer_id, torrent_peer); + let event = if is_seed { StatsEvent::Seeds } else { StatsEvent::Peers }; + self.update_stats(event, 1 - evicted); } entry.updated = std::time::Instant::now(); AnnounceEntry::from_entry(entry) diff --git a/src/tracker/impls/torrent_tracker_rtctorrent.rs b/src/tracker/impls/torrent_tracker_rtctorrent.rs index 4115ecdb..7def1596 100644 --- a/src/tracker/impls/torrent_tracker_rtctorrent.rs +++ b/src/tracker/impls/torrent_tracker_rtctorrent.rs @@ -72,7 +72,12 @@ impl TorrentTracker { /// /// The answer is compressed in memory and delivered on the seeder's next announce poll. /// Returns `false` when the seeder is not present or has no RTC state. + /// + /// The queue is filled by *other* peers naming this one as the target and survives the + /// target's re-announces, so it is capped at `max_rtc_pending_answers`, dropping the oldest + /// entry to make room. pub fn store_rtc_answer(&self, info_hash: InfoHash, seeder_peer_id: PeerId, answerer_peer_id: PeerId, sdp_answer: &str) -> bool { + let max_pending = self.config.tracker_config.max_rtc_pending_answers as usize; let shard = self.torrents_sharding.get_shard(info_hash.0[0]).unwrap(); let mut lock = shard.write(); if let Some(torrent_entry) = lock.get_mut(&info_hash) { @@ -80,6 +85,13 @@ impl TorrentTracker { .or_else(|| torrent_entry.rtc_peers.get_mut(&seeder_peer_id)); if let Some(seeder) = peer && let Some(ref mut rtc) = seeder.rtc_data { + if let Some(slot) = rtc.pending_answers.iter_mut().find(|(id, _)| *id == answerer_peer_id) { + slot.1 = CompressedBytes::compress(sdp_answer); + return true; + } + while rtc.pending_answers.len() >= max_pending { + rtc.pending_answers.remove(0); + } rtc.pending_answers.push((answerer_peer_id, CompressedBytes::compress(sdp_answer))); return true; } diff --git a/src/tracker/impls/torrent_tracker_torrents.rs b/src/tracker/impls/torrent_tracker_torrents.rs index a360a827..e2758abb 100644 --- a/src/tracker/impls/torrent_tracker_torrents.rs +++ b/src/tracker/impls/torrent_tracker_torrents.rs @@ -26,7 +26,7 @@ impl TorrentTracker { /// # Errors /// /// Returns `Err(())` when the database write fails; the caller re-queues the batch. - pub async fn save_torrents(&self, tracker: Arc, torrents: BTreeMap) -> Result<(), ()> + pub async fn save_torrents(&self, tracker: Arc, torrents: &BTreeMap) -> Result<(), ()> { let torrents_count = torrents.len(); if let Ok(()) = self.sqlx.save_torrents(tracker, torrents).await { diff --git a/src/tracker/impls/torrent_tracker_torrents_updates.rs b/src/tracker/impls/torrent_tracker_torrents_updates.rs index caec84eb..5b222c55 100644 --- a/src/tracker/impls/torrent_tracker_torrents_updates.rs +++ b/src/tracker/impls/torrent_tracker_torrents_updates.rs @@ -11,11 +11,7 @@ use log::{ info, warn }; -use std::collections::hash_map::Entry; -use std::collections::{ - BTreeMap, - HashMap -}; +use std::collections::BTreeMap; use std::sync::Arc; use std::sync::atomic::{AtomicU64, Ordering}; @@ -33,12 +29,11 @@ fn next_seq() -> u128 { impl TorrentTracker { /// Queues a torrent update for the next database/cache flush. /// - /// Returns `true` when a new queue slot was created. + /// Returns `true` when a new queue slot was created; a `false` means an update for the same + /// info-hash was already pending and has been superseded. pub fn add_torrent_update(&self, info_hash: InfoHash, torrent_update_data: TorrentUpdateData, updates_action: UpdatesAction) -> bool { - let mut lock = self.torrents_updates.write(); - let timestamp = next_seq(); - if lock.insert(timestamp, (info_hash, torrent_update_data, updates_action)).is_none() { + if self.torrents_updates.insert(next_seq(), info_hash, torrent_update_data, updates_action) { self.update_stats(StatsEvent::TorrentsUpdates, 1); true } else { @@ -46,59 +41,39 @@ impl TorrentTracker { } } - /// Re-queues a batch of torrent updates under fresh sequence numbers, removing the old slots. + /// Queues a batch of torrent updates. /// /// Returns, per info-hash, whether the insert created a new slot. - pub fn add_torrent_updates(&self, hashes: HashMap) -> BTreeMap + pub fn add_torrent_updates(&self, hashes: BTreeMap) -> BTreeMap { - let mut lock = self.torrents_updates.write(); let mut returned_data = BTreeMap::new(); let mut success_count = 0i64; - let mut remove_count = 0i64; - for (timestamp, (info_hash, torrent_entry, updates_action)) in hashes { - let new_timestamp = next_seq(); - let success = lock.insert(new_timestamp, (info_hash, torrent_entry, updates_action)).is_none(); + for (info_hash, (torrent_entry, updates_action)) in hashes { + let success = self.torrents_updates.insert(next_seq(), info_hash, torrent_entry, updates_action); if success { success_count += 1; } returned_data.insert(info_hash, success); - if lock.remove(×tamp).is_some() { - remove_count += 1; - } } if success_count > 0 { self.update_stats(StatsEvent::TorrentsUpdates, success_count); } - if remove_count > 0 { - self.update_stats(StatsEvent::TorrentsUpdates, -remove_count); - } returned_data } - /// Returns a clone of the pending torrent-update queue. - pub fn get_torrent_updates(&self) -> HashMap + /// Returns a copy of the pending torrent-update queue, keyed by info-hash. + pub fn get_torrent_updates(&self) -> BTreeMap { - let lock = self.torrents_updates.read_recursive(); - lock.clone() - } - - /// Removes a single queued update by its sequence key; returns `true` when it existed. - pub fn remove_torrent_update(&self, timestamp: &u128) -> bool - { - let mut lock = self.torrents_updates.write(); - if lock.remove(timestamp).is_some() { - self.update_stats(StatsEvent::TorrentsUpdates, -1); - true - } else { - false - } + self.torrents_updates.snapshot() + .into_iter() + .map(|(info_hash, (_, data, action))| (info_hash, (data, action))) + .collect() } /// Drops all queued torrent updates and resets the queue statistic. pub fn clear_torrent_updates(&self) { - let mut lock = self.torrents_updates.write(); - lock.clear(); + self.torrents_updates.clear(); self.set_stats(StatsEvent::TorrentsUpdates, 0); } @@ -111,36 +86,19 @@ impl TorrentTracker { /// queue so no data is lost. pub async fn save_torrent_updates(&self, torrent_tracker: Arc) -> Result<(), ()> { - let updates: HashMap = { - let mut lock = self.torrents_updates.write(); - std::mem::take(&mut *lock) - }; - if updates.is_empty() { + let drained_updates = self.torrents_updates.drain(); + if drained_updates.is_empty() { return Ok(()); } - let drained = updates.len() as i64; - self.update_stats(StatsEvent::TorrentsUpdates, -drained); - let mut mapping: HashMap = HashMap::with_capacity(updates.len()); - for (timestamp, (info_hash, torrent_entry, updates_action)) in updates { - match mapping.entry(info_hash) { - Entry::Occupied(mut o) => { - if timestamp > o.get().0 { - o.insert((timestamp, torrent_entry, updates_action)); - } - } - Entry::Vacant(v) => { - v.insert((timestamp, torrent_entry, updates_action)); - } - } - } - let mapping_len = mapping.len(); + let mapping_len = drained_updates.len(); + self.update_stats(StatsEvent::TorrentsUpdates, -(mapping_len as i64)); let is_persistent = torrent_tracker.config.database_structure.torrents.persistent.unwrap_or(torrent_tracker.config.database.persistent); - let torrents_to_save: BTreeMap = mapping + let torrents_to_save: BTreeMap = drained_updates .iter() .map(|(info_hash, (_, torrent_update_data, updates_action))| (*info_hash, (*torrent_update_data, *updates_action))) .collect(); let db_result = if is_persistent { - self.save_torrents(torrent_tracker.clone(), torrents_to_save.clone()).await + self.save_torrents(torrent_tracker.clone(), &torrents_to_save).await } else { Ok(()) }; @@ -188,14 +146,7 @@ impl TorrentTracker { Ok(()) } else { error!("[SYNC TORRENT UPDATES] Unable to sync {mapping_len} torrents"); - let mut lock = self.torrents_updates.write(); - let mut restored = 0i64; - for (info_hash, (timestamp, torrent_entry, updates_action)) in mapping { - if let Entry::Vacant(v) = lock.entry(timestamp) { - v.insert((info_hash, torrent_entry, updates_action)); - restored += 1; - } - } + let restored = self.torrents_updates.restore(drained_updates); self.update_stats(StatsEvent::TorrentsUpdates, restored); Err(()) } diff --git a/src/tracker/impls/torrent_tracker_users.rs b/src/tracker/impls/torrent_tracker_users.rs index c1e308f6..fdc81614 100644 --- a/src/tracker/impls/torrent_tracker_users.rs +++ b/src/tracker/impls/torrent_tracker_users.rs @@ -142,11 +142,17 @@ impl TorrentTracker { /// Resolves a user announce key to its [`UserId`] via the O(1) key index. /// - /// Returns `None` when no user owns the given key. + /// Returns `None` when no user owns the given key, or when the owning user is not active. + /// The `active` flag is the tracker's only way to turn a user off, so a deactivated user + /// must stop resolving here rather than merely being reported as inactive by the API. pub fn check_user_key(&self, key: UserId) -> Option { - let lock = self.users_key_index.read_recursive(); - lock.get(&key).copied() + let user_id = { + let lock = self.users_key_index.read_recursive(); + lock.get(&key).copied()? + }; + let users = self.users.read_recursive(); + users.get(&user_id).filter(|user| user.active != 0).map(|_| user_id) } /// Mutates a user entry in place under a single write lock, avoiding the diff --git a/src/tracker/impls/torrent_update_data.rs b/src/tracker/impls/torrent_update_data.rs index 121b094e..3e79d002 100644 --- a/src/tracker/impls/torrent_update_data.rs +++ b/src/tracker/impls/torrent_update_data.rs @@ -23,8 +23,9 @@ impl From<&AnnounceEntry> for TorrentUpdateData { seeds_ipv6: entry.counts.seeds_ipv6 as u64, peers_ipv4: entry.counts.peers_ipv4 as u64, peers_ipv6: entry.counts.peers_ipv6 as u64, - rtc_seeds: entry.rtc_seeds.len() as u64, - rtc_peers: entry.rtc_peers.len() as u64, + // From `counts`, never `len()`: the snapshot's peer maps are bounded copies. + rtc_seeds: entry.counts.rtc_seeds as u64, + rtc_peers: entry.counts.rtc_peers as u64, completed: entry.completed, } } diff --git a/src/tracker/structs/torrent_counts.rs b/src/tracker/structs/torrent_counts.rs index 45a7e2ec..38898323 100644 --- a/src/tracker/structs/torrent_counts.rs +++ b/src/tracker/structs/torrent_counts.rs @@ -4,5 +4,10 @@ pub struct TorrentCounts { pub seeds_ipv6: usize, pub peers_ipv4: usize, pub peers_ipv6: usize, + /// RtcTorrent seeders, counted separately because RtcTorrent responses report only the + /// WebRTC side of the swarm. + pub rtc_seeds: usize, + /// RtcTorrent leechers. + pub rtc_peers: usize, pub completed: u64, } \ No newline at end of file diff --git a/src/tracker/types/torrents_updates.rs b/src/tracker/types/torrents_updates.rs index eb8c76dd..6fbe9afe 100644 --- a/src/tracker/types/torrents_updates.rs +++ b/src/tracker/types/torrents_updates.rs @@ -1,8 +1,122 @@ -use std::collections::HashMap; -use std::sync::Arc; -use parking_lot::RwLock; use crate::tracker::enums::updates_action::UpdatesAction; use crate::tracker::structs::info_hash::InfoHash; use crate::tracker::structs::torrent_update_data::TorrentUpdateData; +use crate::tracker::types::ahash_map::AHashMap; +use parking_lot::RwLock; +use std::collections::hash_map::Entry; +use std::collections::BTreeMap; +use std::sync::Arc; + +/// Number of independently locked shards, matching the torrent store. +const SHARDS: usize = 256; + +/// A pending torrent update: the sequence number that ordered it, the counters to write and +/// what to do with them. +pub type PendingUpdate = (u128, TorrentUpdateData, UpdatesAction); + +/// Queue of torrent stat changes awaiting the next database/cache flush. +/// +/// Sharded by the first byte of the info-hash, like the torrent store, so the announce path does +/// not serialise on one lock. Entries are keyed by info-hash and deduplicated on insert, highest +/// sequence number winning, which bounds the queue by distinct torrents touched rather than by +/// announces served between flushes. +#[derive(Debug)] +pub struct TorrentUpdateQueue { + shards: [RwLock>; SHARDS], +} + +impl Default for TorrentUpdateQueue { + fn default() -> Self { + Self::new() + } +} + +impl TorrentUpdateQueue { + /// Creates an empty queue with 256 independently locked shards. + pub fn new() -> Self { + Self { + shards: std::array::from_fn(|_| RwLock::new(AHashMap::default())), + } + } + + #[inline] + fn shard(&self, info_hash: InfoHash) -> &RwLock> { + &self.shards[info_hash.0[0] as usize] + } + + /// Queues an update for `info_hash`, replacing any pending update with a lower sequence + /// number. + /// + /// Returns `true` when this created a new queue slot, so the caller can keep the + /// `torrents_updates` statistic equal to the number of pending entries. + #[inline] + pub fn insert(&self, seq: u128, info_hash: InfoHash, data: TorrentUpdateData, action: UpdatesAction) -> bool { + let mut lock = self.shard(info_hash).write(); + match lock.entry(info_hash) { + Entry::Occupied(mut o) => { + // Threads can take sequence numbers in one order and reach this lock in another, + // so the later writer is not necessarily the newer update. + if seq > o.get().0 { + o.insert((seq, data, action)); + } + false + } + Entry::Vacant(v) => { + v.insert((seq, data, action)); + true + } + } + } + + /// Removes and returns every pending update. + pub fn drain(&self) -> BTreeMap { + let mut drained = BTreeMap::new(); + for shard in &self.shards { + let taken = std::mem::take(&mut *shard.write()); + drained.extend(taken); + } + drained + } + + /// Puts drained updates back after a failed flush, skipping any info-hash that has since + /// been given a newer update. + /// + /// Returns how many entries were restored. + pub fn restore(&self, updates: BTreeMap) -> i64 { + let mut restored = 0i64; + for (info_hash, (seq, data, action)) in updates { + if self.insert(seq, info_hash, data, action) { + restored += 1; + } + } + restored + } + + /// Returns a copy of the pending updates without removing them. + pub fn snapshot(&self) -> BTreeMap { + let mut out = BTreeMap::new(); + for shard in &self.shards { + out.extend(shard.read().iter().map(|(k, v)| (*k, *v))); + } + out + } + + /// Drops every pending update. + pub fn clear(&self) { + for shard in &self.shards { + shard.write().clear(); + } + } + + /// Returns the number of pending updates across all shards. + pub fn len(&self) -> usize { + self.shards.iter().map(|shard| shard.read().len()).sum() + } + + /// Returns `true` when no update is pending. + pub fn is_empty(&self) -> bool { + self.shards.iter().all(|shard| shard.read().is_empty()) + } +} -pub type TorrentsUpdates = Arc>>; \ No newline at end of file +pub type TorrentsUpdates = Arc; diff --git a/src/udp/enums/server_error.rs b/src/udp/enums/server_error.rs index 85d069a3..bc3ecfb5 100644 --- a/src/udp/enums/server_error.rs +++ b/src/udp/enums/server_error.rs @@ -18,6 +18,8 @@ pub enum ServerError { TorrentBlacklisted, #[error("unknown key")] UnknownKey, + #[error("connection id missing or expired")] + InvalidConnectionId, #[error("peer not authenticated")] PeerNotAuthenticated, #[error("invalid authentication key")] diff --git a/src/udp/impls/parse_pool.rs b/src/udp/impls/parse_pool.rs index e25f0549..8f318b67 100644 --- a/src/udp/impls/parse_pool.rs +++ b/src/udp/impls/parse_pool.rs @@ -4,7 +4,10 @@ use crate::udp::enums::simple_proxy_protocol::SppParseResult; use crate::udp::structs::parse_pool::ParsePool; use crate::udp::structs::udp_packet::UdpPacket; use crate::udp::structs::udp_server::UdpServer; -use crate::udp::udp::parse_spp_header; +use crate::udp::udp::{ + has_spp_magic, + parse_spp_header +}; use crate::websocket::enums::protocol_type::ProtocolType; use crate::websocket::enums::request_type::RequestType; use crate::websocket::websocket::forward_request; @@ -45,13 +48,14 @@ impl ParsePool { /// Spawns `threads` async workers that drain the packet queue, run the UDP request pipeline /// and send replies, until the shutdown watch channel fires. - pub async fn start_thread(&self, threads: usize, tracker: Arc, shutdown_handler: tokio::sync::watch::Receiver, use_payload_ip: bool, simple_proxy_protocol: bool) { + pub async fn start_thread(&self, threads: usize, tracker: Arc, shutdown_handler: tokio::sync::watch::Receiver, use_payload_ip: bool, simple_proxy_protocol: bool, proxy_addrs: Arc>) { let is_slave_mode = tracker.config.tracker_config.cluster == ClusterMode::slave; for i in 0..threads { let payload = self.payload.clone(); let tracker_cloned = tracker.clone(); let mut shutdown_handler = shutdown_handler.clone(); let runtime = self.udp_runtime.clone(); + let proxy_addrs = proxy_addrs.clone(); runtime.spawn(async move { info!("[UDP] Start Parse Pool thread {i}..."); let mut batch: Vec = Vec::with_capacity(BATCH_SIZE); @@ -75,10 +79,11 @@ impl ParsePool { &tracker_cloned, packet, simple_proxy_protocol, + &proxy_addrs, ).await; } else { let (effective_addr, payload_slice) = if simple_proxy_protocol { - Self::extract_spp_info(&packet) + Self::extract_spp_info(&packet, &proxy_addrs) } else { (packet.remote_addr, packet.data.as_slice()) }; @@ -116,8 +121,23 @@ impl ParsePool { } } - fn extract_spp_info(packet: &UdpPacket) -> (SocketAddr, &[u8]) { + /// Resolves the effective client address for a datagram carrying a Simple Proxy Protocol + /// header. + /// + /// `proxy_addrs` lists the sources permitted to speak SPP; the header is unauthenticated, so + /// honouring it from an arbitrary sender lets anyone choose their recorded address. An empty + /// list trusts any sender, which configuration validation warns about at startup. + fn extract_spp_info<'a>(packet: &'a UdpPacket, proxy_addrs: &[std::net::IpAddr]) -> (SocketAddr, &'a [u8]) { let data = packet.data.as_slice(); + if !proxy_addrs.is_empty() && !proxy_addrs.contains(&packet.remote_addr.ip()) { + if has_spp_magic(data) { + debug!( + "[UDP SPP] Ignoring proxy header from untrusted source {}", + packet.remote_addr + ); + } + return (packet.remote_addr, data); + } match parse_spp_header(data) { SppParseResult::Found { header, payload_offset } => { debug!( @@ -137,9 +157,9 @@ impl ParsePool { } } - async fn handle_slave_forward(tracker: &Arc, packet: UdpPacket, simple_proxy_protocol: bool) { + async fn handle_slave_forward(tracker: &Arc, packet: UdpPacket, simple_proxy_protocol: bool, proxy_addrs: &[std::net::IpAddr]) { let (effective_addr, payload_data) = if simple_proxy_protocol { - let (addr, slice) = Self::extract_spp_info(&packet); + let (addr, slice) = Self::extract_spp_info(&packet, proxy_addrs); (addr, slice.to_vec()) } else { (packet.remote_addr, packet.data.to_vec()) diff --git a/src/udp/impls/udp_server.rs b/src/udp/impls/udp_server.rs index d39baf49..1b1f468f 100644 --- a/src/udp/impls/udp_server.rs +++ b/src/udp/impls/udp_server.rs @@ -44,11 +44,26 @@ use std::net::{ Ipv6Addr, SocketAddr }; -use std::sync::Arc; +use std::sync::{ + Arc, + LazyLock +}; use std::time::Duration; use tokio::net::UdpSocket; use tokio::runtime::Builder; +/// Lifetime of a UDP connection id in seconds. BEP 15 suggests two minutes. +const CONNECTION_ID_WINDOW_SECS: u64 = 120; + +/// Per-process secret keying the connection-id hash. Regenerated on every start, so ids do +/// not survive a restart and cannot be precomputed by a client. +static CONNECTION_ID_SECRET: LazyLock = LazyLock::new(ahash::RandomState::new); + +/// Byte range of the 40-character hex announce key inside `/announce//`. +const KEY_PATH_RANGE: std::ops::Range = 10..50; +/// Byte range of the 40-character hex user key inside `/announce//`. +const USER_KEY_PATH_RANGE: std::ops::Range = 51..91; + impl UdpServer { /// Binds the UDP tracker sockets for `bind_address` and prepares the configured receive /// backend (blocking recv, `recvmmsg`, `io_uring` or Windows RIO). @@ -57,7 +72,7 @@ impl UdpServer { /// /// Returns the I/O error when the socket cannot be bound or configured. #[allow(clippy::too_many_arguments)] - pub async fn new(tracker: Arc, bind_address: SocketAddr, udp_threads: usize, worker_threads: usize, recv_buffer_size: usize, send_buffer_size: usize, reuse_address: bool, use_payload_ip: bool, simple_proxy_protocol: bool, receive_method: UdpReceiveMethod) -> tokio::io::Result + pub async fn new(tracker: Arc, bind_address: SocketAddr, udp_threads: usize, worker_threads: usize, recv_buffer_size: usize, send_buffer_size: usize, reuse_address: bool, use_payload_ip: bool, simple_proxy_protocol: bool, proxy_addrs: Arc>, receive_method: UdpReceiveMethod) -> tokio::io::Result { #[cfg(windows)] let use_rio = receive_method == UdpReceiveMethod::rio && { @@ -86,6 +101,7 @@ impl UdpServer { tracker, use_payload_ip, simple_proxy_protocol, + proxy_addrs, receive_method, }) } @@ -128,7 +144,7 @@ impl UdpServer { /// Runs the UDP receive/parse/respond loops until the shutdown watch channel fires. pub async fn start(&self, mut rx: tokio::sync::watch::Receiver) { let parse_pool = Arc::new(ParsePool::new(1_000_000, self.worker_threads)); - parse_pool.start_thread(self.worker_threads, self.tracker.clone(), rx.clone(), self.use_payload_ip, self.simple_proxy_protocol).await; + parse_pool.start_thread(self.worker_threads, self.tracker.clone(), rx.clone(), self.use_payload_ip, self.simple_proxy_protocol, self.proxy_addrs.clone()).await; let payload = parse_pool.payload.clone(); let tracker_queue = self.tracker.clone(); let mut rx_queue = rx.clone(); @@ -339,33 +355,50 @@ impl UdpServer { } } - /// Derives the BEP 15 connection id for a client address from the current time window, - /// so ids expire automatically without server-side state. - pub async fn get_connection_id(remote_address: &SocketAddr) -> ConnectionId { + /// Derives the BEP 15 connection id for a client address in a given time window. + /// + /// A keyed hash of `(window, port, ip)`, so a client cannot forge one and ids expire with + /// the window without the tracker keeping per-client state. + fn derive_connection_id(remote_address: &SocketAddr, window: u64) -> ConnectionId { use std::hash::{ - DefaultHasher, + BuildHasher, Hasher }; - use std::time::{ - SystemTime, - UNIX_EPOCH - }; - let mut hasher = DefaultHasher::new(); - let timestamp = SystemTime::now() - .duration_since(UNIX_EPOCH) - .unwrap_or_default() - .as_nanos() as u64; - hasher.write_u64(timestamp); + let mut hasher = CONNECTION_ID_SECRET.build_hasher(); + hasher.write_u64(window); hasher.write_u16(remote_address.port()); - if let std::net::IpAddr::V4(ipv4) = remote_address.ip() { - hasher.write(&ipv4.octets()); - } else if let std::net::IpAddr::V6(ipv6) = remote_address.ip() { - hasher.write(&ipv6.octets()); + match remote_address.ip() { + std::net::IpAddr::V4(ipv4) => hasher.write(&ipv4.octets()), + std::net::IpAddr::V6(ipv6) => hasher.write(&ipv6.octets()), } ConnectionId(hasher.finish() as i64) } + /// Returns the current time window index used to derive connection ids. + #[inline] + fn connection_id_window() -> u64 { + crate::common::common::current_time() / CONNECTION_ID_WINDOW_SECS + } + + /// Derives the BEP 15 connection id to hand a client in response to a connect request. + pub async fn get_connection_id(remote_address: &SocketAddr) -> ConnectionId { + Self::derive_connection_id(remote_address, Self::connection_id_window()) + } + + /// Checks a connection id supplied in an announce or scrape request against the ids this + /// tracker would have issued to `remote_address`. + /// + /// Required by BEP 15: without it the tracker answers announces from forged source + /// addresses, acting as a UDP reflector. + #[inline] + pub fn connection_id_valid(remote_address: &SocketAddr, connection_id: ConnectionId) -> bool { + let window = Self::connection_id_window(); + // The previous window is accepted so an id stays usable across a window boundary. + connection_id == Self::derive_connection_id(remote_address, window) + || connection_id == Self::derive_connection_id(remote_address, window.wrapping_sub(1)) + } + /// Parses one datagram and produces the tracker response, mapping malformed input and /// handler failures to BEP 15 error responses. pub async fn handle_packet(remote_addr: SocketAddr, payload: &[u8], tracker: Arc, use_payload_ip: bool) -> Response { @@ -471,6 +504,10 @@ impl UdpServer { /// /// Returns a [`ServerError`] when access rules reject the request or the swarm update fails. pub async fn handle_udp_announce(remote_addr: SocketAddr, request: &AnnounceRequest, tracker: Arc, use_payload_ip: bool) -> Result { + if !Self::connection_id_valid(&remote_addr, request.connection_id) { + debug!("[UDP ERROR] Invalid connection id from {remote_addr}"); + return Err(ServerError::InvalidConnectionId); + } let config = &tracker.config.tracker_config; let effective_remote_addr = if use_payload_ip { if let Some(payload_ip) = request.ip_address { @@ -489,49 +526,41 @@ impl UdpServer { debug!("[UDP ERROR] Torrent Blacklisted"); return Err(ServerError::TorrentBlacklisted); } + // Slice the bytes, not the String: `path` is arbitrary client-supplied UTF-8, and a + // String slice at a byte offset inside a multi-byte character panics. if config.keys_enabled { - if request.path.len() < 50 { - debug!("[UDP ERROR] Unknown Key"); + let Some(key_hex) = request.path.as_bytes().get(KEY_PATH_RANGE) else { + debug!("[UDP ERROR] Unknown Key - path too short"); return Err(ServerError::UnknownKey); - } - let key_path_extract = &request.path[10..50]; - if let Ok(result) = hex::decode(key_path_extract) { - if result.len() >= 20 { - let key = <[u8; 20]>::try_from(&result[0..20]).unwrap(); + }; + match hex::decode(key_hex) { + Ok(result) if result.len() >= 20 => { + let key: [u8; 20] = result[..20].try_into().expect("length checked above"); if !tracker.check_key(InfoHash::from(key)) { debug!("[UDP ERROR] Unknown Key"); return Err(ServerError::UnknownKey); } - } else { - debug!("[UDP ERROR] Unknown Key - insufficient bytes"); + } + _ => { + debug!("[UDP ERROR] Unknown Key - not valid hex"); return Err(ServerError::UnknownKey); } - } else { - debug!("[UDP ERROR] Unknown Key"); - return Err(ServerError::UnknownKey); } } let user_key = if config.users_enabled { - let user_key_path_extract = if request.path.len() >= 91 { - Some(&request.path[51..=91]) - } else if !config.users_enabled && request.path.len() >= 50 { - Some(&request.path[10..=50]) - } else { - None + let Some(user_key_hex) = request.path.as_bytes().get(USER_KEY_PATH_RANGE) else { + debug!("[UDP ERROR] Peer Key Not Valid - path too short"); + return Err(ServerError::PeerKeyNotValid); }; - if let Some(path) = user_key_path_extract { - match hex::decode(path) { - Ok(result) if result.len() >= 20 => { - let key = <[u8; 20]>::try_from(&result[0..20]).unwrap(); - tracker.check_user_key(UserId::from(key)) - } - _ => { - debug!("[UDP ERROR] Peer Key Not Valid"); - return Err(ServerError::PeerKeyNotValid); - } + match hex::decode(user_key_hex) { + Ok(result) if result.len() >= 20 => { + let key: [u8; 20] = result[..20].try_into().expect("length checked above"); + tracker.check_user_key(UserId::from(key)) + } + _ => { + debug!("[UDP ERROR] Peer Key Not Valid"); + return Err(ServerError::PeerKeyNotValid); } - } else { - None } } else { None @@ -640,6 +669,10 @@ impl UdpServer { /// /// Currently infallible; kept as `Result` for interface symmetry. pub async fn handle_udp_scrape(remote_addr: SocketAddr, request: &ScrapeRequest, tracker: Arc) -> Result { + if !Self::connection_id_valid(&remote_addr, request.connection_id) { + debug!("[UDP ERROR] Invalid connection id from {remote_addr}"); + return Err(ServerError::InvalidConnectionId); + } let mut torrent_stats = Vec::with_capacity(request.info_hashes.len()); for info_hash in &request.info_hashes { let scrape_entry = match tracker.get_torrent_counts(InfoHash(info_hash.0)) { diff --git a/src/udp/structs/udp_server.rs b/src/udp/structs/udp_server.rs index bae13341..58a6367c 100644 --- a/src/udp/structs/udp_server.rs +++ b/src/udp/structs/udp_server.rs @@ -1,4 +1,7 @@ -use std::net::SocketAddr; +use std::net::{ + IpAddr, + SocketAddr +}; use std::sync::Arc; use tokio::net::UdpSocket; use crate::config::enums::udp_receive_method::UdpReceiveMethod; @@ -16,5 +19,8 @@ pub struct UdpServer { pub(crate) tracker: Arc, pub(crate) use_payload_ip: bool, pub(crate) simple_proxy_protocol: bool, + /// Sources permitted to prepend a Simple Proxy Protocol header. Empty means "any", + /// which is the legacy behaviour. + pub(crate) proxy_addrs: Arc>, pub(crate) receive_method: UdpReceiveMethod, } \ No newline at end of file diff --git a/src/udp/udp.rs b/src/udp/udp.rs index 0351dc99..f160bfbe 100644 --- a/src/udp/udp.rs +++ b/src/udp/udp.rs @@ -49,9 +49,9 @@ pub fn read_be(r: &mut impl Read) -> Result<[u8; N], Error> { /// Spawns the UDP tracker service on `addr` using the selected receive backend and returns /// its join handle. The service runs until the shutdown watch channel fires. #[allow(clippy::too_many_arguments)] -pub async fn udp_service(addr: SocketAddr, udp_threads: usize, worker_threads: usize, recv_buffer_size: usize, send_buffer_size: usize, reuse_address: bool, use_payload_ip: bool, simple_proxy_protocol: bool, receive_method: UdpReceiveMethod, data: Arc, rx: tokio::sync::watch::Receiver, tokio_udp: Arc) -> JoinHandle<()> +pub async fn udp_service(addr: SocketAddr, udp_threads: usize, worker_threads: usize, recv_buffer_size: usize, send_buffer_size: usize, reuse_address: bool, use_payload_ip: bool, simple_proxy_protocol: bool, proxy_addrs: Arc>, receive_method: UdpReceiveMethod, data: Arc, rx: tokio::sync::watch::Receiver, tokio_udp: Arc) -> JoinHandle<()> { - let udp_server = UdpServer::new(data, addr, udp_threads, worker_threads, recv_buffer_size, send_buffer_size, reuse_address, use_payload_ip, simple_proxy_protocol, receive_method).await.unwrap_or_else(|e| { + let udp_server = UdpServer::new(data, addr, udp_threads, worker_threads, recv_buffer_size, send_buffer_size, reuse_address, use_payload_ip, simple_proxy_protocol, proxy_addrs, receive_method).await.unwrap_or_else(|e| { error!("Could not listen to the UDP port: {e}"); exit(1); }); diff --git a/src/websocket/websocket.rs b/src/websocket/websocket.rs index 715b9fb1..d0252f89 100644 --- a/src/websocket/websocket.rs +++ b/src/websocket/websocket.rs @@ -177,8 +177,9 @@ pub async fn process_announce(tracker: &Arc, request: &ClusterRe }; if announce.rtctorrent.unwrap_or(false) { let rtc_interval = tracker_config.rtc_interval as i64; - let seeds_count = torrent_entry.rtc_seeds.len() as i64; - let peers_count = torrent_entry.rtc_peers.len() as i64; + // Counts come from `counts`, never from the peer maps: those are bounded copies. + let seeds_count = torrent_entry.counts.rtc_seeds as i64; + let peers_count = torrent_entry.counts.rtc_peers as i64; let completed_count = torrent_entry.completed as i64; let mut rtc_peers_list = ben_list!(); { @@ -255,7 +256,6 @@ pub fn build_compact_announce_response( stats: &AnnounceResponseStats, ) -> Vec { let mut peers_list: Vec = Vec::with_capacity(72 * 6); - let port_bytes = announce.port.to_be_bytes(); match client_ip { IpAddr::V4(_) => { if announce.left != 0 { @@ -268,7 +268,7 @@ pub fn build_compact_announce_response( for torrent_peer in seeds.values() { if let IpAddr::V4(ipv4) = torrent_peer.peer_addr.ip() { let _ = peers_list.write(&ipv4.octets()); - let _ = peers_list.write(&port_bytes); + let _ = peers_list.write(&torrent_peer.peer_addr.port().to_be_bytes()); } } } @@ -285,7 +285,7 @@ pub fn build_compact_announce_response( } if let IpAddr::V4(ipv4) = torrent_peer.peer_addr.ip() { let _ = peers_list.write(&ipv4.octets()); - let _ = peers_list.write(&port_bytes); + let _ = peers_list.write(&torrent_peer.peer_addr.port().to_be_bytes()); } } } @@ -309,7 +309,7 @@ pub fn build_compact_announce_response( for torrent_peer in seeds.values() { if let IpAddr::V6(ipv6) = torrent_peer.peer_addr.ip() { let _ = peers_list.write(&ipv6.octets()); - let _ = peers_list.write(&port_bytes); + let _ = peers_list.write(&torrent_peer.peer_addr.port().to_be_bytes()); } } } @@ -326,7 +326,7 @@ pub fn build_compact_announce_response( } if let IpAddr::V6(ipv6) = torrent_peer.peer_addr.ip() { let _ = peers_list.write(&ipv6.octets()); - let _ = peers_list.write(&port_bytes); + let _ = peers_list.write(&torrent_peer.peer_addr.port().to_be_bytes()); } } } diff --git a/tests/common/mod.rs b/tests/common/mod.rs index 153d5441..31077abb 100644 --- a/tests/common/mod.rs +++ b/tests/common/mod.rs @@ -11,10 +11,16 @@ pub type TestTracker = Arc; pub type TestConfig = Arc; pub async fn create_test_config() -> TestConfig { + Arc::new(build_test_config(|_| {})) +} + +/// Builds a test configuration, letting the caller flip the settings a test needs. +pub fn build_test_config(customise: impl FnOnce(&mut Configuration)) -> Configuration { let mut config: Configuration = Configuration::init(); config.database.path = ":memory:".to_string(); config.database.persistent = false; - Arc::new(config) + customise(&mut config); + config } pub fn create_test_http_config() -> Arc { @@ -27,6 +33,8 @@ pub fn create_test_http_config_with_rtctorrent(rtctorrent: bool) -> Arc Arc { bind_address: "127.0.0.1:8081".to_string(), real_ip: String::new(), trusted_proxies: false, + trusted_proxy_ips: Vec::new(), + trusted_proxy_addrs: Vec::new(), keep_alive: 5, request_timeout: 10, disconnect_timeout: 5, diff --git a/tests/config_tests.rs b/tests/config_tests.rs index 23d83885..b744c897 100644 --- a/tests/config_tests.rs +++ b/tests/config_tests.rs @@ -2,6 +2,7 @@ mod common; use std::fs; use tempfile::TempDir; +use torrust_actix::config::structs::configuration::Configuration; #[tokio::test] async fn test_config_default_values() { @@ -411,4 +412,94 @@ rtctorrent = true cfg.http_server[0].rtctorrent, "'rtctorrent = true' in TOML should deserialise as true" ); -} \ No newline at end of file +} +#[tokio::test] +async fn test_generated_config_round_trips() { + // Every field the defaults emit has to parse back, and `#[serde(skip)]` fields must not + // appear at all. + let defaults = Configuration::init(); + let annotated = Configuration::generate_annotated_config(&defaults); + assert!(annotated.contains("max_peers_per_torrent"), "new tracker limits must be written out"); + assert!(annotated.contains("trusted_proxy_ips"), "proxy allowlist must be written out"); + assert!(annotated.contains("proxy_addresses"), "UDP proxy allowlist must be written out"); + assert!(!annotated.contains("trusted_proxy_addrs"), "the parsed-only field must not be serialised"); + assert!(!annotated.contains("proxy_addrs ="), "the parsed-only field must not be serialised"); + + let reparsed: Configuration = toml::from_str(&annotated).expect("generated config must parse"); + assert_eq!(reparsed.tracker_config.max_peers_per_torrent, defaults.tracker_config.max_peers_per_torrent); + assert_eq!(reparsed.tracker_config.max_rtc_pending_answers, defaults.tracker_config.max_rtc_pending_answers); + assert_eq!(reparsed.http_server[0].trusted_proxy_ips, defaults.http_server[0].trusted_proxy_ips); + assert_eq!(reparsed.udp_server[0].proxy_addresses, defaults.udp_server[0].proxy_addresses); +} + +#[tokio::test] +async fn test_new_limits_default_when_absent_from_toml() { + // Config files predating these keys must fall back to safe defaults, not a disabled limit. + let toml_str = r#" +log_level = "info" +log_console_interval = 60 + +[tracker_config] +api_key = "SomeVeryStrongApiKey123456789abc" +request_interval = 1800 +request_interval_minimum = 1800 +peers_timeout = 2700 +peers_cleanup_interval = 900 +peers_cleanup_threads = 256 +total_downloads = 0 + +[database] +engine = "sqlite3" +path = "sqlite://data.db" +persistent = false +persistent_interval = 60 +insert_vacant = false +remove_action = false +update_completed = true +update_peers = false + +[database_structure] + +[[http_server]] +enabled = true +bind_address = "0.0.0.0:6969" +real_ip = "X-Real-IP" +keep_alive = 60 +request_timeout = 15 +disconnect_timeout = 15 +max_connections = 25000 +threads = 4 +ssl = false +ssl_key = "" +ssl_cert = "" +tls_connection_rate = 256 + +[[udp_server]] +enabled = true +bind_address = "0.0.0.0:6969" +udp_threads = 2 +worker_threads = 4 +receive_buffer_size = 134217728 +send_buffer_size = 67108864 +reuse_address = true + +[[api_server]] +enabled = true +bind_address = "0.0.0.0:8080" +real_ip = "X-Real-IP" +keep_alive = 60 +request_timeout = 30 +disconnect_timeout = 30 +max_connections = 25000 +threads = 4 +ssl = false +ssl_key = "" +ssl_cert = "" +tls_connection_rate = 256 +"#; + let config: Configuration = toml::from_str(toml_str).expect("legacy config must still parse"); + assert_eq!(config.tracker_config.max_peers_per_torrent, 10_000, "peer cap must default on, not off"); + assert_eq!(config.tracker_config.max_rtc_pending_answers, 32); + assert!(config.http_server[0].trusted_proxy_ips.is_empty()); + assert!(config.udp_server[0].proxy_addresses.is_empty()); +} diff --git a/tests/tracker_tests.rs b/tests/tracker_tests.rs index daaea923..939e5822 100644 --- a/tests/tracker_tests.rs +++ b/tests/tracker_tests.rs @@ -472,4 +472,165 @@ async fn test_announce_entry_counts_exact_when_peer_map_capped() { assert_eq!(snapshot.counts.total_peers(), total, "total_peers must be exact"); assert!(snapshot.peers.len() <= SNAPSHOT_PEER_CAP, "returned peer map must be capped"); assert!(snapshot.peers.len() >= 72, "bounded map must still cover a full response"); -} \ No newline at end of file +} +#[tokio::test] +async fn test_peer_map_is_capped_per_torrent() { + use std::sync::Arc; + use torrust_actix::tracker::structs::torrent_tracker::TorrentTracker; + + // Peer ids are client-chosen, so an uncapped map grows once per id announced. + const CAP: u64 = 8; + let config = Arc::new(common::build_test_config(|config| { + config.tracker_config.max_peers_per_torrent = CAP; + })); + let tracker = Arc::new(TorrentTracker::new(config, false).await); + let info_hash = common::random_info_hash(); + + // Seed the torrent so later announces take the "existing torrent" path. + let first = common::random_peer_id(); + tracker.add_torrent_peer(info_hash, first, common::create_test_peer(first, IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)), 6881), false); + + for _ in 0..(CAP * 5) { + let peer_id = common::random_peer_id(); + let peer = common::create_test_peer(peer_id, IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)), 6881); + tracker.add_torrent_peer(info_hash, peer_id, peer, false); + } + + let entry = tracker.get_torrent(info_hash).expect("torrent should exist"); + assert_eq!(entry.peers.len() as u64, CAP, "leecher map must stay at the cap"); + + // The global counter has to track evictions or it drifts. + let stats = tracker.get_stats(); + assert_eq!(stats.peers, CAP as i64, "peer statistic must match the peers actually held"); +} + +#[tokio::test] +async fn test_peer_cap_refreshes_existing_peer_without_evicting() { + use std::sync::Arc; + use torrust_actix::tracker::structs::torrent_tracker::TorrentTracker; + + const CAP: u64 = 4; + let config = Arc::new(common::build_test_config(|config| { + config.tracker_config.max_peers_per_torrent = CAP; + })); + let tracker = Arc::new(TorrentTracker::new(config, false).await); + let info_hash = common::random_info_hash(); + + let peer_ids: Vec<_> = (0..CAP).map(|_| common::random_peer_id()).collect(); + for peer_id in &peer_ids { + let peer = common::create_test_peer(*peer_id, IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)), 6881); + tracker.add_torrent_peer(info_hash, *peer_id, peer, false); + } + assert_eq!(tracker.get_torrent(info_hash).unwrap().peers.len() as u64, CAP); + + // A peer already in the map is a refresh, not an insert, so nobody is pushed out. + let peer = common::create_test_peer(peer_ids[0], IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)), 6882); + tracker.add_torrent_peer(info_hash, peer_ids[0], peer, false); + let entry = tracker.get_torrent(info_hash).unwrap(); + assert_eq!(entry.peers.len() as u64, CAP, "a refresh must not change the map size"); + for peer_id in &peer_ids { + assert!(entry.peers.contains_key(peer_id), "no peer should have been evicted"); + } + assert_eq!(tracker.get_stats().peers, CAP as i64); +} + +#[tokio::test] +async fn test_rtc_pending_answers_are_capped() { + use std::sync::Arc; + use torrust_actix::tracker::structs::torrent_tracker::TorrentTracker; + + // Filled by *other* peers naming this one, so the cap is what stops anyone making the + // tracker hold arbitrary SDP on a victim's behalf. + const CAP: u64 = 4; + let config = Arc::new(common::build_test_config(|config| { + config.tracker_config.max_rtc_pending_answers = CAP; + })); + let tracker = Arc::new(TorrentTracker::new(config, false).await); + let info_hash = common::random_info_hash(); + let seeder_id = common::random_peer_id(); + + let mut seeder = common::create_test_peer(seeder_id, IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)), 6881); + seeder.left = NumberOfBytes(0); + seeder.rtc_data = Some(Box::new(torrust_actix::tracker::structs::rtc_data::RtcData::new(Some("v=0")))); + tracker.add_torrent_peer(info_hash, seeder_id, seeder, false); + + for _ in 0..(CAP * 10) { + assert!(tracker.store_rtc_answer(info_hash, seeder_id, common::random_peer_id(), "v=0\r\nanswer")); + } + let queued = tracker.take_rtc_pending_answers(info_hash, seeder_id); + assert_eq!(queued.len() as u64, CAP, "pending answers must not grow past the cap"); + + assert!(tracker.take_rtc_pending_answers(info_hash, seeder_id).is_empty()); +} + +#[tokio::test] +async fn test_rtc_pending_answer_replaces_same_peer() { + use std::sync::Arc; + use torrust_actix::tracker::structs::torrent_tracker::TorrentTracker; + + let config = Arc::new(common::build_test_config(|config| { + config.tracker_config.max_rtc_pending_answers = 4; + })); + let tracker = Arc::new(TorrentTracker::new(config, false).await); + let info_hash = common::random_info_hash(); + let seeder_id = common::random_peer_id(); + let answerer_id = common::random_peer_id(); + + let mut seeder = common::create_test_peer(seeder_id, IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)), 6881); + seeder.left = NumberOfBytes(0); + seeder.rtc_data = Some(Box::new(torrust_actix::tracker::structs::rtc_data::RtcData::new(Some("v=0")))); + tracker.add_torrent_peer(info_hash, seeder_id, seeder, false); + + tracker.store_rtc_answer(info_hash, seeder_id, answerer_id, "first"); + tracker.store_rtc_answer(info_hash, seeder_id, answerer_id, "second"); + let queued = tracker.take_rtc_pending_answers(info_hash, seeder_id); + assert_eq!(queued.len(), 1, "the same answerer must occupy one slot"); + assert_eq!(queued[0].1, "second", "the newest answer must win"); +} + +#[tokio::test] +async fn test_cleanup_survives_timeout_longer_than_uptime() { + use std::sync::Arc; + use torrust_actix::tracker::structs::torrent_tracker::TorrentTracker; + + let config = Arc::new(common::build_test_config(|_| {})); + let tracker = Arc::new(TorrentTracker::new(config, false).await); + let info_hash = common::random_info_hash(); + let peer_id = common::random_peer_id(); + tracker.add_torrent_peer(info_hash, peer_id, common::create_test_peer(peer_id, IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)), 6881), false); + + // `Instant` counts from boot on Linux, so a timeout longer than uptime underflows. + let absurd_timeout = Duration::from_secs(60 * 60 * 24 * 365 * 100); + TorrentSharding::cleanup_once(tracker.clone(), absurd_timeout, absurd_timeout, false).await; + + let entry = tracker.get_torrent(info_hash).expect("torrent must survive the skipped pass"); + assert_eq!(entry.peers.len(), 1, "no peer should have been removed"); +} + +#[tokio::test] +async fn test_torrent_updates_dedupe_per_info_hash() { + use torrust_actix::tracker::enums::updates_action::UpdatesAction; + use torrust_actix::tracker::structs::torrent_update_data::TorrentUpdateData; + + let tracker: common::TestTracker = common::create_test_tracker().await; + let info_hash = common::random_info_hash(); + + // Repeat updates for one torrent collapse, so queue size tracks distinct torrents. + assert!(tracker.add_torrent_update(info_hash, TorrentUpdateData { seeds_ipv4: 1, ..Default::default() }, UpdatesAction::Add)); + for seeds in 2..50u64 { + assert!(!tracker.add_torrent_update(info_hash, TorrentUpdateData { seeds_ipv4: seeds, ..Default::default() }, UpdatesAction::Add)); + } + let pending = tracker.get_torrent_updates(); + assert_eq!(pending.len(), 1, "one torrent must occupy one queue slot"); + assert_eq!(pending[&info_hash].0.seeds_ipv4, 49, "the newest update must win"); + assert_eq!(tracker.get_stats().torrents_updates, 1, "statistic must match the queue length"); + + let other = common::random_info_hash(); + assert!(tracker.add_torrent_update(other, TorrentUpdateData::default(), UpdatesAction::Add)); + assert_eq!(tracker.get_torrent_updates().len(), 2, "distinct torrents get distinct slots"); + assert_eq!(tracker.get_stats().torrents_updates, 2); + + tracker.clear_torrent_updates(); + assert!(tracker.get_torrent_updates().is_empty()); + assert_eq!(tracker.get_stats().torrents_updates, 0); +} diff --git a/tests/udp_tests.rs b/tests/udp_tests.rs index 20e10edb..ff08a735 100644 --- a/tests/udp_tests.rs +++ b/tests/udp_tests.rs @@ -1,5 +1,6 @@ mod common; +use std::sync::Arc; use std::net::{ IpAddr, Ipv4Addr, @@ -134,17 +135,133 @@ fn test_response_estimated_size() { } #[tokio::test] -async fn test_connection_id_generation() { +async fn test_connection_id_is_stable_per_client_within_window() { use torrust_actix::udp::structs::udp_server::UdpServer; + // An id must be reproducible per client, or the tracker cannot verify the one echoed back. let addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)), 6881); let conn_id1 = UdpServer::get_connection_id(&addr).await; tokio::time::sleep(tokio::time::Duration::from_millis(10)).await; let conn_id2 = UdpServer::get_connection_id(&addr).await; - assert_ne!(conn_id1.0, conn_id2.0, "Connection IDs should be unique"); + assert_eq!(conn_id1.0, conn_id2.0, "Connection ID should be stable within a time window"); +} + +#[tokio::test] +async fn test_connection_id_differs_per_client() { + use torrust_actix::udp::structs::udp_server::UdpServer; + + let addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)), 6881); + let other_port = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)), 6882); + let other_ip = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(127, 0, 0, 2)), 6881); + let base = UdpServer::get_connection_id(&addr).await; + assert_ne!(base.0, UdpServer::get_connection_id(&other_port).await.0); + assert_ne!(base.0, UdpServer::get_connection_id(&other_ip).await.0); +} + +#[tokio::test] +async fn test_connection_id_validation_rejects_forged_ids() { + use torrust_actix::udp::structs::connection_id::ConnectionId; + use torrust_actix::udp::structs::udp_server::UdpServer; + + let addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(203, 0, 113, 5)), 6881); + let issued = UdpServer::get_connection_id(&addr).await; + assert!(UdpServer::connection_id_valid(&addr, issued), "an id we issued must validate"); + assert!(!UdpServer::connection_id_valid(&addr, ConnectionId(0)), "a guessed id must not validate"); + assert!( + !UdpServer::connection_id_valid(&addr, ConnectionId(issued.0 ^ 1)), + "a tampered id must not validate" + ); + + // Ids are bound to their address, which is what stops spoofed announces being answered. + let victim = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(198, 51, 100, 9)), 6881); + assert!( + !UdpServer::connection_id_valid(&victim, issued), + "an id issued to one address must not validate for another" + ); } #[test] fn test_protocol_identifier_constant() { assert_eq!(PROTOCOL_IDENTIFIER, 0x41727101980, "Protocol ID should match BEP 15 spec"); -} \ No newline at end of file +} +/// Builds a BEP 15 announce datagram, optionally with a trailing option-2 (path) field. +fn build_announce_packet(connection_id: i64, path: Option<&[u8]>) -> Vec { + let mut packet = Vec::with_capacity(120); + packet.extend_from_slice(&connection_id.to_be_bytes()); + packet.extend_from_slice(&1i32.to_be_bytes()); // action: announce + packet.extend_from_slice(&7777i32.to_be_bytes()); // transaction id + packet.extend_from_slice(&[0xAAu8; 20]); // info_hash + packet.extend_from_slice(&[0xBBu8; 20]); // peer_id + packet.extend_from_slice(&0i64.to_be_bytes()); // downloaded + packet.extend_from_slice(&100i64.to_be_bytes()); // left + packet.extend_from_slice(&0i64.to_be_bytes()); // uploaded + packet.extend_from_slice(&0i32.to_be_bytes()); // event + packet.extend_from_slice(&[0u8; 4]); // ip + packet.extend_from_slice(&0u32.to_be_bytes()); // key + packet.extend_from_slice(&72i32.to_be_bytes()); // peers wanted + packet.extend_from_slice(&6881u16.to_be_bytes()); // port + if let Some(path) = path { + packet.push(2); // option: URL data + packet.push(path.len() as u8); + packet.extend_from_slice(path); + } + packet +} + +#[tokio::test] +async fn test_udp_announce_requires_valid_connection_id() { + use torrust_actix::tracker::structs::torrent_tracker::TorrentTracker; + use torrust_actix::udp::enums::response::Response; + use torrust_actix::udp::structs::udp_server::UdpServer; + + let config = Arc::new(common::build_test_config(|_| {})); + let tracker = Arc::new(TorrentTracker::new(config, false).await); + let addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(192, 0, 2, 10)), 6881); + + // An id the tracker never issued must be refused, or it acts as a UDP reflector. + let forged = build_announce_packet(0x0123_4567_89ab_cdef, None); + let response = UdpServer::handle_packet(addr, &forged, tracker.clone(), false).await; + assert!(matches!(response, Response::Error(_)), "forged connection id must be rejected"); + + let issued = UdpServer::get_connection_id(&addr).await; + let valid = build_announce_packet(issued.0, None); + let response = UdpServer::handle_packet(addr, &valid, tracker.clone(), false).await; + assert!(matches!(response, Response::AnnounceIpv4(_)), "issued connection id must be accepted, got {response:?}"); +} + +#[tokio::test] +async fn test_udp_announce_path_slicing_does_not_panic() { + use torrust_actix::tracker::structs::torrent_tracker::TorrentTracker; + use torrust_actix::udp::structs::udp_server::UdpServer; + + // A path whose bytes straddle the key offsets must not panic: under panic = 'abort' one + // datagram would take the whole tracker down. + let config = Arc::new(common::build_test_config(|config| { + config.tracker_config.keys_enabled = true; + config.tracker_config.users_enabled = true; + })); + let tracker = Arc::new(TorrentTracker::new(config, false).await); + let addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(192, 0, 2, 11)), 6881); + let issued = UdpServer::get_connection_id(&addr).await; + + let mut paths: Vec> = Vec::new(); + // A multi-byte character crossing byte offset 50 (the end of the key field). + let mut multibyte = b"a".repeat(49); + multibyte.extend_from_slice("\u{00e9}".as_bytes()); + paths.push(multibyte); + // A multi-byte character crossing byte offset 91 (the end of the user-key field). + let mut multibyte_user = b"b".repeat(90); + multibyte_user.extend_from_slice("\u{00e9}".as_bytes()); + paths.push(multibyte_user); + // Lengths that land exactly on a field boundary. + paths.push(b"c".repeat(50)); + paths.push(b"d".repeat(91)); + paths.push(Vec::new()); + paths.push(b"/announce/".to_vec()); + + for path in paths { + let packet = build_announce_packet(issued.0, Some(&path)); + // Any outcome is fine as long as it is a response and not a panic. + let _ = UdpServer::handle_packet(addr, &packet, tracker.clone(), false).await; + } +} From 5407a2b3ab8baf6f82d4866ed33812acbaa676d0 Mon Sep 17 00:00:00 2001 From: Power2All Date: Sat, 25 Jul 2026 23:32:17 +0200 Subject: [PATCH 2/4] Forgot a version bump --- Cargo.lock | 46 +++++++++++++++++++++---------------- Cargo.toml | 6 ++--- docker/Dockerfile | 2 +- docker/build.bat | 4 ++-- src/udp/impls/udp_server.rs | 2 +- 5 files changed, 33 insertions(+), 27 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 024f5c47..6aad47be 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -70,7 +70,7 @@ dependencies = [ "actix-service", "actix-tls", "actix-utils", - "base64", + "base64 0.22.1", "bitflags", "brotli", "bytes", @@ -215,7 +215,7 @@ dependencies = [ "foldhash", "futures-core", "futures-util", - "impl-more 0.3.2", + "impl-more 0.3.5", "itoa", "language-tags", "log", @@ -572,6 +572,12 @@ version = "0.22.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "72b3254f16251a8381aa12e40e3c4d2f0199f8c6508fbecb9d91f575e0fbb8c6" +[[package]] +name = "base64" +version = "0.23.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b25655df2c3cdd83c5e5b293b88acd880332b2ddadd7c30ac43144fdc0033da9" + [[package]] name = "base64ct" version = "1.8.3" @@ -688,9 +694,9 @@ checksum = "37b2a672a2cb129a2e41c10b1224bb368f9f37a2b16b612598138befd7b37eb5" [[package]] name = "cc" -version = "1.3.0" +version = "1.4.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c89588d05638b5b4594a3348a2d6c20277e43a7f5c5202b05cc56888475a47b8" +checksum = "5add81bb678e6cb321aff7fa0dc7689ad82b112dbc032cea19f91d6b8e3582b9" dependencies = [ "find-msvc-tools", "jobserver", @@ -1220,9 +1226,9 @@ checksum = "92773504d58c093f6de2459af4af33faa518c13451eb8f2b5698ed3d36e7c813" [[package]] name = "either" -version = "1.16.0" +version = "1.17.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "91622ff5e7162018101f2fea40d6ebf4a78bbe5a49736a2020649edf9693679e" +checksum = "9e5e8f6c15a24b9a3ee5efec809ccd006d3b30e8b3bb63c39af737c7f87daa1d" dependencies = [ "serde", ] @@ -1785,7 +1791,7 @@ version = "0.1.20" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "96547c2556ec9d12fb1578c4eaf448b04993e7fb79cbaad930a656880a6bdfa0" dependencies = [ - "base64", + "base64 0.22.1", "bytes", "futures-channel", "futures-util", @@ -1937,9 +1943,9 @@ checksum = "e8a5a9a0ff0086c7a148acb942baaabeadf9504d10400b5a05645853729b9cd2" [[package]] name = "impl-more" -version = "0.3.2" +version = "0.3.5" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "134d2c4324d61664107020b79019cf6a6aec153f0b79bc9619ee9e794a5fb021" +checksum = "277ff51754a3f68f12f58446c5d006aa8baa4914ea273cce24a599cfaff33d4f" [[package]] name = "indexmap" @@ -2594,7 +2600,7 @@ version = "3.0.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1d30c53c26bc5b31a98cd02d20f25a7c8567146caf63ed593a9d87b2775291be" dependencies = [ - "base64", + "base64 0.22.1", "serde_core", ] @@ -2972,7 +2978,7 @@ version = "0.13.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "219c5811de6525e5416c7d5d53bb656d3afdbc6c5af816e0802bcfa42dbdc1c3" dependencies = [ - "base64", + "base64 0.22.1", "bytes", "futures-channel", "futures-core", @@ -3177,9 +3183,9 @@ dependencies = [ [[package]] name = "rustls-pki-types" -version = "1.15.0" +version = "1.15.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "764899a24af3980067ee14bc143654f297b22eaebfe3c7b6b211920a5a59b046" +checksum = "2f4925028c7eb5d1fcdaf196971378ed9d2c1c4efc7dc5d011256f76c99c0a96" dependencies = [ "web-time", "zeroize", @@ -3664,7 +3670,7 @@ version = "0.9.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "05b44e85bf579a8eeb4ceaa77a3a523baf2bf0e9bac7e40f405d537b5d2d5ccb" dependencies = [ - "base64", + "base64 0.22.1", "bytes", "cfg-if", "crc", @@ -3768,7 +3774,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "87a2bdd6e83f6b3ea525ca9fee568030508b58355a43d0b2c1674d5f79dcd65e" dependencies = [ "atoi", - "base64", + "base64 0.22.1", "bitflags", "byteorder", "crc", @@ -4138,7 +4144,7 @@ checksum = "7d56353a2a665ad0f41a421187180aab746c8c325620617ad883a99a1cbe66d2" [[package]] name = "torrust-actix" -version = "4.2.16" +version = "4.2.17" dependencies = [ "actix", "actix-cors", @@ -4146,7 +4152,7 @@ dependencies = [ "actix-web-actors", "ahash", "async-trait", - "base64", + "base64 0.23.0", "bip_bencode", "chrono", "clap", @@ -4374,7 +4380,7 @@ version = "3.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "dea7109cdcd5864d4eeb1b58a1648dc9bf520360d7af16ec26d0a9354bafcfc0" dependencies = [ - "base64", + "base64 0.22.1", "der", "log", "native-tls", @@ -4393,7 +4399,7 @@ version = "0.6.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e994ba84b0bd1b1b0cf92878b7ef898a5c1760108fe7b6010327e274917a808c" dependencies = [ - "base64", + "base64 0.22.1", "http 1.4.2", "httparse", "log", @@ -4460,7 +4466,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d047458f1b5b65237c2f6dc6db136945667f40a7668627b3490b9513a3d43a55" dependencies = [ "actix-web", - "base64", + "base64 0.22.1", "mime_guess", "regex", "rust-embed", diff --git a/Cargo.toml b/Cargo.toml index 6fb73401..0db8b5d3 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -5,7 +5,7 @@ resolver = "3" [package] name = "torrust-actix" -version = "4.2.16" +version = "4.2.17" edition = "2024" rust-version = "1.88.0" license = "MIT" @@ -67,7 +67,7 @@ memcache = { version = "0.20.0", default-features = false } thiserror = "2.0.19" tokio = { version = "1.53.1", features = ["full"] } rand = "0.10.2" -base64 = "0.22.1" +base64 = "0.23.0" tokio-shutdown = "0.1.5" toml = "1.1.3" utoipa-swagger-ui = { version = "9.0.2", features = ["actix-web"] } @@ -106,7 +106,7 @@ harness = false name = "Torrust Actix" identifier = "com.power2all.torrust-actix" icon = ["icon.ico"] -version = "4.2.16" +version = "4.2.17" copyright = "Copyright (c) 2024-2026 Power2All" category = "Public Utility" short_description = "BitTorrent Tracker" diff --git a/docker/Dockerfile b/docker/Dockerfile index 2502933a..85dfabca 100644 --- a/docker/Dockerfile +++ b/docker/Dockerfile @@ -3,7 +3,7 @@ FROM rust:alpine RUN apk update --no-interactive RUN apk add git musl-dev curl pkgconfig openssl-dev openssl-libs-static --no-interactive RUN git clone https://github.com/Power2All/torrust-actix.git /app/torrust-actix -RUN cd /app/torrust-actix && git checkout tags/v4.2.16 +RUN cd /app/torrust-actix && git checkout tags/v4.2.17 WORKDIR /app/torrust-actix RUN cd /app/torrust-actix RUN cargo build --release && rm -Rf target/release/.fingerprint target/release/build target/release/deps target/release/examples target/release/incremental diff --git a/docker/build.bat b/docker/build.bat index c5f453be..774e56e4 100644 --- a/docker/build.bat +++ b/docker/build.bat @@ -1,5 +1,5 @@ @echo off -docker build --no-cache -t power2all/torrust-actix:v4.2.16 -t power2all/torrust-actix:latest . -docker push power2all/torrust-actix:v4.2.16 +docker build --no-cache -t power2all/torrust-actix:v4.2.17 -t power2all/torrust-actix:latest . +docker push power2all/torrust-actix:v4.2.17 docker push power2all/torrust-actix:latest \ No newline at end of file diff --git a/src/udp/impls/udp_server.rs b/src/udp/impls/udp_server.rs index 1b1f468f..0fa38caa 100644 --- a/src/udp/impls/udp_server.rs +++ b/src/udp/impls/udp_server.rs @@ -325,7 +325,7 @@ impl UdpServer { /// Encodes a tracker [`Response`] and sends it to the client, logging failures. pub async fn send_response(tracker: Arc, reply: UdpReply, remote_addr: SocketAddr, response: Response) { - debug!("sending response to: {:?}", &remote_addr); + debug!("sending response to: {remote_addr:?}"); let estimated_size = response.estimated_size(); let mut buffer = Vec::with_capacity(estimated_size); match response.write(&mut buffer) { From f50c46248aa091f800efebfbbbaf84cc8c144d23 Mon Sep 17 00:00:00 2001 From: Power2All Date: Sun, 26 Jul 2026 00:40:14 +0200 Subject: [PATCH 3/4] Applying recommended fixes --- Cargo.lock | 1 + Cargo.toml | 1 + src/config/impls/configuration.rs | 2 +- src/config/structs/tracker_config.rs | 9 ++- src/database/database.rs | 2 +- .../impls/database_connector_mysql.rs | 2 +- .../impls/database_connector_pgsql.rs | 2 +- .../impls/database_connector_sqlite.rs | 2 +- src/database/impls/query_builder.rs | 11 +-- src/security/security.rs | 12 ++- src/security/tests.rs | 13 ++++ src/tracker/impls/torrent_sharding.rs | 30 +++++--- src/tracker/impls/torrent_tracker_keys.rs | 24 ++++-- src/udp/impls/udp_server.rs | 33 ++++---- tests/tracker_tests.rs | 75 +++++++++++++++++++ 15 files changed, 166 insertions(+), 53 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 6aad47be..ba13a7cf 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4172,6 +4172,7 @@ dependencies = [ "rcgen", "redis", "reqwest", + "ring", "rmp-serde", "rustls", "rustls-pemfile", diff --git a/Cargo.toml b/Cargo.toml index 0db8b5d3..dc0b7ac3 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -51,6 +51,7 @@ log = "0.4.33" parking_lot = { version = "0.12.5", features = ["arc_lock", "hardware-lock-elision", "serde", "deadlock_detection"] } percent-encoding = "2.3.2" rcgen = "0.14.7" +ring = "0.17.14" rustls = { version = "0.23.42", default-features = false, features = ["std", "ring"] } rustls-pemfile = "2.2.0" sentry = { version = "0.48.5", default-features = false, features = ["rustls", "backtrace", "contexts", "panic", "transport", "debug-images", "reqwest"] } diff --git a/src/config/impls/configuration.rs b/src/config/impls/configuration.rs index f355f4f1..b7e863cc 100644 --- a/src/config/impls/configuration.rs +++ b/src/config/impls/configuration.rs @@ -1086,7 +1086,7 @@ impl Configuration { remarks.insert(("tracker_config", "keys_enabled"), "# Optional: defaults to false -- require announce keys"); remarks.insert(("tracker_config", "keys_cleanup_interval"), "# Optional: defaults to 60 -- expired-key cleanup interval (seconds)"); remarks.insert(("tracker_config", "users_enabled"), "# Optional: defaults to false -- enable per-user statistics"); - remarks.insert(("tracker_config", "max_peers_per_torrent"), "# Optional: defaults to 10000 -- max peers kept per torrent per peer map (0 = unlimited, not recommended)"); + remarks.insert(("tracker_config", "max_peers_per_torrent"), "# Optional: defaults to 10000 -- max peers per torrent in EACH of its 6 peer maps, so up to 6x this per torrent (0 = unlimited, not recommended)"); remarks.insert(("tracker_config", "max_rtc_pending_answers"), "# Optional: defaults to 32 -- max SDP answers queued for one RtcTorrent peer"); remarks.insert(("tracker_config", "swagger"), "# Optional: defaults to false -- expose Swagger UI at /swagger-ui/"); remarks.insert(("tracker_config", "prometheus_id"), "# Optional: defaults to \"torrust_actix\" -- Prometheus metric label"); diff --git a/src/config/structs/tracker_config.rs b/src/config/structs/tracker_config.rs index a1009d7b..7558c250 100644 --- a/src/config/structs/tracker_config.rs +++ b/src/config/structs/tracker_config.rs @@ -36,10 +36,11 @@ pub struct TrackerConfig { pub peers_cleanup_interval: u64, /// Number of parallel threads used for peer cleanup. pub peers_cleanup_threads: u64, - /// Maximum peers kept per torrent in each peer map (IPv4 seeds, IPv6 seeds, IPv4 - /// leechers, IPv6 leechers, RTC seeds, RTC leechers). When a map is full a stale peer - /// is evicted to make room. `0` disables the limit, leaving memory bounded only by - /// announce rate times `peers_timeout`. + /// Maximum peers kept in *each* of a torrent's six peer maps (IPv4 seeds, IPv6 seeds, IPv4 + /// leechers, IPv6 leechers, RTC seeds, RTC leechers), so a single torrent can hold up to + /// six times this value. When a map is full a stale peer is evicted to make room. + /// `0` disables the limit, leaving memory bounded only by announce rate times + /// `peers_timeout`. #[serde(default = "crate::config::config::default_max_peers_per_torrent")] pub max_peers_per_torrent: u64, /// Maximum SDP answers queued for a single RtcTorrent peer before the oldest is dropped. diff --git a/src/database/database.rs b/src/database/database.rs index b88622a3..64adf8a6 100644 --- a/src/database/database.rs +++ b/src/database/database.rs @@ -40,7 +40,7 @@ pub fn format_hex_select(engine: DatabaseDrivers, column: &str, is_binary: bool) /// SQLite/MySQL, double quotes for PostgreSQL). pub fn quote_identifier(engine: DatabaseDrivers, identifier: &str) -> String { match engine { - DatabaseDrivers::sqlite3 | DatabaseDrivers::mysql => format!("`{identifier}`"), + DatabaseDrivers::sqlite3 | DatabaseDrivers::mysql => format!("`{}`", identifier.replace('`', "``")), DatabaseDrivers::pgsql => format!("\"{}\"", identifier.replace('"', "\"\"")), } } diff --git a/src/database/impls/database_connector_mysql.rs b/src/database/impls/database_connector_mysql.rs index 09916cc3..c7f90d77 100644 --- a/src/database/impls/database_connector_mysql.rs +++ b/src/database/impls/database_connector_mysql.rs @@ -563,7 +563,7 @@ impl DatabaseConnectorMySQL { let hash: [u8; 20] = <[u8; 20]>::try_from(&hex::decode(hash_data).unwrap()[0..20]).unwrap(); let timeout: i64 = result.get(structure.column_timeout.as_str()); - tracker.add_key(InfoHash(hash), timeout); + tracker.add_key_absolute(InfoHash(hash), timeout); hashes += 1; } start += length; diff --git a/src/database/impls/database_connector_pgsql.rs b/src/database/impls/database_connector_pgsql.rs index 3389fd7f..0e2e532c 100644 --- a/src/database/impls/database_connector_pgsql.rs +++ b/src/database/impls/database_connector_pgsql.rs @@ -571,7 +571,7 @@ impl DatabaseConnectorPgSQL { <[u8; 20]>::try_from(&hex::decode(text).unwrap()[0..20]).unwrap() }; let timeout: i64 = result.get(structure.column_timeout.as_str()); - tracker.add_key(InfoHash(hash), timeout); + tracker.add_key_absolute(InfoHash(hash), timeout); hashes += 1; } start += length; diff --git a/src/database/impls/database_connector_sqlite.rs b/src/database/impls/database_connector_sqlite.rs index 2589beb0..e51a6b81 100644 --- a/src/database/impls/database_connector_sqlite.rs +++ b/src/database/impls/database_connector_sqlite.rs @@ -589,7 +589,7 @@ impl DatabaseConnectorSQLite { let hash: [u8; 20] = <[u8; 20]>::try_from(&hex::decode(hash_data).unwrap()[0..20]).unwrap(); let timeout: i64 = result.get(structure.column_timeout.as_str()); - tracker.add_key(InfoHash(hash), timeout); + tracker.add_key_absolute(InfoHash(hash), timeout); hashes += 1; } start += length; diff --git a/src/database/impls/query_builder.rs b/src/database/impls/query_builder.rs index 0ea8d42a..f36720f1 100644 --- a/src/database/impls/query_builder.rs +++ b/src/database/impls/query_builder.rs @@ -7,15 +7,10 @@ impl QueryBuilder { Self { engine } } - /// Quotes a table or column identifier for the bound engine. - /// - /// Quote characters are stripped so an identifier can never break out of its quoting. + /// Quotes a table or column identifier for the bound engine: backticks for SQLite/MySQL, + /// double quotes for PostgreSQL. pub fn quote_identifier(&self, identifier: &str) -> String { - let safe: String = identifier.chars().filter(|c| *c != '`' && *c != '"' && *c != '\0').collect(); - match self.engine { - DatabaseDrivers::sqlite3 | DatabaseDrivers::mysql => format!("`{safe}`"), - DatabaseDrivers::pgsql => safe, - } + crate::database::database::quote_identifier(self.engine, identifier) } /// Formats a hex string as an engine-specific binary literal. diff --git a/src/security/security.rs b/src/security/security.rs index 54aff3b5..9f2cbe49 100644 --- a/src/security/security.rs +++ b/src/security/security.rs @@ -124,8 +124,9 @@ pub fn validate_query_string_length(query: &str) -> Result<(), CustomError> { /// Validates a client-supplied IP string (from a proxy header) before parsing it. /// -/// With `trusted_proxies_enabled = false`, loopback/private/unspecified values are rejected so -/// an untrusted sender cannot claim an internal address. +/// With `trusted_proxies_enabled = false`, loopback, unspecified and private values are +/// rejected (IPv4 private/link-local, IPv6 `fc00::/7` and `fe80::/10`) so an untrusted sender +/// cannot claim an internal address. /// /// # Errors /// @@ -140,7 +141,12 @@ pub fn validate_remote_ip(ip: &str, trusted_proxies_enabled: bool) -> Result<(), ipv4.is_loopback() || ipv4.is_private() || ipv4.is_link_local() || ipv4.is_unspecified() } IpAddr::V6(ipv6) => { - ipv6.is_loopback() || ipv6.is_unspecified() + ipv6.is_loopback() + || ipv6.is_unspecified() + // fc00::/7 and fe80::/10, the IPv6 counterparts of the private and + // link-local ranges rejected above for IPv4. + || ipv6.is_unique_local() + || ipv6.is_unicast_link_local() } }; if is_private { diff --git a/src/security/tests.rs b/src/security/tests.rs index 38c2f13c..3f886006 100644 --- a/src/security/tests.rs +++ b/src/security/tests.rs @@ -104,4 +104,17 @@ mod security_tests { assert!(validate_remote_ip("8.8.8.8", false).is_ok()); assert!(validate_remote_ip("192.168.1.1", true).is_ok()); } + + #[test] + fn test_validate_remote_ip_rejects_internal_ipv6() { + // The IPv6 counterparts of the IPv4 private and link-local ranges rejected above. + assert!(validate_remote_ip("fd00::1", false).is_err()); + assert!(validate_remote_ip("fc00::1", false).is_err()); + assert!(validate_remote_ip("fe80::1", false).is_err()); + assert!(validate_remote_ip("::1", false).is_err()); + assert!(validate_remote_ip("::", false).is_err()); + assert!(validate_remote_ip("2606:4700:4700::1111", false).is_ok()); + // A configured proxy may legitimately forward an internal address. + assert!(validate_remote_ip("fd00::1", true).is_ok()); + } } \ No newline at end of file diff --git a/src/tracker/impls/torrent_sharding.rs b/src/tracker/impls/torrent_sharding.rs index b7b67b34..53cfd8a7 100644 --- a/src/tracker/impls/torrent_sharding.rs +++ b/src/tracker/impls/torrent_sharding.rs @@ -115,18 +115,24 @@ impl TorrentSharding { let (mut torrents_removed, mut seeds_removed, mut peers_removed) = (0u64, 0u64, 0u64); if let Some(shard_arc) = torrent_tracker.torrents_sharding.shards.get(shard as usize) { let now = std::time::Instant::now(); - // `Instant`'s origin is boot time on Linux, so this underflows while uptime is below - // the timeout. Nothing can be expired yet then; unwrapping here would abort. - let (Some(cutoff), Some(rtc_cutoff)) = (now.checked_sub(peer_timeout), now.checked_sub(rtc_peer_timeout)) else { + // `Instant`'s origin is boot time on Linux, so a cutoff underflows while uptime is + // below its timeout. The two are independent: `rtc_peer_timeout` is typically much + // shorter, so RTC peers must still expire while the longer BT cutoff is unavailable. + let cutoff = now.checked_sub(peer_timeout); + let rtc_cutoff = now.checked_sub(rtc_peer_timeout); + if cutoff.is_none() && rtc_cutoff.is_none() { return; - }; + } + // An unavailable cutoff means nothing of that kind can have expired yet. + let bt_expired = |updated: std::time::Instant| cutoff.is_some_and(|c| updated < c); + let rtc_expired = |updated: std::time::Instant| rtc_cutoff.is_some_and(|c| updated < c); let mut expired_full: Vec = Vec::new(); #[allow(clippy::type_complexity)] let mut expired_partial: Vec<(InfoHash, Vec, Vec, Vec, Vec, Vec, Vec)> = Vec::new(); { let shard_read = shard_arc.read(); for (info_hash, torrent_entry) in shard_read.iter() { - if torrent_entry.updated < cutoff { + if bt_expired(torrent_entry.updated) { expired_full.push(*info_hash); continue; } @@ -138,37 +144,37 @@ impl TorrentSharding { let mut expired_rtc_seeds = Vec::new(); let mut expired_rtc_peers = Vec::new(); for (peer_id, torrent_peer) in &torrent_entry.seeds { - if torrent_peer.updated < cutoff { + if bt_expired(torrent_peer.updated) { expired_seeds.push(*peer_id); has_expired = true; } } for (peer_id, torrent_peer) in &torrent_entry.seeds_ipv6 { - if torrent_peer.updated < cutoff { + if bt_expired(torrent_peer.updated) { expired_seeds_ipv6.push(*peer_id); has_expired = true; } } for (peer_id, torrent_peer) in &torrent_entry.peers { - if torrent_peer.updated < cutoff { + if bt_expired(torrent_peer.updated) { expired_peers.push(*peer_id); has_expired = true; } } for (peer_id, torrent_peer) in &torrent_entry.peers_ipv6 { - if torrent_peer.updated < cutoff { + if bt_expired(torrent_peer.updated) { expired_peers_ipv6.push(*peer_id); has_expired = true; } } for (peer_id, torrent_peer) in &torrent_entry.rtc_seeds { - if torrent_peer.updated < rtc_cutoff { + if rtc_expired(torrent_peer.updated) { expired_rtc_seeds.push(*peer_id); has_expired = true; } } for (peer_id, torrent_peer) in &torrent_entry.rtc_peers { - if torrent_peer.updated < rtc_cutoff { + if rtc_expired(torrent_peer.updated) { expired_rtc_peers.push(*peer_id); has_expired = true; } @@ -229,7 +235,7 @@ impl TorrentSharding { for info_hash in &expired_full { if let Entry::Occupied(entry) = shard_write.entry(*info_hash) { let mut entry = entry; - if entry.get().updated >= cutoff { continue; } + if !bt_expired(entry.get().updated) { continue; } let torrent_entry = entry.get_mut(); let seeds_len = torrent_entry.seeds.len() as u64; let seeds_ipv6_len = torrent_entry.seeds_ipv6.len() as u64; diff --git a/src/tracker/impls/torrent_tracker_keys.rs b/src/tracker/impls/torrent_tracker_keys.rs index 64921305..fca7665d 100644 --- a/src/tracker/impls/torrent_tracker_keys.rs +++ b/src/tracker/impls/torrent_tracker_keys.rs @@ -43,25 +43,35 @@ impl TorrentTracker { } } - /// Adds an announce key whose expiry is set to now + `timeout` seconds. + /// Adds an announce key expiring `timeout` seconds from now. /// - /// This stores an absolute timestamp, so `timeout = 0` means "expires now". Only a *stored* - /// expiry of `0`, as loaded from a database row, means permanent. + /// Use [`TorrentTracker::add_key_absolute`] when the expiry is already an absolute + /// timestamp, such as a value read back from the database. /// /// Returns `true` when the key was newly inserted, `false` when it was refreshed. pub fn add_key(&self, hash: InfoHash, timeout: i64) -> bool { - let mut lock = self.keys.write(); let timestamp = SystemTime::now().duration_since(UNIX_EPOCH).unwrap(); - let timeout_unix = timestamp.as_secs() as i64 + timeout; + self.add_key_absolute(hash, timestamp.as_secs() as i64 + timeout) + } + + /// Adds an announce key with an absolute expiry timestamp, stored verbatim. + /// + /// An expiry of `0` means the key never expires; any other value is a Unix timestamp, + /// including ones already in the past. + /// + /// Returns `true` when the key was newly inserted, `false` when it was refreshed. + pub fn add_key_absolute(&self, hash: InfoHash, expiry: i64) -> bool + { + let mut lock = self.keys.write(); match lock.entry(hash) { Entry::Vacant(v) => { self.update_stats(StatsEvent::Key, 1); - v.insert(timeout_unix); + v.insert(expiry); true } Entry::Occupied(mut o) => { - o.insert(timeout_unix); + o.insert(expiry); false } } diff --git a/src/udp/impls/udp_server.rs b/src/udp/impls/udp_server.rs index 0fa38caa..2bac722d 100644 --- a/src/udp/impls/udp_server.rs +++ b/src/udp/impls/udp_server.rs @@ -55,9 +55,17 @@ use tokio::runtime::Builder; /// Lifetime of a UDP connection id in seconds. BEP 15 suggests two minutes. const CONNECTION_ID_WINDOW_SECS: u64 = 120; -/// Per-process secret keying the connection-id hash. Regenerated on every start, so ids do -/// not survive a restart and cannot be precomputed by a client. -static CONNECTION_ID_SECRET: LazyLock = LazyLock::new(ahash::RandomState::new); +/// Per-process HMAC-SHA256 key for connection ids. Regenerated on every start, so ids do not +/// survive a restart. +/// +/// This must be a keyed MAC, not a fast hash: `ahash` and friends are built for hash-map +/// distribution and make no key-recovery guarantee, and recovering the key would let an +/// attacker mint ids for addresses they do not control, restoring the reflection attack the +/// connection id exists to prevent. +static CONNECTION_ID_KEY: LazyLock = LazyLock::new(|| { + ring::hmac::Key::generate(ring::hmac::HMAC_SHA256, &ring::rand::SystemRandom::new()) + .expect("failed to generate the UDP connection-id key from the system RNG") +}); /// Byte range of the 40-character hex announce key inside `/announce//`. const KEY_PATH_RANGE: std::ops::Range = 10..50; @@ -360,19 +368,16 @@ impl UdpServer { /// A keyed hash of `(window, port, ip)`, so a client cannot forge one and ids expire with /// the window without the tracker keeping per-client state. fn derive_connection_id(remote_address: &SocketAddr, window: u64) -> ConnectionId { - use std::hash::{ - BuildHasher, - Hasher - }; - - let mut hasher = CONNECTION_ID_SECRET.build_hasher(); - hasher.write_u64(window); - hasher.write_u16(remote_address.port()); + let mut context = ring::hmac::Context::with_key(&CONNECTION_ID_KEY); + context.update(&window.to_be_bytes()); + context.update(&remote_address.port().to_be_bytes()); match remote_address.ip() { - std::net::IpAddr::V4(ipv4) => hasher.write(&ipv4.octets()), - std::net::IpAddr::V6(ipv6) => hasher.write(&ipv6.octets()), + std::net::IpAddr::V4(ipv4) => context.update(&ipv4.octets()), + std::net::IpAddr::V6(ipv6) => context.update(&ipv6.octets()), } - ConnectionId(hasher.finish() as i64) + let tag = context.sign(); + let truncated: [u8; 8] = tag.as_ref()[..8].try_into().expect("HMAC-SHA256 tag is 32 bytes"); + ConnectionId(i64::from_be_bytes(truncated)) } /// Returns the current time window index used to derive connection ids. diff --git a/tests/tracker_tests.rs b/tests/tracker_tests.rs index 939e5822..c385eb1a 100644 --- a/tests/tracker_tests.rs +++ b/tests/tracker_tests.rs @@ -634,3 +634,78 @@ async fn test_torrent_updates_dedupe_per_info_hash() { assert!(tracker.get_torrent_updates().is_empty()); assert_eq!(tracker.get_stats().torrents_updates, 0); } + +#[tokio::test] +async fn test_keys_loaded_from_database_keep_absolute_expiry() { + use torrust_actix::common::common::current_time; + + let tracker: common::TestTracker = common::create_test_tracker().await; + + // Database rows store an absolute expiry. Feeding one through the relative `add_key` would + // add it to the current time, pushing a valid key decades out and turning a permanent key + // (0) into one that expired the instant it loaded. + let permanent = common::random_info_hash(); + tracker.add_key_absolute(permanent, 0); + assert_eq!(tracker.get_key(permanent).unwrap().1, 0, "expiry must be stored verbatim"); + assert!(tracker.check_key(permanent), "a stored expiry of 0 means permanent"); + + let future = common::random_info_hash(); + let future_expiry = current_time() as i64 + 600; + tracker.add_key_absolute(future, future_expiry); + assert_eq!(tracker.get_key(future).unwrap().1, future_expiry); + assert!(tracker.check_key(future)); + + let past = common::random_info_hash(); + tracker.add_key_absolute(past, current_time() as i64 - 600); + assert!(!tracker.check_key(past), "an already-expired stored key must not validate"); + + // The relative form stays relative for callers that pass a duration. + let relative = common::random_info_hash(); + tracker.add_key(relative, 600); + assert!(tracker.get_key(relative).unwrap().1 >= current_time() as i64 + 599); + assert!(tracker.check_key(relative)); +} + +#[tokio::test] +async fn test_permanent_keys_survive_cleanup() { + let tracker: common::TestTracker = common::create_test_tracker().await; + let permanent = common::random_info_hash(); + let expired = common::random_info_hash(); + tracker.add_key_absolute(permanent, 0); + tracker.add_key_absolute(expired, 1); + + tracker.clean_keys(); + + assert!(tracker.get_key(permanent).is_some(), "permanent keys must not be swept"); + assert!(tracker.get_key(expired).is_none(), "expired keys must be swept"); +} + +#[tokio::test] +async fn test_rtc_peers_expire_while_bt_cutoff_unavailable() { + use std::sync::Arc; + use torrust_actix::tracker::structs::rtc_data::RtcData; + use torrust_actix::tracker::structs::torrent_tracker::TorrentTracker; + + // `rtc_peers_timeout` is far shorter than `peers_timeout`, so shortly after boot the BT + // cutoff underflows while the RTC one is fine. RTC peers must still be swept then. + let config = Arc::new(common::build_test_config(|_| {})); + let tracker = Arc::new(TorrentTracker::new(config, false).await); + let info_hash = common::random_info_hash(); + + let bt_peer_id = common::random_peer_id(); + tracker.add_torrent_peer(info_hash, bt_peer_id, common::create_test_peer(bt_peer_id, IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)), 6881), false); + + let rtc_peer_id = common::random_peer_id(); + let mut rtc_peer = common::create_test_peer(rtc_peer_id, IpAddr::V4(Ipv4Addr::new(127, 0, 0, 2)), 6882); + rtc_peer.rtc_data = Some(Box::new(RtcData::new(Some("v=0")))); + tracker.add_torrent_peer(info_hash, rtc_peer_id, rtc_peer, false); + + tokio::time::sleep(Duration::from_millis(20)).await; + + let unavailable_bt_timeout = Duration::from_secs(60 * 60 * 24 * 365 * 100); + TorrentSharding::cleanup_once(tracker.clone(), unavailable_bt_timeout, Duration::from_millis(1), false).await; + + let entry = tracker.get_torrent(info_hash).expect("torrent must survive"); + assert!(entry.rtc_peers.is_empty(), "RTC peers must expire on their own cutoff"); + assert_eq!(entry.peers.len(), 1, "BT peers must survive an unavailable BT cutoff"); +} From b72607552326b591f8e7dd9b85ea40bfabf32c4b Mon Sep 17 00:00:00 2001 From: Power2All Date: Sun, 26 Jul 2026 01:03:32 +0200 Subject: [PATCH 4/4] Last recommendations --- src/database/impls/query_builder.rs | 20 ++++++++++----- src/database/tests.rs | 40 +++++++++++++++++++++++++++++ 2 files changed, 53 insertions(+), 7 deletions(-) diff --git a/src/database/impls/query_builder.rs b/src/database/impls/query_builder.rs index f36720f1..9d0793ee 100644 --- a/src/database/impls/query_builder.rs +++ b/src/database/impls/query_builder.rs @@ -22,17 +22,23 @@ impl QueryBuilder { } } - /// Formats a value as a single-quoted SQL string literal, escaping embedded quotes (and - /// backslashes on MySQL, which treats them as escapes by default). + /// Formats a value as an SQL string literal for the bound engine. /// /// Statements here are assembled with `format!` and passed to sqlx via `AssertSqlSafe`, so /// nothing downstream escapes anything: literals must be self-contained. + /// + /// PostgreSQL gets an `E''` literal, where backslashes are always escapes, so doubling both + /// backslash and quote is correct whichever way the server has `standard_conforming_strings` + /// set. An ordinary `''` literal would be safe only under the modern default of `on`, and + /// doubling backslashes inside one would corrupt the value. pub fn text_literal(&self, value: &str) -> String { - let escaped = match self.engine { - DatabaseDrivers::mysql => value.replace('\\', "\\\\").replace('\'', "''"), - DatabaseDrivers::sqlite3 | DatabaseDrivers::pgsql => value.replace('\'', "''"), - }; - format!("'{escaped}'") + match self.engine { + // MySQL treats backslash as an escape in ordinary literals unless NO_BACKSLASH_ESCAPES. + DatabaseDrivers::mysql => format!("'{}'", value.replace('\\', "\\\\").replace('\'', "''")), + // SQLite has no backslash escapes at all; doubling the quote is the whole contract. + DatabaseDrivers::sqlite3 => format!("'{}'", value.replace('\'', "''")), + DatabaseDrivers::pgsql => format!("E'{}'", value.replace('\\', "\\\\").replace('\'', "''")), + } } /// Builds the engine-specific upsert conflict clause updating the given columns. diff --git a/src/database/tests.rs b/src/database/tests.rs index 1273f0a6..4365efab 100644 --- a/src/database/tests.rs +++ b/src/database/tests.rs @@ -63,5 +63,45 @@ mod database_tests { assert_eq!(database::limit_offset(DatabaseDrivers::mysql, 100, 50), "LIMIT 100, 50"); assert_eq!(database::limit_offset(DatabaseDrivers::pgsql, 100, 50), "LIMIT 50 OFFSET 100"); } + + #[test] + fn test_quote_identifier_escapes_its_own_quote_character() { + assert_eq!(database::quote_identifier(DatabaseDrivers::sqlite3, "a`b"), "`a``b`"); + assert_eq!(database::quote_identifier(DatabaseDrivers::mysql, "a`b"), "`a``b`"); + assert_eq!(database::quote_identifier(DatabaseDrivers::pgsql, "a\"b"), "\"a\"\"b\""); + } + } + + mod query_builder_tests { + use super::*; + use crate::database::structs::query_builder::QueryBuilder; + + #[test] + fn test_text_literal_escaping_per_engine() { + let value = r"O'Brien\x"; + + // SQLite has no backslash escapes: only the quote is doubled. + assert_eq!(QueryBuilder::new(DatabaseDrivers::sqlite3).text_literal(value), r"'O''Brien\x'"); + + // MySQL treats backslash as an escape by default, so it is doubled too. + assert_eq!(QueryBuilder::new(DatabaseDrivers::mysql).text_literal(value), r"'O''Brien\\x'"); + + // PostgreSQL uses an E'' literal so the result is correct regardless of the + // server's `standard_conforming_strings` setting. + assert_eq!(QueryBuilder::new(DatabaseDrivers::pgsql).text_literal(value), r"E'O''Brien\\x'"); + } + + #[test] + fn test_text_literal_cannot_terminate_the_literal_early() { + // A trailing backslash is the case an ordinary PostgreSQL literal gets wrong when + // `standard_conforming_strings` is off: the backslash would escape the closing quote. + for engine in [DatabaseDrivers::sqlite3, DatabaseDrivers::mysql, DatabaseDrivers::pgsql] { + let literal = QueryBuilder::new(engine).text_literal(r"trailing\"); + assert!(literal.ends_with('\''), "{engine:?} literal must be closed: {literal}"); + if engine != DatabaseDrivers::sqlite3 { + assert!(literal.ends_with(r"\\'"), "{engine:?} must escape the trailing backslash: {literal}"); + } + } + } } } \ No newline at end of file