Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 6 additions & 0 deletions .changeset/identity-tokens.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,6 @@
---
"@chatcops/server": minor
"@chatcops/widget": minor
---

Signed-in visitors. With `identity: { secret }` on the server, your site signs a `userToken` (an HS256 JWT with `sub`, `exp` and an optional `tier`) and the widget sends it with `/session` (`userToken` or `data-user-token`). The user id goes into the session token and is stored on the conversation, so the same user can resume it with a fresh token and nobody else can. `identity.tiers` gives a tier its own budgets.
8 changes: 8 additions & 0 deletions packages/server/src/config.ts
Original file line number Diff line number Diff line change
Expand Up @@ -109,6 +109,14 @@ export interface ChatCopsServerConfig {
* the widget's `turnstileSiteKey`.
*/
turnstile?: TurnstileConfig;
/**
* Signed-in visitors. Your site signs a \`userToken\` (an HS256 JWT with
* \`sub\`, \`exp\` and an optional \`tier\`) with \`secret\`, and the widget
* sends it with \`/session\`. The user id is stored on the conversation, so
* the same user can resume it with a fresh token, and \`tiers\` gives a
* tier its own budgets (merged over \`budgets\`). Needs \`sessionSecrets\`.
*/
identity?: { secret: string; tiers?: Record<string, Budgets> };
/**
* Adds the reply's token usage to the `finish` part of protocol v1
* streams. Off by default, since visitors can read the stream.
Expand Down
48 changes: 37 additions & 11 deletions packages/server/src/handler.ts
Original file line number Diff line number Diff line change
Expand Up @@ -98,7 +98,22 @@ export function createChatHandler(config: ChatCopsServerConfig) {
// rather than as a provider error on the first message.
validateTools(toolDefs);
const tools = toolRegistry(config.tools);
const sessions = createSessions(config, async (id) => (await conversations.getMessages(id)).length === 0);
const sessions = createSessions(config, async (id) => {
const conversation = await conversations.getOrCreate(id);
const uid = conversation.metadata?.uid;
return { unused: conversation.messages.length === 0, ...(typeof uid === 'string' ? { uid } : {}) };
});
// A tier's budgets replace those scopes of the base ones; counts are shared.
// Built here so a mistake in one fails at startup.
const tierGuards = new Map(
budgets
? Object.entries(config.identity?.tiers ?? {}).map(([tier, tierBudgets]) => [
tier,
createBudgetGuard(counters, { ...budgets.budgets, ...tierBudgets }),
])
: []
);
const guardFor = (tier: string | undefined) => (tier && tierGuards.get(tier)) || budgetGuard;
if (!loggedSecurity) {
loggedSecurity = true;
logSecurityAtStartup({
Expand Down Expand Up @@ -137,10 +152,12 @@ export function createChatHandler(config: ChatCopsServerConfig) {
);
}
const req: ValidatedChatRequest = parsed.data;
let user: SignedInUser | undefined;

if (sessions) {
const session = await sessions.check(options.authorization, options.origin ?? null, req.conversationId);
if (!session.ok) return refuse(session.code, { message: session.message }, { error: 'provider_error' });
user = session.claims?.uid ? { uid: session.claims.uid, tier: session.claims.tier } : undefined;
}

if (clientIp) {
Expand All @@ -167,19 +184,20 @@ export function createChatHandler(config: ChatCopsServerConfig) {
return refuse('conversation_busy', {}, { error: 'provider_error' });
}

return { ok: true, parts: replyParts(provider, req, options.signal, clientIp, release) };
return { ok: true, parts: replyParts(provider, req, options.signal, clientIp, release, user) };
}

async function* replyParts(
provider: Awaited<typeof providerPromise>,
req: ValidatedChatRequest,
signal: AbortSignal | undefined,
clientIp: string | undefined,
release: () => Promise<void>
release: () => Promise<void>,
user?: SignedInUser
): AsyncGenerator<WirePart, void> {
const pendingWebhooks = new Set<Promise<void>>();
try {
for await (const part of reply(provider, req, signal, pendingWebhooks, clientIp)) {
for await (const part of reply(provider, req, signal, pendingWebhooks, clientIp, user)) {
// Tells the widget to get a session before its next request.
yield part.type === 'start' && sessions ? { ...part, capabilities: { ...part.capabilities, session: true } } : part;
}
Expand Down Expand Up @@ -230,7 +248,8 @@ export function createChatHandler(config: ChatCopsServerConfig) {
req: ValidatedChatRequest,
signal: AbortSignal | undefined,
pendingWebhooks: Set<Promise<void>>,
clientIp: string | undefined
clientIp: string | undefined,
user?: SignedInUser
): AsyncGenerator<WirePart, void> {
const conversation = await conversations.getOrCreate(req.conversationId);
const conversationMessages = conversation.messages;
Expand Down Expand Up @@ -298,11 +317,12 @@ export function createChatHandler(config: ChatCopsServerConfig) {
content: req.message,
timestamp: Date.now(),
};
await conversations.appendMessages(
req.conversationId,
[userMessage],
newVisitor ? { metadata: { visitor: newVisitor } } : undefined
);
// The signed-in user it belongs to, set once with the first message.
const metadata: Record<string, unknown> = {
...(newVisitor ? { visitor: newVisitor } : {}),
...(user && conversation.metadata?.uid === undefined ? { uid: user.uid } : {}),
};
await conversations.appendMessages(req.conversationId, [userMessage], Object.keys(metadata).length ? { metadata } : undefined);
}

// The system prompt is the operator's text and nothing else, so it's the
Expand Down Expand Up @@ -350,7 +370,7 @@ export function createChatHandler(config: ChatCopsServerConfig) {
exposeUsage: config.exposeUsage,
signal,
...(budgetGuard
? { budget: { guard: budgetGuard, context: { conversationId: req.conversationId, ...(clientIp ? { ip: clientIp } : {}) } } }
? { budget: { guard: guardFor(user?.tier) ?? budgetGuard, context: { conversationId: req.conversationId, ...(clientIp ? { ip: clientIp } : {}) } } }
: {}),
onPart: (part) => {
if (part.type !== 'lead-captured') return;
Expand Down Expand Up @@ -444,6 +464,12 @@ export function createChatHandler(config: ChatCopsServerConfig) {
/** The stream a caller writes: protocol v1, the 0.5 chunks, or the AI SDK's UI message stream. */
export type ChatFormat = 'v1' | 'legacy' | 'ai-sdk';

/** A visitor the host site signed in, from the session token. */
interface SignedInUser {
uid: string;
tier?: string;
}

export interface StartChatOptions {
signal?: AbortSignal;
/** Which stream the caller writes, for `protocolStats()`. Defaults to `'v1'`. */
Expand Down
76 changes: 70 additions & 6 deletions packages/server/src/security/sessions.ts
Original file line number Diff line number Diff line change
@@ -1,5 +1,5 @@
import { errorBody, type WireErrorBody, type WireErrorCode } from '@chatcops/core';
import { importSessionKey, signSession, verifySession, type SessionClaims } from './session-token.js';
import { base64UrlDecode, importSessionKey, signSession, verifySession, verifySessionSignature, type SessionClaims } from './session-token.js';
import { verifyTurnstile, type TurnstileConfig } from './turnstile.js';

/**
Expand All @@ -21,6 +21,15 @@ export interface SessionConfig {
sessionTtlSec?: number;
requireSession?: boolean;
turnstile?: TurnstileConfig;
identity?: { secret: string };
}

/** What `/session` needs to know about a conversation id. */
export interface ConversationInfo {
/** Nothing is stored under it yet. */
unused: boolean;
/** The signed-in user it belongs to, from a `userToken`. */
uid?: string;
}

export type SessionCheck =
Expand All @@ -43,19 +52,24 @@ export interface Sessions {

/**
* `undefined` when `sessionSecrets` isn't set. Throws at startup on a bad
* config. `isUnused` says whether a conversation has nothing stored yet.
* config. `lookup` says what's stored under a conversation id.
*/
export function createSessions(
config: SessionConfig,
isUnused: (conversationId: string) => Promise<boolean> = async () => false,
lookup: (conversationId: string) => Promise<ConversationInfo> = async () => ({ unused: false }),
now: () => number = Date.now
): Sessions | undefined {
const secrets = config.sessionSecrets;
if (secrets === undefined) {
if (config.requireSession) throw new Error('`requireSession` needs `sessionSecrets` to sign tokens with.');
if (config.turnstile) throw new Error('`turnstile` is checked when a session token is issued, so it needs `sessionSecrets`.');
if (config.identity) throw new Error('`identity` goes into session tokens, so it needs `sessionSecrets`.');
return undefined;
}
const identitySecret = config.identity?.secret;
if (config.identity && (typeof identitySecret !== 'string' || new TextEncoder().encode(identitySecret).byteLength < MIN_SECRET_BYTES)) {
throw new Error(`\`identity.secret\` needs at least ${MIN_SECRET_BYTES} bytes, like \`sessionSecrets\`.`);
}
const turnstile = config.turnstile;
if (turnstile && !turnstile.secretKey) throw new Error('`turnstile` needs its `secretKey`.');
if (!Array.isArray(secrets) || secrets.length === 0) throw new Error('`sessionSecrets` needs at least one secret.');
Expand All @@ -71,6 +85,8 @@ export function createSessions(
// Imported on first use: Workers don't allow crypto.subtle at global scope.
let keys: Promise<CryptoKey[]> | undefined;
const getKeys = () => (keys ??= Promise.all(secrets.map((secret) => importSessionKey(secret))));
let identityKey: Promise<CryptoKey> | undefined;
const getIdentityKey = () => (identityKey ??= importSessionKey(identitySecret as string));
const nowSec = () => Math.floor(now() / 1000);
const required = config.requireSession ?? true;

Expand All @@ -86,7 +102,8 @@ export function createSessions(
// Resuming a conversation needs a token that was issued for it, from
// this origin. Expired is fine within the grace window. An id with
// nothing stored yet (the widget makes its own) is given out as is,
// since there's nothing in it to read. Anything else gets a new
// since there's nothing in it to read, and so is a signed-in user's own
// conversation (with a \`userToken\` for them). Anything else gets a new
// conversation, so a guessed or leaked id is useless.
let resumed: SessionClaims | undefined;
let cid: string | undefined;
Expand All @@ -99,6 +116,17 @@ export function createSessions(
});
if (result.ok) resumed = result.claims;
}
// A visitor the host site signed in. A token that doesn't verify is an
// error, not an anonymous session, so a bug on the host site shows.
let user: UserClaims | undefined;
if (request.userToken !== undefined) {
if (!identitySecret) return { status: 400, body: errorBody('invalid_request', { message: '`userToken` needs `identity` on the server.' }) };
user = await verifyUserToken(request.userToken, await getIdentityKey(), time);
if (!user) return { status: 401, body: errorBody('session_required', { message: 'The userToken is invalid or has expired.' }) };
}
// A token for another user doesn't carry over.
if (resumed?.uid && resumed.uid !== user?.sub) resumed = undefined;

// A refresh was checked when its first token was issued; a new session
// has to pass Turnstile first.
if (turnstile && !resumed) {
Expand All @@ -116,7 +144,11 @@ export function createSessions(
}

if (resumed) cid = resumed.cid;
else if (request.conversationId && (await isUnused(request.conversationId))) cid = request.conversationId;
else if (request.conversationId) {
const info = await lookup(request.conversationId);
// Nothing stored yet, or this signed-in user's own conversation.
if (info.unused || (user && info.uid === user.sub)) cid = request.conversationId;
}

const claims: SessionClaims = {
v: 1,
Expand All @@ -125,6 +157,7 @@ export function createSessions(
org,
iat: time,
exp: time + ttlSec,
...(user ? { uid: user.sub, ...(user.tier ? { tier: user.tier } : {}) } : {}),
};
const token = await signSession(claims, await getKeys());
return { status: 200, body: { token, expiresAt: claims.exp * 1000, conversationId: claims.cid } };
Expand Down Expand Up @@ -160,6 +193,8 @@ function bearer(header: string | null | undefined): string | undefined {
interface MintRequest {
conversationId?: string;
previousToken?: string;
/** Signed by the host site with `identity.secret`: `{ sub, exp, tier? }`. */
userToken?: string;
turnstileToken?: string;
/** Sent again with a retried request, so Cloudflare accepts the same Turnstile token twice. */
idempotencyKey?: string;
Expand All @@ -168,8 +203,12 @@ interface MintRequest {
function readMintRequest(body: unknown): MintRequest | undefined {
if (body === undefined || body === null) return {};
if (typeof body !== 'object' || Array.isArray(body)) return undefined;
const { conversationId, previousToken, turnstileToken, idempotencyKey } = body as Record<string, unknown>;
const { conversationId, previousToken, turnstileToken, idempotencyKey, userToken } = body as Record<string, unknown>;
const request: MintRequest = {};
if (userToken !== undefined) {
if (typeof userToken !== 'string' || userToken.length > 4096) return undefined;
request.userToken = userToken;
}
if (turnstileToken !== undefined) {
if (typeof turnstileToken !== 'string' || turnstileToken.length > 4096) return undefined;
request.turnstileToken = turnstileToken;
Expand All @@ -188,3 +227,28 @@ function readMintRequest(body: unknown): MintRequest | undefined {
}
return request;
}

interface UserClaims {
sub: string;
tier?: string;
}

/**
* A host site's `userToken`: an HS256 JWT signed with `identity.secret`, with
* `sub` (the user id, up to 128 characters), `exp` in seconds and an optional
* `tier`. `undefined` when it doesn't verify or has expired.
*/
export async function verifyUserToken(token: string, key: CryptoKey, nowSec: number): Promise<UserClaims | undefined> {
if (!(await verifySessionSignature(token, [key]))) return undefined;
let payload: Record<string, unknown>;
try {
payload = JSON.parse(new TextDecoder().decode(base64UrlDecode(token.split('.')[1] ?? '')));
} catch {
return undefined;
}
const { sub, exp, tier } = payload ?? {};
if (typeof sub !== 'string' || sub.length === 0 || sub.length > 128) return undefined;
if (typeof exp !== 'number' || !Number.isFinite(exp) || exp + 60 < nowSec) return undefined;
if (tier !== undefined && (typeof tier !== 'string' || tier.length > 64)) return undefined;
return { sub, ...(typeof tier === 'string' ? { tier } : {}) };
}
87 changes: 87 additions & 0 deletions packages/server/tests/sessions.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@ import { scriptedProvider, textStep } from '@chatcops/core/testing/provider';
import { chatcopsVercelHandler } from '../src/adapters/vercel.js';
import { chatcopsMiddleware } from '../src/adapters/express.js';
import { createSessions, SESSION_RESUME_GRACE_SEC } from '../src/security/sessions.js';
import { importSessionKey, signSession } from '../src/security/session-token.js';
import type { ChatCopsServerConfig } from '../src/config.js';

const SECRET = 'a-session-secret-that-is-long-enough-1';
Expand Down Expand Up @@ -287,6 +288,92 @@ describe('Turnstile when a session is issued', () => {
});
});

describe('signed-in visitors (identity)', () => {
const IDENTITY_SECRET = 'the-host-site-identity-secret-of-32b';
const now = () => Math.floor(Date.now() / 1000);

/** What a host site does on its own server to vouch for a signed-in user. */
async function userToken(claims: Record<string, unknown>, secret = IDENTITY_SECRET) {
return signSession(claims as never, [await importSessionKey(secret)]);
}
const claimsOf = (token: string) => JSON.parse(Buffer.from(token.split('.')[1], 'base64url').toString());

it('puts the user in the session and on the conversation', async () => {
const storage = memoryStorage();
const { session, chat } = setup({ storage, identity: { secret: IDENTITY_SECRET } });
const minted = await session({ userToken: await userToken({ sub: 'user-42', exp: now() + 3600, tier: 'pro' }) });
expect(minted.status).toBe(200);
expect(claimsOf(minted.body.token)).toMatchObject({ uid: 'user-42', tier: 'pro' });

expect((await chat(minted.body.conversationId, minted.body.token)).status).toBe(200);
expect((await storage.conversations.get(minted.body.conversationId))?.metadata?.uid).toBe('user-42');
});

it('lets the same user resume their conversation with a fresh token, and nobody else', async () => {
const { session, chat } = setup({ storage: memoryStorage(), identity: { secret: IDENTITY_SECRET } });
const first = await session({ userToken: await userToken({ sub: 'user-42', exp: now() + 3600 }) });
await chat(first.body.conversationId, first.body.token);

// A new device: no previous token, just the user's own token.
const sameUser = await session({ conversationId: first.body.conversationId, userToken: await userToken({ sub: 'user-42', exp: now() + 3600 }) });
expect(sameUser.body.conversationId).toBe(first.body.conversationId);

const otherUser = await session({ conversationId: first.body.conversationId, userToken: await userToken({ sub: 'user-7', exp: now() + 3600 }) });
expect(otherUser.body.conversationId).not.toBe(first.body.conversationId);

// Another user's old session token doesn't carry over either.
const swapped = await session({
conversationId: first.body.conversationId,
previousToken: first.body.token,
userToken: await userToken({ sub: 'user-7', exp: now() + 3600 }),
});
expect(swapped.body.conversationId).not.toBe(first.body.conversationId);
});

it('refuses a userToken that does not verify, has expired or has no user id', async () => {
const { session } = setup({ identity: { secret: IDENTITY_SECRET } });
for (const token of [
await userToken({ sub: 'user-42', exp: now() + 3600 }, 'a-different-secret-that-is-long-enough'),
await userToken({ sub: 'user-42', exp: now() - 3600 }),
await userToken({ exp: now() + 3600 }),
'not.a.token',
]) {
const res = await session({ userToken: token });
expect(res.status).toBe(401);
expect(res.body.error.code).toBe('session_required');
}
});

it('refuses a userToken when identity is off', async () => {
const { session } = setup();
expect((await session({ userToken: 'eyJ.x.y' })).status).toBe(400);
});

it('gives a tier its own budgets', async () => {
const { session, chat } = setup({
storage: memoryStorage(),
budgets: { perConversation: { messages: 1 } },
identity: { secret: IDENTITY_SECRET, tiers: { pro: { perConversation: { messages: 3 } } } },
});
const send = async (tier?: string) => {
const minted = await session({ userToken: await userToken({ sub: `user-${tier ?? 'free'}`, exp: now() + 3600, ...(tier ? { tier } : {}) }) });
const replies = [];
for (let i = 0; i < 2; i++) replies.push(await chat(minted.body.conversationId, minted.body.token));
return replies.map((reply) => (reply.text.includes('budget_exceeded') ? 'refused' : 'answered'));
};
expect(await send()).toEqual(['answered', 'refused']);
expect(await send('pro')).toEqual(['answered', 'answered']);
});

it('needs sessionSecrets and a long enough secret, and checks tier budgets at startup', () => {
expect(() => createSessions({ identity: { secret: IDENTITY_SECRET } })).toThrow('needs `sessionSecrets`');
expect(() => createSessions({ sessionSecrets: [SECRET], identity: { secret: 'short' } })).toThrow('at least 32 bytes');
expect(() => setup({ budgets: { perConversation: { messages: 5 } }, identity: { secret: IDENTITY_SECRET, tiers: { pro: { perConversation: { messages: -1 } } } } })).toThrow(
'budgets.perConversation.messages must be a positive number'
);
});
});

describe('the Express middleware', () => {
function run(config: ChatCopsServerConfig, path: string, body: unknown, headers: Record<string, string> = {}) {
const chunks: string[] = [];
Expand Down
Loading
Loading