OXYGENHUB
# dd0788b756f2 oxygenhub / crates / hub / src / api / images.rs
5.4 KiB · Rust · last changed The hub reads what each image is for, and learns the rest from passes 22 days ago source blame history raw
1
//! Images by hash over the API (oxygenflow#39): `flow rootfs push` sends
2
//! an image's missing chunks and then its manifest; runners, and `flow
3
//! rootfs pull`, fetch a manifest and the chunks they lack.
4
5
use super::{caller, not_found, ApiError};
6
use crate::Shared;
7
use axum::body::Bytes;
8
use axum::extract::{Path, State};
9
use axum::http::{header, HeaderMap, StatusCode};
10
use axum::response::{IntoResponse, Response};
11
use axum::Json;
12
use jet::images::{is_hash, Manifest};
13
use serde::{Deserialize, Serialize};
14
15
/// Most chunk names one `missing` question may ask about.
16
const MAX_ASK: usize = 200_000;
17
/// The body that many names make: 64 hex digits, quotes and a comma each,
18
/// with room to spare (#147: the 2 MB default refused an 11 GB image's).
19
pub const MAX_ASK_BYTES: usize = MAX_ASK * 80;
20
21
/// Who sends images: an account whose token may write repositories.
22
async fn sender(app: &Shared, headers: &HeaderMap) -> Result<String, ApiError> {
23
    let c = caller(app, headers).await?;
24
    Ok(c.require("repo:write")?.username.clone())
25
}
26
27
/// Who fetches images: a runner, or any account (for `flow rootfs pull`).
28
async fn fetcher(app: &Shared, headers: &HeaderMap) -> Result<(), ApiError> {
29
    if crate::flow::runner_auth(app, headers).await.is_ok() {
30
        return Ok(());
31
    }
32
    let c = caller(app, headers).await?;
33
    c.account.as_ref().map(|_| ()).ok_or_else(|| ApiError(StatusCode::UNAUTHORIZED, "a runner or personal access token is required".into()))
34
}
35
36
#[derive(Deserialize)]
37
pub struct Ask {
38
    chunks: Vec<String>,
39
}
40
41
#[derive(Serialize)]
42
pub struct Missing {
43
    missing: Vec<String>,
44
}
45
46
/// `POST /flow/images/missing {"chunks": [..]}`: which of these chunks the
47
/// hub lacks.
48
pub async fn missing(State(app): State<Shared>, headers: HeaderMap, Json(ask): Json<Ask>) -> Result<Json<Missing>, ApiError> {
49
    sender(&app, &headers).await?;
50
    if ask.chunks.len() > MAX_ASK {
51
        return Err(ApiError(StatusCode::BAD_REQUEST, format!("ask about at most {MAX_ASK} chunks at once")));
52
    }
53
    let store = app.images();
54
    let missing = tokio::task::spawn_blocking(move || {
55
        let mut seen = std::collections::HashSet::new();
56
        ask.chunks.into_iter().filter(|h| is_hash(h) && seen.insert(h.clone()) && !store.has_chunk(h)).collect()
57
    })
58
    .await
59
    .map_err(|e| ApiError(StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?;
60
    Ok(Json(Missing { missing }))
61
}
62
63
/// `PUT /flow/images/chunks/{hash}`: one chunk, checked against its name.
64
pub async fn put_chunk(
65
    State(app): State<Shared>,
66
    headers: HeaderMap,
67
    Path(hash): Path<String>,
68
    body: Bytes,
69
) -> Result<StatusCode, ApiError> {
70
    sender(&app, &headers).await?;
71
    if !is_hash(&hash) || body.is_empty() || body.len() > oxy_core::chunk::MAX_SIZE {
72
        return Err(ApiError(StatusCode::BAD_REQUEST, "not a chunk: a name is 64 hex digits, a chunk 1 byte to 256 KiB".into()));
73
    }
74
    if blake3::hash(&body).to_hex().as_str() != hash {
75
        return Err(ApiError(StatusCode::BAD_REQUEST, format!("those bytes are not chunk {}", &hash[..12])));
76
    }
77
    let store = app.images();
78
    tokio::task::spawn_blocking(move || store.put_chunk(&hash, &body))
79
        .await
80
        .map_err(|e| ApiError(StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?
81
        .map_err(|e| ApiError(StatusCode::INTERNAL_SERVER_ERROR, format!("keep the chunk: {e}")))?;
82
    Ok(StatusCode::NO_CONTENT)
83
}
84
85
/// `PUT /flow/images/{hash}`: the manifest, accepted once every chunk is
86
/// here and the image they make has this hash.
87
pub async fn put_manifest(
88
    State(app): State<Shared>,
89
    headers: HeaderMap,
90
    Path(hash): Path<String>,
91
    Json(m): Json<Manifest>,
92
) -> Result<(StatusCode, Json<serde_json::Value>), ApiError> {
93
    let who = sender(&app, &headers).await?;
94
    if m.hash != hash {
95
        return Err(ApiError(StatusCode::BAD_REQUEST, format!("the manifest is for {}, not {}", m.hash, hash)));
96
    }
97
    let store = app.images();
98
    // Kept with the architecture its own files name (oxygenhub#173).
99
    let m = tokio::task::spawn_blocking(move || store.keep_reading_arch(&m))
100
        .await
101
        .map_err(|e| ApiError(StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?
102
        .map_err(|e| ApiError(StatusCode::CONFLICT, e))?;
103
    crate::images::record(&app.db, &m, &who).await?;
104
    crate::db::accounts::audit(&app.db, None, "flow.image", &who, &format!("{} ({} MiB)", &m.hash[..12], m.size >> 20)).await.ok();
105
    Ok((StatusCode::CREATED, Json(serde_json::json!({ "hash": m.hash, "size": m.size, "chunks": m.chunks.len(), "arch": m.arch }))))
106
}
107
108
/// `GET /flow/images/{hash}`: an image's manifest.
109
pub async fn get_manifest(State(app): State<Shared>, headers: HeaderMap, Path(hash): Path<String>) -> Result<Json<Manifest>, ApiError> {
110
    fetcher(&app, &headers).await?;
111
    app.images().manifest(&hash).map(Json).ok_or_else(|| not_found("this hub does not hold that image; send it with flow rootfs push"))
112
}
113
114
/// `GET /flow/images/chunks/{hash}`: one chunk's bytes.
115
pub async fn get_chunk(State(app): State<Shared>, headers: HeaderMap, Path(hash): Path<String>) -> Result<Response, ApiError> {
116
    fetcher(&app, &headers).await?;
117
    let store = app.images();
118
    let bytes = tokio::task::spawn_blocking(move || store.chunk(&hash))
119
        .await
120
        .map_err(|e| ApiError(StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?
121
        .map_err(|_| not_found("no such chunk"))?;
122
    Ok(([(header::CONTENT_TYPE, "application/octet-stream"), (header::CACHE_CONTROL, "private, max-age=31536000, immutable")], bytes)
123
        .into_response())
124
}