'use strict'; const nconf = require('nconf'); const winston = require('winston'); const { createHash, createSign, createVerify, getHashes } = require('crypto'); const request = require('../request'); const db = require('../database'); const meta = require('../meta'); const user = require('../user'); const utils = require('../utils'); const ttl = require('../cache/ttl'); const lru = require('../cache/lru'); const batch = require('../batch'); const pubsub = require('../pubsub'); const analytics = require('../analytics'); const requestCache = ttl({ max: 5000, ttl: 1000 * 60 * 5, // 5 minutes }); const ActivityPub = module.exports; ActivityPub._constants = Object.freeze({ uid: -2, publicAddress: 'https://www.w3.org/ns/activitystreams#Public', acceptableTypes: [ 'application/activity+json', 'application/ld+json; profile="https://www.w3.org/ns/activitystreams"', ], acceptedPostTypes: [ 'Note', 'Page', 'Article', 'Question', ], acceptedProtocols: ['https', ...(process.env.CI === 'true' ? ['http'] : [])], }); ActivityPub._cache = requestCache; ActivityPub.helpers = require('./helpers'); ActivityPub.inbox = require('./inbox'); ActivityPub.mocks = require('./mocks'); ActivityPub.notes = require('./notes'); ActivityPub.actors = require('./actors'); ActivityPub.resolveId = async (uid, id) => { try { const query = new URL(id); ({ id } = await ActivityPub.get('uid', uid, id)); const response = new URL(id); if (query.host !== response.host) { winston.warn(`[activitypub/resolveId] id resolution domain mismatch: ${query.href} != ${response.href}`); return null; } return id; } catch (e) { return null; } }; ActivityPub.resolveInboxes = async (ids) => { const inboxes = new Set(); if (!meta.config.activitypubAllowLoopback) { ids = ids.filter((id) => { const { hostname } = new URL(id); return hostname !== nconf.get('url_parsed').hostname; }); } await ActivityPub.actors.assert(ids); await Promise.all(ids.map(async (id) => { const { inbox, sharedInbox } = await user.getUserFields(id, ['inbox', 'sharedInbox']); if (sharedInbox || inbox) { inboxes.add(sharedInbox || inbox); } })); return Array.from(inboxes); }; ActivityPub.getPublicKey = async (type, id) => { let publicKey; try { ({ publicKey } = await db.getObject(`${type}:${id}:keys`)); } catch (e) { ({ publicKey } = await ActivityPub.helpers.generateKeys(type, id)); } return publicKey; }; ActivityPub.getPrivateKey = async (type, id) => { // Sanity checking if (!['cid', 'uid'].includes(type) || !utils.isNumber(id) || parseInt(id, 10) < 0) { throw new Error('[[error:invalid-data]]'); } id = parseInt(id, 10); let privateKey; try { ({ privateKey } = await db.getObject(`${type}:${id}:keys`)); } catch (e) { ({ privateKey } = await ActivityPub.helpers.generateKeys(type, id)); } let keyId; if (type === 'uid') { keyId = `${nconf.get('url')}${id > 0 ? `/uid/${id}` : '/actor'}#key`; } else { keyId = `${nconf.get('url')}/category/${id}#key`; } return { key: privateKey, keyId }; }; ActivityPub.fetchPublicKey = async (uri) => { // Used for retrieving the public key from the passed-in keyId uri const body = await ActivityPub.get('uid', 0, uri); if (!body.hasOwnProperty('publicKey')) { throw new Error('[[error:activitypub.pubKey-not-found]]'); } return body.publicKey; }; ActivityPub.sign = async ({ key, keyId }, url, payload) => { // Returns string for use in 'Signature' header const { host, pathname } = new URL(url); const date = new Date().toUTCString(); let digest = null; let headers = '(request-target) host date'; let signed_string = `(request-target): ${payload ? 'post' : 'get'} ${pathname}\nhost: ${host}\ndate: ${date}`; // Calculate payload hash if payload present if (payload) { const payloadHash = createHash('sha256'); payloadHash.update(JSON.stringify(payload)); digest = `SHA-256=${payloadHash.digest('base64')}`; headers += ' digest'; signed_string += `\ndigest: ${digest}`; } // Sign string using private key let signature = createSign('sha256'); signature.update(signed_string); signature.end(); signature = signature.sign(key, 'base64'); // Construct signature header return { date, digest, signature: `keyId="${keyId}",headers="${headers}",signature="${signature}",algorithm="hs2019"`, }; }; ActivityPub.verify = async (req) => { winston.verbose('[activitypub/verify] Starting signature verification...'); if (!req.headers.hasOwnProperty('signature')) { winston.verbose('[activitypub/verify] Failed, no signature header.'); return false; } // Break the signature apart let { keyId, headers, signature, algorithm, created, expires } = req.headers.signature.split(',').reduce((memo, cur) => { const split = cur.split('="'); const key = split.shift(); const value = split.join('="'); memo[key] = value.slice(0, -1); return memo; }, {}); const acceptableHashes = getHashes(); if (algorithm === 'hs2019' || !acceptableHashes.includes(algorithm)) { algorithm = 'sha256'; } // Re-construct signature string const signed_string = headers.split(' ').reduce((memo, cur) => { switch (cur) { case '(request-target)': { memo.push(`${cur}: ${String(req.method).toLowerCase()} ${req.baseUrl}${req.path}`); break; } case '(created)': { memo.push(`${cur}: ${created}`); break; } case '(expires)': { memo.push(`${cur}: ${expires}`); break; } default: { memo.push(`${cur}: ${req.headers[cur]}`); break; } } return memo; }, []).join('\n'); // Verify the signature string via public key try { // Retrieve public key from remote instance winston.verbose(`[activitypub/verify] Retrieving pubkey for ${keyId}`); const { publicKeyPem } = await ActivityPub.fetchPublicKey(keyId); const verify = createVerify('sha256'); verify.update(signed_string); verify.end(); winston.verbose('[activitypub/verify] Attempting signed string verification'); const verified = verify.verify(publicKeyPem, signature, 'base64'); return verified; } catch (e) { winston.verbose('[activitypub/verify] Failed, key retrieval or verification failure.'); return false; } }; ActivityPub.get = async (type, id, uri) => { const cacheKey = [id, uri].join(';'); const cached = requestCache.get(cacheKey); if (cached !== undefined) { return cached; } const keyData = await ActivityPub.getPrivateKey(type, id); const headers = id >= 0 ? await ActivityPub.sign(keyData, uri) : {}; winston.verbose(`[activitypub/get] ${uri}`); try { const { response, body } = await request.get(uri, { headers: { ...headers, Accept: 'application/ld+json; profile="https://www.w3.org/ns/activitystreams"', }, timeout: 5000, }); if (!String(response.statusCode).startsWith('2')) { winston.error(`[activitypub/get] Received ${response.statusCode} when querying ${uri}`); if (body.hasOwnProperty('error')) { winston.error(`[activitypub/get] Error received: ${body.error}`); } throw new Error(`[[error:activitypub.get-failed]]`); } requestCache.set(cacheKey, body); return body; } catch (e) { // Handle things like non-json body, etc. throw new Error(`[[error:activitypub.get-failed]]`); } }; ActivityPub.retryQueue = lru({ name: 'activitypub-retry-queue', max: 4000, ttl: 1000 * 60 * 60 * 24 * 60 }); // handle clearing retry queue from another member of the cluster pubsub.on(`activitypub-retry-queue:lruCache:del`, (keys) => { if (Array.isArray(keys)) { keys.forEach(key => clearTimeout(ActivityPub.retryQueue.get(key))); } }); async function sendMessage(uri, id, type, payload, attempts = 1) { const keyData = await ActivityPub.getPrivateKey(type, id); const headers = await ActivityPub.sign(keyData, uri, payload); winston.verbose(`[activitypub/send] ${uri}`); try { const { response, body } = await request.post(uri, { headers: { ...headers, 'content-type': 'application/ld+json; profile="https://www.w3.org/ns/activitystreams"', }, body: payload, }); if (String(response.statusCode).startsWith('2')) { winston.verbose(`[activitypub/send] Successfully sent ${payload.type} to ${uri}`); } else { throw new Error(String(body)); } } catch (e) { winston.warn(`[activitypub/send] Could not send ${payload.type} to ${uri}; error: ${e.message}`); // add to retry queue if (attempts < 12) { // stop attempting after ~2 months const timeout = (4 ** attempts) * 1000; // exponential backoff const queueId = `${payload.type}:${payload.id}:${new URL(uri).hostname}`; const timeoutId = setTimeout(() => sendMessage(uri, id, type, payload, attempts + 1), timeout); ActivityPub.retryQueue.set(queueId, timeoutId); winston.verbose(`[activitypub/send] Added ${payload.type} to ${uri} to retry queue for ${timeout}ms`); } else { winston.warn(`[activitypub/send] Max attempts reached for ${payload.type} to ${uri}; giving up on sending`); } } } ActivityPub.send = async (type, id, targets, payload) => { if (!Array.isArray(targets)) { targets = [targets]; } const inboxes = await ActivityPub.resolveInboxes(targets); const actor = ActivityPub.helpers.resolveActor(type, id); payload = { '@context': 'https://www.w3.org/ns/activitystreams', actor, ...payload, }; await batch.processArray( inboxes, async inboxBatch => Promise.all(inboxBatch.map(async uri => sendMessage(uri, id, type, payload))), { batch: 50, interval: 100, }, ); }; ActivityPub.record = async ({ id, type, actor }) => { const now = Date.now(); const { hostname } = new URL(actor); await Promise.all([ db.sortedSetAdd(`activities:datetime`, now, id), db.sortedSetAdd('domains:lastSeen', now, hostname), analytics.increment(['activities', `activities:byType:${type}`, `activities:byHost:${hostname}`]), ]); };