wassname/talk
1
0
Fork 0
mirror of https://github.com/wassname/talk.git synced 2026-08-17 11:27:52 +08:00
Code Issues Packages Projects Releases Wiki Activity

Replaced node_redis with ioredis

Browse Source
This commit is contained in:
Wyatt Johnson
2017-08-29 17:52:29 -06:00
parent 2c3f5385ae
commit 8f3bfd2bd4
7 changed files with 276 additions and 366 deletions
Download Patch File Download Diff File Expand all files Collapse all files
services/cache.js
+64 -180
View File
@@ -1,6 +1,5 @@
const redis = require('./redis');
const debug = require('debug')('talk:services:cache');
const crypto = require('crypto');
const cache = module.exports = {};
@@ -52,60 +51,6 @@ cache.wrap = async (key, expiry, work, kf = keyfunc) => {
return value;
};
// This is designed to increment a key and add an expiry iff the key already
// exists.
const INCR_SCRIPT = `
if redis.call('GET', KEYS[1]) ~= false then
redis.call('INCR', KEYS[1])
redis.call('EXPIRE', KEYS[1], ARGV[1])
end
`;
// This is designed to decrement a key and add an expiry iff the key already
// exists.
const DECR_SCRIPT = `
if redis.call('GET', KEYS[1]) ~= false then
redis.call('DECR', KEYS[1])
redis.call('EXPIRE', KEYS[1], ARGV[1])
end
`;
// Load the script into redis and track the script hash that we will use to exec
// increments on.
const loadScript = (name, script) => new Promise((resolve, reject) => {
let shasum = crypto.createHash('sha1');
shasum.update(script);
let hash = shasum.digest('hex');
cache.client
.script('EXISTS', hash, (err, [exists]) => {
if (err) {
return reject(err);
}
if (exists) {
debug(`already loaded ${name} as SHA[${hash}], not loading again`);
return resolve(hash);
}
debug(`${name} not loaded as SHA[${hash}], loading`);
cache.client
.script('load', script, (err, hash) => {
if (err) {
return reject(err);
}
debug(`loaded ${name} as SHA[${hash}]`);
resolve(hash);
});
});
});
/**
* Init sets up the scripts used in Redis with the incr/decr commands.
*/
@@ -114,93 +59,77 @@ cache.init = async () => {
// Create the redis instance.
cache.client = redis.createClient();
// Load the INCR_SCRIPT and DECR_SCRIPT into Redis.
let [incrScriptHash, decrScriptHash] = await Promise.all([
loadScript('INCR_SCRIPT', INCR_SCRIPT),
loadScript('DECR_SCRIPT', DECR_SCRIPT)
]);
// This is designed to increment a key and add an expiry iff the key already
// exists.
const INCR_SCRIPT = `
if redis.call('GET', KEYS[1]) ~= false then
redis.call('INCR', KEYS[1])
redis.call('EXPIRE', KEYS[1], ARGV[1])
end
`;
// Set the globally scoped cache hashes.
cache.INCR_SCRIPT_HASH = incrScriptHash;
cache.DECR_SCRIPT_HASH = decrScriptHash;
cache.client.defineCommand('increx', {
numberOfKeys: 1,
lua: INCR_SCRIPT,
});
// This is designed to decrement a key and add an expiry iff the key already
// exists.
const DECR_SCRIPT = `
if redis.call('GET', KEYS[1]) ~= false then
redis.call('DECR', KEYS[1])
redis.call('EXPIRE', KEYS[1], ARGV[1])
end
`;
cache.client.defineCommand('decrex', {
numberOfKeys: 1,
lua: DECR_SCRIPT,
});
};
/**
* This will increment a key in redis and update the expiry iff it already
* exists, otherwise it will do nothing.
*/
cache.incr = (key, expiry, kf = keyfunc) => new Promise((resolve, reject) => {
cache.client
.evalsha(cache.INCR_SCRIPT_HASH, 1, kf(key), expiry, (err) => {
if (err) {
return reject(err);
}
return resolve();
});
});
cache.incr = async (key, expiry, kf = keyfunc) => cache.client.increx(kf(key), expiry);
/**
* This will decrement a key in redis and update the expiry iff it already
* exists, otherwise it will do nothing.
*/
cache.decr = (key, expiry, kf = keyfunc) => new Promise((resolve, reject) => {
cache.client
.evalsha(cache.DECR_SCRIPT_HASH, 1, kf(key), expiry, (err) => {
if (err) {
return reject(err);
}
return resolve();
});
});
cache.decr = async (key, expiry, kf = keyfunc) => cache.client.decrex(kf(key, expiry));
/**
* This will increment many keys in redis and update the expiry iff it already
* exists, otherwise it will do nothing.
*/
cache.incrMany = (keys, expiry, kf = keyfunc) => {
cache.incrMany = async (keys, expiry, kf = keyfunc) => {
let multi = cache.client.multi();
keys.forEach((key) => {
for (const key of keys) {
// Queue up the evalsha command.
multi.evalsha(cache.INCR_SCRIPT_HASH, 1, kf(key), expiry);
});
multi.increx(kf(key), expiry);
}
return new Promise((resolve, reject) => {
multi.exec((err) => {
if (err) {
return reject(err);
}
resolve();
});
});
return multi.exec();
};
/**
* This will decrement many keys in redis and update the expiry iff it already
* exists, otherwise it will do nothing.
*/
cache.decrMany = (keys, expiry, kf = keyfunc) => {
cache.decrMany = async (keys, expiry, kf = keyfunc) => {
let multi = cache.client.multi();
keys.forEach((key) => {
for (const key of keys) {
// Queue up the evalsha command.
multi.evalsha(cache.DECR_SCRIPT_HASH, 1, kf(key), expiry);
});
multi.decrex(kf(key), expiry);
}
return new Promise((resolve, reject) => {
multi.exec((err) => {
if (err) {
return reject(err);
}
resolve();
});
});
return multi.exec();
};
/**
@@ -257,28 +186,12 @@ cache.wrapMany = async (keys, expiry, work, kf = keyfunc) => {
* @param {Mixed} key Either an array of items composing a key or a string
* @return {Promise}
*/
cache.get = (key, kf = keyfunc) => new Promise((resolve, reject) => {
cache.client.get(kf(key), (err, reply) => {
if (err) {
return reject(err);
}
cache.get = async (key, kf = keyfunc) => cache.client.get(kf(key)).then((reply) => {
if (reply !== null) {
if (reply !== null) {
let value;
try {
// Parse the stored cache value from JSON.
value = JSON.parse(reply);
} catch (e) {
return reject(e);
}
return resolve(value);
}
resolve(null);
});
// Parse the stored cache value from JSON.
return JSON.parse(reply);
}
});
/**
@@ -288,31 +201,22 @@ cache.get = (key, kf = keyfunc) => new Promise((resolve, reject) => {
* @param {Function} [kf=keyfunc] optional key function to use to turn the
* provided key into a string for the cache.
*/
cache.getMany = (keys, kf = keyfunc) => new Promise((resolve, reject) => {
cache.client.mget(keys.map(kf), (err, replies) => {
if (err) {
return reject(err);
cache.getMany = async (keys, kf = keyfunc) => cache.client.mget(keys.map(kf)).then((replies) => {
// Parse the replies.
for (let i = 0; i < replies.length; i++) {
let value = null;
if (replies[i] != null) {
// Parse the stored cache value from JSON.
value = JSON.parse(replies[i]);
}
// Parse the replies.
for (let i = 0; i < replies.length; i++) {
let value = null;
replies[i] = value;
}
if (replies[i] != null) {
try {
// Parse the stored cache value from JSON.
value = JSON.parse(replies[i]);
} catch (e) {
return reject(e);
}
}
replies[i] = value;
}
return resolve(replies);
});
return replies;
});
/**
@@ -322,7 +226,7 @@ cache.getMany = (keys, kf = keyfunc) => new Promise((resolve, reject) => {
* @param {Function} [kf=keyfunc] optional key function to use to turn the
* provided key into a string for the cache.
*/
cache.setMany = (keys, values, expiry, kf = keyfunc) => {
cache.setMany = async (keys, values, expiry, kf = keyfunc) => {
let multi = cache.client.multi();
keys.forEach((key, index) => {
@@ -334,15 +238,7 @@ cache.setMany = (keys, values, expiry, kf = keyfunc) => {
multi.set(kf(key), reply, 'EX', expiry);
});
return new Promise((resolve, reject) => {
multi.exec((err) => {
if (err) {
return reject(err);
}
resolve();
});
});
return multi.exec();
};
/**
@@ -350,18 +246,12 @@ cache.setMany = (keys, values, expiry, kf = keyfunc) => {
* @param {Mixed} key Either an array of items composing a key or a string
* @return {Promise}
*/
cache.invalidate = (key, kf = keyfunc) => new Promise((resolve, reject) => {
cache.invalidate = async (key, kf = keyfunc) => {
debug(`invalidate: ${kf(key)}`);
cache.client.del(kf(key), (err) => {
if (err) {
return reject(err);
}
resolve();
});
});
return cache.client.del(kf(key));
};
/**
* This sets a value on the key with the expiry and then resolves once it is
@@ -371,16 +261,10 @@ cache.invalidate = (key, kf = keyfunc) => new Promise((resolve, reject) => {
* @param {Integer} expiry Time in seconds for the cache entry to live for
* @return {Promise}
*/
cache.set = (key, value, expiry, kf = keyfunc) => new Promise((resolve, reject) => {
cache.set = async (key, value, expiry, kf = keyfunc) => {
// Serialize the value as JSON.
let reply = JSON.stringify(value);
cache.client.set(kf(key), reply, 'EX', expiry, (err) => {
if (err) {
return reject(err);
}
return resolve();
});
});
return cache.client.set(kf(key), reply, 'EX', expiry);
};
services/passport.js
+15 -22
View File
@@ -159,40 +159,33 @@ async function ValidateUserLogin(loginProfile, user, done) {
/**
* Revoke the token on the request.
*/
const HandleLogout = (req, res, next) => {
const HandleLogout = async (req, res, next) => {
const {jwt} = req;
const now = new Date();
const expiry = (jwt.exp - now.getTime() / 1000).toFixed(0);
client().set(`jtir[${jwt.jti}]`, now.toISOString(), 'EX', expiry, (err) => {
if (err) {
return next(err);
}
try {
await client().set(`jtir[${jwt.jti}]`, now.toISOString(), 'EX', expiry);
} catch (err) {
return next(err);
}
// Only clear the cookie on logout if enabled.
if (JWT_CLEAR_COOKIE_LOGOUT) {
debug('clearing the login cookie');
res.clearCookie(JWT_SIGNING_COOKIE_NAME);
}
// Only clear the cookie on logout if enabled.
if (JWT_CLEAR_COOKIE_LOGOUT) {
debug('clearing the login cookie');
res.clearCookie(JWT_SIGNING_COOKIE_NAME);
}
res.status(204).end();
});
res.status(204).end();
};
const checkGeneralTokenBlacklist = (jwt) => new Promise((resolve, reject) => {
client().get(`jtir[${jwt.jti}]`, (err, expiry) => {
if (err) {
return reject(err);
}
const checkGeneralTokenBlacklist = (jwt) => client().get(`jtir[${jwt.jti}]`)
.then((expiry) => {
if (expiry != null) {
return reject(new errors.ErrAuthentication('token was revoked'));
throw new errors.ErrAuthentication('token was revoked');
}
return resolve();
});
});
/**
* Check if the given token is already blacklisted, throw an error if it is.
services/redis.js
+4 -11
View File
@@ -1,4 +1,4 @@
const redis = require('redis');
const Redis = require('ioredis');
const debug = require('debug')('talk:services:redis');
const enabled = require('debug').enabled('talk:services:redis');
const {
@@ -14,9 +14,10 @@ const attachMonitors = (client) => {
// Debug events.
if (enabled) {
client.on('ready', () => debug('client ready'));
client.on('connect', () => debug('client connected'));
client.on('ready', () => debug('client ready'));
client.on('reconnecting', () => debug('client connection lost, attempting to reconnect'));
client.on('close', () => debug('client closed the connection'));
client.on('end', () => debug('client ended'));
}
@@ -64,19 +65,11 @@ const connectionOptions = {
};
const createClient = () => {
let client = redis.createClient(connectionOptions);
let client = new Redis(connectionOptions);
// Attach the monitors that will print debug messages to the console.
attachMonitors(client);
client.ping((err) => {
if (err) {
console.error('Can\'t ping the redis server!');
throw err;
}
});
return client;
};
services/users.js
+104 -138
View File
@@ -69,33 +69,25 @@ module.exports = class UsersService {
* Indicating that the account should be flagged as "login recaptcha required"
* where future login attempts must be made with the recaptcha flag.
*/
static recordLoginAttempt(email) {
static async recordLoginAttempt(email) {
const rdskey = `la[${email.toLowerCase().trim()}]`;
return new Promise((resolve, reject) => {
client()
.multi()
.incr(rdskey)
.expire(rdskey, RECAPTCHA_WINDOW_SECONDS)
.exec((err, replies) => {
if (err) {
return reject(err);
}
const replies = await client()
.multi()
.incr(rdskey)
.expire(rdskey, RECAPTCHA_WINDOW_SECONDS)
.exec();
// if this is new or has no expiry
if (replies[0] === 1 || replies[1] === -1) {
// if this is new or has no expiry
if (replies[0] === 1 || replies[1] === -1) {
// then expire it after the timeout
client().expire(rdskey, RECAPTCHA_WINDOW_SECONDS);
}
// then expire it after the timeout
client().expire(rdskey, RECAPTCHA_WINDOW_SECONDS);
}
if (replies[0] >= RECAPTCHA_INCORRECT_TRIGGER) {
return reject(errors.ErrLoginAttemptMaximumExceeded);
}
resolve();
});
});
if (replies[0] >= RECAPTCHA_INCORRECT_TRIGGER) {
throw errors.ErrLoginAttemptMaximumExceeded;
}
}
/**
@@ -104,27 +96,17 @@ module.exports = class UsersService {
*
* errors.ErrLoginAttemptMaximumExceeded
*/
static checkLoginAttempts(email) {
static async checkLoginAttempts(email) {
const rdskey = `la[${email.toLowerCase().trim()}]`;
return new Promise((resolve, reject) => {
client()
.get(rdskey, (err, reply) => {
if (err) {
return reject(err);
}
const attempts = await client().get(rdskey);
if (!attempts) {
return;
}
if (!reply) {
return resolve();
}
if (reply >= RECAPTCHA_INCORRECT_TRIGGER) {
return reject(errors.ErrLoginAttemptMaximumExceeded);
}
resolve();
});
});
if (attempts >= RECAPTCHA_INCORRECT_TRIGGER) {
throw errors.ErrLoginAttemptMaximumExceeded;
}
}
/**
@@ -217,24 +199,15 @@ module.exports = class UsersService {
});
}
static changePassword(id, password) {
return new Promise((resolve, reject) => {
bcrypt.hash(password, SALT_ROUNDS, (err, hashedPassword) => {
if (err) {
return reject(err);
}
static async changePassword(id, password) {
const hashedPassword = await bcrypt.hash(password, SALT_ROUNDS);
resolve(hashedPassword);
});
})
.then((hashedPassword) => {
return UserModel.update({id}, {
$inc: {__v: 1},
$set: {
password: hashedPassword
}
});
});
return UserModel.update({id}, {
$inc: {__v: 1},
$set: {
password: hashedPassword
}
});
}
/**
@@ -301,54 +274,48 @@ module.exports = class UsersService {
* @param {String} username name of the display user
* @param {Function} done callback
*/
static createLocalUser(email, password, username) {
static async createLocalUser(email, password, username) {
if (!email) {
return Promise.reject(errors.ErrMissingEmail);
throw errors.ErrMissingEmail;
}
email = email.toLowerCase().trim();
username = username.trim();
return Promise.all([
await Promise.all([
UsersService.isValidUsername(username),
UsersService.isValidPassword(password)
])
.then(() => { // username is valid
return new Promise((resolve, reject) => {
bcrypt.hash(password, SALT_ROUNDS, (err, hashedPassword) => {
if (err) {
return reject(err);
}
]);
let user = new UserModel({
username,
lowercaseUsername: username.toLowerCase(),
password: hashedPassword,
roles: [],
profiles: [
{
id: email,
provider: 'local'
}
]
});
const hashedPassword = await bcrypt.hash(password, SALT_ROUNDS);
user.save((err) => {
if (err) {
if (err.code === 11000) {
if (err.message.match('Username')) {
return reject(errors.ErrUsernameTaken);
}
return reject(errors.ErrEmailTaken);
}
return reject(err);
}
return resolve(user);
});
});
});
});
let user = new UserModel({
username,
lowercaseUsername: username.toLowerCase(),
password: hashedPassword,
roles: [],
profiles: [
{
id: email,
provider: 'local'
}
]
});
try {
user = await user.save();
} catch (err) {
if (err.code === 11000) {
if (err.message.match('Username')) {
throw errors.ErrUsernameTaken;
}
throw errors.ErrEmailTaken;
}
throw err;
}
return user;
}
/**
@@ -387,14 +354,14 @@ module.exports = class UsersService {
* @param {String} role role to add
* @param {Function} done callback after the operation is complete
*/
static addRoleToUser(id, role) {
static async addRoleToUser(id, role) {
const roles = [];
// Check to see if the user role is in the allowable set of roles.
if (role && USER_ROLES.indexOf(role) === -1) {
// User role is not supported! Error out here.
return Promise.reject(new Error(`role ${role} is not supported`));
throw new Error(`role ${role} is not supported`);
} else if(role) {
roles.push(role);
}
@@ -408,13 +375,13 @@ module.exports = class UsersService {
* @param {String} role role to remove
* @param {Function} done callback after the operation is complete
*/
static removeRoleFromUser(id, role) {
static async removeRoleFromUser(id, role) {
// Check to see if the user role is in the allowable set of roles.
if (USER_ROLES.indexOf(role) === -1) {
// User role is not supported! Error out here.
return Promise.reject(new Error(`role ${role} is not supported`));
throw new Error(`role ${role} is not supported`);
}
return UserModel.update({id}, {
@@ -430,13 +397,13 @@ module.exports = class UsersService {
* @param {String} status status to set
* @param {Function} done callback after the operation is complete
*/
static setStatus(id, status) {
static async setStatus(id, status) {
// Check to see if the user status is in the allowable set of roles.
if (USER_STATUS.indexOf(status) === -1) {
// User status is not supported! Error out here.
return Promise.reject(new Error(`status ${status} is not supported`));
throw new Error(`status ${status} is not supported`);
}
// TODO: current updating status behavior is weird.
@@ -583,53 +550,52 @@ module.exports = class UsersService {
* Creates a JWT from a user email. Only works for local accounts.
* @param {String} email of the local user
*/
static createPasswordResetToken(email, loc) {
static async createPasswordResetToken(email, loc) {
if (!email || typeof email !== 'string') {
return Promise.reject('email is required when creating a JWT for resetting passord');
throw new Error('email is required when creating a JWT for resetting passord');
}
email = email.toLowerCase();
return Promise.all([
const [user, settings] = await Promise.all([
UserModel.findOne({profiles: {$elemMatch: {id: email}}}),
SettingsService.retrieve()
])
.then(([user, settings]) => {
if (!user) {
SettingsService.retrieve(),
]);
// Since we don't want to reveal that the email does/doesn't exist
// just go ahead and resolve the Promise with null and check in the
// endpoint.
return;
}
let redirectDomain;
try {
const {hostname, port} = url.parse(loc);
redirectDomain = hostname;
if (port) {
redirectDomain += `:${port}`;
}
} catch (e) {
return Promise.reject('redirect location is invalid');
}
if (!user) {
if (settings.domains.whitelist.indexOf(redirectDomain) === -1) {
return Promise.reject('redirect location is not on the list of acceptable domains');
}
// Since we don't want to reveal that the email does/doesn't exist
// just go ahead and resolve the Promise with null and check in the
// endpoint.
return;
}
let redirectDomain;
try {
const {hostname, port} = url.parse(loc);
redirectDomain = hostname;
if (port) {
redirectDomain += `:${port}`;
}
} catch (e) {
throw new Error('redirect location is invalid');
}
const payload = {
jti: uuid.v4(),
email,
loc,
userId: user.id,
version: user.__v
};
if (settings.domains.whitelist.indexOf(redirectDomain) === -1) {
throw new Error('redirect location is not on the list of acceptable domains');
}
return JWT_SECRET.sign(payload, {
expiresIn: '1d',
subject: PASSWORD_RESET_JWT_SUBJECT
});
});
const payload = {
jti: uuid.v4(),
email,
loc,
userId: user.id,
version: user.__v
};
return JWT_SECRET.sign(payload, {
expiresIn: '1d',
subject: PASSWORD_RESET_JWT_SUBJECT
});
}
/**
@@ -755,7 +721,7 @@ module.exports = class UsersService {
*/
static async createEmailConfirmToken(userID = null, email, referer = ROOT_URL) {
if (!email || typeof email !== 'string') {
return Promise.reject('email is required when creating a JWT for resetting passord');
throw new Error('email is required when creating a JWT for resetting passord');
}
// Conform the email to lowercase.