cryptpad/lib/commands/admin-rpc.js

/*jshint esversion: 6 */
/* globals process */
const nThen = require("nthen");
const getFolderSize = require("get-folder-size");
const Util = require("../common-util");
const Ulimit = require("ulimit");
const Decrees = require("../decrees");
const Pinning = require("./pin-rpc");

var Fs = require("fs");

var Admin = module.exports;

var getFileDescriptorCount = function (Env, server, cb) {
    Fs.readdir('/proc/self/fd', function(err, list) {
        if (err) { return void cb(err); }
        cb(void 0, list.length);
    });
};

var getFileDescriptorLimit = function (env, server, cb) {
    Ulimit(cb);
};

var getCacheStats = function (env, server, cb) {
    var metaSize = 0;
    var channelSize = 0;
    var metaCount = 0;
    var channelCount = 0;

    try {
        var meta = env.metadata_cache;
        for (var x in meta) {
            if (meta.hasOwnProperty(x)) {
                metaCount++;
                metaSize += JSON.stringify(meta[x]).length;
            }
        }

        var channels = env.channel_cache;
        for (var y in channels) {
            if (channels.hasOwnProperty(y)) {
                channelCount++;
                channelSize += JSON.stringify(channels[y]).length;
            }
        }
    } catch (err) {
        return void cb(err && err.message);
    }

    cb(void 0, {
        metadata: metaCount,
        metaSize: metaSize,
        channel: channelCount,
        channelSize: channelSize,
        memoryUsage: process.memoryUsage(),
    });
};

// CryptPad_AsyncStore.rpc.send('ADMIN', ['GET_WORKER_PROFILES'], console.log)
var getWorkerProfiles = function (Env, Server, cb) {
    cb(void 0, Env.commandTimers);
};

var getActiveSessions = function (Env, Server, cb) {
    var stats = Server.getSessionStats();
    cb(void 0, [
        stats.total,
        stats.unique
    ]);
};

var shutdown = function (Env, Server, cb) {
    if (true) {
        return void cb('E_NOT_IMPLEMENTED');
    }

    // disconnect all users and reject new connections
    Server.shutdown();

    // stop all intervals that may be running
    Object.keys(Env.intervals).forEach(function (name) {
        clearInterval(Env.intervals[name]);
    });

    // set a flag to prevent incoming database writes
    // wait until all pending writes are complete
    // then process.exit(0);
    // and allow system functionality to restart the server
};

var getRegisteredUsers = function (Env, Server, cb) {
    Env.batchRegisteredUsers('', cb, function (done) {
        var dir = Env.paths.pin;
        var folders;
        var users = 0;
        nThen(function (waitFor) {
            Fs.readdir(dir, waitFor(function (err, list) {
                if (err) {
                    waitFor.abort();
                    return void done(err);
                }
                folders = list;
            }));
        }).nThen(function (waitFor) {
            folders.forEach(function (f) {
                var dir = Env.paths.pin + '/' + f;
                Fs.readdir(dir, waitFor(function (err, list) {
                    if (err) { return; }
                    users += list.length;
                }));
            });
        }).nThen(function () {
            done(void 0, users);
        });
    });
};

var getDiskUsage = function (Env, Server, cb) {
    Env.batchDiskUsage('', cb, function (done) {
        var data = {};
        nThen(function (waitFor) {
            getFolderSize('./', waitFor(function(err, info) {
                data.total = info;
            }));
            getFolderSize(Env.paths.pin, waitFor(function(err, info) {
                data.pin = info;
            }));
            getFolderSize(Env.paths.blob, waitFor(function(err, info) {
                data.blob = info;
            }));
            getFolderSize(Env.paths.staging, waitFor(function(err, info) {
                data.blobstage = info;
            }));
            getFolderSize(Env.paths.block, waitFor(function(err, info) {
                data.block = info;
            }));
            getFolderSize(Env.paths.data, waitFor(function(err, info) {
                data.datastore = info;
            }));
        }).nThen(function () {
            done(void 0, data);
        });
    });
};

var getActiveChannelCount = function (Env, Server, cb) {
    cb(void 0, Server.getActiveChannelCount());
};

var flushCache = function (Env, Server,  cb) {
    Env.flushCache();
    cb(void 0, true);
};

// CryptPad_AsyncStore.rpc.send('ADMIN', ['ARCHIVE_DOCUMENT', documentID], console.log)
var archiveDocument = function (Env, Server, cb, data) {
    if (!Array.isArray(data)) { return void cb("EINVAL"); }
    var args = data[1];

    var id, reason;
    if (typeof(args) === 'string') {
        id = args;
    } else if (args && typeof(args) === 'object') {
        id = args.id;
        reason = args.reason;
    }

    if (typeof(id) !== 'string' || id.length < 32) { return void cb("EINVAL"); }

    switch (id.length) {
        case 32:
        // TODO disconnect users from active sessions
            return void Env.msgStore.archiveChannel(id, Util.both(cb, function (err) {
                Env.Log.info("ARCHIVAL_CHANNEL_BY_ADMIN_RPC", {
                    channelId: id,
                    reason: reason,
                    status: err? String(err): "SUCCESS",
                });
            }));
        case 48:
            return void Env.blobStore.archive.blob(id, Util.both(cb, function (err) {
                Env.Log.info("ARCHIVAL_BLOB_BY_ADMIN_RPC", {
                    id: id,
                    reason: reason,
                    status: err? String(err): "SUCCESS",
                });
            }));
        default:
            return void cb("INVALID_ID_LENGTH");
    }

    // archival for blob proofs isn't automated, but evict-inactive.js will
    // clean up orpaned blob proofs
    // Env.blobStore.archive.proof(userSafeKey, blobId, cb)
};

var restoreArchivedDocument = function (Env, Server, cb, data) {
    if (!Array.isArray(data)) { return void cb("EINVAL"); }
    var args = data[1];

    var id, reason;
    if (typeof(args) === 'string') {
        id = args;
    } else if (args && typeof(args) === 'object') {
        id = args.id;
        reason = args.reason;
    }

    if (typeof(id) !== 'string' || id.length < 32) { return void cb("EINVAL"); }

    switch (id.length) {
        case 32:
            return void Env.msgStore.restoreArchivedChannel(id, Util.both(cb, function (err) {
                Env.Log.info("RESTORATION_CHANNEL_BY_ADMIN_RPC", {
                    id: id,
                    reason: reason,
                    status: err? String(err): 'SUCCESS',
                });
            }));
        case 48:
            // FIXME this does not yet restore blob ownership
            // Env.blobStore.restore.proof(userSafekey, id, cb)
            return void Env.blobStore.restore.blob(id, Util.both(cb, function (err) {
                Env.Log.info("RESTORATION_BLOB_BY_ADMIN_RPC", {
                    id: id,
                    reason: reason,
                    status: err? String(err): 'SUCCESS',
                });
            }));
        default:
            return void cb("INVALID_ID_LENGTH");
    }
};

// CryptPad_AsyncStore.rpc.send('ADMIN', ['CLEAR_CACHED_CHANNEL_INDEX', documentID], console.log)
var clearChannelIndex = function (Env, Server, cb, data) {
    var id = Array.isArray(data) && data[1];
    if (typeof(id) !== 'string' || id.length < 32) { return void cb("EINVAL"); }
    delete Env.channel_cache[id];
    cb();
};

// CryptPad_AsyncStore.rpc.send('ADMIN', ['GET_CACHED_CHANNEL_INDEX', documentID], console.log)
var getChannelIndex = function (Env, Server, cb, data) {
    var id = Array.isArray(data) && data[1];
    if (typeof(id) !== 'string' || id.length < 32) { return void cb("EINVAL"); }

    var index = Util.find(Env, ['channel_cache', id]);
    if (!index) { return void cb("ENOENT"); }
    cb(void 0, index);
};

// CryptPad_AsyncStore.rpc.send('ADMIN', ['CLEAR_CACHED_CHANNEL_METADATA', documentID], console.log)
var clearChannelMetadata = function (Env, Server, cb, data) {
    var id = Array.isArray(data) && data[1];
    if (typeof(id) !== 'string' || id.length < 32) { return void cb("EINVAL"); }
    delete Env.metadata_cache[id];
    cb();
};

// CryptPad_AsyncStore.rpc.send('ADMIN', ['GET_CACHED_CHANNEL_METADATA', documentID], console.log)
var getChannelMetadata = function (Env, Server, cb, data) {
    var id = Array.isArray(data) && data[1];
    if (typeof(id) !== 'string' || id.length < 32) { return void cb("EINVAL"); }

    var index = Util.find(Env, ['metadata_cache', id]);
    if (!index) { return void cb("ENOENT"); }
    cb(void 0, index);
};

// CryptPad_AsyncStore.rpc.send('ADMIN', [ 'ADMIN_DECREE', ['RESTRICT_REGISTRATION', [true]]], console.log)
var adminDecree = function (Env, Server, cb, data, unsafeKey) {
    var value = data[1];
    if (!Array.isArray(value)) { return void cb('INVALID_DECREE'); }

    var command = value[0];
    var args = value[1];

/*

The admin should have sent a command to be run:

the server adds two pieces of information to the supplied decree:

* the unsafeKey of the admin who uploaded it
* the current time

1. test the command to see if it's valid and will result in a change
2. if so, apply it and write it to the log for persistence
3. respond to the admin with an error or nothing

*/

    var decree = [command, args, unsafeKey, +new Date()];
    var changed;
    try {
        changed = Decrees.handleCommand(Env, decree) || false;
    } catch (err) {
        return void cb(err);
    }

    if (!changed) { return void cb(); }
    Env.Log.info('ADMIN_DECREE', decree);
    Decrees.write(Env, decree, cb);
};

// CryptPad_AsyncStore.rpc.send('ADMIN', ['SET_LAST_EVICTION', 0], console.log)
var setLastEviction = function (Env, Server, cb, data, unsafeKey) {
    var time = data && data[1];
    if (typeof(time) !== 'number') {
        return void cb('INVALID_ARGS');
    }

    Env.lastEviction = time;
    cb();
    Env.Log.info('LAST_EVICTION_TIME_SET', {
        author: unsafeKey,
        time: time,
    });
};

// CryptPad_AsyncStore.rpc.send('ADMIN', ['INSTANCE_STATUS], console.log)
var instanceStatus = function (Env, Server, cb) {
    cb(void 0, {
        restrictRegistration: Env.restrictRegistration,
        launchTime: Env.launchTime,
        currentTime: +new Date(),

        inactiveTime: Env.inactiveTime,
        accountRetentionTime: Env.accountRetentionTime,
        archiveRetentionTime: Env.archiveRetentionTime,

        defaultStorageLimit: Env.defaultStorageLimit,

        lastEviction: Env.lastEviction,
        evictionReport: Env.evictionReport,

        disableIntegratedEviction: Env.disableIntegratedEviction,
        disableIntegratedTasks: Env.disableIntegratedTasks,

        enableProfiling: Env.enableProfiling,
        profilingWindow: Env.profilingWindow,

        maxUploadSize: Env.maxUploadSize,
        premiumUploadSize: Env.premiumUploadSize,

        consentToContact: Env.consentToContact,
        listMyInstance: Env.listMyInstance,
        provideAggregateStatistics: Env.provideAggregateStatistics,

        removeDonateButton: Env.removeDonateButton,
        blockDailyCheck: Env.blockDailyCheck,

        updateAvailable: Env.updateAvailable,
        instancePurpose: Env.instancePurpose,

        instanceDescription: Env.instanceDescription,
        instanceJurisdiction: Env.instanceJurisdiction,
        instanceName: Env.instanceName,
    });
};

// CryptPad_AsyncStore.rpc.send('ADMIN', ['GET_LIMITS'], console.log)
var getLimits = function (Env, Server, cb) {
    cb(void 0, Env.limits);
};

// CryptPad_AsyncStore.rpc.send('ADMIN', ['GET_USER_TOTAL_SIZE', "CrufexqXcY/z+eKJlEbNELVy5Sb7E/EAAEFI8GnEtZ0="], console.log)
var getUserTotalSize = function (Env, Server, cb, data) {
    var signingKey = Array.isArray(data) && data[1];
    if (typeof(signingKey) !== 'string' || signingKey.length < 44) { return void cb("EINVAL"); } // FIXME use a standard check for this
    Pinning.getTotalSize(Env, signingKey, cb);
};

var commands = {
    ACTIVE_SESSIONS: getActiveSessions,
    ACTIVE_PADS: getActiveChannelCount,
    REGISTERED_USERS: getRegisteredUsers,
    DISK_USAGE: getDiskUsage,
    FLUSH_CACHE: flushCache,
    SHUTDOWN: shutdown,
    GET_FILE_DESCRIPTOR_COUNT: getFileDescriptorCount,
    GET_FILE_DESCRIPTOR_LIMIT: getFileDescriptorLimit,
    GET_CACHE_STATS: getCacheStats,

    ARCHIVE_DOCUMENT: archiveDocument,
    RESTORE_ARCHIVED_DOCUMENT: restoreArchivedDocument,

    CLEAR_CACHED_CHANNEL_INDEX: clearChannelIndex,
    GET_CACHED_CHANNEL_INDEX: getChannelIndex,
    // TODO implement admin historyTrim
    // TODO implement kick from channel
    // TODO implement force-disconnect user(s)?

    CLEAR_CACHED_CHANNEL_METADATA: clearChannelMetadata,
    GET_CACHED_CHANNEL_METADATA: getChannelMetadata,

    ADMIN_DECREE: adminDecree,
    INSTANCE_STATUS: instanceStatus,
    GET_LIMITS: getLimits,
    SET_LAST_EVICTION: setLastEviction,
    GET_WORKER_PROFILES: getWorkerProfiles,
    GET_USER_TOTAL_SIZE: getUserTotalSize,
};

Admin.command = function (Env, safeKey, data, _cb, Server) {
    var cb = Util.once(Util.mkAsync(_cb));

    var admins = Env.admins;

    var unsafeKey = Util.unescapeKeyCharacters(safeKey);
    if (admins.indexOf(unsafeKey) === -1) {
        return void cb("FORBIDDEN");
    }

    var command = commands[data[0]];

    if (typeof(command) === 'function') {
        return void command(Env, Server, cb, data, unsafeKey);
    }

    return void cb('UNHANDLED_ADMIN_COMMAND');
};