gfeh_http/
server.rs

1//! The HTTP listener and its one route.
2
3use axum::Router;
4use axum::body::Body;
5use axum::extract::{Path, State};
6use axum::http::{HeaderMap, HeaderValue, StatusCode, header};
7use axum::response::{IntoResponse, Response};
8use axum::routing::get;
9use futures::StreamExt;
10use gfeh_core::{Meta, ObjectStore, OpCtx, OpenMode, Perm, RangeSpec, parse_range};
11use std::net::SocketAddr;
12use std::sync::Arc;
13use tokio::net::TcpListener;
14
15use crate::exposure::{Exposed, Exposures};
16
17/// A running HTTP view.
18#[derive(Debug)]
19pub struct HttpView {
20    address: SocketAddr,
21    shutdown: Option<tokio::sync::oneshot::Sender<()>>,
22}
23
24impl HttpView {
25    /// Start building.
26    #[must_use]
27    pub fn builder(store: Arc<dyn ObjectStore>, exposures: Arc<dyn Exposures>) -> HttpBuilder {
28        HttpBuilder {
29            store,
30            exposures,
31            bind: SocketAddr::from(([127, 0, 0, 1], 0)),
32        }
33    }
34
35    /// Where it is listening.
36    #[must_use]
37    pub fn address(&self) -> SocketAddr {
38        self.address
39    }
40
41    /// The base URL, without a trailing slash.
42    #[must_use]
43    pub fn base_url(&self) -> String {
44        format!("http://{}", self.address)
45    }
46
47    /// The URL a token resolves at.
48    #[must_use]
49    pub fn url_for(&self, token: &str) -> String {
50        format!("{}/f/{token}", self.base_url())
51    }
52}
53
54impl Drop for HttpView {
55    fn drop(&mut self) {
56        if let Some(shutdown) = self.shutdown.take() {
57            let _ = shutdown.send(());
58        }
59    }
60}
61
62/// Assembles an [`HttpView`].
63pub struct HttpBuilder {
64    store: Arc<dyn ObjectStore>,
65    exposures: Arc<dyn Exposures>,
66    bind: SocketAddr,
67}
68
69impl HttpBuilder {
70    /// Listen on this address. Port zero means an ephemeral port.
71    #[must_use]
72    pub fn bind(mut self, addr: SocketAddr) -> Self {
73        self.bind = addr;
74        self
75    }
76
77    /// Bind and serve.
78    ///
79    /// # Errors
80    ///
81    /// Returns the bind failure if the address is unavailable.
82    pub async fn start(self) -> std::io::Result<HttpView> {
83        let listener = TcpListener::bind(self.bind).await?;
84        let address = listener.local_addr()?;
85        let (tx, rx) = tokio::sync::oneshot::channel();
86
87        let state = Arc::new(Served {
88            store: self.store,
89            exposures: self.exposures,
90        });
91        let app = Router::new()
92            .route("/f/{token}", get(serve).head(serve))
93            .with_state(state);
94
95        tokio::spawn(async move {
96            let served = axum::serve(listener, app).with_graceful_shutdown(async {
97                let _ = rx.await;
98            });
99            if let Err(e) = served.await {
100                tracing::error!(error = %e, "the http view stopped serving");
101            }
102        });
103
104        Ok(HttpView {
105            address,
106            shutdown: Some(tx),
107        })
108    }
109}
110
111struct Served {
112    store: Arc<dyn ObjectStore>,
113    exposures: Arc<dyn Exposures>,
114}
115
116/// The exposure context.
117///
118/// A published link is not a login. It carries exactly the right to read the one node
119/// it names, which is what `granted` clamps it to -- so a link that outlives a change
120/// in the owner's permissions cannot do more than read, and the enforcement layer
121/// below still has the final say.
122fn link_context() -> OpCtx {
123    OpCtx {
124        principal: "public".into(),
125        on_behalf_of: None,
126        protocol: "http",
127        granted: Perm::READ | Perm::META_READ,
128    }
129}
130
131async fn serve(
132    State(state): State<Arc<Served>>,
133    Path(token): Path<String>,
134    headers: HeaderMap,
135) -> Response {
136    let Some(exposed) = state.exposures.resolve(&token) else {
137        return not_found();
138    };
139    // A withdrawn link is indistinguishable from one that never existed. Answering 403
140    // would confirm that the token is real, which is information the holder of a
141    // revoked link should not get.
142    if !exposed.enabled {
143        return not_found();
144    }
145
146    let cx = link_context();
147    let meta = match state.store.stat(&cx, &exposed.node).await {
148        Ok(meta) => meta,
149        Err(_) => return not_found(),
150    };
151
152    let etag = meta
153        .etag
154        .as_ref()
155        .map(|tag| format!("\"{tag}\""))
156        .unwrap_or_default();
157
158    // A conditional request is answered before the object is opened: the point of both
159    // conditional headers is to avoid moving the bytes at all.
160    if is_unmodified(&headers, &meta, &etag) {
161        let mut response = Response::new(Body::empty());
162        *response.status_mut() = StatusCode::NOT_MODIFIED;
163        apply_validators(response.headers_mut(), &meta, &etag);
164        return response;
165    }
166
167    let requested = parse_range(
168        headers.get(header::RANGE).and_then(|v| v.to_str().ok()),
169        meta.size,
170    );
171    if requested == RangeSpec::Unsatisfiable {
172        let mut response = Response::new(Body::empty());
173        *response.status_mut() = StatusCode::RANGE_NOT_SATISFIABLE;
174        // The header a client needs to correct itself: how long the object actually is.
175        insert(
176            response.headers_mut(),
177            header::CONTENT_RANGE,
178            &format!("bytes */{}", meta.size),
179        );
180        return response;
181    }
182
183    let handle = match state.store.open(&cx, &exposed.node, OpenMode::Read).await {
184        Ok(handle) => handle,
185        Err(_) => return not_found(),
186    };
187
188    let body_range = requested.byte_range();
189    let stream = match handle.read_stream(body_range) {
190        Ok(stream) => stream,
191        Err(_) => return not_found(),
192    };
193
194    let (status, length) = match requested {
195        RangeSpec::Partial { start, end } => (StatusCode::PARTIAL_CONTENT, end - start + 1),
196        RangeSpec::Whole | RangeSpec::Unsatisfiable => (StatusCode::OK, meta.size),
197    };
198
199    // The handle has to outlive the stream that reads through it, so it rides along
200    // until the last chunk is produced.
201    let body = Body::from_stream(stream.map(move |chunk| {
202        let _keep_alive = &handle;
203        chunk.map_err(std::io::Error::other)
204    }));
205
206    let mut response = Response::new(body);
207    *response.status_mut() = status;
208    let out = response.headers_mut();
209    apply_validators(out, &meta, &etag);
210    insert(out, header::CONTENT_LENGTH, &length.to_string());
211    insert(
212        out,
213        header::CONTENT_TYPE,
214        meta.mime.as_deref().unwrap_or("application/octet-stream"),
215    );
216    insert(out, header::ACCEPT_RANGES, "bytes");
217    insert(
218        out,
219        header::CONTENT_DISPOSITION,
220        &disposition(&exposed, &meta),
221    );
222    if let RangeSpec::Partial { start, end } = requested {
223        insert(
224            out,
225            header::CONTENT_RANGE,
226            &format!("bytes {start}-{end}/{}", meta.size),
227        );
228    }
229    response
230}
231
232/// `ETag` and `Last-Modified`, on every response that carries either.
233///
234/// `Last-Modified` is an HTTP-date. Sending ISO 8601 here is the defect that made an
235/// S3 object visible in a listing and unreadable by the AWS SDK for Go, and it is one
236/// shared helper away from happening again.
237fn apply_validators(out: &mut HeaderMap, meta: &Meta, etag: &str) {
238    if !etag.is_empty() {
239        insert(out, header::ETAG, etag);
240    }
241    insert(
242        out,
243        header::LAST_MODIFIED,
244        &gfeh_core::http_date(meta.times.modified),
245    );
246}
247
248/// Whether the client's copy is still current, by whichever validator it sent.
249///
250/// `If-None-Match` wins outright when both are present, and not as a preference: RFC 9110
251/// §13.1.3 requires a recipient that evaluates an entity tag to ignore the date. The
252/// reason is worth knowing, because it is the reason to prefer the tag generally -- an
253/// HTTP-date has one-second resolution, so two writes within the same second produce one
254/// date and a client holding the first would be told it is current. An ETag is a function
255/// of the bytes and cannot make that mistake.
256///
257/// A date this server cannot parse is treated as no condition at all, which answers with
258/// the object: the cost of being strict is a re-download, and the cost of being lax is a
259/// client shown a stale file until something else changes.
260fn is_unmodified(headers: &HeaderMap, meta: &Meta, etag: &str) -> bool {
261    if let Some(candidate) = headers
262        .get(header::IF_NONE_MATCH)
263        .and_then(|v| v.to_str().ok())
264    {
265        return !etag.is_empty() && matches_etag(candidate, etag);
266    }
267
268    headers
269        .get(header::IF_MODIFIED_SINCE)
270        .and_then(|v| v.to_str().ok())
271        .and_then(gfeh_core::parse_http_date)
272        // The comparison is against the *sent* modification time rather than the stored
273        // one. `Last-Modified` is whole seconds, so an object modified at .400 is
274        // advertised at .000 and echoed back at .000; comparing the millisecond value
275        // would find it newer than its own header and re-send it on every request.
276        .is_some_and(|since| meta.times.modified.div_euclid(1_000) * 1_000 <= since)
277}
278
279/// Whether an `If-None-Match` value matches the object's tag.
280fn matches_etag(candidate: &str, etag: &str) -> bool {
281    if candidate.trim() == "*" {
282        return true;
283    }
284    candidate.split(',').any(|each| {
285        let each = each.trim();
286        // A weak validator compares equal to its strong counterpart for the purpose of
287        // a cache revalidation, which is the only comparison this route performs.
288        let each = each.strip_prefix("W/").unwrap_or(each);
289        each == etag
290    })
291}
292
293/// The filename a browser should save the object under.
294///
295/// Quoted and stripped of anything that would break out of the quoting: a name is
296/// user-controlled, and a header injection here would be a header injection into every
297/// response the daemon serves.
298fn disposition(exposed: &Exposed, meta: &Meta) -> String {
299    let name = exposed.filename.as_deref().unwrap_or(meta.name.as_str());
300    let safe: String = name
301        .chars()
302        .filter(|c| !matches!(c, '"' | '\\' | '\r' | '\n'))
303        .collect();
304    let safe = if safe.is_empty() {
305        "download".to_string()
306    } else {
307        safe
308    };
309    format!("inline; filename=\"{safe}\"")
310}
311
312fn not_found() -> Response {
313    (StatusCode::NOT_FOUND, "not found").into_response()
314}
315
316/// Set a header, dropping it rather than failing if the value is unrepresentable.
317fn insert(out: &mut HeaderMap, name: header::HeaderName, value: &str) {
318    if let Ok(value) = HeaderValue::from_str(value) {
319        out.insert(name, value);
320    }
321}
gfeh_http/server.rs

gfeh_http/
server.rs
