refactor(federation): ♻️ Move incoming federation handling to custom class

This commit is contained in:
Jesse Wierzbinski 2024-11-01 20:42:32 +01:00
parent d570e8c200
commit f26493140f
No known key found for this signature in database
9 changed files with 1066 additions and 622 deletions

View file

@ -3,7 +3,6 @@ import { createRoute } from "@hono/zod-openapi";
import { z } from "zod";
import { Relationship } from "~/classes/database/relationship";
import { User } from "~/classes/database/user";
import { sendFollowAccept } from "~/classes/functions/user";
import { RolePermissions } from "~/drizzle/schema";
import { ErrorSchema } from "~/types/api";
@ -97,7 +96,7 @@ export default apiRoute((app) =>
// Check if accepting remote follow
if (account.isRemote()) {
// Federate follow accept
await sendFollowAccept(account, user);
await user.sendFollowAccept(account);
}
return context.json(foundRelationship.toApi(), 200);

View file

@ -3,7 +3,6 @@ import { createRoute } from "@hono/zod-openapi";
import { z } from "zod";
import { Relationship } from "~/classes/database/relationship";
import { User } from "~/classes/database/user";
import { sendFollowReject } from "~/classes/functions/user";
import { RolePermissions } from "~/drizzle/schema";
import { ErrorSchema } from "~/types/api";
@ -97,7 +96,7 @@ export default apiRoute((app) =>
// Check if rejecting remote follow
if (account.isRemote()) {
// Federate follow reject
await sendFollowReject(account, user);
await user.sendFollowReject(account);
}
return context.json(foundRelationship.toApi(), 200);

View file

@ -1,36 +1,10 @@
import { apiRoute, applyConfig, debugRequest } from "@/api";
import { sentry } from "@/sentry";
import { apiRoute, applyConfig } from "@/api";
import { createRoute } from "@hono/zod-openapi";
import { type Logger, getLogger } from "@logtape/logtape";
import {
EntityValidator,
RequestParserHandler,
SignatureValidator,
} from "@versia/federation";
import type {
Entity,
Delete as VersiaDelete,
Follow as VersiaFollow,
FollowAccept as VersiaFollowAccept,
FollowReject as VersiaFollowReject,
LikeExtension as VersiaLikeExtension,
Note as VersiaNote,
User as VersiaUser,
} from "@versia/federation/types";
import type { SocketAddress } from "bun";
import { eq } from "drizzle-orm";
import type { Context, TypedResponse } from "hono";
import { matches } from "ip-matching";
import { getLogger } from "@logtape/logtape";
import type { Entity } from "@versia/federation/types";
import { z } from "zod";
import { type ValidationError, isValidationError } from "zod-validation-error";
import { Like } from "~/classes/database/like";
import { Note } from "~/classes/database/note";
import { Relationship } from "~/classes/database/relationship";
import { User } from "~/classes/database/user";
import { sendFollowAccept } from "~/classes/functions/user";
import { db } from "~/drizzle/db";
import { Likes, Notes, Notifications } from "~/drizzle/schema";
import { config } from "~/packages/config-manager";
import { InboxProcessor } from "~/classes/inbox/processor";
import { ErrorSchema } from "~/types/api";
export const meta = applyConfig({
@ -127,60 +101,16 @@ const route = createRoute({
export default apiRoute((app) =>
app.openapi(route, async (context) => {
const { uuid } = context.req.valid("param");
const {
"x-signature": signature,
"x-nonce": nonce,
"x-signed-by": signedBy,
authorization,
} = context.req.valid("header");
const logger = getLogger(["federation", "inbox"]);
const body: Entity = await context.req.valid("json");
if (config.debug.federation) {
// Debug request
await debugRequest(
new Request(context.req.url, {
method: context.req.method,
headers: context.req.raw.headers,
body: await context.req.text(),
}),
);
}
const user = await User.fromId(uuid);
if (!user) {
return context.json({ error: "User not found" }, 404);
}
if (user.isRemote()) {
return context.json(
{ error: "Cannot view users from remote instances" },
403,
);
}
const requestIp = context.env?.ip ?? null;
let checkSignature = true;
if (config.federation.bridge.enabled) {
const token = authorization?.split("Bearer ")[1];
if (token) {
const bridgeResponse = await handleBridgeRequest(
token,
requestIp,
context,
);
if (bridgeResponse) {
return bridgeResponse;
}
checkSignature = false;
}
}
const sender = await User.resolve(signedBy);
if (!sender) {
@ -197,532 +127,18 @@ export default apiRoute((app) =>
);
}
const hostname = sender?.data.instance?.baseUrl ?? "";
// Check if Origin is defederated
if (isDefederated(hostname)) {
// Return 201 to not make the sender think there's an error
return context.newResponse(null, 201);
}
// Verify request signature
if (checkSignature) {
const signatureResponse = await verifySignature(
const processor = new InboxProcessor(
context,
body,
sender,
{
signature,
nonce,
context,
authorization,
},
logger,
);
if (signatureResponse) {
return signatureResponse;
}
}
const validator = new EntityValidator();
const handler = new RequestParserHandler(body, validator);
try {
return await handler.parseBody<Response>({
note: async (note) => handleNoteRequest(note, context, logger),
follow: async (follow) =>
handleFollowRequest(follow, user, context),
followAccept: async (followAccept) =>
handleFollowAcceptRequest(followAccept, user, context),
followReject: async (followReject) =>
handleFollowRejectRequest(followReject, user, context),
"pub.versia:likes/Like": async (like) =>
handleLikeRequest(like, context),
delete: async (delete_) =>
handleDeleteRequest(delete_, user, context),
user: async (user) => handleUserRequest(user, context),
unknown: () =>
context.json({ error: "Unknown entity type" }, 400),
});
} catch (e) {
return handleError(e as Error, context, logger);
}
return await processor.process();
}),
);
/**
* Handles bridge requests.
* @param {string} token - The authorization token.
* @param {SocketAddress | null} requestIp - The request IP address.
* @param {Context} context - Hono request context.
* @returns {Promise<Response | null>} - The response or null if no error.
*/
function handleBridgeRequest(
token: string,
requestIp: SocketAddress | null,
context: Context,
): (Response & TypedResponse<{ error: string }, 401 | 500, "json">) | null {
if (token !== config.federation.bridge.token) {
return context.json(
{
error: "An invalid token was passed in the Authorization header. Please use the correct token, or remove the Authorization header.",
},
401,
);
}
if (requestIp?.address) {
if (config.federation.bridge.allowed_ips.length > 0) {
for (const ip of config.federation.bridge.allowed_ips) {
if (matches(ip, requestIp?.address)) {
return null;
}
}
}
} else {
return context.json(
{
error: "Request IP address is not available",
},
500,
);
}
return null;
}
/**
* Checks if the hostname is defederated using glob matching.
* @param {string} hostname - The hostname to check. Can contain glob patterns.
* @returns {boolean} - True if defederated, false otherwise.
*/
function isDefederated(hostname: string): boolean {
const pattern = new Bun.Glob(hostname);
return (
config.federation.blocked.find(
(blocked) => pattern.match(blocked) !== null,
) !== undefined
);
}
/**
* Verifies the request signature.
* @param {User} sender - The sender user.
* @param {string} signature - The request signature.
* @param {string} nonce - The request nonce.
* @param {Context} context - Hono request context.
* @param {Logger} logger - LogTape logger.
* @returns {Promise<Response | null>} - The response or null if no error.
*/
async function verifySignature(
sender: User,
signature: string,
nonce: string,
context: Context,
logger: Logger,
): Promise<
(Response & TypedResponse<{ error: string }, 401 | 400, "json">) | null
> {
if (!sender) {
return context.json({ error: "Could not resolve sender" }, 400);
}
if (config.debug.federation) {
logger.debug`Sender public key: ${sender.data.publicKey}`;
}
const validator = await SignatureValidator.fromStringKey(
sender.data.publicKey,
);
const isValid = await validator
.validate(
new Request(context.req.url, {
method: context.req.method,
headers: {
"X-Signature": signature,
"X-Nonce": nonce,
},
body: await context.req.text(),
}),
)
.catch((e) => {
logger.error`${e}`;
sentry?.captureException(e);
return false;
});
if (!isValid) {
return context.json({ error: "Signature could not be verified" }, 401);
}
return null;
}
/**
* Handles Note entity processing.
*
* @param {VersiaNote} note - Note entity to process.
* @param {Context} context - Hono request context.
* @param {Logger} logger - LogTape logger.
* @returns {Promise<Response>} - The response.
*/
async function handleNoteRequest(
note: VersiaNote,
context: Context,
logger: Logger,
): Promise<
Response &
TypedResponse<
| {
error: string;
}
| string,
404 | 500 | 201,
"json" | "text"
>
> {
const account = await User.resolve(note.author);
if (!account) {
return context.json({ error: "Author not found" }, 404);
}
const newStatus = await Note.fromVersia(note, account).catch((e) => {
logger.error`${e}`;
sentry?.captureException(e);
return null;
});
if (!newStatus) {
return context.json({ error: "Failed to add status" }, 500);
}
return context.text("Note created", 201);
}
/**
* Handles Follow entity processing.
*
* @param {VersiaFollow} follow - Follow entity to process.
* @param {User} user - Owner of this inbox.
* @param {Context} context - Hono request context.
* @returns {Promise<Response>} - The response.
*/
async function handleFollowRequest(
follow: VersiaFollow,
user: User,
context: Context,
): Promise<
Response &
TypedResponse<
| {
error: string;
}
| string,
200 | 400,
"text" | "json"
>
> {
const account = await User.resolve(follow.author);
if (!account) {
return context.json({ error: "Author not found" }, 400);
}
const foundRelationship = await Relationship.fromOwnerAndSubject(
account,
user,
);
if (foundRelationship.data.following) {
return context.text("Already following", 200);
}
await foundRelationship.update({
following: !user.data.isLocked,
requested: user.data.isLocked,
showingReblogs: true,
notifying: true,
languages: [],
});
await db.insert(Notifications).values({
accountId: account.id,
type: user.data.isLocked ? "follow_request" : "follow",
notifiedId: user.id,
});
if (!user.data.isLocked) {
await sendFollowAccept(account, user);
}
return context.text("Follow request sent", 200);
}
/**
* Handles FollowAccept entity processing
*
* @param {VersiaFollowAccept} followAccept - FollowAccept entity to process.
* @param {User} user - Owner of this inbox.
* @param {Context} context - Hono request context.
* @returns {Promise<Response>} - The response.
*/
async function handleFollowAcceptRequest(
followAccept: VersiaFollowAccept,
user: User,
context: Context,
): Promise<
Response &
TypedResponse<{ error: string } | string, 200 | 400, "text" | "json">
> {
const account = await User.resolve(followAccept.author);
if (!account) {
return context.json({ error: "Author not found" }, 400);
}
const foundRelationship = await Relationship.fromOwnerAndSubject(
user,
account,
);
if (!foundRelationship.data.requested) {
return context.text("There is no follow request to accept", 200);
}
await foundRelationship.update({
requested: false,
following: true,
});
return context.text("Follow request accepted", 200);
}
/**
* Handles FollowReject entity processing
*
* @param {VersiaFollowReject} followReject - FollowReject entity to process.
* @param {User} user - Owner of this inbox.
* @param {Context} context - Hono request context.
* @returns {Promise<Response>} - The response.
*/
async function handleFollowRejectRequest(
followReject: VersiaFollowReject,
user: User,
context: Context,
): Promise<
Response &
TypedResponse<{ error: string } | string, 200 | 400, "text" | "json">
> {
const account = await User.resolve(followReject.author);
if (!account) {
return context.json({ error: "Author not found" }, 400);
}
const foundRelationship = await Relationship.fromOwnerAndSubject(
user,
account,
);
if (!foundRelationship.data.requested) {
return context.text("There is no follow request to reject", 200);
}
await foundRelationship.update({
requested: false,
following: false,
});
return context.text("Follow request rejected", 200);
}
/**
* Handles Like entity processing.
*
* @param {VersiaLikeExtension} like - Like entity to process.
* @param {Context} context - Hono request context.
* @returns {Promise<Response>} - The response.
*/
async function handleLikeRequest(
like: VersiaLikeExtension,
context: Context,
): Promise<
Response &
TypedResponse<{ error: string } | string, 200 | 400, "text" | "json">
> {
const author = await User.resolve(like.author);
if (!author) {
return context.json({ error: "Author not found" }, 400);
}
const note = await Note.resolve(like.liked);
if (!note) {
return context.json({ error: "Note not found" }, 400);
}
await author.like(note, like.uri);
return context.text("Like added", 200);
}
/**
* Handles Delete entity processing.
*
* @param {VersiaDelete} delete_ - Delete entity to process.
* @param {User} user - Owner of this inbox.
* @param {Context} context - Hono request context.
* @returns {Promise<Response>} - The response.
*/
async function handleDeleteRequest(
delete_: VersiaDelete,
user: User,
context: Context,
): Promise<
Response &
TypedResponse<
{ error: string } | string,
200 | 400 | 404,
"text" | "json"
>
> {
const toDelete = delete_.deleted;
switch (delete_.deleted_type) {
case "Note": {
const note = await Note.fromSql(
eq(Notes.uri, toDelete),
eq(Notes.authorId, user.id),
);
if (note) {
await note.delete();
return context.text("Note deleted", 200);
}
break;
}
case "User": {
const otherUser = await User.resolve(toDelete);
if (otherUser) {
if (otherUser.id === user.id) {
await user.delete();
return context.text("Account deleted", 200);
}
return context.json(
{
error: "Cannot delete other users than self",
},
400,
);
}
break;
}
case "pub.versia:likes/Like": {
const like = await Like.fromSql(
eq(Likes.uri, toDelete),
eq(Likes.likerId, user.id),
);
if (like) {
await like.delete();
return context.text("Like deleted", 200);
}
return context.json(
{
error: "Like not found or not owned by user",
},
404,
);
}
default: {
return context.json(
{
error: `Deletion of object ${toDelete} not implemented`,
},
400,
);
}
}
return context.json(
{ error: "Object not found or not owned by user" },
404,
);
}
/**
* Handles User entity processing (profile edits).
*
* @param {VersiaUser} user - User entity to process.
* @param {Context} context - Hono request context.
* @returns {Promise<Response>} - The response.
*/
async function handleUserRequest(
user: VersiaUser,
context: Context,
): Promise<
Response &
TypedResponse<{ error: string } | string, 200 | 500, "text" | "json">
> {
const updatedAccount = await User.saveFromRemote(user.uri);
if (!updatedAccount) {
return context.json({ error: "Failed to update user" }, 500);
}
return context.text("User refreshed", 200);
}
/**
* Processes Errors into the appropriate HTTP response.
*
* @param {Error} e - The error object.
* @param {Context} context - Hono request context.
* @param {any} logger - LogTape logger.
* @returns {Response} - The error response.
*/
function handleError(
e: Error,
context: Context,
logger: Logger,
):
| (Response &
TypedResponse<
{
error: string;
error_description: string;
},
400,
"json"
>)
| (Response &
TypedResponse<
{
error: string;
message: string;
},
500,
"json"
>) {
if (isValidationError(e)) {
return context.json(
{
error: "Failed to process request",
error_description: (e as ValidationError).message,
},
400,
);
}
logger.error`${e}`;
sentry?.captureException(e);
return context.json(
{
error: "Failed to process request",
message: (e as Error).message,
},
500,
);
}

View file

@ -20,7 +20,7 @@ import { config } from "~/packages/config-manager/index.ts";
import { BaseInterface } from "./base.ts";
import { User } from "./user.ts";
export type AttachmentType = InferSelectModel<typeof Instances>;
export type InstanceType = InferSelectModel<typeof Instances>;
export class Instance extends BaseInterface<typeof Instances> {
async reload(): Promise<void> {
@ -78,9 +78,7 @@ export class Instance extends BaseInterface<typeof Instances> {
return found.map((s) => new Instance(s));
}
async update(
newInstance: Partial<AttachmentType>,
): Promise<AttachmentType> {
async update(newInstance: Partial<InstanceType>): Promise<InstanceType> {
await db
.update(Instances)
.set(newInstance)
@ -96,7 +94,7 @@ export class Instance extends BaseInterface<typeof Instances> {
return updated.data;
}
save(): Promise<AttachmentType> {
save(): Promise<InstanceType> {
return this.update(this.data);
}
@ -108,6 +106,14 @@ export class Instance extends BaseInterface<typeof Instances> {
}
}
public static async fromUser(user: User): Promise<Instance | null> {
if (!user.data.instanceId) {
return null;
}
return await Instance.fromId(user.data.instanceId);
}
public static async insert(
data: InferInsertModel<typeof Instances>,
): Promise<Instance> {

View file

@ -38,6 +38,8 @@ import { z } from "zod";
import {
type UserWithRelations,
findManyUsers,
followAcceptToVersia,
followRejectToVersia,
followRequestToVersia,
} from "~/classes/functions/user";
import { searchManager } from "~/classes/search/search-manager";
@ -313,6 +315,20 @@ export class User extends BaseInterface<typeof Users, UserWithRelations> {
};
}
public async sendFollowAccept(follower: User): Promise<void> {
await this.federateToUser(
followAcceptToVersia(follower, this),
follower,
);
}
public async sendFollowReject(follower: User): Promise<void> {
await this.federateToUser(
followRejectToVersia(follower, this),
follower,
);
}
static async webFinger(
manager: FederationRequester,
username: string,

View file

@ -111,20 +111,6 @@ export const getFromHeader = async (value: string): Promise<AuthData> => {
return { user, token, application };
};
export const sendFollowAccept = async (follower: User, followee: User) => {
await follower.federateToUser(
followAcceptToVersia(follower, followee),
followee,
);
};
export const sendFollowReject = async (follower: User, followee: User) => {
await follower.federateToUser(
followRejectToVersia(follower, followee),
followee,
);
};
export const transformOutputToUserWithRelations = (
user: Omit<UserType, "endpoints"> & {
followerCount: unknown;

View file

@ -0,0 +1,418 @@
import { beforeEach, describe, expect, jest, mock, test } from "bun:test";
import { SignatureValidator } from "@versia/federation";
import type { Entity, Note as VersiaNote } from "@versia/federation/types";
import { Note, Relationship, User } from "@versia/kit/db";
import { db } from "@versia/kit/db";
import type { Context } from "hono";
import { ValidationError } from "zod-validation-error";
import { config } from "~/packages/config-manager/index.ts";
import { InboxProcessor } from "./processor.ts";
// Mock dependencies
mock.module("@versia/kit/db", () => ({
db: {
insert: jest.fn(
// Return something with a `.values()` method
() => ({ values: jest.fn() }),
),
},
User: {
resolve: jest.fn(),
saveFromRemote: jest.fn(),
sendFollowAccept: jest.fn(),
},
Instance: {
fromUser: jest.fn(),
},
Note: {
resolve: jest.fn(),
fromVersia: jest.fn(),
fromSql: jest.fn(),
},
Relationship: {
fromOwnerAndSubject: jest.fn(),
},
Like: {
fromSql: jest.fn(),
},
}));
mock.module("@versia/federation", () => ({
SignatureValidator: {
fromStringKey: jest.fn(() => ({
validate: jest.fn(),
})),
},
EntityValidator: jest.fn(() => ({
validate: jest.fn(),
})),
RequestParserHandler: jest.fn(),
}));
mock.module("~/packages/config-manager/index.ts", () => ({
config: {
debug: {
federation: false,
},
federation: {
blocked: [],
bridge: {
enabled: false,
token: "test-token",
allowed_ips: [],
},
},
},
}));
describe("InboxProcessor", () => {
let mockContext: Context;
let mockBody: Entity;
let mockSender: User;
let mockHeaders: {
signature: string;
nonce: string;
authorization?: string;
};
let processor: InboxProcessor;
beforeEach(() => {
// Reset all mocks
mock.restore();
// Setup basic mock context
mockContext = {
json: jest.fn(),
text: jest.fn(),
req: {
url: "https://test.com",
method: "POST",
text: jest.fn().mockResolvedValue("test-body"),
},
env: {
ip: { address: "127.0.0.1" },
},
} as unknown as Context;
// Setup basic mock sender
mockSender = {
id: "test-id",
data: {
publicKey: "test-key",
},
} as unknown as User;
// Setup basic mock headers
mockHeaders = {
signature: "test-signature",
nonce: "test-nonce",
};
// Setup basic mock body
mockBody = {} as Entity;
// Create processor instance
processor = new InboxProcessor(
mockContext,
mockBody,
mockSender,
mockHeaders,
);
});
describe("isSignatureValid", () => {
test("returns true for valid signature", async () => {
const mockValidator = {
validate: jest.fn().mockResolvedValue(true),
};
SignatureValidator.fromStringKey = jest
.fn()
.mockResolvedValue(mockValidator);
// biome-ignore lint/complexity/useLiteralKeys: Private method
const result = await processor["isSignatureValid"]();
expect(result).toBe(true);
expect(mockValidator.validate).toHaveBeenCalled();
});
test("returns false for invalid signature", async () => {
const mockValidator = {
validate: jest.fn().mockResolvedValue(false),
};
SignatureValidator.fromStringKey = jest
.fn()
.mockResolvedValue(mockValidator);
// biome-ignore lint/complexity/useLiteralKeys: Private method
const result = await processor["isSignatureValid"]();
expect(result).toBe(false);
});
});
describe("shouldCheckSignature", () => {
test("returns true when bridge is disabled", () => {
// biome-ignore lint/complexity/useLiteralKeys: Private method
const result = processor["shouldCheckSignature"]();
expect(result).toBe(true);
});
test("returns false for valid bridge request", () => {
config.federation.bridge.enabled = true;
config.federation.bridge.token = "valid-token";
config.federation.bridge.allowed_ips = ["127.0.0.1"];
mockHeaders.authorization = "Bearer valid-token";
// biome-ignore lint/complexity/useLiteralKeys: Private method
const result = processor["shouldCheckSignature"]();
expect(result).toBe(false);
});
test("returns error response for invalid token", () => {
config.federation.bridge.enabled = true;
mockHeaders.authorization = "Bearer invalid-token";
// biome-ignore lint/complexity/useLiteralKeys: Private method
const result = processor["shouldCheckSignature"]() as {
code: number;
};
expect(result.code).toBe(401);
});
});
describe("processNote", () => {
test("successfully processes valid note", async () => {
const mockNote = { author: "test-author" };
const mockAuthor = { id: "test-id" };
User.resolve = jest.fn().mockResolvedValue(mockAuthor);
Note.fromVersia = jest.fn().mockResolvedValue(true);
mockContext.text = jest.fn().mockReturnValue({ status: 201 });
// biome-ignore lint/complexity/useLiteralKeys: Private variable
processor["body"] = mockNote as VersiaNote;
// biome-ignore lint/complexity/useLiteralKeys: Private method
await processor["processNote"]();
expect(User.resolve).toHaveBeenCalledWith("test-author");
expect(Note.fromVersia).toHaveBeenCalledWith(mockNote, mockAuthor);
expect(mockContext.text).toHaveBeenCalledWith("Note created", 201);
});
test("returns 404 when author not found", async () => {
User.resolve = jest.fn().mockResolvedValue(null);
mockContext.json = jest.fn().mockReturnValue({ status: 404 });
// biome-ignore lint/complexity/useLiteralKeys: Private method
await processor["processNote"]();
expect(mockContext.json).toHaveBeenCalledWith(
{ error: "Author not found" },
404,
);
});
});
describe("processFollowRequest", () => {
test("successfully processes follow request for unlocked account", async () => {
const mockFollow = {
author: "test-author",
followee: "test-followee",
};
const mockAuthor = { id: "author-id" };
const mockFollowee = {
id: "followee-id",
data: { isLocked: false },
sendFollowAccept: jest.fn(),
};
const mockRelationship = {
data: { following: false },
update: jest.fn(),
};
User.resolve = jest
.fn()
.mockResolvedValueOnce(mockAuthor)
.mockResolvedValueOnce(mockFollowee);
Relationship.fromOwnerAndSubject = jest
.fn()
.mockResolvedValue(mockRelationship);
mockContext.text = jest.fn().mockReturnValue({ status: 200 });
// biome-ignore lint/complexity/useLiteralKeys: Private variable
processor["body"] = mockFollow as unknown as Entity;
// biome-ignore lint/complexity/useLiteralKeys: Private method
await processor["processFollowRequest"]();
expect(mockRelationship.update).toHaveBeenCalledWith({
following: true,
requested: false,
showingReblogs: true,
notifying: true,
languages: [],
});
expect(db.insert).toHaveBeenCalled();
});
test("returns 404 when author not found", async () => {
User.resolve = jest.fn().mockResolvedValue(null);
mockContext.json = jest.fn().mockReturnValue({ status: 404 });
// biome-ignore lint/complexity/useLiteralKeys: Private method
await processor["processFollowRequest"]();
expect(mockContext.json).toHaveBeenCalledWith(
{ error: "Author not found" },
404,
);
});
});
describe("processDelete", () => {
test("successfully deletes a note", async () => {
const mockDelete = {
deleted_type: "Note",
deleted: "test-uri",
};
const mockNote = {
delete: jest.fn(),
};
Note.fromSql = jest.fn().mockResolvedValue(mockNote);
mockContext.text = jest.fn().mockReturnValue({ status: 200 });
// biome-ignore lint/complexity/useLiteralKeys: Private variable
processor["body"] = mockDelete as unknown as Entity;
// biome-ignore lint/complexity/useLiteralKeys: Private method
await processor["processDelete"]();
expect(mockNote.delete).toHaveBeenCalled();
expect(mockContext.text).toHaveBeenCalledWith("Note deleted", 200);
});
test("returns 404 when note not found", async () => {
const mockDelete = {
deleted_type: "Note",
deleted: "test-uri",
};
Note.fromSql = jest.fn().mockResolvedValue(null);
mockContext.json = jest.fn().mockReturnValue({ status: 404 });
// biome-ignore lint/complexity/useLiteralKeys: Private variable
processor["body"] = mockDelete as unknown as Entity;
// biome-ignore lint/complexity/useLiteralKeys: Private method
await processor["processDelete"]();
expect(mockContext.json).toHaveBeenCalledWith(
{ error: "Note to delete not found or not owned by sender" },
404,
);
});
});
describe("processLikeRequest", () => {
test("successfully processes like request", async () => {
const mockLike = {
author: "test-author",
liked: "test-note",
uri: "test-uri",
};
const mockAuthor = {
like: jest.fn(),
};
const mockNote = { id: "note-id" };
User.resolve = jest.fn().mockResolvedValue(mockAuthor);
Note.resolve = jest.fn().mockResolvedValue(mockNote);
mockContext.text = jest.fn().mockReturnValue({ status: 200 });
// biome-ignore lint/complexity/useLiteralKeys: Private variable
processor["body"] = mockLike as unknown as Entity;
// biome-ignore lint/complexity/useLiteralKeys: Private method
await processor["processLikeRequest"]();
expect(mockAuthor.like).toHaveBeenCalledWith(mockNote, "test-uri");
expect(mockContext.text).toHaveBeenCalledWith("Like created", 200);
});
test("returns 404 when author not found", async () => {
User.resolve = jest.fn().mockResolvedValue(null);
mockContext.json = jest.fn().mockReturnValue({ status: 404 });
// biome-ignore lint/complexity/useLiteralKeys: Private method
await processor["processLikeRequest"]();
expect(mockContext.json).toHaveBeenCalledWith(
{ error: "Author not found" },
404,
);
});
});
describe("processUserRequest", () => {
test("successfully processes user update", async () => {
const mockUser = {
uri: "test-uri",
};
const mockUpdatedUser = { id: "user-id" };
User.saveFromRemote = jest.fn().mockResolvedValue(mockUpdatedUser);
mockContext.text = jest.fn().mockReturnValue({ status: 200 });
// biome-ignore lint/complexity/useLiteralKeys: Private variable
processor["body"] = mockUser as unknown as Entity;
// biome-ignore lint/complexity/useLiteralKeys: Private method
await processor["processUserRequest"]();
expect(User.saveFromRemote).toHaveBeenCalledWith("test-uri");
expect(mockContext.text).toHaveBeenCalledWith("User updated", 200);
});
test("returns 500 when update fails", async () => {
User.saveFromRemote = jest.fn().mockResolvedValue(null);
mockContext.json = jest.fn().mockReturnValue({ status: 500 });
// biome-ignore lint/complexity/useLiteralKeys: Private method
await processor["processUserRequest"]();
expect(mockContext.json).toHaveBeenCalledWith(
{ error: "Failed to update user" },
500,
);
});
});
describe("handleError", () => {
test("handles validation errors", () => {
const validationError = new ValidationError("Invalid data");
mockContext.json = jest.fn().mockReturnValue({ status: 400 });
// biome-ignore lint/complexity/useLiteralKeys: Private method
processor["handleError"](validationError);
expect(mockContext.json).toHaveBeenCalledWith(
{
error: "Failed to process request",
error_description: "Invalid data",
},
400,
);
});
test("handles general errors", () => {
const error = new Error("Something went wrong");
mockContext.json = jest.fn().mockReturnValue({ status: 500 });
// biome-ignore lint/complexity/useLiteralKeys: Private method
processor["handleError"](error);
expect(mockContext.json).toHaveBeenCalledWith(
{
error: "Failed to process request",
message: "Something went wrong",
},
500,
);
});
});
});

602
classes/inbox/processor.ts Normal file
View file

@ -0,0 +1,602 @@
import { sentry } from "@/sentry";
import { type Logger, getLogger } from "@logtape/logtape";
import {
EntityValidator,
RequestParserHandler,
SignatureValidator,
} from "@versia/federation";
import type {
Entity,
Delete as VersiaDelete,
Follow as VersiaFollow,
FollowAccept as VersiaFollowAccept,
FollowReject as VersiaFollowReject,
LikeExtension as VersiaLikeExtension,
Note as VersiaNote,
User as VersiaUser,
} from "@versia/federation/types";
import { Instance, Like, Note, Relationship, User, db } from "@versia/kit/db";
import { Likes, Notes, Notifications } from "@versia/kit/tables";
import type { SocketAddress } from "bun";
import { eq } from "drizzle-orm";
import type { Context, TypedResponse } from "hono";
import type { StatusCode } from "hono/utils/http-status";
import { matches } from "ip-matching";
import { type ValidationError, isValidationError } from "zod-validation-error";
import { config } from "~/packages/config-manager/index.ts";
type ResponseBody = {
message?: string;
code: StatusCode;
};
/**
* Checks if the hostname is defederated using glob matching.
* @param {string} hostname - The hostname to check. Can contain glob patterns.
* @returns {boolean} - True if defederated, false otherwise.
*/
function isDefederated(hostname: string): boolean {
const pattern = new Bun.Glob(hostname);
return (
config.federation.blocked.find(
(blocked) => pattern.match(blocked) !== null,
) !== undefined
);
}
/**
* Processes incoming federation inbox messages.
*
* @example
* ```typescript
* const processor = new InboxProcessor(context, body, sender, headers);
*
* const response = await processor.process();
*
* return response;
* ```
*/
export class InboxProcessor {
/**
* Creates a new InboxProcessor instance.
*
* @param context Hono request context.
* @param body Entity JSON body.
* @param sender Sender of the request (from X-Signed-By header).
* @param headers Various request headers.
* @param logger LogTape logger instance.
* @param requestIp Request IP address. Grabs it from the Hono context if not provided.
*/
constructor(
private context: Context,
private body: Entity,
private sender: User,
private headers: {
signature: string;
nonce: string;
authorization?: string;
},
private logger: Logger = getLogger(["federation", "inbox"]),
private requestIp: SocketAddress | null = context.env?.ip ?? null,
) {}
/**
* Verifies the request signature.
*
* @returns {Promise<boolean>} - Whether the signature is valid.
*/
private async isSignatureValid(): Promise<boolean> {
if (config.debug.federation) {
this.logger.debug`Sender public key: ${this.sender.data.publicKey}`;
}
const validator = await SignatureValidator.fromStringKey(
this.sender.data.publicKey,
);
// HACK: Making a fake Request object instead of passing the values directly is necessary because otherwise the validation breaks for some unknown reason
const isValid = await validator.validate(
new Request(this.context.req.url, {
method: this.context.req.method,
headers: {
"X-Signature": this.headers.signature,
"X-Nonce": this.headers.nonce,
},
body: await this.context.req.text(),
}),
);
return isValid;
}
/**
* Determines if signature checks can be skipped.
* Useful for requests from federation bridges.
*
* @returns {boolean | ResponseBody} - Whether to skip signature checks. May include a response body if there are errors.
*/
private shouldCheckSignature(): boolean | ResponseBody {
if (config.federation.bridge.enabled) {
const token = this.headers.authorization?.split("Bearer ")[1];
if (token) {
const isBridge = this.isRequestFromBridge(token);
if (isBridge === true) {
return false;
}
return isBridge;
}
}
return true;
}
/**
* Checks if a request is from a federation bridge.
*
* @param token - Authorization token to check.
* @returns
*/
private isRequestFromBridge(token: string): boolean | ResponseBody {
if (token !== config.federation.bridge.token) {
return {
message:
"An invalid token was passed in the Authorization header. Please use the correct token, or remove the Authorization header.",
code: 401,
};
}
if (!this.requestIp) {
return {
message: "The request IP address could not be determined.",
code: 500,
};
}
if (config.federation.bridge.allowed_ips.length > 0) {
for (const ip of config.federation.bridge.allowed_ips) {
if (matches(ip, this.requestIp.address)) {
return true;
}
}
}
return {
message: "The request is not from a trusted bridge IP address.",
code: 403,
};
}
/**
* Performs request processing.
*
* @returns {Promise<Response>} - HTTP response to send back.
*/
public async process(): Promise<
(Response & TypedResponse<{ error: string }, 500, "json">) | Response
> {
const remoteInstance = await Instance.fromUser(this.sender);
if (!remoteInstance) {
return this.context.json(
{ error: "Could not resolve the remote instance." },
500,
);
}
if (isDefederated(remoteInstance.data.baseUrl)) {
// Return 201 to avoid
// 1. Leaking defederated instance information
// 2. Preventing the sender from thinking the message was not delivered and retrying
return this.context.text("", 201);
}
const shouldCheckSignature = this.shouldCheckSignature();
if (shouldCheckSignature !== true && shouldCheckSignature !== false) {
return this.context.json(
{ error: shouldCheckSignature.message },
shouldCheckSignature.code,
);
}
if (shouldCheckSignature) {
const isValid = await this.isSignatureValid();
if (!isValid) {
return this.context.json(
{ error: "Signature is not valid" },
401,
);
}
}
const validator = new EntityValidator();
const handler = new RequestParserHandler(this.body, validator);
try {
return await handler.parseBody<Response>({
note: () => this.processNote(),
follow: () => this.processFollowRequest(),
followAccept: () => this.processFollowAccept(),
followReject: () => this.processFollowReject(),
"pub.versia:likes/Like": () => this.processLikeRequest(),
delete: () => this.processDelete(),
user: () => this.processUserRequest(),
unknown: () =>
this.context.json({ error: "Unknown entity type" }, 400),
});
} catch (e) {
return this.handleError(e as Error);
}
}
/**
* Handles Note entity processing.
*
* @returns {Promise<Response>} - The response.
*/
private async processNote(): Promise<
Response &
TypedResponse<
| {
error: string;
}
| string,
404 | 500 | 201,
"json" | "text"
>
> {
const note = this.body as VersiaNote;
const author = await User.resolve(note.author);
if (!author) {
return this.context.json({ error: "Author not found" }, 404);
}
await Note.fromVersia(note, author);
return this.context.text("Note created", 201);
}
/**
* Handles Follow entity processing.
*
* @returns {Promise<Response>} - The response.
*/
private async processFollowRequest(): Promise<
Response &
TypedResponse<
| {
error: string;
}
| string,
200 | 404,
"text" | "json"
>
> {
const follow = this.body as unknown as VersiaFollow;
const author = await User.resolve(follow.author);
const followee = await User.resolve(follow.followee);
if (!author) {
return this.context.json({ error: "Author not found" }, 404);
}
if (!followee) {
return this.context.json({ error: "Followee not found" }, 404);
}
const foundRelationship = await Relationship.fromOwnerAndSubject(
author,
followee,
);
if (foundRelationship.data.following) {
return this.context.text("Already following", 200);
}
await foundRelationship.update({
// If followee is not "locked" (doesn't manually approves follow requests), set following to true
following: !followee.data.isLocked,
requested: followee.data.isLocked,
showingReblogs: true,
notifying: true,
languages: [],
});
await db.insert(Notifications).values({
accountId: author.id,
type: followee.data.isLocked ? "follow_request" : "follow",
notifiedId: followee.id,
});
if (!followee.data.isLocked) {
await followee.sendFollowAccept(author);
}
return this.context.text("Follow request sent", 200);
}
/**
* Handles FollowAccept entity processing
*
* @returns {Promise<Response>} - The response.
*/
private async processFollowAccept(): Promise<
Response &
TypedResponse<
{ error: string } | string,
200 | 404,
"text" | "json"
>
> {
const followAccept = this.body as unknown as VersiaFollowAccept;
const author = await User.resolve(followAccept.author);
const follower = await User.resolve(followAccept.follower);
if (!author) {
return this.context.json({ error: "Author not found" }, 404);
}
if (!follower) {
return this.context.json({ error: "Follower not found" }, 404);
}
const foundRelationship = await Relationship.fromOwnerAndSubject(
follower,
author,
);
if (!foundRelationship.data.requested) {
return this.context.text(
"There is no follow request to accept",
200,
);
}
await foundRelationship.update({
requested: false,
following: true,
});
return this.context.text("Follow request accepted", 200);
}
/**
* Handles FollowReject entity processing
*
* @returns {Promise<Response>} - The response.
*/
private async processFollowReject(): Promise<
Response &
TypedResponse<
{ error: string } | string,
200 | 404,
"text" | "json"
>
> {
const followReject = this.body as unknown as VersiaFollowReject;
const author = await User.resolve(followReject.author);
const follower = await User.resolve(followReject.follower);
if (!author) {
return this.context.json({ error: "Author not found" }, 404);
}
if (!follower) {
return this.context.json({ error: "Follower not found" }, 404);
}
const foundRelationship = await Relationship.fromOwnerAndSubject(
follower,
author,
);
if (!foundRelationship.data.requested) {
return this.context.text(
"There is no follow request to reject",
200,
);
}
await foundRelationship.update({
requested: false,
following: false,
});
return this.context.text("Follow request rejected", 200);
}
/**
* Handles Delete entity processing.
*
* @returns {Promise<Response>} - The response.
*/
public async processDelete(): Promise<
Response &
TypedResponse<
{ error: string } | string,
200 | 400 | 404,
"text" | "json"
>
> {
// JS doesn't allow the use of `delete` as a variable name
const delete_ = this.body as unknown as VersiaDelete;
const toDelete = delete_.deleted;
switch (delete_.deleted_type) {
case "Note": {
const note = await Note.fromSql(
eq(Notes.uri, toDelete),
eq(Notes.authorId, this.sender.id),
);
if (!note) {
return this.context.json(
{
error: "Note to delete not found or not owned by sender",
},
404,
);
}
await note.delete();
return this.context.text("Note deleted", 200);
}
case "User": {
const userToDelete = await User.resolve(toDelete);
if (!userToDelete) {
return this.context.json(
{ error: "User to delete not found" },
404,
);
}
if (userToDelete.id === this.sender.id) {
await this.sender.delete();
return this.context.text(
"Account deleted, goodbye 👋",
200,
);
}
return this.context.json(
{
error: "Cannot delete other users than self",
},
400,
);
}
case "pub.versia:likes/Like": {
const like = await Like.fromSql(
eq(Likes.uri, toDelete),
eq(Likes.likerId, this.sender.id),
);
if (!like) {
return this.context.json(
{ error: "Like not found or not owned by sender" },
404,
);
}
await like.delete();
return this.context.text("Like deleted", 200);
}
default: {
return this.context.json(
{
error: `Deletion of object ${toDelete} not implemented`,
},
400,
);
}
}
}
/**
* Handles Like entity processing.
*
* @returns {Promise<Response>} - The response.
*/
private async processLikeRequest(): Promise<
Response &
TypedResponse<
{ error: string } | string,
200 | 404,
"text" | "json"
>
> {
const like = this.body as unknown as VersiaLikeExtension;
const author = await User.resolve(like.author);
const likedNote = await Note.resolve(like.liked);
if (!author) {
return this.context.json({ error: "Author not found" }, 404);
}
if (!likedNote) {
return this.context.json({ error: "Liked Note not found" }, 404);
}
await author.like(likedNote, like.uri);
return this.context.text("Like created", 200);
}
/**
* Handles User entity processing (profile edits).
*
* @returns {Promise<Response>} - The response.
*/
private async processUserRequest(): Promise<
Response &
TypedResponse<
{ error: string } | string,
200 | 500,
"text" | "json"
>
> {
const user = this.body as unknown as VersiaUser;
// FIXME: Instead of refetching the remote user, we should read the incoming json and update from that
const updatedAccount = await User.saveFromRemote(user.uri);
if (!updatedAccount) {
return this.context.json({ error: "Failed to update user" }, 500);
}
return this.context.text("User updated", 200);
}
/**
* Processes Errors into the appropriate HTTP response.
*
* @param {Error} e - The error object.
* @returns {Response} - The error response.
*/
private handleError(e: Error):
| (Response &
TypedResponse<
{
error: string;
error_description: string;
},
400,
"json"
>)
| (Response &
TypedResponse<
{
error: string;
message: string;
},
500,
"json"
>) {
if (isValidationError(e)) {
return this.context.json(
{
error: "Failed to process request",
error_description: (e as ValidationError).message,
},
400,
);
}
this.logger.error`${e}`;
sentry?.captureException(e);
return this.context.json(
{
error: "Failed to process request",
message: (e as Error).message,
},
500,
);
}
}

View file

@ -8,3 +8,5 @@ export { Note } from "~/classes/database/note";
export { Timeline } from "~/classes/database/timeline";
export { Application } from "~/classes/database/application";
export { db } from "~/drizzle/db";
export { Relationship } from "~/classes/database/relationship";
export { Like } from "~/classes/database/like";