chadtmiller.com/pds.js
An atproto PDS in JavaScript, for Cloudflare Workers and Node
0
0
/**
* OCI distribution HTTP API served from dev.pdsjs.oci.repo records, so
* stock docker can push and pull:
*
* docker pull <pds-host>/<name>:<tag>
* docker push <pds-host>/<name>:<tag>
*
* A layer is one blob: its OCI digest and its blob CID hold the same
* SHA-256 bytes, so a blob GET is a digest re-encoding plus a getBlob.
* Manifest bytes are stored as blobs untouched, because a digest covers the
* exact serialization. The record keeps a blob reference for every stored
* manifest, config and layer, which is what holds them against orphan
* cleanup.
*
* Reads go through an injected fetch, typically the server's own fetch
* function. Writes go straight to the ports, the way the drive writer
* writes. Without the write ports the registry serves pulls only.
*
* This module must stay free of node-only imports; it runs on Cloudflare
* Workers.
*/
import { cborEncodeDagCbor, cidToString, createCid } from '@pdsjs/core/repo';
import { createSha256 } from '@pdsjs/core/sha256';
import { digestFromHashBytes, digestToCid, isValidDigest } from './digest.js';
import {
isValidName,
isValidTag,
nameToRkey,
OCI_REPO_COLLECTION,
} from './lexicon.js';
import { MAX_MANIFEST_SIZE, parseManifest } from './manifest.js';
import {
blobRef,
parseRepoRecord,
recordBlobCids,
removeManifest,
serializeRepoRecord,
} from './record.js';
const API_VERSION_HEADER = 'Docker-Distribution-API-Version';
const API_VERSION = 'registry/2.0';
const BLOB_MIME_TYPE = 'application/octet-stream';
/** How long an upload session waits between requests before it is swept. */
const UPLOAD_SESSION_TTL = 60 * 60 * 1000;
/** How long one verified credential pair stays good without re-checking. */
const AUTH_CACHE_TTL = 5 * 60 * 1000;
/** Largest accepted layer, when the caller sets no other bound. */
const DEFAULT_MAX_BLOB_SIZE = 10 * 1024 * 1024 * 1024;
/**
* @typedef {Object} RegistryContext
* @property {(request: Request) => Promise<Response>} xrpc - the PDS fetch
* function; requests are self-addressed XRPC calls
* @property {() => Promise<string|null>} getDid
* @property {import('@pdsjs/core/ports').ActorStoragePort} [actorStorage] - pushes need it
* @property {import('@pdsjs/core/ports').BlobPort} [blobs] - pushes need it
* @property {(did: string, ops: Array<{action: string, uri: string, cid?: string}>) => Promise<Object>} [createCommit] - pushes need it
* @property {import('@pdsjs/core/ports').HasherFactory} [hasher] - SHA-256 factory; absent, the plain JavaScript one
* @property {number} [maxBlobSize] - largest accepted layer, in bytes
* @property {string} [collection] - record collection, for lexicon forks
* @property {string} [realm] - realm named in the Basic challenge
*/
/**
* @typedef {Object} UploadSession
* @property {string} name
* @property {import('@pdsjs/core/ports').BlobWriter} writer
* @property {import('@pdsjs/core/ports').Hasher} hash
* @property {number} size
* @property {number} touched
* @property {boolean} busy
*/
/**
* @param {string} code - error code from the distribution spec
* @param {string} message
* @param {number} status
* @param {Record<string, string>} [headers]
* @returns {Response}
*/
function ociError(code, message, status, headers = {}) {
return Response.json(
{ errors: [{ code, message }] },
{ status, headers: { [API_VERSION_HEADER]: API_VERSION, ...headers } },
);
}
/**
* @param {Uint8Array[]} chunks
* @param {number} size
* @returns {Uint8Array}
*/
function joinChunks(chunks, size) {
const joined = new Uint8Array(size);
let offset = 0;
for (const chunk of chunks) {
joined.set(chunk, offset);
offset += chunk.length;
}
return joined;
}
/**
* The `start-end` byte range of an upload PATCH. The `bytes ` prefix some
* clients send is accepted.
* @param {string|null} header
* @returns {{start: number, end: number}|null}
*/
function parseContentRange(header) {
if (!header) return null;
const match = /^(?:bytes )?(\d+)-(\d+)/.exec(header);
if (!match) return null;
return { start: Number(match[1]), end: Number(match[2]) };
}
/**
* @param {RegistryContext} ctx
* @returns {{handle: (request: Request) => Promise<Response|null>}}
*/
export function createRegistry(ctx) {
if (!ctx || typeof ctx.xrpc !== 'function') {
throw new Error('createRegistry requires an xrpc fetch function');
}
if (typeof ctx.getDid !== 'function') {
throw new Error('createRegistry requires getDid');
}
const {
xrpc,
getDid,
actorStorage,
blobs,
createCommit,
hasher = createSha256,
maxBlobSize = DEFAULT_MAX_BLOB_SIZE,
collection = OCI_REPO_COLLECTION,
realm = 'registry',
} = ctx;
const pushEnabled = Boolean(actorStorage && blobs && createCommit);
/**
* @param {string} path
* @param {RequestInit} [init]
* @returns {Promise<Response>}
*/
const xrpcFetch = (path, init) =>
xrpc(new Request(`https://pds.internal${path}`, init));
// ── Authentication ─────────────────────────────────────────────────────
//
// Docker only pulls anonymously through the Bearer token flow: a Basic
// challenge with no stored credentials fails outright, where a Bearer
// challenge sends the client to the token endpoint, which grants pull to
// anyone and push to valid Basic credentials. Requests carrying Basic
// credentials directly are accepted too, so curl works without a token.
/** Verified credential digests, value = epoch ms the entry expires. */
const authCache = new Map();
/** Issued Bearer tokens, value = {push, expires}. */
const tokens = new Map();
/** How long one issued token stays good. */
const TOKEN_TTL = 30 * 60 * 1000;
/**
* @param {string} decoded - `identifier:password`
* @returns {Promise<string>}
*/
async function credentialKey(decoded) {
const hash = hasher();
await hash.update(new TextEncoder().encode(decoded));
return digestFromHashBytes(await hash.digest());
}
/**
* True when the request carries Basic credentials createSession accepts.
* The check goes through createSession itself, so account passwords and
* app passwords both work, exactly as docker login sends them.
* @param {string} header - the Authorization header
* @returns {Promise<boolean>}
*/
async function basicAuthorized(header) {
let decoded;
try {
decoded = atob(header.slice(5).trim());
} catch {
return false;
}
// The username may be a DID, which holds colons of its own, so the
// split runs after the DID prefix when one is present.
const didMatch = /^(did:[a-z0-9]+:[^:]+):(.*)$/s.exec(decoded);
const at = decoded.indexOf(':');
if (!didMatch && at < 0) return false;
const identifier = didMatch ? didMatch[1] : decoded.slice(0, at);
const password = didMatch ? didMatch[2] : decoded.slice(at + 1);
const key = await credentialKey(decoded);
const expires = authCache.get(key);
if (expires && expires > Date.now()) return true;
authCache.delete(key);
const res = await xrpcFetch('/xrpc/com.atproto.server.createSession', {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({ identifier, password }),
});
if (!res.ok) return false;
authCache.set(key, Date.now() + AUTH_CACHE_TTL);
return true;
}
/**
* What the request may do: null for no valid credentials at all.
* @param {Request} request
* @returns {Promise<{push: boolean}|null>}
*/
async function resolveAccess(request) {
const header = request.headers.get('authorization') ?? '';
if (/^basic\s/i.test(header)) {
return (await basicAuthorized(header)) ? { push: true } : null;
}
if (/^bearer\s/i.test(header)) {
const grant = tokens.get(header.slice(6).trim());
if (!grant || grant.expires < Date.now()) return null;
return { push: grant.push };
}
return null;
}
/**
* GET /v2/token, the realm the Bearer challenge names. Basic credentials
* grant push and pull; a request without credentials gets pull only,
* which is what makes anonymous docker pulls work. Bad credentials are
* refused rather than downgraded, so docker login fails loudly.
* @param {Request} request
* @returns {Promise<Response>}
*/
async function serveToken(request) {
const header = request.headers.get('authorization') ?? '';
let push = false;
if (/^basic\s/i.test(header)) {
if (!(await basicAuthorized(header))) {
return ociError('UNAUTHORIZED', 'invalid credentials', 401);
}
push = true;
}
for (const [token, grant] of tokens) {
if (grant.expires < Date.now()) tokens.delete(token);
}
const token = crypto.randomUUID();
tokens.set(token, { push, expires: Date.now() + TOKEN_TTL });
return Response.json(
{
token,
access_token: token,
expires_in: Math.floor(TOKEN_TTL / 1000),
},
{ headers: { [API_VERSION_HEADER]: API_VERSION } },
);
}
/**
* @param {URL} url
* @returns {Response}
*/
function challenge(url) {
return ociError('UNAUTHORIZED', 'authentication required', 401, {
'WWW-Authenticate': `Bearer realm="${url.origin}/v2/token",service="${realm}"`,
});
}
// ── Reads ──────────────────────────────────────────────────────────────
/**
* @param {string} did
* @param {string} name
* @returns {Promise<import('./record.js').OciRepoRecord|null>}
*/
async function loadRepo(did, name) {
const params = new URLSearchParams({
repo: did,
collection,
rkey: nameToRkey(name),
});
const res = await xrpcFetch(`/xrpc/com.atproto.repo.getRecord?${params}`);
if (!res.ok) return null;
const body = /** @type {{value: unknown}} */ (await res.json());
return parseRepoRecord(body.value);
}
/**
* Size and type of one stored blob, or null when it is absent. The
* metadata row answers when the port is present; otherwise a one-byte
* ranged getBlob reads the size out of Content-Range.
* @param {string} did
* @param {string} digest
* @returns {Promise<{size: number, mimeType: string}|null>}
*/
async function blobMeta(did, digest) {
const cid = digestToCid(digest);
if (actorStorage?.getBlob) {
const meta = await actorStorage.getBlob(cid);
return meta ? { size: meta.size, mimeType: meta.mimeType } : null;
}
const params = new URLSearchParams({ did, cid });
const res = await xrpcFetch(`/xrpc/com.atproto.sync.getBlob?${params}`, {
headers: { Range: 'bytes=0-0' },
});
if (res.status === 206) {
await res.body?.cancel();
const match = /\/(\d+)$/.exec(res.headers.get('content-range') ?? '');
return match
? {
size: Number(match[1]),
mimeType: res.headers.get('content-type') ?? BLOB_MIME_TYPE,
}
: null;
}
if (res.status === 416) {
// An empty blob satisfies no byte range; the 416 still proves it
// exists.
return { size: 0, mimeType: BLOB_MIME_TYPE };
}
if (res.ok) {
const data = await res.arrayBuffer();
return {
size: data.byteLength,
mimeType: res.headers.get('content-type') ?? BLOB_MIME_TYPE,
};
}
return null;
}
/**
* @param {string} did
* @param {string} cid
* @returns {Promise<Uint8Array|null>}
*/
async function readBlob(did, cid) {
const params = new URLSearchParams({ did, cid });
const res = await xrpcFetch(`/xrpc/com.atproto.sync.getBlob?${params}`);
if (!res.ok) return null;
return new Uint8Array(await res.arrayBuffer());
}
/**
* GET or HEAD one blob, streaming and honouring Range.
* @param {Request} request
* @param {string} did
* @param {string} digest
* @returns {Promise<Response>}
*/
async function serveBlob(request, did, digest) {
if (!isValidDigest(digest)) {
return ociError('DIGEST_INVALID', `unsupported digest: ${digest}`, 400);
}
const meta = await blobMeta(did, digest);
if (!meta) {
return ociError('BLOB_UNKNOWN', `blob unknown: ${digest}`, 404);
}
/** @type {Record<string, string>} */
const headers = {
[API_VERSION_HEADER]: API_VERSION,
'Docker-Content-Digest': digest,
'Content-Type': BLOB_MIME_TYPE,
'Accept-Ranges': 'bytes',
ETag: `"${digest}"`,
};
if (request.method === 'HEAD') {
headers['Content-Length'] = String(meta.size);
return new Response(null, { status: 200, headers });
}
const params = new URLSearchParams({ did, cid: digestToCid(digest) });
/** @type {HeadersInit} */
const forward = {};
const range = request.headers.get('range');
if (range) forward.Range = range;
const res = await xrpcFetch(`/xrpc/com.atproto.sync.getBlob?${params}`, {
headers: forward,
});
if (!res.ok && res.status !== 206 && res.status !== 416) {
return ociError('BLOB_UNKNOWN', `blob unknown: ${digest}`, 404);
}
for (const name of ['Content-Length', 'Content-Range']) {
const value = res.headers.get(name);
if (value) headers[name] = value;
}
return new Response(res.body, { status: res.status, headers });
}
/**
* The manifest entry a reference names: a digest directly, a tag through
* the tag list.
* @param {import('./record.js').OciRepoRecord} record
* @param {string} reference
* @returns {import('./record.js').ManifestEntry|null}
*/
function resolveManifest(record, reference) {
const digest = reference.includes(':')
? reference
: record.tags.find((t) => t.tag === reference)?.digest;
if (!digest) return null;
return record.manifests.find((m) => m.digest === digest) ?? null;
}
/**
* @param {Request} request
* @param {string} did
* @param {string} name
* @param {string} reference - tag or digest
* @returns {Promise<Response>}
*/
async function serveManifest(request, did, name, reference) {
const record = await loadRepo(did, name);
if (!record) {
return ociError('NAME_UNKNOWN', `repository unknown: ${name}`, 404);
}
const entry = resolveManifest(record, reference);
if (!entry) {
return ociError(
'MANIFEST_UNKNOWN',
`manifest unknown: ${reference}`,
404,
);
}
/** @type {Record<string, string>} */
const headers = {
[API_VERSION_HEADER]: API_VERSION,
'Docker-Content-Digest': entry.digest,
'Content-Type': entry.mediaType,
ETag: `"${entry.digest}"`,
};
const ifNoneMatch = request.headers.get('if-none-match');
if (ifNoneMatch?.includes(entry.digest)) {
return new Response(null, { status: 304, headers });
}
const bytes = await readBlob(did, entry.blob.ref.$link);
if (!bytes) {
return ociError(
'MANIFEST_UNKNOWN',
`manifest unknown: ${reference}`,
404,
);
}
headers['Content-Length'] = String(bytes.length);
return new Response(
request.method === 'HEAD' ? null : /** @type {BodyInit} */ (bytes),
{ status: 200, headers },
);
}
/**
* @param {URL} url
* @param {string} did
* @param {string} name
* @returns {Promise<Response>}
*/
async function serveTagsList(url, did, name) {
const record = await loadRepo(did, name);
if (!record) {
return ociError('NAME_UNKNOWN', `repository unknown: ${name}`, 404);
}
let tags = record.tags.map((t) => t.tag).sort();
const last = url.searchParams.get('last');
if (last) tags = tags.filter((t) => t > last);
const rawN = url.searchParams.get('n');
const n = rawN === null ? Number.NaN : Number(rawN);
if (Number.isFinite(n) && n >= 0) tags = tags.slice(0, n);
return Response.json(
{ name, tags },
{ headers: { [API_VERSION_HEADER]: API_VERSION } },
);
}
/**
* @param {string} did
* @returns {Promise<Response>}
*/
async function serveCatalog(did) {
const params = new URLSearchParams({
repo: did,
collection,
limit: '100',
});
const res = await xrpcFetch(`/xrpc/com.atproto.repo.listRecords?${params}`);
/** @type {string[]} */
let repositories = [];
if (res.ok) {
const body = /** @type {{records: Array<{value: unknown}>}} */ (
await res.json()
);
repositories = body.records
.map((r) => parseRepoRecord(r.value)?.name)
.filter((name) => typeof name === 'string')
.sort();
}
return Response.json(
{ repositories },
{ headers: { [API_VERSION_HEADER]: API_VERSION } },
);
}
// ── Writes ─────────────────────────────────────────────────────────────
/**
* Multi-request uploads in flight, keyed by session id. In memory: a
* restart loses them, and the client starts the upload over.
* @type {Map<string, UploadSession>}
*/
const uploads = new Map();
/**
* One write to a repository record at a time, so concurrent pushes
* read-modify-write in sequence rather than losing updates.
* @type {Map<string, Promise<unknown>>}
*/
const recordLocks = new Map();
/**
* @template T
* @param {string} name
* @param {() => Promise<T>} fn
* @returns {Promise<T>}
*/
function withRecordLock(name, fn) {
const previous = recordLocks.get(name) ?? Promise.resolve();
const next = previous.then(fn, fn);
recordLocks.set(
name,
next.catch(() => {}),
);
return next;
}
/**
* @param {string} id
* @returns {UploadSession}
*/
function takeUpload(id) {
const session = uploads.get(id);
if (!session || session.busy) {
throw new Error('upload unknown');
}
return session;
}
/** @param {string} id */
async function dropUpload(id) {
const session = uploads.get(id);
if (!session) return;
uploads.delete(id);
try {
await session.writer.abort();
} catch {
// an abort that fails leaves staged bytes for the platform to expire
}
}
/** Abandon every session that has waited past the TTL. */
async function sweepUploads() {
const cutoff = Date.now() - UPLOAD_SESSION_TTL;
for (const [id, session] of uploads) {
if (session.touched < cutoff && !session.busy) await dropUpload(id);
}
}
/**
* Feed one request body into a session. A body that fails partway has
* already fed the hash, so the session ends and the client starts over.
* @param {UploadSession} session
* @param {string} id
* @param {ReadableStream<Uint8Array>|null} body
* @returns {Promise<Response|null>} an error response, or null on success
*/
async function appendBody(session, id, body) {
if (!body) return null;
session.busy = true;
try {
const reader = body.getReader();
while (true) {
const { done, value } = await reader.read();
if (done) break;
const chunk =
value instanceof Uint8Array ? value : new Uint8Array(value);
session.size += chunk.length;
if (session.size > maxBlobSize) {
await reader.cancel();
await dropUpload(id);
return ociError(
'SIZE_INVALID',
`blob exceeds the maximum size of ${maxBlobSize} bytes`,
413,
);
}
await session.hash.update(chunk);
await session.writer.write(chunk);
}
session.busy = false;
session.touched = Date.now();
return null;
} catch (err) {
await dropUpload(id);
const message = err instanceof Error ? err.message : String(err);
return ociError('BLOB_UPLOAD_INVALID', message, 400);
}
}
/**
* Land a finished session as the blob its digest names.
* @param {UploadSession} session
* @param {string} id
* @param {string} digest - declared digest the bytes must hash to
* @returns {Promise<Response>}
*/
async function finishUpload(session, id, digest) {
uploads.delete(id);
if (!isValidDigest(digest)) {
await session.writer.abort();
return ociError('DIGEST_INVALID', `unsupported digest: ${digest}`, 400);
}
const computed = digestFromHashBytes(await session.hash.digest());
if (computed !== digest) {
await session.writer.abort();
return ociError(
'DIGEST_INVALID',
`digest mismatch: body hashes to ${computed}`,
400,
);
}
const cid = digestToCid(digest);
await /** @type {NonNullable<typeof actorStorage>} */ (
actorStorage
).putBlob(cid, BLOB_MIME_TYPE, session.size);
await session.writer.commit(cid, BLOB_MIME_TYPE);
return new Response(null, {
status: 201,
headers: {
[API_VERSION_HEADER]: API_VERSION,
Location: `/v2/${session.name}/blobs/${digest}`,
'Docker-Content-Digest': digest,
},
});
}
/**
* @param {string} name
* @returns {Promise<{id: string, session: UploadSession}>}
*/
async function createUpload(name) {
await sweepUploads();
const id = crypto.randomUUID();
const session = {
name,
writer:
await /** @type {NonNullable<import('@pdsjs/core/ports').BlobPort['createWriter']>} */ (
/** @type {NonNullable<typeof blobs>} */ (blobs).createWriter
)(/** @type {string} */ (await getDid())),
hash: hasher(),
size: 0,
touched: Date.now(),
busy: false,
};
uploads.set(id, session);
return { id, session };
}
/**
* @param {string} name
* @param {string} id
* @param {UploadSession} session
* @returns {Response}
*/
function uploadAccepted(name, id, session) {
return new Response(null, {
status: 202,
headers: {
[API_VERSION_HEADER]: API_VERSION,
Location: `/v2/${name}/blobs/uploads/${id}`,
Range: `0-${Math.max(0, session.size - 1)}`,
'Docker-Upload-UUID': id,
},
});
}
/**
* Store the record, link its blobs and commit, the way the drive writer
* does. The link runs before the commit, so a reap between the two cannot
* take a blob the record already holds.
* @param {string} did
* @param {string} name
* @param {import('./record.js').OciRepoRecord} record
* @param {boolean} existed
*/
async function writeRepoRecord(did, name, record, existed) {
const storage = /** @type {NonNullable<typeof actorStorage>} */ (
actorStorage
);
const rkey = nameToRkey(name);
const uri = `at://${did}/${collection}/${rkey}`;
const encoded = cborEncodeDagCbor(serializeRepoRecord(record));
const cid = cidToString(await createCid(encoded));
await storage.putBlock(cid, encoded);
await storage.putRecord(uri, cid, collection, rkey, encoded);
await storage.unlinkBlobsFromRecord(uri);
for (const blobCid of recordBlobCids(record)) {
await storage.linkBlobToRecord(blobCid, uri, Date.now());
}
await /** @type {NonNullable<typeof createCommit>} */ (createCommit)(did, [
{ action: existed ? 'update' : 'create', uri, cid },
]);
}
/**
* @param {Request} request
* @param {string} did
* @param {string} name
* @param {string} reference - tag or digest
* @returns {Promise<Response>}
*/
async function putManifest(request, did, name, reference) {
const declared = request.headers.get('content-type') ?? BLOB_MIME_TYPE;
const chunks = [];
let size = 0;
if (request.body) {
const reader = request.body.getReader();
while (true) {
const { done, value } = await reader.read();
if (done) break;
const chunk =
value instanceof Uint8Array ? value : new Uint8Array(value);
size += chunk.length;
if (size > MAX_MANIFEST_SIZE) {
await reader.cancel();
return ociError('MANIFEST_INVALID', 'manifest too large', 400);
}
chunks.push(chunk);
}
}
const bytes = joinChunks(chunks, size);
/** @type {import('./manifest.js').ParsedManifest} */
let parsed;
try {
parsed = parseManifest(bytes, declared);
} catch (err) {
const message = err instanceof Error ? err.message : String(err);
return ociError('MANIFEST_INVALID', message, 400);
}
const hash = hasher();
await hash.update(bytes);
const digest = digestFromHashBytes(await hash.digest());
if (reference.includes(':')) {
if (reference !== digest) {
return ociError(
'DIGEST_INVALID',
`manifest hashes to ${digest}, not ${reference}`,
400,
);
}
} else if (!isValidTag(reference)) {
return ociError('TAG_INVALID', `invalid tag: ${reference}`, 400);
}
return withRecordLock(name, async () => {
const existing = await loadRepo(did, name);
// Everything the manifest references must already be here. Layers
// with external urls or a foreign media type are distributed outside
// this registry, so their absence is expected.
/** @type {import('./record.js').LayerEntry[]} */
const layerEntries = [];
for (const descriptor of parsed.blobs) {
if (!isValidDigest(descriptor.digest)) {
return ociError(
'DIGEST_INVALID',
`unsupported digest: ${descriptor.digest}`,
400,
);
}
if (
(descriptor.urls?.length ?? 0) > 0 ||
descriptor.mediaType?.includes('foreign')
) {
continue;
}
const meta = await blobMeta(did, descriptor.digest);
if (!meta) {
return ociError(
'MANIFEST_BLOB_UNKNOWN',
`blob unknown: ${descriptor.digest}`,
404,
);
}
layerEntries.push({
digest: descriptor.digest,
blob: blobRef(
digestToCid(descriptor.digest),
descriptor.mediaType ?? meta.mimeType,
meta.size,
),
});
}
for (const child of parsed.children) {
if (!existing?.manifests.some((m) => m.digest === child.digest)) {
return ociError(
'MANIFEST_BLOB_UNKNOWN',
`manifest unknown: ${child.digest}`,
404,
);
}
}
const manifestCid = digestToCid(digest);
const storage = /** @type {NonNullable<typeof actorStorage>} */ (
actorStorage
);
await /** @type {NonNullable<typeof blobs>} */ (blobs).put(
/** @type {string} */ (did),
manifestCid,
bytes,
parsed.mediaType,
);
await storage.putBlob(manifestCid, parsed.mediaType, bytes.length);
const now = new Date().toISOString();
/** @type {import('./record.js').OciRepoRecord} */
const record = existing ?? {
name,
tags: [],
manifests: [],
layers: [],
createdAt: now,
};
record.updatedAt = now;
if (!record.manifests.some((m) => m.digest === digest)) {
record.manifests.push({
digest,
mediaType: parsed.mediaType,
blob: blobRef(manifestCid, parsed.mediaType, bytes.length),
});
}
for (const entry of layerEntries) {
if (!record.layers.some((l) => l.digest === entry.digest)) {
record.layers.push(entry);
}
}
if (!reference.includes(':')) {
const tag = record.tags.find((t) => t.tag === reference);
if (tag) tag.digest = digest;
else record.tags.push({ tag: reference, digest });
}
await writeRepoRecord(did, name, record, Boolean(existing));
return new Response(null, {
status: 201,
headers: {
[API_VERSION_HEADER]: API_VERSION,
Location: `/v2/${name}/manifests/${digest}`,
'Docker-Content-Digest': digest,
},
});
});
}
/**
* Delete a tag, or a manifest with every tag naming it. Removing a
* manifest drops the layer references no remaining manifest holds, and
* the orphan sweep frees the bytes.
* @param {string} did
* @param {string} name
* @param {string} reference - tag or digest
* @returns {Promise<Response>}
*/
async function deleteManifest(did, name, reference) {
return withRecordLock(name, async () => {
const record = await loadRepo(did, name);
if (!record) {
return ociError('NAME_UNKNOWN', `repository unknown: ${name}`, 404);
}
if (!reference.includes(':')) {
if (!record.tags.some((t) => t.tag === reference)) {
return ociError('MANIFEST_UNKNOWN', `tag unknown: ${reference}`, 404);
}
record.tags = record.tags.filter((t) => t.tag !== reference);
record.updatedAt = new Date().toISOString();
await writeRepoRecord(did, name, record, true);
return new Response(null, {
status: 202,
headers: { [API_VERSION_HEADER]: API_VERSION },
});
}
if (!record.manifests.some((m) => m.digest === reference)) {
return ociError(
'MANIFEST_UNKNOWN',
`manifest unknown: ${reference}`,
404,
);
}
// The record does not hold which layers each manifest references, so
// the remaining manifests are re-read to compute what stays.
const stillReferenced = new Set();
for (const entry of record.manifests) {
if (entry.digest === reference) continue;
const bytes = await readBlob(did, entry.blob.ref.$link);
if (!bytes) continue;
const parsed = parseManifest(bytes, entry.mediaType);
for (const descriptor of parsed.blobs) {
stillReferenced.add(descriptor.digest);
}
}
const next = removeManifest(record, reference, stillReferenced);
next.updatedAt = new Date().toISOString();
await writeRepoRecord(did, name, next, true);
return new Response(null, {
status: 202,
headers: { [API_VERSION_HEADER]: API_VERSION },
});
});
}
// ── Dispatch ───────────────────────────────────────────────────────────
/**
* @param {string[]} segments - path segments after /v2/
* @returns {{route: string, name: string, arg: string}|null}
*/
function matchRoute(segments) {
const s = segments;
const n = s.length;
if (n >= 3 && s[n - 2] === 'manifests') {
return {
route: 'manifests',
name: s.slice(0, n - 2).join('/'),
arg: s[n - 1],
};
}
if (n >= 3 && s[n - 2] === 'tags' && s[n - 1] === 'list') {
return { route: 'tags', name: s.slice(0, n - 2).join('/'), arg: '' };
}
if (n >= 4 && s[n - 3] === 'blobs' && s[n - 2] === 'uploads') {
return {
route: 'upload',
name: s.slice(0, n - 3).join('/'),
arg: s[n - 1],
};
}
if (n >= 3 && s[n - 2] === 'blobs' && s[n - 1] === 'uploads') {
return { route: 'uploads', name: s.slice(0, n - 2).join('/'), arg: '' };
}
if (n >= 3 && s[n - 2] === 'blobs') {
return {
route: 'blobs',
name: s.slice(0, n - 2).join('/'),
arg: s[n - 1],
};
}
if (n >= 3 && s[n - 2] === 'referrers') {
return {
route: 'referrers',
name: s.slice(0, n - 2).join('/'),
arg: s[n - 1],
};
}
return null;
}
return {
/**
* @param {Request} request
* @returns {Promise<Response|null>} null when the path is not a
* registry route
*/
async handle(request) {
const url = new URL(request.url);
const pathname = decodeURIComponent(url.pathname);
if (pathname !== '/v2' && !pathname.startsWith('/v2/')) return null;
try {
const did = await getDid();
if (!did) {
return ociError('NAME_UNKNOWN', 'server not initialized', 404);
}
const segments = pathname.split('/').filter((part) => part !== '');
// ['v2', ...]. The ping answers the Bearer challenge to a request
// without valid credentials: docker reads its auth scheme from this
// response, and without the challenge it never attaches credentials
// to a push. A client with no credentials takes a pull-only token
// from the realm, which is how anonymous pulls work.
if (segments.length === 1) {
if (!(await resolveAccess(request))) return challenge(url);
return Response.json(
{},
{ headers: { [API_VERSION_HEADER]: API_VERSION } },
);
}
if (segments.length === 2 && segments[1] === 'token') {
return await serveToken(request);
}
if (segments.length === 2 && segments[1] === '_catalog') {
return await serveCatalog(did);
}
const match = matchRoute(segments.slice(1));
if (!match) {
return ociError('UNSUPPORTED', 'unsupported route', 404);
}
const { route, name, arg } = match;
if (!isValidName(name)) {
return ociError('NAME_INVALID', `invalid name: ${name}`, 400);
}
const mutating = !['GET', 'HEAD'].includes(request.method);
if (mutating) {
if (!pushEnabled) {
return ociError('UNSUPPORTED', 'registry is read-only', 405);
}
const access = await resolveAccess(request);
if (!access) return challenge(url);
if (!access.push) {
return ociError('DENIED', 'pull-only credentials', 403);
}
}
if (route === 'blobs') {
if (mutating) {
return ociError(
'UNSUPPORTED',
'blobs are removed with their manifests',
405,
);
}
return await serveBlob(request, did, arg);
}
if (route === 'manifests') {
if (request.method === 'PUT') {
return await putManifest(request, did, name, arg);
}
if (request.method === 'DELETE') {
return await deleteManifest(did, name, arg);
}
if (mutating) {
return ociError('UNSUPPORTED', 'unsupported method', 405);
}
return await serveManifest(request, did, name, arg);
}
if (route === 'tags') {
if (mutating) {
return ociError('UNSUPPORTED', 'unsupported method', 405);
}
return await serveTagsList(url, did, name);
}
if (route === 'referrers') {
// The referrers API is not implemented; clients fall back to the
// tag schema on 404.
return ociError('UNSUPPORTED', 'referrers not supported', 404);
}
if (route === 'uploads' && request.method === 'POST') {
const mount = url.searchParams.get('mount');
if (mount && isValidDigest(mount)) {
// Blobs are stored per account rather than per repository, so a
// mount is an existence check.
if (await blobMeta(did, mount)) {
return new Response(null, {
status: 201,
headers: {
[API_VERSION_HEADER]: API_VERSION,
Location: `/v2/${name}/blobs/${mount}`,
'Docker-Content-Digest': mount,
},
});
}
}
const { id, session } = await createUpload(name);
const digest = url.searchParams.get('digest');
if (digest) {
const failed = await appendBody(session, id, request.body);
if (failed) return failed;
return await finishUpload(session, id, digest);
}
return uploadAccepted(name, id, session);
}
if (route === 'upload') {
/** @type {UploadSession} */
let session;
try {
session = takeUpload(arg);
} catch {
return ociError('BLOB_UPLOAD_UNKNOWN', 'upload unknown', 404);
}
if (request.method === 'PATCH') {
const range = parseContentRange(
request.headers.get('content-range'),
);
if (range && range.start !== session.size) {
return ociError(
'BLOB_UPLOAD_INVALID',
`next chunk starts at ${session.size}`,
416,
);
}
const failed = await appendBody(session, arg, request.body);
if (failed) return failed;
return uploadAccepted(name, arg, session);
}
if (request.method === 'PUT') {
const digest = url.searchParams.get('digest');
if (!digest) {
return ociError('DIGEST_INVALID', 'digest missing', 400);
}
const failed = await appendBody(session, arg, request.body);
if (failed) return failed;
return await finishUpload(session, arg, digest);
}
if (request.method === 'DELETE') {
await dropUpload(arg);
return new Response(null, {
status: 204,
headers: { [API_VERSION_HEADER]: API_VERSION },
});
}
if (request.method === 'GET') {
return new Response(null, {
status: 204,
headers: {
[API_VERSION_HEADER]: API_VERSION,
Range: `0-${Math.max(0, session.size - 1)}`,
'Docker-Upload-UUID': arg,
},
});
}
return ociError('UNSUPPORTED', 'unsupported method', 405);
}
return ociError('UNSUPPORTED', 'unsupported route', 404);
} catch (err) {
const message = err instanceof Error ? err.message : String(err);
return ociError('UNKNOWN', message, 500);
}
},
};
}