torq_core/
api.rs

1//! REST + SSE API for the daemon.
2//!
3//! Bound to 127.0.0.1 only; routes require `Authorization: Bearer <token>`
4//! (token lives in config.toml so local clients can read it), except the
5//! stream route, which also accepts the short-lived capability token that
6//! `/play` embeds in the URL — real players can't send headers, so the URL
7//! itself is the ticket.
8
9use std::collections::HashMap;
10use std::convert::Infallible;
11use std::path::PathBuf;
12use std::sync::{Arc, Mutex};
13use std::time::{Duration, Instant};
14
15use axum::body::Body;
16use axum::extract::{Path, Query, State};
17use axum::http::{HeaderMap, HeaderValue, Request, StatusCode, header};
18use axum::middleware::Next;
19use axum::response::sse::{Event as SseEvent, KeepAlive, Sse};
20use axum::response::{IntoResponse, Response};
21use axum::routing::{delete, get, patch, post};
22use axum::{Json, Router};
23use futures::stream::{Stream, StreamExt};
24use librqbit::api::TorrentIdOrHash;
25use serde::{Deserialize, Serialize};
26use tokio::io::AsyncReadExt;
27use tokio_stream::wrappers::BroadcastStream;
28use tokio_util::io::ReaderStream;
29
30use crate::VERSION;
31use crate::daemon::{Daemon, Event, TorrentView};
32
33#[derive(Clone)]
34pub struct AppState {
35    pub daemon: Arc<Daemon>,
36    pub sources: Arc<torq_sources::Registry>,
37    pub client: reqwest::Client,
38    api_port: u16,
39    auth_token: String,
40    /// Capability tokens minted by `/play`, keyed to their issue time. The
41    /// stream route accepts one of these (in the URL) in place of the API
42    /// bearer header, since players can't send headers.
43    stream_tokens: Arc<Mutex<HashMap<String, Instant>>>,
44}
45
46/// How long a stream URL stays valid. Long enough for a player to pause and
47/// reconnect mid-session; `/play` sweeps expired entries as it mints new
48/// ones, so the map stays small.
49const STREAM_TOKEN_TTL: Duration = Duration::from_secs(60 * 60);
50
51pub fn router(
52    daemon: Arc<Daemon>,
53    auth_token: String,
54    sources: Arc<torq_sources::Registry>,
55    client: reqwest::Client,
56    api_port: u16,
57) -> Router {
58    let state = Arc::new(AppState {
59        daemon,
60        sources,
61        client,
62        api_port,
63        auth_token,
64        stream_tokens: Arc::new(Mutex::new(HashMap::new())),
65    });
66    Router::new()
67        .route("/health", get(health))
68        .route("/torrents", get(list_torrents).post(add_torrent))
69        .route("/torrents/{id}", delete(remove_torrent))
70        .route("/torrents/{id}/pause", post(pause_torrent))
71        .route("/torrents/{id}/resume", post(resume_torrent))
72        .route("/torrents/{id}/files", get(torrent_files))
73        .route("/torrents/{id}/play", get(play_file))
74        .route("/torrents/{id}/stream/{file_id}", get(stream_file))
75        .route("/search", get(search))
76        .route("/rss", get(list_rss).post(add_rss))
77        .route("/rss/{id}", delete(remove_rss))
78        .route("/library", get(library_status).post(library_scan))
79        .route("/config", get(get_config))
80        .route("/config/limits", patch(set_limits))
81        .route("/events", get(events))
82        .route_layer(axum::middleware::from_fn_with_state(
83            state.clone(),
84            require_auth,
85        ))
86        .with_state(state)
87}
88
89async fn require_auth(
90    State(state): State<Arc<AppState>>,
91    req: Request<Body>,
92    next: Next,
93) -> Response {
94    let header_ok = req
95        .headers()
96        .get(header::AUTHORIZATION)
97        .and_then(|v| v.to_str().ok())
98        .and_then(|v| v.strip_prefix("Bearer "))
99        .map(|token| constant_time_eq(token.as_bytes(), state.auth_token.as_bytes()))
100        .unwrap_or(false);
101    // The stream route is also reachable with the capability token `/play`
102    // embeds in the URL — players (VLC, IINA, mpv) can't send headers.
103    let token_ok = req
104        .uri()
105        .query()
106        .and_then(|q| {
107            q.split('&')
108                .find_map(|kv| kv.strip_prefix("token="))
109                .map(|t| stream_token_valid(&state.stream_tokens, t))
110        })
111        .unwrap_or(false);
112    if header_ok || token_ok {
113        next.run(req).await
114    } else {
115        (StatusCode::UNAUTHORIZED, "missing or invalid bearer token").into_response()
116    }
117}
118
119fn stream_token_valid(map: &Mutex<HashMap<String, Instant>>, token: &str) -> bool {
120    map.lock()
121        .ok()
122        .and_then(|m| m.get(token).copied())
123        .is_some_and(|issued| issued.elapsed() < STREAM_TOKEN_TTL)
124}
125
126/// Fresh capability token for a stream URL (16 random bytes, hex).
127fn new_stream_token() -> String {
128    let mut b = [0u8; 16];
129    getrandom::getrandom(&mut b).expect("os rng");
130    b.iter().map(|x| format!("{x:02x}")).collect()
131}
132
133fn constant_time_eq(a: &[u8], b: &[u8]) -> bool {
134    if a.len() != b.len() {
135        return false;
136    }
137    a.iter().zip(b).fold(0u8, |acc, (x, y)| acc | (x ^ y)) == 0
138}
139
140// -- handlers ---------------------------------------------------------------
141
142#[derive(Serialize)]
143struct Health {
144    version: &'static str,
145    torrents: usize,
146}
147
148async fn health(State(state): State<Arc<AppState>>) -> Json<Health> {
149    Json(Health {
150        version: VERSION,
151        torrents: state.daemon.views().len(),
152    })
153}
154
155async fn list_torrents(State(state): State<Arc<AppState>>) -> Json<Vec<TorrentView>> {
156    Json(state.daemon.views())
157}
158
159#[derive(Serialize)]
160struct ConfigInfo {
161    /// Concurrent transfer slots; torrents beyond this wait in queue.
162    max_active: usize,
163}
164
165async fn get_config(State(state): State<Arc<AppState>>) -> Json<ConfigInfo> {
166    Json(ConfigInfo {
167        max_active: state.daemon.max_active(),
168    })
169}
170
171#[derive(Serialize, Debug, PartialEq)]
172struct FileInfo {
173    id: usize,
174    name: String,
175    length: u64,
176    included: bool,
177}
178
179fn file_list(details: &librqbit::api::TorrentDetailsResponse) -> Vec<FileInfo> {
180    details
181        .files
182        .as_deref()
183        .unwrap_or_default()
184        .iter()
185        .enumerate()
186        .map(|(i, f)| FileInfo {
187            id: i,
188            name: f.components.join("/"),
189            length: f.length,
190            included: f.included,
191        })
192        .collect()
193}
194
195async fn torrent_files(
196    State(state): State<Arc<AppState>>,
197    Path(id): Path<String>,
198) -> Result<Json<Vec<FileInfo>>, ApiError> {
199    let details = state
200        .daemon
201        .engine()
202        .api()
203        .api_torrent_details(TorrentIdOrHash::parse(&id)?)?;
204    Ok(Json(file_list(&details)))
205}
206
207const VIDEO_EXTS: &[&str] = &[
208    "mp4", "mkv", "webm", "avi", "mov", "m4v", "ts", "wmv", "flv",
209];
210
211/// Largest video file, else the largest file overall — what a player wants.
212fn pick_play_file(files: &[FileInfo]) -> Option<&FileInfo> {
213    let is_video = |f: &FileInfo| {
214        f.name
215            .rsplit('.')
216            .next()
217            .is_some_and(|e| VIDEO_EXTS.contains(&e.to_ascii_lowercase().as_str()))
218    };
219    files
220        .iter()
221        .filter(|f| is_video(f))
222        .max_by_key(|f| f.length)
223        .or_else(|| files.iter().max_by_key(|f| f.length))
224}
225
226#[derive(Serialize)]
227struct PlayResponse {
228    url: String,
229    name: String,
230    file_id: usize,
231    length: u64,
232}
233
234/// Resolve the playable stream URL for a torrent: the largest video file
235/// (fallback: largest file), served by the range endpoint. One implementation
236/// shared by `torq play` and the TUI's `P` key.
237async fn play_file(
238    State(state): State<Arc<AppState>>,
239    Path(id): Path<String>,
240) -> Result<Json<PlayResponse>, ApiError> {
241    let details = state
242        .daemon
243        .engine()
244        .api()
245        .api_torrent_details(TorrentIdOrHash::parse(&id)?)?;
246    let files = file_list(&details);
247    let file =
248        pick_play_file(&files).ok_or_else(|| ApiError::NotFound("torrent has no files".into()))?;
249    let mut tokens = state.stream_tokens.lock().expect("stream tokens");
250    let now = Instant::now();
251    tokens.retain(|_, issued| now.duration_since(*issued) < STREAM_TOKEN_TTL);
252    let token = new_stream_token();
253    tokens.insert(token.clone(), now);
254    drop(tokens);
255    let url = format!(
256        "http://127.0.0.1:{}/torrents/{id}/stream/{}?token={token}",
257        state.api_port, file.id
258    );
259    Ok(Json(PlayResponse {
260        url,
261        name: file.name.clone(),
262        file_id: file.id,
263        length: file.length,
264    }))
265}
266
267/// HTTP range streaming of a torrent file, works mid-download: librqbit's
268/// `FileStream` reads pieces on demand (32MB lookahead), so a player can start
269/// before the file completes — mpv/VLC over this endpoint are the target.
270async fn stream_file(
271    State(state): State<Arc<AppState>>,
272    Path((id, file_id)): Path<(String, usize)>,
273    headers: HeaderMap,
274) -> Result<Response, ApiError> {
275    let parsed = TorrentIdOrHash::parse(&id)?;
276    let api = state.daemon.engine().api();
277    let details = api.api_torrent_details(parsed)?;
278    let files = details.files.as_deref().unwrap_or_default();
279    let file = files
280        .get(file_id)
281        .ok_or_else(|| ApiError::NotFound(format!("file {file_id} not found")))?;
282    let total = file.length;
283
284    let mut stream = api.api_stream(parsed, file_id)?;
285    let range = headers
286        .get(header::RANGE)
287        .and_then(|v| v.to_str().ok())
288        .and_then(|r| parse_range(r, total));
289    let (status, start, end) = match range {
290        Some((s, e)) => (StatusCode::PARTIAL_CONTENT, s, e),
291        None => (StatusCode::OK, 0, total.saturating_sub(1)),
292    };
293    if start > 0 {
294        use tokio::io::AsyncSeekExt;
295        stream.seek(std::io::SeekFrom::Start(start)).await?;
296    }
297    let len = end - start + 1;
298
299    let mut headers = HeaderMap::new();
300    headers.insert("Accept-Ranges", HeaderValue::from_static("bytes"));
301    headers.insert(
302        header::CONTENT_TYPE,
303        HeaderValue::from_static(mime_for(&file.name)),
304    );
305    if status == StatusCode::PARTIAL_CONTENT {
306        headers.insert(
307            header::CONTENT_RANGE,
308            HeaderValue::from_str(&format!("bytes {start}-{end}/{total}")).expect("valid header"),
309        );
310    }
311    headers.insert(
312        header::CONTENT_LENGTH,
313        HeaderValue::from_str(&len.to_string()).expect("valid header"),
314    );
315
316    let body = Body::from_stream(ReaderStream::with_capacity(stream.take(len), 64 * 1024));
317    Ok((status, headers, body).into_response())
318}
319
320/// Parse a single-range `bytes=` header; returns (start, end), inclusive.
321fn parse_range(header: &str, total: u64) -> Option<(u64, u64)> {
322    if total == 0 {
323        return None;
324    }
325    let spec = header.strip_prefix("bytes=")?;
326    let (start_str, end_str) = spec.split_once('-')?;
327    if start_str.is_empty() {
328        // Suffix range: last N bytes.
329        let n = end_str.parse::<u64>().ok()?;
330        if n == 0 {
331            return None;
332        }
333        let start = total.saturating_sub(n);
334        return Some((start, total - 1));
335    }
336    let start = start_str.parse::<u64>().ok()?;
337    if start >= total {
338        return None;
339    }
340    let end = if end_str.is_empty() {
341        total - 1
342    } else {
343        end_str.parse::<u64>().ok()?.min(total - 1)
344    };
345    (end >= start).then_some((start, end))
346}
347
348fn mime_for(name: &str) -> &'static str {
349    match name
350        .rsplit('.')
351        .next()
352        .unwrap_or("")
353        .to_ascii_lowercase()
354        .as_str()
355    {
356        "mp4" | "m4v" => "video/mp4",
357        "mkv" => "video/x-matroska",
358        "webm" => "video/webm",
359        "avi" => "video/x-msvideo",
360        "mov" => "video/quicktime",
361        "ts" => "video/mp2t",
362        "wmv" => "video/x-ms-wmv",
363        "flv" => "video/x-flv",
364        "mp3" => "audio/mpeg",
365        "m4a" | "aac" => "audio/mp4",
366        "flac" => "audio/flac",
367        "ogg" | "opus" => "audio/ogg",
368        _ => "application/octet-stream",
369    }
370}
371
372#[derive(Deserialize)]
373struct SearchReq {
374    q: String,
375    /// Comma-separated source ids; empty = all.
376    #[serde(default)]
377    sources: Option<String>,
378}
379
380/// Aggregated search across enabled sources, deduped by infohash. Failing
381/// sources are reported in `offline`, never fatal.
382async fn search(
383    State(state): State<Arc<AppState>>,
384    Query(req): Query<SearchReq>,
385) -> Json<torq_sources::SearchReport> {
386    let only = req
387        .sources
388        .as_deref()
389        .map(|s| s.split(',').map(str::to_string).collect::<Vec<_>>());
390    let report = torq_sources::aggregate::search_all(
391        &state.sources.sources,
392        &state.client,
393        &req.q,
394        only.as_deref(),
395    )
396    .await;
397    Json(report)
398}
399
400#[derive(Deserialize)]
401struct AddReq {
402    #[serde(default)]
403    magnet: String,
404    #[serde(default)]
405    paused: bool,
406    /// Base64-encoded .torrent bytes (mutually exclusive with magnet).
407    #[serde(default)]
408    torrent_b64: Option<String>,
409}
410
411async fn add_torrent(
412    State(state): State<Arc<AppState>>,
413    Json(req): Json<AddReq>,
414) -> Result<Json<TorrentView>, ApiError> {
415    let view = match req.torrent_b64 {
416        Some(b64) => {
417            use base64::Engine;
418            let bytes = base64::engine::general_purpose::STANDARD
419                .decode(b64.trim())
420                .map_err(|e| ApiError::BadRequest(format!("invalid torrent_b64: {e}")))?;
421            state.daemon.add_torrent_bytes(bytes, req.paused).await?
422        }
423        None if req.magnet.trim().is_empty() => {
424            return Err(ApiError::BadRequest(
425                "provide a magnet or torrent_b64".into(),
426            ));
427        }
428        None => state.daemon.add_magnet(&req.magnet, req.paused).await?,
429    };
430    Ok(Json(view))
431}
432
433#[derive(Deserialize, Default)]
434struct RemoveReq {
435    #[serde(default, deserialize_with = "deserialize_bool_flag")]
436    delete_files: bool,
437}
438
439/// serde_urlencoded only parses `true`/`false` for bools; scripts and curl
440/// users naturally write `?delete_files=1`, so accept both spellings.
441fn deserialize_bool_flag<'de, D>(d: D) -> Result<bool, D::Error>
442where
443    D: serde::Deserializer<'de>,
444{
445    let s = String::deserialize(d)?;
446    match s.as_str() {
447        "true" | "1" => Ok(true),
448        "false" | "0" => Ok(false),
449        other => Err(serde::de::Error::custom(format!(
450            "expected true/false/1/0, got {other:?}"
451        ))),
452    }
453}
454
455async fn remove_torrent(
456    State(state): State<Arc<AppState>>,
457    Path(id): Path<String>,
458    Query(req): Query<RemoveReq>,
459) -> Result<StatusCode, ApiError> {
460    let parsed = TorrentIdOrHash::parse(&id)?;
461    state.daemon.remove(parsed, req.delete_files).await?;
462    Ok(StatusCode::NO_CONTENT)
463}
464
465async fn pause_torrent(
466    State(state): State<Arc<AppState>>,
467    Path(id): Path<String>,
468) -> Result<StatusCode, ApiError> {
469    state.daemon.pause(TorrentIdOrHash::parse(&id)?).await?;
470    Ok(StatusCode::NO_CONTENT)
471}
472
473async fn resume_torrent(
474    State(state): State<Arc<AppState>>,
475    Path(id): Path<String>,
476) -> Result<StatusCode, ApiError> {
477    state.daemon.resume(TorrentIdOrHash::parse(&id)?).await?;
478    Ok(StatusCode::NO_CONTENT)
479}
480
481async fn list_rss(State(state): State<Arc<AppState>>) -> Json<Vec<crate::rss::Subscription>> {
482    Json(state.daemon.rss.list())
483}
484
485#[derive(Deserialize)]
486struct AddRssReq {
487    url: String,
488    #[serde(default)]
489    title_re: Option<String>,
490    #[serde(default)]
491    min_size: Option<u64>,
492    #[serde(default)]
493    max_size: Option<u64>,
494    #[serde(default = "default_sub_interval")]
495    interval_secs: u64,
496}
497
498fn default_sub_interval() -> u64 {
499    300
500}
501
502async fn add_rss(
503    State(state): State<Arc<AppState>>,
504    Json(req): Json<AddRssReq>,
505) -> Result<Json<crate::rss::Subscription>, ApiError> {
506    let sub = state.daemon.rss.add(
507        &req.url,
508        req.title_re,
509        req.min_size,
510        req.max_size,
511        req.interval_secs,
512    )?;
513    Ok(Json(sub))
514}
515
516async fn remove_rss(
517    State(state): State<Arc<AppState>>,
518    Path(id): Path<u64>,
519) -> Result<StatusCode, ApiError> {
520    if state.daemon.rss.remove(id) {
521        Ok(StatusCode::NO_CONTENT)
522    } else {
523        Err(ApiError::NotFound(format!("subscription {id} not found")))
524    }
525}
526
527#[derive(Serialize)]
528struct LibraryStatus {
529    indexed: usize,
530    dirs: Vec<PathBuf>,
531}
532
533async fn library_status(State(state): State<Arc<AppState>>) -> Json<LibraryStatus> {
534    Json(LibraryStatus {
535        indexed: state.daemon.library.count(),
536        dirs: state.daemon.library.dirs(),
537    })
538}
539
540/// Rescan library dirs; returns the number of torrents indexed.
541async fn library_scan(State(state): State<Arc<AppState>>) -> Result<Json<LibraryStatus>, ApiError> {
542    state.daemon.library.scan()?;
543    Ok(Json(LibraryStatus {
544        indexed: state.daemon.library.count(),
545        dirs: state.daemon.library.dirs(),
546    }))
547}
548
549#[derive(Deserialize)]
550struct LimitsReq {
551    upload_bps: Option<u32>,
552    download_bps: Option<u32>,
553}
554
555/// Apply rate limits live (None clears the limit). Persists to config so the
556/// daemon restarts with them.
557async fn set_limits(
558    State(state): State<Arc<AppState>>,
559    Json(req): Json<LimitsReq>,
560) -> Result<StatusCode, ApiError> {
561    state
562        .daemon
563        .engine()
564        .set_limits(req.upload_bps, req.download_bps);
565    let mut config = crate::config::Config::load()?;
566    config.upload_bps = req.upload_bps;
567    config.download_bps = req.download_bps;
568    config.save()?;
569    Ok(StatusCode::NO_CONTENT)
570}
571
572async fn events(
573    State(state): State<Arc<AppState>>,
574) -> Sse<impl Stream<Item = Result<SseEvent, Infallible>>> {
575    let stream = BroadcastStream::new(state.daemon.subscribe()).map(|item| {
576        let data = match item {
577            Ok(Event::TorrentFailed { id, error }) => {
578                serde_json::json!({"type": "torrent_failed", "id": id, "error": error}).to_string()
579            }
580            Ok(ev) => serde_json::to_string(&ev).unwrap_or_else(|_| "{}".into()),
581            Err(_) => "{}".into(), // lagged behind; client should re-poll /torrents
582        };
583        Ok(SseEvent::default().event("torq").data(data))
584    });
585    Sse::new(stream).keep_alive(KeepAlive::default())
586}
587
588// -- errors ------------------------------------------------------------------
589
590pub enum ApiError {
591    NotFound(String),
592    BadRequest(String),
593    Internal(anyhow::Error),
594}
595
596impl IntoResponse for ApiError {
597    fn into_response(self) -> Response {
598        match self {
599            Self::NotFound(m) => (StatusCode::NOT_FOUND, m).into_response(),
600            Self::BadRequest(m) => (StatusCode::BAD_REQUEST, m).into_response(),
601            Self::Internal(e) => {
602                tracing::error!("api error: {e:#}");
603                (StatusCode::INTERNAL_SERVER_ERROR, "internal error").into_response()
604            }
605        }
606    }
607}
608
609impl From<anyhow::Error> for ApiError {
610    fn from(e: anyhow::Error) -> Self {
611        let msg = e.to_string();
612        if msg.contains("not found") || msg.contains("not managed") {
613            Self::NotFound(msg)
614        } else if msg.contains("not a valid magnet") || msg.contains("failed to parse") {
615            Self::BadRequest(msg)
616        } else {
617            Self::Internal(e)
618        }
619    }
620}
621
622impl From<librqbit::ApiError> for ApiError {
623    fn from(e: librqbit::ApiError) -> Self {
624        let msg = e.to_string();
625        if msg.contains("not found") {
626            Self::NotFound(msg)
627        } else {
628            Self::Internal(anyhow::anyhow!(msg))
629        }
630    }
631}
632
633impl From<std::io::Error> for ApiError {
634    fn from(e: std::io::Error) -> Self {
635        Self::Internal(anyhow::anyhow!(e))
636    }
637}
638
639#[cfg(test)]
640mod tests {
641    use super::*;
642
643    fn file(id: usize, name: &str, length: u64) -> FileInfo {
644        FileInfo {
645            id,
646            name: name.into(),
647            length,
648            included: true,
649        }
650    }
651
652    #[test]
653    fn stream_tokens_expire_and_reject_unknown() {
654        let map = Mutex::new(HashMap::new());
655        assert!(!stream_token_valid(&map, "nope"));
656        map.lock().unwrap().insert("fresh".into(), Instant::now());
657        assert!(stream_token_valid(&map, "fresh"));
658        map.lock()
659            .unwrap()
660            .insert("stale".into(), Instant::now() - STREAM_TOKEN_TTL - Duration::from_secs(1));
661        assert!(!stream_token_valid(&map, "stale"));
662        // A fresh token survives alongside the stale one.
663        assert!(stream_token_valid(&map, "fresh"));
664    }
665
666    #[test]
667    fn play_picks_largest_video_else_largest_file() {
668        let files = vec![
669            file(0, "cover.jpg", 500_000),
670            file(1, "movie.mkv", 5_000_000_000),
671            file(2, "sample.mp4", 200_000_000),
672        ];
673        assert_eq!(pick_play_file(&files).unwrap().id, 1);
674        // No video: largest file wins.
675        let only_audio = vec![file(0, "song.mp3", 10_000_000), file(1, "notes.txt", 1)];
676        assert_eq!(pick_play_file(&only_audio).unwrap().id, 0);
677        assert_eq!(pick_play_file(&[]), None);
678    }
679}
torq_core/api.rs

torq_core/
api.rs
