259 lines
11 KiB
TypeScript
259 lines
11 KiB
TypeScript
import type { ErrorCode } from "./error";
|
|
|
|
export const ARCHIVE_EXTRACTION_SCHEMA = 1 as const;
|
|
|
|
export interface ArchiveProjectIdentityIR {
|
|
projectId: string;
|
|
revision: number;
|
|
sha256: string;
|
|
}
|
|
|
|
export interface ArchiveExtractionEntryIR {
|
|
path: string;
|
|
uncompressedBytes: number;
|
|
sha256: string;
|
|
}
|
|
|
|
export interface ArchiveExtractionRequestIR {
|
|
schemaVersion: typeof ARCHIVE_EXTRACTION_SCHEMA;
|
|
transactionId: string;
|
|
archiveId: string;
|
|
committed: ArchiveProjectIdentityIR;
|
|
candidate: ArchiveProjectIdentityIR;
|
|
entries: ArchiveExtractionEntryIR[];
|
|
}
|
|
|
|
export interface ArchiveExtractionStorage {
|
|
readCommitted(projectId: string): Promise<ArchiveProjectIdentityIR>;
|
|
createStaging(transactionId: string, projectId: string): Promise<void>;
|
|
writeStaging(transactionId: string, path: string, bytes: Uint8Array): Promise<void>;
|
|
countStagingEntries(transactionId: string): Promise<number>;
|
|
discardStaging(transactionId: string): Promise<number>;
|
|
commitStaging(
|
|
transactionId: string,
|
|
expected: ArchiveProjectIdentityIR,
|
|
candidate: ArchiveProjectIdentityIR,
|
|
): Promise<ArchiveProjectIdentityIR>;
|
|
}
|
|
|
|
export type ArchiveExtractionEntryReader = (
|
|
entry: ArchiveExtractionEntryIR,
|
|
signal: AbortSignal,
|
|
) => Promise<Uint8Array>;
|
|
|
|
export type ArchiveExtractionReceiptIR =
|
|
| {
|
|
status: "COMMITTED";
|
|
transactionId: string;
|
|
committed: ArchiveProjectIdentityIR;
|
|
stagingEntriesAfter: 0;
|
|
}
|
|
| {
|
|
status: "CANCELLED";
|
|
code: "IO_ARCHIVE_CANCELLED";
|
|
transactionId: string;
|
|
committedBefore: ArchiveProjectIdentityIR;
|
|
committedAfter: ArchiveProjectIdentityIR;
|
|
removedStagingEntries: number;
|
|
stagingEntriesAfter: 0;
|
|
publishedProjects: 0;
|
|
};
|
|
|
|
export class ArchiveExtractionError extends Error {
|
|
readonly code: ErrorCode;
|
|
readonly path?: string;
|
|
|
|
constructor(code: ErrorCode, message: string, path?: string) {
|
|
super(`${code}: ${message}`);
|
|
this.name = "ArchiveExtractionError";
|
|
this.code = code;
|
|
this.path = path;
|
|
}
|
|
}
|
|
|
|
const ID = /^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$/;
|
|
const SHA256 = /^[a-f0-9]{64}$/;
|
|
const CONTROL_CHARACTER = /[\u0000-\u001f\u007f]/;
|
|
const DRIVE_PATH = /^[A-Za-z]:\//;
|
|
const URI_SCHEME = /^[A-Za-z][A-Za-z0-9+.-]*:/;
|
|
|
|
function record(value: unknown, path: string): Record<string, unknown> {
|
|
if (typeof value !== "object" || value === null || Array.isArray(value)) {
|
|
throw new ArchiveExtractionError("IO_ARCHIVE_UNSAFE", `${path} must be an object`, path);
|
|
}
|
|
return value as Record<string, unknown>;
|
|
}
|
|
|
|
function exactKeys(value: Record<string, unknown>, expected: readonly string[], path: string): void {
|
|
const actual = Object.keys(value).sort();
|
|
const allowed = [...expected].sort();
|
|
if (actual.length !== allowed.length || actual.some((key, index) => key !== allowed[index])) {
|
|
throw new ArchiveExtractionError("IO_ARCHIVE_UNSAFE", `${path} contains undeclared fields`, path);
|
|
}
|
|
}
|
|
|
|
function parseIdentity(value: unknown, path: string): ArchiveProjectIdentityIR {
|
|
const input = record(value, path);
|
|
exactKeys(input, ["projectId", "revision", "sha256"], path);
|
|
if (typeof input.projectId !== "string" || !ID.test(input.projectId)) {
|
|
throw new ArchiveExtractionError("IO_ARCHIVE_UNSAFE", `${path}.projectId is invalid`, `${path}.projectId`);
|
|
}
|
|
if (typeof input.revision !== "number" || !Number.isSafeInteger(input.revision) || input.revision < 0) {
|
|
throw new ArchiveExtractionError("IO_ARCHIVE_UNSAFE", `${path}.revision is invalid`, `${path}.revision`);
|
|
}
|
|
if (typeof input.sha256 !== "string" || !SHA256.test(input.sha256)) {
|
|
throw new ArchiveExtractionError("IO_ARCHIVE_UNSAFE", `${path}.sha256 is invalid`, `${path}.sha256`);
|
|
}
|
|
return { projectId: input.projectId, revision: input.revision, sha256: input.sha256 };
|
|
}
|
|
|
|
function parseArchivePath(value: unknown, path: string): string {
|
|
if (typeof value !== "string" || value.length === 0 || value.length > 2_048 || CONTROL_CHARACTER.test(value)) {
|
|
throw new ArchiveExtractionError("IO_ARCHIVE_UNSAFE", `${path} is invalid`, path);
|
|
}
|
|
const normalized = value.normalize("NFC");
|
|
if (normalized !== value || value.startsWith("/") || value.startsWith("\\") || value.includes("\\") || DRIVE_PATH.test(value) || URI_SCHEME.test(value)) {
|
|
throw new ArchiveExtractionError("IO_ARCHIVE_UNSAFE", `${path} is not canonical`, path);
|
|
}
|
|
const segments = value.split("/");
|
|
if (segments.some((segment) => segment.length === 0 || segment === "." || segment === "..")) {
|
|
throw new ArchiveExtractionError("IO_ARCHIVE_UNSAFE", `${path} escapes staging`, path);
|
|
}
|
|
return value;
|
|
}
|
|
|
|
export function parseArchiveExtractionRequest(value: unknown): ArchiveExtractionRequestIR {
|
|
const input = record(value, "input");
|
|
exactKeys(input, ["schemaVersion", "transactionId", "archiveId", "committed", "candidate", "entries"], "input");
|
|
if (input.schemaVersion !== ARCHIVE_EXTRACTION_SCHEMA) {
|
|
throw new ArchiveExtractionError("IO_ARCHIVE_UNSAFE", "unsupported archive extraction schema", "input.schemaVersion");
|
|
}
|
|
if (typeof input.transactionId !== "string" || !ID.test(input.transactionId)) {
|
|
throw new ArchiveExtractionError("IO_ARCHIVE_UNSAFE", "transactionId is invalid", "input.transactionId");
|
|
}
|
|
if (typeof input.archiveId !== "string" || !ID.test(input.archiveId)) {
|
|
throw new ArchiveExtractionError("IO_ARCHIVE_UNSAFE", "archiveId is invalid", "input.archiveId");
|
|
}
|
|
const committed = parseIdentity(input.committed, "input.committed");
|
|
const candidate = parseIdentity(input.candidate, "input.candidate");
|
|
if (candidate.projectId !== committed.projectId || candidate.revision !== committed.revision + 1) {
|
|
throw new ArchiveExtractionError("REVISION_CONFLICT", "candidate must advance the same project by one revision", "input.candidate");
|
|
}
|
|
if (!Array.isArray(input.entries) || input.entries.length === 0 || input.entries.length > 100_000) {
|
|
throw new ArchiveExtractionError("IO_ARCHIVE_UNSAFE", "entries exceeds its bound", "input.entries");
|
|
}
|
|
const paths = new Set<string>();
|
|
const entries = input.entries.map((value, index): ArchiveExtractionEntryIR => {
|
|
const path = `input.entries[${index}]`;
|
|
const entry = record(value, path);
|
|
exactKeys(entry, ["path", "uncompressedBytes", "sha256"], path);
|
|
const entryPath = parseArchivePath(entry.path, `${path}.path`);
|
|
if (paths.has(entryPath)) throw new ArchiveExtractionError("IO_ARCHIVE_UNSAFE", `duplicate entry ${entryPath}`, `${path}.path`);
|
|
paths.add(entryPath);
|
|
if (typeof entry.uncompressedBytes !== "number" || !Number.isSafeInteger(entry.uncompressedBytes) || entry.uncompressedBytes < 0 || entry.uncompressedBytes > 2 * 1024 * 1024 * 1024) {
|
|
throw new ArchiveExtractionError("IO_ARCHIVE_UNSAFE", `${path}.uncompressedBytes is outside its bound`, `${path}.uncompressedBytes`);
|
|
}
|
|
if (typeof entry.sha256 !== "string" || !SHA256.test(entry.sha256)) {
|
|
throw new ArchiveExtractionError("IO_ARCHIVE_UNSAFE", `${path}.sha256 is invalid`, `${path}.sha256`);
|
|
}
|
|
return { path: entryPath, uncompressedBytes: entry.uncompressedBytes, sha256: entry.sha256 };
|
|
});
|
|
const orderedPaths = [...paths].sort();
|
|
for (let index = 1; index < orderedPaths.length; index++) {
|
|
if (orderedPaths[index].startsWith(`${orderedPaths[index - 1]}/`)) {
|
|
throw new ArchiveExtractionError("IO_ARCHIVE_UNSAFE", `file/directory prefix conflict at ${orderedPaths[index]}`, "input.entries");
|
|
}
|
|
}
|
|
return {
|
|
schemaVersion: ARCHIVE_EXTRACTION_SCHEMA,
|
|
transactionId: input.transactionId,
|
|
archiveId: input.archiveId,
|
|
committed,
|
|
candidate,
|
|
entries,
|
|
};
|
|
}
|
|
|
|
function sameIdentity(left: ArchiveProjectIdentityIR, right: ArchiveProjectIdentityIR): boolean {
|
|
return left.projectId === right.projectId && left.revision === right.revision && left.sha256 === right.sha256;
|
|
}
|
|
|
|
async function digest(bytes: Uint8Array): Promise<string> {
|
|
const source = bytes.buffer.slice(bytes.byteOffset, bytes.byteOffset + bytes.byteLength) as ArrayBuffer;
|
|
const result = await crypto.subtle.digest("SHA-256", source);
|
|
return [...new Uint8Array(result)].map((value) => value.toString(16).padStart(2, "0")).join("");
|
|
}
|
|
|
|
function cancellation(error: unknown, signal: AbortSignal): boolean {
|
|
return signal.aborted || typeof error === "object" && error !== null && "name" in error && error.name === "AbortError";
|
|
}
|
|
|
|
function cancelled(): never {
|
|
throw new DOMException("Archive extraction was cancelled", "AbortError");
|
|
}
|
|
|
|
export async function runArchiveExtractionTransaction(
|
|
value: unknown,
|
|
storage: ArchiveExtractionStorage,
|
|
readEntry: ArchiveExtractionEntryReader,
|
|
signal: AbortSignal,
|
|
): Promise<ArchiveExtractionReceiptIR> {
|
|
const request = parseArchiveExtractionRequest(value);
|
|
const committedBefore = await storage.readCommitted(request.committed.projectId);
|
|
if (!sameIdentity(committedBefore, request.committed)) {
|
|
throw new ArchiveExtractionError("REVISION_CONFLICT", "committed project identity is stale", "input.committed");
|
|
}
|
|
|
|
let stagingCreated = false;
|
|
let published = false;
|
|
try {
|
|
if (signal.aborted) cancelled();
|
|
stagingCreated = true;
|
|
await storage.createStaging(request.transactionId, request.committed.projectId);
|
|
for (const entry of request.entries) {
|
|
if (signal.aborted) cancelled();
|
|
const bytes = await readEntry(entry, signal);
|
|
if (signal.aborted) cancelled();
|
|
if (!(bytes instanceof Uint8Array) || bytes.byteLength !== entry.uncompressedBytes || await digest(bytes) !== entry.sha256) {
|
|
throw new ArchiveExtractionError("ASSET_SOURCE_HASH_MISMATCH", `payload identity mismatch for ${entry.path}`, entry.path);
|
|
}
|
|
await storage.writeStaging(request.transactionId, entry.path, bytes);
|
|
if (signal.aborted) cancelled();
|
|
}
|
|
|
|
// Cancellation linearizes here. Once the atomic commit starts, it completes as a commit.
|
|
if (signal.aborted) cancelled();
|
|
const committed = await storage.commitStaging(request.transactionId, committedBefore, request.candidate);
|
|
published = true;
|
|
if (!sameIdentity(committed, request.candidate)) {
|
|
throw new ArchiveExtractionError("STORAGE_TRANSACTION", "storage published an unexpected project identity");
|
|
}
|
|
if (await storage.countStagingEntries(request.transactionId) !== 0) {
|
|
throw new ArchiveExtractionError("STORAGE_TRANSACTION", "committed extraction retained staging entries");
|
|
}
|
|
return { status: "COMMITTED", transactionId: request.transactionId, committed, stagingEntriesAfter: 0 };
|
|
}
|
|
catch (error) {
|
|
if (published) throw error;
|
|
const removedStagingEntries = stagingCreated ? await storage.discardStaging(request.transactionId) : 0;
|
|
const stagingEntriesAfter = stagingCreated ? await storage.countStagingEntries(request.transactionId) : 0;
|
|
const committedAfter = await storage.readCommitted(request.committed.projectId);
|
|
if (stagingEntriesAfter !== 0 || !sameIdentity(committedBefore, committedAfter)) {
|
|
throw new ArchiveExtractionError("STORAGE_TRANSACTION", "archive rollback did not preserve the committed project");
|
|
}
|
|
if (cancellation(error, signal)) {
|
|
return {
|
|
status: "CANCELLED",
|
|
code: "IO_ARCHIVE_CANCELLED",
|
|
transactionId: request.transactionId,
|
|
committedBefore,
|
|
committedAfter,
|
|
removedStagingEntries,
|
|
stagingEntriesAfter: 0,
|
|
publishedProjects: 0,
|
|
};
|
|
}
|
|
throw error;
|
|
}
|
|
}
|