lanflare_core/server/
handlers.rs

1//! HTTP request handlers for LocalSend v2.2 endpoints.
2
3use std::collections::HashMap;
4use std::net::IpAddr;
5use std::sync::Arc;
6use bytes::Bytes;
7use http_body_util::{BodyExt, Full, Limited};
8use hyper::body::Incoming;
9use hyper::{Request, Response, StatusCode};
10use lanflare_proto::discovery::{InfoResponseDtoV2, ProtocolType, RegisterDtoV2, RegisterResponseDtoV2, PROTOCOL_VERSION_V2};
11use lanflare_proto::transfer::{
12    PrepareDownloadResponseDtoV2, PrepareUploadRequestDtoV2, PrepareUploadResponseDtoV2,
13};
14use tokio::sync::{mpsc, oneshot};
15use tokio_util::sync::CancellationToken;
16use uuid::Uuid;
17
18use crate::config::AppConfig;
19use crate::discovery::registry::PeerRegistry;
20use crate::server::session::{
21    DownloadManager, FileTransferStatus, PrepareUploadDecision, SessionEndReason, SessionFile,
22    SessionManager,
23};
24use crate::storage::path::{resolve_unique_destination_path, sanitize_filename};
25use crate::storage::writer::{StorageCommitResult, StreamingFileWriter};
26
27/// Application events emitted by the HTTP server for GUI/TUI integration.
28#[derive(Debug)]
29pub enum ServerTransferEvent {
30    /// Peer registration received.
31    DeviceRegistered {
32        ip: IpAddr,
33        info: RegisterDtoV2,
34    },
35    /// An incoming upload preparation request.
36    PrepareUpload {
37        session_id: String,
38        sender_ip: IpAddr,
39        sender_info: RegisterDtoV2,
40        cert_fingerprint: Option<String>,
41        files: HashMap<String, lanflare_proto::transfer::FileDto>,
42        decision_tx: oneshot::Sender<PrepareUploadDecision>,
43    },
44    /// A file upload started.
45    FileStarted {
46        session_id: String,
47        file_id: String,
48        file_name: String,
49        total_bytes: u64,
50    },
51    /// A file upload completed and verified.
52    FileCompleted {
53        session_id: String,
54        file_id: String,
55        result: StorageCommitResult,
56    },
57    /// A file upload failed.
58    FileFailed {
59        session_id: String,
60        file_id: String,
61        error: String,
62    },
63    /// The transfer session ended.
64    SessionEnded {
65        session_id: String,
66        reason: SessionEndReason,
67    },
68}
69
70/// Shared application state injected into hyper request handlers.
71#[derive(Clone)]
72pub struct ServerState {
73    pub config: Arc<AppConfig>,
74    pub cert_fingerprint: String,
75    pub session_manager: Arc<SessionManager>,
76    pub download_manager: Arc<DownloadManager>,
77    pub registry: Arc<PeerRegistry>,
78    pub event_tx: Option<mpsc::Sender<ServerTransferEvent>>,
79}
80
81/// Helper creating an empty hyper response with given status code.
82pub fn empty_response(status: StatusCode) -> Response<Full<Bytes>> {
83    Response::builder()
84        .status(status)
85        .body(Full::new(Bytes::new()))
86        .unwrap()
87}
88
89/// Helper creating a JSON hyper response with given status code.
90pub fn json_response<T: serde::Serialize>(status: StatusCode, data: &T) -> Response<Full<Bytes>> {
91    let body = match serde_json::to_vec(data) {
92        Ok(bytes) => bytes,
93        Err(_) => {
94            return empty_response(StatusCode::INTERNAL_SERVER_ERROR);
95        }
96    };
97
98    Response::builder()
99        .status(status)
100        .header(hyper::header::CONTENT_TYPE, "application/json")
101        .body(Full::new(Bytes::from(body)))
102        .unwrap()
103}
104
105/// Maximum allowed size for JSON control bodies (10 MB).
106pub const MAX_JSON_BODY_SIZE: usize = 10 * 1024 * 1024;
107
108/// Helper to read and deserialize a bounded JSON request body, preventing memory exhaustion attacks.
109pub async fn read_json_body<T: serde::de::DeserializeOwned>(
110    body: Incoming,
111) -> Result<T, StatusCode> {
112    let limited = Limited::new(body, MAX_JSON_BODY_SIZE);
113    let collected = match limited.collect().await {
114        Ok(c) => c.to_bytes(),
115        Err(err) => {
116            if err.downcast_ref::<http_body_util::LengthLimitError>().is_some() {
117                return Err(StatusCode::PAYLOAD_TOO_LARGE);
118            }
119            return Err(StatusCode::BAD_REQUEST);
120        }
121    };
122    serde_json::from_slice(&collected).map_err(|_| StatusCode::BAD_REQUEST)
123}
124
125/// Helper parsing query parameters into a HashMap.
126pub fn parse_query(query_str: Option<&str>) -> HashMap<String, String> {
127    let mut map = HashMap::new();
128    let Some(query) = query_str else {
129        return map;
130    };
131
132    for pair in query.split('&') {
133        if pair.is_empty() {
134            continue;
135        }
136        let mut parts = pair.splitn(2, '=');
137        let key = parts.next().unwrap_or("").to_string();
138        let value = parts.next().unwrap_or("").to_string();
139        if !key.is_empty() {
140            map.insert(key, value);
141        }
142    }
143    map
144}
145
146/// Handler for `GET /api/localsend/v2/info` and `GET /api/localsend/v1/info`.
147pub async fn handle_info(state: &ServerState) -> Response<Full<Bytes>> {
148    let files = state.download_manager.get_file_dtos().await;
149    let resp = InfoResponseDtoV2 {
150        alias: state.config.alias.clone(),
151        version: PROTOCOL_VERSION_V2.to_string(),
152        device_model: Some(std::env::consts::OS.to_string()),
153        device_type: Some(lanflare_proto::discovery::DeviceType::Desktop),
154        fingerprint: state.cert_fingerprint.clone(),
155        port: Some(state.config.port),
156        protocol: Some(ProtocolType::Https),
157        download: !files.is_empty(),
158    };
159    json_response(StatusCode::OK, &resp)
160}
161
162/// Handler for `POST /api/localsend/v2/register`.
163pub async fn handle_register(
164    req: Request<Incoming>,
165    remote_ip: IpAddr,
166    client_cert_fingerprint: Option<String>,
167    state: &ServerState,
168) -> Response<Full<Bytes>> {
169    let dto: RegisterDtoV2 = match read_json_body(req.into_body()).await {
170        Ok(d) => d,
171        Err(status) => return empty_response(status),
172    };
173
174    // If client cert fingerprint is present (mTLS), ensure claimed fingerprint matches or log warning
175    if let Some(ref cert_fp) = client_cert_fingerprint.filter(|fp| !dto.fingerprint.eq_ignore_ascii_case(fp)) {
176        tracing::warn!(
177            "Claimed fingerprint {} does not match client certificate fingerprint {}",
178            dto.fingerprint,
179            cert_fp
180        );
181    }
182
183    // Ingest into PeerRegistry
184    state.registry.handle_http_register(remote_ip, &dto).await;
185
186    // Emit event if configured
187    if let Some(ref tx) = state.event_tx {
188        let _ = tx
189            .send(ServerTransferEvent::DeviceRegistered {
190                ip: remote_ip,
191                info: dto,
192            })
193            .await;
194    }
195
196    let files = state.download_manager.get_file_dtos().await;
197    let resp = RegisterResponseDtoV2 {
198        alias: state.config.alias.clone(),
199        version: PROTOCOL_VERSION_V2.to_string(),
200        device_model: Some(std::env::consts::OS.to_string()),
201        device_type: Some(lanflare_proto::discovery::DeviceType::Desktop),
202        fingerprint: state.cert_fingerprint.clone(),
203        download: !files.is_empty(),
204    };
205    json_response(StatusCode::OK, &resp)
206}
207
208/// Handler for `POST /api/localsend/v2/prepare-upload`.
209pub async fn handle_prepare_upload(
210    req: Request<Incoming>,
211    remote_ip: IpAddr,
212    client_cert_fingerprint: Option<String>,
213    state: &ServerState,
214) -> Response<Full<Bytes>> {
215    let query = parse_query(req.uri().query());
216    let provided_pin = query.get("pin").map(|s| s.as_str());
217
218    // Validate PIN
219    if let Err(status) = state
220        .session_manager
221        .pin_tracker()
222        .check_pin(state.config.pin.as_deref(), provided_pin, remote_ip)
223        .await
224    {
225        return empty_response(status);
226    }
227
228    let payload: PrepareUploadRequestDtoV2 = match read_json_body(req.into_body()).await {
229        Ok(p) => p,
230        Err(status) => return empty_response(status),
231    };
232
233    if payload.files.is_empty() {
234        return empty_response(StatusCode::BAD_REQUEST);
235    }
236
237    let session_id = Uuid::new_v4().to_string();
238    let cancel = CancellationToken::new();
239
240    // Claim pending session slot
241    if let Err(status) = state
242        .session_manager
243        .claim_pending(
244            session_id.clone(),
245            remote_ip,
246            client_cert_fingerprint.clone(),
247            cancel.clone(),
248        )
249        .await
250    {
251        return empty_response(status);
252    }
253
254    // Determine decision: if auto_accept is enabled in AppConfig, accept all files immediately
255    let decision = if state.config.auto_accept {
256        PrepareUploadDecision::Accept(payload.files.keys().cloned().collect())
257    } else if let Some(ref tx) = state.event_tx {
258        let (decision_tx, decision_rx) = oneshot::channel();
259        let event = ServerTransferEvent::PrepareUpload {
260            session_id: session_id.clone(),
261            sender_ip: remote_ip,
262            sender_info: payload.info.clone(),
263            cert_fingerprint: client_cert_fingerprint.clone(),
264            files: payload.files.clone(),
265            decision_tx,
266        };
267
268        if tx.send(event).await.is_err() {
269            state.session_manager.clear_pending(&session_id).await;
270            return empty_response(StatusCode::INTERNAL_SERVER_ERROR);
271        }
272
273        tokio::select! {
274            res = decision_rx => {
275                match res {
276                    Ok(d) => d,
277                    Err(_) => {
278                        state.session_manager.clear_pending(&session_id).await;
279                        return empty_response(StatusCode::INTERNAL_SERVER_ERROR);
280                    }
281                }
282            }
283            _ = cancel.cancelled() => {
284                state.session_manager.clear_pending(&session_id).await;
285                return empty_response(StatusCode::FORBIDDEN);
286            }
287        }
288    } else {
289        // Without an event channel or auto_accept, accept all files by default for headless operation
290        PrepareUploadDecision::Accept(payload.files.keys().cloned().collect())
291    };
292
293    let accepted_file_ids = match decision {
294        PrepareUploadDecision::Decline => {
295            state.session_manager.clear_pending(&session_id).await;
296            return empty_response(StatusCode::FORBIDDEN);
297        }
298        PrepareUploadDecision::Accept(ids) => ids,
299    };
300
301    if accepted_file_ids.is_empty() {
302        state.session_manager.clear_pending(&session_id).await;
303        // LocalSend spec Section 4.1: If empty or no files accepted, return 204 No Content
304        return empty_response(StatusCode::NO_CONTENT);
305    }
306
307    // Prepare session files with tokens and safe target paths
308    let mut session_files = HashMap::new();
309    let mut response_tokens = HashMap::new();
310
311    for (file_id, file_dto) in payload.files {
312        if !accepted_file_ids.contains(&file_id) {
313            continue;
314        }
315
316        let token = Uuid::new_v4().to_string();
317        let safe_name = match sanitize_filename(&file_dto.file_name) {
318            Ok(name) => name,
319            Err(_) => format!("file_{}.bin", Uuid::new_v4().simple()),
320        };
321
322        let target_path = match resolve_unique_destination_path(&state.config.download_dir, &safe_name)
323        {
324            Ok(p) => p,
325            Err(_) => {
326                state.session_manager.clear_pending(&session_id).await;
327                return empty_response(StatusCode::INTERNAL_SERVER_ERROR);
328            }
329        };
330
331        session_files.insert(
332            file_id.clone(),
333            SessionFile {
334                dto: file_dto,
335                token: token.clone(),
336                status: FileTransferStatus::Pending,
337                attempts: 0,
338                target_path,
339            },
340        );
341        response_tokens.insert(file_id, token);
342    }
343
344    if let Err(status) = state
345        .session_manager
346        .activate_session(
347            &session_id,
348            remote_ip,
349            client_cert_fingerprint,
350            session_files,
351        )
352        .await
353    {
354        state.session_manager.clear_pending(&session_id).await;
355        return empty_response(status);
356    }
357
358    let resp = PrepareUploadResponseDtoV2 {
359        session_id,
360        files: response_tokens,
361    };
362    json_response(StatusCode::OK, &resp)
363}
364
365/// Handler for `POST /api/localsend/v2/upload`.
366pub async fn handle_upload(
367    req: Request<Incoming>,
368    remote_ip: IpAddr,
369    state: &ServerState,
370) -> Response<Full<Bytes>> {
371    let query = parse_query(req.uri().query());
372    let (Some(session_id), Some(file_id), Some(token)) = (
373        query.get("sessionId"),
374        query.get("fileId"),
375        query.get("token"),
376    ) else {
377        return empty_response(StatusCode::BAD_REQUEST);
378    };
379
380    // Validate authorization and mark file InProgress
381    let session_file = match state
382        .session_manager
383        .validate_and_start_upload(session_id, file_id, token, remote_ip)
384        .await
385    {
386        Ok(f) => f,
387        Err(status) => return empty_response(status),
388    };
389
390    if let Some(ref tx) = state.event_tx {
391        let _ = tx
392            .send(ServerTransferEvent::FileStarted {
393                session_id: session_id.clone(),
394                file_id: file_id.clone(),
395                file_name: session_file.dto.file_name.clone(),
396                total_bytes: session_file.dto.size,
397            })
398            .await;
399    }
400
401    // Create streaming disk writer
402    let expected_size = Some(session_file.dto.size);
403    let expected_sha256 = session_file.dto.sha256.clone();
404
405    let mut writer = match StreamingFileWriter::create(
406        &session_file.target_path,
407        expected_size,
408        expected_sha256.clone(),
409    )
410    .await
411    {
412        Ok(w) => w,
413        Err(err) => {
414            tracing::error!("Failed to initialize streaming writer: {err}");
415            state
416                .session_manager
417                .mark_file_status(session_id, file_id, FileTransferStatus::Failed)
418                .await;
419            return empty_response(StatusCode::INTERNAL_SERVER_ERROR);
420        }
421    };
422
423    // Stream chunks directly from request body into disk writer
424    let mut body = req.into_body();
425    let mut stream_error = None;
426
427    while let Some(chunk_res) = body.frame().await {
428        match chunk_res {
429            Ok(frame) => {
430                let data = match frame.into_data() {
431                    Ok(d) => d,
432                    Err(_) => continue,
433                };
434                if let Err(err) = writer.write_chunk(&data).await {
435                    stream_error = Some(err);
436                    break;
437                }
438            }
439            Err(e) => {
440                stream_error = Some(crate::storage::writer::StorageError::StreamError(e.to_string()));
441                break;
442            }
443        }
444    }
445
446    if let Some(err) = stream_error {
447        tracing::warn!("Upload stream failed for file {file_id}: {err}");
448        let _ = writer.abort().await;
449        state
450            .session_manager
451            .mark_file_status(session_id, file_id, FileTransferStatus::Failed)
452            .await;
453        if let Some(ref tx) = state.event_tx {
454            let _ = tx
455                .send(ServerTransferEvent::FileFailed {
456                    session_id: session_id.clone(),
457                    file_id: file_id.clone(),
458                    error: err.to_string(),
459                })
460                .await;
461        }
462
463        return match err {
464            crate::storage::writer::StorageError::ChecksumMismatch { .. } => {
465                empty_response(StatusCode::UNPROCESSABLE_ENTITY)
466            }
467            crate::storage::writer::StorageError::SizeExceeded { .. } => {
468                empty_response(StatusCode::BAD_REQUEST)
469            }
470            _ => empty_response(StatusCode::INTERNAL_SERVER_ERROR),
471        };
472    }
473
474    // Commit file verification (validates length and expected SHA-256)
475    match writer.commit().await {
476        Ok(commit_result) => {
477            let session_finished = state
478                .session_manager
479                .mark_file_status(session_id, file_id, FileTransferStatus::Completed)
480                .await;
481
482            if let Some(ref tx) = state.event_tx {
483                let _ = tx
484                    .send(ServerTransferEvent::FileCompleted {
485                        session_id: session_id.clone(),
486                        file_id: file_id.clone(),
487                        result: commit_result,
488                    })
489                    .await;
490
491                if session_finished {
492                    let _ = tx
493                        .send(ServerTransferEvent::SessionEnded {
494                            session_id: session_id.clone(),
495                            reason: SessionEndReason::Finished,
496                        })
497                        .await;
498                }
499            }
500
501            empty_response(StatusCode::OK)
502        }
503        Err(err) => {
504            tracing::warn!("Commit verification failed for file {file_id}: {err}");
505            state
506                .session_manager
507                .mark_file_status(session_id, file_id, FileTransferStatus::Failed)
508                .await;
509
510            if let Some(ref tx) = state.event_tx {
511                let _ = tx
512                    .send(ServerTransferEvent::FileFailed {
513                        session_id: session_id.clone(),
514                        file_id: file_id.clone(),
515                        error: err.to_string(),
516                    })
517                    .await;
518            }
519
520            match err {
521                crate::storage::writer::StorageError::ChecksumMismatch { .. } => {
522                    empty_response(StatusCode::UNPROCESSABLE_ENTITY)
523                }
524                crate::storage::writer::StorageError::SizeMismatch { .. } => {
525                    empty_response(StatusCode::BAD_REQUEST)
526                }
527                _ => empty_response(StatusCode::INTERNAL_SERVER_ERROR),
528            }
529        }
530    }
531}
532
533/// Handler for `POST /api/localsend/v2/cancel`.
534pub async fn handle_cancel(
535    req: Request<Incoming>,
536    remote_ip: IpAddr,
537    state: &ServerState,
538) -> Response<Full<Bytes>> {
539    let query = parse_query(req.uri().query());
540    let session_id = query.get("sessionId").map(|s| s.as_str());
541
542    let reason = state
543        .session_manager
544        .cancel_session(session_id, remote_ip)
545        .await;
546
547    if let (Some(reason), Some(tx)) = (reason, &state.event_tx) {
548        let _ = tx
549            .send(ServerTransferEvent::SessionEnded {
550                session_id: session_id.unwrap_or("unknown").to_string(),
551                reason,
552            })
553            .await;
554    }
555
556    empty_response(StatusCode::OK)
557}
558
559/// Handler for `POST /api/localsend/v2/prepare-download` (Reverse file transfer / browser download metadata).
560pub async fn handle_prepare_download(
561    req: Request<Incoming>,
562    remote_ip: IpAddr,
563    state: &ServerState,
564) -> Response<Full<Bytes>> {
565    let query = parse_query(req.uri().query());
566    let provided_pin = query.get("pin").map(|s| s.as_str());
567
568    // Check if client is already accepted for this session
569    if let Some(session_id) = query.get("sessionId")
570        && state
571            .download_manager
572            .is_session_valid(session_id, remote_ip)
573            .await
574    {
575        let files = state.download_manager.get_file_dtos().await;
576        let resp = PrepareDownloadResponseDtoV2 {
577            info: InfoResponseDtoV2 {
578                alias: state.config.alias.clone(),
579                version: PROTOCOL_VERSION_V2.to_string(),
580                device_model: Some(std::env::consts::OS.to_string()),
581                device_type: Some(lanflare_proto::discovery::DeviceType::Desktop),
582                fingerprint: state.cert_fingerprint.clone(),
583                port: Some(state.config.port),
584                protocol: Some(ProtocolType::Https),
585                download: true,
586            },
587            session_id: session_id.clone(),
588            files,
589        };
590        return json_response(StatusCode::OK, &resp);
591    }
592
593    // Validate PIN
594    if let Err(status) = state
595        .session_manager
596        .pin_tracker()
597        .check_pin(state.config.pin.as_deref(), provided_pin, remote_ip)
598        .await
599    {
600        return empty_response(status);
601    }
602
603    let files = state.download_manager.get_file_dtos().await;
604    if files.is_empty() {
605        return empty_response(StatusCode::FORBIDDEN);
606    }
607
608    let session_id = state
609        .download_manager
610        .prepare_session(query.get("sessionId").map(|s| s.as_str()), remote_ip, true)
611        .await;
612
613    let resp = PrepareDownloadResponseDtoV2 {
614        info: InfoResponseDtoV2 {
615            alias: state.config.alias.clone(),
616            version: PROTOCOL_VERSION_V2.to_string(),
617            device_model: Some(std::env::consts::OS.to_string()),
618            device_type: Some(lanflare_proto::discovery::DeviceType::Desktop),
619            fingerprint: state.cert_fingerprint.clone(),
620            port: Some(state.config.port),
621            protocol: Some(ProtocolType::Https),
622            download: true,
623        },
624        session_id,
625        files,
626    };
627    json_response(StatusCode::OK, &resp)
628}
629
630/// Handler for `GET /api/localsend/v2/download?sessionId=...&fileId=...`.
631pub async fn handle_download(
632    req: Request<Incoming>,
633    remote_ip: IpAddr,
634    state: &ServerState,
635) -> Response<Full<Bytes>> {
636    let query = parse_query(req.uri().query());
637    let (Some(session_id), Some(file_id)) = (query.get("sessionId"), query.get("fileId")) else {
638        return empty_response(StatusCode::BAD_REQUEST);
639    };
640
641    let file_path = match state
642        .download_manager
643        .get_file_path(session_id, file_id, remote_ip)
644        .await
645    {
646        Ok(p) => p,
647        Err(status) => return empty_response(status),
648    };
649
650    let content = match tokio::fs::read(&file_path).await {
651        Ok(c) => c,
652        Err(e) => {
653            tracing::error!("Failed to read download file {:?}: {e}", file_path);
654            return empty_response(StatusCode::INTERNAL_SERVER_ERROR);
655        }
656    };
657
658    let file_name = file_path
659        .file_name()
660        .and_then(|n| n.to_str())
661        .unwrap_or("download.bin");
662
663    Response::builder()
664        .status(StatusCode::OK)
665        .header(hyper::header::CONTENT_TYPE, "application/octet-stream")
666        .header(
667            hyper::header::CONTENT_DISPOSITION,
668            format!("attachment; filename=\"{file_name}\""),
669        )
670        .header(hyper::header::CONTENT_LENGTH, content.len())
671        .body(Full::new(Bytes::from(content)))
672        .unwrap()
673}
lanflare_core/server/handlers.rs

lanflare_core/server/
handlers.rs
