feat(communication): federate tenant collaboration with signed delivery
This commit is contained in:
448
backend/src/modules/collaboration/federation.ts
Normal file
448
backend/src/modules/collaboration/federation.ts
Normal file
@@ -0,0 +1,448 @@
|
||||
import { generateKeyPairSync, randomUUID } from "node:crypto"
|
||||
import { sql } from "drizzle-orm"
|
||||
import type { FastifyInstance } from "fastify"
|
||||
import { CollaborationService, rows } from "./service"
|
||||
import { fail, hash, overview, Principal, sequence, text, UserContext, uuid } from "./types"
|
||||
import { FederationKey, Headers, instanceUrl, safeTransport, signedHeaders, Transport, validatePublicKey, verifyHeaders } from "./transport"
|
||||
|
||||
type Options = {
|
||||
apiUrl: string; enabled: boolean;
|
||||
seal: (value: string) => any; unseal: (value: any) => string;
|
||||
transport?: Transport;
|
||||
}
|
||||
|
||||
export class FederationService {
|
||||
readonly origin: string
|
||||
readonly transport: Transport
|
||||
private keyPromise?: Promise<FederationKey & { instanceId: string }>
|
||||
private syncing = new Map<string, Promise<any>>()
|
||||
private working = false
|
||||
constructor(readonly service: CollaborationService, readonly options: Options) {
|
||||
this.origin = options.apiUrl.replace(/\/$/, "")
|
||||
this.transport = options.transport || safeTransport
|
||||
}
|
||||
enabled() {
|
||||
if (!this.options.enabled) fail("Der Betreiber muss die Föderation zuerst aktivieren", 503)
|
||||
instanceUrl(this.origin)
|
||||
}
|
||||
async keys() {
|
||||
this.enabled()
|
||||
if (!this.keyPromise) this.keyPromise = (async () => {
|
||||
let [identity] = await rows(this.service.db, sql`SELECT * FROM collaboration_federation_identity WHERE singleton`)
|
||||
if (!identity) {
|
||||
const pair = generateKeyPairSync("ed25519")
|
||||
const pub = pair.publicKey.export({ format: "pem", type: "spki" }).toString()
|
||||
const secret = this.options.seal(pair.privateKey.export({ format: "pem", type: "pkcs8" }).toString())
|
||||
await this.service.db.execute(sql`INSERT INTO collaboration_federation_identity(instance_id,key_id,public_key,private_key_cipher)
|
||||
VALUES (${randomUUID()},${randomUUID()},${pub},${JSON.stringify(secret)}::jsonb) ON CONFLICT DO NOTHING`)
|
||||
;[identity] = await rows(this.service.db, sql`SELECT * FROM collaboration_federation_identity WHERE singleton`)
|
||||
}
|
||||
return { instanceId: identity.instance_id, keyId: identity.key_id, publicKey: identity.public_key, privateKey: this.options.unseal(identity.private_key_cipher) }
|
||||
})().catch(error => { this.keyPromise = undefined; throw error })
|
||||
return this.keyPromise
|
||||
}
|
||||
async discovery() {
|
||||
const keys = await this.keys()
|
||||
return { version: 1, origin: this.origin, instanceId: keys.instanceId, keyId: keys.keyId, publicKey: keys.publicKey }
|
||||
}
|
||||
contact(principal: Principal) { return `${this.origin}/federation/actors/${principal.id}` }
|
||||
async allowed(tenantId: number, origin: string, db: any = this.service.db) {
|
||||
const [partner] = await rows(db, sql`SELECT * FROM collaboration_partners WHERE tenant_id=${tenantId} AND origin=${origin} AND active`)
|
||||
if (!partner) fail("Partnerinstanz ist für diesen Mandanten nicht freigegeben", 403)
|
||||
return partner
|
||||
}
|
||||
async settings(ctx: UserContext, input?: any) {
|
||||
if (!ctx.manage || !(await this.service.currentUser(ctx.userId, ctx.tenantId)).manage) fail("Keine Berechtigung", 403)
|
||||
if (input) {
|
||||
this.enabled()
|
||||
const origin = instanceUrl(text(input.origin, 1000))
|
||||
if (origin === this.origin) fail("Diese Instanz benötigt keine Partnerfreigabe")
|
||||
if (typeof input.active !== "boolean") fail("Ungültiger Partnerstatus")
|
||||
if (!input.active) {
|
||||
await this.service.db.execute(sql`UPDATE collaboration_partners SET active=false,updated_at=now() WHERE tenant_id=${ctx.tenantId} AND origin=${origin}`)
|
||||
} else {
|
||||
await this.service.rateLimit(`partner-discovery:${ctx.tenantId}`, 10, 600)
|
||||
const data = await this.transport(`${origin}/.well-known/fedeo-federation`, "GET", {})
|
||||
if (data.version !== 1 || data.origin !== origin) fail("Inkompatible Instanzadresse oder Protokollversion")
|
||||
const instanceId = uuid(data.instanceId), keyId = uuid(data.keyId), publicKey = validatePublicKey(data.publicKey)
|
||||
const existing = await rows(this.service.db, sql`SELECT instance_id,key_id,public_key FROM collaboration_partners WHERE origin=${origin}`)
|
||||
if (existing.some(p => p.instance_id !== instanceId || p.key_id !== keyId || p.public_key !== publicKey)) fail("Instanzschlüssel hat sich geändert. Eine erneute Vertrauensprüfung durch den Betreiber ist erforderlich.", 409)
|
||||
await this.service.db.execute(sql`INSERT INTO collaboration_partners(tenant_id,origin,instance_id,key_id,public_key)
|
||||
VALUES (${ctx.tenantId},${origin},${instanceId},${keyId},${publicKey}) ON CONFLICT(tenant_id,origin) DO UPDATE SET active=true,updated_at=now()`)
|
||||
}
|
||||
}
|
||||
return { enabled: this.options.enabled, origin: this.origin, partners: (await rows(this.service.db,
|
||||
sql`SELECT origin,active,public_key FROM collaboration_partners WHERE tenant_id=${ctx.tenantId} ORDER BY origin`)).map(p => ({ origin: p.origin, active: p.active, fingerprint: hash(p.public_key) })) }
|
||||
}
|
||||
async call(tenantId: number, origin: string, action: string, payload: any) {
|
||||
this.enabled()
|
||||
await this.allowed(tenantId, origin)
|
||||
const body = JSON.stringify({ version: 1, action, ...payload })
|
||||
const target = `${origin}/federation/v1/rpc`
|
||||
const result = await this.transport(target, "POST", signedHeaders(await this.keys(), this.origin, target, body), body)
|
||||
if (result?.version !== 1 || !("data" in result)) fail("Inkompatible Föderationsantwort", 502)
|
||||
return result.data
|
||||
}
|
||||
async authenticate(body: string, headers: Headers) {
|
||||
this.enabled()
|
||||
const origin = instanceUrl(headers["x-fedeo-origin"] || "")
|
||||
const partners = await rows(this.service.db, sql`SELECT DISTINCT key_id,public_key FROM collaboration_partners WHERE origin=${origin} AND active`)
|
||||
if (partners.length !== 1) fail("Unbekannte oder gesperrte Partnerinstanz", 401)
|
||||
const verified = verifyHeaders({ keyId: partners[0].key_id, publicKey: partners[0].public_key }, `${this.origin}/federation/v1/rpc`, body, headers)
|
||||
const nonce = uuid(verified.nonce)
|
||||
const [accepted] = await rows(this.service.db, sql`INSERT INTO collaboration_federation_nonces(origin,nonce,expires_at)
|
||||
VALUES (${origin},${nonce},now()+interval '10 minutes') ON CONFLICT DO NOTHING RETURNING nonce`)
|
||||
if (!accepted) fail("Signatur wurde bereits verwendet", 401)
|
||||
await this.service.rateLimit(`federation:${origin}`, 600, 60)
|
||||
return origin
|
||||
}
|
||||
async resolveContact(ctx: UserContext, address: string): Promise<Principal> {
|
||||
const localPrefix = `${this.origin}/federation/actors/`
|
||||
if (address.startsWith(localPrefix)) {
|
||||
const actorId = uuid(address.slice(localPrefix.length))
|
||||
const [principal] = await rows(this.service.db, sql`SELECT * FROM collaboration_principals WHERE id=${actorId} AND kind='user'`)
|
||||
if (!principal) fail("Kontakt nicht gefunden", 404)
|
||||
return this.service.assertPrincipal(principal)
|
||||
}
|
||||
const match = /^(https:\/\/.+)\/federation\/actors\/([0-9a-f-]{36})$/i.exec(address)
|
||||
if (!match) fail("Ungültige persönliche Kontaktadresse")
|
||||
const origin = instanceUrl(match[1]), actorId = uuid(match[2])
|
||||
const actor = await this.call(ctx.tenantId, origin, "actor.lookup", { actorId })
|
||||
if (uuid(actor.id) !== actorId) fail("Kontaktidentität stimmt nicht überein", 502)
|
||||
const name = text(actor.name), organisation = text(actor.organisation)
|
||||
const [principal] = await rows(this.service.db, sql`INSERT INTO collaboration_principals(kind,remote_origin,remote_actor_id,display_name)
|
||||
VALUES ('remote',${origin},${actorId},${`${name} · ${organisation}`}) ON CONFLICT(remote_origin,remote_actor_id) WHERE kind='remote'
|
||||
DO UPDATE SET display_name=excluded.display_name RETURNING *`)
|
||||
return principal
|
||||
}
|
||||
async enqueue(db: any, entry: { key: string; ctx: UserContext; origin: string; action: string; payload: any; roomId?: string; bindingId?: string }) {
|
||||
const [job] = await rows(db, sql`INSERT INTO collaboration_outbox(dedup_key,tenant_id,principal_id,origin,action,payload,room_id,binding_id)
|
||||
VALUES (${entry.key},${entry.ctx.tenantId},${entry.ctx.principal.id},${entry.origin},${entry.action},${JSON.stringify(entry.payload)}::jsonb,${entry.roomId || null},${entry.bindingId || null})
|
||||
ON CONFLICT(dedup_key) DO UPDATE SET dedup_key=excluded.dedup_key RETURNING *`)
|
||||
if (JSON.stringify(job.payload) !== JSON.stringify(entry.payload)) {
|
||||
// JSONB key order differs: compare the actual command values instead of its serialization.
|
||||
if (job.payload.input?.text !== entry.payload.input?.text || job.payload.actorId !== entry.payload.actorId || job.payload.roomId !== entry.payload.roomId || job.payload.input?.revision !== entry.payload.input?.revision) fail("Auftrags-ID wurde bereits anders verwendet", 409)
|
||||
}
|
||||
return job
|
||||
}
|
||||
async invite(ctx: UserContext, roomId: string, target: Principal, invitationId: string, db: any = this.service.db) {
|
||||
this.enabled()
|
||||
await this.allowed(ctx.tenantId, target.remote_origin!, db)
|
||||
await this.enqueue(db, { key: `invite:${invitationId}`, ctx, origin: target.remote_origin!, roomId,
|
||||
action: "invitation.deliver", payload: { actorId: ctx.principal.id, targetActorId: target.remote_actor_id, invitationId } })
|
||||
}
|
||||
async deliveries(ctx: UserContext, roomId: string) {
|
||||
await this.service.manager(this.service.db, ctx, await this.service.roomRow(this.service.db, roomId))
|
||||
return rows(this.service.db, sql`SELECT payload->>'invitationId' AS id,status,last_error AS error FROM collaboration_outbox
|
||||
WHERE tenant_id=${ctx.tenantId} AND room_id=${roomId} AND action='invitation.deliver' ORDER BY created_at DESC LIMIT 100`)
|
||||
}
|
||||
async retryInvitation(ctx: UserContext, roomId: string, invitationId: string) {
|
||||
await this.service.manager(this.service.db, ctx, await this.service.roomRow(this.service.db, roomId))
|
||||
const [job] = await rows(this.service.db, sql`UPDATE collaboration_outbox o SET status='pending',attempts=0,next_attempt_at=now(),last_error=null
|
||||
FROM collaboration_invitations i WHERE i.id=${uuid(invitationId)} AND i.room_id=${roomId} AND i.status='pending' AND i.expires_at>now()
|
||||
AND o.dedup_key='invite:' || i.id::text AND o.tenant_id=${ctx.tenantId} AND o.room_id=${roomId} AND o.status='failed' RETURNING o.*`)
|
||||
if (!job) fail("Einladung kann nicht erneut zugestellt werden", 409)
|
||||
await this.deliver(job)
|
||||
return { queued: true }
|
||||
}
|
||||
async changed(roomId: string, db: any) {
|
||||
if (!this.options.enabled) return
|
||||
const [room] = await rows(db, sql`SELECT r.tenant_id,r.created_by,p.id AS actor_id FROM collaboration_rooms r
|
||||
JOIN collaboration_principals p ON p.tenant_id=r.tenant_id AND p.user_id=r.created_by AND p.kind='user' WHERE r.id=${roomId}`)
|
||||
if (!room) return
|
||||
const members = await rows(db, sql`SELECT p.* FROM collaboration_members m JOIN collaboration_principals p ON p.id=m.principal_id
|
||||
WHERE m.room_id=${roomId} AND p.kind='remote'`)
|
||||
for (const p of members) await db.execute(sql`INSERT INTO collaboration_outbox(dedup_key,tenant_id,principal_id,origin,action,payload,room_id)
|
||||
VALUES (${`change:${randomUUID()}`},${room.tenant_id},${room.actor_id},${p.remote_origin},'room.changed',
|
||||
${JSON.stringify({ roomId, targetActorId: p.remote_actor_id })}::jsonb,${roomId})`)
|
||||
const withdrawn = await rows(db, sql`SELECT i.id,p.remote_origin,p.remote_actor_id FROM collaboration_invitations i
|
||||
JOIN collaboration_principals p ON p.id=i.target_principal_id WHERE i.room_id=${roomId} AND i.status='revoked' AND p.kind='remote'`)
|
||||
for (const i of withdrawn) await db.execute(sql`INSERT INTO collaboration_outbox(dedup_key,tenant_id,principal_id,origin,action,payload,room_id)
|
||||
VALUES (${`withdraw:${i.id}`},${room.tenant_id},${room.actor_id},${i.remote_origin},'invitation.withdrawn',
|
||||
${JSON.stringify({ invitationId: i.id, targetActorId: i.remote_actor_id })}::jsonb,${roomId}) ON CONFLICT(dedup_key) DO NOTHING`)
|
||||
}
|
||||
async rpc(origin: string, request: any) {
|
||||
if (!request || request.version !== 1 || typeof request.action !== "string") fail("Protokollversion nicht unterstützt")
|
||||
const action = request.action
|
||||
if (["actor.lookup", "invitation.deliver", "invitation.withdrawn", "room.changed"].includes(action)) {
|
||||
const actorId = uuid(action === "actor.lookup" ? request.actorId : request.targetActorId)
|
||||
const [principal] = await rows(this.service.db, sql`SELECT * FROM collaboration_principals WHERE id=${actorId} AND kind='user'`)
|
||||
if (!principal) fail("Kontakt nicht gefunden", 404)
|
||||
await this.allowed(Number(principal.tenant_id), origin)
|
||||
await this.service.assertPrincipal(principal)
|
||||
if (action === "actor.lookup") {
|
||||
const [tenant] = await rows(this.service.db, sql`SELECT name FROM tenants WHERE id=${principal.tenant_id}`)
|
||||
return { id: principal.id, name: principal.display_name, organisation: tenant.name }
|
||||
}
|
||||
if (action === "room.changed") {
|
||||
await this.service.db.execute(sql`UPDATE collaboration_remote_bindings SET sync_at=null
|
||||
WHERE principal_id=${principal.id} AND origin=${origin} AND room_id=${uuid(request.roomId)} AND status='active'`)
|
||||
return { received: true }
|
||||
}
|
||||
if (action === 'invitation.withdrawn') {
|
||||
const [binding] = await rows(this.service.db, sql`SELECT * FROM collaboration_remote_bindings WHERE principal_id=${principal.id}
|
||||
AND origin=${origin} AND invitation_id=${uuid(request.invitationId)} AND status='pending'`)
|
||||
if (binding) await this.revokeBinding(binding)
|
||||
return { received: true }
|
||||
}
|
||||
const offer = request.offer
|
||||
if (!offer || !["participant", "reader"].includes(offer.role) || typeof offer.overviewAllowed !== "boolean") fail("Ungültige Einladung")
|
||||
const invitationId = uuid(request.invitationId), roomId = uuid(offer.roomId)
|
||||
const snapshot = { name: text(offer.name), topic: text(offer.topic, 1000, true), ownerName: text(offer.ownerName),
|
||||
role: offer.role, overviewAllowed: offer.overviewAllowed, expiresAt: new Date(offer.expiresAt).toISOString(),
|
||||
createdAt: new Date(offer.createdAt).toISOString(), generation: sequence(offer.generation) }
|
||||
if (!snapshot.generation) fail("Einladungsgeneration fehlt")
|
||||
if (new Date(snapshot.expiresAt).getTime() <= Date.now()) fail("Einladung ist abgelaufen", 410)
|
||||
await this.service.db.transaction(async db => {
|
||||
const [updated] = await rows(db, sql`INSERT INTO collaboration_remote_bindings(tenant_id,principal_id,origin,room_id,invitation_id,invitation)
|
||||
VALUES (${principal.tenant_id},${principal.id},${origin},${roomId},${invitationId},${JSON.stringify(snapshot)}::jsonb)
|
||||
ON CONFLICT(principal_id,origin,room_id) DO UPDATE SET invitation_id=excluded.invitation_id,invitation=excluded.invitation,
|
||||
status='pending',snapshot=null,revision=0,last_read_seq=0,sync_seq=0
|
||||
WHERE collaboration_remote_bindings.invitation_id<>excluded.invitation_id
|
||||
AND coalesce((collaboration_remote_bindings.invitation->>'generation')::bigint,0)<${snapshot.generation} RETURNING id`)
|
||||
if (updated) {
|
||||
await db.execute(sql`DELETE FROM collaboration_remote_messages WHERE binding_id=${updated.id}`)
|
||||
await db.execute(sql`UPDATE collaboration_outbox SET status='failed',last_error='Neue Teilnahme erforderlich' WHERE binding_id=${updated.id} AND status='pending'`)
|
||||
}
|
||||
})
|
||||
return { received: true }
|
||||
}
|
||||
const actorId = uuid(request.actorId)
|
||||
const [principal] = await rows(this.service.db, sql`SELECT * FROM collaboration_principals WHERE kind='remote' AND remote_origin=${origin} AND remote_actor_id=${actorId}`)
|
||||
if (!principal) fail("Teilnehmer nicht eingeladen", 403)
|
||||
if (action === "invitation.accept" || action === "invitation.decline") {
|
||||
const [invite] = await rows(this.service.db, sql`SELECT i.*,r.tenant_id FROM collaboration_invitations i JOIN collaboration_rooms r ON r.id=i.room_id WHERE i.id=${uuid(request.invitationId)} AND i.target_principal_id=${principal.id}`)
|
||||
if (!invite) fail("Einladung nicht gefunden", 404)
|
||||
await this.allowed(Number(invite.tenant_id), origin)
|
||||
if (action === "invitation.decline") return this.service.decline(principal, invite.id)
|
||||
return this.service.accept(principal, invite.id)
|
||||
}
|
||||
const room = await this.service.roomRow(this.service.db, uuid(request.roomId))
|
||||
await this.allowed(Number(room.tenant_id), origin)
|
||||
if (action === "room.get") return this.service.room(principal, room.id)
|
||||
if (action === "messages.get") return this.service.messages(principal, room.id, request.input)
|
||||
if (action === "messages.send") return this.service.send(principal, room.id, request.input)
|
||||
if (action === "members.get") return { members: await this.service.members(principal, room.id) }
|
||||
if (action === "read") return this.service.markRead(principal, room.id, request.input?.seq)
|
||||
fail("Unbekannte Föderationsaktion")
|
||||
}
|
||||
async list(ctx: UserContext) {
|
||||
if (!this.options.enabled) return { rooms: [], invitations: [] }
|
||||
const bindings = await rows(this.service.db, sql`SELECT b.* FROM collaboration_remote_bindings b JOIN collaboration_partners p ON p.tenant_id=b.tenant_id AND p.origin=b.origin AND p.active
|
||||
WHERE b.principal_id=${ctx.principal.id} AND b.tenant_id=${ctx.tenantId} AND b.status IN ('active','pending')`)
|
||||
const result: { rooms: any[]; invitations: any[] } = { rooms: [], invitations: [] }
|
||||
for (const b of bindings) {
|
||||
if (b.status === "pending" && new Date(b.invitation.expiresAt).getTime() > Date.now()) result.invitations.push({ ...b.invitation, id: b.id, remote: true })
|
||||
if (b.status === "active") {
|
||||
const [unread] = await rows(this.service.db, sql`SELECT count(*)::int AS count FROM collaboration_remote_messages WHERE binding_id=${b.id} AND seq>${b.last_read_seq} AND (payload->>'own')='false'`)
|
||||
result.rooms.push({ id: b.id, name: b.snapshot?.name || b.invitation.name, topic: b.snapshot?.topic || b.invitation.topic,
|
||||
ownerName: b.invitation.ownerName, status: b.snapshot?.status || 'active', remote: true, unread: unread.count })
|
||||
}
|
||||
}
|
||||
return result
|
||||
}
|
||||
async binding(ctx: UserContext, bindingId: string) {
|
||||
const [binding] = await rows(this.service.db, sql`SELECT * FROM collaboration_remote_bindings WHERE id=${uuid(bindingId)} AND principal_id=${ctx.principal.id} AND tenant_id=${ctx.tenantId}`)
|
||||
return binding
|
||||
}
|
||||
async accept(ctx: UserContext, id: string, decline = false) {
|
||||
const binding = await this.binding(ctx, id)
|
||||
if (!binding) return { found: false }
|
||||
this.enabled()
|
||||
if (!['pending', 'active'].includes(binding.status)) fail("Einladung ist nicht mehr gültig", 410)
|
||||
const result = await this.call(ctx.tenantId, binding.origin, decline ? "invitation.decline" : "invitation.accept", { actorId: ctx.principal.id, invitationId: binding.invitation_id })
|
||||
if (!decline && uuid(result.roomId) !== binding.room_id) fail("Falscher Unterraum in Föderationsantwort", 502)
|
||||
await this.service.db.transaction(async db => {
|
||||
const [updated] = await rows(db, sql`UPDATE collaboration_remote_bindings SET status=${decline ? "declined" : "active"},sync_at=null
|
||||
WHERE id=${binding.id} AND invitation_id=${binding.invitation_id} AND status IN ('pending','active') RETURNING id`)
|
||||
if (!updated) fail("Einladung ist nicht mehr aktuell", 410)
|
||||
if (!decline && binding.status === 'pending') {
|
||||
await db.execute(sql`DELETE FROM collaboration_remote_messages WHERE binding_id=${binding.id}`)
|
||||
await db.execute(sql`UPDATE collaboration_outbox SET status='failed',last_error='Neue Teilnahme erforderlich' WHERE binding_id=${binding.id} AND status='pending'`)
|
||||
}
|
||||
})
|
||||
if (!decline) await this.sync(ctx, { ...binding, status: 'active' })
|
||||
return { found: true, result: decline ? { declined: true } : { roomId: binding.id } }
|
||||
}
|
||||
async revokeBinding(binding: any) {
|
||||
await this.service.db.transaction(async db => {
|
||||
const [revoked] = await rows(db, sql`UPDATE collaboration_remote_bindings SET status='revoked',snapshot=null WHERE id=${binding.id} AND invitation_id=${binding.invitation_id} RETURNING id`)
|
||||
if (!revoked) return
|
||||
await db.execute(sql`DELETE FROM collaboration_remote_messages WHERE binding_id=${binding.id}`)
|
||||
await db.execute(sql`UPDATE collaboration_outbox SET status='failed',last_error='Zugang entzogen' WHERE binding_id=${binding.id} AND status='pending'`)
|
||||
})
|
||||
}
|
||||
async sync(ctx: UserContext, binding: any) {
|
||||
if (this.syncing.has(binding.id)) return this.syncing.get(binding.id)
|
||||
const operation = this.syncOnce(ctx, binding).finally(() => this.syncing.delete(binding.id))
|
||||
this.syncing.set(binding.id, operation)
|
||||
return operation
|
||||
}
|
||||
async syncOnce(ctx: UserContext, binding: any) {
|
||||
try {
|
||||
const detail = await this.call(ctx.tenantId, binding.origin, "room.get", { actorId: ctx.principal.id, roomId: binding.room_id })
|
||||
if (uuid(detail.id) !== binding.room_id || !['active', 'archived'].includes(detail.status) || typeof detail.canSend !== 'boolean') fail("Ungültige Unterraumantwort", 502)
|
||||
const snapshot = { id: binding.id, name: text(detail.name), topic: text(detail.topic, 1000, true), status: detail.status,
|
||||
canSend: detail.canSend, historyFromSeq: sequence(detail.historyFromSeq), revision: sequence(detail.revision),
|
||||
lastReadSeq: sequence(detail.lastReadSeq), overview: null }
|
||||
const page = await this.call(ctx.tenantId, binding.origin, "messages.get", { actorId: ctx.principal.id, roomId: binding.room_id, input: { after: Math.max(Number(binding.sync_seq || 0), snapshot.historyFromSeq - 1) } })
|
||||
await this.cache(binding, page.messages, snapshot)
|
||||
return { ...snapshot, overview: detail.overview ? overview(detail.overview) : null, offline: false }
|
||||
} catch (error: any) {
|
||||
if ([401,403,404,410].includes(error.statusCode)) { await this.revokeBinding(binding); throw error }
|
||||
const current = await this.binding(ctx, binding.id)
|
||||
if (!current || current.status !== 'active' || current.invitation_id !== binding.invitation_id) fail('Zugang ist nicht mehr aktiv', 410)
|
||||
await this.service.db.execute(sql`UPDATE collaboration_remote_bindings SET last_error='Partnerinstanz nicht erreichbar',sync_at=now() WHERE id=${binding.id} AND invitation_id=${binding.invitation_id}`)
|
||||
if (current.snapshot) return { ...current.snapshot, id: binding.id, overview: null, offline: true }
|
||||
throw error
|
||||
}
|
||||
}
|
||||
async cache(binding: any, messages: any[], snapshot?: any) {
|
||||
if (!Array.isArray(messages) || messages.length > 50) fail("Ungültige Nachrichtenantwort", 502)
|
||||
const projected = messages.map(m => ({ id: uuid(m.id), clientId: uuid(m.clientId), seq: sequence(m.seq), body: text(m.body, 10000),
|
||||
senderName: text(m.senderName, 400), senderKind: ['user','guest','remote'].includes(m.senderKind) ? m.senderKind : 'remote',
|
||||
own: m.own === true, createdAt: new Date(m.createdAt).toISOString() }))
|
||||
await this.service.db.transaction(async db => {
|
||||
const [current] = await rows(db, sql`SELECT * FROM collaboration_remote_bindings WHERE id=${binding.id} FOR UPDATE`)
|
||||
if (current.status !== 'active' || current.invitation_id !== binding.invitation_id) fail("Zugang ist nicht mehr aktiv", 410)
|
||||
if (snapshot && Number(current.revision) > snapshot.revision) fail('Teilnahmerechte haben sich geändert', 409)
|
||||
if (snapshot) {
|
||||
await db.execute(sql`UPDATE collaboration_remote_bindings SET snapshot=${JSON.stringify(snapshot)}::jsonb,revision=${snapshot.revision},
|
||||
sync_at=now(),last_error=null,sync_seq=GREATEST(sync_seq,${Math.max(0, ...projected.map(m => m.seq))}),last_read_seq=GREATEST(last_read_seq,${snapshot.lastReadSeq}) WHERE id=${binding.id}`)
|
||||
await db.execute(sql`DELETE FROM collaboration_remote_messages WHERE binding_id=${binding.id} AND seq<${snapshot.historyFromSeq}`)
|
||||
}
|
||||
const from = snapshot?.historyFromSeq || current.snapshot?.historyFromSeq || 0
|
||||
for (const message of projected.filter(m => m.seq >= from)) await db.execute(sql`INSERT INTO collaboration_remote_messages(binding_id,id,seq,payload)
|
||||
VALUES (${binding.id},${message.id},${message.seq},${JSON.stringify(message)}::jsonb)
|
||||
ON CONFLICT(binding_id,id) DO UPDATE SET payload=excluded.payload`)
|
||||
})
|
||||
}
|
||||
async memberRequest(ctx: UserContext, id: string, action: string, input: any = {}) {
|
||||
let binding = await this.binding(ctx, id)
|
||||
if (!binding) return { found: false }
|
||||
this.enabled()
|
||||
if (binding.status !== 'active') fail("Unterraum nicht freigegeben", 404)
|
||||
await this.allowed(ctx.tenantId, binding.origin)
|
||||
if (action === "send") {
|
||||
if (!binding.snapshot?.canSend) fail("Unterraum ist nicht zum Schreiben freigegeben", 403)
|
||||
const command = { actorId: ctx.principal.id, roomId: binding.room_id,
|
||||
input: { clientId: uuid(input.clientId), text: text(input.text, 10000), revision: Number(binding.revision) } }
|
||||
const job = await this.enqueue(this.service.db, { key: `message:${binding.id}:${command.input.clientId}`, ctx, origin: binding.origin,
|
||||
action: 'messages.send', payload: command, bindingId: binding.id })
|
||||
await this.deliver(job)
|
||||
const [message] = await rows(this.service.db, sql`SELECT payload FROM collaboration_remote_messages WHERE binding_id=${binding.id} AND payload->>'clientId'=${command.input.clientId}`)
|
||||
const [currentJob] = await rows(this.service.db, sql`SELECT * FROM collaboration_outbox WHERE id=${job.id}`)
|
||||
if (currentJob.status === 'failed') fail(currentJob.last_error || "Nachricht konnte nicht gesendet werden", 409)
|
||||
return { found: true, result: message?.payload || this.pendingMessage(currentJob) }
|
||||
}
|
||||
const detail = await this.sync(ctx, binding)
|
||||
binding = await this.binding(ctx, id)
|
||||
if (action === "room") return { found: true, result: detail }
|
||||
if (action === "members") {
|
||||
if (detail.offline) return { found: true, result: { members: [] } }
|
||||
const data = await this.call(ctx.tenantId, binding.origin, 'members.get', { actorId: ctx.principal.id, roomId: binding.room_id })
|
||||
if (!Array.isArray(data.members) || data.members.length > 500) fail("Ungültige Teilnehmerliste", 502)
|
||||
return { found: true, result: { members: data.members.map((m: any) => ({ id: uuid(m.id), name: text(m.name,400), kind: text(m.kind,10), own: m.own === true })) } }
|
||||
}
|
||||
if (action === "read") {
|
||||
const seq = sequence(input.seq)
|
||||
if (!detail.offline) await this.call(ctx.tenantId, binding.origin, 'read', { actorId: ctx.principal.id, roomId: binding.room_id, input: { seq } })
|
||||
const [max] = await rows(this.service.db, sql`SELECT max(seq) AS seq FROM collaboration_remote_messages WHERE binding_id=${binding.id}`)
|
||||
await this.service.db.execute(sql`UPDATE collaboration_remote_bindings SET last_read_seq=GREATEST(last_read_seq,${Math.min(seq,Number(max.seq || 0))}) WHERE id=${binding.id}`)
|
||||
return { found: true, result: { read: true } }
|
||||
}
|
||||
const before = sequence(input.before), after = sequence(input.after)
|
||||
if (before && !detail.offline) {
|
||||
const page = await this.call(ctx.tenantId, binding.origin, 'messages.get', { actorId: ctx.principal.id, roomId: binding.room_id, input: { before } })
|
||||
await this.cache(binding, page.messages)
|
||||
}
|
||||
const page = await rows(this.service.db, sql`SELECT payload FROM collaboration_remote_messages WHERE binding_id=${binding.id}
|
||||
${before ? sql`AND seq<${before}` : sql``} ${after ? sql`AND seq>${after}` : sql``}
|
||||
ORDER BY seq ${after ? sql`ASC` : sql`DESC`} LIMIT 50`)
|
||||
const messages = (after ? page : page.reverse()).map(r => r.payload)
|
||||
if (!before) {
|
||||
const pending = await rows(this.service.db, sql`SELECT * FROM collaboration_outbox WHERE binding_id=${binding.id} AND status IN ('pending','failed') AND (payload->'input'->>'revision')::bigint=${binding.revision} ORDER BY created_at LIMIT 100`)
|
||||
messages.push(...pending.map(job => this.pendingMessage(job)))
|
||||
}
|
||||
return { found: true, result: { messages, hasMore: page.length === 50, offline: detail.offline } }
|
||||
}
|
||||
pendingMessage(job: any) {
|
||||
return { id: job.payload.input.clientId, clientId: job.payload.input.clientId, seq: null, body: job.payload.input.text,
|
||||
createdAt: job.created_at, own: true, senderName: 'Du', senderKind: 'user', pending: job.status === 'pending', failed: job.status === 'failed' }
|
||||
}
|
||||
async deliver(job: any) {
|
||||
const [claimed] = await rows(this.service.db, sql`UPDATE collaboration_outbox SET lease_until=now()+interval '30 seconds',attempts=attempts+1
|
||||
WHERE id=${job.id} AND status='pending' AND (lease_until IS NULL OR lease_until<now()) RETURNING *`)
|
||||
if (!claimed) return
|
||||
try {
|
||||
let payload = claimed.payload
|
||||
if (claimed.action === 'invitation.deliver') {
|
||||
const [invitation] = await rows(this.service.db, sql`SELECT i.*,r.name,r.topic,t.name AS owner_name,r.status AS room_status,p.archived,
|
||||
(SELECT id FROM collaboration_audit WHERE subject_id=i.id AND action='invitation.created' LIMIT 1) AS generation
|
||||
FROM collaboration_invitations i JOIN collaboration_rooms r ON r.id=i.room_id JOIN tenants t ON t.id=r.tenant_id
|
||||
LEFT JOIN projects p ON p.id=r.project_id WHERE i.id=${claimed.payload.invitationId}`)
|
||||
if (!invitation || invitation.status !== 'pending' || new Date(invitation.expires_at).getTime() <= Date.now() || invitation.room_status !== 'active' || invitation.archived) fail("Einladung ist nicht mehr aktiv",410)
|
||||
const inviter = await this.service.currentUser(invitation.invited_by, Number(claimed.tenant_id))
|
||||
if (!inviter.manage) fail("Einladung ist nicht mehr gültig",410)
|
||||
payload = { ...payload, offer: { roomId: invitation.room_id, name: invitation.name, topic: invitation.topic, ownerName: invitation.owner_name,
|
||||
role: invitation.role, overviewAllowed: invitation.overview_allowed, expiresAt: invitation.expires_at, createdAt: invitation.created_at, generation: Number(invitation.generation) } }
|
||||
}
|
||||
if (claimed.action === 'messages.send') {
|
||||
const [principal] = await rows(this.service.db, sql`SELECT * FROM collaboration_principals WHERE id=${claimed.principal_id}`)
|
||||
await this.service.assertPrincipal(principal)
|
||||
const [binding] = await rows(this.service.db, sql`SELECT * FROM collaboration_remote_bindings WHERE id=${claimed.binding_id} AND status='active'`)
|
||||
if (!binding || Number(binding.revision) !== payload.input.revision) fail("Teilnahmerechte haben sich geändert",409)
|
||||
}
|
||||
const result = await this.call(Number(claimed.tenant_id), claimed.origin, claimed.action, payload)
|
||||
if (claimed.action === 'messages.send') {
|
||||
const [binding] = await rows(this.service.db, sql`SELECT * FROM collaboration_remote_bindings WHERE id=${claimed.binding_id}`)
|
||||
await this.cache(binding, [result])
|
||||
}
|
||||
await this.service.db.execute(sql`UPDATE collaboration_outbox SET status='sent',lease_until=null,last_error=null WHERE id=${claimed.id}`)
|
||||
} catch (error: any) {
|
||||
const terminal = (error.statusCode >= 400 && error.statusCode < 500 && error.statusCode !== 429) || claimed.attempts >= 20
|
||||
await this.service.db.execute(sql`UPDATE collaboration_outbox SET status=${terminal ? 'failed' : 'pending'},lease_until=null,
|
||||
last_error=${String(error.message || 'Zustellung fehlgeschlagen').slice(0,300)},
|
||||
next_attempt_at=now()+${Math.min(3600, 2 ** Math.min(claimed.attempts,12)) + Math.floor(Math.random()*5)}*interval '1 second' WHERE id=${claimed.id}`)
|
||||
}
|
||||
}
|
||||
async tick() {
|
||||
if (this.working) return
|
||||
this.working = true
|
||||
try {
|
||||
if (this.options.enabled) {
|
||||
const jobs = await rows(this.service.db, sql`SELECT * FROM collaboration_outbox WHERE status='pending' AND next_attempt_at<=now()
|
||||
AND (lease_until IS NULL OR lease_until<now()) ORDER BY next_attempt_at LIMIT 3`)
|
||||
await Promise.allSettled(jobs.map(job => this.deliver(job)))
|
||||
const bindings = await rows(this.service.db, sql`SELECT b.*,p.user_id FROM collaboration_remote_bindings b
|
||||
JOIN collaboration_principals p ON p.id=b.principal_id
|
||||
WHERE b.status='active' AND (b.sync_at IS NULL OR b.sync_at<now()-interval '10 seconds') ORDER BY b.sync_at NULLS FIRST LIMIT 3`)
|
||||
await Promise.allSettled(bindings.map(async b => {
|
||||
try { await this.sync(await this.service.userContext(b.user_id, Number(b.tenant_id)), b) }
|
||||
catch (error: any) { if ([401,403,404,410].includes(error.statusCode)) await this.revokeBinding(b) }
|
||||
}))
|
||||
}
|
||||
await this.service.db.execute(sql`DELETE FROM collaboration_federation_nonces WHERE expires_at<now()`)
|
||||
await this.service.db.execute(sql`DELETE FROM collaboration_guest_challenges WHERE expires_at<now()-interval '1 day'`)
|
||||
await this.service.db.execute(sql`DELETE FROM collaboration_guest_sessions WHERE expires_at<now()`)
|
||||
await this.service.db.execute(sql`DELETE FROM collaboration_rate_limits WHERE resets_at<now()-interval '1 day'`)
|
||||
} finally { this.working = false }
|
||||
}
|
||||
}
|
||||
|
||||
export async function federationRoutes(server: FastifyInstance, { federation }: { federation: FederationService }) {
|
||||
server.removeContentTypeParser('application/json')
|
||||
server.addContentTypeParser('application/json', { parseAs: 'string', bodyLimit: 128 * 1024 }, (req: any, body: string, done) => {
|
||||
req.federationBody = body
|
||||
try { done(null, JSON.parse(body)) } catch { done(Object.assign(new Error('Ungültiges JSON'), { statusCode: 400 })) }
|
||||
})
|
||||
server.get('/.well-known/fedeo-federation', async () => federation.discovery())
|
||||
server.post('/federation/v1/rpc', { bodyLimit: 128 * 1024 }, async (req: any, reply) => {
|
||||
reply.header('Cache-Control','no-store')
|
||||
// Duplicate signature-bearing fields are rejected rather than silently combined.
|
||||
const names = req.raw.rawHeaders.filter((_: string, index: number) => index % 2 === 0).map((name: string) => name.toLowerCase())
|
||||
for (const header of ['signature','signature-input','content-digest','content-type','x-fedeo-origin']) if (names.filter((name: string) => name === header).length !== 1) fail('Mehrdeutige Signaturfelder',401)
|
||||
const origin = await federation.authenticate(req.federationBody, req.headers)
|
||||
return { version: 1, data: await federation.rpc(origin, req.body) }
|
||||
})
|
||||
}
|
||||
@@ -2,6 +2,8 @@ import type { FastifyInstance } from "fastify"
|
||||
import { secrets } from "../../utils/secrets"
|
||||
import { sendMail } from "../../utils/mailer"
|
||||
import { CollaborationService } from "./service"
|
||||
import { FederationService } from "./federation"
|
||||
import { encrypt, decrypt } from "../../utils/crypt"
|
||||
|
||||
export function collaborationRuntime(server: FastifyInstance) {
|
||||
const apiUrl = String(process.env.FEDEO_FEDERATION_URL || secrets.API_BASE_URL || "").replace(/\/$/, "")
|
||||
@@ -11,5 +13,15 @@ export function collaborationRuntime(server: FastifyInstance) {
|
||||
guestPepper: () => secrets.ENCRYPTION_KEY || "",
|
||||
sendMail,
|
||||
})
|
||||
return { service, apiUrl }
|
||||
const federation = new FederationService(service, { apiUrl, enabled: process.env.FEDEO_FEDERATION_ENABLED === "true", seal: encrypt, unseal: decrypt })
|
||||
service.options.changed = (roomId, db) => federation.changed(roomId, db)
|
||||
service.options.invited = (ctx, roomId, target, invitationId, db) => federation.invite(ctx, roomId, target, invitationId, db)
|
||||
let timer: ReturnType<typeof setInterval>
|
||||
server.addHook("onReady", async () => {
|
||||
if (federation.options.enabled) await federation.keys()
|
||||
timer = setInterval(() => federation.tick().catch(error => server.log.error({ err: error }, "Collaboration worker failed")), 5000)
|
||||
timer.unref()
|
||||
})
|
||||
server.addHook("onClose", async () => { clearInterval(timer) })
|
||||
return { service, federation, apiUrl }
|
||||
}
|
||||
|
||||
@@ -10,6 +10,7 @@ export type CollaborationOptions = {
|
||||
guestPepper: () => string;
|
||||
sendMail: (to: string, subject: string, html: string) => Promise<{ success: boolean }>;
|
||||
changed?: (roomId: string, db: any) => Promise<void>;
|
||||
invited?: (ctx: UserContext, roomId: string, target: Principal, invitationId: string, db: any) => Promise<void>;
|
||||
}
|
||||
|
||||
/** Owner-side authority. All membership changes and sends serialize on the room row. */
|
||||
@@ -184,7 +185,8 @@ export class CollaborationService {
|
||||
const body = text(input.text, 10000), clientId = uuid(input.clientId)
|
||||
return this.db.transaction(async (db) => {
|
||||
const room = await this.roomRow(db, roomId, true)
|
||||
await this.membership(db, room, principal, true)
|
||||
const member = await this.membership(db, room, principal, true)
|
||||
if (principal.kind === "remote" && sequence(input.revision) !== Number(member.revision)) fail("Teilnahmerechte haben sich geändert", 409)
|
||||
const [existing] = await rows(db, sql`SELECT id,client_id AS "clientId",seq::float8 AS seq,body,created_at AS "createdAt" FROM collaboration_messages
|
||||
WHERE room_id=${room.id} AND principal_id=${principal.id} AND client_id=${clientId}`)
|
||||
if (existing) {
|
||||
@@ -271,6 +273,10 @@ export class CollaborationService {
|
||||
VALUES (${room.id},${target.id},${hash(token)},${role},${input.overviewAllowed === true},${ctx.userId},
|
||||
LEAST(now()+interval '7 days',${accessExpires}::timestamptz),${accessExpires}) RETURNING id,expires_at AS "expiresAt"`)
|
||||
await this.audit(db, room, ctx.principal, "invitation.created", invite.id)
|
||||
if (target.kind === "remote") {
|
||||
if (!this.options.invited) fail("Föderation ist nicht konfiguriert", 503)
|
||||
await this.options.invited(ctx, room.id, target, invite.id, db)
|
||||
}
|
||||
return invite
|
||||
})
|
||||
return { ...result, token }
|
||||
|
||||
105
backend/src/modules/collaboration/transport.ts
Normal file
105
backend/src/modules/collaboration/transport.ts
Normal file
@@ -0,0 +1,105 @@
|
||||
import { createHash, createPublicKey, sign, verify, randomUUID } from "node:crypto"
|
||||
import { lookup } from "node:dns/promises"
|
||||
import { BlockList, isIP } from "node:net"
|
||||
import https from "node:https"
|
||||
import { fail } from "./types"
|
||||
|
||||
const covered = '("@method" "@target-uri" "content-type" "content-digest" "x-fedeo-origin")'
|
||||
export type FederationKey = { keyId: string; publicKey: string; privateKey?: string }
|
||||
export type Headers = Record<string, string>
|
||||
export type Transport = (url: string, method: "GET" | "POST", headers: Headers, body?: string) => Promise<any>
|
||||
|
||||
export function instanceUrl(value: string) {
|
||||
let url: URL
|
||||
try { url = new URL(value) } catch { fail("Ungültige Instanzadresse") }
|
||||
if (url.protocol !== "https:" || url.username || url.password || url.search || url.hash
|
||||
|| (url.port && url.port !== "443") || !/^\/(?:[A-Za-z0-9_-]+\/?)*$/.test(url.pathname)) {
|
||||
fail("Die Instanz benötigt eine öffentliche HTTPS-Adresse auf Port 443")
|
||||
}
|
||||
return url.href.replace(/\/$/, "")
|
||||
}
|
||||
|
||||
export const digest = (body: string) => `sha-256=:${createHash("sha256").update(body, "utf8").digest("base64")}:`
|
||||
export function signatureBase(method: string, target: string, headers: Headers, params: string) {
|
||||
return `"@method": ${method}\n"@target-uri": ${target}\n"content-type": ${headers["content-type"]}\n"content-digest": ${headers["content-digest"]}\n"x-fedeo-origin": ${headers["x-fedeo-origin"]}\n"@signature-params": ${params}`
|
||||
}
|
||||
|
||||
/** Deliberately narrow RFC 9421 application profile; no algorithm negotiation or remote key URLs. */
|
||||
export function signedHeaders(key: FederationKey, origin: string, target: string, body: string, now = Math.floor(Date.now() / 1000), nonce = randomUUID()): Headers {
|
||||
const params = `${covered};created=${now};expires=${now + 120};nonce="${nonce}";keyid="${key.keyId}";alg="ed25519"`
|
||||
const headers = { "content-type": "application/json", "content-digest": digest(body), "x-fedeo-origin": origin }
|
||||
const signature = sign(null, Buffer.from(signatureBase("POST", target, headers, params)), key.privateKey!).toString("base64")
|
||||
return { ...headers, "signature-input": `fedeo=${params}`, signature: `fedeo=:${signature}:` }
|
||||
}
|
||||
|
||||
export function verifyHeaders(key: FederationKey, target: string, body: string, headers: Headers, now = Math.floor(Date.now() / 1000)) {
|
||||
const raw = headers["signature-input"] || ""
|
||||
const suffix = raw.slice(`fedeo=${covered}`.length)
|
||||
const match = /^;created=(\d+);expires=(\d+);nonce="([0-9a-f-]{36})";keyid="([0-9a-f-]{36})";alg="ed25519"$/.exec(suffix)
|
||||
if (!raw.startsWith(`fedeo=${covered}`) || !match || match[4] !== key.keyId) fail("Ungültiges Signaturprofil", 401)
|
||||
const created = Number(match[1]), expires = Number(match[2])
|
||||
if (created > now + 30 || created < now - 300 || expires < now || expires > created + 300 || expires <= created) fail("Signatur ist abgelaufen", 401)
|
||||
if (headers["content-type"] !== "application/json" || headers["content-digest"] !== digest(body)) fail("Ungültiger Inhaltsdigest", 401)
|
||||
const signature = /^fedeo=:([A-Za-z0-9+/]+={0,2}):$/.exec(headers.signature || "")
|
||||
if (!signature) fail("Ungültige Signatur", 401)
|
||||
const bytes = Buffer.from(signature[1], "base64")
|
||||
if (bytes.length !== 64 || !verify(null, Buffer.from(signatureBase("POST", target, headers, raw.slice(6))), key.publicKey, bytes)) fail("Ungültige Signatur", 401)
|
||||
return { nonce: match[3], expires }
|
||||
}
|
||||
|
||||
const privateV4 = new BlockList()
|
||||
for (const [address, bits] of [["0.0.0.0",8],["10.0.0.0",8],["100.64.0.0",10],["127.0.0.0",8],["169.254.0.0",16],["172.16.0.0",12],
|
||||
["192.0.0.0",24],["192.0.2.0",24],["192.168.0.0",16],["198.18.0.0",15],["198.51.100.0",24],["203.0.113.0",24],["224.0.0.0",3]] as [string,number][]) privateV4.addSubnet(address,bits)
|
||||
const publicV6 = new BlockList(); publicV6.addSubnet("2000::",3,"ipv6")
|
||||
const specialV6 = new BlockList()
|
||||
for (const [address, bits] of [["2001::",23],["2001:db8::",32],["2002::",16],["3fff::",20]] as [string,number][]) specialV6.addSubnet(address,bits,"ipv6")
|
||||
export function publicAddress(address: string) {
|
||||
return isIP(address) === 4 ? !privateV4.check(address) : isIP(address) === 6 && publicV6.check(address,"ipv6") && !specialV6.check(address,"ipv6")
|
||||
}
|
||||
|
||||
export function validatePublicKey(pem: unknown) {
|
||||
if (typeof pem !== "string" || pem.length > 2000) fail("Ungültiger Instanzschlüssel")
|
||||
try { if (createPublicKey(pem).asymmetricKeyType !== "ed25519") fail("Ed25519-Schlüssel erforderlich") }
|
||||
catch { fail("Ungültiger Instanzschlüssel") }
|
||||
return pem
|
||||
}
|
||||
|
||||
export const safeTransport: Transport = async (target, method, headers, body) => {
|
||||
const url = new URL(target)
|
||||
if (url.protocol !== "https:" || url.username || url.password || (url.port && url.port !== "443")) fail("Unzulässiges Föderationsziel")
|
||||
let timer: ReturnType<typeof setTimeout>
|
||||
const addresses = await Promise.race([
|
||||
lookup(url.hostname, { all: true }),
|
||||
new Promise<never>((_, reject) => { timer = setTimeout(() => reject(Object.assign(new Error("DNS-Zeitüberschreitung"), { statusCode: 503 })), 3000) }),
|
||||
]).finally(() => clearTimeout(timer))
|
||||
if (!addresses.length || addresses.some(item => !publicAddress(item.address))) fail("Föderation darf keine privaten oder reservierten Netzwerkadressen aufrufen", 400)
|
||||
const address = addresses[0]
|
||||
return new Promise((resolve, reject) => {
|
||||
// Pin the checked address on this connection; preserve hostname for TLS verification and SNI.
|
||||
const request = https.request(url, {
|
||||
method, headers: { ...headers, ...(body ? { "content-length": Buffer.byteLength(body).toString() } : {}) },
|
||||
agent: false, lookup: ((_hostname: string, options: any, callback: any) => {
|
||||
if (options?.all) callback(null, [address]); else callback(null, address.address, address.family)
|
||||
}) as any,
|
||||
}, response => {
|
||||
const chunks: Buffer[] = []; let length = 0
|
||||
response.on("data", chunk => {
|
||||
length += chunk.length
|
||||
if (length > 1024 * 1024) request.destroy(new Error("Föderationsantwort zu groß"))
|
||||
else chunks.push(chunk)
|
||||
})
|
||||
response.on("end", () => {
|
||||
const status = response.statusCode || 502
|
||||
// Redirects are intentionally not followed, including during discovery.
|
||||
if (status < 200 || status >= 300) {
|
||||
reject(Object.assign(new Error(status >= 500 ? "Partnerinstanz ist nicht erreichbar" : "Partnerinstanz hat die Anfrage abgelehnt"), { statusCode: status })); return
|
||||
}
|
||||
try { resolve(JSON.parse(Buffer.concat(chunks).toString("utf8"))) } catch { reject(Object.assign(new Error("Ungültige Föderationsantwort"), { statusCode: 502 })) }
|
||||
})
|
||||
})
|
||||
const deadline = setTimeout(() => request.destroy(new Error("Föderations-Zeitüberschreitung")), 6000)
|
||||
request.on("error", error => { clearTimeout(deadline); reject(Object.assign(error, { statusCode: 503 })) })
|
||||
request.on("close", () => clearTimeout(deadline))
|
||||
request.end(body)
|
||||
})
|
||||
}
|
||||
Reference in New Issue
Block a user