diff --git a/crates/directory/src/api.rs b/crates/directory/src/api.rs new file mode 100644 index 0000000..ff6dcd5 --- /dev/null +++ b/crates/directory/src/api.rs @@ -0,0 +1,428 @@ +//! The HTTP side: the router, the token check, and `/v1`, where ipx servers send snapshots and read the +//! catalogue. + +use crate::config::Access; +use crate::directory::Shared; +use crate::wire::{Blocked, FeedPage, Instance, Listing, MAX_SNAPSHOT, Me, SnapshotEntry, SnapshotResult}; +use crate::{otel, site}; +use axum::extract::{Path, Query, Request, State}; +use axum::http::{StatusCode, header}; +use axum::middleware::Next; +use axum::response::{IntoResponse, Response}; +use axum::routing::{delete, get, post}; +use axum::{Extension, Json, Router}; +use serde::Deserialize; +use std::time::Instant; +use tower_http::compression::CompressionLayer; +use tower_http::decompression::RequestDecompressionLayer; +use tracing::Instrument; + +/// Every route: the API under `/v1`, behind a token, and the public pages. +pub fn router(dir: Shared) -> Router { + let v1 = Router::new() + .route("/v1/me", get(me)) + .route("/v1/feeds", get(list)) + .route("/v1/popular", get(popular)) + .route("/v1/feeds/{id}", get(feed).delete(remove)) + .route("/v1/feeds/{id}/refresh", post(refresh)) + .route("/v1/snapshot", post(snapshot)) + .route("/v1/blocked", get(blocked)) + .route("/v1/blocked/{id}", delete(unblock)) + .route("/v1/instances", get(instances)) + .route("/v1/instances/{name}", delete(drop_instance)) + .layer(axum::middleware::from_fn_with_state(dir.clone(), authenticate)); + Router::new() + .route("/healthz", get(healthz)) + .merge(v1) + // Anyone may browse, no token, read only (#215). + .merge(site::router()) + // Brotli, zstd or gzip as the caller takes them: the feed list went to ipx as 603 KB of + // JSON, most of a Directory page's time (#253). Inside the access log, which records what + // was sent. + .layer(CompressionLayer::new().br(true).zstd(true).gzip(true)) + // And takes a request's body gzipped: ipx sends its snapshot so, about 100 KB as JSON. + .layer(RequestDecompressionLayer::new().gzip(true)) + .layer(axum::middleware::from_fn(access_log)) + .with_state(dir) +} + +pub(crate) struct ApiError(StatusCode, String); + +impl IntoResponse for ApiError { + fn into_response(self) -> Response { + (self.0, Json(serde_json::json!({ "error": self.1 }))).into_response() + } +} + +impl From for ApiError { + fn from(e: anyhow::Error) -> Self { + tracing::error!(error = %format!("{e:#}"), "request failed"); + ApiError(StatusCode::INTERNAL_SERVER_ERROR, "something went wrong; the directory's log says what".into()) + } +} + +/// Who made the request, for the access log, set by `authenticate`. +#[derive(Clone)] +pub(crate) struct Who(String); + +/// One line a request, with the token's name and never the token: what the Grafana panels read. +pub(crate) async fn access_log(req: Request, next: Next) -> Response { + let (method, path) = (req.method().clone(), req.uri().path().to_owned()); + let start = Instant::now(); + // A span for each request but the healthcheck's, a child of the caller's when it sent a + // traceparent, as ipx does on its calls here (#213). + let span = if path == "/healthz" { + tracing::Span::none() + } else { + tracing::info_span!( + "http", + otel.name = %format!("{method} {path}"), + http.request.method = %method, + url.path = %path, + http.response.status_code = tracing::field::Empty, + ) + }; + if let Some(parent) = otel::remote_parent(req.headers(), None) { + use tracing_opentelemetry::OpenTelemetrySpanExt; + let _ = span.set_parent(parent); + } + let resp = Instrument::instrument(next.run(req), span.clone()).await; + span.record("http.response.status_code", resp.status().as_u16()); + let _in = span.enter(); + let who = resp.extensions().get::().map_or("", |w| w.0.as_str()).to_owned(); + let status = resp.status().as_u16(); + if path != "/healthz" { + tracing::info!( + target: "ipx_directory::http", + method = %method, path = %path, status, duration_ms = start.elapsed().as_millis() as u64, token = %who, + "{method} {path} -> {status}" + ); + } + resp +} + +pub(crate) async fn authenticate(State(dir): State, mut req: Request, next: Next) -> Response { + let presented = req + .headers() + .get(header::AUTHORIZATION) + .and_then(|v| v.to_str().ok()) + .and_then(|v| v.strip_prefix("Bearer ")) + .map(str::trim) + .filter(|t| !t.is_empty()); + let Some(me) = presented.and_then(|t| dir.caller(t)) else { + return ApiError(StatusCode::UNAUTHORIZED, "a valid token is needed: Authorization: Bearer ".into()) + .into_response(); + }; + let who = Who(me.name.clone()); + req.extensions_mut().insert(me); + let mut resp = next.run(req).await; + resp.extensions_mut().insert(who); + resp +} + +pub(crate) fn need(me: &Me, access: Access) -> Result<(), ApiError> { + if me.access >= access { + Ok(()) + } else { + let what = match access { + Access::Ro => "browse", + Access::Rw => "send snapshots", + Access::Rwa => "administer the directory", + }; + Err(ApiError(StatusCode::FORBIDDEN, format!("this token may not {what}"))) + } +} + +pub(crate) async fn healthz(State(dir): State) -> Result<&'static str, ApiError> { + dir.store.ping().await?; + Ok("ok") +} + +pub(crate) async fn me(Extension(me): Extension) -> Json { + Json(me) +} + +pub(crate) async fn list(State(dir): State) -> Result>, ApiError> { + Ok(Json(dir.store.list().await?)) +} + +#[derive(Deserialize)] +pub(crate) struct Limit { + limit: Option, +} + +pub(crate) async fn popular(State(dir): State, Query(q): Query) -> Result>, ApiError> { + Ok(Json(dir.store.popular(q.limit.unwrap_or(10).clamp(1, 100)).await?)) +} + +pub(crate) async fn feed(State(dir): State, Path(id): Path) -> Result, ApiError> { + dir.store + .feed(&id) + .await? + .map(Json) + .ok_or_else(|| ApiError(StatusCode::NOT_FOUND, format!("{id:?} is not in the directory"))) +} + +pub(crate) async fn snapshot( + State(dir): State, + Extension(me): Extension, + Json(entries): Json>, +) -> Result, ApiError> { + need(&me, Access::Rw)?; + if entries.len() > MAX_SNAPSHOT { + return Err(ApiError(StatusCode::PAYLOAD_TOO_LARGE, format!("a snapshot lists at most {MAX_SNAPSHOT} feeds"))); + } + let r = dir.store.snapshot(&me.name, &entries).await?; + dir.wake.notify_one(); + tracing::info!( + instance = %me.name, accepted = r.accepted, refused_private = r.refused_private, + refused_blocked = r.refused_blocked, refused_invalid = r.refused_invalid, "snapshot" + ); + Ok(Json(r)) +} + +pub(crate) async fn remove( + State(dir): State, + Extension(me): Extension, + Path(id): Path, +) -> Result { + need(&me, Access::Rwa)?; + dir.store.block(&id).await?; + tracing::info!(feed = %id, by = %me.name, "feed removed and blocked"); + Ok(StatusCode::NO_CONTENT) +} + +pub(crate) async fn blocked( + State(dir): State, + Extension(me): Extension, +) -> Result>, ApiError> { + need(&me, Access::Rwa)?; + Ok(Json(dir.store.blocked().await?)) +} + +pub(crate) async fn unblock( + State(dir): State, + Extension(me): Extension, + Path(id): Path, +) -> Result { + need(&me, Access::Rwa)?; + if !dir.store.unblock(&id).await? { + return Err(ApiError(StatusCode::NOT_FOUND, format!("{id:?} is not blocked"))); + } + tracing::info!(feed = %id, by = %me.name, "feed unblocked"); + Ok(StatusCode::NO_CONTENT) +} + +pub(crate) async fn instances( + State(dir): State, + Extension(me): Extension, +) -> Result>, ApiError> { + need(&me, Access::Rwa)?; + Ok(Json(dir.store.instances().await?)) +} + +pub(crate) async fn drop_instance( + State(dir): State, + Extension(me): Extension, + Path(name): Path, +) -> Result { + need(&me, Access::Rwa)?; + let n = dir.store.drop_instance(&name).await?; + tracing::info!(instance = %name, feeds = n, by = %me.name, "snapshot dropped"); + Ok(StatusCode::NO_CONTENT) +} + +pub(crate) async fn refresh( + State(dir): State, + Extension(me): Extension, + Path(id): Path, +) -> Result, ApiError> { + need(&me, Access::Rwa)?; + let due = dir + .store + .due_one(&id) + .await? + .ok_or_else(|| ApiError(StatusCode::NOT_FOUND, format!("{id:?} is not in the directory")))?; + dir.refresh(&due).await?; + feed(State(dir), Path(id)).await +} + +#[cfg(test)] +mod tests { + + use super::*; + use crate::config; + use crate::config::Token; + use crate::directory::Directory; + use crate::store; + use crate::store::Store; + use axum::body::Body; + use tower::ServiceExt; + + async fn app() -> (Router, Shared) { + let tokens = ["ro", "rw", "rwa"] + .iter() + .map(|n| Token { + name: format!("{n}-token"), + access: serde_json::from_value(serde_json::json!(n)).unwrap(), + sha256: config::hash(&format!("secret-{n}")), + }) + .collect(); + let dir = Directory::new(Store::scratch().await, tokens, Default::default()).unwrap(); + (router(dir.clone()), dir) + } + + async fn call( + app: &Router, + method: &str, + path: &str, + token: Option<&str>, + body: Option, + ) -> (u16, serde_json::Value) { + let mut req = axum::http::Request::builder().method(method).uri(path); + if let Some(t) = token { + req = req.header("authorization", format!("Bearer secret-{t}")); + } + let req = match body { + Some(b) => req.header("content-type", "application/json").body(Body::from(b.to_string())), + None => req.body(Body::empty()), + } + .unwrap(); + let resp = app.clone().oneshot(req).await.unwrap(); + let status = resp.status().as_u16(); + let bytes = axum::body::to_bytes(resp.into_body(), 1 << 20).await.unwrap(); + (status, serde_json::from_slice(&bytes).unwrap_or(serde_json::Value::Null)) + } + + #[tokio::test] + async fn a_snapshot_may_come_gzipped() { + use std::io::Write as _; + let (app, _) = app().await; + let json = serde_json::json!([{"url": "https://example.com/feed.xml", "subscribers": 2}]).to_string(); + let mut gz = flate2::write::GzEncoder::new(Vec::new(), flate2::Compression::default()); + gz.write_all(json.as_bytes()).unwrap(); + // As ipx sends it (#253). + let req = axum::http::Request::builder() + .method("POST") + .uri("/v1/snapshot") + .header("authorization", "Bearer secret-rw") + .header("content-type", "application/json") + .header("content-encoding", "gzip") + .body(Body::from(gz.finish().unwrap())) + .unwrap(); + let resp = app.clone().oneshot(req).await.unwrap(); + assert_eq!(resp.status(), 200); + let got: serde_json::Value = + serde_json::from_slice(&axum::body::to_bytes(resp.into_body(), 1 << 20).await.unwrap()).unwrap(); + assert_eq!(got["accepted"], 1); + } + + #[tokio::test] + async fn answers_come_compressed_as_the_caller_asks() { + let (app, _) = app().await; + for (asked, sent) in [("br", Some("br")), ("zstd", Some("zstd")), ("gzip", Some("gzip")), ("identity", None)] { + // A page, not an empty feed list: under 32 bytes is sent as it is. + let req = + axum::http::Request::builder().uri("/").header("accept-encoding", asked).body(Body::empty()).unwrap(); + let resp = app.clone().oneshot(req).await.unwrap(); + assert_eq!(resp.status(), 200); + let got = resp.headers().get("content-encoding").map(|v| v.to_str().unwrap().to_owned()); + assert_eq!(got.as_deref(), sent, "asked for {asked} (#253)"); + } + } + + async fn get_page(app: &Router, path: &str) -> (u16, String, Option) { + let req = axum::http::Request::builder().uri(path).body(Body::empty()).unwrap(); + let resp = app.clone().oneshot(req).await.unwrap(); + let status = resp.status().as_u16(); + let csp = resp.headers().get("content-security-policy").map(|v| v.to_str().unwrap().to_owned()); + let bytes = axum::body::to_bytes(resp.into_body(), 1 << 22).await.unwrap(); + (status, String::from_utf8(bytes.to_vec()).unwrap(), csp) + } + + #[tokio::test] + async fn anyone_browses_the_pages_and_the_api_still_needs_a_token() { + let (app, dir) = app().await; + let snap = serde_json::json!([{ "url": "https://x.test/a.xml", "subscribers": 3 }]); + call(&app, "POST", "/v1/snapshot", Some("rw"), Some(snap)).await; + store::tests::fetch_all(&dir.store).await; + let id = dir.store.list().await.unwrap()[0].id.clone(); + + let (s, home, csp) = get_page(&app, "/").await; + assert_eq!(s, 200); + assert!(csp.unwrap().contains("script-src 'self'")); + assert!(home.contains("Most subscribed") && home.contains(&format!("/feed/{id}")), "{home}"); + assert!(get_page(&app, "/all").await.1.contains("3 people subscribe")); + let (s, page, _) = get_page(&app, &format!("/feed/{id}")).await; + assert_eq!(s, 200); + // Its address, to subscribe to anywhere. + assert!(page.contains("value=\"https://x.test/a.xml\""), "{page}"); + assert!(!page.contains("