openvibers/openvibe.media Commits
Media S3: multipart uploads (Create, UploadPart, Complete, Abort, ListParts, ListMultipartUploads) with S3's rules (#109)
Changes
@@@ -19,13 +19,14 @@ a named object zone of one project environment. Zones are made and deleted only**Supported:**- `PutObject`, in one request, including aws-chunked bodies. `Content-MD5`, `x-amz-content-sha256` and the `x-amz-checksum-*` headers and trailers are checked.++- Multipart uploads (`server/s3/multipart.js`): `CreateMultipartUpload`, `UploadPart`, `CompleteMultipartUpload`, `AbortMultipartUpload`, `ListParts` and `ListMultipartUploads`. They follow S3's rules: every part but the last is at least 5 MiB, Complete names parts in ascending order with their ETags, and the object's ETag is the MD5 of the parts' MD5s with `-<count>`. Parts count toward the quota as they arrive. An upload expires after `MEDIA_MULTIPART_TTL_HOURS`.- `GetObject` (with ranges, conditional headers and presigned response overrides) and `HeadObject`.- `DeleteObject` and `DeleteObjects`.- `ListObjectsV2` and `ListObjects` (prefix, delimiter and pages).- `ListBuckets`, `HeadBucket` and `GetBucketLocation`.- Presigned GET and PUT, valid for 7 days at most.--**Not implemented:** multipart uploads, `CopyObject`, ACLs, policies, versioning, tagging and bucket deletion.++**Not implemented:** `CopyObject`, `UploadPartCopy`, ACLs, policies, versioning, tagging and bucket deletion.`CreateBucket` answers 200 for a zone that already exists, so tools that probe this way keep working.## Rules@@@ -40,6 +41,7 @@ a named object zone of one project environment. Zones are made and deleted only## Tests- `test/s3-zones.test.js` runs end to end with the AWS SDK for JavaScript v3: every operation, presigned URLs, isolation, quotas, holds, two doors and zone deletion.++- `test/s3-multipart.test.js` uses the SDK's `Upload` helper (parallel parts) and the raw commands: S3's rules, ETags, abort, quota and expiry.- `test/s3-sigv4.test.js` checks the worked examples of the S3 API reference, including the chunk signature chain.## Using it@@@ -47,8 +49,7 @@ a named object zone of one project environment. Zones are made and deleted only```shaws configure set aws_access_key_id ovk_… # issued on openvibe.zoneaws configure set aws_secret_access_key … # shown once--aws configure set default.s3.multipart_threshold 256MB # no multipart yet: one request per objectaws --endpoint-url https://s3.openvibe.media s3 ls s3://assets/rclone config create oz s3 provider Other endpoint https://s3.openvibe.media access_key_id ovk_… secret_access_key … \-- upload_cutoff 256M no_check_bucket true++ no_check_bucket true```nnew file mode 100644@@@ -0,0 +1,26 @@++-- phase: expand++-- S3 multipart uploads (ADR-031, server/s3/multipart.js): an upload in progress and the parts it has received. The++-- parts' bytes are files under OBJECTS_PATH/.tmp/s3mp/<upload id>/; CompleteMultipartUpload joins them into one object++-- (the same commit as PutObject), AbortMultipartUpload and the expiry sweep remove them.++CREATE TABLE media_s3_uploads (++ id text COLLATE "C" PRIMARY KEY, -- the S3 UploadId (mpu_<ULID>)++ zone_id text COLLATE "C" NOT NULL,++ object_key text COLLATE "C" NOT NULL,++ app_id text COLLATE "C" NOT NULL, -- the project tenant it counts against++ access_key_id text COLLATE "C" NOT NULL, -- who started it++ content_type text COLLATE "C" NOT NULL,++ metadata jsonb NOT NULL DEFAULT '{}', -- the object metadata the completed object takes++ created_at bigint NOT NULL,++ expires_at bigint NOT NULL++);++CREATE INDEX media_s3_uploads_zone ON media_s3_uploads (zone_id, object_key, id);++++CREATE TABLE media_s3_parts (++ upload_id text COLLATE "C" NOT NULL REFERENCES media_s3_uploads (id) ON DELETE CASCADE,++ part_number integer NOT NULL CHECK (part_number BETWEEN 1 AND 10000),++ size_bytes bigint NOT NULL,++ md5 text COLLATE "C" NOT NULL, -- hex: the part's ETag++ file_path text COLLATE "C" NOT NULL,++ created_at bigint NOT NULL,++ PRIMARY KEY (upload_id, part_number)++);@@@ -305,6 +305,8 @@ const ready = (async () => {try { const r = await require('./objects/timeline-cache').sweep(); if (r.removed) console.log(`[Objects] Timeline cache: evicted ${r.removed} local file(s)`); } catch (err) { console.warn('[Objects] timeline cache:', err.message); }// Incomplete multipart uploads past MEDIA_MULTIPART_TTL_HOURS: their parts are deleted (the objects stay uploading).try { const r = await require('./objects/multipart').purgeExpired(); if (r.expired || r.orphan_dirs) console.log(`[Objects] Multipart: ${r.expired} expired session(s), ${r.orphan_dirs} orphan part dir(s) removed`); } catch (err) { console.warn('[Objects] multipart purge:', err.message); }++ // S3 multipart uploads past their expiry (MEDIA_MULTIPART_TTL_HOURS): their parts are removed (server/s3/multipart.js).++ try { const n = await require('./s3/multipart').sweep(); if (n) console.log(`[S3] ${n} expired multipart upload(s) removed`); } catch (err) { console.warn('[S3] multipart sweep:', err.message); }// Native objects' popularity (ADR-021): a finished UTC day's viewer hashes and salt are deleted, only its counts stay (30 days).try { const r = await require('./objects/popularity').rotate(); if (r.hashes || r.salts || r.counts) console.log(`[Popularity] ${r.hashes} viewer hash(es) and ${r.salts} salt(s) of finished days deleted, ${r.counts} old daily count(s) pruned`); } catch (err) { console.warn('[Popularity] rotate:', err.message); }// Namespaces: uploads whose quota reservation expired are failed and their bytes freed; then everynnew file mode 100644@@@ -0,0 +1,119 @@++'use strict';++/**++ * S3 multipart uploads on Media's S3 surface (ADR-031): what the AWS CLI, the SDKs' Upload helpers and rclone use for++ * large files, and the way past a proxy's per-request size limit (each part is its own request).++ *++ * create({ zone, key, appId, accessKeyId, contentType, metadata }) → upload row (UploadId mpu_<ULID>)++ * find(uploadId, zone, key) → the upload, or null (another zone or key: null)++ * storePart(upload, n, tmp, { bytes, md5 }) → the part row (a re-sent part replaces the old)++ * parts(upload) / bytesSoFar(upload, exceptPart)++ * assemble(upload, wanted) wanted = [{ n, etag }] ascending → { tmp, bytes, sha256, etag } | { error }++ * abort(upload) / sweep() remove the parts (files and rows); sweep ends uploads past expires_at++ *++ * S3's rules: part numbers 1 to 10000; every part but the last is at least 5 MiB; Complete names its parts in++ * ascending order with their ETags; the object's ETag is the MD5 of the parts' MD5s, then "-<count>".++ */++const crypto = require('crypto');++const fs = require('fs');++const path = require('path');++const { ids } = require('openvibe-contracts');++const db = require('../db/database');++const config = require('../config');++++const MIN_PART = 5 * 1024 * 1024;++const TTL_MS = Math.max(1, config.objects.multipartTtlHours || 24) * 3600 * 1000;++const dirOf = (uploadId) => path.join(config.objects.path, '.tmp', 's3mp', uploadId);++++async function create({ zone, key, appId, accessKeyId, contentType, metadata, now = Date.now() }) {++ const id = `mpu_${ids.ulid(now)}`;++ await db.run(`INSERT INTO media_s3_uploads (id, zone_id, object_key, app_id, access_key_id, content_type, metadata, created_at, expires_at)++ VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)`, [id, zone.id, key, appId, accessKeyId, contentType, JSON.stringify(metadata || {}), now, now + TTL_MS]);++ fs.mkdirSync(dirOf(id), { recursive: true });++ return await db.get('SELECT * FROM media_s3_uploads WHERE id = ?', [id]);++}++++async function find(uploadId, zone, key) {++ if (!/^mpu_[0-9A-HJKMNP-TV-Z]{26}$/.test(String(uploadId || ''))) return null;++ const u = await db.get('SELECT * FROM media_s3_uploads WHERE id = ?', [uploadId]);++ return u && u.zone_id === zone.id && u.object_key === key ? u : null;++}++++async function parts(upload) {++ return await db.all('SELECT * FROM media_s3_parts WHERE upload_id = ? ORDER BY part_number', [upload.id]);++}++++async function bytesSoFar(upload, exceptPart = 0) {++ const r = await db.get('SELECT COALESCE(SUM(size_bytes), 0) AS b FROM media_s3_parts WHERE upload_id = ? AND part_number <> ?', [upload.id, exceptPart]);++ return Number(r.b);++}++++/** Move a received part (the file at `tmp`) into the upload; a part sent again replaces the earlier one. */++async function storePart(upload, n, tmp, { bytes, md5, now = Date.now() }) {++ const dest = path.join(dirOf(upload.id), String(n).padStart(5, '0'));++ fs.mkdirSync(path.dirname(dest), { recursive: true });++ fs.renameSync(tmp, dest);++ await db.run(`INSERT INTO media_s3_parts (upload_id, part_number, size_bytes, md5, file_path, created_at) VALUES (?, ?, ?, ?, ?, ?)++ ON CONFLICT (upload_id, part_number) DO UPDATE SET size_bytes = excluded.size_bytes, md5 = excluded.md5,++ file_path = excluded.file_path, created_at = excluded.created_at`, [upload.id, n, bytes, md5, dest, now]);++ return await db.get('SELECT * FROM media_s3_parts WHERE upload_id = ? AND part_number = ?', [upload.id, n]);++}++++/** Join the named parts into one file, checking S3's rules. The upload's own part files stay until abort or commit. */++async function assemble(upload, wanted) {++ if (!wanted.length) return { error: ['MalformedXML', 'The XML you provided was not well-formed or did not validate'] };++ for (let i = 1; i < wanted.length; i++) {++ if (wanted[i].n <= wanted[i - 1].n) return { error: ['InvalidPartOrder', 'The list of parts was not in ascending order'] };++ }++ const have = new Map((await parts(upload)).map((p) => [p.part_number, p]));++ const chosen = [];++ for (const [i, w] of wanted.entries()) {++ const p = have.get(w.n);++ const etag = String(w.etag || '').replace(/^"|"$/g, '').toLowerCase();++ if (!p || p.md5 !== etag) return { error: ['InvalidPart', `Part ${w.n} was not uploaded, or its ETag does not match`] };++ if (i < wanted.length - 1 && Number(p.size_bytes) < MIN_PART) return { error: ['EntityTooSmall', `Part ${w.n} is smaller than the 5 MiB every part but the last must be`] };++ if (!fs.existsSync(p.file_path)) return { error: ['InvalidPart', `Part ${w.n} is no longer stored`] };++ chosen.push(p);++ }++ const tmp = path.join(config.objects.path, '.tmp', `s3mp-${upload.id}-${crypto.randomBytes(4).toString('hex')}`);++ const sha = crypto.createHash('sha256');++ const md5s = crypto.createHash('md5');++ let bytes = 0;++ const out = fs.createWriteStream(tmp);++ try {++ for (const p of chosen) {++ md5s.update(Buffer.from(p.md5, 'hex'));++ await new Promise((resolve, reject) => {++ const src = fs.createReadStream(p.file_path);++ src.on('data', (c) => { sha.update(c); bytes += c.length; });++ src.on('error', reject);++ out.once('error', reject);++ src.on('end', resolve);++ src.pipe(out, { end: false });++ });++ }++ await new Promise((resolve, reject) => out.end((err) => (err ? reject(err) : resolve())));++ } catch (err) {++ out.destroy();++ try { fs.unlinkSync(tmp); } catch { /* not written */ }++ return { error: ['InternalError', `The parts could not be joined: ${err.message}`] };++ }++ return { tmp, bytes, sha256: sha.digest('hex'), etag: `${md5s.digest('hex')}-${chosen.length}` };++}++++async function abort(upload) {++ fs.rmSync(dirOf(upload.id), { recursive: true, force: true });++ await db.run('DELETE FROM media_s3_uploads WHERE id = ?', [upload.id]);++}++++async function sweep({ now = Date.now() } = {}) {++ const old = await db.all('SELECT * FROM media_s3_uploads WHERE expires_at < ? LIMIT 500', [now]);++ for (const u of old) await abort(u);++ return old.length;++}++++async function listForZone(zone, { prefix = '', limit = 1000 } = {}) {++ return await db.all(`SELECT * FROM media_s3_uploads WHERE zone_id = ? AND object_key LIKE ? ESCAPE '\\' ORDER BY object_key, id LIMIT ?`,++ [zone.id, `${prefix.replace(/[\\%_]/g, (c) => `\\${c}`)}%`, limit]);++}++++module.exports = { create, find, parts, bytesSoFar, storePart, assemble, abort, sweep, listForZone, MIN_PART };@@@ -12,10 +12,12 @@* PUT /<zone> CreateBucket: 200 for a zone that exists (clients probe this way), else NotImplemented* POST /<zone>?delete DeleteObjects* PUT /<zone>/<key> PutObject (one request; aws-chunked bodies too)++ * POST /<zone>/<key>?uploads CreateMultipartUpload; PUT ?partNumber&uploadId UploadPart; POST ?uploadId Complete;++ * DELETE ?uploadId Abort; GET ?uploadId ListParts; GET /<zone>?uploads ListMultipartUploads* GET /<zone>/<key> GetObject (Range)* HEAD /<zone>/<key> HeadObject* DELETE /<zone>/<key> DeleteObject (Media's soft delete; refused under a retention hold)-- * everything else NotImplemented (multipart, copy, ACLs, policies, versioning, bucket deletion)++ * everything else NotImplemented (copy, part copy, ACLs, policies, versioning, bucket deletion)** Every request is signed (SigV4, header or presigned) with an access key (server/s3/keys.js). The key fixes the* project and environment, so a bucket name resolves to one zone id of that environment (server/s3/zones.js); a key@@@ -42,6 +44,7 @@ const namespaces = require('../objects/namespaces');const sigv4 = require('./sigv4');const keys = require('./keys');const zones = require('./zones');++const mp = require('./multipart');const MAX_KEY_BYTES = 1024;const MAX_LIST = 1000;@@@ -55,6 +58,7 @@ const STATUS = {KeyTooLongError: 400, MissingContentLength: 411, EntityTooLarge: 400, BadDigest: 400, XAmzContentSHA256Mismatch: 400,IncompleteBody: 400, MalformedXML: 400, InvalidRange: 416, NoSuchBucket: 404, NoSuchKey: 404, QuotaExceeded: 403,NotImplemented: 501, MethodNotAllowed: 405, InternalError: 500, ServiceUnavailable: 503, InvalidBucketName: 400,++ NoSuchUpload: 404, EntityTooSmall: 400, InvalidPart: 400, InvalidPartOrder: 400, PreconditionFailed: 412,};const esc = (s) => String(s).replace(/&/g, '&').replace(/</g, '<').replace(/>/g, '>').replace(/"/g, '"').replace(/'/g, ''');@@@ -87,6 +91,7 @@ function s3error(req, res, code, message, extra = '') {const etagOf = (obj) => {const md = model.parseJson(obj.metadata, {});++ if (md.etag) return `"${md.etag}"`; // a multipart object: md5 of the parts' md5s, then -<parts>return md.md5 ? `"${md.md5}"` : `"${String(obj.content_hash || obj.id).slice(0, 32)}-1"`;};const lastModified = (obj) => new Date(String(obj.updated_at || obj.created_at).replace(' ', 'T') + (String(obj.updated_at || obj.created_at).endsWith('Z') ? '' : 'Z'));@@@ -303,48 +308,31 @@ function integrityProblem(req, parsed, got) {return null;}--async function putObject(req, res, zone, key) {-- const { key: akey, parsed } = req.s3;-- if (!mayWrite(akey)) { req.resume(); return s3error(req, res, 'AccessDenied', 'This access key is read-only'); }-- if (req.headers['x-amz-copy-source']) { req.resume(); return s3error(req, res, 'NotImplemented', 'CopyObject is not supported yet: read the object and write it again'); }-- const acl = String(req.headers['x-amz-acl'] || 'private');-- if (acl !== 'private' && acl !== 'bucket-owner-full-control') { req.resume(); return s3error(req, res, 'AccessDenied', 'Zones are private: objects cannot be made public over S3'); }-- if (DERIVED_KEY.test(key)) {-- req.resume();-- return s3error(req, res, 'InvalidArgument', 'Keys of the form [path/]med_<id> are the ids of objects written through the native API; choose another key');-- }-- const meta = userMeta(req);-- if (!meta) { req.resume(); return s3error(req, res, 'InvalidArgument', `User metadata is limited to ${MAX_META_BYTES} bytes`); }-- const declared = Number(sigv4.isStreaming(parsed) ? req.headers['x-amz-decoded-content-length'] : req.headers['content-length']);-- const cap = config.objects.maxUploadMb * 1024 * 1024;-- if (!Number.isFinite(declared)) { req.resume(); return s3error(req, res, 'MissingContentLength', 'You must provide the Content-Length HTTP header'); }-- if (declared > cap) { res.set('Connection', 'close'); req.resume(); return s3error(req, res, 'EntityTooLarge', `Objects sent in one request are at most ${cap} bytes`); }-- // Room for the declared size before a byte is stored; checked again with the real size below.-- const pre = await namespaces.checkQuota(req.appRow, zone.namespace, { bytes: declared, objects: 0 });-- if (pre) { res.set('Connection', 'close'); req.resume(); return s3error(req, res, 'QuotaExceeded', pre.detail); }---- const tmpDir = path.join(config.objects.path, '.tmp');-- fs.mkdirSync(tmpDir, { recursive: true });-- const tmp = path.join(tmpDir, `s3-${crypto.randomBytes(8).toString('hex')}`);-- const drop = () => { try { fs.unlinkSync(tmp); } catch { /* not written */ } };-- const got = await receive(req, tmp, Math.max(declared, 0), parsed);-- if (got.error) { drop(); return s3error(req, res, got.error.code, got.error.message); }-- if (got.bytes !== declared) { drop(); return s3error(req, res, 'IncompleteBody', `You did not provide the number of bytes specified by the Content-Length HTTP header (${declared}; ${got.bytes} received)`); }-- const bad = integrityProblem(req, parsed, got);-- if (bad) { drop(); return s3error(req, res, bad[0], bad[1]); }++const mimeOf = (req) => String(req.headers['content-type'] || 'application/octet-stream').split(';')[0].trim().toLowerCase().slice(0, 127) || 'application/octet-stream';-- const mime = String(req.headers['content-type'] || 'application/octet-stream').split(';')[0].trim().toLowerCase().slice(0, 127) || 'application/octet-stream';-- const md = { filename: key.split('/').pop().slice(0, 200), md5: got.md5.toString('hex'), s3: true, created_by: akey.id };-- if (Object.keys(meta).length) md.s3_meta = meta;++/** The metadata an S3 write keeps on its object: its file name, user metadata and the response headers it set. */++function objectMeta(req, key, meta) {++ const md = { filename: key.split('/').pop().slice(0, 200), s3: true, created_by: req.s3.key.id };++ if (meta && Object.keys(meta).length) md.s3_meta = meta;for (const [h, f] of [['content-disposition', 'content_disposition'], ['content-encoding', 'content_encoding'], ['cache-control', 'cache_control']]) {if (req.headers[h] && String(req.headers[h]).length <= 512) md[f] = String(req.headers[h]);}++ return md;++}++++/**++ * Make stored bytes (the file at `tmp`) the object under `key`, in one transaction under the tenant's quota lock: the++ * quota with the real size, the zone still active, the key's previous object through the deletion path (refused while++ * held), then the new object, ready, with its location, invariant and media.object.uploaded event. PutObject and++ * CompleteMultipartUpload both end here. → { obj } | { error: [code, message] } (the caller removes `tmp` on error).++ */++async function commitObject(req, zone, key, { tmp, bytes, sha256, mime, md }) {let outcome;try {const events = require('../events');outcome = await db.getDb().tx(async () => {const previous = await db.get(`SELECT * FROM media_objects WHERE zone_id = ? AND object_key = ? AND lifecycle_status = 'ready'`, [zone.id, key]);-- const q = await namespaces.checkQuota(req.appRow, zone.namespace, { bytes: got.bytes, objects: 1, excludeId: previous ? previous.id : null });++ const q = await namespaces.checkQuota(req.appRow, zone.namespace, { bytes, objects: 1, excludeId: previous ? previous.id : null });if (q) return { error: ['QuotaExceeded', q.detail] };const fresh = await zones.byId(zone.id);if (!fresh || fresh.state !== 'active') return { error: ['NoSuchBucket', 'The specified bucket does not exist'] };@@@ -353,7 +341,7 @@ async function putObject(req, res, zone, key) {const pmd = model.parseJson(previous.metadata, {});pmd.pre_delete_status = previous.lifecycle_status;pmd.retention_until = new Date(Date.now() + config.objects.retentionDays * 864e5).toISOString();-- pmd.deleted_by = `${akey.id} (overwrite)`;++ pmd.deleted_by = `${req.s3.key.id} (overwrite)`;await db.run(`UPDATE media_objects SET lifecycle_status = 'deleted', deleted_at = ov_now(), metadata = ?, updated_at = ov_now() WHERE id = ?`,[JSON.stringify(pmd), previous.id]);await namespaces.settle(previous.id);@@@ -361,15 +349,15 @@ async function putObject(req, res, zone, key) {await namespaces.ensure(req.appRow, db.rootNamespace(req.appRow));const id = await model.createObject({app_id: req.appId, namespace: zone.namespace, kind: 'file', owner_subject: null, owner_app: req.appId, owner_user_id: null,-- visibility: 'private', lifecycle_status: 'uploading', mime_type: mime, size_bytes: got.bytes, metadata: md,++ visibility: 'private', lifecycle_status: 'uploading', mime_type: mime, size_bytes: bytes, metadata: md,});const obj = await model.getObject(id);const dest = model.objectFilePath(obj);fs.mkdirSync(path.dirname(dest), { recursive: true });fs.renameSync(tmp, dest);await db.run(`UPDATE media_objects SET lifecycle_status = 'ready', content_hash = ?, canonical_provider = 'local', canonical_key = ?,-- zone_id = ?, object_key = ?, updated_at = ov_now() WHERE id = ?`, [got.sha256, dest, zone.id, key, id]);-- await model.upsertLocation(id, { provider: 'local', key: dest, state: 'present', size_bytes: got.bytes, checksum: got.sha256, verified: true });++ zone_id = ?, object_key = ?, updated_at = ov_now() WHERE id = ?`, [sha256, dest, zone.id, key, id]);++ await model.upsertLocation(id, { provider: 'local', key: dest, state: 'present', size_bytes: bytes, checksum: sha256, verified: true });const ready = await model.getObject(id);await require('../objects/invariant').record(ready);await events.record('media.object.uploaded', req.appId, await model.objectPublic(ready));@@@ -378,18 +366,174 @@ async function putObject(req, res, zone, key) {});require('../events').kick();} catch (err) {-- drop();-- console.error('[S3] put error:', err.message);-- return s3error(req, res, 'InternalError', 'We encountered an internal error. Please try again.');++ console.error('[S3] commit error:', err.message);++ return { error: ['InternalError', 'We encountered an internal error. Please try again.'] };}++ if (!outcome.error) await namespaces.reconcileChain(req.appRow, zone.namespace).catch(() => {});++ return outcome;++}++++async function putObject(req, res, zone, key) {++ const { key: akey, parsed } = req.s3;++ if (!mayWrite(akey)) { req.resume(); return s3error(req, res, 'AccessDenied', 'This access key is read-only'); }++ if (req.headers['x-amz-copy-source']) { req.resume(); return s3error(req, res, 'NotImplemented', 'CopyObject is not supported yet: read the object and write it again'); }++ const acl = String(req.headers['x-amz-acl'] || 'private');++ if (acl !== 'private' && acl !== 'bucket-owner-full-control') { req.resume(); return s3error(req, res, 'AccessDenied', 'Zones are private: objects cannot be made public over S3'); }++ if (DERIVED_KEY.test(key)) {++ req.resume();++ return s3error(req, res, 'InvalidArgument', 'Keys of the form [path/]med_<id> are the ids of objects written through the native API; choose another key');++ }++ const meta = userMeta(req);++ if (!meta) { req.resume(); return s3error(req, res, 'InvalidArgument', `User metadata is limited to ${MAX_META_BYTES} bytes`); }++ const declared = Number(sigv4.isStreaming(parsed) ? req.headers['x-amz-decoded-content-length'] : req.headers['content-length']);++ const cap = config.objects.maxUploadMb * 1024 * 1024;++ if (!Number.isFinite(declared)) { req.resume(); return s3error(req, res, 'MissingContentLength', 'You must provide the Content-Length HTTP header'); }++ if (declared > cap) { res.set('Connection', 'close'); req.resume(); return s3error(req, res, 'EntityTooLarge', `Objects sent in one request are at most ${cap} bytes`); }++ // Room for the declared size before a byte is stored; checked again with the real size below.++ const pre = await namespaces.checkQuota(req.appRow, zone.namespace, { bytes: declared, objects: 0 });++ if (pre) { res.set('Connection', 'close'); req.resume(); return s3error(req, res, 'QuotaExceeded', pre.detail); }++++ const tmpDir = path.join(config.objects.path, '.tmp');++ fs.mkdirSync(tmpDir, { recursive: true });++ const tmp = path.join(tmpDir, `s3-${crypto.randomBytes(8).toString('hex')}`);++ const drop = () => { try { fs.unlinkSync(tmp); } catch { /* not written */ } };++ const got = await receive(req, tmp, Math.max(declared, 0), parsed);++ if (got.error) { drop(); return s3error(req, res, got.error.code, got.error.message); }++ if (got.bytes !== declared) { drop(); return s3error(req, res, 'IncompleteBody', `You did not provide the number of bytes specified by the Content-Length HTTP header (${declared}; ${got.bytes} received)`); }++ const bad = integrityProblem(req, parsed, got);++ if (bad) { drop(); return s3error(req, res, bad[0], bad[1]); }++++ const md = { ...objectMeta(req, key, meta), md5: got.md5.toString('hex') };++ const outcome = await commitObject(req, zone, key, { tmp, bytes: got.bytes, sha256: got.sha256, mime: mimeOf(req), md });if (outcome.error) { drop(); return s3error(req, res, outcome.error[0], outcome.error[1]); }-- await namespaces.reconcileChain(req.appRow, zone.namespace).catch(() => {});baseHeaders(res);res.set('ETag', etagOf(outcome.obj));for (const k of ['x-amz-checksum-crc32', 'x-amz-checksum-sha256']) if (req.headers[k] || got.trailers[k]) res.set(k, req.headers[k] || got.trailers[k]);res.status(200).end();}++// ── Multipart uploads ────────────────────────────────────────++++/** Read a small XML request body (Complete, DeleteObjects), aws-chunked or not, checking x-amz-content-sha256. */++async function readXmlBody(req, limit = 2 * 1024 * 1024) {++ const chunks = [];++ let n = 0;++ const body = sigv4.isStreaming(req.s3.parsed) ? req.pipe(sigv4.chunkDecoder(req.s3.parsed)) : req;++ for await (const c of body) {++ n += c.length;++ if (n > limit) return { error: ['MalformedXML', 'The request body is too large'] };++ chunks.push(c);++ }++ const raw = Buffer.concat(chunks);++ if (/^[0-9a-f]{64}$/.test(req.s3.parsed.payloadHash) && crypto.createHash('sha256').update(raw).digest('hex') !== req.s3.parsed.payloadHash) {++ return { error: ['XAmzContentSHA256Mismatch', 'The provided x-amz-content-sha256 header does not match what was computed'] };++ }++ return { text: raw.toString('utf8') };++}++++async function createMultipart(req, res, zone, key) {++ req.resume();++ if (!mayWrite(req.s3.key)) return s3error(req, res, 'AccessDenied', 'This access key is read-only');++ const acl = String(req.headers['x-amz-acl'] || 'private');++ if (acl !== 'private' && acl !== 'bucket-owner-full-control') return s3error(req, res, 'AccessDenied', 'Zones are private: objects cannot be made public over S3');++ if (DERIVED_KEY.test(key)) return s3error(req, res, 'InvalidArgument', 'Keys of the form [path/]med_<id> are the ids of objects written through the native API; choose another key');++ const meta = userMeta(req);++ if (!meta) return s3error(req, res, 'InvalidArgument', `User metadata is limited to ${MAX_META_BYTES} bytes`);++ const up = await mp.create({ zone, key, appId: req.appId, accessKeyId: req.s3.key.id, contentType: mimeOf(req), metadata: objectMeta(req, key, meta) });++ xml(res, 200, `<InitiateMultipartUploadResult xmlns="${XMLNS}"><Bucket>${esc(zone.name)}</Bucket><Key>${esc(key)}</Key><UploadId>${up.id}</UploadId></InitiateMultipartUploadResult>`);++}++++async function uploadPart(req, res, zone, key) {++ const { key: akey, parsed } = req.s3;++ if (!mayWrite(akey)) { req.resume(); return s3error(req, res, 'AccessDenied', 'This access key is read-only'); }++ if (req.headers['x-amz-copy-source']) { req.resume(); return s3error(req, res, 'NotImplemented', 'UploadPartCopy is not supported yet'); }++ const n = Number(req.query.partNumber);++ if (!Number.isInteger(n) || n < 1 || n > 10000) { req.resume(); return s3error(req, res, 'InvalidArgument', 'Part number must be an integer between 1 and 10000, inclusive'); }++ const up = await mp.find(req.query.uploadId, zone, key);++ if (!up) { req.resume(); return s3error(req, res, 'NoSuchUpload', 'The specified upload does not exist. The upload ID may be invalid, or the upload may have been aborted or completed.'); }++ const declared = Number(sigv4.isStreaming(parsed) ? req.headers['x-amz-decoded-content-length'] : req.headers['content-length']);++ const cap = (config.objects.multipartMaxPartMb || 256) * 1024 * 1024;++ if (!Number.isFinite(declared)) { req.resume(); return s3error(req, res, 'MissingContentLength', 'You must provide the Content-Length HTTP header'); }++ if (declared > cap) { res.set('Connection', 'close'); req.resume(); return s3error(req, res, 'EntityTooLarge', `A part is at most ${cap} bytes`); }++ const total = await mp.bytesSoFar(up, n) + declared;++ if (total > (config.objects.multipartMaxMb || 20480) * 1024 * 1024) { res.set('Connection', 'close'); req.resume(); return s3error(req, res, 'EntityTooLarge', 'The object would be larger than this surface accepts'); }++ const pre = await namespaces.checkQuota(req.appRow, zone.namespace, { bytes: total, objects: 0 });++ if (pre) { res.set('Connection', 'close'); req.resume(); return s3error(req, res, 'QuotaExceeded', pre.detail); }++ const tmpDir = path.join(config.objects.path, '.tmp');++ fs.mkdirSync(tmpDir, { recursive: true });++ const tmp = path.join(tmpDir, `s3p-${crypto.randomBytes(8).toString('hex')}`);++ const drop = () => { try { fs.unlinkSync(tmp); } catch { /* not written */ } };++ const got = await receive(req, tmp, Math.max(declared, 0), parsed);++ if (got.error) { drop(); return s3error(req, res, got.error.code, got.error.message); }++ if (got.bytes !== declared) { drop(); return s3error(req, res, 'IncompleteBody', `You did not provide the number of bytes specified by the Content-Length HTTP header (${declared}; ${got.bytes} received)`); }++ const bad = integrityProblem(req, parsed, got);++ if (bad) { drop(); return s3error(req, res, bad[0], bad[1]); }++ if (!await mp.find(up.id, zone, key)) { drop(); return s3error(req, res, 'NoSuchUpload', 'The upload was aborted or completed while this part arrived'); }++ const part = await mp.storePart(up, n, tmp, { bytes: got.bytes, md5: got.md5.toString('hex') });++ baseHeaders(res);++ res.set('ETag', `"${part.md5}"`);++ for (const k of ['x-amz-checksum-crc32', 'x-amz-checksum-sha256']) if (req.headers[k] || got.trailers[k]) res.set(k, req.headers[k] || got.trailers[k]);++ res.status(200).end();++}++++async function completeMultipart(req, res, zone, key) {++ if (!mayWrite(req.s3.key)) { req.resume(); return s3error(req, res, 'AccessDenied', 'This access key is read-only'); }++ const up = await mp.find(req.query.uploadId, zone, key);++ if (!up) { req.resume(); return s3error(req, res, 'NoSuchUpload', 'The specified upload does not exist. The upload ID may be invalid, or the upload may have been aborted or completed.'); }++ const body = await readXmlBody(req).catch((err) => ({ error: [err.code && STATUS[err.code] ? err.code : 'IncompleteBody', err.message] }));++ if (body.error) return s3error(req, res, body.error[0], body.error[1]);++ const items = body.text.match(/<Part>[\s\S]*?<\/Part>/g) || [];++ if (!/<CompleteMultipartUpload[\s>]/.test(body.text) || !items.length || items.length > 10000) return s3error(req, res, 'MalformedXML', 'The XML you provided was not well-formed or did not validate');++ const wanted = items.map((x) => ({ n: Number((/<PartNumber>\s*(\d+)\s*<\/PartNumber>/.exec(x) || [])[1]), etag: ((/<ETag>([\s\S]*?)<\/ETag>/.exec(x) || [])[1] || '').replace(/"/g, '"') }));++ if (wanted.some((w) => !Number.isInteger(w.n))) return s3error(req, res, 'MalformedXML', 'Every part names its PartNumber');++ const joined = await mp.assemble(up, wanted);++ if (joined.error) return s3error(req, res, joined.error[0], joined.error[1]);++ const kept = up.metadata && typeof up.metadata === 'object' ? up.metadata : model.parseJson(up.metadata, {}); // jsonb++ const md = { ...kept, etag: joined.etag };++ const outcome = await commitObject(req, zone, key, { tmp: joined.tmp, bytes: joined.bytes, sha256: joined.sha256, mime: up.content_type, md });++ if (outcome.error) { try { fs.unlinkSync(joined.tmp); } catch { /* moved or gone */ } return s3error(req, res, outcome.error[0], outcome.error[1]); }++ await mp.abort(up);++ const host = String(req.headers.host || '');++ xml(res, 200, `<CompleteMultipartUploadResult xmlns="${XMLNS}"><Location>https://${esc(host)}/${esc(zone.name)}/${esc(sigv4.awsEncode(key, true))}</Location>`++ + `<Bucket>${esc(zone.name)}</Bucket><Key>${esc(key)}</Key><ETag>${esc(etagOf(outcome.obj))}</ETag></CompleteMultipartUploadResult>`);++}++++async function abortMultipart(req, res, zone, key) {++ req.resume();++ if (!mayWrite(req.s3.key)) return s3error(req, res, 'AccessDenied', 'This access key is read-only');++ const up = await mp.find(req.query.uploadId, zone, key);++ if (!up) return s3error(req, res, 'NoSuchUpload', 'The specified upload does not exist. The upload ID may be invalid, or the upload may have been aborted or completed.');++ await mp.abort(up);++ baseHeaders(res);++ res.status(204).end();++}++++async function listParts(req, res, zone, key) {++ if (!mayRead(req.s3.key)) return s3error(req, res, 'AccessDenied', 'This access key cannot read');++ const up = await mp.find(req.query.uploadId, zone, key);++ if (!up) return s3error(req, res, 'NoSuchUpload', 'The specified upload does not exist. The upload ID may be invalid, or the upload may have been aborted or completed.');++ const after = Number(req.query['part-number-marker']) || 0;++ const max = Math.min(1000, Math.max(1, Number(req.query['max-parts']) || 1000));++ const all = (await mp.parts(up)).filter((p) => p.part_number > after);++ const page = all.slice(0, max);++ const items = page.map((p) => `<Part><PartNumber>${p.part_number}</PartNumber><LastModified>${new Date(Number(p.created_at)).toISOString()}</LastModified>`++ + `<ETag>"${p.md5}"</ETag><Size>${Number(p.size_bytes)}</Size></Part>`).join('');++ const truncated = all.length > max;++ xml(res, 200, `<ListPartsResult xmlns="${XMLNS}"><Bucket>${esc(zone.name)}</Bucket><Key>${esc(key)}</Key><UploadId>${up.id}</UploadId>`++ + `<PartNumberMarker>${after}</PartNumberMarker>${truncated ? `<NextPartNumberMarker>${page[page.length - 1].part_number}</NextPartNumberMarker>` : ''}`++ + `<MaxParts>${max}</MaxParts><IsTruncated>${truncated}</IsTruncated><StorageClass>STANDARD</StorageClass>${items}</ListPartsResult>`);++}++++async function listUploads(req, res, zone) {++ if (!mayRead(req.s3.key)) return s3error(req, res, 'AccessDenied', 'This access key cannot read');++ const prefix = typeof req.query.prefix === 'string' ? req.query.prefix : '';++ const rows = await mp.listForZone(zone, { prefix });++ const items = rows.map((u) => `<Upload><Key>${esc(u.object_key)}</Key><UploadId>${u.id}</UploadId><StorageClass>STANDARD</StorageClass>`++ + `<Initiated>${new Date(Number(u.created_at)).toISOString()}</Initiated></Upload>`).join('');++ xml(res, 200, `<ListMultipartUploadsResult xmlns="${XMLNS}"><Bucket>${esc(zone.name)}</Bucket><Prefix>${esc(prefix)}</Prefix>`++ + `<MaxUploads>1000</MaxUploads><IsTruncated>false</IsTruncated>${items}</ListMultipartUploadsResult>`);++}++async function deleteOne(req, zone, key) {const obj = await findObject(zone, req.appId, key);if (!obj) return { ok: true }; // S3 answers 204 for a key that is not there@@@ -542,7 +686,8 @@ router.use(async (req, res) => {if (req.method === 'HEAD') { if (!mayRead(req.s3.key)) return s3error(req, res, 'AccessDenied', ''); baseHeaders(res); res.set('x-amz-bucket-region', 'us-east-1'); return res.status(200).end(); }if (req.method === 'GET' && q.location !== undefined) return xml(res, 200, `<LocationConstraint xmlns="${XMLNS}"/>`);if (req.method === 'POST' && q.delete !== undefined) return await deleteObjects(req, res, zone);-- if (req.method === 'GET' && !['acl', 'policy', 'versioning', 'cors', 'lifecycle', 'tagging', 'uploads', 'versions', 'website', 'encryption', 'logging', 'notification', 'replication', 'object-lock', 'ownershipControls', 'publicAccessBlock', 'accelerate', 'requestPayment'].some((k) => q[k] !== undefined)) {++ if (req.method === 'GET' && q.uploads !== undefined) return await listUploads(req, res, zone);++ if (req.method === 'GET' && !['acl', 'policy', 'versioning', 'cors', 'lifecycle', 'tagging', 'versions', 'website', 'encryption', 'logging', 'notification', 'replication', 'object-lock', 'ownershipControls', 'publicAccessBlock', 'accelerate', 'requestPayment'].some((k) => q[k] !== undefined)) {if (!mayRead(req.s3.key)) return s3error(req, res, 'AccessDenied', 'This access key cannot read');return await listObjects(req, res, zone, { v2: q['list-type'] === '2' });}@@@ -553,9 +698,16 @@ router.use(async (req, res) => {if (key == null) { req.resume(); return s3error(req, res, 'KeyTooLongError', `Keys are 1 to ${MAX_KEY_BYTES} bytes`); }const zone = await zoneFor(req, res, bucket);if (!zone) { req.resume(); return; }-- if (q.uploads !== undefined || q.uploadId !== undefined || q.partNumber !== undefined) {++ if (req.method === 'POST' && q.uploads !== undefined) return await createMultipart(req, res, zone, key);++ if (q.uploadId !== undefined) {++ if (req.method === 'PUT' && q.partNumber !== undefined) return await uploadPart(req, res, zone, key);++ if (req.method === 'POST') return await completeMultipart(req, res, zone, key);++ if (req.method === 'DELETE') return await abortMultipart(req, res, zone, key);++ if (req.method === 'GET') return await listParts(req, res, zone, key);++ }++ if (q.partNumber !== undefined) {req.resume();-- return s3error(req, res, 'NotImplemented', 'Multipart uploads are not supported yet: send objects up to the single-request limit in one PUT');++ return s3error(req, res, 'NotImplemented', 'Reading one part of a completed object is not supported; use a Range');}if (['acl', 'tagging', 'retention', 'legal-hold', 'torrent', 'restore', 'attributes', 'versionId'].some((k) => q[k] !== undefined)) {req.resume();@@@ -88,7 +88,7 @@ const http = require('http');lane: 'light', maxAttempts: 3, backoffS: () => 0,run: async (job, ctx) => {flakyCalls++;-- if (!ctx.checkpoint) { ctx.saveCheckpoint({ step: 1 }); throw new Error('transient'); }++ if (!ctx.checkpoint) { await ctx.saveCheckpoint({ step: 1 }); throw new Error('transient'); }return { resumed_from: ctx.checkpoint.step };},});nnew file mode 100644@@@ -0,0 +1,125 @@++'use strict';++// S3 multipart uploads on Media's S3 surface (server/s3/multipart.js) with the AWS SDK v3: the SDK's Upload helper++// (parallel parts) round-trips an 11 MiB object; the raw commands follow S3's rules (5 MiB minimum but for the last++// part, ascending order, matching ETags, a re-sent part replaces the earlier one); the ETag is the MD5 of the parts'++// MD5s; ListParts and ListMultipartUploads; Abort removes the parts; quotas count every part; uploads expire.++const assert = require('assert');++const fs = require('fs');++const os = require('os');++const path = require('path');++const http = require('http');++const crypto = require('crypto');++++(async () => {++ const tmp = fs.mkdtempSync(path.join(os.tmpdir(), 'ov-media-s3mp-'));++ const dir = (n) => { const d = path.join(tmp, n); fs.mkdirSync(d, { recursive: true }); return d; };++ process.env.VOD_PATH = dir('vods');++ process.env.FILES_PATH = dir('files');++ process.env.OBJECTS_PATH = dir('objects');++ process.env.THUMBNAILS_PATH = dir('thumbnails');++ process.env.OV_NETWORK_URL = 'https://openvibe.network';++ process.env.MEDIA_PUBLIC_URL = 'https://media.test';++ process.env.MEDIA_SIGNING_SECRET = 'test-signing-secret';++ process.env.MEDIA_S3_KEY_SECRET = 'k'.repeat(40); // a test value: 32+ characters++ process.env.MEDIA_S3_HOSTS = '127.0.0.1';++ process.env.MEDIA_APP_TENANT_QUOTA_MB = '40';++++ const db = require('../server/db/database');++ const auth = require('../server/auth');++ const mp = require('../server/s3/multipart');++ const { serviceAuth, zones: zoneContracts, ids } = require('openvibe-contracts');++ const express = require('express');++ const S3 = require('@aws-sdk/client-s3');++ const { Upload } = require('@aws-sdk/lib-storage');++++ const keys = crypto.generateKeyPairSync('rsa', { modulusLength: 2048, publicKeyEncoding: { type: 'spki', format: 'pem' }, privateKeyEncoding: { type: 'pkcs8', format: 'pem' } });++ auth._setNetworkPublicKeyForTests(keys.publicKey);++ const now = Math.floor(Date.now() / 1000);++ const svc = (sub, cap) => serviceAuth.signServiceToken({ iss: 'https://openvibe.network', sub, actor_type: 'service', aud: ['openvibe.media'], cap, iat: now, exp: now + 600, jti: `tok_${crypto.randomBytes(6).toString('hex')}` }, keys.privateKey);++ const ADMIN = `usr_${ids.ulid()}`;++ const PRJ = `prj_${ids.ulid()}`;++++ const s3router = require('../server/s3/routes');++ const admin = require('../server/s3/admin-routes');++ const app = express();++ app.use((req, res, next) => (s3router.isS3Host(req) ? s3router(req, res, next) : next()));++ app.use('/internal/zones', admin.internalRouter());++ app.use('/api/v1/projects/:project/access-keys', admin.keysRouter({ projects: { isAdmin: async (_p, s) => s === ADMIN } }));++ const server = http.createServer(app);++ await new Promise((r) => server.listen(0, '127.0.0.1', r));++ const port = server.address().port;++ const internal = async (method, p, token, body) => {++ const res = await fetch(`http://localhost:${port}${p}`, { method, headers: { authorization: `Bearer ${token}`, 'content-type': 'application/json' }, body: body ? JSON.stringify(body) : undefined });++ return { status: res.status, body: await res.json().catch(() => null) };++ };++ const zone = (name, over = {}) => {++ const z = { id: `zon_${ids.ulid()}`, project_id: PRJ, env: 'production', name, default: false, access: 'private', state: 'active', quota_bytes: null, revision: 1, created_at: new Date().toISOString(), ...over };++ return Object.assign(z, zoneContracts.identity(z));++ };++ const MiB = 1024 * 1024;++++ try {++ const big = zone('big');++ const small = zone('small', { quota_bytes: 7 * MiB });++ for (const z of [big, small]) assert.strictEqual((await internal('PUT', `/internal/zones/${z.id}`, svc('svc:zone', ['media.zone.provision']), z)).status, 200);++ const issued = (await internal('POST', `/api/v1/projects/${PRJ}/access-keys`, svc('svc:services', ['media.access_key.manage']), { env: 'production', name: 'mp', on_behalf_of: ADMIN })).body;++ const s3 = new S3.S3Client({ endpoint: `http://127.0.0.1:${port}`, region: 'us-east-1', forcePathStyle: true, credentials: { accessKeyId: issued.key.id, secretAccessKey: issued.secret } });++++ // ── The SDK's Upload helper: 11 MiB in parts of 5 MiB, sent in parallel ──++ const body = crypto.randomBytes(11 * MiB);++ const done = await new Upload({ client: s3, params: { Bucket: 'big', Key: 'videos/clip.bin', Body: body, ContentType: 'video/mp4', Metadata: { origin: 'camera' } }, partSize: 5 * MiB, queueSize: 3 }).done();++ const partMd5s = [0, 1, 2].map((i) => crypto.createHash('md5').update(body.subarray(i * 5 * MiB, Math.min((i + 1) * 5 * MiB, body.length))).digest());++ const want = `"${crypto.createHash('md5').update(Buffer.concat(partMd5s)).digest('hex')}-3"`;++ assert.strictEqual(done.ETag, want, 'the multipart ETag is the MD5 of the parts\' MD5s, then -3');++ let out = await s3.send(new S3.GetObjectCommand({ Bucket: 'big', Key: 'videos/clip.bin' }));++ assert.ok(Buffer.from(await out.Body.transformToByteArray()).equals(body), 'the joined object is the bytes sent');++ assert.deepStrictEqual([out.ContentType, out.Metadata.origin, out.ETag], ['video/mp4', 'camera', want]);++ const row = await db.get("SELECT * FROM media_objects WHERE zone_id = ? AND object_key = 'videos/clip.bin' AND lifecycle_status = 'ready'", [big.id]);++ assert.strictEqual(row.content_hash, crypto.createHash('sha256').update(body).digest('hex'), 'the object\'s sha256 is the whole object\'s');++ assert.strictEqual((await db.get('SELECT COUNT(*) AS n FROM media_s3_uploads')).n, 0, 'a completed upload leaves nothing behind');++ console.log('✅ the SDK\'s Upload helper round-trips 11 MiB in three parallel parts; S3\'s multipart ETag');++++ // ── The raw commands and S3's rules ──++ const create = await s3.send(new S3.CreateMultipartUploadCommand({ Bucket: 'big', Key: 'raw.bin' }));++ const UploadId = create.UploadId;++ assert.ok(/^mpu_/.test(UploadId));++ const p1 = Buffer.alloc(5 * MiB, 1), p2a = Buffer.alloc(1024, 2), p2b = Buffer.alloc(2048, 3), tiny = Buffer.alloc(10, 4);++ const e1 = (await s3.send(new S3.UploadPartCommand({ Bucket: 'big', Key: 'raw.bin', UploadId, PartNumber: 1, Body: p1 }))).ETag;++ await s3.send(new S3.UploadPartCommand({ Bucket: 'big', Key: 'raw.bin', UploadId, PartNumber: 2, Body: p2a }));++ const e2 = (await s3.send(new S3.UploadPartCommand({ Bucket: 'big', Key: 'raw.bin', UploadId, PartNumber: 2, Body: p2b }))).ETag;++ const eT = (await s3.send(new S3.UploadPartCommand({ Bucket: 'big', Key: 'raw.bin', UploadId, PartNumber: 3, Body: tiny }))).ETag;++ const listed = await s3.send(new S3.ListPartsCommand({ Bucket: 'big', Key: 'raw.bin', UploadId }));++ assert.deepStrictEqual(listed.Parts.map((p) => [p.PartNumber, p.Size]), [[1, 5 * MiB], [2, 2048], [3, 10]], 'a re-sent part replaces the earlier one');++ const uploads = await s3.send(new S3.ListMultipartUploadsCommand({ Bucket: 'big' }));++ assert.deepStrictEqual(uploads.Uploads.map((u) => u.UploadId), [UploadId]);++ await assert.rejects(s3.send(new S3.CompleteMultipartUploadCommand({ Bucket: 'big', Key: 'raw.bin', UploadId, MultipartUpload: { Parts: [{ PartNumber: 2, ETag: e2 }, { PartNumber: 1, ETag: e1 }] } })), /InvalidPartOrder/);++ await assert.rejects(s3.send(new S3.CompleteMultipartUploadCommand({ Bucket: 'big', Key: 'raw.bin', UploadId, MultipartUpload: { Parts: [{ PartNumber: 1, ETag: e1 }, { PartNumber: 2, ETag: '"00000000000000000000000000000000"' }] } })), /InvalidPart/);++ await assert.rejects(s3.send(new S3.CompleteMultipartUploadCommand({ Bucket: 'big', Key: 'raw.bin', UploadId, MultipartUpload: { Parts: [{ PartNumber: 1, ETag: e1 }, { PartNumber: 2, ETag: e2 }, { PartNumber: 3, ETag: eT }] } })), /EntityTooSmall/, 'every part but the last is at least 5 MiB');++ out = await s3.send(new S3.CompleteMultipartUploadCommand({ Bucket: 'big', Key: 'raw.bin', UploadId, MultipartUpload: { Parts: [{ PartNumber: 1, ETag: e1 }, { PartNumber: 2, ETag: e2 }] } }));++ assert.ok(out.ETag.endsWith('-2"'));++ out = await s3.send(new S3.GetObjectCommand({ Bucket: 'big', Key: 'raw.bin' }));++ assert.ok(Buffer.from(await out.Body.transformToByteArray()).equals(Buffer.concat([p1, p2b])), 'only the named parts, in order');++ await assert.rejects(s3.send(new S3.UploadPartCommand({ Bucket: 'big', Key: 'raw.bin', UploadId, PartNumber: 4, Body: tiny })), /NoSuchUpload/, 'a completed upload takes no more parts');++ console.log('✅ CreateMultipartUpload, UploadPart (replace), ListParts, ListMultipartUploads, Complete with S3\'s rules');++++ // ── Abort, quota, other keys and zones, expiry ──++ const ab = await s3.send(new S3.CreateMultipartUploadCommand({ Bucket: 'big', Key: 'gone.bin' }));++ await s3.send(new S3.UploadPartCommand({ Bucket: 'big', Key: 'gone.bin', UploadId: ab.UploadId, PartNumber: 1, Body: tiny }));++ await s3.send(new S3.AbortMultipartUploadCommand({ Bucket: 'big', Key: 'gone.bin', UploadId: ab.UploadId }));++ assert.ok(!fs.existsSync(path.join(process.env.OBJECTS_PATH, '.tmp', 's3mp', ab.UploadId)), 'abort removes the parts');++ await assert.rejects(s3.send(new S3.ListPartsCommand({ Bucket: 'big', Key: 'gone.bin', UploadId: ab.UploadId })), /NoSuchUpload/);++ await assert.rejects(s3.send(new S3.ListPartsCommand({ Bucket: 'small', Key: 'raw.bin', UploadId })), /NoSuchUpload/, 'an upload belongs to its zone and key');++ const q = await s3.send(new S3.CreateMultipartUploadCommand({ Bucket: 'small', Key: 'q.bin' }));++ await s3.send(new S3.UploadPartCommand({ Bucket: 'small', Key: 'q.bin', UploadId: q.UploadId, PartNumber: 1, Body: Buffer.alloc(5 * MiB) }));++ await assert.rejects(s3.send(new S3.UploadPartCommand({ Bucket: 'small', Key: 'q.bin', UploadId: q.UploadId, PartNumber: 2, Body: Buffer.alloc(3 * MiB) })), /QuotaExceeded/, 'parts count toward the zone\'s quota as they arrive');++ await db.run('UPDATE media_s3_uploads SET expires_at = 1 WHERE id = ?', [q.UploadId]);++ assert.strictEqual(await mp.sweep(), 1, 'an expired upload is swept');++ assert.ok(!fs.existsSync(path.join(process.env.OBJECTS_PATH, '.tmp', 's3mp', q.UploadId)));++ console.log('✅ abort, an upload bound to its zone and key, quota across parts, expiry');++ } finally {++ server.close();++ fs.rmSync(tmp, { recursive: true, force: true });++ }++ console.log('s3 multipart: all checks passed');++ process.exit(0);++})().catch((err) => { console.error(err); process.exit(1); });@@@ -227,10 +227,10 @@ const { Readable } = require('stream');await model.placeHold({ object_id: current.id, kind: 'evidence', reason: 'test' });await assert.rejects(s3.send(new S3.PutObjectCommand({ Bucket: 'assets', Key: 'readme.txt', Body: 'v3' })), /AccessDenied/, 'an overwrite of a held object is refused');await assert.rejects(s3.send(new S3.DeleteObjectCommand({ Bucket: 'assets', Key: 'readme.txt' })), /AccessDenied/, 'so is its deletion');-- await assert.rejects(s3.send(new S3.CreateMultipartUploadCommand({ Bucket: 'assets', Key: 'mp' })), /NotImplemented/);++ await assert.rejects(s3.send(new S3.UploadPartCopyCommand({ Bucket: 'assets', Key: 'mp', UploadId: (await s3.send(new S3.CreateMultipartUploadCommand({ Bucket: 'assets', Key: 'mp' }))).UploadId, PartNumber: 1, CopySource: 'assets/readme.txt' })), /NotImplemented/);await assert.rejects(s3.send(new S3.CopyObjectCommand({ Bucket: 'assets', Key: 'copy', CopySource: 'assets/readme.txt' })), /NotImplemented/);await assert.rejects(s3.send(new S3.DeleteBucketCommand({ Bucket: 'assets' })), /NotImplemented|MethodNotAllowed/);-- console.log('✅ quotas, overwrite through the deletion path, holds refuse overwrite and delete, NotImplemented answers');++ console.log('✅ quotas, overwrite through the deletion path, holds refuse overwrite and delete, NotImplemented answers (copy, part copy, bucket deletion)');// ── Deleting objects ──await s3.send(new S3.DeleteObjectCommand({ Bucket: 'assets', Key: 'site/index.html' }));
@@ -19,13 +19,14 @@ a named object zone of one project environment. Zones are made and deleted only
@@ -19,13 +19,14 @@ a named object zone of one project environment. Zones are made and deleted only
**Supported:**
**Supported:**
- `PutObject`, in one request, including aws-chunked bodies. `Content-MD5`, `x-amz-content-sha256` and the `x-amz-checksum-*` headers and trailers are checked.
- `PutObject`, in one request, including aws-chunked bodies. `Content-MD5`, `x-amz-content-sha256` and the `x-amz-checksum-*` headers and trailers are checked.
+- Multipart uploads (`server/s3/multipart.js`): `CreateMultipartUpload`, `UploadPart`, `CompleteMultipartUpload`, `AbortMultipartUpload`, `ListParts` and `ListMultipartUploads`. They follow S3's rules: every part but the last is at least 5 MiB, Complete names parts in ascending order with their ETags, and the object's ETag is the MD5 of the parts' MD5s with `-<count>`. Parts count toward the quota as they arrive. An upload expires after `MEDIA_MULTIPART_TTL_HOURS`.
- `GetObject` (with ranges, conditional headers and presigned response overrides) and `HeadObject`.
- `GetObject` (with ranges, conditional headers and presigned response overrides) and `HeadObject`.
- `DeleteObject` and `DeleteObjects`.
- `DeleteObject` and `DeleteObjects`.
- `ListObjectsV2` and `ListObjects` (prefix, delimiter and pages).
- `ListObjectsV2` and `ListObjects` (prefix, delimiter and pages).
- `ListBuckets`, `HeadBucket` and `GetBucketLocation`.
- `ListBuckets`, `HeadBucket` and `GetBucketLocation`.
- Presigned GET and PUT, valid for 7 days at most.
- Presigned GET and PUT, valid for 7 days at most.
-**Not implemented:** multipart uploads, `CopyObject`, ACLs, policies, versioning, tagging and bucket deletion.
+**Not implemented:** `CopyObject`, `UploadPartCopy`, ACLs, policies, versioning, tagging and bucket deletion.
`CreateBucket` answers 200 for a zone that already exists, so tools that probe this way keep working.
`CreateBucket` answers 200 for a zone that already exists, so tools that probe this way keep working.
## Rules
## Rules
@@ -40,6 +41,7 @@ a named object zone of one project environment. Zones are made and deleted only
@@ -40,6 +41,7 @@ a named object zone of one project environment. Zones are made and deleted only
## Tests
## Tests
- `test/s3-zones.test.js` runs end to end with the AWS SDK for JavaScript v3: every operation, presigned URLs, isolation, quotas, holds, two doors and zone deletion.
- `test/s3-zones.test.js` runs end to end with the AWS SDK for JavaScript v3: every operation, presigned URLs, isolation, quotas, holds, two doors and zone deletion.
+- `test/s3-multipart.test.js` uses the SDK's `Upload` helper (parallel parts) and the raw commands: S3's rules, ETags, abort, quota and expiry.
- `test/s3-sigv4.test.js` checks the worked examples of the S3 API reference, including the chunk signature chain.
- `test/s3-sigv4.test.js` checks the worked examples of the S3 API reference, including the chunk signature chain.
## Using it
## Using it
@@ -47,8 +49,7 @@ a named object zone of one project environment. Zones are made and deleted only
@@ -47,8 +49,7 @@ a named object zone of one project environment. Zones are made and deleted only
```sh
```sh
aws configure set aws_access_key_id ovk_… # issued on openvibe.zone
aws configure set aws_access_key_id ovk_… # issued on openvibe.zone
aws configure set aws_secret_access_key … # shown once
aws configure set aws_secret_access_key … # shown once
-aws configure set default.s3.multipart_threshold 256MB # no multipart yet: one request per object
aws --endpoint-url https://s3.openvibe.media s3 ls s3://assets/
aws --endpoint-url https://s3.openvibe.media s3 ls s3://assets/
rclone config create oz s3 provider Other endpoint https://s3.openvibe.media access_key_id ovk_… secret_access_key … \
rclone config create oz s3 provider Other endpoint https://s3.openvibe.media access_key_id ovk_… secret_access_key … \
- upload_cutoff 256M no_check_bucket true
+ no_check_bucket true
```
```
new file mode 100644
new file mode 100644
@@ -0,0 +1,26 @@
@@ -0,0 +1,26 @@
+-- phase: expand
+-- S3 multipart uploads (ADR-031, server/s3/multipart.js): an upload in progress and the parts it has received. The
+-- parts' bytes are files under OBJECTS_PATH/.tmp/s3mp/<upload id>/; CompleteMultipartUpload joins them into one object
+-- (the same commit as PutObject), AbortMultipartUpload and the expiry sweep remove them.
+CREATE TABLE media_s3_uploads (
+ id text COLLATE "C" PRIMARY KEY, -- the S3 UploadId (mpu_<ULID>)
+ zone_id text COLLATE "C" NOT NULL,
+ object_key text COLLATE "C" NOT NULL,
+ app_id text COLLATE "C" NOT NULL, -- the project tenant it counts against
+ access_key_id text COLLATE "C" NOT NULL, -- who started it
+ content_type text COLLATE "C" NOT NULL,
+ metadata jsonb NOT NULL DEFAULT '{}', -- the object metadata the completed object takes
+ created_at bigint NOT NULL,
+ expires_at bigint NOT NULL
+);
+CREATE INDEX media_s3_uploads_zone ON media_s3_uploads (zone_id, object_key, id);
+
+CREATE TABLE media_s3_parts (
+ upload_id text COLLATE "C" NOT NULL REFERENCES media_s3_uploads (id) ON DELETE CASCADE,
+ part_number integer NOT NULL CHECK (part_number BETWEEN 1 AND 10000),
+ size_bytes bigint NOT NULL,
+ md5 text COLLATE "C" NOT NULL, -- hex: the part's ETag
+ file_path text COLLATE "C" NOT NULL,
+ created_at bigint NOT NULL,
+ PRIMARY KEY (upload_id, part_number)
+);
@@ -305,6 +305,8 @@ const ready = (async () => {
@@ -305,6 +305,8 @@ const ready = (async () => {
try { const r = await require('./objects/timeline-cache').sweep(); if (r.removed) console.log(`[Objects] Timeline cache: evicted ${r.removed} local file(s)`); } catch (err) { console.warn('[Objects] timeline cache:', err.message); }
try { const r = await require('./objects/timeline-cache').sweep(); if (r.removed) console.log(`[Objects] Timeline cache: evicted ${r.removed} local file(s)`); } catch (err) { console.warn('[Objects] timeline cache:', err.message); }
// Incomplete multipart uploads past MEDIA_MULTIPART_TTL_HOURS: their parts are deleted (the objects stay uploading).
// Incomplete multipart uploads past MEDIA_MULTIPART_TTL_HOURS: their parts are deleted (the objects stay uploading).
try { const r = await require('./objects/multipart').purgeExpired(); if (r.expired || r.orphan_dirs) console.log(`[Objects] Multipart: ${r.expired} expired session(s), ${r.orphan_dirs} orphan part dir(s) removed`); } catch (err) { console.warn('[Objects] multipart purge:', err.message); }
try { const r = await require('./objects/multipart').purgeExpired(); if (r.expired || r.orphan_dirs) console.log(`[Objects] Multipart: ${r.expired} expired session(s), ${r.orphan_dirs} orphan part dir(s) removed`); } catch (err) { console.warn('[Objects] multipart purge:', err.message); }
+ // S3 multipart uploads past their expiry (MEDIA_MULTIPART_TTL_HOURS): their parts are removed (server/s3/multipart.js).
+ try { const n = await require('./s3/multipart').sweep(); if (n) console.log(`[S3] ${n} expired multipart upload(s) removed`); } catch (err) { console.warn('[S3] multipart sweep:', err.message); }
// Native objects' popularity (ADR-021): a finished UTC day's viewer hashes and salt are deleted, only its counts stay (30 days).
// Native objects' popularity (ADR-021): a finished UTC day's viewer hashes and salt are deleted, only its counts stay (30 days).
try { const r = await require('./objects/popularity').rotate(); if (r.hashes || r.salts || r.counts) console.log(`[Popularity] ${r.hashes} viewer hash(es) and ${r.salts} salt(s) of finished days deleted, ${r.counts} old daily count(s) pruned`); } catch (err) { console.warn('[Popularity] rotate:', err.message); }
try { const r = await require('./objects/popularity').rotate(); if (r.hashes || r.salts || r.counts) console.log(`[Popularity] ${r.hashes} viewer hash(es) and ${r.salts} salt(s) of finished days deleted, ${r.counts} old daily count(s) pruned`); } catch (err) { console.warn('[Popularity] rotate:', err.message); }
// Namespaces: uploads whose quota reservation expired are failed and their bytes freed; then every
// Namespaces: uploads whose quota reservation expired are failed and their bytes freed; then every
new file mode 100644
new file mode 100644
@@ -0,0 +1,119 @@
@@ -0,0 +1,119 @@
+'use strict';
+/**
+ * S3 multipart uploads on Media's S3 surface (ADR-031): what the AWS CLI, the SDKs' Upload helpers and rclone use for
+ * large files, and the way past a proxy's per-request size limit (each part is its own request).
+ *
+ * create({ zone, key, appId, accessKeyId, contentType, metadata }) → upload row (UploadId mpu_<ULID>)
+ * find(uploadId, zone, key) → the upload, or null (another zone or key: null)
+ * storePart(upload, n, tmp, { bytes, md5 }) → the part row (a re-sent part replaces the old)
+ * parts(upload) / bytesSoFar(upload, exceptPart)
+ * assemble(upload, wanted) wanted = [{ n, etag }] ascending → { tmp, bytes, sha256, etag } | { error }
+ * abort(upload) / sweep() remove the parts (files and rows); sweep ends uploads past expires_at
+ *
+ * S3's rules: part numbers 1 to 10000; every part but the last is at least 5 MiB; Complete names its parts in
+ * ascending order with their ETags; the object's ETag is the MD5 of the parts' MD5s, then "-<count>".
+ */
+const crypto = require('crypto');
+const fs = require('fs');
+const path = require('path');
+const { ids } = require('openvibe-contracts');
+const db = require('../db/database');
+const config = require('../config');
+
+const MIN_PART = 5 * 1024 * 1024;
+const TTL_MS = Math.max(1, config.objects.multipartTtlHours || 24) * 3600 * 1000;
+const dirOf = (uploadId) => path.join(config.objects.path, '.tmp', 's3mp', uploadId);
+
+async function create({ zone, key, appId, accessKeyId, contentType, metadata, now = Date.now() }) {
+ const id = `mpu_${ids.ulid(now)}`;
+ await db.run(`INSERT INTO media_s3_uploads (id, zone_id, object_key, app_id, access_key_id, content_type, metadata, created_at, expires_at)
+ VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)`, [id, zone.id, key, appId, accessKeyId, contentType, JSON.stringify(metadata || {}), now, now + TTL_MS]);
+ fs.mkdirSync(dirOf(id), { recursive: true });
+ return await db.get('SELECT * FROM media_s3_uploads WHERE id = ?', [id]);
+}
+
+async function find(uploadId, zone, key) {
+ if (!/^mpu_[0-9A-HJKMNP-TV-Z]{26}$/.test(String(uploadId || ''))) return null;
+ const u = await db.get('SELECT * FROM media_s3_uploads WHERE id = ?', [uploadId]);
+ return u && u.zone_id === zone.id && u.object_key === key ? u : null;
+}
+
+async function parts(upload) {
+ return await db.all('SELECT * FROM media_s3_parts WHERE upload_id = ? ORDER BY part_number', [upload.id]);
+}
+
+async function bytesSoFar(upload, exceptPart = 0) {
+ const r = await db.get('SELECT COALESCE(SUM(size_bytes), 0) AS b FROM media_s3_parts WHERE upload_id = ? AND part_number <> ?', [upload.id, exceptPart]);
+ return Number(r.b);
+}
+
+/** Move a received part (the file at `tmp`) into the upload; a part sent again replaces the earlier one. */
+async function storePart(upload, n, tmp, { bytes, md5, now = Date.now() }) {
+ const dest = path.join(dirOf(upload.id), String(n).padStart(5, '0'));
+ fs.mkdirSync(path.dirname(dest), { recursive: true });
+ fs.renameSync(tmp, dest);
+ await db.run(`INSERT INTO media_s3_parts (upload_id, part_number, size_bytes, md5, file_path, created_at) VALUES (?, ?, ?, ?, ?, ?)
+ ON CONFLICT (upload_id, part_number) DO UPDATE SET size_bytes = excluded.size_bytes, md5 = excluded.md5,
+ file_path = excluded.file_path, created_at = excluded.created_at`, [upload.id, n, bytes, md5, dest, now]);
+ return await db.get('SELECT * FROM media_s3_parts WHERE upload_id = ? AND part_number = ?', [upload.id, n]);
+}
+
+/** Join the named parts into one file, checking S3's rules. The upload's own part files stay until abort or commit. */
+async function assemble(upload, wanted) {
+ if (!wanted.length) return { error: ['MalformedXML', 'The XML you provided was not well-formed or did not validate'] };
+ for (let i = 1; i < wanted.length; i++) {
+ if (wanted[i].n <= wanted[i - 1].n) return { error: ['InvalidPartOrder', 'The list of parts was not in ascending order'] };
+ }
+ const have = new Map((await parts(upload)).map((p) => [p.part_number, p]));
+ const chosen = [];
+ for (const [i, w] of wanted.entries()) {
+ const p = have.get(w.n);
+ const etag = String(w.etag || '').replace(/^"|"$/g, '').toLowerCase();
+ if (!p || p.md5 !== etag) return { error: ['InvalidPart', `Part ${w.n} was not uploaded, or its ETag does not match`] };
+ if (i < wanted.length - 1 && Number(p.size_bytes) < MIN_PART) return { error: ['EntityTooSmall', `Part ${w.n} is smaller than the 5 MiB every part but the last must be`] };
+ if (!fs.existsSync(p.file_path)) return { error: ['InvalidPart', `Part ${w.n} is no longer stored`] };
+ chosen.push(p);
+ }
+ const tmp = path.join(config.objects.path, '.tmp', `s3mp-${upload.id}-${crypto.randomBytes(4).toString('hex')}`);
+ const sha = crypto.createHash('sha256');
+ const md5s = crypto.createHash('md5');
+ let bytes = 0;
+ const out = fs.createWriteStream(tmp);
+ try {
+ for (const p of chosen) {
+ md5s.update(Buffer.from(p.md5, 'hex'));
+ await new Promise((resolve, reject) => {
+ const src = fs.createReadStream(p.file_path);
+ src.on('data', (c) => { sha.update(c); bytes += c.length; });
+ src.on('error', reject);
+ out.once('error', reject);
+ src.on('end', resolve);
+ src.pipe(out, { end: false });
+ });
+ }
+ await new Promise((resolve, reject) => out.end((err) => (err ? reject(err) : resolve())));
+ } catch (err) {
+ out.destroy();
+ try { fs.unlinkSync(tmp); } catch { /* not written */ }
+ return { error: ['InternalError', `The parts could not be joined: ${err.message}`] };
+ }
+ return { tmp, bytes, sha256: sha.digest('hex'), etag: `${md5s.digest('hex')}-${chosen.length}` };
+}
+
+async function abort(upload) {
+ fs.rmSync(dirOf(upload.id), { recursive: true, force: true });
+ await db.run('DELETE FROM media_s3_uploads WHERE id = ?', [upload.id]);
+}
+
+async function sweep({ now = Date.now() } = {}) {
+ const old = await db.all('SELECT * FROM media_s3_uploads WHERE expires_at < ? LIMIT 500', [now]);
+ for (const u of old) await abort(u);
+ return old.length;
+}
+
+async function listForZone(zone, { prefix = '', limit = 1000 } = {}) {
+ return await db.all(`SELECT * FROM media_s3_uploads WHERE zone_id = ? AND object_key LIKE ? ESCAPE '\\' ORDER BY object_key, id LIMIT ?`,
+ [zone.id, `${prefix.replace(/[\\%_]/g, (c) => `\\${c}`)}%`, limit]);
+}
+
+module.exports = { create, find, parts, bytesSoFar, storePart, assemble, abort, sweep, listForZone, MIN_PART };
@@ -12,10 +12,12 @@
@@ -12,10 +12,12 @@
* PUT /<zone> CreateBucket: 200 for a zone that exists (clients probe this way), else NotImplemented
* PUT /<zone> CreateBucket: 200 for a zone that exists (clients probe this way), else NotImplemented
* POST /<zone>?delete DeleteObjects
* POST /<zone>?delete DeleteObjects
* PUT /<zone>/<key> PutObject (one request; aws-chunked bodies too)
* PUT /<zone>/<key> PutObject (one request; aws-chunked bodies too)
+ * POST /<zone>/<key>?uploads CreateMultipartUpload; PUT ?partNumber&uploadId UploadPart; POST ?uploadId Complete;
+ * DELETE ?uploadId Abort; GET ?uploadId ListParts; GET /<zone>?uploads ListMultipartUploads
* GET /<zone>/<key> GetObject (Range)
* GET /<zone>/<key> GetObject (Range)
* HEAD /<zone>/<key> HeadObject
* HEAD /<zone>/<key> HeadObject
* DELETE /<zone>/<key> DeleteObject (Media's soft delete; refused under a retention hold)
* DELETE /<zone>/<key> DeleteObject (Media's soft delete; refused under a retention hold)
- * everything else NotImplemented (multipart, copy, ACLs, policies, versioning, bucket deletion)
+ * everything else NotImplemented (copy, part copy, ACLs, policies, versioning, bucket deletion)
*
*
* Every request is signed (SigV4, header or presigned) with an access key (server/s3/keys.js). The key fixes the
* Every request is signed (SigV4, header or presigned) with an access key (server/s3/keys.js). The key fixes the
* project and environment, so a bucket name resolves to one zone id of that environment (server/s3/zones.js); a key
* project and environment, so a bucket name resolves to one zone id of that environment (server/s3/zones.js); a key
@@ -42,6 +44,7 @@ const namespaces = require('../objects/namespaces');
@@ -42,6 +44,7 @@ const namespaces = require('../objects/namespaces');
const sigv4 = require('./sigv4');
const sigv4 = require('./sigv4');
const keys = require('./keys');
const keys = require('./keys');
const zones = require('./zones');
const zones = require('./zones');
+const mp = require('./multipart');
const MAX_KEY_BYTES = 1024;
const MAX_KEY_BYTES = 1024;
const MAX_LIST = 1000;
const MAX_LIST = 1000;
@@ -55,6 +58,7 @@ const STATUS = {
@@ -55,6 +58,7 @@ const STATUS = {
KeyTooLongError: 400, MissingContentLength: 411, EntityTooLarge: 400, BadDigest: 400, XAmzContentSHA256Mismatch: 400,
KeyTooLongError: 400, MissingContentLength: 411, EntityTooLarge: 400, BadDigest: 400, XAmzContentSHA256Mismatch: 400,
IncompleteBody: 400, MalformedXML: 400, InvalidRange: 416, NoSuchBucket: 404, NoSuchKey: 404, QuotaExceeded: 403,
IncompleteBody: 400, MalformedXML: 400, InvalidRange: 416, NoSuchBucket: 404, NoSuchKey: 404, QuotaExceeded: 403,
NotImplemented: 501, MethodNotAllowed: 405, InternalError: 500, ServiceUnavailable: 503, InvalidBucketName: 400,
NotImplemented: 501, MethodNotAllowed: 405, InternalError: 500, ServiceUnavailable: 503, InvalidBucketName: 400,
+ NoSuchUpload: 404, EntityTooSmall: 400, InvalidPart: 400, InvalidPartOrder: 400, PreconditionFailed: 412,
};
};
const esc = (s) => String(s).replace(/&/g, '&').replace(/</g, '<').replace(/>/g, '>').replace(/"/g, '"').replace(/'/g, ''');
const esc = (s) => String(s).replace(/&/g, '&').replace(/</g, '<').replace(/>/g, '>').replace(/"/g, '"').replace(/'/g, ''');
@@ -87,6 +91,7 @@ function s3error(req, res, code, message, extra = '') {
@@ -87,6 +91,7 @@ function s3error(req, res, code, message, extra = '') {
const etagOf = (obj) => {
const etagOf = (obj) => {
const md = model.parseJson(obj.metadata, {});
const md = model.parseJson(obj.metadata, {});
+ if (md.etag) return `"${md.etag}"`; // a multipart object: md5 of the parts' md5s, then -<parts>
return md.md5 ? `"${md.md5}"` : `"${String(obj.content_hash || obj.id).slice(0, 32)}-1"`;
return md.md5 ? `"${md.md5}"` : `"${String(obj.content_hash || obj.id).slice(0, 32)}-1"`;
};
};
const lastModified = (obj) => new Date(String(obj.updated_at || obj.created_at).replace(' ', 'T') + (String(obj.updated_at || obj.created_at).endsWith('Z') ? '' : 'Z'));
const lastModified = (obj) => new Date(String(obj.updated_at || obj.created_at).replace(' ', 'T') + (String(obj.updated_at || obj.created_at).endsWith('Z') ? '' : 'Z'));
@@ -303,48 +308,31 @@ function integrityProblem(req, parsed, got) {
@@ -303,48 +308,31 @@ function integrityProblem(req, parsed, got) {
return null;
return null;
}
}
-async function putObject(req, res, zone, key) {
- const { key: akey, parsed } = req.s3;
- if (!mayWrite(akey)) { req.resume(); return s3error(req, res, 'AccessDenied', 'This access key is read-only'); }
- if (req.headers['x-amz-copy-source']) { req.resume(); return s3error(req, res, 'NotImplemented', 'CopyObject is not supported yet: read the object and write it again'); }
- const acl = String(req.headers['x-amz-acl'] || 'private');
- if (acl !== 'private' && acl !== 'bucket-owner-full-control') { req.resume(); return s3error(req, res, 'AccessDenied', 'Zones are private: objects cannot be made public over S3'); }
- if (DERIVED_KEY.test(key)) {
- req.resume();
- return s3error(req, res, 'InvalidArgument', 'Keys of the form [path/]med_<id> are the ids of objects written through the native API; choose another key');
- }
- const meta = userMeta(req);
- if (!meta) { req.resume(); return s3error(req, res, 'InvalidArgument', `User metadata is limited to ${MAX_META_BYTES} bytes`); }
- const declared = Number(sigv4.isStreaming(parsed) ? req.headers['x-amz-decoded-content-length'] : req.headers['content-length']);
- const cap = config.objects.maxUploadMb * 1024 * 1024;
- if (!Number.isFinite(declared)) { req.resume(); return s3error(req, res, 'MissingContentLength', 'You must provide the Content-Length HTTP header'); }
- if (declared > cap) { res.set('Connection', 'close'); req.resume(); return s3error(req, res, 'EntityTooLarge', `Objects sent in one request are at most ${cap} bytes`); }
- // Room for the declared size before a byte is stored; checked again with the real size below.
- const pre = await namespaces.checkQuota(req.appRow, zone.namespace, { bytes: declared, objects: 0 });
- if (pre) { res.set('Connection', 'close'); req.resume(); return s3error(req, res, 'QuotaExceeded', pre.detail); }
-
- const tmpDir = path.join(config.objects.path, '.tmp');
- fs.mkdirSync(tmpDir, { recursive: true });
- const tmp = path.join(tmpDir, `s3-${crypto.randomBytes(8).toString('hex')}`);
- const drop = () => { try { fs.unlinkSync(tmp); } catch { /* not written */ } };
- const got = await receive(req, tmp, Math.max(declared, 0), parsed);
- if (got.error) { drop(); return s3error(req, res, got.error.code, got.error.message); }
- if (got.bytes !== declared) { drop(); return s3error(req, res, 'IncompleteBody', `You did not provide the number of bytes specified by the Content-Length HTTP header (${declared}; ${got.bytes} received)`); }
- const bad = integrityProblem(req, parsed, got);
- if (bad) { drop(); return s3error(req, res, bad[0], bad[1]); }
+const mimeOf = (req) => String(req.headers['content-type'] || 'application/octet-stream').split(';')[0].trim().toLowerCase().slice(0, 127) || 'application/octet-stream';
- const mime = String(req.headers['content-type'] || 'application/octet-stream').split(';')[0].trim().toLowerCase().slice(0, 127) || 'application/octet-stream';
- const md = { filename: key.split('/').pop().slice(0, 200), md5: got.md5.toString('hex'), s3: true, created_by: akey.id };
- if (Object.keys(meta).length) md.s3_meta = meta;
+/** The metadata an S3 write keeps on its object: its file name, user metadata and the response headers it set. */
+function objectMeta(req, key, meta) {
+ const md = { filename: key.split('/').pop().slice(0, 200), s3: true, created_by: req.s3.key.id };
+ if (meta && Object.keys(meta).length) md.s3_meta = meta;
for (const [h, f] of [['content-disposition', 'content_disposition'], ['content-encoding', 'content_encoding'], ['cache-control', 'cache_control']]) {
for (const [h, f] of [['content-disposition', 'content_disposition'], ['content-encoding', 'content_encoding'], ['cache-control', 'cache_control']]) {
if (req.headers[h] && String(req.headers[h]).length <= 512) md[f] = String(req.headers[h]);
if (req.headers[h] && String(req.headers[h]).length <= 512) md[f] = String(req.headers[h]);
}
}
+ return md;
+}
+
+/**
+ * Make stored bytes (the file at `tmp`) the object under `key`, in one transaction under the tenant's quota lock: the
+ * quota with the real size, the zone still active, the key's previous object through the deletion path (refused while
+ * held), then the new object, ready, with its location, invariant and media.object.uploaded event. PutObject and
+ * CompleteMultipartUpload both end here. → { obj } | { error: [code, message] } (the caller removes `tmp` on error).
+ */
+async function commitObject(req, zone, key, { tmp, bytes, sha256, mime, md }) {
let outcome;
let outcome;
try {
try {
const events = require('../events');
const events = require('../events');
outcome = await db.getDb().tx(async () => {
outcome = await db.getDb().tx(async () => {
const previous = await db.get(`SELECT * FROM media_objects WHERE zone_id = ? AND object_key = ? AND lifecycle_status = 'ready'`, [zone.id, key]);
const previous = await db.get(`SELECT * FROM media_objects WHERE zone_id = ? AND object_key = ? AND lifecycle_status = 'ready'`, [zone.id, key]);
- const q = await namespaces.checkQuota(req.appRow, zone.namespace, { bytes: got.bytes, objects: 1, excludeId: previous ? previous.id : null });
+ const q = await namespaces.checkQuota(req.appRow, zone.namespace, { bytes, objects: 1, excludeId: previous ? previous.id : null });
if (q) return { error: ['QuotaExceeded', q.detail] };
if (q) return { error: ['QuotaExceeded', q.detail] };
const fresh = await zones.byId(zone.id);
const fresh = await zones.byId(zone.id);
if (!fresh || fresh.state !== 'active') return { error: ['NoSuchBucket', 'The specified bucket does not exist'] };
if (!fresh || fresh.state !== 'active') return { error: ['NoSuchBucket', 'The specified bucket does not exist'] };
@@ -353,7 +341,7 @@ async function putObject(req, res, zone, key) {
@@ -353,7 +341,7 @@ async function putObject(req, res, zone, key) {
const pmd = model.parseJson(previous.metadata, {});
const pmd = model.parseJson(previous.metadata, {});
pmd.pre_delete_status = previous.lifecycle_status;
pmd.pre_delete_status = previous.lifecycle_status;
pmd.retention_until = new Date(Date.now() + config.objects.retentionDays * 864e5).toISOString();
pmd.retention_until = new Date(Date.now() + config.objects.retentionDays * 864e5).toISOString();
- pmd.deleted_by = `${akey.id} (overwrite)`;
+ pmd.deleted_by = `${req.s3.key.id} (overwrite)`;
await db.run(`UPDATE media_objects SET lifecycle_status = 'deleted', deleted_at = ov_now(), metadata = ?, updated_at = ov_now() WHERE id = ?`,
await db.run(`UPDATE media_objects SET lifecycle_status = 'deleted', deleted_at = ov_now(), metadata = ?, updated_at = ov_now() WHERE id = ?`,
[JSON.stringify(pmd), previous.id]);
[JSON.stringify(pmd), previous.id]);
await namespaces.settle(previous.id);
await namespaces.settle(previous.id);
@@ -361,15 +349,15 @@ async function putObject(req, res, zone, key) {
@@ -361,15 +349,15 @@ async function putObject(req, res, zone, key) {
await namespaces.ensure(req.appRow, db.rootNamespace(req.appRow));
await namespaces.ensure(req.appRow, db.rootNamespace(req.appRow));
const id = await model.createObject({
const id = await model.createObject({
app_id: req.appId, namespace: zone.namespace, kind: 'file', owner_subject: null, owner_app: req.appId, owner_user_id: null,
app_id: req.appId, namespace: zone.namespace, kind: 'file', owner_subject: null, owner_app: req.appId, owner_user_id: null,
- visibility: 'private', lifecycle_status: 'uploading', mime_type: mime, size_bytes: got.bytes, metadata: md,
+ visibility: 'private', lifecycle_status: 'uploading', mime_type: mime, size_bytes: bytes, metadata: md,
});
});
const obj = await model.getObject(id);
const obj = await model.getObject(id);
const dest = model.objectFilePath(obj);
const dest = model.objectFilePath(obj);
fs.mkdirSync(path.dirname(dest), { recursive: true });
fs.mkdirSync(path.dirname(dest), { recursive: true });
fs.renameSync(tmp, dest);
fs.renameSync(tmp, dest);
await db.run(`UPDATE media_objects SET lifecycle_status = 'ready', content_hash = ?, canonical_provider = 'local', canonical_key = ?,
await db.run(`UPDATE media_objects SET lifecycle_status = 'ready', content_hash = ?, canonical_provider = 'local', canonical_key = ?,
- zone_id = ?, object_key = ?, updated_at = ov_now() WHERE id = ?`, [got.sha256, dest, zone.id, key, id]);
- await model.upsertLocation(id, { provider: 'local', key: dest, state: 'present', size_bytes: got.bytes, checksum: got.sha256, verified: true });
+ zone_id = ?, object_key = ?, updated_at = ov_now() WHERE id = ?`, [sha256, dest, zone.id, key, id]);
+ await model.upsertLocation(id, { provider: 'local', key: dest, state: 'present', size_bytes: bytes, checksum: sha256, verified: true });
const ready = await model.getObject(id);
const ready = await model.getObject(id);
await require('../objects/invariant').record(ready);
await require('../objects/invariant').record(ready);
await events.record('media.object.uploaded', req.appId, await model.objectPublic(ready));
await events.record('media.object.uploaded', req.appId, await model.objectPublic(ready));
@@ -378,18 +366,174 @@ async function putObject(req, res, zone, key) {
@@ -378,18 +366,174 @@ async function putObject(req, res, zone, key) {
});
});
require('../events').kick();
require('../events').kick();
} catch (err) {
} catch (err) {
- drop();
- console.error('[S3] put error:', err.message);
- return s3error(req, res, 'InternalError', 'We encountered an internal error. Please try again.');
+ console.error('[S3] commit error:', err.message);
+ return { error: ['InternalError', 'We encountered an internal error. Please try again.'] };
}
}
+ if (!outcome.error) await namespaces.reconcileChain(req.appRow, zone.namespace).catch(() => {});
+ return outcome;
+}
+
+async function putObject(req, res, zone, key) {
+ const { key: akey, parsed } = req.s3;
+ if (!mayWrite(akey)) { req.resume(); return s3error(req, res, 'AccessDenied', 'This access key is read-only'); }
+ if (req.headers['x-amz-copy-source']) { req.resume(); return s3error(req, res, 'NotImplemented', 'CopyObject is not supported yet: read the object and write it again'); }
+ const acl = String(req.headers['x-amz-acl'] || 'private');
+ if (acl !== 'private' && acl !== 'bucket-owner-full-control') { req.resume(); return s3error(req, res, 'AccessDenied', 'Zones are private: objects cannot be made public over S3'); }
+ if (DERIVED_KEY.test(key)) {
+ req.resume();
+ return s3error(req, res, 'InvalidArgument', 'Keys of the form [path/]med_<id> are the ids of objects written through the native API; choose another key');
+ }
+ const meta = userMeta(req);
+ if (!meta) { req.resume(); return s3error(req, res, 'InvalidArgument', `User metadata is limited to ${MAX_META_BYTES} bytes`); }
+ const declared = Number(sigv4.isStreaming(parsed) ? req.headers['x-amz-decoded-content-length'] : req.headers['content-length']);
+ const cap = config.objects.maxUploadMb * 1024 * 1024;
+ if (!Number.isFinite(declared)) { req.resume(); return s3error(req, res, 'MissingContentLength', 'You must provide the Content-Length HTTP header'); }
+ if (declared > cap) { res.set('Connection', 'close'); req.resume(); return s3error(req, res, 'EntityTooLarge', `Objects sent in one request are at most ${cap} bytes`); }
+ // Room for the declared size before a byte is stored; checked again with the real size below.
+ const pre = await namespaces.checkQuota(req.appRow, zone.namespace, { bytes: declared, objects: 0 });
+ if (pre) { res.set('Connection', 'close'); req.resume(); return s3error(req, res, 'QuotaExceeded', pre.detail); }
+
+ const tmpDir = path.join(config.objects.path, '.tmp');
+ fs.mkdirSync(tmpDir, { recursive: true });
+ const tmp = path.join(tmpDir, `s3-${crypto.randomBytes(8).toString('hex')}`);
+ const drop = () => { try { fs.unlinkSync(tmp); } catch { /* not written */ } };
+ const got = await receive(req, tmp, Math.max(declared, 0), parsed);
+ if (got.error) { drop(); return s3error(req, res, got.error.code, got.error.message); }
+ if (got.bytes !== declared) { drop(); return s3error(req, res, 'IncompleteBody', `You did not provide the number of bytes specified by the Content-Length HTTP header (${declared}; ${got.bytes} received)`); }
+ const bad = integrityProblem(req, parsed, got);
+ if (bad) { drop(); return s3error(req, res, bad[0], bad[1]); }
+
+ const md = { ...objectMeta(req, key, meta), md5: got.md5.toString('hex') };
+ const outcome = await commitObject(req, zone, key, { tmp, bytes: got.bytes, sha256: got.sha256, mime: mimeOf(req), md });
if (outcome.error) { drop(); return s3error(req, res, outcome.error[0], outcome.error[1]); }
if (outcome.error) { drop(); return s3error(req, res, outcome.error[0], outcome.error[1]); }
- await namespaces.reconcileChain(req.appRow, zone.namespace).catch(() => {});
baseHeaders(res);
baseHeaders(res);
res.set('ETag', etagOf(outcome.obj));
res.set('ETag', etagOf(outcome.obj));
for (const k of ['x-amz-checksum-crc32', 'x-amz-checksum-sha256']) if (req.headers[k] || got.trailers[k]) res.set(k, req.headers[k] || got.trailers[k]);
for (const k of ['x-amz-checksum-crc32', 'x-amz-checksum-sha256']) if (req.headers[k] || got.trailers[k]) res.set(k, req.headers[k] || got.trailers[k]);
res.status(200).end();
res.status(200).end();
}
}
+// ── Multipart uploads ────────────────────────────────────────
+
+/** Read a small XML request body (Complete, DeleteObjects), aws-chunked or not, checking x-amz-content-sha256. */
+async function readXmlBody(req, limit = 2 * 1024 * 1024) {
+ const chunks = [];
+ let n = 0;
+ const body = sigv4.isStreaming(req.s3.parsed) ? req.pipe(sigv4.chunkDecoder(req.s3.parsed)) : req;
+ for await (const c of body) {
+ n += c.length;
+ if (n > limit) return { error: ['MalformedXML', 'The request body is too large'] };
+ chunks.push(c);
+ }
+ const raw = Buffer.concat(chunks);
+ if (/^[0-9a-f]{64}$/.test(req.s3.parsed.payloadHash) && crypto.createHash('sha256').update(raw).digest('hex') !== req.s3.parsed.payloadHash) {
+ return { error: ['XAmzContentSHA256Mismatch', 'The provided x-amz-content-sha256 header does not match what was computed'] };
+ }
+ return { text: raw.toString('utf8') };
+}
+
+async function createMultipart(req, res, zone, key) {
+ req.resume();
+ if (!mayWrite(req.s3.key)) return s3error(req, res, 'AccessDenied', 'This access key is read-only');
+ const acl = String(req.headers['x-amz-acl'] || 'private');
+ if (acl !== 'private' && acl !== 'bucket-owner-full-control') return s3error(req, res, 'AccessDenied', 'Zones are private: objects cannot be made public over S3');
+ if (DERIVED_KEY.test(key)) return s3error(req, res, 'InvalidArgument', 'Keys of the form [path/]med_<id> are the ids of objects written through the native API; choose another key');
+ const meta = userMeta(req);
+ if (!meta) return s3error(req, res, 'InvalidArgument', `User metadata is limited to ${MAX_META_BYTES} bytes`);
+ const up = await mp.create({ zone, key, appId: req.appId, accessKeyId: req.s3.key.id, contentType: mimeOf(req), metadata: objectMeta(req, key, meta) });
+ xml(res, 200, `<InitiateMultipartUploadResult xmlns="${XMLNS}"><Bucket>${esc(zone.name)}</Bucket><Key>${esc(key)}</Key><UploadId>${up.id}</UploadId></InitiateMultipartUploadResult>`);
+}
+
+async function uploadPart(req, res, zone, key) {
+ const { key: akey, parsed } = req.s3;
+ if (!mayWrite(akey)) { req.resume(); return s3error(req, res, 'AccessDenied', 'This access key is read-only'); }
+ if (req.headers['x-amz-copy-source']) { req.resume(); return s3error(req, res, 'NotImplemented', 'UploadPartCopy is not supported yet'); }
+ const n = Number(req.query.partNumber);
+ if (!Number.isInteger(n) || n < 1 || n > 10000) { req.resume(); return s3error(req, res, 'InvalidArgument', 'Part number must be an integer between 1 and 10000, inclusive'); }
+ const up = await mp.find(req.query.uploadId, zone, key);
+ if (!up) { req.resume(); return s3error(req, res, 'NoSuchUpload', 'The specified upload does not exist. The upload ID may be invalid, or the upload may have been aborted or completed.'); }
+ const declared = Number(sigv4.isStreaming(parsed) ? req.headers['x-amz-decoded-content-length'] : req.headers['content-length']);
+ const cap = (config.objects.multipartMaxPartMb || 256) * 1024 * 1024;
+ if (!Number.isFinite(declared)) { req.resume(); return s3error(req, res, 'MissingContentLength', 'You must provide the Content-Length HTTP header'); }
+ if (declared > cap) { res.set('Connection', 'close'); req.resume(); return s3error(req, res, 'EntityTooLarge', `A part is at most ${cap} bytes`); }
+ const total = await mp.bytesSoFar(up, n) + declared;
+ if (total > (config.objects.multipartMaxMb || 20480) * 1024 * 1024) { res.set('Connection', 'close'); req.resume(); return s3error(req, res, 'EntityTooLarge', 'The object would be larger than this surface accepts'); }
+ const pre = await namespaces.checkQuota(req.appRow, zone.namespace, { bytes: total, objects: 0 });
+ if (pre) { res.set('Connection', 'close'); req.resume(); return s3error(req, res, 'QuotaExceeded', pre.detail); }
+ const tmpDir = path.join(config.objects.path, '.tmp');
+ fs.mkdirSync(tmpDir, { recursive: true });
+ const tmp = path.join(tmpDir, `s3p-${crypto.randomBytes(8).toString('hex')}`);
+ const drop = () => { try { fs.unlinkSync(tmp); } catch { /* not written */ } };
+ const got = await receive(req, tmp, Math.max(declared, 0), parsed);
+ if (got.error) { drop(); return s3error(req, res, got.error.code, got.error.message); }
+ if (got.bytes !== declared) { drop(); return s3error(req, res, 'IncompleteBody', `You did not provide the number of bytes specified by the Content-Length HTTP header (${declared}; ${got.bytes} received)`); }
+ const bad = integrityProblem(req, parsed, got);
+ if (bad) { drop(); return s3error(req, res, bad[0], bad[1]); }
+ if (!await mp.find(up.id, zone, key)) { drop(); return s3error(req, res, 'NoSuchUpload', 'The upload was aborted or completed while this part arrived'); }
+ const part = await mp.storePart(up, n, tmp, { bytes: got.bytes, md5: got.md5.toString('hex') });
+ baseHeaders(res);
+ res.set('ETag', `"${part.md5}"`);
+ for (const k of ['x-amz-checksum-crc32', 'x-amz-checksum-sha256']) if (req.headers[k] || got.trailers[k]) res.set(k, req.headers[k] || got.trailers[k]);
+ res.status(200).end();
+}
+
+async function completeMultipart(req, res, zone, key) {
+ if (!mayWrite(req.s3.key)) { req.resume(); return s3error(req, res, 'AccessDenied', 'This access key is read-only'); }
+ const up = await mp.find(req.query.uploadId, zone, key);
+ if (!up) { req.resume(); return s3error(req, res, 'NoSuchUpload', 'The specified upload does not exist. The upload ID may be invalid, or the upload may have been aborted or completed.'); }
+ const body = await readXmlBody(req).catch((err) => ({ error: [err.code && STATUS[err.code] ? err.code : 'IncompleteBody', err.message] }));
+ if (body.error) return s3error(req, res, body.error[0], body.error[1]);
+ const items = body.text.match(/<Part>[\s\S]*?<\/Part>/g) || [];
+ if (!/<CompleteMultipartUpload[\s>]/.test(body.text) || !items.length || items.length > 10000) return s3error(req, res, 'MalformedXML', 'The XML you provided was not well-formed or did not validate');
+ const wanted = items.map((x) => ({ n: Number((/<PartNumber>\s*(\d+)\s*<\/PartNumber>/.exec(x) || [])[1]), etag: ((/<ETag>([\s\S]*?)<\/ETag>/.exec(x) || [])[1] || '').replace(/"/g, '"') }));
+ if (wanted.some((w) => !Number.isInteger(w.n))) return s3error(req, res, 'MalformedXML', 'Every part names its PartNumber');
+ const joined = await mp.assemble(up, wanted);
+ if (joined.error) return s3error(req, res, joined.error[0], joined.error[1]);
+ const kept = up.metadata && typeof up.metadata === 'object' ? up.metadata : model.parseJson(up.metadata, {}); // jsonb
+ const md = { ...kept, etag: joined.etag };
+ const outcome = await commitObject(req, zone, key, { tmp: joined.tmp, bytes: joined.bytes, sha256: joined.sha256, mime: up.content_type, md });
+ if (outcome.error) { try { fs.unlinkSync(joined.tmp); } catch { /* moved or gone */ } return s3error(req, res, outcome.error[0], outcome.error[1]); }
+ await mp.abort(up);
+ const host = String(req.headers.host || '');
+ xml(res, 200, `<CompleteMultipartUploadResult xmlns="${XMLNS}"><Location>https://${esc(host)}/${esc(zone.name)}/${esc(sigv4.awsEncode(key, true))}</Location>`
+ + `<Bucket>${esc(zone.name)}</Bucket><Key>${esc(key)}</Key><ETag>${esc(etagOf(outcome.obj))}</ETag></CompleteMultipartUploadResult>`);
+}
+
+async function abortMultipart(req, res, zone, key) {
+ req.resume();
+ if (!mayWrite(req.s3.key)) return s3error(req, res, 'AccessDenied', 'This access key is read-only');
+ const up = await mp.find(req.query.uploadId, zone, key);
+ if (!up) return s3error(req, res, 'NoSuchUpload', 'The specified upload does not exist. The upload ID may be invalid, or the upload may have been aborted or completed.');
+ await mp.abort(up);
+ baseHeaders(res);
+ res.status(204).end();
+}
+
+async function listParts(req, res, zone, key) {
+ if (!mayRead(req.s3.key)) return s3error(req, res, 'AccessDenied', 'This access key cannot read');
+ const up = await mp.find(req.query.uploadId, zone, key);
+ if (!up) return s3error(req, res, 'NoSuchUpload', 'The specified upload does not exist. The upload ID may be invalid, or the upload may have been aborted or completed.');
+ const after = Number(req.query['part-number-marker']) || 0;
+ const max = Math.min(1000, Math.max(1, Number(req.query['max-parts']) || 1000));
+ const all = (await mp.parts(up)).filter((p) => p.part_number > after);
+ const page = all.slice(0, max);
+ const items = page.map((p) => `<Part><PartNumber>${p.part_number}</PartNumber><LastModified>${new Date(Number(p.created_at)).toISOString()}</LastModified>`
+ + `<ETag>"${p.md5}"</ETag><Size>${Number(p.size_bytes)}</Size></Part>`).join('');
+ const truncated = all.length > max;
+ xml(res, 200, `<ListPartsResult xmlns="${XMLNS}"><Bucket>${esc(zone.name)}</Bucket><Key>${esc(key)}</Key><UploadId>${up.id}</UploadId>`
+ + `<PartNumberMarker>${after}</PartNumberMarker>${truncated ? `<NextPartNumberMarker>${page[page.length - 1].part_number}</NextPartNumberMarker>` : ''}`
+ + `<MaxParts>${max}</MaxParts><IsTruncated>${truncated}</IsTruncated><StorageClass>STANDARD</StorageClass>${items}</ListPartsResult>`);
+}
+
+async function listUploads(req, res, zone) {
+ if (!mayRead(req.s3.key)) return s3error(req, res, 'AccessDenied', 'This access key cannot read');
+ const prefix = typeof req.query.prefix === 'string' ? req.query.prefix : '';
+ const rows = await mp.listForZone(zone, { prefix });
+ const items = rows.map((u) => `<Upload><Key>${esc(u.object_key)}</Key><UploadId>${u.id}</UploadId><StorageClass>STANDARD</StorageClass>`
+ + `<Initiated>${new Date(Number(u.created_at)).toISOString()}</Initiated></Upload>`).join('');
+ xml(res, 200, `<ListMultipartUploadsResult xmlns="${XMLNS}"><Bucket>${esc(zone.name)}</Bucket><Prefix>${esc(prefix)}</Prefix>`
+ + `<MaxUploads>1000</MaxUploads><IsTruncated>false</IsTruncated>${items}</ListMultipartUploadsResult>`);
+}
+
async function deleteOne(req, zone, key) {
async function deleteOne(req, zone, key) {
const obj = await findObject(zone, req.appId, key);
const obj = await findObject(zone, req.appId, key);
if (!obj) return { ok: true }; // S3 answers 204 for a key that is not there
if (!obj) return { ok: true }; // S3 answers 204 for a key that is not there
@@ -542,7 +686,8 @@ router.use(async (req, res) => {
@@ -542,7 +686,8 @@ router.use(async (req, res) => {
if (req.method === 'HEAD') { if (!mayRead(req.s3.key)) return s3error(req, res, 'AccessDenied', ''); baseHeaders(res); res.set('x-amz-bucket-region', 'us-east-1'); return res.status(200).end(); }
if (req.method === 'HEAD') { if (!mayRead(req.s3.key)) return s3error(req, res, 'AccessDenied', ''); baseHeaders(res); res.set('x-amz-bucket-region', 'us-east-1'); return res.status(200).end(); }
if (req.method === 'GET' && q.location !== undefined) return xml(res, 200, `<LocationConstraint xmlns="${XMLNS}"/>`);
if (req.method === 'GET' && q.location !== undefined) return xml(res, 200, `<LocationConstraint xmlns="${XMLNS}"/>`);
if (req.method === 'POST' && q.delete !== undefined) return await deleteObjects(req, res, zone);
if (req.method === 'POST' && q.delete !== undefined) return await deleteObjects(req, res, zone);
- if (req.method === 'GET' && !['acl', 'policy', 'versioning', 'cors', 'lifecycle', 'tagging', 'uploads', 'versions', 'website', 'encryption', 'logging', 'notification', 'replication', 'object-lock', 'ownershipControls', 'publicAccessBlock', 'accelerate', 'requestPayment'].some((k) => q[k] !== undefined)) {
+ if (req.method === 'GET' && q.uploads !== undefined) return await listUploads(req, res, zone);
+ if (req.method === 'GET' && !['acl', 'policy', 'versioning', 'cors', 'lifecycle', 'tagging', 'versions', 'website', 'encryption', 'logging', 'notification', 'replication', 'object-lock', 'ownershipControls', 'publicAccessBlock', 'accelerate', 'requestPayment'].some((k) => q[k] !== undefined)) {
if (!mayRead(req.s3.key)) return s3error(req, res, 'AccessDenied', 'This access key cannot read');
if (!mayRead(req.s3.key)) return s3error(req, res, 'AccessDenied', 'This access key cannot read');
return await listObjects(req, res, zone, { v2: q['list-type'] === '2' });
return await listObjects(req, res, zone, { v2: q['list-type'] === '2' });
}
}
@@ -553,9 +698,16 @@ router.use(async (req, res) => {
@@ -553,9 +698,16 @@ router.use(async (req, res) => {
if (key == null) { req.resume(); return s3error(req, res, 'KeyTooLongError', `Keys are 1 to ${MAX_KEY_BYTES} bytes`); }
if (key == null) { req.resume(); return s3error(req, res, 'KeyTooLongError', `Keys are 1 to ${MAX_KEY_BYTES} bytes`); }
const zone = await zoneFor(req, res, bucket);
const zone = await zoneFor(req, res, bucket);
if (!zone) { req.resume(); return; }
if (!zone) { req.resume(); return; }
- if (q.uploads !== undefined || q.uploadId !== undefined || q.partNumber !== undefined) {
+ if (req.method === 'POST' && q.uploads !== undefined) return await createMultipart(req, res, zone, key);
+ if (q.uploadId !== undefined) {
+ if (req.method === 'PUT' && q.partNumber !== undefined) return await uploadPart(req, res, zone, key);
+ if (req.method === 'POST') return await completeMultipart(req, res, zone, key);
+ if (req.method === 'DELETE') return await abortMultipart(req, res, zone, key);
+ if (req.method === 'GET') return await listParts(req, res, zone, key);
+ }
+ if (q.partNumber !== undefined) {
req.resume();
req.resume();
- return s3error(req, res, 'NotImplemented', 'Multipart uploads are not supported yet: send objects up to the single-request limit in one PUT');
+ return s3error(req, res, 'NotImplemented', 'Reading one part of a completed object is not supported; use a Range');
}
}
if (['acl', 'tagging', 'retention', 'legal-hold', 'torrent', 'restore', 'attributes', 'versionId'].some((k) => q[k] !== undefined)) {
if (['acl', 'tagging', 'retention', 'legal-hold', 'torrent', 'restore', 'attributes', 'versionId'].some((k) => q[k] !== undefined)) {
req.resume();
req.resume();
@@ -88,7 +88,7 @@ const http = require('http');
@@ -88,7 +88,7 @@ const http = require('http');
lane: 'light', maxAttempts: 3, backoffS: () => 0,
lane: 'light', maxAttempts: 3, backoffS: () => 0,
run: async (job, ctx) => {
run: async (job, ctx) => {
flakyCalls++;
flakyCalls++;
- if (!ctx.checkpoint) { ctx.saveCheckpoint({ step: 1 }); throw new Error('transient'); }
+ if (!ctx.checkpoint) { await ctx.saveCheckpoint({ step: 1 }); throw new Error('transient'); }
return { resumed_from: ctx.checkpoint.step };
return { resumed_from: ctx.checkpoint.step };
},
},
});
});
new file mode 100644
new file mode 100644
@@ -0,0 +1,125 @@
@@ -0,0 +1,125 @@
+'use strict';
+// S3 multipart uploads on Media's S3 surface (server/s3/multipart.js) with the AWS SDK v3: the SDK's Upload helper
+// (parallel parts) round-trips an 11 MiB object; the raw commands follow S3's rules (5 MiB minimum but for the last
+// part, ascending order, matching ETags, a re-sent part replaces the earlier one); the ETag is the MD5 of the parts'
+// MD5s; ListParts and ListMultipartUploads; Abort removes the parts; quotas count every part; uploads expire.
+const assert = require('assert');
+const fs = require('fs');
+const os = require('os');
+const path = require('path');
+const http = require('http');
+const crypto = require('crypto');
+
+(async () => {
+ const tmp = fs.mkdtempSync(path.join(os.tmpdir(), 'ov-media-s3mp-'));
+ const dir = (n) => { const d = path.join(tmp, n); fs.mkdirSync(d, { recursive: true }); return d; };
+ process.env.VOD_PATH = dir('vods');
+ process.env.FILES_PATH = dir('files');
+ process.env.OBJECTS_PATH = dir('objects');
+ process.env.THUMBNAILS_PATH = dir('thumbnails');
+ process.env.OV_NETWORK_URL = 'https://openvibe.network';
+ process.env.MEDIA_PUBLIC_URL = 'https://media.test';
+ process.env.MEDIA_SIGNING_SECRET = 'test-signing-secret';
+ process.env.MEDIA_S3_KEY_SECRET = 'k'.repeat(40); // a test value: 32+ characters
+ process.env.MEDIA_S3_HOSTS = '127.0.0.1';
+ process.env.MEDIA_APP_TENANT_QUOTA_MB = '40';
+
+ const db = require('../server/db/database');
+ const auth = require('../server/auth');
+ const mp = require('../server/s3/multipart');
+ const { serviceAuth, zones: zoneContracts, ids } = require('openvibe-contracts');
+ const express = require('express');
+ const S3 = require('@aws-sdk/client-s3');
+ const { Upload } = require('@aws-sdk/lib-storage');
+
+ const keys = crypto.generateKeyPairSync('rsa', { modulusLength: 2048, publicKeyEncoding: { type: 'spki', format: 'pem' }, privateKeyEncoding: { type: 'pkcs8', format: 'pem' } });
+ auth._setNetworkPublicKeyForTests(keys.publicKey);
+ const now = Math.floor(Date.now() / 1000);
+ const svc = (sub, cap) => serviceAuth.signServiceToken({ iss: 'https://openvibe.network', sub, actor_type: 'service', aud: ['openvibe.media'], cap, iat: now, exp: now + 600, jti: `tok_${crypto.randomBytes(6).toString('hex')}` }, keys.privateKey);
+ const ADMIN = `usr_${ids.ulid()}`;
+ const PRJ = `prj_${ids.ulid()}`;
+
+ const s3router = require('../server/s3/routes');
+ const admin = require('../server/s3/admin-routes');
+ const app = express();
+ app.use((req, res, next) => (s3router.isS3Host(req) ? s3router(req, res, next) : next()));
+ app.use('/internal/zones', admin.internalRouter());
+ app.use('/api/v1/projects/:project/access-keys', admin.keysRouter({ projects: { isAdmin: async (_p, s) => s === ADMIN } }));
+ const server = http.createServer(app);
+ await new Promise((r) => server.listen(0, '127.0.0.1', r));
+ const port = server.address().port;
+ const internal = async (method, p, token, body) => {
+ const res = await fetch(`http://localhost:${port}${p}`, { method, headers: { authorization: `Bearer ${token}`, 'content-type': 'application/json' }, body: body ? JSON.stringify(body) : undefined });
+ return { status: res.status, body: await res.json().catch(() => null) };
+ };
+ const zone = (name, over = {}) => {
+ const z = { id: `zon_${ids.ulid()}`, project_id: PRJ, env: 'production', name, default: false, access: 'private', state: 'active', quota_bytes: null, revision: 1, created_at: new Date().toISOString(), ...over };
+ return Object.assign(z, zoneContracts.identity(z));
+ };
+ const MiB = 1024 * 1024;
+
+ try {
+ const big = zone('big');
+ const small = zone('small', { quota_bytes: 7 * MiB });
+ for (const z of [big, small]) assert.strictEqual((await internal('PUT', `/internal/zones/${z.id}`, svc('svc:zone', ['media.zone.provision']), z)).status, 200);
+ const issued = (await internal('POST', `/api/v1/projects/${PRJ}/access-keys`, svc('svc:services', ['media.access_key.manage']), { env: 'production', name: 'mp', on_behalf_of: ADMIN })).body;
+ const s3 = new S3.S3Client({ endpoint: `http://127.0.0.1:${port}`, region: 'us-east-1', forcePathStyle: true, credentials: { accessKeyId: issued.key.id, secretAccessKey: issued.secret } });
+
+ // ── The SDK's Upload helper: 11 MiB in parts of 5 MiB, sent in parallel ──
+ const body = crypto.randomBytes(11 * MiB);
+ const done = await new Upload({ client: s3, params: { Bucket: 'big', Key: 'videos/clip.bin', Body: body, ContentType: 'video/mp4', Metadata: { origin: 'camera' } }, partSize: 5 * MiB, queueSize: 3 }).done();
+ const partMd5s = [0, 1, 2].map((i) => crypto.createHash('md5').update(body.subarray(i * 5 * MiB, Math.min((i + 1) * 5 * MiB, body.length))).digest());
+ const want = `"${crypto.createHash('md5').update(Buffer.concat(partMd5s)).digest('hex')}-3"`;
+ assert.strictEqual(done.ETag, want, 'the multipart ETag is the MD5 of the parts\' MD5s, then -3');
+ let out = await s3.send(new S3.GetObjectCommand({ Bucket: 'big', Key: 'videos/clip.bin' }));
+ assert.ok(Buffer.from(await out.Body.transformToByteArray()).equals(body), 'the joined object is the bytes sent');
+ assert.deepStrictEqual([out.ContentType, out.Metadata.origin, out.ETag], ['video/mp4', 'camera', want]);
+ const row = await db.get("SELECT * FROM media_objects WHERE zone_id = ? AND object_key = 'videos/clip.bin' AND lifecycle_status = 'ready'", [big.id]);
+ assert.strictEqual(row.content_hash, crypto.createHash('sha256').update(body).digest('hex'), 'the object\'s sha256 is the whole object\'s');
+ assert.strictEqual((await db.get('SELECT COUNT(*) AS n FROM media_s3_uploads')).n, 0, 'a completed upload leaves nothing behind');
+ console.log('✅ the SDK\'s Upload helper round-trips 11 MiB in three parallel parts; S3\'s multipart ETag');
+
+ // ── The raw commands and S3's rules ──
+ const create = await s3.send(new S3.CreateMultipartUploadCommand({ Bucket: 'big', Key: 'raw.bin' }));
+ const UploadId = create.UploadId;
+ assert.ok(/^mpu_/.test(UploadId));
+ const p1 = Buffer.alloc(5 * MiB, 1), p2a = Buffer.alloc(1024, 2), p2b = Buffer.alloc(2048, 3), tiny = Buffer.alloc(10, 4);
+ const e1 = (await s3.send(new S3.UploadPartCommand({ Bucket: 'big', Key: 'raw.bin', UploadId, PartNumber: 1, Body: p1 }))).ETag;
+ await s3.send(new S3.UploadPartCommand({ Bucket: 'big', Key: 'raw.bin', UploadId, PartNumber: 2, Body: p2a }));
+ const e2 = (await s3.send(new S3.UploadPartCommand({ Bucket: 'big', Key: 'raw.bin', UploadId, PartNumber: 2, Body: p2b }))).ETag;
+ const eT = (await s3.send(new S3.UploadPartCommand({ Bucket: 'big', Key: 'raw.bin', UploadId, PartNumber: 3, Body: tiny }))).ETag;
+ const listed = await s3.send(new S3.ListPartsCommand({ Bucket: 'big', Key: 'raw.bin', UploadId }));
+ assert.deepStrictEqual(listed.Parts.map((p) => [p.PartNumber, p.Size]), [[1, 5 * MiB], [2, 2048], [3, 10]], 'a re-sent part replaces the earlier one');
+ const uploads = await s3.send(new S3.ListMultipartUploadsCommand({ Bucket: 'big' }));
+ assert.deepStrictEqual(uploads.Uploads.map((u) => u.UploadId), [UploadId]);
+ await assert.rejects(s3.send(new S3.CompleteMultipartUploadCommand({ Bucket: 'big', Key: 'raw.bin', UploadId, MultipartUpload: { Parts: [{ PartNumber: 2, ETag: e2 }, { PartNumber: 1, ETag: e1 }] } })), /InvalidPartOrder/);
+ await assert.rejects(s3.send(new S3.CompleteMultipartUploadCommand({ Bucket: 'big', Key: 'raw.bin', UploadId, MultipartUpload: { Parts: [{ PartNumber: 1, ETag: e1 }, { PartNumber: 2, ETag: '"00000000000000000000000000000000"' }] } })), /InvalidPart/);
+ await assert.rejects(s3.send(new S3.CompleteMultipartUploadCommand({ Bucket: 'big', Key: 'raw.bin', UploadId, MultipartUpload: { Parts: [{ PartNumber: 1, ETag: e1 }, { PartNumber: 2, ETag: e2 }, { PartNumber: 3, ETag: eT }] } })), /EntityTooSmall/, 'every part but the last is at least 5 MiB');
+ out = await s3.send(new S3.CompleteMultipartUploadCommand({ Bucket: 'big', Key: 'raw.bin', UploadId, MultipartUpload: { Parts: [{ PartNumber: 1, ETag: e1 }, { PartNumber: 2, ETag: e2 }] } }));
+ assert.ok(out.ETag.endsWith('-2"'));
+ out = await s3.send(new S3.GetObjectCommand({ Bucket: 'big', Key: 'raw.bin' }));
+ assert.ok(Buffer.from(await out.Body.transformToByteArray()).equals(Buffer.concat([p1, p2b])), 'only the named parts, in order');
+ await assert.rejects(s3.send(new S3.UploadPartCommand({ Bucket: 'big', Key: 'raw.bin', UploadId, PartNumber: 4, Body: tiny })), /NoSuchUpload/, 'a completed upload takes no more parts');
+ console.log('✅ CreateMultipartUpload, UploadPart (replace), ListParts, ListMultipartUploads, Complete with S3\'s rules');
+
+ // ── Abort, quota, other keys and zones, expiry ──
+ const ab = await s3.send(new S3.CreateMultipartUploadCommand({ Bucket: 'big', Key: 'gone.bin' }));
+ await s3.send(new S3.UploadPartCommand({ Bucket: 'big', Key: 'gone.bin', UploadId: ab.UploadId, PartNumber: 1, Body: tiny }));
+ await s3.send(new S3.AbortMultipartUploadCommand({ Bucket: 'big', Key: 'gone.bin', UploadId: ab.UploadId }));
+ assert.ok(!fs.existsSync(path.join(process.env.OBJECTS_PATH, '.tmp', 's3mp', ab.UploadId)), 'abort removes the parts');
+ await assert.rejects(s3.send(new S3.ListPartsCommand({ Bucket: 'big', Key: 'gone.bin', UploadId: ab.UploadId })), /NoSuchUpload/);
+ await assert.rejects(s3.send(new S3.ListPartsCommand({ Bucket: 'small', Key: 'raw.bin', UploadId })), /NoSuchUpload/, 'an upload belongs to its zone and key');
+ const q = await s3.send(new S3.CreateMultipartUploadCommand({ Bucket: 'small', Key: 'q.bin' }));
+ await s3.send(new S3.UploadPartCommand({ Bucket: 'small', Key: 'q.bin', UploadId: q.UploadId, PartNumber: 1, Body: Buffer.alloc(5 * MiB) }));
+ await assert.rejects(s3.send(new S3.UploadPartCommand({ Bucket: 'small', Key: 'q.bin', UploadId: q.UploadId, PartNumber: 2, Body: Buffer.alloc(3 * MiB) })), /QuotaExceeded/, 'parts count toward the zone\'s quota as they arrive');
+ await db.run('UPDATE media_s3_uploads SET expires_at = 1 WHERE id = ?', [q.UploadId]);
+ assert.strictEqual(await mp.sweep(), 1, 'an expired upload is swept');
+ assert.ok(!fs.existsSync(path.join(process.env.OBJECTS_PATH, '.tmp', 's3mp', q.UploadId)));
+ console.log('✅ abort, an upload bound to its zone and key, quota across parts, expiry');
+ } finally {
+ server.close();
+ fs.rmSync(tmp, { recursive: true, force: true });
+ }
+ console.log('s3 multipart: all checks passed');
+ process.exit(0);
+})().catch((err) => { console.error(err); process.exit(1); });
@@ -227,10 +227,10 @@ const { Readable } = require('stream');
@@ -227,10 +227,10 @@ const { Readable } = require('stream');
await model.placeHold({ object_id: current.id, kind: 'evidence', reason: 'test' });
await model.placeHold({ object_id: current.id, kind: 'evidence', reason: 'test' });
await assert.rejects(s3.send(new S3.PutObjectCommand({ Bucket: 'assets', Key: 'readme.txt', Body: 'v3' })), /AccessDenied/, 'an overwrite of a held object is refused');
await assert.rejects(s3.send(new S3.PutObjectCommand({ Bucket: 'assets', Key: 'readme.txt', Body: 'v3' })), /AccessDenied/, 'an overwrite of a held object is refused');
await assert.rejects(s3.send(new S3.DeleteObjectCommand({ Bucket: 'assets', Key: 'readme.txt' })), /AccessDenied/, 'so is its deletion');
await assert.rejects(s3.send(new S3.DeleteObjectCommand({ Bucket: 'assets', Key: 'readme.txt' })), /AccessDenied/, 'so is its deletion');
- await assert.rejects(s3.send(new S3.CreateMultipartUploadCommand({ Bucket: 'assets', Key: 'mp' })), /NotImplemented/);
+ await assert.rejects(s3.send(new S3.UploadPartCopyCommand({ Bucket: 'assets', Key: 'mp', UploadId: (await s3.send(new S3.CreateMultipartUploadCommand({ Bucket: 'assets', Key: 'mp' }))).UploadId, PartNumber: 1, CopySource: 'assets/readme.txt' })), /NotImplemented/);
await assert.rejects(s3.send(new S3.CopyObjectCommand({ Bucket: 'assets', Key: 'copy', CopySource: 'assets/readme.txt' })), /NotImplemented/);
await assert.rejects(s3.send(new S3.CopyObjectCommand({ Bucket: 'assets', Key: 'copy', CopySource: 'assets/readme.txt' })), /NotImplemented/);
await assert.rejects(s3.send(new S3.DeleteBucketCommand({ Bucket: 'assets' })), /NotImplemented|MethodNotAllowed/);
await assert.rejects(s3.send(new S3.DeleteBucketCommand({ Bucket: 'assets' })), /NotImplemented|MethodNotAllowed/);
- console.log('✅ quotas, overwrite through the deletion path, holds refuse overwrite and delete, NotImplemented answers');
+ console.log('✅ quotas, overwrite through the deletion path, holds refuse overwrite and delete, NotImplemented answers (copy, part copy, bucket deletion)');
// ── Deleting objects ──
// ── Deleting objects ──
await s3.send(new S3.DeleteObjectCommand({ Bucket: 'assets', Key: 'site/index.html' }));
await s3.send(new S3.DeleteObjectCommand({ Bucket: 'assets', Key: 'site/index.html' }));