From ea8ae075dd6a7cbc552f2cad7b1f8f35565c8c04 Mon Sep 17 00:00:00 2001 From: Tien Do Nam Date: Sun, 12 Jul 2026 03:59:26 +0200 Subject: [PATCH] feat: add v2 http server --- core/Cargo.lock | 2 + core/Cargo.toml | 4 +- core/src/http/dto_v2.rs | 85 +++- core/src/http/server/controller/v2.rs | 491 ++++++++++++++++++++++- core/src/http/server/error.rs | 7 + core/src/http/server/event.rs | 90 +++++ core/src/http/server/mod.rs | 152 ++++--- core/src/http/server/query.rs | 45 +++ core/src/http/server/session.rs | 48 +++ core/src/main.rs | 64 ++- core/tests/v2_server.rs | 551 ++++++++++++++++++++++++++ 11 files changed, 1472 insertions(+), 67 deletions(-) create mode 100644 core/src/http/server/event.rs create mode 100644 core/src/http/server/query.rs create mode 100644 core/src/http/server/session.rs create mode 100644 core/tests/v2_server.rs diff --git a/core/Cargo.lock b/core/Cargo.lock index c5723555..18943aa3 100644 --- a/core/Cargo.lock +++ b/core/Cargo.lock @@ -1242,6 +1242,7 @@ dependencies = [ "bytes", "ed25519-dalek", "flate2", + "form_urlencoded", "futures-util", "http-body-util", "hyper", @@ -1255,6 +1256,7 @@ dependencies = [ "serde", "serde_json", "sha2", + "socket2 0.6.2", "thiserror 2.0.18", "tokio", "tokio-rustls", diff --git a/core/Cargo.toml b/core/Cargo.toml index 223fd986..11eb4fc4 100644 --- a/core/Cargo.toml +++ b/core/Cargo.toml @@ -9,6 +9,7 @@ base64 = "0.22.1" bytes = "1.11" ed25519-dalek = { version = "2.2", features = ["pem", "rand_core"], optional = true } flate2 = { version = "1.1", optional = true } +form_urlencoded = { version = "1.2", optional = true } futures-util = { version = "0.3.31", features = ["sink"] } http-body-util = { version = "0.1.3", optional = true } hyper = { version = "1.8.1", optional = true } @@ -22,6 +23,7 @@ rustls = { version = "0.23.32", default-features = false, features = ["ring", "t serde = { version = "1.0", features = ["derive"] } serde_json = "1.0" sha2 = { version = "0.10.9", optional = true } +socket2 = { version = "0.6.2", optional = true } thiserror = "2.0.18" tokio = { version = "1.49.0", features = ["full"] } tokio-rustls = { version = "0.26.4", default-features = false, features = ["ring", "tls12"], optional = true } @@ -37,7 +39,7 @@ x509-parser = { version = "0.18.0", features = ["verify"], optional = true } [features] default = [] crypto = ["ed25519-dalek", "rsa", "sha2"] -http = ["crypto", "http-body-util", "hyper", "hyper-util", "pem", "reqwest", "rustls", "tokio-rustls", "x509-parser"] +http = ["crypto", "form_urlencoded", "http-body-util", "hyper", "hyper-util", "pem", "reqwest", "rustls", "socket2", "tokio-rustls", "x509-parser"] webrtc-signaling = ["tokio-tungstenite"] webrtc = ["crypto", "flate2", "dep:webrtc", "webrtc-signaling", "x509-parser"] full = ["crypto", "http", "webrtc"] diff --git a/core/src/http/dto_v2.rs b/core/src/http/dto_v2.rs index faba4fe6..b116f518 100644 --- a/core/src/http/dto_v2.rs +++ b/core/src/http/dto_v2.rs @@ -3,6 +3,48 @@ use crate::model::transfer::FileDto; use serde::{Deserialize, Serialize}; use std::collections::HashMap; +/// The protocol version (major.minor) implemented by this crate for the v2 protocol. +pub const PROTOCOL_VERSION_V2: &str = "2.1"; + +/// Serde helpers for `DeviceType` in the v2 protocol. +/// +/// The v2 protocol uses lowercase values (e.g. "desktop") on the wire. +/// Unknown values fall back to `Desktop` as required by the protocol (section 7.1). +pub(crate) mod device_type_v2 { + use crate::model::discovery::DeviceType; + use serde::{Deserialize, Deserializer, Serializer}; + + pub fn serialize( + value: &Option, + serializer: S, + ) -> Result { + match value { + Some(device_type) => serializer.serialize_str(match device_type { + DeviceType::Mobile => "mobile", + DeviceType::Desktop => "desktop", + DeviceType::Web => "web", + DeviceType::Headless => "headless", + DeviceType::Server => "server", + }), + None => serializer.serialize_none(), + } + } + + pub fn deserialize<'de, D: Deserializer<'de>>( + deserializer: D, + ) -> Result, D::Error> { + let value = Option::::deserialize(deserializer)?; + Ok(value.map(|value| match value.to_lowercase().as_str() { + "mobile" => DeviceType::Mobile, + "desktop" => DeviceType::Desktop, + "web" => DeviceType::Web, + "headless" => DeviceType::Headless, + "server" => DeviceType::Server, + _ => DeviceType::Desktop, + })) + } +} + /// Protocol type for HTTP or HTTPS connections. #[derive(Clone, Debug, Deserialize, Eq, Serialize, PartialEq)] #[serde(rename_all = "lowercase")] @@ -39,7 +81,11 @@ pub struct MulticastMessageV2 { pub device_model: Option, /// Device type category. Optional. - #[serde(skip_serializing_if = "Option::is_none")] + #[serde( + default, + with = "device_type_v2", + skip_serializing_if = "Option::is_none" + )] pub device_type: Option, /// Fingerprint for device identification. @@ -79,7 +125,11 @@ pub struct RegisterDtoV2 { pub device_model: Option, /// Device type category. Optional. - #[serde(skip_serializing_if = "Option::is_none")] + #[serde( + default, + with = "device_type_v2", + skip_serializing_if = "Option::is_none" + )] pub device_type: Option, /// Fingerprint for device identification. @@ -114,7 +164,11 @@ pub struct RegisterResponseDtoV2 { pub device_model: Option, /// Device type category. Optional. - #[serde(skip_serializing_if = "Option::is_none")] + #[serde( + default, + with = "device_type_v2", + skip_serializing_if = "Option::is_none" + )] pub device_type: Option, /// Fingerprint for device identification. @@ -194,7 +248,11 @@ pub struct InfoResponseDtoV2 { pub device_model: Option, /// Device type category. Optional. - #[serde(skip_serializing_if = "Option::is_none")] + #[serde( + default, + with = "device_type_v2", + skip_serializing_if = "Option::is_none" + )] pub device_type: Option, /// Fingerprint for device identification. @@ -230,6 +288,7 @@ mod tests { assert!(json.contains("\"announce\":true")); assert!(json.contains("\"download\":true")); assert!(json.contains("\"protocol\":\"https\"")); + assert!(json.contains("\"deviceType\":\"mobile\"")); } #[test] @@ -238,7 +297,7 @@ mod tests { "alias": "Secret Banana", "version": "2.0", "deviceModel": "Windows", - "deviceType": "DESKTOP", + "deviceType": "desktop", "fingerprint": "random string", "port": 53317, "protocol": "https", @@ -256,6 +315,22 @@ mod tests { assert!(dto.download); } + #[test] + fn test_device_type_unknown_falls_back_to_desktop() { + // Unknown device types must fall back to desktop (protocol section 7.1). + let json = r#"{ + "alias": "Test Device", + "version": "2.0", + "deviceType": "fridge", + "fingerprint": "abc123", + "port": 53317, + "protocol": "http" + }"#; + + let dto: RegisterDtoV2 = serde_json::from_str(json).unwrap(); + assert_eq!(dto.device_type, Some(DeviceType::Desktop)); + } + #[test] fn test_register_response_without_download_field() { // Test that download defaults to false when not present diff --git a/core/src/http/server/controller/v2.rs b/core/src/http/server/controller/v2.rs index fa1a76ef..f32bc298 100644 --- a/core/src/http/server/controller/v2.rs +++ b/core/src/http/server/controller/v2.rs @@ -1,30 +1,501 @@ -use crate::http::dto::{RegisterDto, RegisterResponseDto}; +use crate::http::dto_v2::{ + InfoResponseDtoV2, PrepareUploadRequestDtoV2, PrepareUploadResponseDtoV2, RegisterDtoV2, + RegisterResponseDtoV2, PROTOCOL_VERSION_V2, +}; use crate::http::server::collect_to_json::CollectToJson; use crate::http::server::error::AppError; +use crate::http::server::event::{PrepareUploadDecisionV2, ServerEventV2, SessionEndReasonV2}; +use crate::http::server::query::parse_query; use crate::http::server::response::JsonResponse; -use crate::http::server::{AppState, RequestClientInfo}; +use crate::http::server::session::{FileStatusV2, SessionFileV2, SessionStateV2, UploadSessionV2}; +use crate::http::server::{AppState, RequestClientInfo, V2State}; +use bytes::Bytes; +use http_body_util::{BodyExt, Full}; use hyper::body::Incoming; -use hyper::StatusCode; +use hyper::{Request, Response, StatusCode}; +use std::collections::HashMap; +use std::sync::Arc; +use tokio::sync::{mpsc, oneshot}; +use uuid::Uuid; + +/// Maximum failed PIN attempts per IP before requests are blocked with 429. +const MAX_PIN_ATTEMPTS: u32 = 3; + +/// Channel capacity for file upload chunks (provides backpressure). +const UPLOAD_CHANNEL_CAPACITY: usize = 16; pub(crate) async fn register( body: Incoming, state: AppState, client_info: RequestClientInfo, -) -> Result, AppError> { - let payload = body.collect_to_json::().await?; +) -> Result, AppError> { + let payload = body.collect_to_json::().await?; + + if let Some(v2) = &state.v2 { + let _ = v2 + .event_tx + .send(ServerEventV2::Register { + ip: client_info.ip, + info: payload, + }) + .await; + } let info = state.info.lock().await.clone(); - let has_web_interface = state.web.lock().await.is_some(); + let download = state.web.lock().await.is_some(); Ok(JsonResponse { status: StatusCode::OK, - body: RegisterResponseDto { + body: RegisterResponseDtoV2 { alias: info.alias, - version: info.version, + version: PROTOCOL_VERSION_V2.to_string(), device_model: info.device_model, device_type: info.device_type, - token: info.token, - has_web_interface, + fingerprint: info.token, + download, }, }) } + +pub(crate) async fn info(state: AppState) -> Result, AppError> { + let info = state.info.lock().await.clone(); + let download = state.web.lock().await.is_some(); + + Ok(JsonResponse { + status: StatusCode::OK, + body: InfoResponseDtoV2 { + alias: info.alias, + version: PROTOCOL_VERSION_V2.to_string(), + device_model: info.device_model, + device_type: info.device_type, + fingerprint: info.token, + download, + }, + }) +} + +pub(crate) async fn prepare_upload( + req: Request, + state: AppState, + client_info: RequestClientInfo, +) -> Result>, AppError> { + let v2 = require_v2(&state)?; + let query = parse_query(req.uri().query()); + + check_pin(&v2, &query, &client_info).await?; + + let payload = req + .into_body() + .collect_to_json::() + .await?; + + if payload.files.is_empty() { + return Err(AppError::BadRequest("No files provided".to_string())); + } + + // Claim the single session slot. + { + let mut slot = v2.session.lock().await; + if slot.is_some() { + return Err(AppError::Message( + StatusCode::CONFLICT, + "Blocked by another session".to_string(), + )); + } + *slot = Some(SessionStateV2::Pending); + } + + // Frees the slot again if this request is aborted before a session is created. + let mut pending_guard = PendingSessionGuard::new(v2.clone()); + + let (decision_tx, decision_rx) = oneshot::channel(); + let event = ServerEventV2::PrepareUpload { + ip: client_info.ip, + info: payload.info, + files: payload.files.clone(), + decision_tx, + }; + if v2.event_tx.send(event).await.is_err() { + return Err(AppError::Status(StatusCode::INTERNAL_SERVER_ERROR)); + } + + let decision = decision_rx + .await + .map_err(|_| AppError::Status(StatusCode::INTERNAL_SERVER_ERROR))?; + + let accepted_ids = match decision { + PrepareUploadDecisionV2::Decline => { + pending_guard.clear().await; + return Err(AppError::Message( + StatusCode::FORBIDDEN, + "Rejected".to_string(), + )); + } + PrepareUploadDecisionV2::Accept(ids) => ids, + }; + + let files: HashMap = payload + .files + .into_iter() + .filter(|(id, _)| accepted_ids.contains(id)) + .map(|(id, dto)| { + let file = SessionFileV2 { + dto, + token: Uuid::new_v4().to_string(), + status: FileStatusV2::Pending, + }; + (id, file) + }) + .collect(); + + if files.is_empty() { + // Nothing to transfer. + pending_guard.clear().await; + let mut res = Response::new(Full::default()); + *res.status_mut() = StatusCode::NO_CONTENT; + return Ok(res); + } + + let session_id = Uuid::new_v4().to_string(); + let tokens: HashMap = files + .iter() + .map(|(id, file)| (id.clone(), file.token.clone())) + .collect(); + + { + let mut slot = v2.session.lock().await; + *slot = Some(SessionStateV2::Active(UploadSessionV2 { + session_id: session_id.clone(), + sender_ip: client_info.ip, + files, + })); + } + pending_guard.disarm(); + + tracing::info!("Upload session created: {session_id}"); + + Ok(JsonResponse { + status: StatusCode::OK, + body: PrepareUploadResponseDtoV2 { + session_id, + files: tokens, + }, + } + .into_response()) +} + +pub(crate) async fn upload( + req: Request, + state: AppState, + client_info: RequestClientInfo, +) -> Result>, AppError> { + let v2 = require_v2(&state)?; + let query = parse_query(req.uri().query()); + + let (Some(session_id), Some(file_id), Some(token)) = ( + query.get("sessionId"), + query.get("fileId"), + query.get("token"), + ) else { + return Err(AppError::Message( + StatusCode::BAD_REQUEST, + "Missing parameters".to_string(), + )); + }; + + // Validate the request and mark the file as in progress. + let file_dto = { + let mut slot = v2.session.lock().await; + let Some(SessionStateV2::Active(session)) = slot.as_mut() else { + return Err(invalid_token_error()); + }; + if session.session_id != *session_id || session.sender_ip != client_info.ip { + return Err(invalid_token_error()); + } + let Some(file) = session.files.get_mut(file_id) else { + return Err(invalid_token_error()); + }; + if file.token != *token || file.status != FileStatusV2::Pending { + return Err(invalid_token_error()); + } + file.status = FileStatusV2::InProgress; + file.dto.clone() + }; + + // Marks the file as failed if this request is aborted mid-transfer. + let mut upload_guard = UploadGuard::new(v2.clone(), session_id.clone(), file_id.clone()); + + let (binary_tx, binary_rx) = mpsc::channel::(UPLOAD_CHANNEL_CAPACITY); + let (result_tx, result_rx) = oneshot::channel::>(); + + let event = ServerEventV2::FileUpload { + session_id: session_id.clone(), + file_id: file_id.clone(), + file: file_dto, + binary_rx, + result_tx, + }; + if v2.event_tx.send(event).await.is_err() { + upload_guard.finish(false).await; + return Err(AppError::Status(StatusCode::INTERNAL_SERVER_ERROR)); + } + + // Forward the request body to the application. + let mut body = req.into_body(); + let mut stream_error = false; + while let Some(frame) = body.frame().await { + match frame { + Ok(frame) => { + let Ok(data) = frame.into_data() else { + continue; // ignore non-data frames (e.g. trailers) + }; + if data.is_empty() { + continue; + } + if binary_tx.send(data).await.is_err() { + // The application dropped the receiver. + stream_error = true; + break; + } + } + Err(err) => { + tracing::warn!("Error reading upload body of file {file_id}: {err:#}"); + stream_error = true; + break; + } + } + } + + // Signal end of file to the application. + drop(binary_tx); + + let success = match stream_error { + true => false, + false => match result_rx.await { + Ok(Ok(())) => true, + Ok(Err(err)) => { + tracing::warn!("Application failed to process file {file_id}: {err}"); + false + } + Err(_) => false, + }, + }; + + upload_guard.finish(success).await; + + match success { + true => Ok(Response::new(Full::default())), + false => Err(AppError::Status(StatusCode::INTERNAL_SERVER_ERROR)), + } +} + +pub(crate) async fn cancel( + req: Request, + state: AppState, +) -> Result>, AppError> { + let v2 = require_v2(&state)?; + let query = parse_query(req.uri().query()); + + if let Some(session_id) = query.get("sessionId") { + let cancelled = { + let mut slot = v2.session.lock().await; + match slot.as_ref() { + Some(SessionStateV2::Active(session)) if session.session_id == *session_id => { + *slot = None; + true + } + _ => false, + } + }; + + if cancelled { + tracing::info!("Upload session cancelled by sender: {session_id}"); + let _ = v2 + .event_tx + .send(ServerEventV2::SessionEnd { + session_id: session_id.clone(), + reason: SessionEndReasonV2::Cancelled, + }) + .await; + } + } + + Ok(Response::new(Full::default())) +} + +fn require_v2(state: &AppState) -> Result, AppError> { + state + .v2 + .clone() + .ok_or(AppError::Status(StatusCode::NOT_FOUND)) +} + +fn invalid_token_error() -> AppError { + AppError::Message( + StatusCode::FORBIDDEN, + "Invalid token or IP address".to_string(), + ) +} + +async fn check_pin( + v2: &Arc, + query: &HashMap, + client_info: &RequestClientInfo, +) -> Result<(), AppError> { + let Some(required_pin) = &v2.pin else { + return Ok(()); + }; + + let mut attempts = v2.pin_attempts.lock().await; + let count = attempts.get(&client_info.ip).copied().unwrap_or(0); + if count >= MAX_PIN_ATTEMPTS { + return Err(AppError::Message( + StatusCode::TOO_MANY_REQUESTS, + "Too many requests".to_string(), + )); + } + + match query.get("pin") { + Some(pin) if pin == required_pin => { + attempts.pop(&client_info.ip); + Ok(()) + } + Some(_) => { + attempts.put(client_info.ip, count + 1); + Err(AppError::Message( + StatusCode::UNAUTHORIZED, + "Invalid PIN".to_string(), + )) + } + None => Err(AppError::Message( + StatusCode::UNAUTHORIZED, + "PIN required".to_string(), + )), + } +} + +/// Frees a claimed `Pending` session slot unless a session was created. +/// +/// The cleanup also runs on drop so the slot is not leaked +/// when the request future is cancelled (e.g. the sender disconnected +/// while the application was still deciding). +struct PendingSessionGuard { + v2: Arc, + armed: bool, +} + +impl PendingSessionGuard { + fn new(v2: Arc) -> Self { + Self { v2, armed: true } + } + + /// Disarms the guard after the pending slot was replaced by an active session. + fn disarm(&mut self) { + self.armed = false; + } + + /// Frees the pending slot immediately. + async fn clear(&mut self) { + self.armed = false; + clear_pending_session(&self.v2).await; + } +} + +impl Drop for PendingSessionGuard { + fn drop(&mut self) { + if !self.armed { + return; + } + let v2 = self.v2.clone(); + tokio::spawn(async move { + clear_pending_session(&v2).await; + }); + } +} + +async fn clear_pending_session(v2: &V2State) { + let mut slot = v2.session.lock().await; + if matches!(*slot, Some(SessionStateV2::Pending)) { + *slot = None; + } +} + +/// Sets the final status of a file after an upload attempt. +/// +/// The cleanup also runs on drop (as a failure) so the file is not stuck +/// in progress when the request future is cancelled mid-transfer. +struct UploadGuard { + v2: Arc, + session_id: String, + file_id: String, + armed: bool, +} + +impl UploadGuard { + fn new(v2: Arc, session_id: String, file_id: String) -> Self { + Self { + v2, + session_id, + file_id, + armed: true, + } + } + + async fn finish(&mut self, success: bool) { + self.armed = false; + finalize_file(&self.v2, &self.session_id, &self.file_id, success).await; + } +} + +impl Drop for UploadGuard { + fn drop(&mut self) { + if !self.armed { + return; + } + let v2 = self.v2.clone(); + let session_id = std::mem::take(&mut self.session_id); + let file_id = std::mem::take(&mut self.file_id); + tokio::spawn(async move { + finalize_file(&v2, &session_id, &file_id, false).await; + }); + } +} + +/// Sets the final status of a file and ends the session once all files are done. +async fn finalize_file(v2: &V2State, session_id: &str, file_id: &str, success: bool) { + let session_ended = { + let mut slot = v2.session.lock().await; + let Some(SessionStateV2::Active(session)) = slot.as_mut() else { + return; + }; + if session.session_id != session_id { + return; + } + if let Some(file) = session.files.get_mut(file_id) { + if file.status == FileStatusV2::InProgress { + file.status = match success { + true => FileStatusV2::Finished, + false => FileStatusV2::Failed, + }; + } + } + match session.is_complete() { + true => { + *slot = None; + true + } + false => false, + } + }; + + if session_ended { + tracing::info!("Upload session finished: {session_id}"); + let _ = v2 + .event_tx + .send(ServerEventV2::SessionEnd { + session_id: session_id.to_string(), + reason: SessionEndReasonV2::Finished, + }) + .await; + } +} diff --git a/core/src/http/server/error.rs b/core/src/http/server/error.rs index c18e5518..28b5f723 100644 --- a/core/src/http/server/error.rs +++ b/core/src/http/server/error.rs @@ -14,6 +14,9 @@ pub enum AppError { #[error("Invalid request: {0}")] BadRequest(String), + + #[error("{0}: {1}")] + Message(StatusCode, String), } impl AppError { @@ -35,6 +38,10 @@ impl AppError { status: StatusCode::BAD_REQUEST, body: ErrorResponse { message }, }, + AppError::Message(status, message) => JsonResponse { + status, + body: ErrorResponse { message }, + }, }; json.into_response() diff --git a/core/src/http/server/event.rs b/core/src/http/server/event.rs new file mode 100644 index 00000000..f6d696ac --- /dev/null +++ b/core/src/http/server/event.rs @@ -0,0 +1,90 @@ +use crate::http::dto_v2::RegisterDtoV2; +use crate::model::transfer::FileDto; +use bytes::Bytes; +use std::collections::{HashMap, HashSet}; +use std::net::IpAddr; +use tokio::sync::{mpsc, oneshot}; + +/// Events emitted by the v2 HTTP server that must be handled by the application. +#[derive(Debug)] +pub enum ServerEventV2 { + /// A device registered itself via `POST /api/localsend/v2/register`. + Register { + /// The IP address of the remote device. + ip: IpAddr, + + /// The device information sent by the remote device. + info: RegisterDtoV2, + }, + + /// A sender requests to upload files via `POST /api/localsend/v2/prepare-upload`. + /// + /// The application must answer on `decision_tx`. + /// Dropping `decision_tx` results in a 500 response. + PrepareUpload { + /// The IP address of the sender. + ip: IpAddr, + + /// The device information of the sender. + info: RegisterDtoV2, + + /// The offered files, mapped by file ID. + files: HashMap, + + /// Channel to send the decision (accept all, a subset, or decline). + decision_tx: oneshot::Sender, + }, + + /// An accepted file is being uploaded via `POST /api/localsend/v2/upload`. + /// + /// Binary chunks arrive on `binary_rx` until the channel is closed. + /// The application should compare the number of received bytes with `file.size` + /// and report the result on `result_tx` which determines the HTTP response + /// (200 on `Ok`, 500 on `Err` or when `result_tx` is dropped). + FileUpload { + /// The session ID of the upload session. + session_id: String, + + /// The ID of the file being uploaded. + file_id: String, + + /// The metadata of the file being uploaded. + file: FileDto, + + /// Channel receiving the binary chunks of the file. + binary_rx: mpsc::Receiver, + + /// Channel to report whether the file was processed successfully. + result_tx: oneshot::Sender>, + }, + + /// An upload session ended. + SessionEnd { + /// The session ID of the ended session. + session_id: String, + + /// Why the session ended. + reason: SessionEndReasonV2, + }, +} + +/// The application's decision for a prepare-upload request. +#[derive(Debug)] +pub enum PrepareUploadDecisionV2 { + /// Accept the given file IDs (a subset of the offered files). + /// An empty set responds with 204 (no file transfer needed). + Accept(HashSet), + + /// Decline the request (403). + Decline, +} + +/// Why an upload session ended. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum SessionEndReasonV2 { + /// All accepted files reached a final state (finished or failed). + Finished, + + /// The sender cancelled the session via `POST /api/localsend/v2/cancel`. + Cancelled, +} diff --git a/core/src/http/server/mod.rs b/core/src/http/server/mod.rs index b2ef00d4..8e42cd02 100644 --- a/core/src/http/server/mod.rs +++ b/core/src/http/server/mod.rs @@ -2,12 +2,17 @@ mod client_cert_verifier; mod collect_to_json; mod controller; mod error; +pub mod event; +mod query; mod response; +mod session; use crate::crypto::cert::public_key_from_cert_der; use crate::http::server::client_cert_verifier::CustomClientCertVerifier; use crate::http::server::controller::web::WebPageState; use crate::http::server::error::AppError; +use crate::http::server::event::ServerEventV2; +use crate::http::server::session::SessionStateV2; use crate::http::state::ClientInfo; use bytes::Bytes; use http_body_util::Full; @@ -23,7 +28,31 @@ use std::net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr}; use std::num::NonZeroUsize; use std::ops::Deref; use std::sync::Arc; -use tokio::sync::{oneshot, Mutex}; +use tokio::sync::{mpsc, oneshot, Mutex}; + +/// Configuration for the v2 (legacy) protocol endpoints. +pub struct ServerConfigV2 { + /// Optional PIN that senders must provide via the `pin` query parameter. + pub pin: Option, + + /// Channel on which the server emits events that must be handled by the application. + pub event_tx: mpsc::Sender, +} + +/// Runtime state of the v2 protocol endpoints. +pub(crate) struct V2State { + /// Optional PIN required for prepare-upload requests. + pub(crate) pin: Option, + + /// Channel on which server events are emitted to the application. + pub(crate) event_tx: mpsc::Sender, + + /// The single upload session slot. Only one session can be active at a time. + pub(crate) session: Mutex>, + + /// Maps client IPs to the number of failed PIN attempts. + pub(crate) pin_attempts: Mutex>, +} #[derive(Clone)] struct AppState { @@ -38,10 +67,13 @@ struct AppState { /// Maps client identifiers to nonces that are expected to be received from remote. generated_nonce_map: Arc>>>, + + /// State of the v2 protocol endpoints. `None` disables the v2 routes. + v2: Option>, } impl AppState { - fn new(info: Arc>) -> Self { + fn new(info: Arc>, v2_config: Option) -> Self { Self { info, web: Arc::new(Mutex::new(None)), @@ -51,6 +83,14 @@ impl AppState { generated_nonce_map: Arc::new(Mutex::new(LruCache::new( NonZeroUsize::new(200).unwrap(), ))), + v2: v2_config.map(|config| { + Arc::new(V2State { + pin: config.pin, + event_tx: config.event_tx, + session: Mutex::new(None), + pin_attempts: Mutex::new(LruCache::new(NonZeroUsize::new(200).unwrap())), + }) + }), } } } @@ -60,28 +100,37 @@ pub async fn start_with_port( port: u16, tls_config: Option, info: ClientInfo, - legacy_enabled: bool, + v2_config: Option, stop_rx: oneshot::Receiver<()>, ) -> anyhow::Result<()> { let ipv4_socket_addr = SocketAddr::new(Ipv4Addr::UNSPECIFIED.into(), port); let ipv6_socket_addr = SocketAddr::new(Ipv6Addr::UNSPECIFIED.into(), port); let info = Arc::new(Mutex::new(info)); - let state = AppState::new(info.clone()); + let state = AppState::new(info.clone(), v2_config); + + let ipv4_listener = tokio::net::TcpListener::bind(ipv4_socket_addr).await?; + let ipv6_listener = match bind_ipv6_only(ipv6_socket_addr) { + Ok(listener) => Some(listener), + Err(err) => { + tracing::warn!("Failed to start server on {}: {err:#}", ipv6_socket_addr); + None + } + }; tokio::spawn({ let state = state.clone(); async move { tokio::select! { - _ = start_server_with_addr(ipv4_socket_addr, tls_config.clone(), state.clone(), legacy_enabled) => { + _ = start_server_with_listener(ipv4_listener, tls_config.clone(), state.clone()) => { tracing::info!("Server stopped on: {}", ipv4_socket_addr); } _ = async { - if start_server_with_addr(ipv6_socket_addr, tls_config, state, legacy_enabled).await.is_err() { - tracing::warn!("Failed to start server on: {}", ipv6_socket_addr); - - // Keep the future running forever, so we continue using "ipv4 only" even if ipv6 fails. - tokio::time::sleep(std::time::Duration::from_secs(u64::MAX)).await; + if let Some(listener) = ipv6_listener { + let _ = start_server_with_listener(listener, tls_config, state).await; } + + // Keep the future running forever, so we continue using "ipv4 only" even if ipv6 fails. + tokio::time::sleep(std::time::Duration::from_secs(u64::MAX)).await; } => {} _ = stop_rx => {} } @@ -91,22 +140,38 @@ pub async fn start_with_port( Ok(()) } +/// Binds an IPv6 listener with `IPV6_V6ONLY` enabled. +/// +/// Without this flag, some systems (e.g. macOS) bind IPv6 wildcard sockets in +/// dual-stack mode, which conflicts with the separate IPv4 listener on the same port. +fn bind_ipv6_only(socket_addr: SocketAddr) -> anyhow::Result { + let socket = socket2::Socket::new( + socket2::Domain::IPV6, + socket2::Type::STREAM, + Some(socket2::Protocol::TCP), + )?; + socket.set_only_v6(true)?; + #[cfg(not(windows))] + socket.set_reuse_address(true)?; + socket.set_nonblocking(true)?; + socket.bind(&socket_addr.into())?; + socket.listen(1024)?; + Ok(tokio::net::TcpListener::from_std(socket.into())?) +} + #[derive(Clone, Debug)] pub struct TlsConfig { pub cert: String, pub private_key: String, } -async fn start_server_with_addr( - socket_addr: SocketAddr, +async fn start_server_with_listener( + incoming: tokio::net::TcpListener, tls_config: Option, app_state: AppState, - legacy_enabled: bool, ) -> anyhow::Result<()> { let _ = rustls::crypto::ring::default_provider().install_default(); - let incoming = tokio::net::TcpListener::bind(socket_addr).await?; - let tls_acceptor = match tls_config { Some(tls_config) => Some(create_tls_config(&tls_config).inspect_err(|err| { tracing::error!("failed to create tls config: {err:#}"); @@ -116,7 +181,7 @@ async fn start_server_with_addr( tracing::info!( "Started server on {} (TLS: {})", - socket_addr, + incoming.local_addr()?, tls_acceptor.is_some() ); @@ -155,7 +220,7 @@ async fn start_server_with_addr( req.extensions_mut() .insert::(client_info.clone()); req.extensions_mut().insert::(app_state.clone()); - handle_request(req, legacy_enabled) + handle_request(req) }), ) .await @@ -172,7 +237,7 @@ async fn start_server_with_addr( }, ); req.extensions_mut().insert::(app_state.clone()); - handle_request(req, legacy_enabled) + handle_request(req) }), ) .await @@ -229,21 +294,15 @@ impl RequestClientInfo { } } -async fn handle_request( - req: Request, - legacy_enabled: bool, -) -> Result>, hyper::Error> { - Ok(handle_request_inner(req, legacy_enabled) - .await - .unwrap_or_else(|err| { - tracing::error!("Error handling request: {err:?}"); - err.to_response() - })) +async fn handle_request(req: Request) -> Result>, hyper::Error> { + Ok(handle_request_inner(req).await.unwrap_or_else(|err| { + tracing::error!("Error handling request: {err:?}"); + err.to_response() + })) } async fn handle_request_inner( mut req: Request, - legacy_enabled: bool, ) -> Result>, AppError> { let Some(state) = req.extensions_mut().remove::() else { return Err(AppError::Status(StatusCode::INTERNAL_SERVER_ERROR)); @@ -253,9 +312,11 @@ async fn handle_request_inner( return Err(AppError::Status(StatusCode::INTERNAL_SERVER_ERROR)); }; + let v2_enabled = state.v2.is_some(); + match (req.method(), req.uri().path()) { (&Method::POST, "/api/localsend/v2/register") => { - if !legacy_enabled { + if !v2_enabled { return Err(AppError::Status(StatusCode::NOT_FOUND)); } @@ -265,38 +326,33 @@ async fn handle_request_inner( .into_response(), ) } + (&Method::GET, "/api/localsend/v2/info") => { + if !v2_enabled { + return Err(AppError::Status(StatusCode::NOT_FOUND)); + } + + Ok(controller::v2::info(state).await?.into_response()) + } (&Method::POST, "/api/localsend/v2/prepare-upload") => { - if !legacy_enabled { + if !v2_enabled { return Err(AppError::Status(StatusCode::NOT_FOUND)); } - Ok( - controller::v2::register(req.into_body(), state, client_info) - .await? - .into_response(), - ) + controller::v2::prepare_upload(req, state, client_info).await } (&Method::POST, "/api/localsend/v2/upload") => { - if !legacy_enabled { + if !v2_enabled { return Err(AppError::Status(StatusCode::NOT_FOUND)); } - Ok( - controller::v2::register(req.into_body(), state, client_info) - .await? - .into_response(), - ) + controller::v2::upload(req, state, client_info).await } (&Method::POST, "/api/localsend/v2/cancel") => { - if !legacy_enabled { + if !v2_enabled { return Err(AppError::Status(StatusCode::NOT_FOUND)); } - Ok( - controller::v2::register(req.into_body(), state, client_info) - .await? - .into_response(), - ) + controller::v2::cancel(req, state).await } (&Method::POST, "/api/localsend/v3/nonce") => { Ok( diff --git a/core/src/http/server/query.rs b/core/src/http/server/query.rs new file mode 100644 index 00000000..b26b4fea --- /dev/null +++ b/core/src/http/server/query.rs @@ -0,0 +1,45 @@ +use std::collections::HashMap; + +/// Parses a URL query string into a key-value map with percent-decoding. +pub(crate) fn parse_query(query: Option<&str>) -> HashMap { + let Some(query) = query else { + return HashMap::new(); + }; + + form_urlencoded::parse(query.as_bytes()) + .map(|(key, value)| (key.into_owned(), value.into_owned())) + .collect() +} + +#[cfg(test)] +mod tests { + use super::parse_query; + + #[test] + fn test_parse_query() { + let query = parse_query(Some("sessionId=abc&fileId=some%20file&token=a%2Bb+c")); + assert_eq!(query.get("sessionId").unwrap(), "abc"); + assert_eq!(query.get("fileId").unwrap(), "some file"); + assert_eq!(query.get("token").unwrap(), "a+b c"); + } + + #[test] + fn test_parse_query_empty() { + assert!(parse_query(None).is_empty()); + assert!(parse_query(Some("")).is_empty()); + } + + #[test] + fn test_parse_query_no_value() { + let query = parse_query(Some("flag&pin=123456")); + assert_eq!(query.get("flag").unwrap(), ""); + assert_eq!(query.get("pin").unwrap(), "123456"); + } + + #[test] + fn test_parse_query_invalid_percent() { + let query = parse_query(Some("a=%zz&b=%4")); + assert_eq!(query.get("a").unwrap(), "%zz"); + assert_eq!(query.get("b").unwrap(), "%4"); + } +} \ No newline at end of file diff --git a/core/src/http/server/session.rs b/core/src/http/server/session.rs new file mode 100644 index 00000000..d0530905 --- /dev/null +++ b/core/src/http/server/session.rs @@ -0,0 +1,48 @@ +use crate::model::transfer::FileDto; +use std::collections::HashMap; +use std::net::IpAddr; + +/// State of the single v2 upload session slot. +pub(crate) enum SessionStateV2 { + /// A prepare-upload request is waiting for the application's decision. + Pending, + + /// An accepted upload session. + Active(UploadSessionV2), +} + +pub(crate) struct UploadSessionV2 { + pub(crate) session_id: String, + + /// The IP address of the sender. Uploads are only accepted from this address. + pub(crate) sender_ip: IpAddr, + + /// The accepted files, mapped by file ID. + pub(crate) files: HashMap, +} + +impl UploadSessionV2 { + /// Whether all files reached a final state. + pub(crate) fn is_complete(&self) -> bool { + self.files + .values() + .all(|file| matches!(file.status, FileStatusV2::Finished | FileStatusV2::Failed)) + } +} + +pub(crate) struct SessionFileV2 { + pub(crate) dto: FileDto, + + /// The file-specific token required for the upload request. + pub(crate) token: String, + + pub(crate) status: FileStatusV2, +} + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub(crate) enum FileStatusV2 { + Pending, + InProgress, + Finished, + Failed, +} diff --git a/core/src/main.rs b/core/src/main.rs index 994461e3..99113945 100644 --- a/core/src/main.rs +++ b/core/src/main.rs @@ -7,7 +7,8 @@ mod webrtc; use crate::crypto::token; use crate::http::client::LsHttpClientV3; use crate::http::dto::{PrepareUploadRequestDto, ProtocolType, RegisterDto}; -use crate::http::server::TlsConfig; +use crate::http::server::event::{PrepareUploadDecisionV2, ServerEventV2}; +use crate::http::server::{ServerConfigV2, TlsConfig}; use crate::model::discovery::DeviceType; use crate::webrtc::signaling::{ClientInfo, WsServerMessage}; use crate::webrtc::webrtc::{PinConfig, RTCFile, RTCFileError, RTCSendFileResponse, RTCStatus}; @@ -138,6 +139,57 @@ async fn server_test() -> Result<()> { }; let (stop_tx, stop_rx) = oneshot::channel::<()>(); + let (event_tx, mut event_rx) = mpsc::channel::(16); + + tokio::spawn(async move { + while let Some(event) = event_rx.recv().await { + match event { + ServerEventV2::Register { ip, info } => { + tracing::info!("Device registered from {ip}: {}", info.alias); + } + ServerEventV2::PrepareUpload { + info, + files, + decision_tx, + .. + } => { + tracing::info!( + "Prepare upload from {}: {} file(s)", + info.alias, + files.len() + ); + + // Accept all files. + let _ = decision_tx.send(PrepareUploadDecisionV2::Accept( + files.keys().cloned().collect(), + )); + } + ServerEventV2::FileUpload { + file, + mut binary_rx, + result_tx, + .. + } => { + tokio::spawn(async move { + let mut received: u64 = 0; + while let Some(chunk) = binary_rx.recv().await { + received += chunk.len() as u64; + } + tracing::info!( + "Received {}/{} bytes of {}", + received, + file.size, + file.file_name + ); + let _ = result_tx.send(Ok(())); + }); + } + ServerEventV2::SessionEnd { session_id, reason } => { + tracing::info!("Session {session_id} ended: {reason:?}"); + } + } + } + }); http::server::start_with_port( 53317, @@ -146,7 +198,10 @@ async fn server_test() -> Result<()> { private_key: PRIVATE_KEY.to_string(), }), client_info, - true, + Some(ServerConfigV2 { + pin: None, + event_tx, + }), stop_rx, ) .await?; @@ -218,7 +273,10 @@ async fn client_test() -> Result<()> { ) .await?; - println!("Prepare Upload Response: {:?}", prepare_upload_response.response); + println!( + "Prepare Upload Response: {:?}", + prepare_upload_response.response + ); Ok(()) } diff --git a/core/tests/v2_server.rs b/core/tests/v2_server.rs new file mode 100644 index 00000000..eb062929 --- /dev/null +++ b/core/tests/v2_server.rs @@ -0,0 +1,551 @@ +#![cfg(feature = "http")] + +use localsend::http::client::{ClientError, LsHttpClientV2}; +use localsend::http::dto::ProtocolType; +use localsend::http::dto_v2::{PrepareUploadRequestDtoV2, ProtocolTypeV2, RegisterDtoV2}; +use localsend::http::server::event::{PrepareUploadDecisionV2, ServerEventV2, SessionEndReasonV2}; +use localsend::http::server::{start_with_port, ServerConfigV2}; +use localsend::http::state::ClientInfo; +use localsend::model::transfer::FileDto; +use std::collections::HashMap; +use std::sync::atomic::{AtomicU16, Ordering}; +use std::sync::Arc; +use std::time::Duration; +use tokio::sync::{mpsc, oneshot, Mutex}; + +struct TestServer { + port: u16, + /// Uploaded file contents, mapped by file ID. + received: Arc>>>, + /// Ended sessions with their reasons. + session_ends: Arc>>, + _stop_tx: oneshot::Sender<()>, +} + +async fn start_test_server(pin: Option, accept: bool) -> TestServer { + let _ = tracing_subscriber::fmt().with_test_writer().try_init(); + let port = free_port(); + let received: Arc>>> = Arc::new(Mutex::new(HashMap::new())); + let session_ends: Arc>> = + Arc::new(Mutex::new(Vec::new())); + + let (event_tx, mut event_rx) = mpsc::channel::(16); + + tokio::spawn({ + let received = received.clone(); + let session_ends = session_ends.clone(); + async move { + while let Some(event) = event_rx.recv().await { + match event { + ServerEventV2::Register { .. } => {} + ServerEventV2::PrepareUpload { + files, decision_tx, .. + } => { + let decision = match accept { + true => { + PrepareUploadDecisionV2::Accept(files.keys().cloned().collect()) + } + false => PrepareUploadDecisionV2::Decline, + }; + let _ = decision_tx.send(decision); + } + ServerEventV2::FileUpload { + file_id, + mut binary_rx, + result_tx, + .. + } => { + let received = received.clone(); + tokio::spawn(async move { + let mut bytes = Vec::new(); + while let Some(chunk) = binary_rx.recv().await { + bytes.extend_from_slice(&chunk); + } + received.lock().await.insert(file_id, bytes); + let _ = result_tx.send(Ok(())); + }); + } + ServerEventV2::SessionEnd { session_id, reason } => { + session_ends.lock().await.push((session_id, reason)); + } + } + } + } + }); + + let (stop_tx, stop_rx) = oneshot::channel::<()>(); + + start_with_port( + port, + None, // plain HTTP + ClientInfo { + alias: "Test Server".to_string(), + version: "2.1".to_string(), + device_model: Some("Rust".to_string()), + device_type: None, + token: "server-fingerprint".to_string(), + }, + Some(ServerConfigV2 { pin, event_tx }), + stop_rx, + ) + .await + .expect("Failed to start server"); + + wait_until_reachable(port).await; + + TestServer { + port, + received, + session_ends, + _stop_tx: stop_tx, + } +} + +/// Returns a free port. +/// +/// A counter is used instead of binding to port 0 because the OS may hand out +/// the same just-freed ephemeral port to multiple tests running in parallel. +fn free_port() -> u16 { + static PORT_COUNTER: AtomicU16 = AtomicU16::new(40551); + + loop { + let port = PORT_COUNTER.fetch_add(1, Ordering::SeqCst); + if std::net::TcpListener::bind(("127.0.0.1", port)).is_ok() { + return port; + } + } +} + +async fn wait_until_reachable(port: u16) { + for _ in 0..100 { + if tokio::net::TcpStream::connect(("127.0.0.1", port)) + .await + .is_ok() + { + return; + } + tokio::time::sleep(Duration::from_millis(20)).await; + } + panic!("Server did not become reachable on port {port}"); +} + +fn sender_info() -> RegisterDtoV2 { + RegisterDtoV2 { + alias: "Test Sender".to_string(), + version: "2.1".to_string(), + device_model: Some("Rust".to_string()), + device_type: None, + fingerprint: "sender-fingerprint".to_string(), + port: 53317, + protocol: ProtocolTypeV2::Http, + download: false, + } +} + +fn file_dto(id: &str, name: &str, size: u64) -> FileDto { + FileDto { + id: id.to_string(), + file_name: name.to_string(), + size, + file_type: "application/octet-stream".to_string(), + sha256: None, + preview: None, + metadata: None, + } +} + +fn prepare_upload_request(files: &[FileDto]) -> PrepareUploadRequestDtoV2 { + PrepareUploadRequestDtoV2 { + info: sender_info(), + files: files + .iter() + .map(|file| (file.id.clone(), file.clone())) + .collect(), + } +} + +async fn upload_bytes( + client: &LsHttpClientV2, + port: u16, + session_id: &str, + file_id: &str, + token: &str, + bytes: &[u8], +) -> Result<(), ClientError> { + let (tx, rx) = mpsc::channel::>(4); + let chunks: Vec> = bytes.chunks(1024).map(|chunk| chunk.to_vec()).collect(); + tokio::spawn(async move { + for chunk in chunks { + if tx.send(chunk).await.is_err() { + break; + } + } + }); + + client + .upload( + ProtocolType::Http, + "127.0.0.1", + port, + None, + session_id, + file_id, + token, + rx, + ) + .await +} + +fn assert_status(result: Result, expected_status: u16) { + match result { + Err(ClientError::StatusCode(err)) => assert_eq!(err.status, expected_status), + Err(err) => panic!("Expected status code {expected_status}, got error: {err:?}"), + Ok(_) => panic!("Expected status code {expected_status}, got success"), + } +} + +#[tokio::test] +async fn test_register_and_info() { + let server = start_test_server(None, true).await; + let client = LsHttpClientV2::try_new_without_cert().unwrap(); + + let response = client + .register(ProtocolType::Http, "127.0.0.1", server.port, sender_info()) + .await + .unwrap(); + assert_eq!(response.body.alias, "Test Server"); + assert_eq!(response.body.fingerprint, "server-fingerprint"); + assert!(!response.body.download); + + let info = client + .info(ProtocolType::Http, "127.0.0.1", server.port) + .await + .unwrap(); + assert_eq!(info.alias, "Test Server"); + assert_eq!(info.fingerprint, "server-fingerprint"); +} + +#[tokio::test] +async fn test_register_over_ipv6() { + let server = start_test_server(None, true).await; + let client = LsHttpClientV2::try_new_without_cert().unwrap(); + + let response = client + .register(ProtocolType::Http, "::1", server.port, sender_info()) + .await + .unwrap(); + assert_eq!(response.body.alias, "Test Server"); +} + +#[tokio::test] +async fn test_full_upload_flow() { + let server = start_test_server(None, true).await; + let client = LsHttpClientV2::try_new_without_cert().unwrap(); + + let file_a = file_dto("file-a", "a.bin", 100_000); + let file_b = file_dto("file-b", "b.bin", 5); + + let result = client + .prepare_upload( + ProtocolType::Http, + "127.0.0.1", + server.port, + None, + prepare_upload_request(&[file_a.clone(), file_b.clone()]), + None, + ) + .await + .unwrap(); + + assert_eq!(result.status_code, 200); + let response = result.response.unwrap(); + assert_eq!(response.files.len(), 2); + + let bytes_a: Vec = (0..100_000u32).map(|i| i as u8).collect(); + let bytes_b = b"hello".to_vec(); + + upload_bytes( + &client, + server.port, + &response.session_id, + "file-a", + &response.files["file-a"], + &bytes_a, + ) + .await + .unwrap(); + + upload_bytes( + &client, + server.port, + &response.session_id, + "file-b", + &response.files["file-b"], + &bytes_b, + ) + .await + .unwrap(); + + let received = server.received.lock().await; + assert_eq!(received["file-a"], bytes_a); + assert_eq!(received["file-b"], bytes_b); + drop(received); + + // The session should have ended after all files were uploaded. + tokio::time::sleep(Duration::from_millis(100)).await; + let session_ends = server.session_ends.lock().await; + assert_eq!( + *session_ends, + vec![(response.session_id.clone(), SessionEndReasonV2::Finished)] + ); + + // The session is gone, so uploading again is rejected. + let result = upload_bytes( + &client, + server.port, + &response.session_id, + "file-a", + &response.files["file-a"], + b"again", + ) + .await; + assert_status(result, 403); +} + +#[tokio::test] +async fn test_upload_with_invalid_token() { + let server = start_test_server(None, true).await; + let client = LsHttpClientV2::try_new_without_cert().unwrap(); + + let file = file_dto("file-a", "a.bin", 5); + let response = client + .prepare_upload( + ProtocolType::Http, + "127.0.0.1", + server.port, + None, + prepare_upload_request(&[file]), + None, + ) + .await + .unwrap() + .response + .unwrap(); + + let result = upload_bytes( + &client, + server.port, + &response.session_id, + "file-a", + "wrong-token", + b"hello", + ) + .await; + assert_status(result, 403); + + // The correct token still works afterwards. + upload_bytes( + &client, + server.port, + &response.session_id, + "file-a", + &response.files["file-a"], + b"hello", + ) + .await + .unwrap(); +} + +#[tokio::test] +async fn test_upload_missing_parameters() { + let server = start_test_server(None, true).await; + + let response = localsend::reqwest::Client::new() + .post(format!( + "http://127.0.0.1:{}/api/localsend/v2/upload?sessionId=abc", + server.port + )) + .body("data") + .send() + .await + .unwrap(); + assert_eq!(response.status().as_u16(), 400); +} + +#[tokio::test] +async fn test_second_session_blocked_and_cancel() { + let server = start_test_server(None, true).await; + let client = LsHttpClientV2::try_new_without_cert().unwrap(); + + let file = file_dto("file-a", "a.bin", 5); + let response = client + .prepare_upload( + ProtocolType::Http, + "127.0.0.1", + server.port, + None, + prepare_upload_request(&[file.clone()]), + None, + ) + .await + .unwrap() + .response + .unwrap(); + + // A second session is blocked while the first one is active. + let result = client + .prepare_upload( + ProtocolType::Http, + "127.0.0.1", + server.port, + None, + prepare_upload_request(&[file.clone()]), + None, + ) + .await; + assert_status(result, 409); + + client + .cancel( + ProtocolType::Http, + "127.0.0.1", + server.port, + &response.session_id, + ) + .await + .unwrap(); + + tokio::time::sleep(Duration::from_millis(100)).await; + let session_ends = server.session_ends.lock().await.clone(); + assert_eq!( + session_ends, + vec![(response.session_id.clone(), SessionEndReasonV2::Cancelled)] + ); + + // After cancelling, a new session can be created. + client + .prepare_upload( + ProtocolType::Http, + "127.0.0.1", + server.port, + None, + prepare_upload_request(&[file]), + None, + ) + .await + .unwrap(); +} + +#[tokio::test] +async fn test_prepare_upload_declined() { + let server = start_test_server(None, false).await; + let client = LsHttpClientV2::try_new_without_cert().unwrap(); + + let file = file_dto("file-a", "a.bin", 5); + let result = client + .prepare_upload( + ProtocolType::Http, + "127.0.0.1", + server.port, + None, + prepare_upload_request(&[file.clone()]), + None, + ) + .await; + assert_status(result, 403); + + // A declined request must not block subsequent sessions. + let result = client + .prepare_upload( + ProtocolType::Http, + "127.0.0.1", + server.port, + None, + prepare_upload_request(&[file]), + None, + ) + .await; + assert_status(result, 403); +} + +#[tokio::test] +async fn test_pin() { + let server = start_test_server(Some("123456".to_string()), true).await; + let client = LsHttpClientV2::try_new_without_cert().unwrap(); + + let file = file_dto("file-a", "a.bin", 5); + + // Missing PIN. + let result = client + .prepare_upload( + ProtocolType::Http, + "127.0.0.1", + server.port, + None, + prepare_upload_request(&[file.clone()]), + None, + ) + .await; + assert_status(result, 401); + + // Wrong PIN. + let result = client + .prepare_upload( + ProtocolType::Http, + "127.0.0.1", + server.port, + None, + prepare_upload_request(&[file.clone()]), + Some("000000"), + ) + .await; + assert_status(result, 401); + + // Correct PIN. + client + .prepare_upload( + ProtocolType::Http, + "127.0.0.1", + server.port, + None, + prepare_upload_request(&[file]), + Some("123456"), + ) + .await + .unwrap(); +} + +#[tokio::test] +async fn test_pin_too_many_attempts() { + let server = start_test_server(Some("123456".to_string()), true).await; + let client = LsHttpClientV2::try_new_without_cert().unwrap(); + + let file = file_dto("file-a", "a.bin", 5); + + for _ in 0..3 { + let result = client + .prepare_upload( + ProtocolType::Http, + "127.0.0.1", + server.port, + None, + prepare_upload_request(&[file.clone()]), + Some("000000"), + ) + .await; + assert_status(result, 401); + } + + // Blocked even with the correct PIN. + let result = client + .prepare_upload( + ProtocolType::Http, + "127.0.0.1", + server.port, + None, + prepare_upload_request(&[file]), + Some("123456"), + ) + .await; + assert_status(result, 429); +}