284 lines
18 KiB
JavaScript
284 lines
18 KiB
JavaScript
import { createRequire } from 'node:module';
|
|
import { createHash, randomUUID } from 'node:crypto';
|
|
import { createReadStream, createWriteStream } from 'node:fs';
|
|
import { mkdir, open, readFile, rename, rm } from 'node:fs/promises';
|
|
import { Transform } from 'node:stream';
|
|
import { Readable } from 'node:stream';
|
|
import { pipeline } from 'node:stream/promises';
|
|
import { join } from 'node:path';
|
|
|
|
// A pilot also runs this contract inside the exact, already installed backend
|
|
// image. Its own SDK dependency tree must pass before the host can cut over.
|
|
const requireSdk = createRequire(process.env.FEDEO_STORAGE_SDK_ROOT
|
|
? join(process.env.FEDEO_STORAGE_SDK_ROOT, 'package.json') : import.meta.url);
|
|
const S3 = requireSdk('@aws-sdk/client-s3');
|
|
const { getSignedUrl } = requireSdk('@aws-sdk/s3-request-presigner');
|
|
|
|
// The pilot deliberately has a tested bound; no untested multipart conversion.
|
|
export const MAX_OBJECT_BYTES = 1024 * 1024 * 1024;
|
|
export const MAX_OBJECTS = 200_000;
|
|
export const client = config => new S3.S3Client({ ...config, forcePathStyle: true, maxAttempts: 3 });
|
|
export const call = (s3, operation, input) => s3.send(new S3[`${operation}Command`](input));
|
|
export const canonical = value => JSON.stringify(value, (_, v) => v && typeof v === 'object' && !Array.isArray(v)
|
|
? Object.fromEntries(Object.entries(v).sort(([a], [b]) => a.localeCompare(b))) : v);
|
|
export const digest = value => createHash('sha256').update(canonical(value)).digest('hex');
|
|
const requireThat = (condition, reason) => { if (!condition) throw new Error(reason); };
|
|
|
|
export async function atomicJson(path, value) {
|
|
const temp = `${path}.tmp`;
|
|
const file = await open(temp, 'w', 0o600);
|
|
try { await file.writeFile(JSON.stringify(value, null, 2) + '\n'); await file.sync(); }
|
|
finally { await file.close(); }
|
|
await rename(temp, path);
|
|
const parent = await open(join(path, '..'), 'r');
|
|
try { await parent.sync(); } finally { await parent.close(); }
|
|
}
|
|
|
|
export function validateKeys(objects) {
|
|
requireThat(objects.length <= MAX_OBJECTS, `Pilotgrenze: höchstens ${MAX_OBJECTS} Objekte.`);
|
|
const keys = new Set(objects.map(o => o.Key));
|
|
requireThat(keys.size === objects.length, 'Doppelte Objektschlüssel.');
|
|
for (const { Key: key, Size: size } of objects) {
|
|
requireThat(typeof key === 'string' && key.length > 0 && Buffer.byteLength(key) <= 1024,
|
|
'Ungültiger Objektschlüssel.');
|
|
const segments = key.split('/');
|
|
requireThat(!/[\x00-\x1f\x7f\\]/.test(key) && segments.every(s => s && s !== '.' && s !== '..'),
|
|
`Nicht verlustfrei übertragbarer Schlüssel: ${JSON.stringify(key)}`);
|
|
requireThat(Number.isSafeInteger(size) && size >= 0 && size <= MAX_OBJECT_BYTES,
|
|
`Objekt außerhalb der getesteten Größe (${MAX_OBJECT_BYTES / 1024 / 1024} MiB): ${JSON.stringify(key)}`);
|
|
for (let i = 1; i < segments.length; i++) {
|
|
requireThat(!keys.has(segments.slice(0, i).join('/')), `Datei-/Verzeichniskonflikt: ${JSON.stringify(key)}`);
|
|
}
|
|
}
|
|
}
|
|
|
|
export async function list(s3, Bucket) {
|
|
const result = [];
|
|
let ContinuationToken;
|
|
do {
|
|
const page = await call(s3, 'ListObjectsV2', { Bucket, ContinuationToken });
|
|
result.push(...(page.Contents || []).map(o => ({ Key: o.Key, Size: o.Size, ETag: o.ETag,
|
|
LastModified: o.LastModified?.toISOString() })));
|
|
requireThat(result.length <= MAX_OBJECTS, `Pilotgrenze: höchstens ${MAX_OBJECTS} Objekte.`);
|
|
requireThat(!page.IsTruncated || (page.NextContinuationToken && page.NextContinuationToken !== ContinuationToken),
|
|
'Ungültige S3-Paginierung.');
|
|
ContinuationToken = page.IsTruncated ? page.NextContinuationToken : undefined;
|
|
} while (ContinuationToken);
|
|
return result.sort((a, b) => Buffer.compare(Buffer.from(a.Key), Buffer.from(b.Key)));
|
|
}
|
|
|
|
async function optional(s3, operation, Bucket, absentCodes) {
|
|
try { return await call(s3, operation, { Bucket }); }
|
|
catch (error) { if (absentCodes.includes(error.name)) return {}; throw error; }
|
|
}
|
|
|
|
function privateAcl(acl) {
|
|
return (acl.Grants || []).every(g => g.Grantee?.Type === 'CanonicalUser' &&
|
|
(g.Grantee.ID || '') === (acl.Owner?.ID || '') && g.Permission === 'FULL_CONTROL');
|
|
}
|
|
|
|
export async function inspectBucket(s3, Bucket) {
|
|
const buckets = await call(s3, 'ListBuckets', {});
|
|
requireThat(buckets.Buckets?.length === 1 && buckets.Buckets[0].Name === Bucket,
|
|
'Die Quelle muss genau den FEDEO-Bucket enthalten; weitere Buckets separat klären.');
|
|
const v = await call(s3, 'GetBucketVersioning', { Bucket });
|
|
requireThat(!v.Status, 'Aktivierte oder ausgesetzte Versionierung wird nicht unterstützt.');
|
|
const versions = await call(s3, 'ListObjectVersions', { Bucket, MaxKeys: 1000 });
|
|
// Even for an unversioned bucket MinIO returns current versions with VersionId=null.
|
|
requireThat(!(versions.DeleteMarkers?.length) && (versions.Versions || []).every(x =>
|
|
x.IsLatest && x.VersionId === 'null'), 'Alte Versionen oder Delete-Marker vorhanden.');
|
|
// A versioning flag above is mandatory; still inspect all version pages to detect old history.
|
|
let page = versions;
|
|
while (page.IsTruncated) {
|
|
const next = await call(s3, 'ListObjectVersions', { Bucket, KeyMarker: page.NextKeyMarker,
|
|
VersionIdMarker: page.NextVersionIdMarker });
|
|
requireThat(!(next.DeleteMarkers?.length) && (next.Versions || []).every(x => x.IsLatest && x.VersionId === 'null'),
|
|
'Alte Versionen oder Delete-Marker vorhanden.');
|
|
requireThat(next.NextKeyMarker !== page.NextKeyMarker || !next.IsTruncated, 'Ungültige Versionspaginierung.');
|
|
page = next;
|
|
}
|
|
const lock = await optional(s3, 'GetObjectLockConfiguration', Bucket, ['ObjectLockConfigurationNotFoundError', 'ObjectLockConfigurationNotFound']);
|
|
requireThat(!lock.ObjectLockConfiguration?.ObjectLockEnabled, 'Object Lock wird nicht unterstützt.');
|
|
const encryption = await optional(s3, 'GetBucketEncryption', Bucket, ['ServerSideEncryptionConfigurationNotFoundError']);
|
|
requireThat(!encryption.ServerSideEncryptionConfiguration?.Rules?.length, 'Bucket-Verschlüsselung benötigt einen eigenen Adapter.');
|
|
const policy = await optional(s3, 'GetBucketPolicy', Bucket, ['NoSuchBucketPolicy']);
|
|
requireThat(!policy.Policy, 'Eigene Bucket-Policy benötigt eine gesonderte Übernahme.');
|
|
const lifecycle = await optional(s3, 'GetBucketLifecycleConfiguration', Bucket, ['NoSuchLifecycleConfiguration']);
|
|
requireThat(!lifecycle.Rules?.length, 'Lifecycle-Regeln vorhanden.');
|
|
const replication = await optional(s3, 'GetBucketReplication', Bucket, ['ReplicationConfigurationNotFoundError']);
|
|
requireThat(!replication.ReplicationConfiguration, 'Bucket-Replikation vorhanden.');
|
|
const notification = await call(s3, 'GetBucketNotificationConfiguration', { Bucket });
|
|
requireThat(!['TopicConfigurations', 'QueueConfigurations', 'LambdaFunctionConfigurations'].some(k => notification[k]?.length)
|
|
&& !notification.EventBridgeConfiguration, 'Bucket-Benachrichtigungen vorhanden.');
|
|
const uploads = await call(s3, 'ListMultipartUploads', { Bucket, MaxUploads: 1 });
|
|
requireThat(!uploads.Uploads?.length, 'Unvollständige Multipart-Uploads vorhanden.');
|
|
const tags = await optional(s3, 'GetBucketTagging', Bucket, ['NoSuchTagSet', 'NoSuchTagSetError']);
|
|
requireThat(!tags.TagSet?.length, 'Bucket-Tags vorhanden.');
|
|
const cors = await optional(s3, 'GetBucketCors', Bucket, ['NoSuchCORSConfiguration']);
|
|
requireThat(!cors.CORSRules?.length, 'Eigene CORS-Konfiguration vorhanden.');
|
|
requireThat(privateAcl(await call(s3, 'GetBucketAcl', { Bucket })), 'Bucket-ACL ist nicht privat.');
|
|
}
|
|
|
|
export function metadata(head) {
|
|
requireThat(!head.ServerSideEncryption && !head.SSECustomerAlgorithm && !head.ObjectLockMode &&
|
|
!head.ObjectLockLegalHoldStatus && !head.WebsiteRedirectLocation && (!head.StorageClass || head.StorageClass === 'STANDARD'),
|
|
'Nicht unterstützte Objektattribute gefunden.');
|
|
const result = {};
|
|
for (const key of ['ContentType', 'ContentDisposition', 'ContentEncoding', 'ContentLanguage', 'CacheControl', 'Metadata']) {
|
|
if (head[key] !== undefined) result[key] = head[key];
|
|
}
|
|
result.Metadata ||= {};
|
|
if (head.Expires) result.Expires = head.Expires.toISOString();
|
|
return result;
|
|
}
|
|
|
|
export async function inventory(s3, Bucket) {
|
|
await inspectBucket(s3, Bucket);
|
|
const objects = await list(s3, Bucket);
|
|
validateKeys(objects);
|
|
for (const object of objects) {
|
|
const input = { Bucket, Key: object.Key };
|
|
const head = await call(s3, 'HeadObject', input);
|
|
requireThat(head.ETag === object.ETag && head.ContentLength === object.Size, 'Quelle während Inventarisierung verändert.');
|
|
object.metadata = metadata(head);
|
|
requireThat(!(await call(s3, 'GetObjectTagging', input)).TagSet?.length, `Objekt-Tags vorhanden: ${JSON.stringify(object.Key)}`);
|
|
requireThat(privateAcl(await call(s3, 'GetObjectAcl', input)), 'Objekt-ACL ist nicht privat.');
|
|
}
|
|
return { schemaVersion: 1, bucket: Bucket, count: objects.length,
|
|
bytes: objects.reduce((n, o) => n + o.Size, 0), objects };
|
|
}
|
|
|
|
export async function readHash(s3, Bucket, object, spool) {
|
|
const response = await call(s3, 'GetObject', { Bucket, Key: object.Key, ...(object.ETag ? { IfMatch: object.ETag } : {}) });
|
|
const hash = createHash('sha256');
|
|
let bytes = 0;
|
|
const inspect = new Transform({ transform(chunk, _, next) {
|
|
bytes += chunk.length;
|
|
if (bytes > MAX_OBJECT_BYTES) return next(new Error('Objekt überschreitet Größenlimit.'));
|
|
hash.update(chunk); next(null, chunk);
|
|
}});
|
|
if (spool) await pipeline(response.Body, inspect, createWriteStream(spool, { mode: 0o600 }));
|
|
else for await (const chunk of response.Body) { bytes += chunk.length; hash.update(chunk);
|
|
requireThat(bytes <= MAX_OBJECT_BYTES, 'Objekt überschreitet Größenlimit.'); }
|
|
requireThat(bytes === object.Size, 'Objektgröße während Lesen verändert.');
|
|
return { sha256: hash.digest('hex'), size: bytes, metadata: metadata(response) };
|
|
}
|
|
|
|
export async function transfer(source, target, Bucket, directory, { verifyOnly = false } = {}) {
|
|
await mkdir(directory, { recursive: true, mode: 0o700 });
|
|
const current = await inventory(source, Bucket);
|
|
const manifestPath = join(directory, 'inventory.json');
|
|
let previous;
|
|
try { previous = JSON.parse(await readFile(manifestPath, 'utf8')); }
|
|
catch (error) { if (error.code !== 'ENOENT') throw error; }
|
|
requireThat(!previous || canonical(previous) === canonical(current), 'Quellinventar seit Beginn verändert.');
|
|
const targetObjects = await list(target, Bucket);
|
|
requireThat(previous || targetObjects.length === 0, 'Ziel ist bei erster Kopie nicht leer.');
|
|
const sourceKeys = new Set(current.objects.map(o => o.Key));
|
|
requireThat(targetObjects.every(o => sourceKeys.has(o.Key)), 'Unerklärte zusätzliche Zielobjekte vorhanden.');
|
|
if (!previous) await atomicJson(manifestPath, current);
|
|
const existing = new Map(targetObjects.map(o => [o.Key, o]));
|
|
const journal = await open(join(directory, 'objects.jsonl'), 'a', 0o600);
|
|
const hashes = [];
|
|
try {
|
|
for (const object of current.objects) {
|
|
const spool = join(directory, 'object.spool');
|
|
try {
|
|
const original = await readHash(source, Bucket, object, verifyOnly ? undefined : spool);
|
|
requireThat(canonical(original.metadata) === canonical(object.metadata), 'Quellmetadaten verändert.');
|
|
let copied;
|
|
if (existing.has(object.Key) && existing.get(object.Key).Size === object.Size) {
|
|
copied = await readHash(target, Bucket, { Key: object.Key, Size: object.Size });
|
|
}
|
|
if (canonical(copied) !== canonical(original)) {
|
|
requireThat(!verifyOnly, `Integritätsfehler: ${JSON.stringify(object.Key)}`);
|
|
const meta = { ...object.metadata };
|
|
if (meta.Expires) meta.Expires = new Date(meta.Expires);
|
|
await call(target, 'PutObject', { Bucket, Key: object.Key, ...meta,
|
|
ContentLength: object.Size, ChecksumSHA256: Buffer.from(original.sha256, 'hex').toString('base64'),
|
|
Body: object.Size === 0 ? Buffer.alloc(0) : createReadStream(spool) });
|
|
copied = await readHash(target, Bucket, { Key: object.Key, Size: object.Size });
|
|
requireThat(canonical(copied) === canonical(original), `Integritätsfehler nach Kopie: ${JSON.stringify(object.Key)}`);
|
|
}
|
|
hashes.push({ key: object.Key, ...original });
|
|
await journal.writeFile(JSON.stringify({ key: object.Key, ...original, verifiedAt: new Date().toISOString() }) + '\n');
|
|
await journal.sync();
|
|
process.stdout.write(JSON.stringify({ phase: verifyOnly ? 'verifying' : 'copying', completed: hashes.length, total: current.count }) + '\n');
|
|
} finally { await rm(spool, { force: true }); }
|
|
}
|
|
} finally { await journal.close(); }
|
|
const after = await inventory(source, Bucket);
|
|
requireThat(canonical(after) === canonical(current), 'Quelle während Kopie/Prüfung verändert.');
|
|
const result = await list(target, Bucket);
|
|
requireThat(canonical(result.map(({ Key, Size }) => ({ Key, Size }))) ===
|
|
canonical(current.objects.map(({ Key, Size }) => ({ Key, Size }))), 'Schlüsselmenge oder Größen weichen ab.');
|
|
const report = { schemaVersion: 1, verifiedAt: new Date().toISOString(), count: current.count, bytes: current.bytes,
|
|
inventorySha256: digest(current), objectsSha256: digest(hashes), objects: hashes };
|
|
await atomicJson(join(directory, 'verified.json'), report);
|
|
return report;
|
|
}
|
|
|
|
export async function verifyManifest(target, Bucket, report) {
|
|
const objects = await list(target, Bucket);
|
|
requireThat(canonical(objects.map(o => ({ key: o.Key, size: o.Size }))) ===
|
|
canonical(report.objects.map(o => ({ key: o.key, size: o.size }))), 'Restore: Schlüsselmenge/Größen abweichend.');
|
|
for (const object of report.objects) {
|
|
requireThat(canonical(await readHash(target, Bucket, { Key: object.key, Size: object.size })) ===
|
|
canonical({ sha256: object.sha256, size: object.size, metadata: object.metadata }), 'Restore: Hash/Metadaten abweichend.');
|
|
}
|
|
return { count: report.count, bytes: report.bytes, restored: true };
|
|
}
|
|
|
|
export async function cleanupProbe(s3, Bucket, probePath) {
|
|
let record;
|
|
try { record = JSON.parse(await readFile(probePath, 'utf8')); }
|
|
catch (error) { if (error.code === 'ENOENT') return; throw error; }
|
|
requireThat(/^fedeo-contract-[0-9a-f-]{36}\//.test(record.key), 'Ungültiges Probejournal.');
|
|
await call(s3, 'DeleteObject', { Bucket, Key: record.key });
|
|
await call(s3, 'DeleteObject', { Bucket, Key: record.key + '.stream' });
|
|
}
|
|
|
|
export async function contract(s3, config, Bucket, { foreignBucket, probePath } = {}) {
|
|
if (probePath) await cleanupProbe(s3, Bucket, probePath);
|
|
const Key = `fedeo-contract-${randomUUID()}/Mandant ä + % #/Prüfung.pdf`;
|
|
if (probePath) await atomicJson(probePath, { key: Key });
|
|
const body = Buffer.from('FEDEO S3 Prüfung\n');
|
|
try {
|
|
await call(s3, 'PutObject', { Bucket, Key, Body: body, ContentType: 'application/pdf',
|
|
ContentDisposition: 'inline; filename="test.pdf"', CacheControl: 'private, max-age=0', Metadata: { test: 'fedeo' } });
|
|
const head = await call(s3, 'HeadObject', { Bucket, Key });
|
|
requireThat(head.ContentLength === body.length && head.Metadata.test === 'fedeo', 'HEAD-Vertrag fehlgeschlagen.');
|
|
const got = await call(s3, 'GetObject', { Bucket, Key });
|
|
requireThat(Buffer.from(await got.Body.transformToByteArray()).equals(body), 'GET-Vertrag fehlgeschlagen.');
|
|
// Application exports use streams. Exercise exactly the SDK environment setting
|
|
// applied by the pilot Compose file, including empty streams.
|
|
for (const bytes of [Buffer.alloc(0), Buffer.alloc(128 * 1024, 42)]) {
|
|
await call(s3, 'PutObject', { Bucket, Key: Key + '.stream', ContentLength: bytes.length,
|
|
Body: Readable.from([bytes]), ContentType: 'application/zip' });
|
|
const streamHead = await call(s3, 'HeadObject', { Bucket, Key: Key + '.stream' });
|
|
requireThat(!streamHead.ContentEncoding && streamHead.ContentLength === bytes.length, 'Streaming-Metadaten abweichend.');
|
|
const downloaded = await call(s3, 'GetObject', { Bucket, Key: Key + '.stream' });
|
|
requireThat(Buffer.from(await downloaded.Body.transformToByteArray()).equals(bytes), 'Streaming-Inhalt abweichend.');
|
|
}
|
|
const url = await getSignedUrl(s3, new S3.GetObjectCommand({ Bucket, Key }), { expiresIn: 86400 });
|
|
const signed = await fetch(url);
|
|
requireThat(signed.ok && Buffer.from(await signed.arrayBuffer()).equals(body), 'Presigned-URL fehlgeschlagen.');
|
|
const anonymous = await fetch(url.split('?')[0]);
|
|
await anonymous.arrayBuffer();
|
|
requireThat([401, 403].includes(anonymous.status), 'Anonymer Objektzugriff erlaubt!');
|
|
const wrong = client({ ...config, credentials: { accessKeyId: 'fedeo-invalid', secretAccessKey: 'fedeo-invalid-secret' } });
|
|
try { await denied(wrong, 'GetObject', { Bucket, Key }); } finally { wrong.destroy(); }
|
|
if (foreignBucket) await denied(s3, 'ListObjectsV2', { Bucket: foreignBucket });
|
|
await call(s3, 'DeleteObject', { Bucket, Key });
|
|
try { await call(s3, 'HeadObject', { Bucket, Key }); throw new Error('DELETE nicht wirksam.'); }
|
|
catch (error) { if (error.$metadata?.httpStatusCode !== 404) throw error; }
|
|
} finally { await call(s3, 'DeleteObject', { Bucket, Key }); await call(s3, 'DeleteObject', { Bucket, Key: Key + '.stream' }); }
|
|
return { operations: ['PUT', 'HEAD', 'GET', 'DELETE', 'SigV4', 'presigned-24h', 'anonymous-denied', 'wrong-key-denied'] };
|
|
}
|
|
|
|
async function denied(s3, operation, input) {
|
|
try { const value = await call(s3, operation, input); value.Body?.destroy(); }
|
|
catch (error) { if ([401, 403].includes(error.$metadata?.httpStatusCode)) return; throw error; }
|
|
throw new Error('Nicht autorisierter Zugriff wurde erlaubt.');
|
|
}
|