The_Great_Archive/cryptpad
1
0
Fork 0
mirror of https://github.com/cryptpad/cryptpad.git synced 2026-09-14 11:05:41 +05:00
Code Issues Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'origin/blob-metadata' into blobmd

Browse Source
This commit is contained in:
yflory 2024-12-17 14:51:06 +01:00
commit 237dc7680e
4 changed files with 124 additions and 29 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
lib/commands/metadata.js
View File

@ -13,7 +13,8 @@ Data.getMetadataRaw = function (Env, channel /* channelName */, _cb) {
const cb = Util.once(Util.mkAsync(_cb));
if (!Core.isValidId(channel)) { return void cb('INVALID_CHAN'); }
if (channel.length !== HK.STANDARD_CHANNEL_LENGTH &&
channel.length !== HK.ADMIN_CHANNEL_LENGTH) { return cb("INVALID_CHAN_LENGTH"); }
channel.length !== HK.ADMIN_CHANNEL_LENGTH &&
channel.length !== HK.BLOB_ID_LENGTH) { return cb("INVALID_CHAN_LENGTH"); }
// return synthetic metadata for admin broadcast channels as a safety net
// in case anybody manages to write metadata
@ -79,6 +80,7 @@ Data.setMetadata = function (Env, safeKey, data, cb, Server) {
var channel = data.channel;
var command = data.command;
// XXX BLOBMD allow blobs
if (!channel || !Core.isValidId(channel)) { return void cb ('INVALID_CHAN'); }
if (!command || typeof (command) !== 'string') { return void cb('INVALID_COMMAND'); }
if (Meta.commands.indexOf(command) === -1) { return void cb('UNSUPPORTED_COMMAND'); }
@ -140,6 +142,7 @@ Data.setMetadata = function (Env, safeKey, data, cb, Server) {
cb(void 0, metadata);
return void next();
}
// XXX BLOBMD use correct store for blobs
Env.msgStore.writeMetadata(channel, JSON.stringify(line), function (e) {
if (e) {
cb(e);
@ -155,6 +158,7 @@ Data.setMetadata = function (Env, safeKey, data, cb, Server) {
// update the cached metadata
metadata_cache[channel] = metadata;
Env.checkCache(channel); // XXX ???
// it's easy to check if the channel is restricted
const isRestricted = metadata.restricted;

2
lib/hk-util.js
View File

@ -42,6 +42,8 @@ const ADMIN_CHANNEL_LENGTH = HK.ADMIN_CHANNEL_LENGTH = 33;
// with a 34 character id
const EPHEMERAL_CHANNEL_LENGTH = HK.EPHEMERAL_CHANNEL_LENGTH = 34;
const BLOB_ID_LENGTH = HK.BLOB_ID_LENGTH = 48;
// Temporary channels are archived X ms after everyone has left them
const TEMPORARY_CHANNEL_LIFETIME = 30 * 1000;

110
lib/storage/blob.js
View File

@ -10,6 +10,10 @@ var BlobStore = module.exports;
var nThen = require("nthen");
var Semaphore = require("saferphore");
var Util = require("../common-util");
var Meta = require("../metadata");
const PERMISSIVE = 511;
const readFileBin = require("../stream-file").readFileBin;
const BLOB_LENGTH = 48;
@ -31,7 +35,7 @@ var prependArchive = function (Env, path) {
return Path.join(Env.archivePath, 'blob', relativePathToBlob);
};
// /blob/<safeKeyPrefix>/<safeKey>/<blobPrefix>/<blobId>
// /blob/<blobPrefix>/<blobId>
var makeBlobPath = function (Env, blobId) {
return Path.join(Env.blobPath, blobId.slice(0, 2), blobId);
};
@ -40,6 +44,11 @@ var makeActivityPath = function (Env, blobId) {
return makeBlobPath(Env, blobId) + '.activity';
};
// /blob/<blobPrefix>/<blobId>.metadata.ndjson
var mkMetadataPath = function (Env, blobId) {
return Path.join(Env.blobPath, blobId.slice(0, 2), blobId) + '.metadata.ndjson';
};
// /blobstate/<safeKeyPrefix>/<safeKey>
var makeStagePath = function (Env, safeKey) {
return Path.join(Env.blobStagingPath, safeKey.slice(0, 2), safeKey);
@ -190,6 +199,71 @@ var getActivity = function (Env, blobId, cb) {
});
};
// destroyStream && createIdleStreamCollector
// copied from lib/storage/file.js
// see comments there
const STREAM_CLOSE_TIMEOUT = 120000;
const STREAM_DESTROY_TIMEOUT = 30000;
const destroyStream = function (stream) {
if (!stream) { return; }
try {
stream.close();
if (stream.closed && stream.fd === null) { return; }
} catch (err) {
console.error(err);
}
setTimeout(function () {
try { stream.destroy(); } catch (err) { console.error(err); }
}, STREAM_DESTROY_TIMEOUT);
};
const createIdleStreamCollector = function (stream) {
var collector = Util.once(Util.mkAsync(Util.bake(destroyStream, [stream])));
collector.keepAlive = Util.throttle(collector, STREAM_CLOSE_TIMEOUT);
collector.keepAlive();
return collector;
};
// writeMetadata appends to the dedicated log of metadata amendments
var writeMetadata = function (env, channelId, data, cb) {
var path = mkMetadataPath(env, channelId);
Fse.mkdirp(Path.dirname(path), PERMISSIVE, function (err) {
if (err && err.code !== 'EEXIST') { return void cb(err); }
Fs.appendFile(path, data + '\n', cb);
});
};
var archiveMetadata = (Env, blobId, cb) => {
var path = mkMetadataPath(Env, blobId);
var archivePath = prependArchive(Env, path);
// XXX eviction clean lone md files
// if we fail to delete the metadata file, it can still be removed later by the eviction script
Fse.move(path, archivePath, { overwrite: true }, cb);
};
var readBlobMetadata = function (env, blobId, handler, _cb) {
var metadataPath = mkMetadataPath(env, blobId);
var stream = Fs.createReadStream(metadataPath, {start: 0});
const collector = createIdleStreamCollector(stream);
var cb = Util.both(_cb, collector);
readFileBin(stream, function (msgObj, readMore) {
collector.keepAlive();
var line = msgObj.buff.toString('utf8');
try {
var parsed = JSON.parse(line);
handler(null, parsed);
} catch (err) {
handler(err, line);
}
readMore();
}, function (err) {
// ENOENT => there is no metadata log
if (!err || err.code === 'ENOENT') { return void cb(); }
// otherwise stream errors?
cb(err);
});
};
/********** METHODS **************/
var upload = function (Env, safeKey, content, cb) {
@ -313,6 +387,9 @@ var tryId = function (path, cb) {
};
// owned_upload_complete
let unescapeKeyCharacters = function (key) {
return key.replace(/\-/g, '/');
};
var owned_upload_complete = function (Env, safeKey, id, cb) {
closeBlobstage(Env, safeKey);
if (!isValidId(id)) {
@ -325,12 +402,10 @@ var owned_upload_complete = function (Env, safeKey, id, cb) {
}
var finalPath = makeBlobPath(Env, id);
let unsafeKey = unescapeKeyCharacters(safeKey);
//var finalOwnPath = makeProofPath(Env, safeKey, id);
var finalOwnPath = makeProofPath(Env, safeKey, id);
// the user wants to move it into blob and create a empty file with the same id
// in their own space:
// /blob/safeKeyPrefix/safeKey/blobPrefix/blobID
// the user wants to move it into blob and create a metadata log with an owner
nThen(function (w) {
// make the requisite directory structure using Mkdirp
@ -340,12 +415,6 @@ var owned_upload_complete = function (Env, safeKey, id, cb) {
return void cb(e.code);
}
}));
Fse.mkdirp(Path.dirname(finalOwnPath), w(function (e /*, path */) {
if (e) { // does not throw error if the directory already existed
w.abort();
return void cb(e.code);
}
}));
}).nThen(function (w) {
// make sure the id does not collide with another
tryId(finalPath, w(function (e) {
@ -355,8 +424,11 @@ var owned_upload_complete = function (Env, safeKey, id, cb) {
}
}));
}).nThen(function (w) {
// Create the empty file proving ownership
Fs.writeFile(finalOwnPath, '', w(function (e) {
// Write the metadata
let md = JSON.stringify({
owners: [unsafeKey]
});
writeMetadata(Env, id, md, w((e) => {
if (e) {
w.abort();
return void cb(e.code);
@ -411,6 +483,7 @@ var archiveBlob = function (Env, blobId, reason, cb) {
var blobPath = makeBlobPath(Env, blobId);
var archivePath = prependArchive(Env, blobPath);
Fse.move(blobPath, archivePath, { overwrite: true }, cb);
archiveMetadata(Env, blobId, () => {});
archiveActivity(Env, blobId, () => {});
addPlaceholder(Env, blobId, reason, () => {});
};
@ -674,6 +747,15 @@ BlobStore.create = function (config, _cb) {
isOwnedBy(Env, safeKey, blobId, cb);
},
readMetadata: (blobId, handler, cb) => {
if (!isValidId(blobId)) { return void cb("INVALID_ID"); }
readBlobMetadata(Env, blobId, handler, cb);
},
writeMetadata: (blobId, data, cb) => {
if (!isValidId(blobId)) { return void cb("INVALID_ID"); }
writeMetadata(Env, blobId, data, cb);
},
remove: {
blob: function (blobId, _cb) {
var cb = Util.once(Util.mkAsync(_cb));

35
lib/workers/db-worker.js
View File

@ -338,7 +338,13 @@ const computeMetadata = function (data, cb) {
const ref = {};
const lineHandler = Meta.createLineHandler(ref, Env.Log.error);
monitoringIncrement('computeMetadata');
return void store.readChannelMetadata(data.channel, lineHandler, function (err) {
let f = store.readChannelMetadata;
if (data.channel.length === HK.BLOB_ID_LENGTH) {
f = blobStore.readMetadata;
}
return void f(data.channel, lineHandler, function (err) {
if (err) {
// stream errors?
return void cb(err);
@ -558,6 +564,7 @@ const removeOwnedBlob = function (data, cb) {
if (typeof(data.safeKey) !== 'string') { return void cb("INVALID_KEY"); }
const blobId = data.blobId;
const safeKey = Util.escapeKeyCharacters(data.safeKey);
const unsafeKey = Util.unescapeKeyCharacters(data.safeKey);
const reason = data.reason || 'ARCHIVE_OWNED';
@ -569,6 +576,18 @@ const removeOwnedBlob = function (data, cb) {
return void cb("INSUFFICIENT_PERMISSIONS");
}
}));
computeMetadata({channel: blobId}, w((err, meta) => {
if (err || !meta) {
w.abort();
return void cb("INSUFFICIENT_PERMISSIONS");
}
let owners = meta.owners;
if (!owners || !owners.includes(unsafeKey)) {
w.abort();
return void cb("INSUFFICIENT_PERMISSIONS");
}
// Owned, continue
}));
}).nThen(function (w) {
// remove the blob
blobStore.archive.blob(blobId, reason, w(function (err) {
@ -581,20 +600,8 @@ const removeOwnedBlob = function (data, cb) {
w.abort();
return void cb(err);
}
}));
}).nThen(function () {
// archive the proof
blobStore.archive.proof(safeKey, blobId, function (err) {
Env.Log.info("ARCHIVAL_PROOF_REMOVAL_BY_OWNER_RPC", {
safeKey: safeKey,
blobId: blobId,
status: err? String(err): 'SUCCESS',
});
if (err) {
return void cb("E_PROOF_REMOVAL");
}
cb(void 0, 'OK');
});
}));
});
};