RPC to do AP fetches

.config/example.yml

@@ -29,6 +29,10 @@ url: https://example.com/
 # The port that your Calckey server should listen on.
 port: 3000
 
+#   ┌────────────────────────────┐
+#───┘ Magnetar RPC configuration └──────────────────────────────
+rpcHost: magnetar:4935
+
 #   ┌──────────────────────────┐
 #───┘ PostgreSQL configuration └────────────────────────────────
 

package.json

@@ -6,7 +6,7 @@
 		"type": "git",
 		"url": "https://git.astolfo.cool/natty/calckey"
 	},
-	"packageManager": "pnpm@8.6.3",
+	"packageManager": "pnpm@8.6.3+sha512.d18e277ae8072091046bccbca0931f77dc3080791cd6122ae890bf504125d8af76b37fb33da287dba9fbbb6da6ebb13e314e9fa4a464c7effe3d8599cebe7243",
 	"private": true,
 	"scripts": {
 		"rebuild": "pnpm run clean && pnpm node ./scripts/build-greet.js && pnpm -r run build",

packages/backend/.swcrc

@@ -1,10 +1,11 @@
 {
-	"$schema": "https://json.schemastore.org/swcrc",
+	"$schema": "https://swc.rs/schema.json",
 	"jsc": {
         "parser": {
 			"syntax": "typescript",
 			"dynamicImport": true,
 			"decorators": true
+
 		},
 		"transform": {
 			"legacyDecorator": true,
@@ -21,5 +22,10 @@
 		},
 		"target": "es2022"
 	},
+    "module": {
+        "type": "es6",
+        "importInterop": "node",
+        "resolveFully": true
+    },
     "minify": false
 }

packages/backend/package.json

@@ -20,15 +20,16 @@
 		"@koa/cors": "3.4.3",
 		"@koa/multer": "3.0.2",
 		"@koa/router": "9.0.1",
+		"@msgpack/msgpack": "3.0.0-beta2",
 		"@peertube/http-signature": "1.7.0",
 		"@redocly/openapi-core": "1.0.0-beta.120",
 		"@sinonjs/fake-timers": "9.1.2",
 		"@syuilo/aiscript": "0.11.1",
-		"adm-zip": "^0.5.10",
+		"adm-zip": "^0.5.16",
 		"ajv": "8.12.0",
 		"archiver": "5.3.1",
 		"argon2": "^0.30.3",
-		"async-mutex": "^0.4.0",
+		"async-mutex": "^0.4.1",
 		"autobind-decorator": "2.4.0",
 		"autolinker": "4.0.0",
 		"autwh": "0.1.0",
@@ -40,7 +41,7 @@
 		"cbor": "8.1.0",
 		"chalk": "5.2.0",
 		"chalk-template": "0.4.0",
-		"chokidar": "3.5.3",
+		"chokidar": "^3.6.0",
 		"cli-highlight": "2.1.11",
 		"color-convert": "2.0.1",
 		"content-disposition": "0.5.4",
@@ -79,7 +80,7 @@
 		"nodemailer": "6.9.3",
 		"oauth": "^0.10.0",
 		"os-utils": "0.0.14",
-		"otpauth": "^9.1.2",
+		"otpauth": "^9.3.2",
 		"parse5": "7.1.2",
 		"pg": "8.11.0",
 		"private-ip": "2.3.4",
@@ -100,6 +101,7 @@
 		"sanitize-html": "2.10.0",
 		"semver": "7.5.1",
 		"sharp": "0.32.6",
+		"smart-buffer": "^4.2.0",
 		"sonic-channel": "^1.3.1",
 		"stringz": "2.1.0",
 		"syslog-pro": "1.0.0",
@@ -116,9 +118,9 @@
 		"xev": "3.0.2"
 	},
 	"devDependencies": {
-		"@swc/cli": "^0.1.62",
-		"@swc/core": "^1.3.62",
-		"@types/adm-zip": "^0.5.0",
+		"@swc/cli": "^0.1.65",
+		"@swc/core": "^1.7.23",
+		"@types/adm-zip": "^0.5.5",
 		"@types/bcryptjs": "2.4.2",
 		"@types/bull": "3.15.9",
 		"@types/cbor": "6.0.0",
@@ -143,7 +145,7 @@
 		"@types/node-fetch": "3.0.3",
 		"@types/nodemailer": "6.4.8",
 		"@types/oauth": "0.9.1",
-		"@types/probe-image-size": "^7.2.0",
+		"@types/probe-image-size": "^7.2.5",
 		"@types/punycode": "2.1.0",
 		"@types/qrcode": "1.5.0",
 		"@types/qs": "6.9.7",
@@ -163,17 +165,17 @@
 		"@types/ws": "8.5.4",
 		"autobind-decorator": "2.4.0",
 		"cross-env": "7.0.3",
-		"eslint": "^8.42.0",
+		"eslint": "^8.57.0",
 		"execa": "6.1.0",
 		"json5": "2.2.3",
 		"json5-loader": "4.0.1",
 		"strict-event-emitter-types": "2.0.0",
-		"swc-loader": "^0.2.3",
+		"swc-loader": "^0.2.6",
 		"ts-loader": "9.4.3",
 		"ts-node": "10.9.1",
 		"tsconfig-paths": "4.2.0",
 		"typescript": "5.1.3",
-		"webpack": "^5.85.1",
+		"webpack": "^5.94.0",
 		"ws": "8.13.0"
 	}
 }

packages/backend/src/config/types.ts

@@ -7,6 +7,7 @@ export type Source = {
 	url: string;
 	port: number;
 	disableHsts?: boolean;
+    rpcHost: string;
 	db: {
 		host: string;
 		port: number;

packages/backend/src/mag/rpc-client.ts

@@ -0,0 +1,142 @@
+import {Socket} from "node:net"
+import {SmartBuffer} from "smart-buffer";
+
+import {decode, encode} from "@msgpack/msgpack";
+import Logger from "@/services/logger.js";
+import config from "@/config/index.js";
+import type {IObject} from "@/remote/activitypub/type";
+
+let client: Socket | null = null;
+let exponentialBackoff = 0;
+let serial: bigint = BigInt(0);
+
+const logger = new Logger("RpcLog");
+
+function getRpcClient(): Socket {
+    if (client) {
+        return client;
+    }
+
+    const [host, portStr] = config.rpcHost.trim().split(/:(?=[0-9]+$)/, 2);
+    const port = parseInt(portStr);
+    client = new Socket();
+    const reconnectWithBackoff = () => {
+        setTimeout(() => {
+            client!.connect(port, host);
+        }, 1000 * (1 + Math.pow(1.5, exponentialBackoff)));
+    };
+
+    const buf = new SmartBuffer({
+        encoding: "binary"
+    });
+
+    client.on("connect", () => {
+        exponentialBackoff = 0;
+    })
+
+    client.on("error", e => {
+        logger.warn(`RPC connection error: ${e}`);
+        client!.removeAllListeners("data");
+        buf.clear();
+        exponentialBackoff = Math.min(12, exponentialBackoff + 1);
+        reconnectWithBackoff();
+    })
+
+    client.on("close", () => {
+        client!.removeAllListeners("data");
+        buf.clear();
+        exponentialBackoff = Math.min(12, exponentialBackoff + 1);
+        reconnectWithBackoff();
+    });
+
+    client.on("data", (recv) => {
+        buf.writeBuffer(recv);
+
+        if (buf.length < 1 + 4 + 8) {
+            return;
+        }
+
+        const header = buf.readUInt8();
+        if (header != 77) {
+            logger.error(`Invalid header: ${header}`);
+            return;
+        }
+
+        const serial = buf.readBigUInt64BE();
+
+        const dataLen = buf.readUInt32BE();
+        if (buf.remaining() < dataLen) {
+            buf.readOffset = 0;
+            return;
+        }
+        const data = buf.readBuffer(dataLen);
+        const dataDecoded: any = decode(data);
+
+        // Move the rest of the data to the beginning of the buffer
+        const rest = buf.readBuffer();
+        buf.clear();
+        buf.writeBuffer(rest);
+        client!.emit(`mag-data:${serial}`, dataDecoded);
+    });
+
+    client.connect(port, host);
+
+    return client;
+}
+
+async function rpcCall<D, T>(method: string, data: D): Promise<T> {
+    const header = new Uint8Array([77]);
+    const textEncoder = new TextEncoder();
+    const methodBuf = textEncoder.encode(method);
+    const dataBuf = encode(data);
+
+    const serialLength = 8;
+    const sizeLength = 4;
+
+    const packetBuf = new Uint8Array(header.length + serialLength + sizeLength + methodBuf.length + sizeLength + dataBuf.length);
+    packetBuf.set(header, 0);
+    packetBuf.set(methodBuf, header.length + serialLength + sizeLength)
+    packetBuf.set(dataBuf, header.length + serialLength + sizeLength + methodBuf.length + sizeLength);
+
+    const packetDataView = new DataView(packetBuf.buffer);
+    packetDataView.setBigUint64(header.length, serial);
+    packetDataView.setUint32(header.length + serialLength, methodBuf.length);
+    packetDataView.setUint32(header.length + serialLength + sizeLength + methodBuf.length, dataBuf.length);
+
+    const client = getRpcClient();
+    client.write(packetBuf);
+
+    const fut = new Promise((resolve, reject) => {
+        client.once(`mag-data:${serial}`, resolve);
+        client.once("close", reject);
+        client.once("error", reject);
+    });
+
+    serial++;
+
+    const result = await fut as { success: false, data: any } | { success: true, data: T };
+
+    if (!result.success) {
+        throw result.data;
+    }
+
+    return result.data;
+}
+
+
+export async function magApGet(userId: string, url: string): Promise<IObject> {
+    logger.info(`AP GET to: ${url}`);
+    return await rpcCall("/ap/get", {
+        user_id: userId,
+        url
+    });
+}
+
+export async function magApPost(userId: string, url: string, body: string): Promise<IObject> {
+    logger.info(`AP POST to: ${url}`);
+    return await rpcCall("/ap/post", {
+        user_id: userId,
+        url,
+        body
+    });
+}

packages/backend/src/misc/gen-key-pair.ts

@@ -18,25 +18,3 @@ export async function genRsaKeyPair(modulusLength = 2048) {
 		},
 	});
 }
-
-export async function genEcKeyPair(
-	namedCurve:
-		| "prime256v1"
-		| "secp384r1"
-		| "secp521r1"
-		| "curve25519" = "prime256v1",
-) {
-	return await generateKeyPair("ec", {
-		namedCurve,
-		publicKeyEncoding: {
-			type: "spki",
-			format: "pem",
-		},
-		privateKeyEncoding: {
-			type: "pkcs8",
-			format: "pem",
-			cipher: undefined,
-			passphrase: undefined,
-		},
-	});
-}

packages/backend/src/prelude/xml.ts

@@ -1,38 +0,0 @@
-const map: Record<string, string> = {
-	"&": "&amp;",
-	"<": "&lt;",
-	">": "&gt;",
-	'"': "&quot;",
-	"'": "&apos;",
-};
-
-const beginingOfCDATA = "<![CDATA[";
-const endOfCDATA = "]]>";
-
-export function escapeValue(x: string): string {
-	let insideOfCDATA = false;
-	let builder = "";
-	for (let i = 0; i < x.length; ) {
-		if (insideOfCDATA) {
-			if (x.slice(i, i + beginingOfCDATA.length) === beginingOfCDATA) {
-				insideOfCDATA = true;
-				i += beginingOfCDATA.length;
-			} else {
-				builder += x[i++];
-			}
-		} else {
-			if (x.slice(i, i + endOfCDATA.length) === endOfCDATA) {
-				insideOfCDATA = false;
-				i += endOfCDATA.length;
-			} else {
-				const b = x[i++];
-				builder += map[b] || b;
-			}
-		}
-	}
-	return builder;
-}
-
-export function escapeAttribute(x: string): string {
-	return Object.entries(map).reduce((a, [k, v]) => a.replace(k, v), x);
-}

packages/backend/src/remote/activitypub/ap-request.ts

@@ -1,152 +0,0 @@
-import * as crypto from "node:crypto";
-import { URL } from "node:url";
-
-type Request = {
-	url: string;
-	method: string;
-	headers: Record<string, string>;
-};
-
-type PrivateKey = {
-	privateKeyPem: string;
-	keyId: string;
-};
-
-export function createSignedPost(args: {
-	key: PrivateKey;
-	url: string;
-	body: string;
-	additionalHeaders: Record<string, string>;
-}) {
-	const u = new URL(args.url);
-	const digestHeader = `SHA-256=${crypto
-		.createHash("sha256")
-		.update(args.body)
-		.digest("base64")}`;
-
-	const request: Request = {
-		url: u.href,
-		method: "POST",
-		headers: objectAssignWithLcKey(
-			{
-				Date: new Date().toUTCString(),
-				Host: u.hostname,
-				"Content-Type": "application/activity+json",
-				Digest: digestHeader,
-			},
-			args.additionalHeaders,
-		),
-	};
-
-	const result = signToRequest(request, args.key, [
-		"(request-target)",
-		"date",
-		"host",
-		"digest",
-	]);
-
-	return {
-		request,
-		signingString: result.signingString,
-		signature: result.signature,
-		signatureHeader: result.signatureHeader,
-	};
-}
-
-export function createSignedGet(args: {
-	key: PrivateKey;
-	url: string;
-	additionalHeaders: Record<string, string>;
-}) {
-	const u = new URL(args.url);
-
-	const request: Request = {
-		url: u.href,
-		method: "GET",
-		headers: objectAssignWithLcKey(
-			{
-				Accept: "application/activity+json, application/ld+json",
-				Date: new Date().toUTCString(),
-				Host: new URL(args.url).hostname,
-			},
-			args.additionalHeaders,
-		),
-	};
-
-	const result = signToRequest(request, args.key, [
-		"(request-target)",
-		"date",
-		"host",
-		"accept",
-	]);
-
-	return {
-		request,
-		signingString: result.signingString,
-		signature: result.signature,
-		signatureHeader: result.signatureHeader,
-	};
-}
-
-function signToRequest(
-	request: Request,
-	key: PrivateKey,
-	includeHeaders: string[],
-) {
-	const signingString = genSigningString(request, includeHeaders);
-	const signature = crypto
-		.sign("sha256", Buffer.from(signingString), key.privateKeyPem)
-		.toString("base64");
-	const signatureHeader = `keyId="${
-		key.keyId
-	}",algorithm="rsa-sha256",headers="${includeHeaders.join(
-		" ",
-	)}",signature="${signature}"`;
-
-	request.headers = objectAssignWithLcKey(request.headers, {
-		Signature: signatureHeader,
-	});
-
-	return {
-		request,
-		signingString,
-		signature,
-		signatureHeader,
-	};
-}
-
-function genSigningString(request: Request, includeHeaders: string[]) {
-	request.headers = lcObjectKey(request.headers);
-
-	const results: string[] = [];
-
-	for (const key of includeHeaders.map((x) => x.toLowerCase())) {
-		if (key === "(request-target)") {
-			results.push(
-				`(request-target): ${request.method.toLowerCase()} ${
-					new URL(request.url).pathname
-				}`,
-			);
-		} else {
-			results.push(`${key}: ${request.headers[key]}`);
-		}
-	}
-
-	return results.join("\n");
-}
-
-function lcObjectKey(src: Record<string, string>) {
-	const dst: Record<string, string> = {};
-	for (const key of Object.keys(src).filter(
-		(x) => x !== "__proto__" && typeof src[x] === "string",
-	))
-		dst[key.toLowerCase()] = src[key];
-	return dst;
-}
-
-function objectAssignWithLcKey(
-	a: Record<string, string>,
-	b: Record<string, string>,
-) {
-	return Object.assign(lcObjectKey(a), lcObjectKey(b));
-}

packages/backend/src/remote/activitypub/models/icon.ts

@@ -1,5 +0,0 @@
-export type IIcon = {
-	type: string;
-	mediaType?: string;
-	url?: string;
-};

packages/backend/src/remote/activitypub/request.ts

@@ -1,32 +1,10 @@
-import config from "@/config/index.js";
-import { getUserKeypair } from "@/misc/keypair-store.js";
 import type {User} from "@/models/entities/user.js";
-import { getResponse } from "../../misc/fetch.js";
-import { createSignedPost, createSignedGet } from "./ap-request.js";
+import {magApGet, magApPost} from "@/mag/rpc-client.js";
 
 export default async (user: { id: User["id"] }, url: string, object: any) => {
 	const body = JSON.stringify(object);
 
-	const keypair = await getUserKeypair(user.id);
-
-	const req = createSignedPost({
-		key: {
-			privateKeyPem: keypair.privateKey,
-			keyId: `${config.url}/users/${user.id}#main-key`,
-		},
-		url,
-		body,
-		additionalHeaders: {
-			"User-Agent": config.userAgent,
-		},
-	});
-
-	await getResponse({
-		url,
-		method: req.request.method,
-		headers: req.request.headers,
-		body,
-	});
+    return await magApPost(user.id, url, body);
 };
 
 /**
@@ -35,24 +13,5 @@ export default async (user: { id: User["id"] }, url: string, object: any) => {
  * @param url URL to fetch
  */
 export async function signedGet(url: string, user: { id: User["id"] }) {
-	const keypair = await getUserKeypair(user.id);
-
-	const req = createSignedGet({
-		key: {
-			privateKeyPem: keypair.privateKey,
-			keyId: `${config.url}/users/${user.id}#main-key`,
-		},
-		url,
-		additionalHeaders: {
-			"User-Agent": config.userAgent,
-		},
-	});
-
-	const res = await getResponse({
-		url,
-		method: req.request.method,
-		headers: req.request.headers,
-	});
-
-	return await res.json();
+    return await magApGet(user.id, url);
 }

packages/backend/src/remote/activitypub/resolver.ts

@@ -1,19 +1,12 @@
 import config from "@/config/index.js";
-import { getJson } from "@/misc/fetch.js";
 import type {ILocalUser} from "@/models/entities/user.js";
 import {getInstanceActor} from "@/services/instance-actor.js";
 import {fetchMeta} from "@/misc/fetch-meta.js";
 import {extractDbHost, isSelfHost} from "@/misc/convert-host.js";
 import {signedGet} from "./request.js";
-import type { IObject, ICollection, IOrderedCollection } from "./type.js";
-import { isCollectionOrOrderedCollection, getApId } from "./type.js";
-import {
-	FollowRequests,
-	Notes,
-	NoteReactions,
-	Polls,
-	Users,
-} from "@/models/index.js";
+import type {ICollection, IObject, IOrderedCollection} from "./type.js";
+import {getApId, isCollectionOrOrderedCollection} from "./type.js";
+import {NoteReactions, Notes, Polls, Users,} from "@/models/index.js";
 import {parseUri} from "./db-resolver.js";
 import renderNote from "@/remote/activitypub/renderer/note.js";
 import {renderLike} from "@/remote/activitypub/renderer/like.js";
@@ -29,9 +22,10 @@ export default class Resolver {
 	private user?: ILocalUser;
 	private recursionLimit?: number;
 
-	constructor(recursionLimit = 100) {
+	constructor(recursionLimit = 100, user?: ILocalUser) {
 		this.history = new Set();
 		this.recursionLimit = recursionLimit;
+        this.user = user;
 	}
 
 	public getHistory(): string[] {
@@ -102,11 +96,7 @@ export default class Resolver {
 			this.user = await getInstanceActor();
 		}
 
-		const object = (
-			this.user
-				? await signedGet(value, this.user)
-				: await getJson(value, "application/activity+json, application/ld+json")
-		) as IObject;
+		const object = await signedGet(value, this.user);
 
 		if (
 			object == null ||

packages/backend/src/server/api/endpoints/ap/get.ts

@@ -29,8 +29,7 @@ export const paramDef = {
 	required: ["uri"],
 } as const;
 
-export default define(meta, paramDef, async (ps) => {
-	const resolver = new Resolver();
-	const object = await resolver.resolve(ps.uri);
-	return object;
+export default define(meta, paramDef, async (ps, me) => {
+	const resolver = new Resolver(undefined, me);
+    return await resolver.resolve(ps.uri);
 });

packages/backend/src/server/api/endpoints/ap/show.ts

@@ -126,7 +126,7 @@ async function fetchAny(
 	}
 
 	// fetching Object once from remote
-	const resolver = new Resolver();
+	const resolver = new Resolver(undefined, me ?? undefined);
 	const object = await resolver.resolve(uri);
 
 	// /@user If a URI other than the id is specified,

packages/backend/src/server/well-known.ts

@@ -1,38 +1,13 @@
 import Router from "@koa/router";
 
 import config from "@/config/index.js";
-import {escapeAttribute, escapeValue} from "@/prelude/xml.js";
 
 // Init router
 const router = new Router();
 
-const XRD = (
-    ...x: {
-        element: string;
-        value?: string;
-        attributes?: Record<string, string>;
-    }[]
-) =>
-    `<?xml version="1.0" encoding="UTF-8"?><XRD xmlns="http://docs.oasis-open.org/ns/xri/xrd-1.0">${x
-        .map(
-            ({ element, value, attributes }) =>
-                `<${Object.entries(
-                    (typeof attributes === "object" && attributes) || {}
-                ).reduce(
-                    (a, [k, v]) => `${a} ${k}="${escapeAttribute(v)}"`,
-                    element
-                )}${
-                    typeof value === "string"
-                        ? `>${escapeValue(value)}</${element}`
-                        : "/"
-                }>`
-        )
-        .reduce((a, c) => a + c, "")}</XRD>`;
-
 const allPath = "/.well-known/(.*)";
 const webFingerPath = "/.well-known/webfinger";
 const jrd = "application/jrd+json";
-const xrd = "application/xrd+xml";
 
 router.use(allPath, async (ctx, next) => {
     ctx.set({
@@ -48,18 +23,6 @@ router.options(allPath, async (ctx) => {
     ctx.status = 204;
 });
 
-router.get("/.well-known/host-meta", async (ctx) => {
-    ctx.set("Content-Type", xrd);
-    ctx.body = XRD({
-        element: "Link",
-        attributes: {
-            rel: "lrdd",
-            type: xrd,
-            template: `${config.url}${webFingerPath}?resource={uri}`,
-        },
-    });
-});
-
 router.get("/.well-known/host-meta.json", async (ctx) => {
     ctx.set("Content-Type", jrd);
     ctx.body = {