rightkit_tauri_shell/
stream.rs

1//! Ranged custom-protocol streaming of broker-granted files.
2//!
3//! Donors: ViewRight `binary_stream.rs` (id resolution, HEAD/GET/OPTIONS, single-range parsing,
4//! heavy reads off the calling thread) and `stream_http.rs` (range parser, fallible response
5//! builders that never panic). The shell registers one or more schemes with
6//! `ShellBuilder::stream_protocol("myapp-stream", policy)`; URLs are `myapp-stream://localhost/<id>`
7//! (`http://myapp-stream.localhost/<id>` on Windows). The only credential is the id minted by
8//! [`crate::broker::PathBroker::grant_stream`]. Revoking the file revokes its ids for good.
9
10use std::collections::BTreeMap;
11use std::fmt;
12use std::fs::File;
13use std::io::{Read, Seek, SeekFrom};
14use std::panic::{catch_unwind, AssertUnwindSafe};
15use std::path::Path;
16use std::sync::Arc;
17
18use tauri::http::response::Builder;
19use tauri::http::{header, Method, Request, Response, StatusCode};
20
21use crate::broker::PathBroker;
22
23/// What happened to one stream request (see [`StreamPolicy::observer`]).
24#[derive(Debug, Clone, Copy, PartialEq, Eq)]
25pub enum StreamEventKind {
26    /// The id resolved, the type is allowed and the file is within its size cap. `size` is the
27    /// file length. Emitted once per request that gets this far; refused requests emit nothing.
28    Resolved { size: u64 },
29    /// A HEAD request was answered. `range` is the inclusive range when the request carried a
30    /// valid `Range` header.
31    Head {
32        size: u64,
33        range: Option<(u64, u64)>,
34    },
35    /// A GET with a valid single range was accepted (emitted before the read). `end` is inclusive.
36    RangeServed { start: u64, end: u64, size: u64 },
37    /// A GET without a `Range` header was accepted (emitted before the read).
38    FullServed { size: u64 },
39    /// A `Range` header was malformed, multi-range or unsatisfiable; the answer is 416.
40    RangeRejected { size: u64 },
41    /// The body was read and handed to the responder: `bytes` is the body length, `ranged` is true
42    /// for a range response. Emitted from the reader thread.
43    BytesRead { bytes: u64, ranged: bool },
44}
45
46/// One observed stream event with the content type of the file it concerns.
47#[derive(Debug, Clone, Copy, PartialEq, Eq)]
48pub struct StreamEvent {
49    /// The MIME type from [`mime_for`] (for example `application/pdf`), so an observer can ignore
50    /// kinds it does not care about.
51    pub mime: &'static str,
52    pub kind: StreamEventKind,
53}
54
55/// A shared callback receiving [`StreamEvent`]s. Cheap to clone. A panic inside it is caught and
56/// dropped; it never reaches the response. It runs inline on the request path (and on the reader
57/// thread for `BytesRead`), so keep it quick: bump a counter, do not do I/O.
58#[derive(Clone)]
59pub struct StreamObserver(Arc<dyn Fn(&StreamEvent) + Send + Sync>);
60
61impl StreamObserver {
62    pub fn new(f: impl Fn(&StreamEvent) + Send + Sync + 'static) -> Self {
63        Self(Arc::new(f))
64    }
65
66    fn emit(&self, mime: &'static str, kind: StreamEventKind) {
67        let event = StreamEvent { mime, kind };
68        let _ = catch_unwind(AssertUnwindSafe(|| (self.0)(&event)));
69    }
70}
71
72impl fmt::Debug for StreamObserver {
73    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
74        f.write_str("StreamObserver(..)")
75    }
76}
77
78fn emit(observer: &Option<StreamObserver>, mime: &'static str, kind: StreamEventKind) {
79    if let Some(o) = observer {
80        o.emit(mime, kind);
81    }
82}
83
84/// Policy for what a stream scheme will serve.
85#[derive(Debug, Clone)]
86pub struct StreamPolicy {
87    /// Largest file served (bytes) when no per-kind cap applies.
88    pub max_bytes: u64,
89    /// Lowercase extensions served; `None` serves any type the mime table knows.
90    pub extensions: Option<Vec<String>>,
91    /// Per-kind size caps (bytes). A key is an extension (`"mov"`, a leading dot is optional), a full
92    /// MIME type (`"application/pdf"`) or a MIME family (`"video"` or `"video/*"`). Lookup order is
93    /// extension, then full MIME, then family; a file matching no key uses `max_bytes`. Empty by
94    /// default, so behaviour is unchanged until a cap is added with [`StreamPolicy::with_kind_cap`].
95    pub max_bytes_by_kind: BTreeMap<String, u64>,
96    /// Optional observer of served requests; `None` (the default) costs one branch per request. Set
97    /// it with [`StreamPolicy::observer`].
98    pub observer: Option<StreamObserver>,
99}
100
101impl Default for StreamPolicy {
102    fn default() -> Self {
103        Self {
104            max_bytes: 150 * 1024 * 1024,
105            extensions: None,
106            max_bytes_by_kind: BTreeMap::new(),
107            observer: None,
108        }
109    }
110}
111
112impl StreamPolicy {
113    /// Add or replace a per-kind cap. `kind` follows the key rules of
114    /// [`StreamPolicy::max_bytes_by_kind`].
115    pub fn with_kind_cap(mut self, kind: &str, max_bytes: u64) -> Self {
116        self.max_bytes_by_kind
117            .insert(normalize_kind(kind), max_bytes);
118        self
119    }
120
121    /// Observe requests served under this policy: resolved size, HEAD, range served, full served,
122    /// rejected range and bytes read (see [`StreamEventKind`]). A panicking observer is contained.
123    pub fn observer(mut self, f: impl Fn(&StreamEvent) + Send + Sync + 'static) -> Self {
124        self.observer = Some(StreamObserver::new(f));
125        self
126    }
127
128    /// The size cap that applies to `path`: the most specific per-kind entry, else `max_bytes`.
129    pub fn max_bytes_for(&self, path: &Path) -> u64 {
130        if self.max_bytes_by_kind.is_empty() {
131            return self.max_bytes;
132        }
133        let ext = path
134            .extension()
135            .and_then(|e| e.to_str())
136            .map(|e| e.to_ascii_lowercase());
137        if let Some(cap) = ext.as_deref().and_then(|e| self.max_bytes_by_kind.get(e)) {
138            return *cap;
139        }
140        let Some(mime) = mime_for(path) else {
141            return self.max_bytes;
142        };
143        let mime = mime.split(';').next().unwrap_or(mime).trim();
144        if let Some(cap) = self.max_bytes_by_kind.get(mime) {
145            return *cap;
146        }
147        let family = mime.split('/').next().unwrap_or(mime);
148        self.max_bytes_by_kind
149            .get(family)
150            .copied()
151            .unwrap_or(self.max_bytes)
152    }
153}
154
155fn normalize_kind(kind: &str) -> String {
156    kind.trim()
157        .trim_start_matches('.')
158        .trim_end_matches("/*")
159        .trim_end_matches('/')
160        .to_ascii_lowercase()
161}
162
163/// Content type by extension (media and document formats the suite streams).
164pub fn mime_for(path: &Path) -> Option<&'static str> {
165    let ext = path.extension()?.to_str()?.to_ascii_lowercase();
166    Some(match ext.as_str() {
167        "pdf" => "application/pdf",
168        "docx" => "application/vnd.openxmlformats-officedocument.wordprocessingml.document",
169        "xlsx" => "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
170        "png" => "image/png",
171        "jpg" | "jpeg" => "image/jpeg",
172        "gif" => "image/gif",
173        "webp" => "image/webp",
174        "avif" => "image/avif",
175        "svg" => "image/svg+xml",
176        "bmp" => "image/bmp",
177        "ico" => "image/x-icon",
178        "tif" | "tiff" => "image/tiff",
179        "mp3" => "audio/mpeg",
180        "wav" => "audio/wav",
181        "m4a" => "audio/mp4",
182        "flac" => "audio/flac",
183        "ogg" | "oga" => "audio/ogg",
184        "mp4" | "m4v" => "video/mp4",
185        "mov" => "video/quicktime",
186        "webm" => "video/webm",
187        "txt" | "log" => "text/plain; charset=utf-8",
188        "md" => "text/markdown; charset=utf-8",
189        "json" => "application/json",
190        "csv" => "text/csv; charset=utf-8",
191        "bin" | "dat" => "application/octet-stream",
192        _ => return None,
193    })
194}
195
196/// Last-resort response when a builder call fails (a malformed header value). Built from
197/// `Response::new`, which cannot fail, so a release build never panics here.
198pub(crate) fn internal_error() -> Response<Vec<u8>> {
199    let mut r = Response::new(Vec::new());
200    *r.status_mut() = StatusCode::INTERNAL_SERVER_ERROR;
201    r
202}
203
204/// Finish a builder, falling back to [`internal_error`] instead of panicking.
205pub(crate) fn finish(b: Builder, body: Vec<u8>) -> Response<Vec<u8>> {
206    b.body(body).unwrap_or_else(|_| internal_error())
207}
208
209/// CORS headers every stream and site response carries, errors included.
210pub(crate) fn cors(status: StatusCode) -> Builder {
211    Response::builder()
212        .status(status)
213        .header(header::ACCESS_CONTROL_ALLOW_ORIGIN, "*")
214        .header(header::ACCESS_CONTROL_ALLOW_METHODS, "GET, HEAD, OPTIONS")
215        .header(header::ACCESS_CONTROL_ALLOW_HEADERS, "Range")
216        .header(
217            header::ACCESS_CONTROL_EXPOSE_HEADERS,
218            "Accept-Ranges, Content-Length, Content-Range",
219        )
220}
221
222/// Body response headers: mime, length, range support, no caching, CORS.
223pub(crate) fn ranged(status: StatusCode, mime: &str, len: u64) -> Builder {
224    Response::builder()
225        .status(status)
226        .header(header::CONTENT_TYPE, mime)
227        .header(header::ACCEPT_RANGES, "bytes")
228        .header(header::CONTENT_LENGTH, len.to_string())
229        .header(header::CACHE_CONTROL, "no-store")
230        .header(header::ACCESS_CONTROL_ALLOW_ORIGIN, "*")
231        .header(
232            header::ACCESS_CONTROL_EXPOSE_HEADERS,
233            "Accept-Ranges, Content-Length, Content-Range",
234        )
235}
236
237/// A bodiless response with CORS headers.
238pub(crate) fn empty(status: StatusCode) -> Response<Vec<u8>> {
239    finish(cors(status), Vec::new())
240}
241
242/// The method gate shared by every custom scheme: OPTIONS answers 204, anything but GET/HEAD is
243/// refused. `None` means keep handling the request.
244pub(crate) fn method_gate_status(method: &Method) -> Option<StatusCode> {
245    if method == Method::OPTIONS {
246        return Some(StatusCode::NO_CONTENT);
247    }
248    if method != Method::GET && method != Method::HEAD {
249        return Some(StatusCode::METHOD_NOT_ALLOWED);
250    }
251    None
252}
253
254/// Parse one `bytes=a-b` / `bytes=a-` / `bytes=-n` range against a known length into an inclusive
255/// `(start, end)`. Multi-range requests, empty bodies and unsatisfiable ranges are refused.
256pub fn parse_range(value: &str, len: u64) -> Option<(u64, u64)> {
257    let spec = value.strip_prefix("bytes=")?;
258    if spec.contains(',') || len == 0 {
259        return None;
260    }
261    let (a, b) = spec.split_once('-')?;
262    if a.is_empty() {
263        let suffix = b.parse::<u64>().ok()?;
264        if suffix == 0 {
265            return None;
266        }
267        return Some((len.saturating_sub(suffix), len - 1));
268    }
269    let start = a.parse::<u64>().ok()?;
270    if start >= len {
271        return None;
272    }
273    let end = if b.is_empty() {
274        len - 1
275    } else {
276        b.parse::<u64>().ok()?.min(len - 1)
277    };
278    (start <= end).then_some((start, end))
279}
280
281/// Read the inclusive `[start, end]` slice of `path`.
282pub(crate) fn read_range(path: &Path, start: u64, end: u64) -> std::io::Result<Vec<u8>> {
283    let mut f = File::open(path)?;
284    f.seek(SeekFrom::Start(start))?;
285    let mut buf = vec![0; (end - start + 1) as usize];
286    f.read_exact(&mut buf)?;
287    Ok(buf)
288}
289
290/// Serve one request. Validation (id, mime, stat, range) is synchronous and cheap; only the body
291/// read runs on a dedicated thread, then `respond` is called from there.
292pub fn serve<F>(broker: &PathBroker, policy: &StreamPolicy, request: &Request<Vec<u8>>, respond: F)
293where
294    F: FnOnce(Response<Vec<u8>>) + Send + 'static,
295{
296    if let Some(status) = method_gate_status(request.method()) {
297        return respond(empty(status));
298    }
299    let id = request.uri().path().trim_start_matches('/').trim();
300    if id.is_empty() {
301        return respond(empty(StatusCode::BAD_REQUEST));
302    }
303    let Ok(target) = broker.resolve_stream(id) else {
304        return respond(empty(StatusCode::FORBIDDEN));
305    };
306    let path = target.path;
307    let allowed_ext = policy.extensions.as_ref().is_none_or(|exts| {
308        path.extension()
309            .and_then(|e| e.to_str())
310            .is_some_and(|e| exts.iter().any(|x| x.eq_ignore_ascii_case(e)))
311    });
312    let mime = match mime_for(&path) {
313        Some(m) if allowed_ext => m,
314        _ => return respond(empty(StatusCode::UNSUPPORTED_MEDIA_TYPE)),
315    };
316    let Ok(meta) = std::fs::metadata(&path) else {
317        return respond(empty(StatusCode::NOT_FOUND));
318    };
319    if !meta.is_file() || meta.len() > policy.max_bytes_for(&path) {
320        return respond(empty(StatusCode::PAYLOAD_TOO_LARGE));
321    }
322    let total = meta.len();
323    let head = method_is_head(request);
324    let observer = &policy.observer;
325    emit(observer, mime, StreamEventKind::Resolved { size: total });
326
327    if let Some(range) = request.headers().get(header::RANGE) {
328        let parsed = range.to_str().ok().and_then(|r| parse_range(r, total));
329        let Some((start, end)) = parsed else {
330            emit(
331                observer,
332                mime,
333                StreamEventKind::RangeRejected { size: total },
334            );
335            return respond(finish(
336                cors(StatusCode::RANGE_NOT_SATISFIABLE)
337                    .header(header::CONTENT_RANGE, format!("bytes */{total}")),
338                Vec::new(),
339            ));
340        };
341        let builder = ranged(StatusCode::PARTIAL_CONTENT, mime, end - start + 1).header(
342            header::CONTENT_RANGE,
343            format!("bytes {start}-{end}/{total}"),
344        );
345        if head {
346            emit(
347                observer,
348                mime,
349                StreamEventKind::Head {
350                    size: total,
351                    range: Some((start, end)),
352                },
353            );
354            return respond(finish(builder, Vec::new()));
355        }
356        emit(
357            observer,
358            mime,
359            StreamEventKind::RangeServed {
360                start,
361                end,
362                size: total,
363            },
364        );
365        let observer = observer.clone();
366        std::thread::spawn(move || {
367            respond(match read_range(&path, start, end) {
368                Ok(bytes) => {
369                    emit(
370                        &observer,
371                        mime,
372                        StreamEventKind::BytesRead {
373                            bytes: bytes.len() as u64,
374                            ranged: true,
375                        },
376                    );
377                    finish(builder, bytes)
378                }
379                Err(_) => empty(StatusCode::NOT_FOUND),
380            })
381        });
382        return;
383    }
384    if head {
385        emit(
386            observer,
387            mime,
388            StreamEventKind::Head {
389                size: total,
390                range: None,
391            },
392        );
393        return respond(finish(ranged(StatusCode::OK, mime, total), Vec::new()));
394    }
395    emit(observer, mime, StreamEventKind::FullServed { size: total });
396    let observer = observer.clone();
397    std::thread::spawn(move || {
398        respond(match std::fs::read(&path) {
399            Ok(bytes) => {
400                emit(
401                    &observer,
402                    mime,
403                    StreamEventKind::BytesRead {
404                        bytes: bytes.len() as u64,
405                        ranged: false,
406                    },
407                );
408                finish(ranged(StatusCode::OK, mime, bytes.len() as u64), bytes)
409            }
410            Err(_) => empty(StatusCode::NOT_FOUND),
411        })
412    });
413}
414
415fn method_is_head<B>(request: &Request<B>) -> bool {
416    request.method() == Method::HEAD
417}
rightkit_tauri_shell/stream.rs

rightkit_tauri_shell/
stream.rs
