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

@@ -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,

docs/dev/server-sent-events.md

@@ -0,0 +1,13 @@
+<!--
+	SPDX-FileCopyrightText: © 2025 Hornwitser <code@hornwitser.no>
+	SPDX-License-Identifier: AGPL-3.0-or-later
+-->
+# 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.

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);

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<number>();
+	// Expire sessions for this user
+	const expiredSessionIds = new Set<number>();
 	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);

server/api/auth/session.delete.ts

@@ -18,8 +18,5 @@ export default defineEventHandler(async (event) => {
 		}
 	}
 
-	if (session) {
-		cancelSessionStreams(session.id);
-	}
 	await clearServerSession(event);
 })

server/api/auth/session.get.ts

@@ -2,23 +2,10 @@
 	SPDX-FileCopyrightText: © 2025 Hornwitser <code@hornwitser.no>
 	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<ApiSession | undefined> => {
+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);
 })

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", "*");

server/database.ts

@@ -13,6 +13,7 @@ export interface ServerSession {
 	expiresAtMs: number,
 	discardAtMs: number,
 	successor?: Id,
+	finished: boolean,
 };
 
 export interface ServerUser {

server/streams.ts

@@ -2,8 +2,10 @@
 	SPDX-FileCopyrightText: © 2025 Hornwitser <code@hornwitser.no>
 	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<string>,
@@ -31,18 +33,30 @@ function sendMessageAndClose(
 	;
 }
 
-const streams = new Map<WritableStream<string>, { sessionId?: number, accountId?: number }>();
+const streams = new Map<WritableStream<string>, { sessionId?: number, accountId?: number, expiresAtMs: number }>();
 
 let keepaliveInterval: ReturnType<typeof setInterval> | null = null
-export function addStream(stream: WritableStream<string>, sessionId?: number, accountId?: number) {
+export async function addStream(
+	event: H3Event,
+	stream: WritableStream<string>,
+	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<string>) {
 	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`);
+		}
 	}
 }

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<ApiSession> {
+	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,
+	};
+}

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
 ;

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,