gfeh-http 0.1.2

//! The HTTP listener and its one route.

use axum::Router;
use axum::body::Body;
use axum::extract::{Path, State};
use axum::http::{HeaderMap, HeaderValue, StatusCode, header};
use axum::response::{IntoResponse, Response};
use axum::routing::get;
use futures::StreamExt;
use gfeh_core::{Meta, ObjectStore, OpCtx, OpenMode, Perm, RangeSpec, parse_range};
use std::net::SocketAddr;
use std::sync::Arc;
use tokio::net::TcpListener;

use crate::exposure::{Exposed, Exposures};

/// A running HTTP view.
#[derive(Debug)]
pub struct HttpView {
    address: SocketAddr,
    shutdown: Option<tokio::sync::oneshot::Sender<()>>,
}

impl HttpView {
    /// Start building.
    #[must_use]
    pub fn builder(store: Arc<dyn ObjectStore>, exposures: Arc<dyn Exposures>) -> HttpBuilder {
        HttpBuilder {
            store,
            exposures,
            bind: SocketAddr::from(([127, 0, 0, 1], 0)),
        }
    }

    /// Where it is listening.
    #[must_use]
    pub fn address(&self) -> SocketAddr {
        self.address
    }

    /// The base URL, without a trailing slash.
    #[must_use]
    pub fn base_url(&self) -> String {
        format!("http://{}", self.address)
    }

    /// The URL a token resolves at.
    #[must_use]
    pub fn url_for(&self, token: &str) -> String {
        format!("{}/f/{token}", self.base_url())
    }
}

impl Drop for HttpView {
    fn drop(&mut self) {
        if let Some(shutdown) = self.shutdown.take() {
            let _ = shutdown.send(());
        }
    }
}

/// Assembles an [`HttpView`].
pub struct HttpBuilder {
    store: Arc<dyn ObjectStore>,
    exposures: Arc<dyn Exposures>,
    bind: SocketAddr,
}

impl HttpBuilder {
    /// Listen on this address. Port zero means an ephemeral port.
    #[must_use]
    pub fn bind(mut self, addr: SocketAddr) -> Self {
        self.bind = addr;
        self
    }

    /// Bind and serve.
    ///
    /// # Errors
    ///
    /// Returns the bind failure if the address is unavailable.
    pub async fn start(self) -> std::io::Result<HttpView> {
        let listener = TcpListener::bind(self.bind).await?;
        let address = listener.local_addr()?;
        let (tx, rx) = tokio::sync::oneshot::channel();

        let state = Arc::new(Served {
            store: self.store,
            exposures: self.exposures,
        });
        let app = Router::new()
            .route("/f/{token}", get(serve).head(serve))
            .with_state(state);

        tokio::spawn(async move {
            let served = axum::serve(listener, app).with_graceful_shutdown(async {
                let _ = rx.await;
            });
            if let Err(e) = served.await {
                tracing::error!(error = %e, "the http view stopped serving");
            }
        });

        Ok(HttpView {
            address,
            shutdown: Some(tx),
        })
    }
}

struct Served {
    store: Arc<dyn ObjectStore>,
    exposures: Arc<dyn Exposures>,
}

/// The exposure context.
///
/// A published link is not a login. It carries exactly the right to read the one node
/// it names, which is what `granted` clamps it to -- so a link that outlives a change
/// in the owner's permissions cannot do more than read, and the enforcement layer
/// below still has the final say.
fn link_context() -> OpCtx {
    OpCtx {
        principal: "public".into(),
        on_behalf_of: None,
        protocol: "http",
        granted: Perm::READ | Perm::META_READ,
    }
}

async fn serve(
    State(state): State<Arc<Served>>,
    Path(token): Path<String>,
    headers: HeaderMap,
) -> Response {
    let Some(exposed) = state.exposures.resolve(&token) else {
        return not_found();
    };
    // A withdrawn link is indistinguishable from one that never existed. Answering 403
    // would confirm that the token is real, which is information the holder of a
    // revoked link should not get.
    if !exposed.enabled {
        return not_found();
    }

    let cx = link_context();
    let meta = match state.store.stat(&cx, &exposed.node).await {
        Ok(meta) => meta,
        Err(_) => return not_found(),
    };

    let etag = meta
        .etag
        .as_ref()
        .map(|tag| format!("\"{tag}\""))
        .unwrap_or_default();

    // A conditional request is answered before the object is opened: the point of both
    // conditional headers is to avoid moving the bytes at all.
    if is_unmodified(&headers, &meta, &etag) {
        let mut response = Response::new(Body::empty());
        *response.status_mut() = StatusCode::NOT_MODIFIED;
        apply_validators(response.headers_mut(), &meta, &etag);
        return response;
    }

    let requested = parse_range(
        headers.get(header::RANGE).and_then(|v| v.to_str().ok()),
        meta.size,
    );
    if requested == RangeSpec::Unsatisfiable {
        let mut response = Response::new(Body::empty());
        *response.status_mut() = StatusCode::RANGE_NOT_SATISFIABLE;
        // The header a client needs to correct itself: how long the object actually is.
        insert(
            response.headers_mut(),
            header::CONTENT_RANGE,
            &format!("bytes */{}", meta.size),
        );
        return response;
    }

    let handle = match state.store.open(&cx, &exposed.node, OpenMode::Read).await {
        Ok(handle) => handle,
        Err(_) => return not_found(),
    };

    let body_range = requested.byte_range();
    let stream = match handle.read_stream(body_range) {
        Ok(stream) => stream,
        Err(_) => return not_found(),
    };

    let (status, length) = match requested {
        RangeSpec::Partial { start, end } => (StatusCode::PARTIAL_CONTENT, end - start + 1),
        RangeSpec::Whole | RangeSpec::Unsatisfiable => (StatusCode::OK, meta.size),
    };

    // The handle has to outlive the stream that reads through it, so it rides along
    // until the last chunk is produced.
    let body = Body::from_stream(stream.map(move |chunk| {
        let _keep_alive = &handle;
        chunk.map_err(std::io::Error::other)
    }));

    let mut response = Response::new(body);
    *response.status_mut() = status;
    let out = response.headers_mut();
    apply_validators(out, &meta, &etag);
    insert(out, header::CONTENT_LENGTH, &length.to_string());
    insert(
        out,
        header::CONTENT_TYPE,
        meta.mime.as_deref().unwrap_or("application/octet-stream"),
    );
    insert(out, header::ACCEPT_RANGES, "bytes");
    insert(
        out,
        header::CONTENT_DISPOSITION,
        &disposition(&exposed, &meta),
    );
    if let RangeSpec::Partial { start, end } = requested {
        insert(
            out,
            header::CONTENT_RANGE,
            &format!("bytes {start}-{end}/{}", meta.size),
        );
    }
    response
}

/// `ETag` and `Last-Modified`, on every response that carries either.
///
/// `Last-Modified` is an HTTP-date. Sending ISO 8601 here is the defect that made an
/// S3 object visible in a listing and unreadable by the AWS SDK for Go, and it is one
/// shared helper away from happening again.
fn apply_validators(out: &mut HeaderMap, meta: &Meta, etag: &str) {
    if !etag.is_empty() {
        insert(out, header::ETAG, etag);
    }
    insert(
        out,
        header::LAST_MODIFIED,
        &gfeh_core::http_date(meta.times.modified),
    );
}

/// Whether the client's copy is still current, by whichever validator it sent.
///
/// `If-None-Match` wins outright when both are present, and not as a preference: RFC 9110
/// §13.1.3 requires a recipient that evaluates an entity tag to ignore the date. The
/// reason is worth knowing, because it is the reason to prefer the tag generally -- an
/// HTTP-date has one-second resolution, so two writes within the same second produce one
/// date and a client holding the first would be told it is current. An ETag is a function
/// of the bytes and cannot make that mistake.
///
/// A date this server cannot parse is treated as no condition at all, which answers with
/// the object: the cost of being strict is a re-download, and the cost of being lax is a
/// client shown a stale file until something else changes.
fn is_unmodified(headers: &HeaderMap, meta: &Meta, etag: &str) -> bool {
    if let Some(candidate) = headers
        .get(header::IF_NONE_MATCH)
        .and_then(|v| v.to_str().ok())
    {
        return !etag.is_empty() && matches_etag(candidate, etag);
    }

    headers
        .get(header::IF_MODIFIED_SINCE)
        .and_then(|v| v.to_str().ok())
        .and_then(gfeh_core::parse_http_date)
        // The comparison is against the *sent* modification time rather than the stored
        // one. `Last-Modified` is whole seconds, so an object modified at .400 is
        // advertised at .000 and echoed back at .000; comparing the millisecond value
        // would find it newer than its own header and re-send it on every request.
        .is_some_and(|since| meta.times.modified.div_euclid(1_000) * 1_000 <= since)
}

/// Whether an `If-None-Match` value matches the object's tag.
fn matches_etag(candidate: &str, etag: &str) -> bool {
    if candidate.trim() == "*" {
        return true;
    }
    candidate.split(',').any(|each| {
        let each = each.trim();
        // A weak validator compares equal to its strong counterpart for the purpose of
        // a cache revalidation, which is the only comparison this route performs.
        let each = each.strip_prefix("W/").unwrap_or(each);
        each == etag
    })
}

/// The filename a browser should save the object under.
///
/// Quoted and stripped of anything that would break out of the quoting: a name is
/// user-controlled, and a header injection here would be a header injection into every
/// response the daemon serves.
fn disposition(exposed: &Exposed, meta: &Meta) -> String {
    let name = exposed.filename.as_deref().unwrap_or(meta.name.as_str());
    let safe: String = name
        .chars()
        .filter(|c| !matches!(c, '"' | '\\' | '\r' | '\n'))
        .collect();
    let safe = if safe.is_empty() {
        "download".to_string()
    } else {
        safe
    };
    format!("inline; filename=\"{safe}\"")
}

fn not_found() -> Response {
    (StatusCode::NOT_FOUND, "not found").into_response()
}

/// Set a header, dropping it rather than failing if the value is unrepresentable.
fn insert(out: &mut HeaderMap, name: header::HeaderName, value: &str) {
    if let Ok(value) = HeaderValue::from_str(value) {
        out.insert(name, value);
    }
}