diff --git a/.gitignore b/.gitignore index 8e1432d..84fd725 100644 --- a/.gitignore +++ b/.gitignore @@ -4,6 +4,7 @@ /.run-data/ /.remote-data/ /data/ +/data-filter-test/ /dht-search.toml **/*.rs.bk diff --git a/TODOS.md b/TODOS.md index 6879de1..b6f5b4c 100644 --- a/TODOS.md +++ b/TODOS.md @@ -213,12 +213,13 @@ - [ ] 统计真实数据的 infohash 重复率和内容重复率 - [ ] 定义可配置的名称路径扩展名和大小过滤规则 -- [ ] 定义 Metadata 最大大小文件数和路径长度限制 -- [ ] 识别空名称异常路径大小溢出和文件数量攻击 +- [x] 定义可配置的 Metadata 最大大小文件数名称路径长度和目录层级限制 +- [x] 识别空名称控制字符异常路径大小溢出总大小不一致和文件数量攻击 - [ ] 设计可解释的名称标准化规则 - [ ] 为模糊相似结果生成聚合候选但不自动删除 - [ ] 支持黑名单规则版本和命中原因 -- [ ] 保留被过滤记录的计数指标但避免保存大内容 +- [x] 使用带规则指纹的 RocksDB 轻量拒绝记录阻止相同异常 infohash 重复下载 +- [x] 保留按原因分类的过滤指标但避免保存名称和大文件列表 - [ ] 增加误判测试和边界数据集 ### 验收标准 @@ -308,6 +309,6 @@ ## 当前下一步 -使用真实采集数据进行 Web 浏览器人工验收并根据使用反馈调整交互细节 +建立可重复的百万级数据基准并测量查询延迟索引速度磁盘占用和内存使用 -随后进行百万级查询基准和更长时间的资源稳定性测试 +随后进行更长时间的真实采集与资源稳定性测试 diff --git a/dht-crawler/CHANGELOG.md b/dht-crawler/CHANGELOG.md index fcd1299..76ce2cb 100644 --- a/dht-crawler/CHANGELOG.md +++ b/dht-crawler/CHANGELOG.md @@ -12,6 +12,7 @@ ### Changed +- Metadata 最大 info 字典大小改为可通过 `max_metadata_size_bytes` 配置，并拒绝负数文件大小和总大小溢出。 - 主动 `get_peers` ingress 改为有界排队，避免突发采样在速率预算耗尽时直接丢弃。 - 默认主动 Peer lookup 提升到每秒 128 个、最多 256 个并发 lookup。 - 空节点池的 Bootstrap 默认改为 30 秒重试、每轮最多 16 个端点，降低坏 DNS diff --git a/dht-crawler/README.md b/dht-crawler/README.md index 6dbc62a..0552c3b 100644 --- a/dht-crawler/README.md +++ b/dht-crawler/README.md @@ -158,6 +158,7 @@ let options = DHTOptions { max_queue_size: 20_000, max_worker_count: 8, max_connects_per_second: 2, + max_metadata_size_bytes: 10 * 1024 * 1024, ..Default::default() }, crawl: CrawlOptions { diff --git a/dht-crawler/src/metadata.rs b/dht-crawler/src/metadata.rs index 91c336c..03fe6a0 100644 --- a/dht-crawler/src/metadata.rs +++ b/dht-crawler/src/metadata.rs @@ -192,6 +192,7 @@ impl PeerFailureCache { /// BEP-9 Metadata fetcher with an end-to-end timeout and shared Peer failure cache. pub struct RbitFetcher { total_timeout: Duration, + max_metadata_size_bytes: usize, runtime_stats: DhtRuntimeStats, peer_failure_cache: Arc, connect_rate_limiter: Arc, @@ -203,18 +204,27 @@ impl RbitFetcher { /// [`DHTServer`](crate::DHTServer) normally constructs this component from /// [`MetadataOptions`](crate::MetadataOptions). pub fn new(timeout_secs: u64) -> Self { - Self::new_with_runtime_stats(timeout_secs, 32, 200_000, 60, DhtRuntimeStats::default()) + Self::new_with_runtime_stats( + timeout_secs, + 32, + 10 * 1024 * 1024, + 200_000, + 60, + DhtRuntimeStats::default(), + ) } pub(crate) fn new_with_runtime_stats( timeout_secs: u64, max_connects_per_second: u32, + max_metadata_size_bytes: usize, peer_failure_cache_capacity: usize, peer_failure_ttl_secs: u64, runtime_stats: DhtRuntimeStats, ) -> Self { Self { total_timeout: Duration::from_secs(if timeout_secs == 0 { 15 } else { timeout_secs }), + max_metadata_size_bytes: max_metadata_size_bytes.max(1), runtime_stats, peer_failure_cache: Arc::new(PeerFailureCache::new( peer_failure_cache_capacity, @@ -457,7 +467,7 @@ impl RbitFetcher { } if metadata_size > 0 && remote_ut_metadata_id > 0 && !request_sent { - if metadata_size > 10 * 1024 * 1024 { + if metadata_size as usize > self.max_metadata_size_bytes { #[cfg(feature = "metrics")] counter!("dht_metadata_fetch_fail_total", "reason" => "size_limit") .increment(1); @@ -578,9 +588,10 @@ fn parse_metadata(info_bytes: &[u8]) -> Option { let piece_length = dict .get(&b"piece length"[..]) .and_then(|value| value.as_integer()) - .unwrap_or(0) as u64; + .and_then(|value| u64::try_from(value).ok()) + .unwrap_or(0); - let mut total_size = 0; + let mut total_size = 0_u64; let mut file_list = Vec::new(); if let Some(files) = dict.get(&b"files"[..]).and_then(|value| value.as_list()) { for file in files { @@ -593,26 +604,23 @@ fn parse_metadata(info_bytes: &[u8]) -> Option { else { continue; }; - let length = length as u64; - total_size += length; - let path = file_dict + let length = u64::try_from(length).ok()?; + total_size = total_size.checked_add(length)?; + let parts = file_dict .get(&b"path"[..]) - .and_then(|value| value.as_list()) - .map(|parts| { - parts - .iter() - .filter_map(|part| part.as_str()) - .collect::>() - .join("/") - }) - .unwrap_or_default(); + .and_then(|value| value.as_list())?; + let path = parts + .iter() + .map(|part| part.as_str()) + .collect::>>()? + .join("/"); file_list.push(FileInfo { path, size: length }); } } else if let Some(length) = dict .get(&b"length"[..]) .and_then(|value| value.as_integer()) { - total_size = length as u64; + total_size = u64::try_from(length).ok()?; file_list.push(FileInfo { path: name.clone(), size: total_size, @@ -625,6 +633,11 @@ fn parse_metadata(info_bytes: &[u8]) -> Option { #[cfg(test)] mod tests { use super::*; + + #[test] + fn parser_rejects_negative_file_size() { + assert!(parse_metadata(b"d6:lengthi-1e4:name1:ae").is_none()); + } use tokio::io::{AsyncReadExt, AsyncWriteExt}; use tokio::net::TcpListener; @@ -677,7 +690,8 @@ mod tests { }); let stats = DhtRuntimeStats::default(); - let fetcher = RbitFetcher::new_with_runtime_stats(1, 10, 10, 60, stats.clone()); + let fetcher = + RbitFetcher::new_with_runtime_stats(1, 10, 10 * 1024 * 1024, 10, 60, stats.clone()); let started = Instant::now(); assert!(matches!( fetcher.fetch(&[7; 20], addr).await, @@ -712,8 +726,14 @@ mod tests { stream.read_exact(&mut handshake).await.unwrap(); stream.write_all(&handshake).await.unwrap(); }); - let fetcher = - RbitFetcher::new_with_runtime_stats(1, 10, 10, 60, DhtRuntimeStats::default()); + let fetcher = RbitFetcher::new_with_runtime_stats( + 1, + 10, + 10 * 1024 * 1024, + 10, + 60, + DhtRuntimeStats::default(), + ); assert!(fetcher.verify_handshake(&[7; 20], addr).await); peer.await.unwrap(); } diff --git a/dht-crawler/src/server.rs b/dht-crawler/src/server.rs index 4ed22ac..5e10d70 100644 --- a/dht-crawler/src/server.rs +++ b/dht-crawler/src/server.rs @@ -358,6 +358,7 @@ impl DHTServer { let fetcher = Arc::new(RbitFetcher::new_with_runtime_stats( options.metadata.timeout_secs, options.metadata.max_connects_per_second, + options.metadata.max_metadata_size_bytes, options.metadata.peer_failure_cache_capacity, options.metadata.peer_failure_ttl_secs, runtime_stats.clone(), diff --git a/dht-crawler/src/types.rs b/dht-crawler/src/types.rs index 004b07c..b3106c8 100644 --- a/dht-crawler/src/types.rs +++ b/dht-crawler/src/types.rs @@ -142,6 +142,8 @@ pub struct MetadataOptions { pub max_worker_count: usize, /// Maximum real TCP connection attempts started per second. pub max_connects_per_second: u32, + /// Maximum accepted BEP-9 info dictionary size in bytes. + pub max_metadata_size_bytes: usize, /// Maximum number of cached bad Peer socket addresses. pub peer_failure_cache_capacity: usize, /// Timeout/connect failure cache lifetime in seconds. @@ -308,6 +310,7 @@ impl Default for MetadataOptions { max_queue_size: 10_000, max_worker_count: 8, max_connects_per_second: 2, + max_metadata_size_bytes: 10 * 1024 * 1024, peer_failure_cache_capacity: 200_000, peer_failure_ttl_secs: 60, } diff --git a/dht-search.example.toml b/dht-search.example.toml index 6b0ad85..1bac639 100644 --- a/dht-search.example.toml +++ b/dht-search.example.toml @@ -6,6 +6,13 @@ stats_interval_secs = 10 index_batch_size = 1024 index_interval_millis = 5000 +[metadata_limits] +max_metadata_bytes = 10485760 +max_files = 20000 +max_name_bytes = 1024 +max_path_bytes = 4096 +max_path_depth = 64 + [dht] port = 12313 netmode = "ipv4-only" diff --git a/dht-search.filter-test.toml b/dht-search.filter-test.toml new file mode 100644 index 0000000..5f13330 --- /dev/null +++ b/dht-search.filter-test.toml @@ -0,0 +1,13 @@ +data_dir = "data-filter-test" +run_duration_secs = 300 + +[metadata_limits] +max_metadata_bytes = 65536 +max_files = 10 +max_name_bytes = 64 +max_path_bytes = 128 +max_path_depth = 4 + +[http] +listen = "127.0.0.1:8080" +web_dir = "web/dist" diff --git a/dht-search/README.md b/dht-search/README.md index 260c4aa..ae20c84 100644 --- a/dht-search/README.md +++ b/dht-search/README.md @@ -80,6 +80,33 @@ Metadata 下载和可用性握手共用 `metadata_connects_per_second` 预算不 Windows 下索引每五秒批量提交 临时文件占用会自动指数退避重试且不会停止采集 HTTP 服务或丢失 RocksDB 待索引状态 +### Metadata 安全限制 + +应用会在 Metadata 下载和进入 RocksDB 前执行两层资源与结构校验 + +| 配置项 | 默认值 | 作用 | +|---|---:|---| +| `metadata_limits.max_metadata_bytes` | `10485760` | 下载阶段允许的最大 info 字典字节数 | +| `metadata_limits.max_files` | `20000` | 单个种子允许的最大文件数量 | +| `metadata_limits.max_name_bytes` | `1024` | 种子名称最大 UTF-8 字节数 | +| `metadata_limits.max_path_bytes` | `4096` | 单个文件路径最大 UTF-8 字节数 | +| `metadata_limits.max_path_depth` | `64` | 单个文件路径最大目录层级 | + +空名称 空文件列表 控制字符 空路径段 `.` `..` 大小溢出和声明总大小不一致会被分类拒绝 + +通过完整 Metadata 校验后被拒绝的 infohash 只在 RocksDB 保存原因规则指纹时间和次数 不保存名称或文件列表 相同规则下再次发现时不会重复下载 修改限制后规则指纹变化并允许重新判断 + +`/stats` 返回 `metadata_filtered` 总数以及 `metadata_filtered_*` 分类计数 Web 运行状态展示本次运行的过滤总数 + +可以使用独立数据目录和严格限制运行五分钟测试 不会污染正式数据目录 + +```powershell +cargo run -p dht-search -- --config dht-search.filter-test.toml +Invoke-RestMethod http://127.0.0.1:8080/stats | ConvertTo-Json -Depth 5 +``` + +严格测试配置仅用于观察过滤效果 不应作为正式采集配置 + ### 采样去重和 Peer 查找 BEP-51 返回的 infohash 会先进入有界批量准入队列并由 RocksDB 精确判断 diff --git a/dht-search/src/api/handlers.rs b/dht-search/src/api/handlers.rs index 79062f8..75b6730 100644 --- a/dht-search/src/api/handlers.rs +++ b/dht-search/src/api/handlers.rs @@ -9,7 +9,7 @@ use axum::{ response::{IntoResponse, Response}, }; use dht_search::{ - domain::InfoHash, + domain::{InfoHash, MetadataRejectionReason}, search::{SearchOptions, SearchPage, SearchSort}, storage::VerificationPriority, }; @@ -35,6 +35,7 @@ pub(crate) async fn stats(State(state): State) -> Json let dht = state.dht_stats.snapshot(); let observability = state.dht_stats.observability_snapshot(); let persistence = state.persistence.snapshot(); + let filtered = persistence.filtered; let verification = state .verification .as_ref() @@ -56,6 +57,24 @@ pub(crate) async fn stats(State(state): State) -> Json metadata_in_flight: dht.metadata_in_flight, metadata_ok: dht.metadata_peer_succeeded, metadata_failed: dht.metadata_peer_failed, + metadata_filtered: observability + .metadata_failure_size_limit + .saturating_add(filtered.total()), + metadata_filtered_too_large: observability.metadata_failure_size_limit, + metadata_filtered_invalid_info_hash: filtered + .count(MetadataRejectionReason::InvalidInfoHash), + metadata_filtered_empty_name: filtered.count(MetadataRejectionReason::EmptyName), + metadata_filtered_name_too_long: filtered.count(MetadataRejectionReason::NameTooLong), + metadata_filtered_invalid_name: filtered.count(MetadataRejectionReason::InvalidName), + metadata_filtered_empty_file_list: filtered.count(MetadataRejectionReason::EmptyFileList), + metadata_filtered_too_many_files: filtered.count(MetadataRejectionReason::TooManyFiles), + metadata_filtered_empty_path: filtered.count(MetadataRejectionReason::EmptyPath), + metadata_filtered_path_too_long: filtered.count(MetadataRejectionReason::PathTooLong), + metadata_filtered_path_too_deep: filtered.count(MetadataRejectionReason::PathTooDeep), + metadata_filtered_invalid_path: filtered.count(MetadataRejectionReason::InvalidPath), + metadata_filtered_size_overflow: filtered.count(MetadataRejectionReason::SizeOverflow), + metadata_filtered_total_size_mismatch: filtered + .count(MetadataRejectionReason::TotalSizeMismatch), persistence_accepted: persistence.accepted, persistence_inserted: persistence.inserted, persistence_updated: persistence.updated, diff --git a/dht-search/src/api/mod.rs b/dht-search/src/api/mod.rs index 49f0d8e..4519d85 100644 --- a/dht-search/src/api/mod.rs +++ b/dht-search/src/api/mod.rs @@ -60,7 +60,7 @@ mod tests { }; use dht_crawler::{DhtRuntimeStats, FileInfo, TorrentInfo}; use dht_search::{ - domain::{InfoHash, TorrentRecord}, + domain::{InfoHash, MetadataLimits, TorrentRecord}, search::SearchEngine, storage::{RocksTorrentRepository, TorrentRepository}, }; @@ -105,7 +105,8 @@ mod tests { let search = SearchEngine::open(directory.path().join("tantivy")).unwrap(); search.index_pending(repository.as_ref(), 10, 20).unwrap(); let repository_trait: Arc = repository.clone(); - let persistence = PersistencePipeline::start(repository_trait.clone(), 4); + let persistence = + PersistencePipeline::start(repository_trait.clone(), 4, MetadataLimits::default()); let verification = VerificationIngress::for_test(repository.clone(), 10); let web_dir = directory.path().join("web"); std::fs::create_dir_all(&web_dir).unwrap(); @@ -130,6 +131,23 @@ mod tests { let body = to_bytes(response.into_body(), usize::MAX).await.unwrap(); assert_eq!(body.as_ref(), b"
DHT Search
"); + let response = app + .clone() + .oneshot( + Request::builder() + .uri("/stats") + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::OK); + let json: serde_json::Value = + serde_json::from_slice(&to_bytes(response.into_body(), usize::MAX).await.unwrap()) + .unwrap(); + assert_eq!(json["metadata_filtered"], 0); + assert_eq!(json["metadata_filtered_too_many_files"], 0); + let response = app .clone() .oneshot( diff --git a/dht-search/src/api/response.rs b/dht-search/src/api/response.rs index 2fee041..5aceee4 100644 --- a/dht-search/src/api/response.rs +++ b/dht-search/src/api/response.rs @@ -29,6 +29,20 @@ pub(crate) struct StatsResponse { pub(crate) metadata_in_flight: usize, pub(crate) metadata_ok: u64, pub(crate) metadata_failed: u64, + pub(crate) metadata_filtered: u64, + pub(crate) metadata_filtered_too_large: u64, + pub(crate) metadata_filtered_invalid_info_hash: u64, + pub(crate) metadata_filtered_empty_name: u64, + pub(crate) metadata_filtered_name_too_long: u64, + pub(crate) metadata_filtered_invalid_name: u64, + pub(crate) metadata_filtered_empty_file_list: u64, + pub(crate) metadata_filtered_too_many_files: u64, + pub(crate) metadata_filtered_empty_path: u64, + pub(crate) metadata_filtered_path_too_long: u64, + pub(crate) metadata_filtered_path_too_deep: u64, + pub(crate) metadata_filtered_invalid_path: u64, + pub(crate) metadata_filtered_size_overflow: u64, + pub(crate) metadata_filtered_total_size_mismatch: u64, pub(crate) persistence_accepted: u64, pub(crate) persistence_inserted: u64, pub(crate) persistence_updated: u64, diff --git a/dht-search/src/app.rs b/dht-search/src/app.rs index 75ae61b..6c948cf 100644 --- a/dht-search/src/app.rs +++ b/dht-search/src/app.rs @@ -8,7 +8,7 @@ use std::{ use dht_crawler::DHTServer; use dht_search::{ - domain::InfoHash, + domain::{InfoHash, MetadataRejectionReason}, search::SearchEngine, storage::{RocksTorrentRepository, TorrentRepository}, }; @@ -25,15 +25,22 @@ use crate::{ pub(crate) async fn run(config: AppConfig) -> Result<(), AppError> { std::fs::create_dir_all(&config.data_dir)?; let database_path = config.data_dir.join("rocksdb"); - let repository = Arc::new(RocksTorrentRepository::open(&database_path)?); + let metadata_limits = config.metadata_limits(); + let repository = Arc::new(RocksTorrentRepository::open_with_rejection_rule( + &database_path, + metadata_limits.rule_id(), + )?); let (search, search_created) = SearchEngine::open_with_status(config.data_dir.join("tantivy"))?; if search_created { let records = repository.prepare_full_reindex()?; tracing::info!(records, "检测到新搜索索引并准备全量重建"); } let repository_api: Arc = repository.clone(); - let mut persistence = - PersistencePipeline::start(repository_api, config.persistence_queue_capacity); + let mut persistence = PersistencePipeline::start( + repository_api, + config.persistence_queue_capacity, + metadata_limits, + ); let ingress = persistence.ingress.clone(); let options = config.dht_options(); @@ -377,6 +384,15 @@ async fn monitor( metadata_in_flight = dht.metadata_in_flight, metadata_ok = dht.metadata_peer_succeeded, metadata_failed = dht.metadata_peer_failed, + metadata_filtered = observability + .metadata_failure_size_limit + .saturating_add(storage.filtered.total()), + metadata_filtered_too_many_files = storage + .filtered + .count(MetadataRejectionReason::TooManyFiles), + metadata_filtered_invalid_path = storage + .filtered + .count(MetadataRejectionReason::InvalidPath), persistence_accepted = storage.accepted, persistence_inserted = storage.inserted, persistence_updated = storage.updated, diff --git a/dht-search/src/config.rs b/dht-search/src/config.rs index c1caa9f..415dd23 100644 --- a/dht-search/src/config.rs +++ b/dht-search/src/config.rs @@ -10,6 +10,7 @@ use dht_crawler::{ use serde::Deserialize; use crate::error::AppError; +use dht_search::domain::MetadataLimits; #[derive(Debug, Parser)] #[command(name = "dht-search", version, about = "DHT 元数据采集和搜索服务")] @@ -31,6 +32,7 @@ pub(crate) struct AppConfig { pub(crate) run_duration_secs: Option, pub(crate) index_batch_size: usize, pub(crate) index_interval_millis: u64, + pub(crate) metadata_limits: MetadataLimitsConfig, pub(crate) dht: DhtConfig, pub(crate) http: HttpConfig, pub(crate) verification: VerificationConfig, @@ -74,6 +76,16 @@ pub(crate) struct VerificationConfig { pub(crate) poll_interval_millis: u64, } +#[derive(Debug, Clone, Deserialize)] +#[serde(default, deny_unknown_fields)] +pub(crate) struct MetadataLimitsConfig { + pub(crate) max_metadata_bytes: usize, + pub(crate) max_files: usize, + pub(crate) max_name_bytes: usize, + pub(crate) max_path_bytes: usize, + pub(crate) max_path_depth: usize, +} + #[derive(Debug, Clone, Copy, Default, Deserialize)] #[serde(rename_all = "kebab-case")] pub(crate) enum NetworkMode { @@ -129,6 +141,7 @@ impl AppConfig { max_queue_size: self.dht.metadata_queue_capacity, max_worker_count: self.dht.metadata_workers, max_connects_per_second: self.dht.metadata_connects_per_second, + max_metadata_size_bytes: self.metadata_limits.max_metadata_bytes, ..defaults.metadata }, peer_lookup: PeerLookupOptions { @@ -164,6 +177,15 @@ impl AppConfig { } } + pub(crate) fn metadata_limits(&self) -> MetadataLimits { + MetadataLimits { + max_files: self.metadata_limits.max_files, + max_name_bytes: self.metadata_limits.max_name_bytes, + max_path_bytes: self.metadata_limits.max_path_bytes, + max_path_depth: self.metadata_limits.max_path_depth, + } + } + fn validate(&self) -> Result<(), AppError> { if self.persistence_queue_capacity == 0 { return Err(AppError::Config( @@ -185,6 +207,16 @@ impl AppConfig { "索引批量大小和执行间隔必须大于零".to_owned(), )); } + if self.metadata_limits.max_metadata_bytes == 0 + || self.metadata_limits.max_files == 0 + || self.metadata_limits.max_name_bytes == 0 + || self.metadata_limits.max_path_bytes == 0 + || self.metadata_limits.max_path_depth == 0 + { + return Err(AppError::Config( + "Metadata 大小文件数名称路径和目录层级上限必须大于零".to_owned(), + )); + } if self.dht.metadata_workers == 0 || self.dht.metadata_queue_capacity == 0 { return Err(AppError::Config( "Metadata worker 和队列容量必须大于零".to_owned(), @@ -221,6 +253,7 @@ impl Default for AppConfig { run_duration_secs: None, index_batch_size: 512, index_interval_millis: 5_000, + metadata_limits: MetadataLimitsConfig::default(), dht: DhtConfig::default(), http: HttpConfig::default(), verification: VerificationConfig::default(), @@ -228,6 +261,18 @@ impl Default for AppConfig { } } +impl Default for MetadataLimitsConfig { + fn default() -> Self { + Self { + max_metadata_bytes: 10 * 1024 * 1024, + max_files: 20_000, + max_name_bytes: 1_024, + max_path_bytes: 4_096, + max_path_depth: 64, + } + } +} + impl Default for DhtConfig { fn default() -> Self { Self { @@ -311,6 +356,7 @@ mod tests { assert_eq!(options.port, 12_313); assert_eq!(options.metadata.max_worker_count, 8); assert_eq!(options.metadata.max_connects_per_second, 2); + assert_eq!(options.metadata.max_metadata_size_bytes, 10 * 1024 * 1024); assert_eq!(options.max_outbound_queries_per_second, 10); assert_eq!(options.crawl.rate_limit.max_find_node_rate_per_sec, 6); } @@ -345,4 +391,11 @@ mod tests { .unwrap_err(); assert!(matches!(error, AppError::Toml(_))); } + + #[test] + fn zero_metadata_limit_is_rejected() { + let mut config = AppConfig::default(); + config.metadata_limits.max_files = 0; + assert!(matches!(config.validate(), Err(AppError::Config(_)))); + } } diff --git a/dht-search/src/crawler/pipeline.rs b/dht-search/src/crawler/pipeline.rs index 421a4c1..4a6afc0 100644 --- a/dht-search/src/crawler/pipeline.rs +++ b/dht-search/src/crawler/pipeline.rs @@ -1,6 +1,7 @@ // 负责定义采集阶段之间的有界队列背压和任务流转规则 use std::{ + str::FromStr, sync::{ Arc, Mutex, atomic::{AtomicU64, AtomicUsize, Ordering}, @@ -11,7 +12,7 @@ use std::{ use dht_crawler::TorrentInfo; use dht_search::{ - domain::TorrentRecord, + domain::{InfoHash, MetadataLimits, MetadataRejectionReason, RejectedMetadata, TorrentRecord}, storage::{StorageError, TorrentRepository, UpsertOutcome}, }; use tokio::sync::oneshot; @@ -20,8 +21,9 @@ use crate::error::AppError; #[derive(Clone)] pub(crate) struct PersistenceIngress { - sender: Arc>>>, + sender: Arc>>>, stats: Arc, + limits: MetadataLimits, } pub(crate) struct PersistencePipeline { @@ -39,6 +41,7 @@ pub(crate) struct PersistenceStats { invalid: AtomicU64, failed: AtomicU64, queue_depth: AtomicUsize, + filtered: [AtomicU64; MetadataRejectionReason::COUNT], } #[derive(Debug, Clone, Copy)] @@ -50,26 +53,48 @@ pub(crate) struct PersistenceSnapshot { pub(crate) invalid: u64, pub(crate) failed: u64, pub(crate) queue_depth: usize, + pub(crate) filtered: MetadataFilterSnapshot, +} + +#[derive(Debug, Clone, Copy)] +pub(crate) struct MetadataFilterSnapshot { + counts: [u64; MetadataRejectionReason::COUNT], +} + +enum PersistenceItem { + Record(TorrentRecord), + Rejection(RejectedMetadata), } impl PersistencePipeline { - pub(crate) fn start(repository: Arc, capacity: usize) -> Self { - let (sender, receiver) = mpsc::sync_channel::(capacity); + pub(crate) fn start( + repository: Arc, + capacity: usize, + limits: MetadataLimits, + ) -> Self { + let (sender, receiver) = mpsc::sync_channel::(capacity); let (fatal_tx, fatal) = oneshot::channel(); let stats = Arc::new(PersistenceStats::default()); let worker_stats = stats.clone(); let worker = thread::Builder::new() .name("torrent-persistence".to_owned()) .spawn(move || { - while let Ok(record) = receiver.recv() { + while let Ok(item) = receiver.recv() { worker_stats.queue_depth.fetch_sub(1, Ordering::Relaxed); - match repository.upsert(record) { - Ok(UpsertOutcome::Inserted) => { + let outcome = match item { + PersistenceItem::Record(record) => repository.upsert(record).map(Some), + PersistenceItem::Rejection(rejection) => { + repository.record_rejection(rejection).map(|()| None) + } + }; + match outcome { + Ok(Some(UpsertOutcome::Inserted)) => { worker_stats.inserted.fetch_add(1, Ordering::Relaxed); } - Ok(UpsertOutcome::Updated { .. }) => { + Ok(Some(UpsertOutcome::Updated { .. })) => { worker_stats.updated.fetch_add(1, Ordering::Relaxed); } + Ok(None) => {} Err(error) => { worker_stats.failed.fetch_add(1, Ordering::Relaxed); let _ = fatal_tx.send(error.to_string()); @@ -85,6 +110,7 @@ impl PersistencePipeline { ingress: PersistenceIngress { sender: Arc::new(Mutex::new(Some(sender))), stats, + limits, }, fatal, worker, @@ -103,14 +129,31 @@ impl PersistencePipeline { impl PersistenceIngress { pub(crate) fn try_enqueue(&self, torrent: TorrentInfo) -> bool { - let record = match TorrentRecord::try_from(torrent) { + let rejected_info_hash = InfoHash::from_str(&torrent.info_hash).ok(); + let rejected_at = torrent.timestamp; + let record = match TorrentRecord::try_from_with_limits(torrent, self.limits) { Ok(record) => record, Err(error) => { self.stats.invalid.fetch_add(1, Ordering::Relaxed); - tracing::warn!(%error, "拒绝无效 Metadata"); + let reason = error.rejection_reason(); + self.stats.filtered[reason.index()].fetch_add(1, Ordering::Relaxed); + tracing::warn!(%error, ?reason, "拒绝无效 Metadata"); + if let Some(info_hash) = rejected_info_hash { + let rejection = RejectedMetadata::new( + info_hash, + reason, + self.limits.rule_id(), + rejected_at, + ); + self.try_send(PersistenceItem::Rejection(rejection), false); + } return false; } }; + self.try_send(PersistenceItem::Record(record), true) + } + + fn try_send(&self, item: PersistenceItem, accepted_record: bool) -> bool { let sender = self .sender .lock() @@ -118,9 +161,11 @@ impl PersistenceIngress { let Some(sender) = sender.as_ref() else { return false; }; - match sender.try_send(record) { + match sender.try_send(item) { Ok(()) => { - self.stats.accepted.fetch_add(1, Ordering::Relaxed); + if accepted_record { + self.stats.accepted.fetch_add(1, Ordering::Relaxed); + } self.stats.queue_depth.fetch_add(1, Ordering::Relaxed); true } @@ -154,10 +199,23 @@ impl PersistenceStats { invalid: self.invalid.load(Ordering::Relaxed), failed: self.failed.load(Ordering::Relaxed), queue_depth: self.queue_depth.load(Ordering::Relaxed), + filtered: MetadataFilterSnapshot { + counts: std::array::from_fn(|index| self.filtered[index].load(Ordering::Relaxed)), + }, } } } +impl MetadataFilterSnapshot { + pub(crate) fn total(self) -> u64 { + self.counts.iter().copied().sum() + } + + pub(crate) fn count(self, reason: MetadataRejectionReason) -> u64 { + self.counts[reason.index()] + } +} + #[cfg(test)] mod tests { use std::sync::Mutex; @@ -176,6 +234,7 @@ mod tests { #[derive(Default)] struct MemoryRepository { records: Mutex>, + rejections: Mutex>, } impl TorrentRepository for MemoryRepository { @@ -194,6 +253,15 @@ mod tests { Ok(UpsertOutcome::Inserted) } + fn rejection(&self, _: InfoHash) -> Result, StorageError> { + Ok(None) + } + + fn record_rejection(&self, rejection: RejectedMetadata) -> Result<(), StorageError> { + self.rejections.lock().unwrap().push(rejection); + Ok(()) + } + fn observe_existing(&self, _: InfoHash, _: u64) -> Result { Ok(false) } @@ -280,10 +348,43 @@ mod tests { #[tokio::test] async fn accepted_record_is_drained_before_shutdown() { let repository = Arc::new(MemoryRepository::default()); - let pipeline = PersistencePipeline::start(repository.clone(), 1); + let pipeline = PersistencePipeline::start(repository.clone(), 1, MetadataLimits::default()); assert!(pipeline.ingress.try_enqueue(torrent())); pipeline.close_and_join().await.unwrap(); let records = repository.records.lock().unwrap(); assert_eq!(records.len(), 1); } + + #[tokio::test] + async fn invalid_record_is_classified_and_persists_only_a_rejection() { + let repository = Arc::new(MemoryRepository::default()); + let limits = MetadataLimits { + max_files: 1, + ..MetadataLimits::default() + }; + let pipeline = PersistencePipeline::start(repository.clone(), 1, limits); + let mut invalid = torrent(); + invalid.files.push(FileInfo { + path: "second".into(), + size: 1, + }); + invalid.total_size = 2; + + assert!(!pipeline.ingress.try_enqueue(invalid)); + let snapshot = pipeline.ingress.snapshot(); + assert_eq!(snapshot.accepted, 0); + assert_eq!(snapshot.filtered.total(), 1); + assert_eq!( + snapshot + .filtered + .count(MetadataRejectionReason::TooManyFiles), + 1 + ); + pipeline.close_and_join().await.unwrap(); + + assert!(repository.records.lock().unwrap().is_empty()); + let rejections = repository.rejections.lock().unwrap(); + assert_eq!(rejections.len(), 1); + assert_eq!(rejections[0].reason, MetadataRejectionReason::TooManyFiles); + } } diff --git a/dht-search/src/domain/mod.rs b/dht-search/src/domain/mod.rs index 0ec45f3..279573a 100644 --- a/dht-search/src/domain/mod.rs +++ b/dht-search/src/domain/mod.rs @@ -8,6 +8,7 @@ pub(crate) use torrent::ContentGroupBuilder; #[cfg(test)] pub(crate) use torrent::test_record; pub use torrent::{ - Availability, AvailabilityStatus, ContentGroup, Heat, HeatLevel, InfoHash, TorrentFile, - TorrentRecord, TorrentRecordError, VerificationResult, + Availability, AvailabilityStatus, ContentGroup, Heat, HeatLevel, InfoHash, MetadataLimits, + MetadataRejectionReason, RejectedMetadata, TorrentFile, TorrentRecord, TorrentRecordError, + VerificationResult, }; diff --git a/dht-search/src/domain/torrent.rs b/dht-search/src/domain/torrent.rs index 8b90df0..de80071 100644 --- a/dht-search/src/domain/torrent.rs +++ b/dht-search/src/domain/torrent.rs @@ -9,6 +9,110 @@ use super::fingerprint::content_key; const MAX_STORED_PEERS: usize = 32; const ACTIVITY_SCALE: u64 = 1_000; const ACTIVITY_HALF_LIFE_SECS: f64 = 86_400.0; +const METADATA_VALIDATION_VERSION: u64 = 1; + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct MetadataLimits { + pub max_files: usize, + pub max_name_bytes: usize, + pub max_path_bytes: usize, + pub max_path_depth: usize, +} + +impl MetadataLimits { + pub fn rule_id(self) -> [u8; 32] { + let mut hasher = blake3::Hasher::new(); + hasher.update(b"dht-search-metadata-limits\0"); + hasher.update(&METADATA_VALIDATION_VERSION.to_be_bytes()); + hasher.update(&(self.max_files as u64).to_be_bytes()); + hasher.update(&(self.max_name_bytes as u64).to_be_bytes()); + hasher.update(&(self.max_path_bytes as u64).to_be_bytes()); + hasher.update(&(self.max_path_depth as u64).to_be_bytes()); + *hasher.finalize().as_bytes() + } +} + +impl Default for MetadataLimits { + fn default() -> Self { + Self { + max_files: 20_000, + max_name_bytes: 1_024, + max_path_bytes: 4_096, + max_path_depth: 64, + } + } +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum MetadataRejectionReason { + InvalidInfoHash, + EmptyName, + NameTooLong, + InvalidName, + EmptyFileList, + TooManyFiles, + EmptyPath, + PathTooLong, + PathTooDeep, + InvalidPath, + SizeOverflow, + TotalSizeMismatch, +} + +impl MetadataRejectionReason { + pub const COUNT: usize = 12; + + pub const fn index(self) -> usize { + match self { + Self::InvalidInfoHash => 0, + Self::EmptyName => 1, + Self::NameTooLong => 2, + Self::InvalidName => 3, + Self::EmptyFileList => 4, + Self::TooManyFiles => 5, + Self::EmptyPath => 6, + Self::PathTooLong => 7, + Self::PathTooDeep => 8, + Self::InvalidPath => 9, + Self::SizeOverflow => 10, + Self::TotalSizeMismatch => 11, + } + } +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct RejectedMetadata { + pub info_hash: InfoHash, + pub reason: MetadataRejectionReason, + pub rule_id: [u8; 32], + pub first_rejected_at: u64, + pub last_seen: u64, + pub seen_count: u64, +} + +impl RejectedMetadata { + pub fn new( + info_hash: InfoHash, + reason: MetadataRejectionReason, + rule_id: [u8; 32], + timestamp: u64, + ) -> Self { + Self { + info_hash, + reason, + rule_id, + first_rejected_at: timestamp, + last_seen: timestamp, + seen_count: 1, + } + } + + pub fn observe_again(&mut self, timestamp: u64) { + self.last_seen = self.last_seen.max(timestamp); + self.seen_count = self.seen_count.saturating_add(1); + } +} #[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)] pub struct InfoHash([u8; 20]); @@ -278,6 +382,91 @@ pub struct TorrentRecord { } impl TorrentRecord { + pub fn try_from_with_limits( + info: TorrentInfo, + limits: MetadataLimits, + ) -> Result { + let info_hash = InfoHash::from_str(&info.info_hash)?; + if info.name.trim().is_empty() { + return Err(TorrentRecordError::EmptyName); + } + if info.name.len() > limits.max_name_bytes { + return Err(TorrentRecordError::NameTooLong { + actual: info.name.len(), + limit: limits.max_name_bytes, + }); + } + if info.name.chars().any(char::is_control) { + return Err(TorrentRecordError::InvalidName); + } + if info.files.is_empty() { + return Err(TorrentRecordError::EmptyFileList); + } + if info.files.len() > limits.max_files { + return Err(TorrentRecordError::TooManyFiles { + actual: info.files.len(), + limit: limits.max_files, + }); + } + for file in &info.files { + validate_path(&file.path, limits)?; + } + let files: Vec<_> = info + .files + .into_iter() + .map(|file| TorrentFile { + path: file.path, + size: file.size, + }) + .collect(); + let calculated_size = files.iter().try_fold(0_u64, |total, file| { + total + .checked_add(file.size) + .ok_or(TorrentRecordError::SizeOverflow) + })?; + if calculated_size != info.total_size { + return Err(TorrentRecordError::TotalSizeMismatch { + declared: info.total_size, + calculated: calculated_size, + }); + } + let content_key = content_key(&files)?; + let mut source_peers = info.peers; + source_peers.sort_unstable(); + source_peers.dedup(); + source_peers.truncate(MAX_STORED_PEERS); + let reachable_peers = source_peers.len().min(u32::MAX as usize) as u32; + let availability = if reachable_peers > 0 { + Availability { + status: AvailabilityStatus::Active, + last_verified_at: Some(info.timestamp), + last_success_at: Some(info.timestamp), + discovered_peers: reachable_peers, + reachable_peers, + consecutive_failures: 0, + next_check_at: info.timestamp.saturating_add(86_400), + } + } else { + Availability::default() + }; + + Ok(Self { + info_hash, + name: info.name, + total_size: info.total_size, + files, + piece_length: info.piece_length, + source_peers, + content_key, + first_seen: info.timestamp, + last_seen: info.timestamp, + seen_count: 1, + availability, + activity_score_millis: ACTIVITY_SCALE, + activity_updated_at: info.timestamp, + }) + } + pub fn observe_again(&mut self, timestamp: u64, peers: &[String]) { self.last_seen = self.last_seen.max(timestamp); self.seen_count = self.seen_count.saturating_add(1); @@ -369,78 +558,59 @@ impl TryFrom for TorrentRecord { type Error = TorrentRecordError; fn try_from(info: TorrentInfo) -> Result { - let info_hash = InfoHash::from_str(&info.info_hash)?; - let files: Vec<_> = info - .files - .into_iter() - .map(|file| TorrentFile { - path: file.path, - size: file.size, - }) - .collect(); - if info.name.trim().is_empty() { - return Err(TorrentRecordError::EmptyName); - } - if files.is_empty() { - return Err(TorrentRecordError::EmptyFileList); - } - let calculated_size = files.iter().try_fold(0_u64, |total, file| { - total - .checked_add(file.size) - .ok_or(TorrentRecordError::SizeOverflow) - })?; - if calculated_size != info.total_size { - return Err(TorrentRecordError::TotalSizeMismatch { - declared: info.total_size, - calculated: calculated_size, - }); - } - let content_key = content_key(&files)?; - let mut source_peers = info.peers; - source_peers.sort_unstable(); - source_peers.dedup(); - source_peers.truncate(MAX_STORED_PEERS); - let reachable_peers = source_peers.len().min(u32::MAX as usize) as u32; - let availability = if reachable_peers > 0 { - Availability { - status: AvailabilityStatus::Active, - last_verified_at: Some(info.timestamp), - last_success_at: Some(info.timestamp), - discovered_peers: reachable_peers, - reachable_peers, - consecutive_failures: 0, - next_check_at: info.timestamp.saturating_add(86_400), - } - } else { - Availability::default() - }; - - Ok(Self { - info_hash, - name: info.name, - total_size: info.total_size, - files, - piece_length: info.piece_length, - source_peers, - content_key, - first_seen: info.timestamp, - last_seen: info.timestamp, - seen_count: 1, - availability, - activity_score_millis: ACTIVITY_SCALE, - activity_updated_at: info.timestamp, - }) + Self::try_from_with_limits(info, MetadataLimits::default()) } } +fn validate_path(path: &str, limits: MetadataLimits) -> Result<(), TorrentRecordError> { + if path.is_empty() { + return Err(TorrentRecordError::EmptyNormalizedPath); + } + if path.len() > limits.max_path_bytes { + return Err(TorrentRecordError::PathTooLong { + actual: path.len(), + limit: limits.max_path_bytes, + }); + } + if path.chars().any(char::is_control) { + return Err(TorrentRecordError::InvalidPath); + } + let mut depth = 0_usize; + for component in path.split(['/', '\\']) { + if component.is_empty() || matches!(component, "." | "..") { + return Err(TorrentRecordError::InvalidPath); + } + depth += 1; + } + if depth > limits.max_path_depth { + return Err(TorrentRecordError::PathTooDeep { + actual: depth, + limit: limits.max_path_depth, + }); + } + Ok(()) +} + #[derive(Debug, thiserror::Error, PartialEq, Eq)] pub enum TorrentRecordError { #[error("infohash 必须是二十字节的十六进制字符串")] InvalidInfoHash, #[error("种子名称不能为空")] EmptyName, + #[error("种子名称长度 {actual} 字节超过上限 {limit}")] + NameTooLong { actual: usize, limit: usize }, + #[error("种子名称包含控制字符")] + InvalidName, #[error("文件列表不能为空")] EmptyFileList, + #[error("文件数量 {actual} 超过上限 {limit}")] + TooManyFiles { actual: usize, limit: usize }, + #[error("文件路径长度 {actual} 字节超过上限 {limit}")] + PathTooLong { actual: usize, limit: usize }, + #[error("文件路径目录层级 {actual} 超过上限 {limit}")] + PathTooDeep { actual: usize, limit: usize }, + #[error("文件路径包含空段上级目录当前目录或控制字符")] + InvalidPath, #[error("文件总大小溢出")] SizeOverflow, #[error("声明大小 {declared} 与文件计算大小 {calculated} 不一致")] @@ -449,6 +619,25 @@ pub enum TorrentRecordError { EmptyNormalizedPath, } +impl TorrentRecordError { + pub fn rejection_reason(&self) -> MetadataRejectionReason { + match self { + Self::InvalidInfoHash => MetadataRejectionReason::InvalidInfoHash, + Self::EmptyName => MetadataRejectionReason::EmptyName, + Self::NameTooLong { .. } => MetadataRejectionReason::NameTooLong, + Self::InvalidName => MetadataRejectionReason::InvalidName, + Self::EmptyFileList => MetadataRejectionReason::EmptyFileList, + Self::TooManyFiles { .. } => MetadataRejectionReason::TooManyFiles, + Self::EmptyNormalizedPath => MetadataRejectionReason::EmptyPath, + Self::PathTooLong { .. } => MetadataRejectionReason::PathTooLong, + Self::PathTooDeep { .. } => MetadataRejectionReason::PathTooDeep, + Self::InvalidPath => MetadataRejectionReason::InvalidPath, + Self::SizeOverflow => MetadataRejectionReason::SizeOverflow, + Self::TotalSizeMismatch { .. } => MetadataRejectionReason::TotalSizeMismatch, + } + } +} + #[cfg(test)] pub(crate) fn test_record(hash_byte: u8, timestamp: u64) -> TorrentRecord { let files = vec![TorrentFile { @@ -477,12 +666,110 @@ mod tests { use super::*; use dht_crawler::FileInfo; + fn torrent_info(files: Vec) -> TorrentInfo { + TorrentInfo { + info_hash: "0101010101010101010101010101010101010101".into(), + magnet_link: String::new(), + name: "Example".into(), + total_size: files + .iter() + .fold(0_u64, |total, file| total.saturating_add(file.size)), + files, + piece_length: 16_384, + peers: Vec::new(), + timestamp: 1, + } + } + #[test] fn infohash_round_trips_as_lowercase_hex() { let hash = InfoHash::from_str("ABABABABABABABABABABABABABABABABABABABAB").unwrap(); assert_eq!(hash.to_string(), "abababababababababababababababababababab"); } + #[test] + fn metadata_limits_accept_boundary_and_reject_excess() { + let limits = MetadataLimits { + max_files: 1, + max_name_bytes: 7, + max_path_bytes: 8, + max_path_depth: 2, + }; + let accepted = torrent_info(vec![FileInfo { + path: "dir/a.rs".into(), + size: 1, + }]); + assert!(TorrentRecord::try_from_with_limits(accepted, limits).is_ok()); + + let too_many = torrent_info(vec![ + FileInfo { + path: "a".into(), + size: 1, + }, + FileInfo { + path: "b".into(), + size: 1, + }, + ]); + let error = TorrentRecord::try_from_with_limits(too_many, limits).unwrap_err(); + assert_eq!( + error.rejection_reason(), + MetadataRejectionReason::TooManyFiles + ); + } + + #[test] + fn unsafe_and_deep_paths_are_rejected_by_reason() { + let limits = MetadataLimits { + max_path_depth: 2, + ..MetadataLimits::default() + }; + for (path, reason) in [ + ("dir/../file", MetadataRejectionReason::InvalidPath), + ("dir//file", MetadataRejectionReason::InvalidPath), + ("a/b/c", MetadataRejectionReason::PathTooDeep), + ("a\0b", MetadataRejectionReason::InvalidPath), + ] { + let info = torrent_info(vec![FileInfo { + path: path.into(), + size: 1, + }]); + let error = TorrentRecord::try_from_with_limits(info, limits).unwrap_err(); + assert_eq!(error.rejection_reason(), reason, "path={path:?}"); + } + } + + #[test] + fn size_overflow_is_rejected_without_panicking() { + let mut info = torrent_info(vec![ + FileInfo { + path: "a".into(), + size: u64::MAX, + }, + FileInfo { + path: "b".into(), + size: 1, + }, + ]); + info.total_size = u64::MAX; + let error = + TorrentRecord::try_from_with_limits(info, MetadataLimits::default()).unwrap_err(); + assert_eq!( + error.rejection_reason(), + MetadataRejectionReason::SizeOverflow + ); + } + + #[test] + fn rule_id_changes_when_a_limit_changes() { + let defaults = MetadataLimits::default(); + let changed = MetadataLimits { + max_files: defaults.max_files - 1, + ..defaults + }; + assert_ne!(defaults.rule_id(), changed.rule_id()); + } + #[test] fn repeated_observation_updates_time_count_and_unique_peers() { let mut record = test_record(1, 10); diff --git a/dht-search/src/storage/keys.rs b/dht-search/src/storage/keys.rs index cca9153..bfb13b7 100644 --- a/dht-search/src/storage/keys.rs +++ b/dht-search/src/storage/keys.rs @@ -6,6 +6,7 @@ pub(crate) const DATABASE_FORMAT_KEY: &[u8] = b"\x00database-format"; pub(crate) const DATABASE_FORMAT_VALUE: &[u8] = b"dht-search"; pub(crate) const VERIFICATION_QUEUE_COUNT_KEY: &[u8] = b"\x00verification-queue-count"; const TORRENT_PREFIX: u8 = b't'; +const REJECTED_METADATA_PREFIX: u8 = b'r'; const CONTENT_PREFIX: u8 = b'c'; const CONTENT_GROUP_PREFIX: u8 = b'g'; const PENDING_INDEX_PREFIX: u8 = b'p'; @@ -18,6 +19,10 @@ pub(crate) fn torrent_key(info_hash: InfoHash) -> [u8; 1 + InfoHash::BYTE_LEN] { prefixed_info_hash(TORRENT_PREFIX, info_hash) } +pub(crate) fn rejected_metadata_key(info_hash: InfoHash) -> [u8; 1 + InfoHash::BYTE_LEN] { + prefixed_info_hash(REJECTED_METADATA_PREFIX, info_hash) +} + pub(crate) fn content_group_key(content_key: &[u8; 32]) -> [u8; 1 + 32] { prefixed_content_key(CONTENT_GROUP_PREFIX, content_key) } diff --git a/dht-search/src/storage/repository.rs b/dht-search/src/storage/repository.rs index c5816b2..90eeacc 100644 --- a/dht-search/src/storage/repository.rs +++ b/dht-search/src/storage/repository.rs @@ -1,6 +1,6 @@ // 负责定义元数据去重状态恢复和索引任务所需的存储接口 -use crate::domain::{ContentGroup, InfoHash, TorrentRecord, VerificationResult}; +use crate::domain::{ContentGroup, InfoHash, RejectedMetadata, TorrentRecord, VerificationResult}; pub trait TorrentRepository: Send + Sync { fn get(&self, info_hash: InfoHash) -> Result, StorageError>; @@ -11,6 +11,10 @@ pub trait TorrentRepository: Send + Sync { fn upsert(&self, observation: TorrentRecord) -> Result; + fn rejection(&self, info_hash: InfoHash) -> Result, StorageError>; + + fn record_rejection(&self, rejection: RejectedMetadata) -> Result<(), StorageError>; + fn observe_existing(&self, info_hash: InfoHash, observed_at: u64) -> Result; diff --git a/dht-search/src/storage/rocks.rs b/dht-search/src/storage/rocks.rs index beea9be..b540b36 100644 --- a/dht-search/src/storage/rocks.rs +++ b/dht-search/src/storage/rocks.rs @@ -7,15 +7,18 @@ use rocksdb::{ SliceTransform, WriteBatch, }; -use crate::domain::{ContentGroupBuilder, InfoHash, TorrentRecord, VerificationResult}; +use crate::domain::{ + ContentGroupBuilder, InfoHash, MetadataLimits, RejectedMetadata, TorrentRecord, + VerificationResult, +}; use super::{ keys::{ DATABASE_FORMAT_KEY, DATABASE_FORMAT_VALUE, VERIFICATION_QUEUE_COUNT_KEY, content_group_key, content_group_prefix, content_member_key, content_members_prefix, decode_content_member_info_hash, decode_pending_content_key, decode_verification_lease, - decode_verification_task, pending_index_key, pending_index_prefix, torrent_key, - verification_lease_key, verification_lease_prefix, verification_locator_key, + decode_verification_task, pending_index_key, pending_index_prefix, rejected_metadata_key, + torrent_key, verification_lease_key, verification_lease_prefix, verification_locator_key, verification_task_key, verification_task_prefix, }, repository::{ @@ -36,10 +39,18 @@ struct ContentGroupState { pub struct RocksTorrentRepository { db: DB, write_lock: Mutex<()>, + rejection_rule_id: [u8; 32], } impl RocksTorrentRepository { pub fn open(path: impl AsRef) -> Result { + Self::open_with_rejection_rule(path, MetadataLimits::default().rule_id()) + } + + pub fn open_with_rejection_rule( + path: impl AsRef, + rejection_rule_id: [u8; 32], + ) -> Result { let mut block_options = BlockBasedOptions::default(); block_options.set_bloom_filter(10.0, false); let block_cache = Cache::new_lru_cache(DEFAULT_BLOCK_CACHE_BYTES); @@ -55,6 +66,7 @@ impl RocksTorrentRepository { let repository = Self { db: DB::open(&options, path)?, write_lock: Mutex::new(()), + rejection_rule_id, }; repository.initialize_format()?; Ok(repository) @@ -79,6 +91,14 @@ impl RocksTorrentRepository { rmp_serde::from_slice(bytes).map_err(Into::into) } + fn encode_rejection(rejection: &RejectedMetadata) -> Result, StorageError> { + rmp_serde::to_vec_named(rejection).map_err(Into::into) + } + + fn decode_rejection(bytes: &[u8]) -> Result { + rmp_serde::from_slice(bytes).map_err(Into::into) + } + fn encode_group(state: ContentGroupState) -> Result, StorageError> { rmp_serde::to_vec_named(&state).map_err(Into::into) } @@ -197,6 +217,7 @@ impl TorrentRepository for RocksTorrentRepository { torrent_key(observation.info_hash), Self::encode(&observation)?, ); + batch.delete(rejected_metadata_key(observation.info_hash)); batch.put( content_member_key(&observation.content_key, observation.info_hash), [], @@ -206,6 +227,34 @@ impl TorrentRepository for RocksTorrentRepository { Ok(UpsertOutcome::Inserted) } + fn rejection(&self, info_hash: InfoHash) -> Result, StorageError> { + self.db + .get(rejected_metadata_key(info_hash))? + .map(|bytes| Self::decode_rejection(&bytes)) + .transpose() + } + + fn record_rejection(&self, mut rejection: RejectedMetadata) -> Result<(), StorageError> { + let _guard = self + .write_lock + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + if self.get(rejection.info_hash)?.is_some() { + return Ok(()); + } + if let Some(mut current) = self.rejection(rejection.info_hash)? + && current.rule_id == rejection.rule_id + { + current.observe_again(rejection.last_seen); + rejection = current; + } + self.db.put( + rejected_metadata_key(rejection.info_hash), + Self::encode_rejection(&rejection)?, + )?; + Ok(()) + } + fn observe_existing( &self, info_hash: InfoHash, @@ -216,7 +265,18 @@ impl TorrentRepository for RocksTorrentRepository { .lock() .unwrap_or_else(std::sync::PoisonError::into_inner); let Some(mut record) = self.get(info_hash)? else { - return Ok(false); + let Some(mut rejection) = self.rejection(info_hash)? else { + return Ok(false); + }; + if rejection.rule_id != self.rejection_rule_id { + return Ok(false); + } + rejection.observe_again(observed_at); + self.db.put( + rejected_metadata_key(info_hash), + Self::encode_rejection(&rejection)?, + )?; + return Ok(true); }; record.observe_again(observed_at, &[]); let mut batch = WriteBatch::default(); @@ -239,21 +299,42 @@ impl TorrentRepository for RocksTorrentRepository { .lock() .unwrap_or_else(std::sync::PoisonError::into_inner); let keys: Vec<_> = info_hashes.iter().copied().map(torrent_key).collect(); + let rejected_keys: Vec<_> = info_hashes + .iter() + .copied() + .map(rejected_metadata_key) + .collect(); let records = self.db.multi_get(keys.iter()); + let rejections = self.db.multi_get(rejected_keys.iter()); let mut unknown = Vec::with_capacity(info_hashes.len()); let mut batch = WriteBatch::default(); let mut updated = 0_usize; let mut dirty_groups = std::collections::BTreeSet::new(); - for ((info_hash, key), record) in info_hashes.iter().zip(&keys).zip(records) { - let Some(bytes) = record? else { - unknown.push(*info_hash); + for ((((info_hash, key), rejected_key), record), rejection) in info_hashes + .iter() + .zip(&keys) + .zip(&rejected_keys) + .zip(records) + .zip(rejections) + { + if let Some(bytes) = record? { + let mut record = Self::decode(&bytes)?; + record.observe_again(observed_at, &[]); + batch.put(key, Self::encode(&record)?); + dirty_groups.insert(record.content_key); + updated += 1; continue; - }; - let mut record = Self::decode(&bytes)?; - record.observe_again(observed_at, &[]); - batch.put(key, Self::encode(&record)?); - dirty_groups.insert(record.content_key); - updated += 1; + } + if let Some(bytes) = rejection? { + let mut rejection = Self::decode_rejection(&bytes)?; + if rejection.rule_id == self.rejection_rule_id { + rejection.observe_again(observed_at); + batch.put(rejected_key, Self::encode_rejection(&rejection)?); + updated += 1; + continue; + } + } + unknown.push(*info_hash); } if updated > 0 { for content_key in dirty_groups { @@ -666,6 +747,78 @@ mod tests { assert_eq!(stored.seen_count, 2); } + #[test] + fn rejection_survives_restart_and_blocks_the_same_rule() { + let directory = TempDir::new().unwrap(); + let limits = MetadataLimits::default(); + let info_hash = InfoHash::from_bytes([9; 20]); + { + let repository = RocksTorrentRepository::open_with_rejection_rule( + directory.path(), + limits.rule_id(), + ) + .unwrap(); + repository + .record_rejection(RejectedMetadata::new( + info_hash, + crate::domain::MetadataRejectionReason::TooManyFiles, + limits.rule_id(), + 10, + )) + .unwrap(); + assert!(repository.get(info_hash).unwrap().is_none()); + assert!(repository.observe_existing(info_hash, 20).unwrap()); + } + + let repository = + RocksTorrentRepository::open_with_rejection_rule(directory.path(), limits.rule_id()) + .unwrap(); + assert!(repository.observe_existing(info_hash, 30).unwrap()); + let rejection = repository.rejection(info_hash).unwrap().unwrap(); + assert_eq!(rejection.first_rejected_at, 10); + assert_eq!(rejection.last_seen, 30); + assert_eq!(rejection.seen_count, 3); + assert!(repository.pending_index(10).unwrap().is_empty()); + } + + #[test] + fn changed_rule_releases_old_rejection_and_valid_upsert_removes_it() { + let directory = TempDir::new().unwrap(); + let old_limits = MetadataLimits::default(); + let info_hash = InfoHash::from_bytes([10; 20]); + { + let repository = RocksTorrentRepository::open_with_rejection_rule( + directory.path(), + old_limits.rule_id(), + ) + .unwrap(); + repository + .record_rejection(RejectedMetadata::new( + info_hash, + crate::domain::MetadataRejectionReason::TooManyFiles, + old_limits.rule_id(), + 10, + )) + .unwrap(); + } + + let new_limits = MetadataLimits { + max_files: old_limits.max_files + 1, + ..old_limits + }; + let repository = RocksTorrentRepository::open_with_rejection_rule( + directory.path(), + new_limits.rule_id(), + ) + .unwrap(); + assert!(!repository.observe_existing(info_hash, 20).unwrap()); + let mut record = test_record(10, 20); + record.info_hash = info_hash; + repository.upsert(record).unwrap(); + assert!(repository.rejection(info_hash).unwrap().is_none()); + assert!(repository.get(info_hash).unwrap().is_some()); + } + #[test] fn equal_content_maps_multiple_infohashes_without_merging_records() { let directory = TempDir::new().unwrap(); diff --git a/web/src/App.vue b/web/src/App.vue index 0291f14..20ecab6 100644 --- a/web/src/App.vue +++ b/web/src/App.vue @@ -212,7 +212,7 @@ onBeforeUnmount(() => {

服务运行状态

-
DHT 节点{{ stats.nodes.toLocaleString() }}
已索引内容{{ stats.indexed_documents.toLocaleString() }}
获取成功{{ stats.metadata_ok.toLocaleString() }}
下载中{{ stats.metadata_in_flight }}
新收录{{ stats.persistence_inserted.toLocaleString() }}
验证成功{{ stats.verification_succeeded.toLocaleString() }}
+
DHT 节点{{ stats.nodes.toLocaleString() }}
已索引内容{{ stats.indexed_documents.toLocaleString() }}
获取成功{{ stats.metadata_ok.toLocaleString() }}
下载中{{ stats.metadata_in_flight }}
新收录{{ stats.persistence_inserted.toLocaleString() }}
已过滤{{ stats.metadata_filtered.toLocaleString() }}
验证成功{{ stats.verification_succeeded.toLocaleString() }}

无法获取服务状态

diff --git a/web/src/types/api.ts b/web/src/types/api.ts index 3d3edba..4f617d4 100644 --- a/web/src/types/api.ts +++ b/web/src/types/api.ts @@ -99,6 +99,7 @@ export interface ServiceStats { metadata_ok: number metadata_failed: number metadata_in_flight: number + metadata_filtered: number persistence_inserted: number persistence_updated: number persistence_queue: number