Sharkey/src/queue/processors/process-inbox.ts

183 lines
5.5 KiB
TypeScript
Raw Normal View History

2019-03-07 16:07:21 +02:00
import * as Bull from 'bull';
import * as httpSignature from 'http-signature';
2019-03-07 16:07:21 +02:00
import parseAcct from '../../misc/acct/parse';
import User, { IRemoteUser } from '../../models/user';
import perform from '../../remote/activitypub/perform';
import { resolvePerson, updatePerson } from '../../remote/activitypub/models/person';
2018-08-30 14:53:41 +03:00
import { toUnicode } from 'punycode';
import { URL } from 'url';
2019-03-07 16:07:21 +02:00
import { publishApLogStream } from '../../services/stream';
import Logger from '../../services/logger';
import { registerOrFetchInstanceDoc } from '../../services/register-or-fetch-instance-doc';
import Instance from '../../models/instance';
import instanceChart from '../../services/chart/instance';
2018-04-04 17:12:35 +03:00
const logger = new Logger('inbox');
2018-04-06 16:40:06 +03:00
2018-04-04 17:12:35 +03:00
// ユーザーのinboxにアクティビティが届いた時の処理
2019-03-07 16:07:21 +02:00
export default async (job: Bull.Job, done: any): Promise<void> => {
2018-04-04 17:12:35 +03:00
const signature = job.data.signature;
const activity = job.data.activity;
2018-04-06 16:40:06 +03:00
//#region Log
const info = Object.assign({}, activity);
delete info['@context'];
delete info['signature'];
2019-02-05 07:04:40 +02:00
logger.debug(JSON.stringify(info, null, 2));
2018-04-06 16:40:06 +03:00
//#endregion
2018-04-04 17:12:35 +03:00
const keyIdLower = signature.keyId.toLowerCase();
2018-06-18 08:28:43 +03:00
let user: IRemoteUser;
2018-04-04 17:12:35 +03:00
if (keyIdLower.startsWith('acct:')) {
const { username, host } = parseAcct(keyIdLower.slice('acct:'.length));
if (host === null) {
logger.warn(`request was made by local user: @${username}`);
2018-04-04 17:12:35 +03:00
done();
2018-04-06 08:35:17 +03:00
return;
2018-04-04 17:12:35 +03:00
}
2018-08-30 14:53:41 +03:00
// アクティビティ内のホストの検証
try {
ValidateActivity(activity, host);
} catch (e) {
logger.warn(e.message);
2018-08-30 14:53:41 +03:00
done();
return;
}
// ブロックしてたら中断
// TODO: いちいちデータベースにアクセスするのはコスト高そうなのでどっかにキャッシュしておく
const instance = await Instance.findOne({ host: host.toLowerCase() });
if (instance && instance.isBlocked) {
logger.warn(`Blocked request: ${host}`);
done();
return;
}
user = await User.findOne({ usernameLower: username, host: host.toLowerCase() }) as IRemoteUser;
2018-04-04 17:12:35 +03:00
} else {
2018-08-30 14:53:41 +03:00
// アクティビティ内のホストの検証
const host = toUnicode(new URL(signature.keyId).hostname.toLowerCase());
try {
ValidateActivity(activity, host);
} catch (e) {
logger.warn(e.message);
2018-08-30 14:53:41 +03:00
done();
return;
}
// ブロックしてたら中断
// TODO: いちいちデータベースにアクセスするのはコスト高そうなのでどっかにキャッシュしておく
const instance = await Instance.findOne({ host: host.toLowerCase() });
if (instance && instance.isBlocked) {
logger.warn(`Blocked request: ${host}`);
done();
return;
}
2018-04-04 17:12:35 +03:00
user = await User.findOne({
host: { $ne: null },
2018-04-07 21:58:11 +03:00
'publicKey.id': signature.keyId
2018-04-04 17:12:35 +03:00
}) as IRemoteUser;
2018-09-01 11:53:38 +03:00
}
2018-04-04 17:12:35 +03:00
// Update Person activityの場合は、ここで署名検証/更新処理まで実施して終了
2018-09-01 11:53:38 +03:00
if (activity.type === 'Update') {
if (activity.object && activity.object.type === 'Person') {
if (user == null) {
logger.warn('Update activity received, but user not registed.');
2018-09-01 11:53:38 +03:00
} else if (!httpSignature.verifySignature(signature, user.publicKey.publicKeyPem)) {
logger.warn('Update activity received, but signature verification failed.');
2018-09-01 11:53:38 +03:00
} else {
updatePerson(activity.actor, null, activity.object);
}
done();
return;
2018-04-04 17:12:35 +03:00
}
2018-09-01 11:53:38 +03:00
}
// アクティビティを送信してきたユーザーがまだMisskeyサーバーに登録されていなかったら登録する
if (user === null) {
user = await resolvePerson(activity.actor) as IRemoteUser;
2018-04-04 17:12:35 +03:00
}
if (user === null) {
done(new Error('failed to resolve user'));
return;
}
2018-04-15 06:51:05 +03:00
if (!httpSignature.verifySignature(signature, user.publicKey.publicKeyPem)) {
logger.error('signature verification failed');
2018-04-06 08:35:17 +03:00
done();
2018-04-04 17:12:35 +03:00
return;
}
2018-11-16 10:04:28 +02:00
2018-11-05 12:40:09 +02:00
//#region Log
publishApLogStream({
direction: 'in',
activity: activity.type,
host: user.host,
actor: user.username
});
//#endregion
2018-04-04 17:12:35 +03:00
2019-02-07 09:05:29 +02:00
// Update stats
registerOrFetchInstanceDoc(user.host).then(i => {
Instance.update({ _id: i._id }, {
$set: {
latestRequestReceivedAt: new Date(),
lastCommunicatedAt: new Date(),
isNotResponding: false
2019-02-07 09:05:29 +02:00
}
});
instanceChart.requestReceived(i.host);
2019-02-07 09:05:29 +02:00
});
2018-04-04 17:12:35 +03:00
// アクティビティを処理
try {
2018-04-08 22:08:56 +03:00
await perform(user, activity);
2018-04-04 17:12:35 +03:00
done();
} catch (e) {
done(e);
}
};
2018-08-30 14:53:41 +03:00
/**
* Validate host in activity
* @param activity Activity
* @param host Expect host
*/
function ValidateActivity(activity: any, host: string) {
// id (if exists)
if (typeof activity.id === 'string') {
const uriHost = toUnicode(new URL(activity.id).hostname.toLowerCase());
2018-08-31 10:46:24 +03:00
if (host !== uriHost) {
const diag = activity.signature ? '. Has LD-Signature. Forwarded?' : '';
throw new Error(`activity.id(${activity.id}) has different host(${host})${diag}`);
}
2018-08-30 14:53:41 +03:00
}
// actor (if exists)
if (typeof activity.actor === 'string') {
const uriHost = toUnicode(new URL(activity.actor).hostname.toLowerCase());
if (host !== uriHost) throw new Error('activity.actor has different host');
}
// For Create activity
if (activity.type === 'Create' && activity.object) {
// object.id (if exists)
if (typeof activity.object.id === 'string') {
const uriHost = toUnicode(new URL(activity.object.id).hostname.toLowerCase());
if (host !== uriHost) throw new Error('activity.object.id has different host');
}
// object.attributedTo (if exists)
if (typeof activity.object.attributedTo === 'string') {
const uriHost = toUnicode(new URL(activity.object.attributedTo).hostname.toLowerCase());
if (host !== uriHost) throw new Error('activity.object.attributedTo has different host');
}
}
}