shadowfacts.net/lib/activitypub/federate.ts

import { promises as fs } from "fs";
import crypto from "crypto";
import uuidv4 from "uuid/v4";
import request from "request";
import {
	Activity,
	ArticleObject,
	ActorObject,
	CreateActivity,
	NoteObject
} from "./activity";
import { URL } from "url";
import { getConnection } from "typeorm";
import Actor from "../entity/Actor";
import Article from "../entity/Article";
import Note from "../entity/Note";

const domain = process.env.DOMAIN;

export function createActivity(object: ArticleObject | NoteObject): CreateActivity {
	const uuid = uuidv4();
	const createObject = {
		"@context": [
			"https://www.w3.org/ns/activitystreams"
		],
		"type": "Create",
		"id": `https://${domain}/ap/${uuid}`,
		"actor": `https://${domain}/ap/actor`,
		"to": object.to,
		"cc": object.cc,
		"object": object
	};
	return createObject;
}

export async function getActor(url: string, forceUpdate: boolean = false): Promise<ActorObject | null> {
	if (!forceUpdate) {
		try {
			const cached = await getCachedActor(url);
			if (cached) return cached;
		} catch (err) {
			console.error(`Encountered error getting cached actor ${url}`, err);
		}
	}
	const remote = await fetchActor(url);
	if (remote) {
		cacheActor(remote);
	} else {
		getConnection().createQueryBuilder()
			.delete()
			.from(Actor, "actor")
			.where("actor.id = :id", { id: url })
			.execute();
		getConnection().createQueryBuilder()
			.delete()
			.from(Note, "note")
			.where("note.\"actorId\" = :actor", { actor: url })
			.execute();
	}
	return remote;
}

export async function getCachedActor(url: string): Promise<ActorObject | null> {
	const result = await getConnection().manager.findByIds(Actor, [url]);
	if (result.length > 0) {
		const actor = result[0];
		return actor.actorObject;
	} else {
		return null;
	}
}

async function cacheActor(actorObject: ActorObject) {
	function getIconUrl(icon: string | object): string {
		return icon instanceof String ? icon : (icon as any).url;
	}
	const iconURL = !actorObject.icon ? null : actorObject.icon instanceof Array ? getIconUrl(actorObject.icon[0]) : getIconUrl(actorObject.icon);
	const actor = new Actor();
	actor.id = actorObject.id;
	actor.actorObject = actorObject;
	actor.displayName = actorObject.name;
	actor.inbox = actorObject.inbox;
	actor.iconURL = iconURL;
	if (actorObject.publicKey && actorObject.publicKey.publicKeyPem) {
		actor.publicKeyPem = actorObject.publicKey.publicKeyPem;
	}
	actor.isFollower = false;
	await getConnection().manager.save(actor);
}

async function fetchActor(url?: string): Promise<ActorObject | null> {
	if (!url) {
		return Promise.resolve(null);
	}
	return new Promise((resolve, reject) => {
		request({
			url,
			headers: {
				"Accept": "application/activity+json"
			},
			method: "GET",
			json: true
		}, (err, res) => {
			if (err) reject(err);
			else if (!res.body || res.body.error || Object.keys(res.body).length === 0) resolve(null);
			else resolve(res.body as ActorObject);
		});
	});
}

export async function signAndSend(activity: Activity, inbox: string) {
	const targetDomain = new URL(inbox).hostname;
	const inboxFragment = inbox.replace("https://" + targetDomain, "");
	const date = new Date();
	const privKey = (await fs.readFile(process.env.PRIV_KEY_PEM!)).toString();
	const bodyDigest = crypto.createHash("sha256").update(JSON.stringify(activity)).digest("base64");
	const stringToSign = `(request-target): post ${inboxFragment}\nhost: ${targetDomain}\ndate: ${date.toUTCString()}\ndigest: SHA-256=${bodyDigest}`;
	const signer = crypto.createSign("sha256");
	signer.update(stringToSign);
	signer.end();
	const signature = signer.sign(privKey, "base64");
	const header = `keyId="https://${domain}/ap/actor#main-key",headers="(request-target) host date digest",signature="${signature}"`;
	console.log("Sending:", activity);
	console.log("stringToSign:", stringToSign);
	console.log("Signature: " + header);
	request({
		url: inbox,
		headers: {
			"Host": targetDomain,
			"Date": date.toUTCString(),
			"Digest": `SHA-256=${bodyDigest}`,
			"Signature": header,
			"Accept": "application/activity+json, application/json"
		},
		method: "POST",
		json: true,
		body: activity,
	}, (err, res) => {
		console.log("Sent message to inbox at", targetDomain);
		if (err) console.log("Error:", err, res);
		if (res) {
			console.log("Response status code", res.statusCode);
			console.log(res.body);
		}
	});
}

async function sendToFollowers(activity: CreateActivity) {
	const followers = await getConnection().getRepository(Actor).find({where: {isFollower: true}});
	const inboxes = followers.map(it => "https://" + new URL(it.inbox).host + "/inbox");
	// convert to a Set to deduplicate inboxes
	(new Set(inboxes)).forEach(inbox => {
		console.log(`Federating ${activity.object.id} to ${inbox}`);
		signAndSend(activity, inbox);
	});
}

export default async function federate(toFederate: [string, ArticleObject][]) {
	for (const [id, article] of toFederate) {
		
		sendToFollowers(createActivity(article));
		await getConnection().manager.update(Article, id, { hasFederated: true });
	}
}