johnandersen777.bsky.social/knotless
git that lives in your Atmosphere account
Forked from chadtmiller.com/knotless
0
0
//! An OCI registry of the images repositories publish as releases
//! (`knotless release publish --oci`, docs/design/packages.md):
//!
//! ```text
//! docker pull knotless.dev/<owner>/<repo>/<package>:<version>
//! docker pull knotless.dev/<owner>/<repo>:latest # the package named as the repository
//! docker login knotless.dev -u <handle> # an app password
//! docker push knotless.dev/<owner>/<repo>/<package>:<version>
//! ```
//!
//! A tag is a version, or `latest`, the highest without a pre-release
//! part; its manifest is the one its release's `index.json` names. Every
//! blob is a file of a release, named for its digest, read from the
//! owner's PDS once, checked against its record, and kept.
//!
//! A push is the owner's alone, with a token `/v2/token` gives for their
//! app password, which logs in to their PDS and is kept nowhere. Each blob
//! pushed is checked against its digest and uploaded to their PDS as it
//! lands; a manifest pushed by a tag, a version, makes the release of it
//! and every blob it reaches, with an `index.json` naming it. A version is
//! never pushed twice, and `latest` is never pushed: it is computed.
// A handler's error is the answer it sends, made once per request, so a
// large one costs nothing to move.
#![allow(clippy::result_large_err)]
use crate::releases::{Records, base_url, kept_dir};
use axum::Router;
use axum::body::Body;
use axum::extract::{Path as Params, Query, State};
use axum::http::{HeaderMap, HeaderName, StatusCode, header};
use axum::response::{IntoResponse, Response};
use axum::routing::get;
use futures::StreamExt;
use knotless::env::Ctx;
use knotless::oci;
use knotless::push;
use knotless::records::{RELEASE_MIME, ReleaseFile, ReleasePart, ReleaseRecord};
use knotless::release::{self, Made, Target};
use knotless::remote::Remote;
use knotless::url::RepoUrl;
use knotless::xrpc::Auth;
use serde_json::{Value, json};
use std::collections::HashMap;
use std::path::PathBuf;
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};
use tokio::io::AsyncWriteExt;
/// The most packages whose releases are kept listed at once, and the most
/// blobs pushed and not yet released: past either, all are forgotten.
const MOST_KEPT: usize = 1024;
/// How long a token for pushing lasts.
const TOKEN_FOR: Duration = Duration::from_secs(3600);
/// The largest manifest taken.
const MOST_MANIFEST: usize = 4 << 20;
const API_VERSION: HeaderName = HeaderName::from_static("docker-distribution-api-version");
const DIGEST: HeaderName = HeaderName::from_static("docker-content-digest");
const UPLOAD: HeaderName = HeaderName::from_static("docker-upload-uuid");
/// An image's name: an owner, a repository of theirs, and a package.
#[derive(Clone, Debug, PartialEq, Eq, Hash)]
struct Name {
owner: String,
repo: String,
package: String,
}
impl std::fmt::Display for Name {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(f, "{}/{}/{}", self.owner, self.repo, self.package)
}
}
/// What a path under `/v2/` asks of an image.
#[derive(Debug, PartialEq)]
enum Ask {
/// A manifest, by tag or digest.
Manifest(String),
/// A blob, by digest.
Blob(String),
/// A blob's upload: a new one, or one under way by its id.
Upload(Option<String>),
/// The tags.
Tags,
}
/// Someone pushing: their account, and their session's authority there.
struct Pusher {
did: String,
auth: Auth,
}
struct Upload {
did: String,
path: PathBuf,
}
struct Registry {
ctx: Ctx,
public_url: Option<String>,
/// Each package's releases, by name.
records: Mutex<HashMap<Name, Arc<Records>>>,
/// Who each token for pushing is, and when it was given.
tokens: Mutex<HashMap<String, (Arc<Pusher>, Instant)>>,
/// The uploads under way, by id.
uploads: Mutex<HashMap<String, Upload>>,
/// The blobs pushed and uploaded, not yet in a release.
staged: Mutex<Staged>,
}
/// The most blobs one account has staged at once: past it, its oldest is
/// forgotten, so one account's pushes never cost another's.
const STAGED_EACH: usize = 256;
/// The blobs pushed and uploaded, not yet in a release, by account and
/// digest, each with when it was staged, counting up. Past [`STAGED_EACH`]
/// for an account, or [`MOST_KEPT`] in all, the oldest is forgotten: a push
/// still under way stages its blobs last, so it keeps them.
#[derive(Default)]
struct Staged {
files: HashMap<(String, String), (u64, ReleaseFile)>,
next: u64,
}
impl Staged {
fn get(&self, did: &str, digest: &str) -> Option<&ReleaseFile> {
let key = (did.to_string(), digest.to_string());
self.files.get(&key).map(|(_, file)| file)
}
fn insert(&mut self, did: &str, digest: String, file: ReleaseFile) {
let own = self.files.keys().filter(|(d, _)| d == did).count();
if own >= STAGED_EACH {
self.forget_oldest(Some(did));
}
if self.files.len() >= MOST_KEPT {
self.forget_oldest(None);
}
self.next += 1;
self.files
.insert((did.to_string(), digest), (self.next, file));
}
/// Forgets the oldest staged, of `did` if given.
fn forget_oldest(&mut self, did: Option<&str>) {
let oldest = self
.files
.iter()
.filter(|((d, _), _)| did.is_none_or(|did| d == did))
.min_by_key(|(_, (at, _))| *at)
.map(|(key, _)| key.clone());
if let Some(key) = oldest {
self.files.remove(&key);
}
}
}
/// The registry's routes, reading as `ctx`, its tokens given out at
/// `public_url` (or else the host it is asked for).
pub fn routes(ctx: Ctx, public_url: Option<String>) -> Router {
let registry = Arc::new(Registry {
ctx,
public_url,
records: Mutex::default(),
tokens: Mutex::default(),
uploads: Mutex::default(),
staged: Mutex::default(),
});
Router::new()
.route("/v2/", get(base))
.route("/v2/token", get(token))
.route(
"/v2/{*rest}",
get(asked).post(posted).patch(patched).put(put),
)
.with_state(registry)
}
/// An error as the distribution spec has them.
fn error(status: StatusCode, code: &str, message: String) -> Response {
let body = json!({"errors": [{"code": code, "message": message}]});
let json = [(header::CONTENT_TYPE, "application/json")];
(
status,
[(API_VERSION, "registry/2.0")],
json,
body.to_string(),
)
.into_response()
}
fn unknown(code: &str, message: String) -> Response {
error(StatusCode::NOT_FOUND, code, message)
}
fn invalid(code: &str, message: String) -> Response {
error(StatusCode::BAD_REQUEST, code, message)
}
fn failed(e: anyhow::Error) -> Response {
error(StatusCode::BAD_GATEWAY, "UNKNOWN", format!("{e:#}"))
}
/// What a path under `/v2/` asks for, of which image.
fn parse(rest: &str) -> Option<(Name, Ask)> {
let (name, ask) = if let Some(name) = rest.strip_suffix("/tags/list") {
(name, Ask::Tags)
} else if let Some((name, id)) = rest.split_once("/blobs/uploads") {
let id = id.trim_start_matches('/');
(
name,
Ask::Upload(Some(id.to_string()).filter(|i| !i.is_empty())),
)
} else if let Some((name, reference)) = rest.rsplit_once("/manifests/") {
(name, Ask::Manifest(reference.into()))
} else {
let (name, digest) = rest.rsplit_once("/blobs/")?;
(name, Ask::Blob(digest.into()))
};
let parts: Vec<&str> = name.split('/').collect();
let (owner, repo, package) = match parts[..] {
[owner, repo] => (owner, repo, repo),
[owner, repo, package] => (owner, repo, package),
_ => return None,
};
let name = Name {
owner: owner.into(),
repo: repo.into(),
package: package.into(),
};
Some((name, ask))
}
/// A fresh value no one can guess, as hex.
fn random() -> String {
use rand_core::{OsRng, RngCore};
let mut bytes = [0u8; 16];
OsRng.fill_bytes(&mut bytes);
bytes.iter().map(|b| format!("{b:02x}")).collect()
}
/// What a handler answers, either way.
type Answer = Result<Response, Response>;
/// Empties `map` once it holds [`MOST_KEPT`].
fn bounded<K, V>(map: &mut HashMap<K, V>) {
if map.len() >= MOST_KEPT {
map.clear();
}
}
impl Registry {
fn url(name: &Name) -> Option<RepoUrl> {
RepoUrl::parse(&format!("knotless://{}/{}", name.owner, name.repo)).ok()
}
fn records(&self, name: &Name) -> Option<Arc<Records>> {
let url = Registry::url(name)?;
Some(self.records_at(name, url))
}
/// `name`'s releases, kept, of the repository at `url`.
fn records_at(&self, name: &Name, url: RepoUrl) -> Arc<Records> {
let mut kept = self.records.lock().unwrap();
bounded(&mut kept);
let records = kept
.entry(name.clone())
.or_insert_with(|| Arc::new(Records::new(self.ctx.clone(), url, name.package.clone())));
records.clone()
}
/// Asks for a token, for `scope` if given.
fn challenge(&self, headers: &HeaderMap, scope: Option<String>) -> Response {
let realm = format!("{}/v2/token", base_url(self.public_url.as_deref(), headers));
let mut asked = format!("Bearer realm=\"{realm}\",service=\"knotless\"");
if let Some(scope) = scope {
asked.push_str(&format!(",scope=\"{scope}\""));
}
let says = "log in with an app password: docker login <host> -u <handle>";
let mut res = error(StatusCode::UNAUTHORIZED, "UNAUTHORIZED", says.into());
let value = asked.parse().expect("a header's value");
res.headers_mut().insert(header::WWW_AUTHENTICATE, value);
res
}
/// Who `headers`' token is, if it is one for pushing still.
fn bearer(&self, headers: &HeaderMap) -> Option<Arc<Pusher>> {
let said = headers.get(header::AUTHORIZATION)?.to_str().ok()?;
let token = said.strip_prefix("Bearer ")?;
let mut tokens = self.tokens.lock().unwrap();
tokens.retain(|_, (_, at)| at.elapsed() < TOKEN_FOR);
tokens.get(token).map(|(pusher, _)| pusher.clone())
}
/// The owner of `name`, pushing with `headers`' token, and the
/// repository: else what to answer instead.
async fn pusher(
&self,
headers: &HeaderMap,
name: &Name,
) -> Result<(Arc<Pusher>, RepoUrl), Response> {
let scope = format!("repository:{name}:push,pull");
let Some(pusher) = self.bearer(headers) else {
return Err(self.challenge(headers, Some(scope)));
};
let Some(url) = Registry::url(name) else {
let says = format!("{}/{}: no such repository", name.owner, name.repo);
return Err(unknown("NAME_INVALID", says));
};
let owner = knotless::identity::resolve(&self.ctx, &name.owner).await;
let owner = owner.map_err(|e| unknown("NAME_UNKNOWN", format!("{e:#}")))?;
if owner.did != pusher.did {
let only = format!("only {} pushes to {name}", name.owner);
return Err(error(StatusCode::FORBIDDEN, "DENIED", only));
}
Ok((pusher, url))
}
/// The blob of `digest` that `did` pushed and no release holds yet.
fn staged(&self, did: &str, digest: &str) -> Option<ReleaseFile> {
let staged = self.staged.lock().unwrap();
staged.get(did, digest).cloned()
}
/// Uploads the file at `path`, of `size` bytes and SHA-256 `sha`, to
/// `pusher`'s PDS, keeps it among the release files, and stages it.
async fn stage(
&self,
pusher: &Pusher,
url: &RepoUrl,
path: &PathBuf,
size: u64,
sha: &str,
) -> anyhow::Result<()> {
let remote = Remote::open(&self.ctx, url).await?;
let auth = &pusher.auth;
let blobs = push::upload_file(&self.ctx, &remote, auth, path, size, RELEASE_MIME).await?;
let kept = kept_dir(&self.ctx);
tokio::fs::create_dir_all(&kept).await?;
tokio::fs::rename(path, kept.join(sha)).await?;
let file = ReleaseFile {
name: format!("sha256-{sha}"),
size,
sha256: sha.to_string(),
parts: blobs.into_iter().map(|blob| ReleasePart { blob }).collect(),
};
let mut staged = self.staged.lock().unwrap();
staged.insert(&pusher.did, format!("sha256:{sha}"), file);
Ok(())
}
/// A new file for an upload's bytes, until it is done.
async fn uploading(&self) -> anyhow::Result<PathBuf> {
let dir = self.ctx.env.cache_dir.join("serve").join("uploads");
tokio::fs::create_dir_all(&dir).await?;
let path = dir.join(random());
tokio::fs::write(&path, b"").await?;
Ok(path)
}
}
/// Says this is a registry, which clients ask first: with a token, any
/// token, as a pull's is anyone's; without one, asking for one.
async fn base(State(registry): State<Arc<Registry>>, headers: HeaderMap) -> Response {
let said = headers
.get(header::AUTHORIZATION)
.and_then(|v| v.to_str().ok());
if !said.is_some_and(|s| s.starts_with("Bearer ")) {
return registry.challenge(&headers, None);
}
let json = [(header::CONTENT_TYPE, "application/json")];
([(API_VERSION, "registry/2.0")], json, "{}").into_response()
}
/// `value` as a JSON answer.
fn json_answer(value: Value) -> Response {
(
[(header::CONTENT_TYPE, "application/json")],
value.to_string(),
)
.into_response()
}
/// A token: anyone's, for pulling, or for an owner's app password, given
/// as Basic, one to push with.
async fn token(State(registry): State<Arc<Registry>>, headers: HeaderMap) -> Answer {
use base64::Engine;
let basic = headers
.get(header::AUTHORIZATION)
.and_then(|v| v.to_str().ok())
.and_then(|v| v.strip_prefix("Basic "))
.and_then(|b| base64::engine::general_purpose::STANDARD.decode(b).ok())
.and_then(|b| String::from_utf8(b).ok());
let Some(basic) = basic else {
return Ok(json_answer(
json!({"token": "anonymous", "access_token": "anonymous"}),
));
};
let (user, password) = basic.split_once(':').unwrap_or((&basic, ""));
let session = knotless::session::create(®istry.ctx, user, password).await;
let refused = |e| {
let says = format!("{user} could not log in: {e:#}");
error(StatusCode::UNAUTHORIZED, "UNAUTHORIZED", says)
};
let session = session.map_err(refused)?;
let token = random();
let pusher = Arc::new(Pusher {
did: session.did.clone(),
auth: session.auth(),
});
let mut tokens = registry.tokens.lock().unwrap();
tokens.insert(token.clone(), (pusher, Instant::now()));
let lasts = TOKEN_FOR.as_secs();
Ok(json_answer(
json!({"token": token, "access_token": token, "expires_in": lasts}),
))
}
/// The file of `digest` among `releases`.
fn by_digest<'a>(releases: &'a [ReleaseRecord], digest: &str) -> Option<&'a ReleaseFile> {
let name = oci::file_name(digest)?;
releases
.iter()
.flat_map(|r| &r.files)
.find(|f| f.name == name)
}
/// What a blob, or a manifest of `media_type`, is handed out with.
fn headers(digest: &str, media_type: &str) -> [(HeaderName, String); 3] {
[
(header::CONTENT_TYPE, media_type.to_string()),
(DIGEST, digest.to_string()),
(API_VERSION, "registry/2.0".to_string()),
]
}
/// `name`'s releases, and where to read their files: else what to answer.
async fn opened(
registry: &Registry,
name: &Name,
) -> Result<(Arc<Records>, Arc<Vec<ReleaseRecord>>), Response> {
let Some(records) = registry.records(name) else {
let invalid = format!("{}/{}: no such repository", name.owner, name.repo);
return Err(unknown("NAME_INVALID", invalid));
};
let listed = records.list().await;
let releases = listed.map_err(|e| unknown("NAME_UNKNOWN", format!("{e:#}")))?;
Ok((records, releases))
}
/// That nothing at `rest` is done by `method`.
fn unsupported(rest: &str, method: &str) -> Response {
let says = format!("{rest}: nothing is {method} here");
error(StatusCode::METHOD_NOT_ALLOWED, "UNSUPPORTED", says)
}
/// `name` and what is asked of it, from a path under `/v2/`, else that
/// nothing is done there by `method`.
fn asking(rest: &str, method: &str) -> Result<(Name, Ask), Response> {
parse(rest).ok_or_else(|| unsupported(rest, method))
}
async fn asked(State(registry): State<Arc<Registry>>, Params(rest): Params<String>) -> Answer {
let Some((name, ask)) = parse(&rest) else {
return Err(unknown(
"NAME_UNKNOWN",
format!("{rest}: not <owner>/<repo>[/<package>]"),
));
};
let (records, releases) = opened(®istry, &name).await?;
match ask {
Ask::Tags => {
let images: Vec<ReleaseRecord> = releases
.iter()
.filter(|r| r.files.iter().any(|f| f.name == oci::INDEX))
.cloned()
.collect();
let mut tags: Vec<&str> = images.iter().map(|r| r.version.as_str()).collect();
if release::latest(&images, &name.package).is_some() {
tags.push("latest");
}
Ok(json_answer(json!({"name": name.to_string(), "tags": tags})))
}
Ask::Blob(digest) => {
let file = match by_digest(&releases, &digest) {
Some(file) => Some(file.clone()),
None => {
let owner = knotless::identity::resolve(®istry.ctx, &name.owner).await;
owner.ok().and_then(|o| registry.staged(&o.did, &digest))
}
};
let Some(file) = file else {
return Err(unknown("BLOB_UNKNOWN", format!("no blob {digest}")));
};
let octets = headers(&digest, "application/octet-stream");
records.send(&file, octets).await.map_err(failed)
}
Ask::Manifest(reference) => manifest(&records, &releases, &name, &reference).await,
Ask::Upload(_) => Err(unknown(
"BLOB_UPLOAD_UNKNOWN",
format!("{rest}: not read here"),
)),
}
}
/// A manifest by its digest, or by a tag, the one its release names.
async fn manifest(
records: &Records,
releases: &[ReleaseRecord],
name: &Name,
reference: &str,
) -> Answer {
let digest = match reference.starts_with("sha256:") {
true => reference.to_string(),
false => {
let tagged = match reference {
"latest" => release::latest(releases, &name.package),
version => releases.iter().find(|r| r.version == version),
};
let index = tagged.and_then(|r| r.files.iter().find(|f| f.name == oci::INDEX));
let Some(index) = index else {
return Err(unknown("MANIFEST_UNKNOWN", format!("no image {reference}")));
};
let read = records.read(index).await.and_then(|i| oci::manifest(&i));
read.map_err(failed)?.digest
}
};
let Some(file) = by_digest(releases, &digest) else {
return Err(unknown("MANIFEST_UNKNOWN", format!("no manifest {digest}")));
};
let bytes = records.read(file).await.map_err(failed)?;
let media_type = oci::media_type(&bytes);
Ok((headers(&digest, &media_type), bytes).into_response())
}
/// Appends `body` to the file at `path`, and says its size now.
async fn append(path: &PathBuf, body: Body) -> anyhow::Result<u64> {
let mut options = tokio::fs::OpenOptions::new();
let mut file = options.append(true).open(path).await?;
let mut chunks = body.into_data_stream();
while let Some(chunk) = chunks.next().await {
file.write_all(&chunk?).await?;
}
file.flush().await?;
Ok(file.metadata().await?.len())
}
/// Where an upload under way goes on, as it is answered.
fn going(name: &Name, id: &str, size: u64) -> Response {
let at = format!("/v2/{name}/blobs/uploads/{id}");
let range = format!("0-{}", size.saturating_sub(1));
let headers = [
(header::LOCATION, at),
(header::RANGE, range),
(UPLOAD, id.to_string()),
(API_VERSION, "registry/2.0".to_string()),
];
(StatusCode::ACCEPTED, headers).into_response()
}
/// Where a blob or manifest pushed is now, as it is answered.
fn made(name: &Name, kind: &str, digest: &str) -> Response {
let headers = [
(header::LOCATION, format!("/v2/{name}/{kind}/{digest}")),
(DIGEST, digest.to_string()),
(API_VERSION, "registry/2.0".to_string()),
];
(StatusCode::CREATED, headers).into_response()
}
/// Begins a blob's upload, and with `?digest=` ends it with the body.
async fn posted(
State(registry): State<Arc<Registry>>,
Params(rest): Params<String>,
Query(query): Query<HashMap<String, String>>,
headers: HeaderMap,
body: Body,
) -> Answer {
let (name, ask) = asking(&rest, "posted")?;
if ask != Ask::Upload(None) {
return Err(unsupported(&rest, "posted"));
}
let (pusher, url) = registry.pusher(&headers, &name).await?;
let path = registry.uploading().await.map_err(failed)?;
let id = path
.file_name()
.expect("a name")
.to_string_lossy()
.into_owned();
let upload = Upload {
did: pusher.did.clone(),
path,
};
registry.uploads.lock().unwrap().insert(id.clone(), upload);
match query.get("digest") {
Some(digest) => finish(®istry, &pusher, &url, &name, &id, digest, body).await,
None => Ok(going(&name, &id, 0)),
}
}
/// Goes on with an upload, with the body as its next part.
async fn patched(
State(registry): State<Arc<Registry>>,
Params(rest): Params<String>,
headers: HeaderMap,
body: Body,
) -> Answer {
let (name, ask) = asking(&rest, "patched")?;
let Ask::Upload(Some(id)) = ask else {
return Err(unsupported(&rest, "patched"));
};
let (pusher, _) = registry.pusher(&headers, &name).await?;
let path = upload_of(®istry, &pusher, &id)?;
let size = append(&path, body).await.map_err(failed)?;
Ok(going(&name, &id, size))
}
/// Where `pusher`'s upload `id` is, if it is theirs.
fn upload_of(registry: &Registry, pusher: &Pusher, id: &str) -> Result<PathBuf, Response> {
let uploads = registry.uploads.lock().unwrap();
match uploads.get(id).filter(|u| u.did == pusher.did) {
Some(upload) => Ok(upload.path.clone()),
None => Err(unknown("BLOB_UPLOAD_UNKNOWN", format!("no upload {id}"))),
}
}
/// Ends an upload, with the body as its last part, or pushes a manifest.
async fn put(
State(registry): State<Arc<Registry>>,
Params(rest): Params<String>,
Query(query): Query<HashMap<String, String>>,
headers: HeaderMap,
body: Body,
) -> Answer {
let (name, ask) = asking(&rest, "put")?;
let (reference, id) = match ask {
Ask::Manifest(reference) => (Some(reference), None),
Ask::Upload(Some(id)) => (None, Some(id)),
_ => return Err(unsupported(&rest, "put")),
};
let (pusher, url) = registry.pusher(&headers, &name).await?;
if let Some(reference) = reference {
let media_type = headers
.get(header::CONTENT_TYPE)
.and_then(|v| v.to_str().ok());
let media_type = media_type.map(String::from);
let pushing = Pushing {
registry: ®istry,
pusher: &pusher,
url: &url,
name: &name,
};
return pushing.manifest(&reference, media_type, body).await;
}
let id = id.unwrap_or_default();
let Some(digest) = query.get("digest") else {
return Err(invalid(
"DIGEST_INVALID",
"an upload ends with ?digest=".into(),
));
};
finish(®istry, &pusher, &url, &name, &id, digest, body).await
}
/// Ends upload `id` with `body`: checked against `digest`, uploaded and
/// staged.
async fn finish(
registry: &Registry,
pusher: &Pusher,
url: &RepoUrl,
name: &Name,
id: &str,
digest: &str,
body: Body,
) -> Answer {
let path = upload_of(registry, pusher, id)?;
registry.uploads.lock().unwrap().remove(id);
let appended = append(&path, body).await;
let measured = appended.and_then(|_| release::sha256_file(&path));
let (size, sha) = measured.map_err(failed)?;
if format!("sha256:{sha}") != digest {
let _ = tokio::fs::remove_file(&path).await;
let says = format!("what was uploaded is sha256:{sha}, not {digest}");
return Err(invalid("DIGEST_INVALID", says));
}
let staged = registry.stage(pusher, url, &path, size, &sha).await;
staged.map_err(failed)?;
Ok(made(name, "blobs", digest))
}
/// The digests `manifest` points at: its blobs, and the manifests an index
/// names, apart.
fn pointed(manifest: &Value) -> (Vec<String>, Vec<String>) {
let digests = |v: &Value| -> Vec<String> {
let all = v.as_array().into_iter().flatten();
all.filter_map(|d| d["digest"].as_str().map(String::from))
.collect()
};
let mut blobs = digests(&manifest["layers"]);
blobs.extend(digests(&manifest["blobs"]));
blobs.extend(manifest["config"]["digest"].as_str().map(String::from));
(blobs, digests(&manifest["manifests"]))
}
/// `file`, or else that `digest` was never pushed.
fn pushed_already(file: Option<ReleaseFile>, digest: &str) -> Result<ReleaseFile, Response> {
let says = format!("{digest} was not pushed");
file.ok_or_else(|| invalid("MANIFEST_BLOB_UNKNOWN", says))
}
/// Adds `file` to `files` unless one of its name is there.
fn add(files: &mut Vec<ReleaseFile>, file: ReleaseFile) {
if !files.iter().any(|f| f.name == file.name) {
files.push(file);
}
}
/// A manifest being pushed, by the owner of a repository.
struct Pushing<'a> {
registry: &'a Registry,
pusher: &'a Pusher,
url: &'a RepoUrl,
name: &'a Name,
}
impl Pushing<'_> {
/// The blob of `digest`, pushed or released.
fn find(&self, releases: &[ReleaseRecord], digest: &str) -> Result<ReleaseFile, Response> {
let staged = self.registry.staged(&self.pusher.did, digest);
pushed_already(
staged.or_else(|| by_digest(releases, digest).cloned()),
digest,
)
}
/// Pushes a manifest: by digest, staged; by tag, a version, released,
/// with every blob it reaches.
async fn manifest(&self, reference: &str, media_type: Option<String>, body: Body) -> Answer {
let (registry, name) = (self.registry, self.name);
let says = format!("a manifest is at most {MOST_MANIFEST} bytes, of JSON");
let read = axum::body::to_bytes(body, MOST_MANIFEST).await.ok();
let value = read
.as_ref()
.and_then(|b| serde_json::from_slice::<Value>(b).ok());
let (Some(bytes), Some(value)) = (read, value) else {
return Err(invalid("MANIFEST_INVALID", says));
};
let sha = release::sha256(&bytes);
let digest = format!("sha256:{sha}");
let tag = !reference.starts_with("sha256:");
if !tag && reference != digest {
let says = format!("the manifest is {digest}, not {reference}");
return Err(invalid("DIGEST_INVALID", says));
}
if tag && release::version(reference).is_err() {
let says =
format!("{reference}: a tag is a version, such as 1.0.0; latest is the highest");
return Err(invalid("TAG_INVALID", says));
}
let path = registry.uploading().await.map_err(failed)?;
let size = bytes.len() as u64;
let staged = async {
tokio::fs::write(&path, &bytes).await?;
registry
.stage(self.pusher, self.url, &path, size, &sha)
.await
};
staged.await.map_err(failed)?;
if !tag {
return Ok(made(name, "manifests", &digest));
}
let (records, releases) = opened(registry, name).await?;
// The manifest, and what it reaches: an index's manifests, and theirs.
let mut files: Vec<ReleaseFile> = Vec::new();
let mut manifests = vec![(digest.clone(), value)];
while let Some((at, manifest)) = manifests.pop() {
add(&mut files, self.find(&releases, &at)?);
let (blobs, children) = pointed(&manifest);
for blob in blobs {
add(&mut files, self.find(&releases, &blob)?);
}
for child in children {
let file = self.find(&releases, &child)?;
let read = records.read(&file).await.ok();
let value = read.and_then(|b| serde_json::from_slice(&b).ok());
manifests.push((child, value.unwrap_or(Value::Null)));
}
}
// As the client said, else as the manifest says.
let said = oci::media_type(&bytes);
let media_type = media_type.unwrap_or(said);
let index = json!({
"schemaVersion": 2,
"mediaType": "application/vnd.oci.image.index.v1+json",
"manifests": [{"mediaType": media_type, "digest": digest, "size": size}],
});
self.release(reference, index, files).await?;
Ok(made(name, "manifests", &digest))
}
/// Makes the release of `version`: its `index`, uploaded, and `files`.
async fn release(
&self,
version: &str,
index: Value,
mut files: Vec<ReleaseFile>,
) -> Result<(), Response> {
let (ctx, auth) = (&self.registry.ctx, &self.pusher.auth);
let denied = |e| error(StatusCode::FORBIDDEN, "DENIED", format!("{e:#}"));
let target = Target::open(ctx, self.url, &self.name.package, version).await;
let target = target.map_err(denied)?;
let index = index.to_string().into_bytes();
let blobs = push::upload(ctx, &target.remote, auth, &index, RELEASE_MIME).await;
let parts = blobs.map_err(failed)?;
let file = ReleaseFile {
name: oci::INDEX.to_string(),
size: index.len() as u64,
sha256: release::sha256(&index),
parts: parts.into_iter().map(|blob| ReleasePart { blob }).collect(),
};
files.insert(0, file);
let made = target.create(auth, files, Made::default()).await;
made.map_err(failed)?;
// Pulled at once, it is among those listed.
let records = self.registry.records_at(self.name, self.url.clone());
records.forget().await;
Ok(())
}
}
#[cfg(test)]
mod tests {
use super::*;
fn name(o: &str, r: &str, p: &str) -> Name {
Name {
owner: o.into(),
repo: r.into(),
package: p.into(),
}
}
#[test]
fn names() {
assert_eq!(
parse("alice.test/tools/manifests/latest"),
Some((
name("alice.test", "tools", "tools"),
Ask::Manifest("latest".into())
))
);
assert_eq!(
parse("alice.test/tools/serve/blobs/sha256:ab"),
Some((
name("alice.test", "tools", "serve"),
Ask::Blob("sha256:ab".into())
))
);
assert_eq!(
parse("alice.test/tools/blobs/uploads/"),
Some((name("alice.test", "tools", "tools"), Ask::Upload(None)))
);
assert_eq!(
parse("alice.test/tools/blobs/uploads/ab12"),
Some((
name("alice.test", "tools", "tools"),
Ask::Upload(Some("ab12".into()))
))
);
assert_eq!(
parse("alice.test/tools/tags/list"),
Some((name("alice.test", "tools", "tools"), Ask::Tags))
);
for bad in [
"alice.test/manifests/1.0.0",
"a/b/c/d/manifests/1",
"a/b/other",
] {
assert_eq!(parse(bad), None, "{bad}");
}
}
#[test]
fn what_a_manifest_points_at() {
let image = json!({
"config": {"digest": "sha256:c"},
"layers": [{"digest": "sha256:l1"}, {"digest": "sha256:l2"}],
});
let blobs = vec![
"sha256:l1".to_string(),
"sha256:l2".into(),
"sha256:c".into(),
];
assert_eq!(pointed(&image), (blobs, vec![]));
let index = json!({"manifests": [{"digest": "sha256:m"}]});
assert_eq!(pointed(&index), (vec![], vec!["sha256:m".to_string()]));
}
#[test]
fn what_is_kept_is_bounded() {
let ctx = Ctx::new(knotless::env::Env::from_vars(|_| None));
let registry = Registry {
ctx,
public_url: None,
records: Mutex::default(),
tokens: Mutex::default(),
uploads: Mutex::default(),
staged: Mutex::default(),
};
for i in 0..=MOST_KEPT {
assert!(
registry
.records(&name("alice.test", &format!("r{i}"), "p"))
.is_some()
);
}
assert_eq!(registry.records.lock().unwrap().len(), 1);
assert!(
registry
.records(&name("alice.test", "no/such", "p"))
.is_none()
);
}
#[test]
fn staging_forgets_only_the_oldest() {
let file = |name: &str| ReleaseFile {
name: name.into(),
size: 1,
sha256: name.into(),
parts: Vec::new(),
};
let mut staged = Staged::default();
staged.insert("did:a", "a0".into(), file("a0"));
// One account past its share loses only its own oldest.
for i in 0..STAGED_EACH {
staged.insert("did:b", format!("b{i}"), file("b"));
}
staged.insert("did:b", "last".into(), file("b"));
assert!(staged.get("did:b", "b0").is_none());
assert!(staged.get("did:b", "b1").is_some());
assert!(staged.get("did:b", "last").is_some());
assert!(staged.get("did:a", "a0").is_some());
// Many accounts past the whole lose the oldest of all.
for i in 0..MOST_KEPT {
staged.insert(&format!("did:{i}"), "x".into(), file("x"));
}
assert_eq!(staged.files.len(), MOST_KEPT);
assert!(staged.get("did:a", "a0").is_none());
assert!(staged.get(&format!("did:{}", MOST_KEPT - 1), "x").is_some());
}
}