//! Images by hash over the API (oxygenflow#39): `flow rootfs push` sends //! an image's missing chunks and then its manifest; runners, and `flow //! rootfs pull`, fetch a manifest and the chunks they lack. use super::{caller, not_found, ApiError}; use crate::Shared; use axum::body::Bytes; use axum::extract::{Path, State}; use axum::http::{header, HeaderMap, StatusCode}; use axum::response::{IntoResponse, Response}; use axum::Json; use jet::images::{is_hash, Manifest}; use serde::{Deserialize, Serialize}; /// Most chunk names one `missing` question may ask about. const MAX_ASK: usize = 200_000; /// The body that many names make: 64 hex digits, quotes and a comma each, /// with room to spare (#147: the 2 MB default refused an 11 GB image's). pub const MAX_ASK_BYTES: usize = MAX_ASK * 80; /// Who sends images: an account whose token may write repositories. async fn sender(app: &Shared, headers: &HeaderMap) -> Result { let c = caller(app, headers).await?; Ok(c.require("repo:write")?.username.clone()) } /// Who fetches images: a runner, or any account (for `flow rootfs pull`). async fn fetcher(app: &Shared, headers: &HeaderMap) -> Result<(), ApiError> { if crate::flow::runner_auth(app, headers).await.is_ok() { return Ok(()); } let c = caller(app, headers).await?; c.account.as_ref().map(|_| ()).ok_or_else(|| ApiError(StatusCode::UNAUTHORIZED, "a runner or personal access token is required".into())) } #[derive(Deserialize)] pub struct Ask { chunks: Vec, } #[derive(Serialize)] pub struct Missing { missing: Vec, } /// `POST /flow/images/missing {"chunks": [..]}`: which of these chunks the /// hub lacks. pub async fn missing(State(app): State, headers: HeaderMap, Json(ask): Json) -> Result, ApiError> { sender(&app, &headers).await?; if ask.chunks.len() > MAX_ASK { return Err(ApiError(StatusCode::BAD_REQUEST, format!("ask about at most {MAX_ASK} chunks at once"))); } let store = app.images(); let missing = tokio::task::spawn_blocking(move || { let mut seen = std::collections::HashSet::new(); ask.chunks.into_iter().filter(|h| is_hash(h) && seen.insert(h.clone()) && !store.has_chunk(h)).collect() }) .await .map_err(|e| ApiError(StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?; Ok(Json(Missing { missing })) } /// `PUT /flow/images/chunks/{hash}`: one chunk, checked against its name. pub async fn put_chunk( State(app): State, headers: HeaderMap, Path(hash): Path, body: Bytes, ) -> Result { sender(&app, &headers).await?; if !is_hash(&hash) || body.is_empty() || body.len() > oxy_core::chunk::MAX_SIZE { return Err(ApiError(StatusCode::BAD_REQUEST, "not a chunk: a name is 64 hex digits, a chunk 1 byte to 256 KiB".into())); } if blake3::hash(&body).to_hex().as_str() != hash { return Err(ApiError(StatusCode::BAD_REQUEST, format!("those bytes are not chunk {}", &hash[..12]))); } let store = app.images(); tokio::task::spawn_blocking(move || store.put_chunk(&hash, &body)) .await .map_err(|e| ApiError(StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))? .map_err(|e| ApiError(StatusCode::INTERNAL_SERVER_ERROR, format!("keep the chunk: {e}")))?; Ok(StatusCode::NO_CONTENT) } /// `PUT /flow/images/{hash}`: the manifest, accepted once every chunk is /// here and the image they make has this hash. pub async fn put_manifest( State(app): State, headers: HeaderMap, Path(hash): Path, Json(m): Json, ) -> Result<(StatusCode, Json), ApiError> { let who = sender(&app, &headers).await?; if m.hash != hash { return Err(ApiError(StatusCode::BAD_REQUEST, format!("the manifest is for {}, not {}", m.hash, hash))); } // In a task of its own (#341): reading a big image can outlast the // sender's wait, and a sender that gives up must not leave an image // kept but never offered. let m = tokio::spawn(async move { let store = app.images(); // A manifest kept before was verified then; only its row may be missing. let m = match store.manifest(&m.hash).filter(|k| k.chunks == m.chunks) { Some(kept) => kept, // Kept with the architecture its own files name (oxygenhub#173). None => tokio::task::spawn_blocking(move || store.keep_reading_arch(&m)) .await .map_err(|e| ApiError(StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))? .map_err(|e| ApiError(StatusCode::CONFLICT, e))?, }; crate::images::record(&app.db, &m, &who).await?; crate::db::accounts::audit(&app.db, None, "flow.image", &who, &format!("{} ({} MiB)", &m.hash[..12], m.size >> 20)).await.ok(); Ok::<_, ApiError>(m) }) .await .map_err(|e| ApiError(StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))??; Ok((StatusCode::CREATED, Json(serde_json::json!({ "hash": m.hash, "size": m.size, "chunks": m.chunks.len(), "arch": m.arch })))) } /// `GET /flow/images/{hash}`: an image's manifest. pub async fn get_manifest(State(app): State, headers: HeaderMap, Path(hash): Path) -> Result, ApiError> { fetcher(&app, &headers).await?; app.images().manifest(&hash).map(Json).ok_or_else(|| not_found("this hub does not hold that image; send it with flow rootfs push")) } /// `GET /flow/images/chunks/{hash}`: one chunk's bytes. pub async fn get_chunk(State(app): State, headers: HeaderMap, Path(hash): Path) -> Result { fetcher(&app, &headers).await?; let store = app.images(); let bytes = tokio::task::spawn_blocking(move || store.chunk(&hash)) .await .map_err(|e| ApiError(StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))? .map_err(|_| not_found("no such chunk"))?; Ok(([(header::CONTENT_TYPE, "application/octet-stream"), (header::CACHE_CONTROL, "private, max-age=31536000, immutable")], bytes) .into_response()) }