From 011687b391c12df4c7066c72591995c9329a3c0c Mon Sep 17 00:00:00 2001 From: Hornwitser Date: Tue, 8 Jul 2025 15:43:14 +0200 Subject: [PATCH] Close event streams for expired sessions When a session expires close any event streams that have been opened with that session. This prevents an attacker with a leaked session cookie from opening a stream and receiving updates indefinitely without being detected. By sending the session the event stream is opened with when the stream is established this closure on session expiry also serves as a way for a user agent to be notified whenever its own access level changes. --- composables/event-source.ts | 5 +++- docs/dev/server-sent-events.md | 13 +++++++++ server/api/admin/user.patch.ts | 4 +++ server/api/auth/account.delete.ts | 29 +++++++++++++------- server/api/auth/session.delete.ts | 3 --- server/api/auth/session.get.ts | 17 ++---------- server/api/events.ts | 4 +-- server/database.ts | 1 + server/streams.ts | 44 ++++++++++++++++++++++++------- server/utils/session.ts | 32 +++++++++++++++++++--- shared/types/api.ts | 12 +++++++++ stores/session.ts | 12 ++++++++- 12 files changed, 132 insertions(+), 44 deletions(-) create mode 100644 docs/dev/server-sent-events.md diff --git a/composables/event-source.ts b/composables/event-source.ts index 379cc07..838b879 100644 --- a/composables/event-source.ts +++ b/composables/event-source.ts @@ -22,8 +22,11 @@ class AppEventSource extends EventTarget { console.log("AppEventSource", event.type, event.data); this.dispatchEvent(new Event(event.type)); } else { - const data = event.data ? JSON.parse(event.data) : undefined; + const data = event.data ? JSON.parse(event.data) as ApiEvent : undefined; console.log("AppEventSource", event.type, data); + if (data?.type === "connected") { + this.#sourceSessionId = data.session?.id; + } this.dispatchEvent(new MessageEvent(event.type, { data, origin: event.origin, diff --git a/docs/dev/server-sent-events.md b/docs/dev/server-sent-events.md new file mode 100644 index 0000000..14c1813 --- /dev/null +++ b/docs/dev/server-sent-events.md @@ -0,0 +1,13 @@ + +# Server-sent events + +To update in real time this application sends a `text/event-source` stream using [Server-sent events](https://developer.mozilla.org/en-US/docs/Web/API/Server-sent_events). These streams use the current session if any to filter restricted resources and ends when the session expires, necessitating a reconnect by the user agent. (If there are no session associated with the connection it ends after the session expiry timeout). + +## Events + +Upon connecting a `"connect"` event is emitted with the session the connection was made under. This is the primary mechanism a user agent discovers its own session having been rotated into a new one, which also happens when the access level of the account associated with the session changes. + +After the `"connect"` event the user agent will start to receive updates to resources it has access to that has changed. There is no filtering for what resoucres the user agent receives updates for at the moment as there's not enough events to justify the complexity of server-side subscriptions and filtering. diff --git a/server/api/admin/user.patch.ts b/server/api/admin/user.patch.ts index bdcf099..42b08ad 100644 --- a/server/api/admin/user.patch.ts +++ b/server/api/admin/user.patch.ts @@ -64,6 +64,10 @@ export default defineEventHandler(async (event) => { for (const session of sessions) { if (session.accountId === user.id) { session.expiresAtMs = 0; + broadcastEvent({ + type: "session-expired", + sessionId: session.id, + }); } } await writeSessions(sessions); diff --git a/server/api/auth/account.delete.ts b/server/api/auth/account.delete.ts index 7506531..f75d067 100644 --- a/server/api/auth/account.delete.ts +++ b/server/api/auth/account.delete.ts @@ -12,16 +12,25 @@ export default defineEventHandler(async (event) => { const serverSession = await requireServerSessionWithUser(event); let users = await readUsers(); - // Remove sessions for this user - const removedSessionIds = new Set(); + // Expire sessions for this user + const expiredSessionIds = new Set(); let sessions = await readSessions(); - sessions = sessions.filter(session => { - if (session.accountId === serverSession.accountId) { - removedSessionIds.add(session.id); - return false; + const nowMs = Date.now(); + for (const session of sessions) { + if ( + !session.finished + && session.successor !== undefined + && session.expiresAtMs < nowMs + && session.accountId === serverSession.accountId + ) { + session.expiresAtMs = nowMs; + broadcastEvent({ + type: "session-expired", + sessionId: session.id, + }); + expiredSessionIds.add(session.id); } - return true; - }); + } cancelAccountStreams(serverSession.accountId); await writeSessions(sessions); await deleteCookie(event, "session"); @@ -29,13 +38,13 @@ export default defineEventHandler(async (event) => { // Remove subscriptions for this user let subscriptions = await readSubscriptions(); subscriptions = subscriptions.filter( - subscription => !removedSessionIds.has(subscription.sessionId) + subscription => !expiredSessionIds.has(subscription.sessionId) ); await writeSubscriptions(subscriptions); // Remove the user const account = users.find(user => user.id === serverSession.accountId)!; - const now = new Date().toISOString(); + const now = new Date(nowMs).toISOString(); account.deleted = true; account.updatedAt = now; await writeUsers(users); diff --git a/server/api/auth/session.delete.ts b/server/api/auth/session.delete.ts index ca24eaf..df4bfd9 100644 --- a/server/api/auth/session.delete.ts +++ b/server/api/auth/session.delete.ts @@ -18,8 +18,5 @@ export default defineEventHandler(async (event) => { } } - if (session) { - cancelSessionStreams(session.id); - } await clearServerSession(event); }) diff --git a/server/api/auth/session.get.ts b/server/api/auth/session.get.ts index f63f355..9ad7a5e 100644 --- a/server/api/auth/session.get.ts +++ b/server/api/auth/session.get.ts @@ -2,23 +2,10 @@ SPDX-FileCopyrightText: © 2025 Hornwitser SPDX-License-Identifier: AGPL-3.0-or-later */ -import { readSubscriptions, readUsers } from "~/server/database"; -import type { ApiSession } from "~/shared/types/api"; - -export default defineEventHandler(async (event): Promise => { +export default defineEventHandler(async event => { const session = await getServerSession(event, false); if (!session) return; - const users = await readUsers(); - const account = users.find(user => user.id === session.accountId); - const subscriptions = await readSubscriptions(); - const push = Boolean( - subscriptions.find(sub => sub.type === "push" && sub.sessionId === session.id) - ); - return { - id: session.id, - account, - push, - }; + return await serverSessionToApi(event, session); }) diff --git a/server/api/events.ts b/server/api/events.ts index 140c52e..12e1f3b 100644 --- a/server/api/events.ts +++ b/server/api/events.ts @@ -24,8 +24,8 @@ export default defineEventHandler(async (event) => { console.log(`cancelled event stream for ${source}`); deleteStream(stream.writable); } - }) - addStream(stream.writable, session?.id, session?.accountId); + }); + addStream(event, stream.writable, session); // Workaround to properly handle stream errors. See https://github.com/unjs/h3/issues/986 setHeader(event, "Access-Control-Allow-Origin", "*"); diff --git a/server/database.ts b/server/database.ts index b369e24..24e30a5 100644 --- a/server/database.ts +++ b/server/database.ts @@ -13,6 +13,7 @@ export interface ServerSession { expiresAtMs: number, discardAtMs: number, successor?: Id, + finished: boolean, }; export interface ServerUser { diff --git a/server/streams.ts b/server/streams.ts index a231e85..7298d73 100644 --- a/server/streams.ts +++ b/server/streams.ts @@ -2,8 +2,10 @@ SPDX-FileCopyrightText: © 2025 Hornwitser SPDX-License-Identifier: AGPL-3.0-or-later */ -import { readUsers } from "~/server/database"; +import { readUsers, ServerSession } from "~/server/database"; import type { ApiAccount, ApiEvent } from "~/shared/types/api"; +import { serverSessionToApi } from "./utils/session"; +import { H3Event } from "h3"; function sendMessage( stream: WritableStream, @@ -31,18 +33,30 @@ function sendMessageAndClose( ; } -const streams = new Map, { sessionId?: number, accountId?: number }>(); +const streams = new Map, { sessionId?: number, accountId?: number, expiresAtMs: number }>(); let keepaliveInterval: ReturnType | null = null -export function addStream(stream: WritableStream, sessionId?: number, accountId?: number) { +export async function addStream( + event: H3Event, + stream: WritableStream, + session?: ServerSession, +) { if (streams.size === 0) { console.log("Starting keepalive") keepaliveInterval = setInterval(sendKeepalive, 4000) } - streams.set(stream, { sessionId, accountId }); - // Produce a response immediately to avoid the reply waiting for content. - const message = `data: connected sid:${sessionId ?? '-'} aid:${accountId ?? '-'}\n\n`; - sendMessage(stream, message); + const runtimeConfig = useRuntimeConfig(event); + streams.set(stream, { + sessionId: session?.id, + accountId: session?.accountId, + expiresAtMs: session?.expiresAtMs ?? Date.now() + runtimeConfig.sessionExpiresTimeout * 1000, + }); + // Produce a response immediately to avoid the reply waiting for content. + const update: ApiEvent = { + type: "connected", + session: session ? await serverSessionToApi(event, session) : undefined, + }; + sendMessage(stream, `event: update\ndata: ${JSON.stringify(update)}\n\n`); } export function deleteStream(stream: WritableStream) { streams.delete(stream); @@ -119,6 +133,13 @@ export async function broadcastEvent(event: ApiEvent) { if (!streams.size) { return; } + + // Session expiry events cause the streams belonging to that session to be terminated + if (event.type === "session-expired") { + cancelSessionStreams(event.sessionId); + return; + } + const users = await readUsers(); for (const [stream, streamData] of streams) { // Account events are specially handled and only sent to the user they belong to. @@ -139,7 +160,12 @@ export async function broadcastEvent(event: ApiEvent) { } function sendKeepalive() { - for (const stream of streams.keys()) { - sendMessage(stream, ": keepalive\n"); + const now = Date.now(); + for (const [stream, streamData] of streams) { + if (streamData.expiresAtMs > now) { + sendMessage(stream, ": keepalive\n"); + } else { + sendMessageAndClose(stream, `data: cancelled\n\n`); + } } } diff --git a/server/utils/session.ts b/server/utils/session.ts index c33d3c7..db819e4 100644 --- a/server/utils/session.ts +++ b/server/utils/session.ts @@ -13,6 +13,8 @@ import { writeSessions, writeSubscriptions } from "~/server/database"; +import { broadcastEvent } from "../streams"; +import type { ApiSession } from "~/shared/types/api"; async function removeSessionSubscription(sessionId: number) { const subscriptions = await readSubscriptions(); @@ -27,9 +29,13 @@ async function clearServerSessionInternal(event: H3Event, sessions: ServerSessio const existingSessionCookie = await getSignedCookie(event, "session"); if (existingSessionCookie) { const sessionId = parseInt(existingSessionCookie, 10); - const sessionIndex = sessions.findIndex(session => session.id === sessionId); - if (sessionIndex !== -1) { - sessions.splice(sessionIndex, 1); + const session = sessions.find(session => session.id === sessionId); + if (session) { + session.finished = true; + broadcastEvent({ + type: "session-expired", + sessionId, + }); await removeSessionSubscription(sessionId); return true; } @@ -56,6 +62,7 @@ export async function setServerSession(event: H3Event, account: ServerUser) { access: account.type, expiresAtMs: now + runtimeConfig.sessionExpiresTimeout * 1000, discardAtMs: now + runtimeConfig.sessionDiscardTimeout * 1000, + finished: false, id: await nextSessionId(), }; @@ -74,6 +81,7 @@ async function rotateSession(event: H3Event, sessions: ServerSession[], session: access: account?.type ?? "anonymous", expiresAtMs: now + runtimeConfig.sessionExpiresTimeout * 1000, discardAtMs: now + runtimeConfig.sessionDiscardTimeout * 1000, + finished: false, id: await nextSessionId(), }; session.successor = newSession.id; @@ -90,6 +98,9 @@ export async function getServerSession(event: H3Event, ignoreExpired: boolean) { const sessions = await readSessions(); const session = sessions.find(session => session.id === sessionId); if (session) { + if (session.finished) { + return undefined; + } if (!ignoreExpired && session.successor !== undefined) { throw createError({ statusCode: 403, @@ -149,3 +160,18 @@ export async function requireServerSessionWithAdmin(event: H3Event) { } return { ...session, accountId: session.accountId }; } + +export async function serverSessionToApi(event: H3Event, session: ServerSession): Promise { + const users = await readUsers(); + const account = users.find(user => user.id === session.accountId); + const subscriptions = await readSubscriptions(); + const push = Boolean( + subscriptions.find(sub => sub.type === "push" && sub.sessionId === session.id) + ); + + return { + id: session.id, + account, + push, + }; +} diff --git a/shared/types/api.ts b/shared/types/api.ts index 321ea3c..3f2d421 100644 --- a/shared/types/api.ts +++ b/shared/types/api.ts @@ -149,12 +149,22 @@ export interface ApiAccountUpdate { data: ApiAccount, } +export interface ApiConnected { + type: "connected", + session?: ApiSession, +} + export interface ApiScheduleUpdate { type: "schedule-update", updatedFrom?: string, data: ApiSchedule | ApiTombstone, } +export interface ApiSessionExpired { + type: "session-expired", + sessionId: Id, +} + export interface ApiUserUpdate { type: "user-update", updatedFrom?: string, @@ -163,6 +173,8 @@ export interface ApiUserUpdate { export type ApiEvent = | ApiAccountUpdate + | ApiConnected | ApiScheduleUpdate + | ApiSessionExpired | ApiUserUpdate ; diff --git a/stores/session.ts b/stores/session.ts index 48f5fda..2545c4f 100644 --- a/stores/session.ts +++ b/stores/session.ts @@ -4,7 +4,7 @@ */ import { appendResponseHeader } from "h3"; import type { H3Event } from "h3"; -import type { ApiAccount } from "~/shared/types/api"; +import type { ApiAccount, ApiSession } from "~/shared/types/api"; const fetchSessionWithCookie = async (event?: H3Event) => { // Client side @@ -33,6 +33,9 @@ export const useSessionStore = defineStore("session", () => { const actions = { async fetch(event?: H3Event) { const session = await fetchSessionWithCookie(event) + actions.update(session); + }, + update(session?: ApiSession) { state.account.value = session?.account; state.id.value = session?.id; state.push.value = session?.push ?? false; @@ -58,6 +61,13 @@ export const useSessionStore = defineStore("session", () => { }, }; + appEventSource?.addEventListener("update", (event) => { + if (event.data.type !== "connected") { + return; + } + actions.update(event.data.session); + }); + return { ...state, ...actions,