diff --git a/Cargo.lock b/Cargo.lock index f3110170..4d4dc6f2 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1847,7 +1847,7 @@ dependencies = [ [[package]] name = "nzb-nntp" -version = "0.2.24" +version = "0.2.26" dependencies = [ "anyhow", "arc-swap", @@ -2816,6 +2816,7 @@ dependencies = [ "clap", "crc32fast", "flate2", + "hex", "http", "libc", "mime_guess", @@ -2839,6 +2840,7 @@ dependencies = [ "serde", "serde_json", "serial_test", + "sha2 0.11.0", "tempfile", "tokio", "tokio-util", diff --git a/apps/rustnzb/Cargo.toml b/apps/rustnzb/Cargo.toml index 28f5565c..a48273db 100644 --- a/apps/rustnzb/Cargo.toml +++ b/apps/rustnzb/Cargo.toml @@ -42,6 +42,8 @@ reqwest = { workspace = true, features = ["json"] } regex = { workspace = true } uuid = { workspace = true } base64 = { workspace = true } +sha2 = "0.11" +hex = { workspace = true } rust-embed = { version = "8", features = ["debug-embed", "interpolate-folder-path"] } mime_guess = "2" libc = "0.2" diff --git a/apps/rustnzb/src/group_observation.rs b/apps/rustnzb/src/group_observation.rs new file mode 100644 index 00000000..e69d07fd --- /dev/null +++ b/apps/rustnzb/src/group_observation.rs @@ -0,0 +1,346 @@ +use std::{collections::BTreeSet, sync::Arc}; + +use axum::{Json, extract::State}; +use base64::{Engine as _, engine::general_purpose::STANDARD as BASE64}; +use nzb_web::{ + error::ApiError, + nzb_core::nzb_nntp::{ + DefectiveOverviewRow, LosslessOverviewRow, NntpConnection, NntpError, OverviewFormat, + }, + state::AppState, +}; +use serde_json::{Value, json}; +use sha2::{Digest, Sha256}; + +mod availability; +mod body_prefix; +mod clear_search; +mod contract; + +pub(crate) use availability::h_article_availability; +pub(crate) use body_prefix::h_article_body_prefix; +pub(crate) use clear_search::h_clear_search; + +use contract::{ArticleHeadInput, OverviewRangeInput}; + +fn blocked(operation: &str, request_id: &str, group: &str, failure_code: &str) -> Json { + Json(json!({ + "status": "blocked", + "operation": operation, + "request_id": request_id, + "group": group, + "failure_code": failure_code + })) +} + +fn nntp_failure(error: &NntpError, operation: &str) -> &'static str { + match error { + NntpError::Auth(_) | NntpError::AuthRequired(_) => "nntp_authentication_failed", + NntpError::PermissionDenied(_) => "nntp_operation_not_authorized", + NntpError::NoSuchGroup(_) => "nntp_group_unavailable", + NntpError::ServiceUnavailable(_) => "nntp_service_unavailable", + NntpError::Timeout(_) => "nntp_operation_timed_out", + NntpError::ResponseTooLarge(_) => "nntp_observation_response_limit_exceeded", + NntpError::UnsupportedCommand(_) => "nntp_operation_unsupported", + NntpError::Connection(_) | NntpError::Io(_) | NntpError::Tls(_) => { + "nntp_transport_unavailable" + } + NntpError::ArticleNotFound(_) + if matches!(operation, "article_head" | "article_body_prefix") => + { + "nntp_article_unavailable" + } + NntpError::Protocol(_) if operation == "article_head" => "nntp_head_unavailable", + NntpError::Protocol(_) if operation == "article_body_prefix" => "nntp_body_unavailable", + NntpError::Protocol(_) if operation == "article_availability" => "nntp_stat_unavailable", + NntpError::Protocol(_) => "nntp_overview_unavailable", + _ => "nntp_operation_failed", + } +} + +pub(crate) async fn h_article_head( + State(state): State>, + Json(input): Json, +) -> Result, ApiError> { + input.validate().map_err(ApiError::bad_request)?; + let servers = state.queue_manager.get_servers(); + let Some(server) = servers.first() else { + return Ok(blocked( + "article_head", + &input.request_id, + &input.group, + "nntp_provider_not_configured", + )); + }; + let mut connection = NntpConnection::new(format!("head-{}", input.request_id)); + if let Err(error) = connection.connect(server).await { + return Ok(blocked( + "article_head", + &input.request_id, + &input.group, + nntp_failure(&error, "article_head"), + )); + } + let group = match connection.group(&input.group).await { + Ok(group) => group, + Err(error) => { + let _ = connection.quit().await; + return Ok(blocked( + "article_head", + &input.request_id, + &input.group, + nntp_failure(&error, "article_head"), + )); + } + }; + if group.name != input.group + || input.article_number < group.first + || input.article_number > group.last + { + let _ = connection.quit().await; + return Ok(blocked( + "article_head", + &input.request_id, + &input.group, + "nntp_article_binding_invalid", + )); + } + let response = match connection + .fetch_head_number(input.article_number, input.max_header_bytes) + .await + { + Ok(response) => response, + Err(error) => { + let _ = connection.quit().await; + return Ok(blocked( + "article_head", + &input.request_id, + &input.group, + nntp_failure(&error, "article_head"), + )); + } + }; + let _ = connection.quit().await; + let Some(headers) = response.data else { + return Ok(blocked( + "article_head", + &input.request_id, + &input.group, + "nntp_head_response_invalid", + )); + }; + if headers.len() > input.max_header_bytes { + return Ok(blocked( + "article_head", + &input.request_id, + &input.group, + "nntp_head_byte_limit_exceeded", + )); + } + let digest = hex::encode(Sha256::digest(&headers)); + Ok(Json(json!({ + "status": "complete", + "operation": "article_head", + "request_id": input.request_id, + "group": group.name, + "group_first_article": group.first, + "group_last_article": group.last, + "article_number": input.article_number, + "header_byte_count": headers.len(), + "headers_base64": BASE64.encode(headers), + "headers_sha256": digest + }))) +} + +fn missing_ranges(start: u64, end: u64, present: &BTreeSet) -> Vec<(u64, u64)> { + let mut ranges = Vec::new(); + let mut missing_start = None; + for article in start..=end { + if present.contains(&article) { + if let Some(first) = missing_start.take() { + ranges.push((first, article - 1)); + } + } else if missing_start.is_none() { + missing_start = Some(article); + } + } + if let Some(first) = missing_start { + ranges.push((first, end)); + } + ranges +} + +fn digest_parts<'a>(prefix: &[u8], parts: impl IntoIterator) -> String { + let mut digest = Sha256::new(); + digest.update(prefix); + for part in parts { + digest.update((part.len() as u64).to_be_bytes()); + digest.update(part); + } + hex::encode(digest.finalize()) +} + +fn format_digest(format: &OverviewFormat) -> String { + digest_parts(b"overview-format", format.fields.iter().map(Vec::as_slice)) +} + +fn row_json(row: LosslessOverviewRow, format_digest: &str) -> Value { + let article = row.article_number.to_be_bytes(); + let digest = digest_parts( + b"overview-row", + std::iter::once(article.as_slice()) + .chain(std::iter::once(format_digest.as_bytes())) + .chain(row.fields.iter().map(Vec::as_slice)), + ); + json!({ + "article_number": row.article_number, + "fields_base64": row.fields.into_iter().map(|field| BASE64.encode(field)).collect::>(), + "row_sha256": digest + }) +} + +fn defective_row_json(row: DefectiveOverviewRow) -> Value { + let digest = digest_parts(b"defective-overview-row", [row.wire_line.as_slice()]); + json!({ + "article_number": row.article_number, + "wire_row_base64": BASE64.encode(row.wire_line), + "raw_sha256": digest, + "failure_code": row.failure_code.as_str() + }) +} + +pub(crate) async fn h_overview_range( + State(state): State>, + Json(input): Json, +) -> Result, ApiError> { + input.validate().map_err(ApiError::bad_request)?; + let servers = state.queue_manager.get_servers(); + let Some(server) = servers.first() else { + return Ok(blocked( + "overview_range", + &input.request_id, + &input.group, + "nntp_provider_not_configured", + )); + }; + let mut connection = NntpConnection::new(format!("overview-{}", input.request_id)); + if let Err(error) = connection.connect(server).await { + return Ok(blocked( + "overview_range", + &input.request_id, + &input.group, + nntp_failure(&error, "overview_range"), + )); + } + let group = match connection.group(&input.group).await { + Ok(group) => group, + Err(error) => { + let _ = connection.quit().await; + return Ok(blocked( + "overview_range", + &input.request_id, + &input.group, + nntp_failure(&error, "overview_range"), + )); + } + }; + if group.name != input.group { + let _ = connection.quit().await; + return Ok(blocked( + "overview_range", + &input.request_id, + &input.group, + "nntp_group_binding_invalid", + )); + } + let format = match connection.overview_format().await { + Ok(format) => format, + Err(error) => { + let _ = connection.quit().await; + return Ok(blocked( + "overview_range", + &input.request_id, + &input.group, + nntp_failure(&error, "overview_range"), + )); + } + }; + let overview = match connection + .xover_lossless(input.start_article, input.end_article, &format) + .await + { + Ok(overview) => overview, + Err(error) => { + let _ = connection.quit().await; + return Ok(blocked( + "overview_range", + &input.request_id, + &input.group, + nntp_failure(&error, "overview_range"), + )); + } + }; + let _ = connection.quit().await; + let returned = overview.rows.len() + overview.defective_rows.len(); + if returned as u64 > input.max_headers { + return Ok(blocked( + "overview_range", + &input.request_id, + &input.group, + "nntp_overview_header_limit_exceeded", + )); + } + let present = overview + .rows + .iter() + .map(|row| row.article_number) + .chain( + overview + .defective_rows + .iter() + .filter_map(|row| row.article_number), + ) + .collect::>(); + let missing = missing_ranges(input.start_article, input.end_article, &present); + let format_digest = format_digest(&format); + let valid = overview.rows.len(); + let defective = overview.defective_rows.len(); + let format_fields = format + .fields + .into_iter() + .map(|field| BASE64.encode(field)) + .collect::>(); + let rows = overview + .rows + .into_iter() + .map(|row| row_json(row, &format_digest)) + .collect::>(); + let defective_rows = overview + .defective_rows + .into_iter() + .map(defective_row_json) + .collect::>(); + Ok(Json(json!({ + "status": "complete", + "operation": "overview_range", + "request_id": input.request_id, + "group": group.name, + "group_first_article": group.first, + "group_last_article": group.last, + "requested_start_article": input.start_article, + "requested_end_article": input.end_article, + "returned_row_count": returned, + "valid_row_count": valid, + "defective_row_count": defective, + "missing_ranges": missing, + "overview_format": { + "fields_base64": format_fields, + "sha256": format_digest + }, + "rows": rows, + "defective_rows": defective_rows + }))) +} + +#[cfg(test)] +mod tests; diff --git a/apps/rustnzb/src/group_observation/availability.rs b/apps/rustnzb/src/group_observation/availability.rs new file mode 100644 index 00000000..6c5e2b4b --- /dev/null +++ b/apps/rustnzb/src/group_observation/availability.rs @@ -0,0 +1,66 @@ +use super::{ + contract::{ArticleAvailabilityInput, article_availability_digest}, + nntp_failure, +}; +use axum::{Json, extract::State}; +use nzb_web::{ + error::ApiError, + nzb_core::nzb_nntp::{NntpConnection, StatPipeline}, + state::AppState, +}; +use serde_json::{Value, json}; +use std::sync::Arc; + +pub(crate) async fn h_article_availability( + State(state): State>, + Json(input): Json, +) -> Result, ApiError> { + input.validate().map_err(ApiError::bad_request)?; + let servers = state.queue_manager.get_servers(); + let Some(server) = servers.first() else { + return Ok(blocked(&input, "nntp_provider_not_configured")); + }; + let mut connection = NntpConnection::new(format!("availability-{}", input.request_id)); + if let Err(error) = connection.connect(server).await { + return Ok(blocked( + &input, + nntp_failure(&error, "article_availability"), + )); + } + let mut pipeline = StatPipeline::new(); + for message_id in &input.message_ids { + pipeline.add(message_id.clone()); + } + let observations = match pipeline.execute(&mut connection).await { + Ok(observations) => observations, + Err(error) => { + let _ = connection.quit().await; + return Ok(blocked( + &input, + nntp_failure(&error, "article_availability"), + )); + } + }; + let _ = connection.quit().await; + let available_segment_count = observations.iter().filter(|item| item.exists).count(); + let unavailable_segment_count = observations.len().saturating_sub(available_segment_count); + Ok(Json(json!({ + "status": "complete", + "operation": "article_availability", + "request_id": input.request_id, + "sample_sha256": input.sample_sha256, + "sampled_segment_count": observations.len(), + "available_segment_count": available_segment_count, + "unavailable_segment_count": unavailable_segment_count + }))) +} + +fn blocked(input: &ArticleAvailabilityInput, failure_code: &str) -> Json { + Json(json!({ + "status": "blocked", + "operation": "article_availability", + "request_id": input.request_id, + "sample_sha256": article_availability_digest(&input.message_ids), + "failure_code": failure_code + })) +} diff --git a/apps/rustnzb/src/group_observation/body_prefix.rs b/apps/rustnzb/src/group_observation/body_prefix.rs new file mode 100644 index 00000000..8dc1b1f7 --- /dev/null +++ b/apps/rustnzb/src/group_observation/body_prefix.rs @@ -0,0 +1,191 @@ +use super::{blocked, contract::ArticleBodyPrefixInput, nntp_failure}; +use axum::{Json, extract::State}; +use base64::{Engine as _, engine::general_purpose::STANDARD as BASE64}; +use nzb_web::{error::ApiError, nzb_core::nzb_nntp::NntpConnection, state::AppState}; +use serde_json::{Value, json}; +use sha2::{Digest, Sha256}; +use std::sync::Arc; + +pub(crate) async fn h_article_body_prefix( + State(state): State>, + Json(input): Json, +) -> Result, ApiError> { + input.validate().map_err(ApiError::bad_request)?; + let servers = state.queue_manager.get_servers(); + let Some(server) = servers.first() else { + return Ok(blocked( + "article_body_prefix", + &input.request_id, + &input.group, + "nntp_provider_not_configured", + )); + }; + let mut server = server.clone(); + server.compress = false; + let mut connection = NntpConnection::new(format!("body-prefix-{}", input.request_id)); + if let Err(error) = connection.connect(&server).await { + return Ok(blocked( + "article_body_prefix", + &input.request_id, + &input.group, + nntp_failure(&error, "article_body_prefix"), + )); + } + let group = match connection.group(&input.group).await { + Ok(group) => group, + Err(error) => { + let _ = connection.quit().await; + return Ok(blocked( + "article_body_prefix", + &input.request_id, + &input.group, + nntp_failure(&error, "article_body_prefix"), + )); + } + }; + let prefix = match connection + .fetch_body_prefix(&input.message_id, input.max_wire_bytes) + .await + { + Ok(prefix) => prefix, + Err(error) => { + let _ = connection.quit().await; + return Ok(blocked( + "article_body_prefix", + &input.request_id, + &input.group, + nntp_failure(&error, "article_body_prefix"), + )); + } + }; + let _ = connection.quit().await; + let decoded = decode_payload_prefix(&prefix.data, prefix.complete, input.max_payload_bytes); + let wire_sha256 = hex::encode(Sha256::digest(&prefix.data)); + let payload_sha256 = hex::encode(Sha256::digest(&decoded.bytes)); + Ok(Json(json!({ + "status": "complete", + "operation": "article_body_prefix", + "request_id": input.request_id, + "group": group.name, + "group_first_article": group.first, + "group_last_article": group.last, + "message_id": input.message_id, + "wire_prefix_byte_count": prefix.data.len(), + "wire_prefix_base64": BASE64.encode(prefix.data), + "wire_prefix_sha256": wire_sha256, + "body_complete": prefix.complete, + "payload_encoding": decoded.encoding, + "payload_prefix_byte_count": decoded.bytes.len(), + "payload_prefix_base64": BASE64.encode(decoded.bytes), + "payload_prefix_sha256": payload_sha256, + "payload_complete": decoded.complete + }))) +} + +struct DecodedPrefix { + bytes: Vec, + encoding: &'static str, + complete: bool, +} + +fn decode_payload_prefix(wire: &[u8], body_complete: bool, maximum: usize) -> DecodedPrefix { + let mut started = false; + let mut ended = false; + let mut truncated = false; + // Grow on demand rather than reserving from a request-derived size. The + // decode loop below already caps output at `maximum` (validated + // <= MAX_PAYLOAD_PREFIX_BYTES), so the buffer stays bounded without seeding + // an allocation from a caller-influenced value. + let mut output = Vec::new(); + for raw_line in wire.split_inclusive(|byte| *byte == b'\n') { + let line = raw_line.strip_suffix(b"\n").unwrap_or(raw_line); + let line = line.strip_suffix(b"\r").unwrap_or(line); + if !started { + if line.starts_with(b"=ybegin ") { + started = true; + } + continue; + } + if line.starts_with(b"=ypart ") { + continue; + } + if line.starts_with(b"=yend ") { + ended = true; + break; + } + let mut index = 0usize; + while index < line.len() { + if output.len() == maximum { + truncated = true; + break; + } + let encoded = if line[index] == b'=' { + let Some(escaped) = line.get(index + 1) else { + truncated = true; + break; + }; + index += 2; + escaped.wrapping_sub(64) + } else { + let value = line[index]; + index += 1; + value + }; + output.push(encoded.wrapping_sub(42)); + } + if truncated { + break; + } + } + if started { + DecodedPrefix { + bytes: output, + encoding: "yenc", + complete: body_complete && ended && !truncated, + } + } else { + let length = wire.len().min(maximum); + DecodedPrefix { + bytes: wire[..length].to_vec(), + encoding: "plain", + complete: body_complete && wire.len() <= maximum, + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn encode(data: &[u8]) -> Vec { + let mut output = b"=ybegin line=128 size=8 name=random.bin\r\n".to_vec(); + for byte in data { + let encoded = byte.wrapping_add(42); + if matches!(encoded, 0 | b'\n' | b'\r' | b'=') { + output.push(b'='); + output.push(encoded.wrapping_add(64)); + } else { + output.push(encoded); + } + } + output.extend_from_slice(b"\r\n=yend size=8\r\n"); + output + } + + #[test] + fn yenc_prefix_decodes_binary_magic_without_requiring_a_complete_article() { + let wire = encode(b"PAR2\0PKT"); + let prefix = decode_payload_prefix(&wire[..wire.len() - 8], false, 64); + assert_eq!(prefix.bytes, b"PAR2\0PKT"); + assert_eq!(prefix.encoding, "yenc"); + assert!(!prefix.complete); + } + + #[test] + fn plain_small_metadata_remains_byte_exact() { + let prefix = decode_payload_prefix(b"release.nfo\r\n", true, 64); + assert_eq!(prefix.bytes, b"release.nfo\r\n"); + assert_eq!(prefix.encoding, "plain"); + assert!(prefix.complete); + } +} diff --git a/apps/rustnzb/src/group_observation/clear_search.rs b/apps/rustnzb/src/group_observation/clear_search.rs new file mode 100644 index 00000000..9a345f22 --- /dev/null +++ b/apps/rustnzb/src/group_observation/clear_search.rs @@ -0,0 +1,707 @@ +use super::{defective_row_json, format_digest, missing_ranges, nntp_failure, row_json}; +use crate::group_observation::contract::{ClearSearchInput, ClearSearchRangeInput, now_unix_ms}; +use axum::{Json, extract::State}; +use base64::{Engine as _, engine::general_purpose::STANDARD as BASE64}; +use nzb_web::{ + error::ApiError, + nzb_core::nzb_nntp::{ + ArticleRange, HeaderEntry, LosslessOverviewRows, NntpConnection, NntpError, OverviewFormat, + }, + state::AppState, +}; +use serde_json::{Value, json}; +use std::{collections::BTreeSet, future::Future, sync::Arc, time::Duration}; +use tokio::time::Instant; + +const MAX_COMMAND_RESPONSE_BYTES: usize = 8 * 1024 * 1024; + +#[derive(Clone, Copy, Eq, PartialEq)] +enum AcceleratorState { + Supported, + Unsupported, + Defective, + Unverified, +} + +impl AcceleratorState { + fn as_str(self) -> &'static str { + match self { + Self::Supported => "supported", + Self::Unsupported => "unsupported", + Self::Defective => "defective", + Self::Unverified => "unverified", + } + } +} + +struct Calibration { + state: AcceleratorState, + positive_matches: usize, + negative_matches: usize, +} + +struct CalibrationResult { + calibration: Calibration, + stop: Option, +} + +struct ObservedRange { + overview: LosslessOverviewRows, + matches: Vec, + accelerator_state: AcceleratorState, + accelerator_failure_code: Option<&'static str>, + response_bytes: usize, + elapsed_ms: u64, +} + +#[derive(Clone, Copy, Debug)] +struct Stop { + receipt_state: &'static str, + failure_code: &'static str, +} + +struct Budget { + maximum: usize, + used: usize, +} + +impl Budget { + fn new(maximum: usize) -> Self { + Self { maximum, used: 0 } + } + + fn command_limit(&self) -> Result { + let remaining = self.maximum.saturating_sub(self.used); + if remaining == 0 { + Err(Stop { + receipt_state: "refused", + failure_code: "nntp_observation_response_limit_exceeded", + }) + } else { + Ok(remaining.min(MAX_COMMAND_RESPONSE_BYTES)) + } + } + + fn consume(&mut self, bytes: usize) -> Result<(), Stop> { + self.used = self.used.checked_add(bytes).ok_or(Stop { + receipt_state: "refused", + failure_code: "nntp_observation_response_limit_exceeded", + })?; + if self.used > self.maximum { + return Err(Stop { + receipt_state: "refused", + failure_code: "nntp_observation_response_limit_exceeded", + }); + } + Ok(()) + } +} + +struct Deadline { + at: Instant, +} + +impl Deadline { + fn from_input(input: &ClearSearchInput) -> Result { + let now = now_unix_ms()?; + let remaining = input.deadline_at_unix_ms.saturating_sub(now); + Ok(Self { + at: Instant::now() + Duration::from_millis(remaining), + }) + } + + async fn run(&self, future: F) -> Result { + tokio::time::timeout_at(self.at, future) + .await + .map_err(|_| Stop { + receipt_state: "cancelled", + failure_code: "nntp_operation_timed_out", + }) + } +} + +enum XpatOutcome { + Complete(Vec), + Unsupported, + Defective(&'static str), + Stopped(Stop), +} + +pub(crate) async fn h_clear_search( + State(state): State>, + Json(input): Json, +) -> Result, ApiError> { + input.validate().map_err(ApiError::bad_request)?; + let deadline = Deadline::from_input(&input).map_err(ApiError::bad_request)?; + Ok(Json(observe(state, &input, &deadline).await)) +} + +async fn observe(state: Arc, input: &ClearSearchInput, deadline: &Deadline) -> Value { + let servers = state.queue_manager.get_servers(); + let Some(server) = servers.first() else { + return failed_response( + input, + Stop { + receipt_state: "refused", + failure_code: "nntp_provider_not_configured", + }, + ); + }; + let mut connection = NntpConnection::new(format!("clear-{}", input.request_id)); + match deadline.run(connection.connect(server)).await { + Ok(Ok(())) => {} + Ok(Err(error)) => return failed_response(input, stop_for_error(&error)), + Err(stop) => return failed_response(input, stop), + } + let group = match deadline.run(connection.group(&input.group)).await { + Ok(Ok(group)) => group, + Ok(Err(error)) => return failed_response(input, stop_for_error(&error)), + Err(stop) => return failed_response(input, stop), + }; + if group.name != input.group + || group.first == 0 + || group.last < group.first + || input + .ranges + .iter() + .any(|range| range.start_article < group.first || range.end_article > group.last) + { + return failed_response( + input, + Stop { + receipt_state: "refused", + failure_code: "nntp_group_binding_invalid", + }, + ); + } + let format = match deadline.run(connection.overview_format()).await { + Ok(Ok(format)) => format, + Ok(Err(error)) => return failed_response(input, stop_for_error(&error)), + Err(stop) => return failed_response(input, stop), + }; + let mut budget = Budget::new(input.max_response_bytes); + let mut calibration = None; + let mut range_rows = Vec::new(); + let patterns = input + .patterns + .iter() + .map(String::as_str) + .collect::>(); + let mut next_range = input.ranges.len(); + let mut following_stop = None; + for (index, range) in input.ranges.iter().enumerate() { + let started = Instant::now(); + let range_bytes_before = budget.used; + let response_limit = match budget.command_limit() { + Ok(limit) => limit, + Err(stop) => { + next_range = index; + following_stop = Some(stop); + break; + } + }; + let overview = match deadline + .run(connection.xover_lossless_bounded( + range.start_article, + range.end_article, + &format, + response_limit, + )) + .await + { + Ok(Ok(response)) => { + if let Err(stop) = budget.consume(response.response_bytes) { + next_range = index; + following_stop = Some(stop); + break; + } + response.value + } + Ok(Err(error)) => { + next_range = index; + following_stop = Some(stop_for_error(&error)); + break; + } + Err(stop) => { + next_range = index; + following_stop = Some(stop); + break; + } + }; + if calibration.is_none() { + let result = calibrate( + &mut connection, + range, + &overview, + input.max_matches_per_range, + deadline, + &mut budget, + ) + .await; + if result.calibration.state != AcceleratorState::Unverified { + calibration = Some(result.calibration); + } + if let Some(stop) = result.stop { + range_rows.push(range_json( + range, + &format, + ObservedRange { + overview, + matches: Vec::new(), + accelerator_state: AcceleratorState::Defective, + accelerator_failure_code: Some(stop.failure_code), + response_bytes: budget.used.saturating_sub(range_bytes_before), + elapsed_ms: elapsed_millis(started), + }, + )); + next_range = index + 1; + following_stop = Some(stop); + break; + } + } + let accelerator_state = calibration + .as_ref() + .map_or(AcceleratorState::Unverified, |value| value.state); + let mut accelerator_failure = None; + let mut matches = Vec::new(); + if accelerator_state == AcceleratorState::Supported { + match bounded_xpat( + &mut connection, + range, + &patterns, + input.max_matches_per_range, + deadline, + &mut budget, + ) + .await + { + XpatOutcome::Complete(observed) => matches = observed, + XpatOutcome::Unsupported => { + accelerator_failure = Some("nntp_header_pattern_unavailable"); + if let Some(calibration) = calibration.as_mut() { + calibration.state = AcceleratorState::Defective; + } + } + XpatOutcome::Defective(code) => { + accelerator_failure = Some(code); + if let Some(calibration) = calibration.as_mut() { + calibration.state = AcceleratorState::Defective; + } + } + XpatOutcome::Stopped(stop) => { + accelerator_failure = Some(stop.failure_code); + if let Some(calibration) = calibration.as_mut() { + calibration.state = AcceleratorState::Defective; + } + range_rows.push(range_json( + range, + &format, + ObservedRange { + overview, + matches, + accelerator_state: AcceleratorState::Defective, + accelerator_failure_code: accelerator_failure, + response_bytes: budget.used.saturating_sub(range_bytes_before), + elapsed_ms: elapsed_millis(started), + }, + )); + next_range = index + 1; + following_stop = Some(stop); + break; + } + } + } + let final_accelerator_state = calibration + .as_ref() + .map_or(AcceleratorState::Unverified, |value| value.state); + range_rows.push(range_json( + range, + &format, + ObservedRange { + overview, + matches, + accelerator_state: final_accelerator_state, + accelerator_failure_code: accelerator_failure, + response_bytes: budget.used.saturating_sub(range_bytes_before), + elapsed_ms: elapsed_millis(started), + }, + )); + } + if let Some(stop) = following_stop { + append_failure_receipts(&mut range_rows, &input.ranges[next_range..], stop); + } + let calibration = calibration.unwrap_or(Calibration { + state: AcceleratorState::Unverified, + positive_matches: 0, + negative_matches: 0, + }); + response( + input, + Some((&group.name, group.first, group.last)), + Some(&format), + calibration, + budget.used, + range_rows, + true, + ) +} + +async fn calibrate( + connection: &mut NntpConnection, + range: &ClearSearchRangeInput, + overview: &LosslessOverviewRows, + max_matches: usize, + deadline: &Deadline, + budget: &mut Budget, +) -> CalibrationResult { + let Some((article_number, token)) = positive_control(overview) else { + return CalibrationResult { + calibration: Calibration { + state: AcceleratorState::Unverified, + positive_matches: 0, + negative_matches: 0, + }, + stop: None, + }; + }; + let positive_pattern = format!("*{token}*"); + let positive = match bounded_xpat( + connection, + range, + &[positive_pattern.as_str()], + max_matches, + deadline, + budget, + ) + .await + { + XpatOutcome::Complete(matches) => matches, + XpatOutcome::Unsupported => { + return calibration_result(AcceleratorState::Unsupported, 0, 0, None); + } + XpatOutcome::Defective(_) => { + return calibration_result(AcceleratorState::Defective, 0, 0, None); + } + XpatOutcome::Stopped(stop) => { + return calibration_result(AcceleratorState::Defective, 0, 0, Some(stop)); + } + }; + let negative_pattern = format!("*newsgroupsnegative{article_number:x}nomatch*"); + let negative = match bounded_xpat( + connection, + range, + &[negative_pattern.as_str()], + max_matches, + deadline, + budget, + ) + .await + { + XpatOutcome::Complete(matches) => matches, + XpatOutcome::Unsupported | XpatOutcome::Defective(_) => { + return calibration_result(AcceleratorState::Defective, positive.len(), 0, None); + } + XpatOutcome::Stopped(stop) => { + return calibration_result(AcceleratorState::Defective, positive.len(), 0, Some(stop)); + } + }; + let positive_is_correct = positive + .iter() + .any(|matched| matched.article_num == article_number); + calibration_result( + if positive_is_correct && negative.is_empty() { + AcceleratorState::Supported + } else { + AcceleratorState::Defective + }, + positive.len(), + negative.len(), + None, + ) +} + +async fn bounded_xpat( + connection: &mut NntpConnection, + range: &ClearSearchRangeInput, + patterns: &[&str], + max_matches: usize, + deadline: &Deadline, + budget: &mut Budget, +) -> XpatOutcome { + let response_limit = match budget.command_limit() { + Ok(limit) => limit, + Err(stop) => return XpatOutcome::Stopped(stop), + }; + let response = match deadline + .run(connection.xpat_bounded( + "Subject", + ArticleRange::Range(range.start_article, range.end_article), + patterns, + response_limit, + )) + .await + { + Ok(Ok(response)) => response, + Ok(Err(NntpError::UnsupportedCommand(_))) => return XpatOutcome::Unsupported, + Ok(Err(error)) => return XpatOutcome::Stopped(stop_for_error(&error)), + Err(stop) => return XpatOutcome::Stopped(stop), + }; + if let Err(stop) = budget.consume(response.response_bytes) { + return XpatOutcome::Stopped(stop); + } + if response.value.len() > max_matches { + XpatOutcome::Defective("nntp_header_pattern_match_limit_exceeded") + } else if response + .value + .iter() + .try_fold(BTreeSet::new(), |mut seen, matched| { + if matched.article_num < range.start_article + || matched.article_num > range.end_article + || !seen.insert(matched.article_num) + { + None + } else { + Some(seen) + } + }) + .is_none() + { + XpatOutcome::Defective("nntp_header_pattern_contract_invalid") + } else { + XpatOutcome::Complete(response.value) + } +} + +fn calibration_result( + state: AcceleratorState, + positive_matches: usize, + negative_matches: usize, + stop: Option, +) -> CalibrationResult { + CalibrationResult { + calibration: Calibration { + state, + positive_matches, + negative_matches, + }, + stop, + } +} + +fn positive_control(overview: &LosslessOverviewRows) -> Option<(u64, String)> { + overview.rows.iter().find_map(|row| { + let subject = std::str::from_utf8(row.fields.first()?).ok()?; + let token = subject + .split(|character: char| !character.is_ascii_alphanumeric()) + .find(|token| { + token.len() >= 4 && token.bytes().all(|byte| byte.is_ascii_alphanumeric()) + })?; + Some((row.article_number, token.to_string())) + }) +} + +fn range_json( + range: &ClearSearchRangeInput, + format: &OverviewFormat, + observed: ObservedRange, +) -> Value { + let ObservedRange { + overview, + matches, + accelerator_state, + accelerator_failure_code, + response_bytes, + elapsed_ms, + } = observed; + let present = overview + .rows + .iter() + .map(|row| row.article_number) + .chain( + overview + .defective_rows + .iter() + .filter_map(|row| row.article_number), + ) + .collect::>(); + let missing = missing_ranges(range.start_article, range.end_article, &present); + let digest = format_digest(format); + let receipt_state = if overview.defective_rows.is_empty() { + "complete" + } else { + "partial" + }; + json!({ + "start_article": range.start_article, + "end_article": range.end_article, + "receipt_state": receipt_state, + "failure_code": Value::Null, + "response_bytes": response_bytes, + "elapsed_ms": elapsed_ms, + "returned_row_count": overview.rows.len() + overview.defective_rows.len(), + "valid_row_count": overview.rows.len(), + "defective_row_count": overview.defective_rows.len(), + "missing_ranges": missing, + "unobserved_ranges": Vec::<(u64, u64)>::new(), + "rows": overview.rows.into_iter().map(|row| row_json(row, &digest)).collect::>(), + "defective_rows": overview.defective_rows.into_iter().map(defective_row_json).collect::>(), + "accelerator_state": accelerator_state.as_str(), + "accelerator_failure_code": accelerator_failure_code, + "xpat_match_count": matches.len(), + "xpat_matches": matches.into_iter().map(|matched| json!({ + "article_number": matched.article_num, + "value": matched.value + })).collect::>() + }) +} + +fn failure_receipt(range: &ClearSearchRangeInput, stop: Stop) -> Value { + json!({ + "start_article": range.start_article, + "end_article": range.end_article, + "receipt_state": stop.receipt_state, + "failure_code": stop.failure_code, + "response_bytes": 0, + "elapsed_ms": 0, + "returned_row_count": 0, + "valid_row_count": 0, + "defective_row_count": 0, + "missing_ranges": Vec::<(u64, u64)>::new(), + "unobserved_ranges": [(range.start_article, range.end_article)], + "rows": Vec::::new(), + "defective_rows": Vec::::new(), + "accelerator_state": "unverified", + "accelerator_failure_code": Value::Null, + "xpat_match_count": 0, + "xpat_matches": Vec::::new() + }) +} + +fn append_failure_receipts( + receipts: &mut Vec, + ranges: &[ClearSearchRangeInput], + stop: Stop, +) { + receipts.extend(ranges.iter().map(|range| failure_receipt(range, stop))); +} + +fn failed_response(input: &ClearSearchInput, stop: Stop) -> Value { + let mut receipts = Vec::new(); + append_failure_receipts(&mut receipts, &input.ranges, stop); + response( + input, + None, + None, + Calibration { + state: AcceleratorState::Unverified, + positive_matches: 0, + negative_matches: 0, + }, + 0, + receipts, + false, + ) +} + +fn response( + input: &ClearSearchInput, + group: Option<(&str, u64, u64)>, + format: Option<&OverviewFormat>, + calibration: Calibration, + response_bytes: usize, + ranges: Vec, + connection_reused: bool, +) -> Value { + let execution_state = if ranges.iter().all(|range| { + matches!( + range.get("receipt_state").and_then(Value::as_str), + Some("complete" | "partial") + ) + }) { + "complete" + } else { + "incomplete" + }; + let overview_format = format.map(|format| { + json!({ + "fields_base64": format.fields.iter().map(|field| BASE64.encode(field)).collect::>(), + "sha256": format_digest(format) + }) + }); + json!({ + "status": "complete", + "operation": "clear_search", + "request_id": input.request_id, + "cancellation_id": input.cancellation_id, + "group": input.group, + "group_first_article": group.map(|value| value.1), + "group_last_article": group.map(|value| value.2), + "predicate_sha256": input.predicate_sha256, + "deadline_at_unix_ms": input.deadline_at_unix_ms, + "execution_state": execution_state, + "connection_reused": connection_reused, + "response_bytes": response_bytes, + "overview_format": overview_format, + "accelerator": { + "state": calibration.state.as_str(), + "positive_match_count": calibration.positive_matches, + "negative_match_count": calibration.negative_matches + }, + "ranges": ranges + }) +} + +fn stop_for_error(error: &NntpError) -> Stop { + let receipt_state = match error { + NntpError::Connection(_) + | NntpError::Io(_) + | NntpError::Tls(_) + | NntpError::Timeout(_) + | NntpError::ServiceUnavailable(_) => "transport_failed", + _ => "refused", + }; + Stop { + receipt_state, + failure_code: match error { + NntpError::ResponseTooLarge(_) => "nntp_observation_response_limit_exceeded", + NntpError::UnsupportedCommand(_) => "nntp_operation_unsupported", + NntpError::Protocol(_) => "nntp_overview_unavailable", + _ => nntp_failure(error, "overview_range"), + }, + } +} + +fn elapsed_millis(started: Instant) -> u64 { + started.elapsed().as_millis().try_into().unwrap_or(u64::MAX) +} + +#[cfg(test)] +mod tests { + use super::*; + use nzb_web::nzb_core::nzb_nntp::LosslessOverviewRow; + + #[test] + fn positive_control_comes_from_an_observed_ascii_subject() { + let overview = LosslessOverviewRows { + rows: vec![LosslessOverviewRow { + article_number: 42, + fields: vec![b"Traitors Espana S02E01".to_vec()], + }], + defective_rows: Vec::new(), + }; + assert_eq!( + positive_control(&overview), + Some((42, "Traitors".to_string())) + ); + } + + #[test] + fn aggregate_budget_never_grants_more_than_the_remaining_bytes() { + let mut budget = Budget::new(10); + assert_eq!(budget.command_limit().expect("initial limit"), 10); + budget.consume(7).expect("consume bytes"); + assert_eq!(budget.command_limit().expect("remaining limit"), 3); + budget.consume(3).expect("consume remaining"); + assert!(budget.command_limit().is_err()); + } +} diff --git a/apps/rustnzb/src/group_observation/contract.rs b/apps/rustnzb/src/group_observation/contract.rs new file mode 100644 index 00000000..e79e930f --- /dev/null +++ b/apps/rustnzb/src/group_observation/contract.rs @@ -0,0 +1,354 @@ +use serde::Deserialize; +use sha2::{Digest, Sha256}; +use std::time::{SystemTime, UNIX_EPOCH}; + +const MAX_OBSERVATION_HEADERS: u64 = 10_000; +const MAX_PATTERNS: usize = 16; +const MAX_PATTERN_COMMAND_BYTES: usize = 400; +const MAX_PATTERN_MATCHES: usize = 1_000; +pub(super) const MAX_HEAD_BYTES: usize = 64 * 1024; +pub(super) const MAX_BODY_PREFIX_BYTES: usize = 256 * 1024; +pub(super) const MAX_PAYLOAD_PREFIX_BYTES: usize = 64 * 1024; +const MAX_CLEAR_SEARCH_RESPONSE_BYTES: usize = 64 * 1024 * 1024; +const MAX_CLEAR_SEARCH_RANGES: usize = 8; +const MAX_CLEAR_SEARCH_ARTICLES_PER_RANGE: u64 = 10_000; +const MAX_CLEAR_SEARCH_DURATION_MS: u64 = 120_000; +const MAX_AVAILABILITY_SEGMENTS: usize = 64; + +#[derive(Deserialize)] +#[serde(deny_unknown_fields)] +pub(crate) struct OverviewRangeInput { + pub(crate) request_id: String, + pub(crate) group: String, + pub(crate) start_article: u64, + pub(crate) end_article: u64, + pub(crate) max_headers: u64, +} + +#[derive(Deserialize)] +#[serde(deny_unknown_fields)] +pub(crate) struct ArticleHeadInput { + pub(crate) request_id: String, + pub(crate) group: String, + pub(crate) article_number: u64, + pub(crate) max_header_bytes: usize, +} + +#[derive(Deserialize)] +#[serde(deny_unknown_fields)] +pub(crate) struct ArticleBodyPrefixInput { + pub(crate) request_id: String, + pub(crate) group: String, + pub(crate) message_id: String, + pub(crate) max_wire_bytes: usize, + pub(crate) max_payload_bytes: usize, +} + +#[derive(Clone, Deserialize)] +#[serde(deny_unknown_fields)] +pub(crate) struct ClearSearchRangeInput { + pub(crate) start_article: u64, + pub(crate) end_article: u64, +} + +#[derive(Deserialize)] +#[serde(deny_unknown_fields)] +pub(crate) struct ClearSearchInput { + pub(crate) request_id: String, + pub(crate) cancellation_id: String, + pub(crate) group: String, + pub(crate) ranges: Vec, + pub(crate) patterns: Vec, + pub(crate) predicate_sha256: String, + pub(crate) max_matches_per_range: usize, + pub(crate) max_response_bytes: usize, + pub(crate) deadline_at_unix_ms: u64, +} + +#[derive(Deserialize)] +#[serde(deny_unknown_fields)] +pub(crate) struct ArticleAvailabilityInput { + pub(crate) request_id: String, + pub(crate) message_ids: Vec, + pub(crate) sample_sha256: String, +} + +impl OverviewRangeInput { + pub(super) fn validate(&self) -> Result<(), &'static str> { + validate_observation_identity(&self.request_id, &self.group)?; + let count = range_count( + self.start_article, + self.end_article, + MAX_OBSERVATION_HEADERS, + )?; + if self.max_headers == 0 + || self.max_headers > MAX_OBSERVATION_HEADERS + || count > self.max_headers + { + return Err("overview range exceeds its admitted header bound"); + } + Ok(()) + } +} + +impl ArticleHeadInput { + pub(super) fn validate(&self) -> Result<(), &'static str> { + validate_observation_identity(&self.request_id, &self.group)?; + if self.article_number == 0 + || self.max_header_bytes == 0 + || self.max_header_bytes > MAX_HEAD_BYTES + { + return Err("article head request is outside its admitted bounds"); + } + Ok(()) + } +} + +impl ArticleBodyPrefixInput { + pub(super) fn validate(&self) -> Result<(), &'static str> { + validate_observation_identity(&self.request_id, &self.group)?; + let message_id = self.message_id.trim_matches(['<', '>']); + if message_id.is_empty() + || message_id.len() > 2048 + || !message_id.contains('@') + || !message_id + .bytes() + .all(|byte| byte.is_ascii_graphic() && !matches!(byte, b'/' | b'\\')) + || self.max_wire_bytes == 0 + || self.max_wire_bytes > MAX_BODY_PREFIX_BYTES + || self.max_payload_bytes == 0 + || self.max_payload_bytes > MAX_PAYLOAD_PREFIX_BYTES + || self.max_payload_bytes > self.max_wire_bytes + { + return Err("article body prefix request is outside its admitted bounds"); + } + Ok(()) + } +} + +impl ClearSearchInput { + pub(super) fn validate(&self) -> Result<(), &'static str> { + validate_observation_identity(&self.request_id, &self.group)?; + let now = now_unix_ms()?; + if !valid_control_id(&self.cancellation_id) + || self.ranges.is_empty() + || self.ranges.len() > MAX_CLEAR_SEARCH_RANGES + || self.ranges.iter().any(|range| { + range_count( + range.start_article, + range.end_article, + MAX_CLEAR_SEARCH_ARTICLES_PER_RANGE, + ) + .is_err() + }) + || self + .ranges + .windows(2) + .any(|pair| pair[0].end_article >= pair[1].start_article) + || self.patterns.is_empty() + || self.patterns.len() > MAX_PATTERNS + || self.max_matches_per_range == 0 + || self.max_matches_per_range > MAX_PATTERN_MATCHES + || self.patterns.iter().map(String::len).sum::() + + self.patterns.len().saturating_sub(1) + > MAX_PATTERN_COMMAND_BYTES + || self.patterns.iter().any(|pattern| { + pattern.is_empty() + || pattern.len() > 256 + || pattern + .chars() + .filter(|character| character.is_alphanumeric()) + .take(3) + .count() + < 3 + || pattern + .bytes() + .any(|byte| byte.is_ascii_control() || byte.is_ascii_whitespace()) + }) + || self.predicate_sha256 != clear_search_predicate_digest(&self.patterns) + || self.max_response_bytes == 0 + || self.max_response_bytes > MAX_CLEAR_SEARCH_RESPONSE_BYTES + || self.deadline_at_unix_ms <= now + || self.deadline_at_unix_ms.saturating_sub(now) > MAX_CLEAR_SEARCH_DURATION_MS + { + return Err("clear search request is outside its admitted bounds"); + } + Ok(()) + } +} + +impl ArticleAvailabilityInput { + pub(super) fn validate(&self) -> Result<(), &'static str> { + if !valid_control_id(&self.request_id) + || self.message_ids.is_empty() + || self.message_ids.len() > MAX_AVAILABILITY_SEGMENTS + || self.message_ids.iter().any(|message_id| { + let message_id = message_id.trim_matches(['<', '>']); + message_id.is_empty() + || message_id.len() > 2048 + || !message_id.contains('@') + || !message_id + .bytes() + .all(|byte| byte.is_ascii_graphic() && !matches!(byte, b'/' | b'\\')) + }) + || self + .message_ids + .iter() + .map(|message_id| message_id.trim_matches(['<', '>'])) + .collect::>() + .len() + != self.message_ids.len() + || self.sample_sha256 != article_availability_digest(&self.message_ids) + { + return Err("article availability request is outside its admitted bounds"); + } + Ok(()) + } +} + +pub(super) fn article_availability_digest(message_ids: &[String]) -> String { + let mut digest = Sha256::new(); + digest.update(b"newsgroups-article-availability-sample"); + for message_id in message_ids { + let message_id = message_id.trim_matches(['<', '>']); + digest.update((message_id.len() as u64).to_be_bytes()); + digest.update(message_id.as_bytes()); + } + hex::encode(digest.finalize()) +} + +pub(super) fn clear_search_predicate_digest(patterns: &[String]) -> String { + let mut digest = Sha256::new(); + digest.update(b"newsgroups-clear-search-predicates"); + for pattern in patterns { + digest.update((pattern.len() as u64).to_be_bytes()); + digest.update(pattern.as_bytes()); + } + hex::encode(digest.finalize()) +} + +pub(super) fn now_unix_ms() -> Result { + SystemTime::now() + .duration_since(UNIX_EPOCH) + .map(|duration| duration.as_millis() as u64) + .map_err(|_| "system clock is before the Unix epoch") +} + +fn valid_control_id(value: &str) -> bool { + !value.is_empty() + && value.len() <= 128 + && value + .bytes() + .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'-' | b'_' | b'.')) +} + +fn validate_observation_identity(request_id: &str, group: &str) -> Result<(), &'static str> { + if request_id.is_empty() + || request_id.len() > 128 + || !request_id + .bytes() + .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'-' | b'_' | b'.')) + || group.is_empty() + || group.len() > 255 + || group.split('.').any(|component| { + component.is_empty() + || !component + .bytes() + .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'-' | b'_')) + }) + { + return Err("observation identity is invalid"); + } + Ok(()) +} + +fn range_count(start: u64, end: u64, maximum: u64) -> Result { + if start == 0 || end < start { + return Err("article range is invalid"); + } + let count = end + .checked_sub(start) + .and_then(|delta| delta.checked_add(1)) + .ok_or("article range is invalid")?; + if count > maximum { + return Err("article range exceeds the operation bound"); + } + Ok(count) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn requests_are_exact_and_bounded() { + let overview = OverviewRangeInput { + request_id: "scan-one".to_string(), + group: "esp.binarios.series.misc".to_string(), + start_article: 10, + end_article: 19, + max_headers: 10, + }; + assert!(overview.validate().is_ok()); + + let mut oversized = overview; + oversized.end_article = 10_010; + assert!(oversized.validate().is_err()); + + let mut head = ArticleHeadInput { + request_id: "head-one".to_string(), + group: "esp.binarios.series.misc".to_string(), + article_number: 42, + max_header_bytes: MAX_HEAD_BYTES, + }; + assert!(head.validate().is_ok()); + head.max_header_bytes += 1; + assert!(head.validate().is_err()); + + let mut body = ArticleBodyPrefixInput { + request_id: "body-one".to_string(), + group: "esp.binarios.series.misc".to_string(), + message_id: "one@example.invalid".to_string(), + max_wire_bytes: MAX_BODY_PREFIX_BYTES, + max_payload_bytes: MAX_PAYLOAD_PREFIX_BYTES, + }; + assert!(body.validate().is_ok()); + body.max_payload_bytes = body.max_wire_bytes + 1; + assert!(body.validate().is_err()); + + let mut availability = ArticleAvailabilityInput { + request_id: "availability-one".to_string(), + message_ids: vec!["one@example.invalid".to_string()], + sample_sha256: article_availability_digest(&["one@example.invalid".to_string()]), + }; + assert!(availability.validate().is_ok()); + availability + .message_ids + .push("one@example.invalid".to_string()); + assert!(availability.validate().is_err()); + + let mut clear = ClearSearchInput { + request_id: "clear-one".to_string(), + cancellation_id: "cancel-one".to_string(), + group: "esp.binarios.series.misc".to_string(), + ranges: vec![ + ClearSearchRangeInput { + start_article: 1, + end_article: 10, + }, + ClearSearchRangeInput { + start_article: 20, + end_article: 30, + }, + ], + patterns: vec!["*Traitors*".to_string()], + predicate_sha256: clear_search_predicate_digest(&["*Traitors*".to_string()]), + max_matches_per_range: 100, + max_response_bytes: 1024, + deadline_at_unix_ms: now_unix_ms().expect("clock") + 1_000, + }; + assert!(clear.validate().is_ok()); + clear.ranges[1].start_article = 10; + assert!(clear.validate().is_err()); + } +} diff --git a/apps/rustnzb/src/group_observation/tests.rs b/apps/rustnzb/src/group_observation/tests.rs new file mode 100644 index 00000000..5523d0ce --- /dev/null +++ b/apps/rustnzb/src/group_observation/tests.rs @@ -0,0 +1,366 @@ +use std::{collections::BTreeSet, sync::Arc}; + +use arc_swap::ArcSwap; +use axum::{Json, extract::State}; +use nzb_web::{ + AppState, QueueManager, + auth::{CredentialStore, TokenStore}, + nzb_core::{ + config::AppConfig, + db::Database, + nzb_nntp::{ + ServerConfig, + testutil::{MockConfig, MockNntpServer, test_config}, + }, + }, +}; +use tempfile::TempDir; + +use super::{ + contract::{ + ArticleBodyPrefixInput, ArticleHeadInput, ClearSearchInput, ClearSearchRangeInput, + OverviewRangeInput, clear_search_predicate_digest, now_unix_ms, + }, + h_article_body_prefix, h_article_head, h_clear_search, h_overview_range, missing_ranges, +}; + +fn state_without_provider() -> (Arc, TempDir) { + state_with_servers(Vec::new()) +} + +fn state_with_servers(servers: Vec) -> (Arc, TempDir) { + let config = AppConfig { + servers, + ..AppConfig::default() + }; + let database = Database::open_memory().expect("open in-memory database"); + let temporary = TempDir::new().expect("create temporary directory"); + let incomplete = temporary.path().join("incomplete"); + let complete = temporary.path().join("complete"); + std::fs::create_dir_all(&incomplete).expect("create incomplete directory"); + std::fs::create_dir_all(&complete).expect("create complete directory"); + let log = nzb_web::LogBuffer::new(); + let queue = QueueManager::new( + config.servers.clone(), + database, + incomplete, + complete, + log.clone(), + config.general.max_active_downloads, + config.categories.clone(), + config.general.min_free_space_bytes, + config.general.speed_limit_bps, + false, + config.general.max_nested_archive_depth, + config.general.abort_hopeless, + config.general.early_failure_check, + config.general.required_completion_pct, + config.general.article_timeout_secs, + ); + let state = Arc::new(AppState::new( + Arc::new(ArcSwap::from_pointee(config)), + temporary.path().join("config.toml"), + queue, + log, + Arc::new(TokenStore::new()), + Arc::new(CredentialStore::new(temporary.path().to_path_buf())), + )); + (state, temporary) +} + +#[tokio::test] +async fn unsupported_xpat_yields_complete_lossless_overview_on_the_same_connection() { + let group = "esp.binarios.series.misc"; + let mut groups = std::collections::HashMap::new(); + groups.insert(group.to_string(), (2, 1, 2)); + let server = MockNntpServer::start(MockConfig { + groups, + xpat_unsupported: true, + xover_entries: vec![ + "1\tTraitors Espana S02E01\tposter\tWed, 07 May 2025 20:50:00 +0000\t<1@test>\t\t100\t1".into(), + "2\tUnrelated\tposter\tWed, 07 May 2025 20:51:00 +0000\t<2@test>\t\t100\t1".into(), + ], + ..MockConfig::default() + }) + .await; + let (state, _temporary) = state_with_servers(vec![test_config(server.port())]); + let Json(response) = h_clear_search( + State(state), + Json(ClearSearchInput { + request_id: "clear-unsupported".to_string(), + cancellation_id: "cancel-unsupported".to_string(), + group: group.to_string(), + ranges: vec![ClearSearchRangeInput { + start_article: 1, + end_article: 2, + }], + patterns: vec!["*Traitors*".to_string()], + predicate_sha256: clear_search_predicate_digest(&["*Traitors*".to_string()]), + max_matches_per_range: 100, + max_response_bytes: 1024 * 1024, + deadline_at_unix_ms: now_unix_ms().expect("clock") + 10_000, + }), + ) + .await + .expect("clear search response"); + assert_eq!(response["status"], "complete"); + assert_eq!(response["execution_state"], "complete"); + assert_eq!(response["connection_reused"], true); + assert_eq!(response["accelerator"]["state"], "unsupported"); + assert_eq!(response["ranges"][0]["valid_row_count"], 2); + assert_eq!(response["ranges"][0]["xpat_matches"], serde_json::json!([])); +} + +#[tokio::test] +async fn bounded_multi_range_search_returns_one_exact_receipt_per_range() { + let group = "esp.binarios.series.misc"; + let mut groups = std::collections::HashMap::new(); + groups.insert(group.to_string(), (4, 1, 4)); + let mut xover_entries_by_range = std::collections::HashMap::new(); + xover_entries_by_range.insert( + "1-2".to_string(), + vec![ + "1\tTraitors Espana S02E01\tposter\tWed, 07 May 2025 20:50:00 +0000\t<1@test>\t\t100\t1".into(), + "2\tUnrelated\tposter\tWed, 07 May 2025 20:51:00 +0000\t<2@test>\t\t100\t1".into(), + ], + ); + xover_entries_by_range.insert( + "3-4".to_string(), + vec![ + "3\tTraitors Espana S02E01 repost\tposter\tWed, 07 May 2025 20:52:00 +0000\t<3@test>\t\t100\t1".into(), + "4\tUnrelated again\tposter\tWed, 07 May 2025 20:53:00 +0000\t<4@test>\t\t100\t1".into(), + ], + ); + let mut xpat_entries_by_request = std::collections::HashMap::new(); + xpat_entries_by_request.insert( + "1-2 *Traitors*".to_string(), + vec!["1 Traitors Espana S02E01".to_string()], + ); + xpat_entries_by_request.insert( + "3-4 *Traitors*".to_string(), + vec!["3 Traitors Espana S02E01 repost".to_string()], + ); + let server = MockNntpServer::start(MockConfig { + groups, + xover_entries_by_range, + xpat_entries_by_request, + ..MockConfig::default() + }) + .await; + let (state, _temporary) = state_with_servers(vec![test_config(server.port())]); + let Json(response) = h_clear_search( + State(state), + Json(ClearSearchInput { + request_id: "clear-multi".to_string(), + cancellation_id: "cancel-multi".to_string(), + group: group.to_string(), + ranges: vec![ + ClearSearchRangeInput { + start_article: 1, + end_article: 2, + }, + ClearSearchRangeInput { + start_article: 3, + end_article: 4, + }, + ], + patterns: vec!["*Traitors*".to_string()], + predicate_sha256: clear_search_predicate_digest(&["*Traitors*".to_string()]), + max_matches_per_range: 100, + max_response_bytes: 1024 * 1024, + deadline_at_unix_ms: now_unix_ms().expect("clock") + 10_000, + }), + ) + .await + .expect("clear search response"); + assert_eq!(response["execution_state"], "complete"); + assert_eq!(response["accelerator"]["state"], "supported"); + assert_eq!(response["ranges"].as_array().map(Vec::len), Some(2)); + assert_eq!(response["ranges"][0]["receipt_state"], "complete"); + assert_eq!(response["ranges"][1]["receipt_state"], "complete"); + assert_eq!(response["ranges"][0]["xpat_match_count"], 1); + assert_eq!(response["ranges"][1]["xpat_match_count"], 1); +} + +#[tokio::test] +async fn deadline_cancellation_receipts_cover_every_admitted_range() { + let group = "esp.binarios.series.misc"; + let mut groups = std::collections::HashMap::new(); + groups.insert(group.to_string(), (4, 1, 4)); + let server = MockNntpServer::start(MockConfig { + groups, + response_delay: Some(std::time::Duration::from_millis(100)), + ..MockConfig::default() + }) + .await; + let (state, _temporary) = state_with_servers(vec![test_config(server.port())]); + let Json(response) = h_clear_search( + State(state), + Json(ClearSearchInput { + request_id: "clear-deadline".to_string(), + cancellation_id: "cancel-deadline".to_string(), + group: group.to_string(), + ranges: vec![ + ClearSearchRangeInput { + start_article: 1, + end_article: 2, + }, + ClearSearchRangeInput { + start_article: 3, + end_article: 4, + }, + ], + patterns: vec!["*Traitors*".to_string()], + predicate_sha256: clear_search_predicate_digest(&["*Traitors*".to_string()]), + max_matches_per_range: 100, + max_response_bytes: 1024 * 1024, + deadline_at_unix_ms: now_unix_ms().expect("clock") + 20, + }), + ) + .await + .expect("clear search response"); + assert_eq!(response["execution_state"], "incomplete"); + assert_eq!(response["ranges"].as_array().map(Vec::len), Some(2)); + assert!(response["ranges"].as_array().is_some_and(|ranges| { + ranges.iter().all(|range| { + range["receipt_state"] == "cancelled" + && range["failure_code"] == "nntp_operation_timed_out" + }) + })); +} + +#[tokio::test] +async fn missing_provider_is_a_typed_blocker_for_every_observation() { + let (state, _temporary) = state_without_provider(); + let Json(overview) = h_overview_range( + State(Arc::clone(&state)), + Json(OverviewRangeInput { + request_id: "overview-one".to_string(), + group: "esp.binarios.series.misc".to_string(), + start_article: 1, + end_article: 10, + max_headers: 10, + }), + ) + .await + .expect("overview response"); + assert_eq!(overview["status"], "blocked"); + assert_eq!(overview["failure_code"], "nntp_provider_not_configured"); + assert_eq!(overview["request_id"], "overview-one"); + + let Json(clear_search) = h_clear_search( + State(Arc::clone(&state)), + Json(ClearSearchInput { + request_id: "clear-one".to_string(), + cancellation_id: "cancel-one".to_string(), + group: "esp.binarios.series.misc".to_string(), + ranges: vec![ClearSearchRangeInput { + start_article: 1, + end_article: 10, + }], + patterns: vec!["*Traitors*".to_string()], + predicate_sha256: clear_search_predicate_digest(&["*Traitors*".to_string()]), + max_matches_per_range: 100, + max_response_bytes: 1024 * 1024, + deadline_at_unix_ms: now_unix_ms().expect("clock") + 10_000, + }), + ) + .await + .expect("clear search response"); + assert_eq!(clear_search["status"], "complete"); + assert_eq!(clear_search["execution_state"], "incomplete"); + assert_eq!(clear_search["ranges"][0]["receipt_state"], "refused"); + assert_eq!( + clear_search["ranges"][0]["failure_code"], + "nntp_provider_not_configured" + ); + assert_eq!(clear_search["request_id"], "clear-one"); + + let Json(head) = h_article_head( + State(Arc::clone(&state)), + Json(ArticleHeadInput { + request_id: "head-one".to_string(), + group: "esp.binarios.series.misc".to_string(), + article_number: 42, + max_header_bytes: 64 * 1024, + }), + ) + .await + .expect("head response"); + assert_eq!(head["status"], "blocked"); + assert_eq!(head["failure_code"], "nntp_provider_not_configured"); + assert_eq!(head["request_id"], "head-one"); + + let Json(body) = h_article_body_prefix( + State(state), + Json(ArticleBodyPrefixInput { + request_id: "body-one".to_string(), + group: "esp.binarios.series.misc".to_string(), + message_id: "one@example.invalid".to_string(), + max_wire_bytes: 64 * 1024, + max_payload_bytes: 32 * 1024, + }), + ) + .await + .expect("body response"); + assert_eq!(body["status"], "blocked"); + assert_eq!(body["failure_code"], "nntp_provider_not_configured"); + assert_eq!(body["request_id"], "body-one"); +} + +#[tokio::test] +async fn body_prefix_returns_digest_bound_decoded_binary_magic() { + use base64::{Engine as _, engine::general_purpose::STANDARD as BASE64}; + + let mut wire = b"=ybegin line=128 size=8 name=random.bin\r\n".to_vec(); + for byte in b"PAR2\0PKT" { + let encoded = byte.wrapping_add(42); + if matches!(encoded, 0 | b'\n' | b'\r' | b'=') { + wire.push(b'='); + wire.push(encoded.wrapping_add(64)); + } else { + wire.push(encoded); + } + } + wire.extend_from_slice(b"\r\n=yend size=8\r\n"); + let mut articles = std::collections::HashMap::new(); + articles.insert("one@example.invalid".to_string(), wire); + let group = "esp.binarios.series.misc"; + let mut groups = std::collections::HashMap::new(); + groups.insert(group.to_string(), (1, 1, 1)); + let server = MockNntpServer::start(MockConfig { + groups, + articles, + ..MockConfig::default() + }) + .await; + let (state, _temporary) = state_with_servers(vec![test_config(server.port())]); + let Json(body) = h_article_body_prefix( + State(state), + Json(ArticleBodyPrefixInput { + request_id: "body-magic".to_string(), + group: group.to_string(), + message_id: "one@example.invalid".to_string(), + max_wire_bytes: 64 * 1024, + max_payload_bytes: 32 * 1024, + }), + ) + .await + .expect("body prefix"); + assert_eq!(body["status"], "complete"); + assert_eq!(body["payload_encoding"], "yenc"); + assert_eq!(body["body_complete"], true); + assert_eq!(body["payload_complete"], true); + assert_eq!( + BASE64 + .decode(body["payload_prefix_base64"].as_str().unwrap()) + .unwrap(), + b"PAR2\0PKT" + ); +} + +#[test] +fn missing_articles_are_compact_ranges() { + let present = BTreeSet::from([2, 3, 6, 8]); + assert_eq!(missing_ranges(1, 8, &present), [(1, 1), (4, 5), (7, 7)]); +} diff --git a/apps/rustnzb/src/lib.rs b/apps/rustnzb/src/lib.rs index a6905f14..9542baa0 100644 --- a/apps/rustnzb/src/lib.rs +++ b/apps/rustnzb/src/lib.rs @@ -1,4 +1,5 @@ pub mod group_handlers; +pub mod group_observation; pub mod handlers; pub mod server; diff --git a/apps/rustnzb/src/server.rs b/apps/rustnzb/src/server.rs index b6c6b78f..ec0abd95 100644 --- a/apps/rustnzb/src/server.rs +++ b/apps/rustnzb/src/server.rs @@ -18,6 +18,7 @@ use utoipa::OpenApi; use utoipa_swagger_ui::SwaggerUi; use crate::group_handlers; +use crate::group_observation; use crate::handlers; use nzb_web::auth; use nzb_web::error::ApiError; @@ -249,6 +250,26 @@ pub fn build_router(state: Arc) -> Router { // Newsgroup browsing .route("/groups", get(group_handlers::h_group_list)) .route("/groups/refresh", post(group_handlers::h_group_refresh)) + .route( + "/groups/overview-range", + post(group_observation::h_overview_range), + ) + .route( + "/groups/article-head", + post(group_observation::h_article_head), + ) + .route( + "/groups/article-body-prefix", + post(group_observation::h_article_body_prefix), + ) + .route( + "/groups/article-availability", + post(group_observation::h_article_availability), + ) + .route( + "/groups/clear-search", + post(group_observation::h_clear_search), + ) .route("/groups/{id}", get(group_handlers::h_group_get)) .route("/groups/{id}/status", get(group_handlers::h_group_status)) .route( diff --git a/crates/nzb-core/Cargo.toml b/crates/nzb-core/Cargo.toml index c127be5b..41f52697 100644 --- a/crates/nzb-core/Cargo.toml +++ b/crates/nzb-core/Cargo.toml @@ -12,7 +12,7 @@ default = [] groups-db = [] [dependencies] -nzb-nntp = { version = "0.2.24", path = "../nzb-nntp" } +nzb-nntp = { version = "0.2.26", path = "../nzb-nntp" } serde = { version = "1", features = ["derive"] } serde_json = "1" toml = "1.1" diff --git a/crates/nzb-dispatch/src/article_failure.rs b/crates/nzb-dispatch/src/article_failure.rs index 230d1361..22407b78 100644 --- a/crates/nzb-dispatch/src/article_failure.rs +++ b/crates/nzb-dispatch/src/article_failure.rs @@ -106,7 +106,9 @@ impl ArticleFailure { NntpError::Connection(_) => ArticleFailureKind::ConnectionClosed, NntpError::Io(_) => ArticleFailureKind::ConnectionClosed, NntpError::Timeout(_) => ArticleFailureKind::Timeout, - NntpError::Protocol(_) => ArticleFailureKind::Protocol, + NntpError::Protocol(_) + | NntpError::UnsupportedCommand(_) + | NntpError::ResponseTooLarge(_) => ArticleFailureKind::Protocol, NntpError::NoSuchGroup(_) | NntpError::NoArticleSelected(_) => { ArticleFailureKind::Protocol } diff --git a/crates/nzb-news/Cargo.toml b/crates/nzb-news/Cargo.toml index da4adf9e..a49a26a9 100644 --- a/crates/nzb-news/Cargo.toml +++ b/crates/nzb-news/Cargo.toml @@ -8,13 +8,13 @@ repository = "https://github.com/TheDancingDeveloper-org/nzb-news" readme = "README.md" [dependencies] -nzb-nntp = { version = "0.2.24", path = "../nzb-nntp" } +nzb-nntp = { version = "0.2.26", path = "../nzb-nntp" } tokio = { version = "1", features = ["sync", "rt", "time", "macros"] } tracing = "0.1" [dev-dependencies] -nzb-nntp = { version = "0.2.24", path = "../nzb-nntp", features = ["test-support"] } +nzb-nntp = { version = "0.2.26", path = "../nzb-nntp", features = ["test-support"] } tokio = { version = "1", features = ["full", "test-util"] } tracing-subscriber = { version = "0.3", features = ["env-filter"] } diff --git a/crates/nzb-news/src/downloader.rs b/crates/nzb-news/src/downloader.rs index c67bc056..6dc68bd3 100644 --- a/crates/nzb-news/src/downloader.rs +++ b/crates/nzb-news/src/downloader.rs @@ -1840,6 +1840,8 @@ fn clone_err(err: &NntpError) -> NntpError { NntpError::NoSuchGroup(m) => NntpError::NoSuchGroup(m.clone()), NntpError::NoArticleSelected(m) => NntpError::NoArticleSelected(m.clone()), NntpError::Protocol(m) => NntpError::Protocol(m.clone()), + NntpError::UnsupportedCommand(m) => NntpError::UnsupportedCommand(m.clone()), + NntpError::ResponseTooLarge(m) => NntpError::ResponseTooLarge(m.clone()), NntpError::Io(e) => NntpError::Io(std::io::Error::new(e.kind(), e.to_string())), NntpError::NoConnectionsAvailable(m) => NntpError::NoConnectionsAvailable(m.clone()), NntpError::Timeout(m) => NntpError::Timeout(m.clone()), diff --git a/crates/nzb-news/src/news_wrapper.rs b/crates/nzb-news/src/news_wrapper.rs index 81d7654d..c27b7b95 100644 --- a/crates/nzb-news/src/news_wrapper.rs +++ b/crates/nzb-news/src/news_wrapper.rs @@ -271,6 +271,8 @@ impl NewsWrapper { | NntpError::PermissionDenied(_) | NntpError::ServiceUnavailable(_) | NntpError::Protocol(_) + | NntpError::UnsupportedCommand(_) + | NntpError::ResponseTooLarge(_) | NntpError::Timeout(_) => { self.bump_bad_cons(); self.hard_reset().await; diff --git a/crates/nzb-news/src/penalty.rs b/crates/nzb-news/src/penalty.rs index bc467909..acf5eae0 100644 --- a/crates/nzb-news/src/penalty.rs +++ b/crates/nzb-news/src/penalty.rs @@ -78,7 +78,9 @@ pub fn penalty_for_error(err: &NntpError) -> PenaltyAction { // Timeouts and unexpected protocol responses: medium cooldown // after bad_cons threshold. A single slow article shouldn't // penalise the whole server, but a run of timeouts should. - Timeout(_) | Protocol(_) => PenaltyAction::BadCons(PENALTY_TIMEOUT), + Timeout(_) | Protocol(_) | UnsupportedCommand(_) | ResponseTooLarge(_) => { + PenaltyAction::BadCons(PENALTY_TIMEOUT) + } // Pool/orchestrator-side errors — shouldn't flow through this // path, but if they do, take the conservative default penalty. diff --git a/crates/nzb-nntp/Cargo.toml b/crates/nzb-nntp/Cargo.toml index 73432684..e9bad3a3 100644 --- a/crates/nzb-nntp/Cargo.toml +++ b/crates/nzb-nntp/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "nzb-nntp" -version = "0.2.24" +version = "0.2.26" edition = "2024" description = "Async NNTP client with TLS, pipelining, connection pooling, and multi-server support" license = "MIT" diff --git a/crates/nzb-nntp/src/connection.rs b/crates/nzb-nntp/src/connection.rs index eb5d0627..a45be2d8 100644 --- a/crates/nzb-nntp/src/connection.rs +++ b/crates/nzb-nntp/src/connection.rs @@ -39,6 +39,9 @@ use crate::capabilities::NntpCapabilities; use crate::config::{ListActiveEntry, ServerConfig}; use crate::error::{NntpError, NntpResult}; +use crate::overview::{ + LosslessOverviewRows, OverviewFormat, parse_lossless_overview_rows, parse_overview_format, +}; // --------------------------------------------------------------------------- // Response @@ -55,6 +58,12 @@ pub struct NntpResponse { pub data: Option>, } +#[derive(Debug, Clone, Eq, PartialEq)] +pub struct BodyPrefixResponse { + pub data: Vec, + pub complete: bool, +} + impl NntpResponse { /// Returns `true` if the response indicates success (2xx). pub fn is_success(&self) -> bool { @@ -118,6 +127,12 @@ pub struct HeaderEntry { pub value: String, } +#[derive(Debug, Clone)] +pub struct BoundedResponse { + pub value: T, + pub response_bytes: usize, +} + // --------------------------------------------------------------------------- // Article range for XHDR / XPAT queries // --------------------------------------------------------------------------- @@ -928,6 +943,34 @@ impl NntpConnection { } } + async fn read_multiline_body_maybe_decompress_bounded( + &mut self, + max_bytes: usize, + ) -> NntpResult> { + let raw = self.read_multiline_body_bounded(max_bytes).await?; + if self.compress_enabled && raw.len() >= 2 && raw[0] == 0x1f && raw[1] == 0x8b { + use flate2::read::GzDecoder; + use std::io::Read; + + let decoder = GzDecoder::new(&raw[..]); + let mut decompressed = Vec::with_capacity(max_bytes.min(raw.len() * 4)); + decoder + .take(max_bytes as u64 + 1) + .read_to_end(&mut decompressed) + .map_err(|error| { + NntpError::Protocol(format!("gzip HEAD decode failed: {error}")) + })?; + if decompressed.len() > max_bytes { + return Err(NntpError::ResponseTooLarge( + "decompressed multi-line response exceeds its byte bound".into(), + )); + } + Ok(decompressed) + } else { + Ok(raw) + } + } + // ------------------------------------------------------------------ // ARTICLE command // ------------------------------------------------------------------ @@ -1038,6 +1081,86 @@ impl NntpConnection { } } + /// Fetch exact headers for one article number in the selected group. + pub async fn fetch_head_number( + &mut self, + article_number: u64, + max_header_bytes: usize, + ) -> NntpResult { + if !self.capabilities.have_head { + return Err(NntpError::Protocol("Server does not support HEAD".into())); + } + if article_number == 0 || max_header_bytes == 0 || max_header_bytes > 64 * 1024 { + return Err(NntpError::Protocol("HEAD request bound is invalid".into())); + } + if self.state != ConnectionState::Ready { + return Err(NntpError::Protocol(format!( + "Cannot HEAD in state {:?}", + self.state + ))); + } + self.state = ConnectionState::Busy; + self.send_command(&format!("HEAD {article_number}")) + .await + .inspect_err(|_| self.state = ConnectionState::Error)?; + let status = self + .read_response_line() + .await + .inspect_err(|_| self.state = ConnectionState::Error)?; + match status.code { + 221 => { + let data = self + .read_multiline_body_maybe_decompress_bounded(max_header_bytes) + .await + .inspect_err(|_| self.state = ConnectionState::Error)?; + self.state = ConnectionState::Ready; + Ok(NntpResponse { + code: status.code, + message: status.message, + data: Some(data), + }) + } + 423 => { + self.state = ConnectionState::Ready; + Err(NntpError::ArticleNotFound(article_number.to_string())) + } + 411 => { + self.state = ConnectionState::Ready; + Err(NntpError::NoSuchGroup(status.message)) + } + 412 | 420 => { + self.state = ConnectionState::Ready; + Err(NntpError::NoArticleSelected(status.message)) + } + 403 => { + self.state = ConnectionState::Error; + Err(NntpError::PermissionDenied(status.message)) + } + 480 => { + self.state = ConnectionState::Error; + Err(NntpError::AuthRequired(status.message)) + } + 481 | 482 => { + self.state = ConnectionState::Error; + Err(NntpError::Auth(format!( + "HEAD rejected ({}): {}", + status.code, status.message + ))) + } + 502 => { + self.state = ConnectionState::Error; + Err(NntpError::ServiceUnavailable(status.message)) + } + _ => { + self.state = ConnectionState::Error; + Err(NntpError::Protocol(format!( + "Unexpected HEAD response {}: {}", + status.code, status.message + ))) + } + } + } + // ------------------------------------------------------------------ // STAT command (pre-check) // ------------------------------------------------------------------ @@ -1172,6 +1295,156 @@ impl NntpConnection { // XOVER command (RFC 2980 Section 2.8) // ------------------------------------------------------------------ + pub async fn overview_format(&mut self) -> NntpResult { + if self.state != ConnectionState::Ready { + return Err(NntpError::Protocol(format!( + "Cannot LIST OVERVIEW.FMT in state {:?}", + self.state + ))); + } + self.state = ConnectionState::Busy; + self.send_command("LIST OVERVIEW.FMT") + .await + .inspect_err(|_| self.state = ConnectionState::Error)?; + let status = self + .read_response_line() + .await + .inspect_err(|_| self.state = ConnectionState::Error)?; + match status.code { + 215 => { + let data = self + .read_multiline_body_maybe_decompress() + .await + .inspect_err(|_| self.state = ConnectionState::Error)?; + self.state = ConnectionState::Ready; + parse_overview_format(&data) + } + 480 => { + self.state = ConnectionState::Error; + Err(NntpError::AuthRequired(status.message)) + } + 481 | 482 => { + self.state = ConnectionState::Error; + Err(NntpError::Auth(format!( + "LIST OVERVIEW.FMT rejected ({}): {}", + status.code, status.message + ))) + } + 502 => { + self.state = ConnectionState::Error; + Err(NntpError::ServiceUnavailable(status.message)) + } + _ => { + self.state = ConnectionState::Error; + Err(NntpError::Protocol(format!( + "Unexpected LIST OVERVIEW.FMT response {}: {}", + status.code, status.message + ))) + } + } + } + + pub async fn xover_lossless( + &mut self, + start: u64, + end: u64, + format: &OverviewFormat, + ) -> NntpResult { + self.xover_lossless_with_limit(start, end, format, None) + .await + .map(|response| response.value) + } + + pub async fn xover_lossless_bounded( + &mut self, + start: u64, + end: u64, + format: &OverviewFormat, + max_response_bytes: usize, + ) -> NntpResult> { + if max_response_bytes == 0 || max_response_bytes > 8 * 1024 * 1024 { + return Err(NntpError::Protocol( + "XOVER response bound is invalid".into(), + )); + } + self.xover_lossless_with_limit(start, end, format, Some(max_response_bytes)) + .await + } + + async fn xover_lossless_with_limit( + &mut self, + start: u64, + end: u64, + format: &OverviewFormat, + max_response_bytes: Option, + ) -> NntpResult> { + if self.state != ConnectionState::Ready { + return Err(NntpError::Protocol(format!( + "Cannot XOVER in state {:?}", + self.state + ))); + } + self.state = ConnectionState::Busy; + self.send_command(&format!("XOVER {start}-{end}")) + .await + .inspect_err(|_| self.state = ConnectionState::Error)?; + let status = self + .read_response_line() + .await + .inspect_err(|_| self.state = ConnectionState::Error)?; + match status.code { + 224 => { + let data = match max_response_bytes { + Some(maximum) => { + self.read_multiline_body_maybe_decompress_bounded(maximum) + .await + } + None => self.read_multiline_body_maybe_decompress().await, + } + .inspect_err(|_| self.state = ConnectionState::Error)?; + self.state = ConnectionState::Ready; + Ok(BoundedResponse { + response_bytes: data.len(), + value: parse_lossless_overview_rows(&data, format, start, end), + }) + } + 420 => { + self.state = ConnectionState::Ready; + Ok(BoundedResponse { + value: LosslessOverviewRows { + rows: Vec::new(), + defective_rows: Vec::new(), + }, + response_bytes: 0, + }) + } + 412 => { + self.state = ConnectionState::Ready; + Err(NntpError::NoSuchGroup( + "No newsgroup selected (send GROUP first)".into(), + )) + } + 481 | 482 => { + self.state = ConnectionState::Error; + Err(NntpError::Auth(format!( + "XOVER rejected ({}): {}", + status.code, status.message + ))) + } + 502 => { + self.state = ConnectionState::Error; + Err(NntpError::ServiceUnavailable(status.message)) + } + _ => { + self.state = ConnectionState::Error; + Err(NntpError::Protocol(format!( + "Unexpected XOVER response {}: {}", + status.code, status.message + ))) + } + } + } + /// Fetch overview data for a range of article numbers. /// /// Sends `XOVER start-end` and parses the tab-delimited multi-line response. @@ -1333,6 +1606,32 @@ impl NntpConnection { range: ArticleRange, patterns: &[&str], ) -> NntpResult> { + self.xpat_with_limit(header, range, patterns, None) + .await + .map(|response| response.value) + } + + pub async fn xpat_bounded( + &mut self, + header: &str, + range: ArticleRange, + patterns: &[&str], + max_response_bytes: usize, + ) -> NntpResult>> { + if max_response_bytes == 0 || max_response_bytes > 8 * 1024 * 1024 { + return Err(NntpError::Protocol("XPAT response bound is invalid".into())); + } + self.xpat_with_limit(header, range, patterns, Some(max_response_bytes)) + .await + } + + async fn xpat_with_limit( + &mut self, + header: &str, + range: ArticleRange, + patterns: &[&str], + max_response_bytes: Option, + ) -> NntpResult>> { if self.state != ConnectionState::Ready { return Err(NntpError::Protocol(format!( "Cannot XPAT in state {:?}", @@ -1358,16 +1657,26 @@ impl NntpConnection { match status.code { 221 => { - let data = self - .read_multiline_body_maybe_decompress() - .await - .inspect_err(|_| self.state = ConnectionState::Error)?; + let data = match max_response_bytes { + Some(maximum) => { + self.read_multiline_body_maybe_decompress_bounded(maximum) + .await + } + None => self.read_multiline_body_maybe_decompress().await, + } + .inspect_err(|_| self.state = ConnectionState::Error)?; self.state = ConnectionState::Ready; - Ok(parse_header_data(&data)) + Ok(BoundedResponse { + response_bytes: data.len(), + value: parse_header_data(&data), + }) } 420 => { self.state = ConnectionState::Ready; - Ok(Vec::new()) // No articles matched + Ok(BoundedResponse { + value: Vec::new(), + response_bytes: 0, + }) } 412 => { self.state = ConnectionState::Ready; @@ -1390,6 +1699,13 @@ impl NntpConnection { self.state = ConnectionState::Error; Err(NntpError::ServiceUnavailable(status.message)) } + 500 | 501 => { + self.state = ConnectionState::Ready; + Err(NntpError::UnsupportedCommand(format!( + "XPAT unsupported ({}): {}", + status.code, status.message + ))) + } _ => { self.state = ConnectionState::Error; Err(NntpError::Protocol(format!( @@ -1491,6 +1807,97 @@ impl NntpConnection { } } + pub async fn fetch_body_prefix( + &mut self, + message_id: &str, + max_bytes: usize, + ) -> NntpResult { + if max_bytes == 0 || max_bytes > 256 * 1024 { + return Err(NntpError::Protocol( + "BODY prefix request bound is invalid".into(), + )); + } + if self.compress_enabled { + return Err(NntpError::UnsupportedCommand( + "BODY prefix requires an uncompressed observation connection".into(), + )); + } + if !self.capabilities.have_body { + return Err(NntpError::UnsupportedCommand( + "Server does not support BODY prefix observation".into(), + )); + } + if self.state != ConnectionState::Ready { + return Err(NntpError::Protocol(format!( + "Cannot BODY prefix in state {:?}", + self.state + ))); + } + self.state = ConnectionState::Busy; + let mid = normalize_message_id(message_id); + self.send_command(&format!("BODY {mid}")) + .await + .inspect_err(|_| self.state = ConnectionState::Error)?; + let status = self + .read_response_line() + .await + .inspect_err(|_| self.state = ConnectionState::Error)?; + match status.code { + 222 => { + let prefix = self + .read_multiline_body_prefix(max_bytes) + .await + .inspect_err(|_| self.state = ConnectionState::Error)?; + if prefix.complete { + self.state = ConnectionState::Ready; + } else { + self.transport.take(); + self.compress_enabled = false; + self.state = ConnectionState::Disconnected; + } + Ok(prefix) + } + 430 => { + self.state = ConnectionState::Ready; + Err(NntpError::ArticleNotFound(mid)) + } + 412 | 420 => { + self.state = ConnectionState::Ready; + Err(NntpError::NoArticleSelected(status.message)) + } + 403 => { + self.state = ConnectionState::Error; + Err(NntpError::PermissionDenied(status.message)) + } + 480 => { + self.state = ConnectionState::Error; + Err(NntpError::AuthRequired(status.message)) + } + 481 | 482 => { + self.state = ConnectionState::Error; + Err(NntpError::Auth(format!( + "BODY prefix rejected ({}): {}", + status.code, status.message + ))) + } + 500 | 501 => { + self.state = ConnectionState::Ready; + Err(NntpError::UnsupportedCommand(status.message)) + } + 502 => { + self.state = ConnectionState::Error; + Err(NntpError::ServiceUnavailable(status.message)) + } + _ => { + self.state = ConnectionState::Error; + Err(NntpError::Protocol(format!( + "Unexpected BODY prefix response {}: {}", + status.code, status.message + ))) + } + } + } + // ------------------------------------------------------------------ // LIST ACTIVE (RFC 3977 Section 7.6.3) // ------------------------------------------------------------------ @@ -1852,10 +2259,67 @@ impl NntpConnection { /// Public for pipeline use. pub(crate) async fn read_multiline_body(&mut self) -> NntpResult> { let mut body = self.checkout_body_buffer(); - self.read_multiline_body_into(&mut body).await?; + self.read_multiline_body_into_bounded(&mut body, None) + .await?; + Ok(body) + } + + async fn read_multiline_body_bounded(&mut self, max_bytes: usize) -> NntpResult> { + let mut body = Vec::with_capacity(max_bytes.min(16 * 1024)); + self.read_multiline_body_into_bounded(&mut body, Some(max_bytes)) + .await?; Ok(body) } + async fn read_multiline_body_prefix( + &mut self, + max_bytes: usize, + ) -> NntpResult { + let mut data = Vec::with_capacity(max_bytes.min(16 * 1024)); + let heartbeat = self.io_heartbeat.clone(); + let transport = self + .transport + .as_mut() + .ok_or(NntpError::Connection("Not connected".into()))?; + let complete = loop { + self.line_scratch.clear(); + let count = tokio::time::timeout( + READ_BODY_LINE_TIMEOUT, + transport.read_line_bytes(&mut self.line_scratch), + ) + .await + .map_err(|_| NntpError::Timeout("BODY prefix line timed out".into()))? + .map_err(NntpError::Io)?; + if count == 0 { + return Err(NntpError::Connection( + "Server closed connection during BODY prefix".into(), + )); + } + if let Some(heartbeat) = &heartbeat { + heartbeat.tick(); + } + if self.line_scratch == b".\r\n" || self.line_scratch == b".\n" { + break true; + } + if self.line_scratch.len() > 16 * 1024 { + return Err(NntpError::ResponseTooLarge( + "BODY prefix line exceeds its byte bound".into(), + )); + } + let line = if self.line_scratch.starts_with(b"..") { + &self.line_scratch[1..] + } else { + &self.line_scratch + }; + let remaining = max_bytes.saturating_sub(data.len()); + data.extend_from_slice(&line[..line.len().min(remaining)]); + if line.len() > remaining || data.len() == max_bytes { + break false; + } + }; + Ok(BodyPrefixResponse { data, complete }) + } + /// Same as [`Self::read_multiline_body`], but fills a caller-owned /// buffer (cleared first) instead of allocating. `out` is expected to /// have been checked out via [`Self::checkout_body_buffer`] so its @@ -1863,7 +2327,11 @@ impl NntpConnection { /// checkout is what actually avoids the page-fault/allocation cost per /// article; passing an arbitrary fresh `Vec::new()` here still works /// correctly, it just forgoes the reuse benefit. - async fn read_multiline_body_into(&mut self, out: &mut Vec) -> NntpResult<()> { + async fn read_multiline_body_into_bounded( + &mut self, + out: &mut Vec, + max_bytes: Option, + ) -> NntpResult<()> { out.clear(); // Clone the heartbeat ref before we take the mutable borrow on // `transport`, so we can tick it inside the loop body. Cheap: just @@ -1918,11 +2386,21 @@ impl NntpConnection { } // Dot-unstuffing: if a line starts with "..", remove the first dot - if self.line_scratch.starts_with(b"..") { - out.extend_from_slice(&self.line_scratch[1..]); + let line = if self.line_scratch.starts_with(b"..") { + &self.line_scratch[1..] } else { - out.extend_from_slice(&self.line_scratch); + &self.line_scratch + }; + if max_bytes.is_some_and(|maximum| { + out.len() + .checked_add(line.len()) + .is_none_or(|length| length > maximum) + }) { + return Err(NntpError::ResponseTooLarge( + "multi-line response exceeds its byte bound".into(), + )); } + out.extend_from_slice(line); } Ok(()) @@ -2551,6 +3029,52 @@ mod tests { assert_eq!(conn.state, ConnectionState::Ready); } + #[tokio::test] + async fn head_by_article_number_is_exact_and_bounded() { + let mut groups = HashMap::new(); + groups.insert("alt.binaries.test".into(), (1, 42, 42)); + let headers = b"Date: Wed, 07 May 2025 20:50:00 +0000\r\nXref: news 42\r\n".to_vec(); + let server = MockNntpServer::start(MockConfig { + groups, + heads: HashMap::from([(42, headers.clone())]), + ..MockConfig::default() + }) + .await; + let mut connection = NntpConnection::new("head-exact".into()); + connection + .connect(&test_config(server.port())) + .await + .unwrap(); + connection.group("alt.binaries.test").await.unwrap(); + let response = connection + .fetch_head_number(42, headers.len()) + .await + .unwrap(); + assert_eq!(response.code, 221); + assert_eq!(response.data, Some(headers)); + + let mut groups = HashMap::new(); + groups.insert("alt.binaries.test".into(), (1, 42, 42)); + let server = MockNntpServer::start(MockConfig { + groups, + heads: HashMap::from([(42, b"Header: value that exceeds the bound\r\n".to_vec())]), + ..MockConfig::default() + }) + .await; + let mut connection = NntpConnection::new("head-bounded".into()); + connection + .connect(&test_config(server.port())) + .await + .unwrap(); + connection.group("alt.binaries.test").await.unwrap(); + let error = connection + .fetch_head_number(42, 8) + .await + .expect_err("bounded HEAD"); + assert!(matches!(error, NntpError::ResponseTooLarge(_))); + assert_eq!(connection.state, ConnectionState::Error); + } + #[tokio::test] async fn test_connect_with_auth() { let server = MockNntpServer::start(MockConfig { @@ -2696,6 +3220,37 @@ mod tests { assert_eq!(conn.state, ConnectionState::Ready); } + #[tokio::test] + async fn lossless_overview_negotiates_format_and_retains_defects() { + let mut groups = HashMap::new(); + groups.insert("alt.binaries.test".into(), (2u64, 1u64, 2u64)); + let server = MockNntpServer::start(MockConfig { + groups, + xover_raw_entries: vec![ + b"1\tEspa\xf1a\tposter@test\tDate\t\t\t10\t1".to_vec(), + b"2\tSub\tject\tposter@test\tDate\t\t\t10\t1".to_vec(), + ], + ..MockConfig::default() + }) + .await; + let config = test_config(server.port()); + let mut conn = NntpConnection::new("test".into()); + conn.connect(&config).await.unwrap(); + conn.group("alt.binaries.test").await.unwrap(); + + let format = conn.overview_format().await.unwrap(); + let rows = conn.xover_lossless(1, 2, &format).await.unwrap(); + + assert_eq!(format.fields.len(), 7); + assert_eq!(rows.rows[0].fields[0], b"Espa\xf1a"); + assert_eq!(rows.defective_rows[0].article_number, Some(2)); + assert_eq!( + rows.defective_rows[0].failure_code, + crate::overview::DefectiveOverviewRowCode::FieldCountInvalid + ); + assert_eq!(conn.state, ConnectionState::Ready); + } + #[tokio::test] async fn test_xover_empty_range() { let mut groups = HashMap::new(); @@ -2890,6 +3445,48 @@ mod tests { assert_eq!(conn.state, ConnectionState::Ready); } + #[tokio::test] + async fn body_prefix_retires_an_incomplete_connection_without_exceeding_the_bound() { + let mut articles = HashMap::new(); + articles.insert( + "prefix@test".into(), + b"first line\r\nsecond line\r\n".to_vec(), + ); + let server = MockNntpServer::start(MockConfig { + articles, + ..MockConfig::default() + }) + .await; + let config = test_config(server.port()); + let mut conn = NntpConnection::new("test".into()); + conn.connect(&config).await.unwrap(); + + let prefix = conn.fetch_body_prefix("prefix@test", 8).await.unwrap(); + assert_eq!(prefix.data, b"first li"); + assert!(!prefix.complete); + assert_eq!(conn.state, ConnectionState::Disconnected); + assert!(!conn.is_connected()); + } + + #[tokio::test] + async fn body_prefix_keeps_a_complete_bounded_connection_ready() { + let mut articles = HashMap::new(); + articles.insert("small@test".into(), b"small\r\n".to_vec()); + let server = MockNntpServer::start(MockConfig { + articles, + ..MockConfig::default() + }) + .await; + let config = test_config(server.port()); + let mut conn = NntpConnection::new("test".into()); + conn.connect(&config).await.unwrap(); + + let prefix = conn.fetch_body_prefix("small@test", 1024).await.unwrap(); + assert_eq!(prefix.data, b"small\r\n"); + assert!(prefix.complete); + assert_eq!(conn.state, ConnectionState::Ready); + } + #[tokio::test] async fn test_io_heartbeat_ticks_on_every_read() { // Verify the socket-liveness heartbeat advances on each NNTP response @@ -3372,6 +3969,66 @@ mod tests { assert_eq!(conn.state, ConnectionState::Ready); } + #[tokio::test] + async fn unsupported_xpat_keeps_the_connection_ready_for_bounded_xover() { + let mut groups = HashMap::new(); + groups.insert("alt.binaries.test".into(), (1, 1, 1)); + let server = MockNntpServer::start(MockConfig { + groups, + xpat_unsupported: true, + xover_entries: vec![ + "1\tTraitors Espana S02E01\tposter\tWed, 07 May 2025 20:50:00 +0000\t<1@test>\t\t100\t1".into(), + ], + ..MockConfig::default() + }) + .await; + let mut connection = NntpConnection::new("xpat-fallback".into()); + connection + .connect(&test_config(server.port())) + .await + .unwrap(); + connection.group("alt.binaries.test").await.unwrap(); + let format = connection.overview_format().await.unwrap(); + let error = connection + .xpat_bounded("Subject", ArticleRange::Range(1, 1), &["*Traitors*"], 1024) + .await + .expect_err("unsupported XPAT"); + assert!(matches!(error, NntpError::UnsupportedCommand(_))); + assert_eq!(connection.state, ConnectionState::Ready); + let overview = connection + .xover_lossless_bounded(1, 1, &format, 1024) + .await + .unwrap(); + assert_eq!(overview.value.rows.len(), 1); + } + + #[tokio::test] + async fn bounded_lossless_xover_fails_before_retaining_an_oversized_response() { + let mut groups = HashMap::new(); + groups.insert("alt.binaries.test".into(), (1, 1, 1)); + let server = MockNntpServer::start(MockConfig { + groups, + xover_entries: vec![ + "1\tTraitors Espana S02E01\tposter\tWed, 07 May 2025 20:50:00 +0000\t<1@test>\t\t100\t1".into(), + ], + ..MockConfig::default() + }) + .await; + let mut connection = NntpConnection::new("xover-bounded".into()); + connection + .connect(&test_config(server.port())) + .await + .unwrap(); + connection.group("alt.binaries.test").await.unwrap(); + let format = connection.overview_format().await.unwrap(); + let error = connection + .xover_lossless_bounded(1, 1, &format, 8) + .await + .expect_err("bounded XOVER"); + assert!(matches!(error, NntpError::ResponseTooLarge(_))); + assert_eq!(connection.state, ConnectionState::Error); + } + #[tokio::test] async fn test_xpat_multiple_patterns() { let mut groups = HashMap::new(); diff --git a/crates/nzb-nntp/src/error.rs b/crates/nzb-nntp/src/error.rs index 2551b6c5..cd154615 100644 --- a/crates/nzb-nntp/src/error.rs +++ b/crates/nzb-nntp/src/error.rs @@ -47,6 +47,14 @@ pub enum NntpError { #[error("Protocol error: {0}")] Protocol(String), + /// The server explicitly rejected a command as unsupported. + #[error("Unsupported command: {0}")] + UnsupportedCommand(String), + + /// A bounded NNTP response exceeded its admitted byte limit. + #[error("Response too large: {0}")] + ResponseTooLarge(String), + /// Underlying I/O error. #[error("I/O error: {0}")] Io(#[from] std::io::Error), diff --git a/crates/nzb-nntp/src/lib.rs b/crates/nzb-nntp/src/lib.rs index b12041fa..533bd76d 100644 --- a/crates/nzb-nntp/src/lib.rs +++ b/crates/nzb-nntp/src/lib.rs @@ -14,6 +14,7 @@ pub mod config; pub mod connection; pub mod downloader; pub mod error; +pub mod overview; pub mod pipeline; pub mod pool; pub mod server; @@ -25,11 +26,15 @@ pub mod testutil; pub use capabilities::NntpCapabilities; pub use config::{Article, ListActiveEntry, ServerConfig}; pub use connection::{ - ArticleRange, ConnectionState, GroupResponse, HeaderEntry, NntpConnection, NntpResponse, - XoverEntry, + ArticleRange, BodyPrefixResponse, BoundedResponse, ConnectionState, GroupResponse, HeaderEntry, + NntpConnection, NntpResponse, XoverEntry, }; pub use downloader::{ArticleResult, Downloader}; pub use error::{NntpError, NntpResult}; +pub use overview::{ + DefectiveOverviewRow, DefectiveOverviewRowCode, LosslessOverviewRow, LosslessOverviewRows, + OverviewFormat, +}; pub use pipeline::{Pipeline, StatPipeline, StatResult}; pub use pool::ConnectionPool; pub use server::ServerState; diff --git a/crates/nzb-nntp/src/overview.rs b/crates/nzb-nntp/src/overview.rs new file mode 100644 index 00000000..21dde71e --- /dev/null +++ b/crates/nzb-nntp/src/overview.rs @@ -0,0 +1,223 @@ +use std::collections::BTreeSet; + +use crate::error::{NntpError, NntpResult}; + +const REQUIRED_HEADER_FIELDS: [&[u8]; 5] = [ + b"Subject:", + b"From:", + b"Date:", + b"Message-ID:", + b"References:", +]; +const MAX_OVERVIEW_FIELDS: usize = 64; +const MAX_FIELD_DESCRIPTOR_BYTES: usize = 256; + +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct OverviewFormat { + pub fields: Vec>, +} + +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct LosslessOverviewRow { + pub article_number: u64, + pub fields: Vec>, +} + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum DefectiveOverviewRowCode { + ArticleNumberInvalid, + ArticleNumberOutOfRange, + DuplicateArticleNumber, + FieldCountInvalid, +} + +impl DefectiveOverviewRowCode { + pub fn as_str(self) -> &'static str { + match self { + Self::ArticleNumberInvalid => "article_number_invalid", + Self::ArticleNumberOutOfRange => "article_number_out_of_range", + Self::DuplicateArticleNumber => "duplicate_article_number", + Self::FieldCountInvalid => "field_count_invalid", + } + } +} + +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct DefectiveOverviewRow { + pub article_number: Option, + pub wire_line: Vec, + pub failure_code: DefectiveOverviewRowCode, +} + +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct LosslessOverviewRows { + pub rows: Vec, + pub defective_rows: Vec, +} + +pub fn parse_overview_format(data: &[u8]) -> NntpResult { + let fields = wire_lines(data) + .map(|line| line.to_vec()) + .collect::>(); + if fields.len() < REQUIRED_HEADER_FIELDS.len() + 2 || fields.len() > MAX_OVERVIEW_FIELDS { + return Err(NntpError::Protocol( + "LIST OVERVIEW.FMT field count is invalid".into(), + )); + } + let metadata_fields_valid = fields[5].eq_ignore_ascii_case(b":bytes") + && fields[6].eq_ignore_ascii_case(b":lines") + || fields[5].eq_ignore_ascii_case(b"Bytes:") && fields[6].eq_ignore_ascii_case(b"Lines:"); + if !metadata_fields_valid { + return Err(NntpError::Protocol( + "LIST OVERVIEW.FMT metadata fields are invalid".into(), + )); + } + for (index, field) in fields.iter().enumerate() { + if field.is_empty() + || field.len() > MAX_FIELD_DESCRIPTOR_BYTES + || !field.iter().all(u8::is_ascii) + { + return Err(NntpError::Protocol( + "LIST OVERVIEW.FMT field descriptor is invalid".into(), + )); + } + if let Some(required) = REQUIRED_HEADER_FIELDS.get(index) + && !field.eq_ignore_ascii_case(required) + { + return Err(NntpError::Protocol( + "LIST OVERVIEW.FMT required field order is invalid".into(), + )); + } + } + Ok(OverviewFormat { fields }) +} + +pub fn parse_lossless_overview_rows( + data: &[u8], + format: &OverviewFormat, + start_article: u64, + end_article: u64, +) -> LosslessOverviewRows { + let mut rows = Vec::new(); + let mut defective_rows = Vec::new(); + let mut seen = BTreeSet::new(); + for line in wire_lines(data) { + let mut parts = line.split(|byte| *byte == b'\t'); + let article_bytes = parts.next().unwrap_or_default(); + let article_number = parse_ascii_u64(article_bytes); + let fields = parts.map(|part| part.to_vec()).collect::>(); + let failure_code = if article_number.is_none() { + Some(DefectiveOverviewRowCode::ArticleNumberInvalid) + } else if article_number + .is_some_and(|number| number < start_article || number > end_article) + { + Some(DefectiveOverviewRowCode::ArticleNumberOutOfRange) + } else if fields.len() != format.fields.len() { + Some(DefectiveOverviewRowCode::FieldCountInvalid) + } else if article_number.is_some_and(|number| !seen.insert(number)) { + Some(DefectiveOverviewRowCode::DuplicateArticleNumber) + } else { + None + }; + if let Some(failure_code) = failure_code { + defective_rows.push(DefectiveOverviewRow { + article_number, + wire_line: line.to_vec(), + failure_code, + }); + } else if let Some(article_number) = article_number { + rows.push(LosslessOverviewRow { + article_number, + fields, + }); + } + } + LosslessOverviewRows { + rows, + defective_rows, + } +} + +fn wire_lines(data: &[u8]) -> impl Iterator { + data.split(|byte| *byte == b'\n').filter_map(|line| { + let line = line.strip_suffix(b"\r").unwrap_or(line); + (!line.is_empty()).then_some(line) + }) +} + +fn parse_ascii_u64(value: &[u8]) -> Option { + if value.is_empty() || !value.iter().all(u8::is_ascii_digit) { + return None; + } + std::str::from_utf8(value).ok()?.parse().ok() +} + +#[cfg(test)] +mod tests { + use super::*; + + fn format() -> OverviewFormat { + parse_overview_format( + b"Subject:\r\nFrom:\r\nDate:\r\nMessage-ID:\r\nReferences:\r\n:bytes\r\n:lines\r\nXref:full\r\n", + ) + .expect("format") + } + + #[test] + fn format_requires_the_standard_prefix_and_preserves_optional_fields() { + let format = format(); + assert_eq!(format.fields.len(), 8); + assert_eq!(format.fields[7], b"Xref:full"); + assert!(parse_overview_format(b"From:\r\nSubject:\r\n").is_err()); + } + + #[test] + fn format_accepts_and_preserves_the_alternative_metadata_names() { + let format = parse_overview_format( + b"Subject:\r\nFrom:\r\nDate:\r\nMessage-ID:\r\nReferences:\r\nBytes:\r\nLines:\r\nXref:full\r\n", + ) + .expect("alternative metadata names"); + assert_eq!(format.fields[5], b"Bytes:"); + assert_eq!(format.fields[6], b"Lines:"); + assert!( + parse_overview_format( + b"Subject:\r\nFrom:\r\nDate:\r\nMessage-ID:\r\nReferences:\r\nBytes:\r\n:lines\r\n" + ) + .is_err() + ); + } + + #[test] + fn rows_preserve_legacy_bytes_and_negotiated_order() { + let rows = parse_lossless_overview_rows( + b"42\tEspa\xf1a\tPoster\tDate\t\t\t10\t1\tserver group:42\r\n", + &format(), + 40, + 50, + ); + assert!(rows.defective_rows.is_empty()); + assert_eq!(rows.rows[0].article_number, 42); + assert_eq!(rows.rows[0].fields[0], b"Espa\xf1a"); + assert_eq!(rows.rows[0].fields[7], b"server group:42"); + } + + #[test] + fn malformed_rows_remain_exact_and_typed() { + let rows = parse_lossless_overview_rows( + b"bad\tSubject\tPoster\tDate\t\t\t10\t1\tserver group:1\r\n43\tSub\tject\tPoster\tDate\t\t\t10\t1\tserver group:43\r\n", + &format(), + 40, + 50, + ); + assert_eq!(rows.defective_rows.len(), 2); + assert_eq!( + rows.defective_rows[0].failure_code, + DefectiveOverviewRowCode::ArticleNumberInvalid + ); + assert_eq!(rows.defective_rows[0].wire_line[0..3], *b"bad"); + assert_eq!( + rows.defective_rows[1].failure_code, + DefectiveOverviewRowCode::FieldCountInvalid + ); + } +} diff --git a/crates/nzb-nntp/src/pipeline.rs b/crates/nzb-nntp/src/pipeline.rs index 952bf67a..06555d18 100644 --- a/crates/nzb-nntp/src/pipeline.rs +++ b/crates/nzb-nntp/src/pipeline.rs @@ -335,12 +335,11 @@ impl StatPipeline { return Err(NntpError::ServiceUnavailable(resp.message)); } _ => { - // Unknown response — treat as missing but don't abort - trace!(code = resp.code, mid = %mid, "Unexpected STAT response"); - results.push(StatResult { - message_id: mid.clone(), - exists: false, - }); + conn.state = ConnectionState::Error; + return Err(NntpError::Protocol(format!( + "Unexpected STAT response {}: {}", + resp.code, resp.message + ))); } } } @@ -700,4 +699,25 @@ mod tests { assert!(results[0].exists); assert!(results[1].exists); } + + #[tokio::test] + async fn test_stat_pipeline_unknown_response_fails_closed() { + let mut overrides = HashMap::new(); + overrides.insert("unknown@test".to_string(), 499); + let server = MockNntpServer::start(MockConfig { + article_response_overrides: overrides, + ..MockConfig::default() + }) + .await; + let config = test_config(server.port()); + let mut conn = NntpConnection::new("test".into()); + conn.connect(&config).await.unwrap(); + + let mut stat = StatPipeline::new(); + stat.add("unknown@test".into()); + assert!(matches!( + stat.execute(&mut conn).await, + Err(NntpError::Protocol(_)) + )); + } } diff --git a/crates/nzb-nntp/src/testutil.rs b/crates/nzb-nntp/src/testutil.rs index ae0530f1..19637972 100644 --- a/crates/nzb-nntp/src/testutil.rs +++ b/crates/nzb-nntp/src/testutil.rs @@ -110,14 +110,26 @@ pub struct MockConfig { pub groups: HashMap, /// Articles: message-id (without angle brackets) -> body bytes. pub articles: HashMap>, + /// Exact HEAD response bytes keyed by article number. + pub heads: HashMap>, /// XOVER entries as pre-formatted tab-delimited lines. pub xover_entries: Vec, + /// XOVER entries keyed by the exact requested article range. + pub xover_entries_by_range: HashMap>, + /// Raw XOVER entries used when exact non-UTF-8 bytes are required. + pub xover_raw_entries: Vec>, /// XHDR entries as pre-formatted `artnum value` lines. pub xhdr_entries: Vec, /// XPAT entries as pre-formatted `artnum value` lines. pub xpat_entries: Vec, + /// XPAT entries keyed by the exact range and pattern arguments. + pub xpat_entries_by_request: HashMap>, + /// If true, XPAT returns an unsupported-command response. + pub xpat_unsupported: bool, /// LIST ACTIVE entries as pre-formatted `groupname last first status` lines. pub list_active_entries: Vec, + /// LIST OVERVIEW.FMT entries in negotiated order. + pub overview_format_entries: Vec, /// If true, `POST` returns 440 instead of accepting an article body. pub post_not_permitted: bool, /// Captured raw articles received via `POST`, after un-dot-stuffing and @@ -173,10 +185,24 @@ impl Default for MockConfig { service_unavailable: false, groups: HashMap::new(), articles: HashMap::new(), + heads: HashMap::new(), xover_entries: Vec::new(), + xover_entries_by_range: HashMap::new(), + xover_raw_entries: Vec::new(), xhdr_entries: Vec::new(), xpat_entries: Vec::new(), + xpat_entries_by_request: HashMap::new(), + xpat_unsupported: false, list_active_entries: Vec::new(), + overview_format_entries: vec![ + "Subject:".into(), + "From:".into(), + "Date:".into(), + "Message-ID:".into(), + "References:".into(), + ":bytes".into(), + ":lines".into(), + ], post_not_permitted: false, posted_articles: None, silent_close_after_bytes: None, @@ -500,15 +526,27 @@ async fn handle_connection(stream: tokio::net::TcpStream, config: Arc { + let range_entries = parts + .get(1) + .and_then(|range| config.xover_entries_by_range.get(*range)); + let entries = if config.xover_entries_by_range.is_empty() { + &config.xover_entries + } else { + range_entries.unwrap_or(&config.xover_entries) + }; if !authenticated { mwrite!(conn, b"480 Authentication required\r\n"); } else if selected_group.is_none() { mwrite!(conn, b"412 No newsgroup selected\r\n"); - } else if config.xover_entries.is_empty() { + } else if entries.is_empty() && config.xover_raw_entries.is_empty() { mwrite!(conn, b"420 No articles in range\r\n"); } else { mwrite!(conn, b"224 Overview data follows\r\n"); - for entry in &config.xover_entries { + for entry in &config.xover_raw_entries { + mwrite!(conn, entry); + mwrite!(conn, b"\r\n"); + } + for entry in entries { mwrite!(conn, entry.as_bytes()); mwrite!(conn, b"\r\n"); } @@ -516,6 +554,27 @@ async fn handle_connection(stream: tokio::net::TcpStream, config: Arc { + if !authenticated { + mwrite!(conn, b"480 Authentication required\r\n"); + } else if selected_group.is_none() { + mwrite!(conn, b"412 No newsgroup selected\r\n"); + } else if let Some(article_number) = + parts.get(1).and_then(|value| value.parse().ok()) + && let Some(headers) = config.heads.get(&article_number) + { + let resp = format!("221 {article_number} article headers follow\r\n"); + mwrite!(conn, resp.as_bytes()); + mwrite!(conn, headers); + if !headers.ends_with(b"\n") { + mwrite!(conn, b"\r\n"); + } + mwrite!(conn, b".\r\n"); + } else { + mwrite!(conn, b"423 No article with that number\r\n"); + } + } + "XHDR" => { if !authenticated { mwrite!(conn, b"480 Authentication required\r\n"); @@ -532,13 +591,23 @@ async fn handle_connection(stream: tokio::net::TcpStream, config: Arc { + let request_entries = parts + .get(2) + .and_then(|request| config.xpat_entries_by_request.get(*request)); + let entries = if config.xpat_entries_by_request.is_empty() { + &config.xpat_entries + } else { + request_entries.unwrap_or(&config.xpat_entries) + }; if !authenticated { mwrite!(conn, b"480 Authentication required\r\n"); - } else if config.xpat_entries.is_empty() { + } else if config.xpat_unsupported { + mwrite!(conn, b"500 XPAT unsupported\r\n"); + } else if entries.is_empty() { mwrite!(conn, b"420 No articles matched\r\n"); } else { mwrite!(conn, b"221 Header data follows\r\n"); - for entry in &config.xpat_entries { + for entry in entries { mwrite!(conn, entry.as_bytes()); mwrite!(conn, b"\r\n"); } @@ -549,6 +618,16 @@ async fn handle_connection(stream: tokio::net::TcpStream, config: Arc { if !authenticated { mwrite!(conn, b"480 Authentication required\r\n"); + } else if parts + .get(1) + .is_some_and(|value| value.eq_ignore_ascii_case("OVERVIEW.FMT")) + { + mwrite!(conn, b"215 Order of fields in overview database\r\n"); + for entry in &config.overview_format_entries { + mwrite!(conn, entry.as_bytes()); + mwrite!(conn, b"\r\n"); + } + mwrite!(conn, b".\r\n"); } else if config.list_active_entries.is_empty() { mwrite!(conn, b"215 List of newsgroups follows\r\n"); mwrite!(conn, b".\r\n"); diff --git a/desktop/src-tauri/Cargo.lock b/desktop/src-tauri/Cargo.lock index 670fc184..edf6e984 100644 --- a/desktop/src-tauri/Cargo.lock +++ b/desktop/src-tauri/Cargo.lock @@ -3069,7 +3069,7 @@ dependencies = [ [[package]] name = "nzb-nntp" -version = "0.2.24" +version = "0.2.26" dependencies = [ "anyhow", "arc-swap", @@ -4363,6 +4363,7 @@ dependencies = [ "chrono", "clap", "flate2", + "hex", "http", "libc", "mime_guess", @@ -4378,6 +4379,7 @@ dependencies = [ "rustls", "serde", "serde_json", + "sha2 0.11.0", "tokio", "tokio-util", "tower-http 0.7.0",