feat: add v2 http server

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",

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"]

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<S: Serializer>(
+        value: &Option<DeviceType>,
+        serializer: S,
+    ) -> Result<S::Ok, S::Error> {
+        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<Option<DeviceType>, D::Error> {
+        let value = Option::<String>::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<String>,
 
     /// 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<DeviceType>,
 
     /// Fingerprint for device identification.
@@ -79,7 +125,11 @@ pub struct RegisterDtoV2 {
     pub device_model: Option<String>,
 
     /// 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<DeviceType>,
 
     /// Fingerprint for device identification.
@@ -114,7 +164,11 @@ pub struct RegisterResponseDtoV2 {
     pub device_model: Option<String>,
 
     /// 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<DeviceType>,
 
     /// Fingerprint for device identification.
@@ -194,7 +248,11 @@ pub struct InfoResponseDtoV2 {
     pub device_model: Option<String>,
 
     /// 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<DeviceType>,
 
     /// 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

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<JsonResponse<RegisterResponseDto>, AppError> {
-    let payload = body.collect_to_json::<RegisterDto>().await?;
+) -> Result<JsonResponse<RegisterResponseDtoV2>, AppError> {
+    let payload = body.collect_to_json::<RegisterDtoV2>().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<JsonResponse<InfoResponseDtoV2>, 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<Incoming>,
+    state: AppState,
+    client_info: RequestClientInfo,
+) -> Result<Response<Full<Bytes>>, 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::<PrepareUploadRequestDtoV2>()
+        .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<String, SessionFileV2> = 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<String, String> = 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<Incoming>,
+    state: AppState,
+    client_info: RequestClientInfo,
+) -> Result<Response<Full<Bytes>>, 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::<Bytes>(UPLOAD_CHANNEL_CAPACITY);
+    let (result_tx, result_rx) = oneshot::channel::<Result<(), String>>();
+
+    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<Incoming>,
+    state: AppState,
+) -> Result<Response<Full<Bytes>>, 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<Arc<V2State>, 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<V2State>,
+    query: &HashMap<String, String>,
+    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<V2State>,
+    armed: bool,
+}
+
+impl PendingSessionGuard {
+    fn new(v2: Arc<V2State>) -> 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<V2State>,
+    session_id: String,
+    file_id: String,
+    armed: bool,
+}
+
+impl UploadGuard {
+    fn new(v2: Arc<V2State>, 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;
+    }
+}

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()

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<String, FileDto>,
+
+        /// Channel to send the decision (accept all, a subset, or decline).
+        decision_tx: oneshot::Sender<PrepareUploadDecisionV2>,
+    },
+
+    /// 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<Bytes>,
+
+        /// Channel to report whether the file was processed successfully.
+        result_tx: oneshot::Sender<Result<(), String>>,
+    },
+
+    /// 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<String>),
+
+    /// 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,
+}

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<String>,
+
+    /// Channel on which the server emits events that must be handled by the application.
+    pub event_tx: mpsc::Sender<ServerEventV2>,
+}
+
+/// Runtime state of the v2 protocol endpoints.
+pub(crate) struct V2State {
+    /// Optional PIN required for prepare-upload requests.
+    pub(crate) pin: Option<String>,
+
+    /// Channel on which server events are emitted to the application.
+    pub(crate) event_tx: mpsc::Sender<ServerEventV2>,
+
+    /// The single upload session slot. Only one session can be active at a time.
+    pub(crate) session: Mutex<Option<SessionStateV2>>,
+
+    /// Maps client IPs to the number of failed PIN attempts.
+    pub(crate) pin_attempts: Mutex<LruCache<IpAddr, u32>>,
+}
 
 #[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<Mutex<LruCache<String, Vec<u8>>>>,
+
+    /// State of the v2 protocol endpoints. `None` disables the v2 routes.
+    v2: Option<Arc<V2State>>,
 }
 
 impl AppState {
-    fn new(info: Arc<Mutex<ClientInfo>>) -> Self {
+    fn new(info: Arc<Mutex<ClientInfo>>, v2_config: Option<ServerConfigV2>) -> 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<TlsConfig>,
     info: ClientInfo,
-    legacy_enabled: bool,
+    v2_config: Option<ServerConfigV2>,
     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<tokio::net::TcpListener> {
+    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<TlsConfig>,
     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::<RequestClientInfo>(client_info.clone());
                                 req.extensions_mut().insert::<AppState>(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::<AppState>(app_state.clone());
-                                handle_request(req, legacy_enabled)
+                                handle_request(req)
                             }),
                         )
                         .await
@@ -229,21 +294,15 @@ impl RequestClientInfo {
     }
 }
 
-async fn handle_request(
-    req: Request<Incoming>,
-    legacy_enabled: bool,
-) -> Result<Response<Full<Bytes>>, 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<Incoming>) -> Result<Response<Full<Bytes>>, 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<Incoming>,
-    legacy_enabled: bool,
 ) -> Result<Response<Full<Bytes>>, AppError> {
     let Some(state) = req.extensions_mut().remove::<AppState>() 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(

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<String, String> {
+    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");
+    }
+}

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<String, SessionFileV2>,
+}
+
+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,
+}

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::<ServerEventV2>(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(())
 }

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<Mutex<HashMap<String, Vec<u8>>>>,
+    /// Ended sessions with their reasons.
+    session_ends: Arc<Mutex<Vec<(String, SessionEndReasonV2)>>>,
+    _stop_tx: oneshot::Sender<()>,
+}
+
+async fn start_test_server(pin: Option<String>, accept: bool) -> TestServer {
+    let _ = tracing_subscriber::fmt().with_test_writer().try_init();
+    let port = free_port();
+    let received: Arc<Mutex<HashMap<String, Vec<u8>>>> = Arc::new(Mutex::new(HashMap::new()));
+    let session_ends: Arc<Mutex<Vec<(String, SessionEndReasonV2)>>> =
+        Arc::new(Mutex::new(Vec::new()));
+
+    let (event_tx, mut event_rx) = mpsc::channel::<ServerEventV2>(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::<Vec<u8>>(4);
+    let chunks: Vec<Vec<u8>> = 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<impl Sized, ClientError>, 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<u8> = (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);
+}