wushenghua/hapi
1
0
Fork 0
mirror of https://github.com/wu736139669/hapi.git synced 2026-08-06 06:41:56 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: add namespace-based multi-user isolation

Browse Source 
Implement namespace support across sessions, machines, and users for multi-user server deployments. Add access control with specific error reasons (namespace-missing, access-denied, not-found) and database schema updates with namespace columns and indexes.
This commit is contained in:
weishu
2025-12-31 21:56:01 +08:00
parent 460c393006
commit e821458af8
30 changed files with 737 additions and 124 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/src/socket/handlers/cli.ts
+105 -11
View File
@@ -1,11 +1,11 @@
import type { Server, Socket } from 'socket.io'
import { z } from 'zod'
import { randomUUID } from 'node:crypto'
import type { Store } from '../../store'
import type { Store, StoredMachine, StoredSession } from '../../store'
import { RpcRegistry } from '../rpcRegistry'
import type { SyncEvent } from '../../sync/syncEngine'
import { extractTodoWriteTodosFromMessageContent } from '../../sync/todos'
import { TerminalRegistry } from '../terminalRegistry'
import type { SocketServer, SocketWithData } from '../socketTypes'
type SessionAlivePayload = {
sid: string
@@ -89,7 +89,7 @@ const terminalErrorSchema = z.object({
})
export type CliHandlersDeps = {
io: Server
io: SocketServer
store: Store
rpcRegistry: RpcRegistry
terminalRegistry: TerminalRegistry
@@ -99,21 +99,64 @@ export type CliHandlersDeps = {
onWebappEvent?: (event: SyncEvent) => void
}
export function registerCliHandlers(socket: Socket, deps: CliHandlersDeps): void {
type AccessErrorReason = 'namespace-missing' | 'access-denied' | 'not-found'
type AccessResult<T> =
| { ok: true; value: T }
| { ok: false; reason: AccessErrorReason }
export function registerCliHandlers(socket: SocketWithData, deps: CliHandlersDeps): void {
const { io, store, rpcRegistry, terminalRegistry, onSessionAlive, onSessionEnd, onMachineAlive, onWebappEvent } = deps
const terminalNamespace = io.of('/terminal')
const namespace = typeof socket.data.namespace === 'string' ? socket.data.namespace : null
const resolveSessionAccess = (sessionId: string): AccessResult<StoredSession> => {
if (!namespace) {
return { ok: false, reason: 'namespace-missing' }
}
const session = store.getSessionByNamespace(sessionId, namespace)
if (session) {
return { ok: true, value: session }
}
if (store.getSession(sessionId)) {
return { ok: false, reason: 'access-denied' }
}
return { ok: false, reason: 'not-found' }
}
const resolveMachineAccess = (machineId: string): AccessResult<StoredMachine> => {
if (!namespace) {
return { ok: false, reason: 'namespace-missing' }
}
const machine = store.getMachineByNamespace(machineId, namespace)
if (machine) {
return { ok: true, value: machine }
}
if (store.getMachine(machineId)) {
return { ok: false, reason: 'access-denied' }
}
return { ok: false, reason: 'not-found' }
}
const auth = socket.handshake.auth as Record<string, unknown> | undefined
const sessionId = typeof auth?.sessionId === 'string' ? auth.sessionId : null
if (sessionId) {
if (sessionId && resolveSessionAccess(sessionId).ok) {
socket.join(`session:${sessionId}`)
}
const machineId = typeof auth?.machineId === 'string' ? auth.machineId : null
if (machineId) {
if (machineId && resolveMachineAccess(machineId).ok) {
socket.join(`machine:${machineId}`)
}
const emitAccessError = (scope: 'session' | 'machine', id: string, reason: AccessErrorReason) => {
const message = reason === 'access-denied'
? `${scope} access denied`
: reason === 'not-found'
? `${scope} not found`
: 'Namespace missing'
socket.emit('error', { message, code: reason, scope, id })
}
socket.on('rpc-register', (data: unknown) => {
const parsed = rpcRegisterSchema.safeParse(data)
if (!parsed.success) {
@@ -161,11 +204,18 @@ export function registerCliHandlers(socket: Socket, deps: CliHandlersDeps): void
})()
: raw
const sessionAccess = resolveSessionAccess(sid)
if (!sessionAccess.ok) {
emitAccessError('session', sid, sessionAccess.reason)
return
}
const session = sessionAccess.value
const msg = store.addMessage(sid, content, localId)
const todos = extractTodoWriteTodosFromMessageContent(content)
if (todos) {
const updated = store.setSessionTodos(sid, todos, msg.createdAt)
const updated = store.setSessionTodos(sid, todos, msg.createdAt, session.namespace)
if (updated) {
onWebappEvent?.({ type: 'session-updated', sessionId: sid, data: { sid } })
}
@@ -211,7 +261,13 @@ export function registerCliHandlers(socket: Socket, deps: CliHandlersDeps): void
}
const { sid, metadata, expectedVersion } = parsed.data
const result = store.updateSessionMetadata(sid, metadata, expectedVersion)
const sessionAccess = resolveSessionAccess(sid)
if (!sessionAccess.ok) {
cb({ result: 'error', reason: sessionAccess.reason })
return
}
const result = store.updateSessionMetadata(sid, metadata, expectedVersion, sessionAccess.value.namespace)
if (result.result === 'success') {
cb({ result: 'success', version: result.version, metadata: result.value })
} else if (result.result === 'version-mismatch') {
@@ -245,7 +301,13 @@ export function registerCliHandlers(socket: Socket, deps: CliHandlersDeps): void
}
const { sid, agentState, expectedVersion } = parsed.data
const result = store.updateSessionAgentState(sid, agentState, expectedVersion)
const sessionAccess = resolveSessionAccess(sid)
if (!sessionAccess.ok) {
cb({ result: 'error', reason: sessionAccess.reason })
return
}
const result = store.updateSessionAgentState(sid, agentState, expectedVersion, sessionAccess.value.namespace)
if (result.result === 'success') {
cb({ result: 'success', version: result.version, agentState: result.value })
} else if (result.result === 'version-mismatch') {
@@ -275,6 +337,11 @@ export function registerCliHandlers(socket: Socket, deps: CliHandlersDeps): void
if (!data || typeof data.sid !== 'string' || typeof data.time !== 'number') {
return
}
const sessionAccess = resolveSessionAccess(data.sid)
if (!sessionAccess.ok) {
emitAccessError('session', data.sid, sessionAccess.reason)
return
}
onSessionAlive?.(data)
})
@@ -282,6 +349,11 @@ export function registerCliHandlers(socket: Socket, deps: CliHandlersDeps): void
if (!data || typeof data.sid !== 'string' || typeof data.time !== 'number') {
return
}
const sessionAccess = resolveSessionAccess(data.sid)
if (!sessionAccess.ok) {
emitAccessError('session', data.sid, sessionAccess.reason)
return
}
onSessionEnd?.(data)
})
@@ -289,6 +361,11 @@ export function registerCliHandlers(socket: Socket, deps: CliHandlersDeps): void
if (!data || typeof data.machineId !== 'string' || typeof data.time !== 'number') {
return
}
const machineAccess = resolveMachineAccess(data.machineId)
if (!machineAccess.ok) {
emitAccessError('machine', data.machineId, machineAccess.reason)
return
}
onMachineAlive?.(data)
})
@@ -300,7 +377,13 @@ export function registerCliHandlers(socket: Socket, deps: CliHandlersDeps): void
}
const { machineId: id, metadata, expectedVersion } = parsed.data
const result = store.updateMachineMetadata(id, metadata, expectedVersion)
const machineAccess = resolveMachineAccess(id)
if (!machineAccess.ok) {
cb({ result: 'error', reason: machineAccess.reason })
return
}
const result = store.updateMachineMetadata(id, metadata, expectedVersion, machineAccess.value.namespace)
if (result.result === 'success') {
cb({ result: 'success', version: result.version, metadata: result.value })
} else if (result.result === 'version-mismatch') {
@@ -334,7 +417,13 @@ export function registerCliHandlers(socket: Socket, deps: CliHandlersDeps): void
}
const { machineId: id, daemonState, expectedVersion } = parsed.data
const result = store.updateMachineDaemonState(id, daemonState, expectedVersion)
const machineAccess = resolveMachineAccess(id)
if (!machineAccess.ok) {
cb({ result: 'error', reason: machineAccess.reason })
return
}
const result = store.updateMachineDaemonState(id, daemonState, expectedVersion, machineAccess.value.namespace)
if (result.result === 'success') {
cb({ result: 'success', version: result.version, daemonState: result.value })
} else if (result.result === 'version-mismatch') {
@@ -381,6 +470,11 @@ export function registerCliHandlers(socket: Socket, deps: CliHandlersDeps): void
if (payload.sessionId !== entry.sessionId) {
return
}
const sessionAccess = resolveSessionAccess(payload.sessionId)
if (!sessionAccess.ok) {
emitAccessError('session', payload.sessionId, sessionAccess.reason)
return
}
const terminalSocket = terminalNamespace.sockets.get(entry.socketId)
if (!terminalSocket) {
return
server/src/socket/handlers/terminal.test.ts
+6 -4
View File
@@ -1,7 +1,7 @@
import { describe, expect, it } from 'bun:test'
import type { Server, Socket } from 'socket.io'
import { registerTerminalHandlers } from './terminal'
import { TerminalRegistry } from '../terminalRegistry'
import type { SocketServer, SocketWithData } from '../socketTypes'
type EmittedEvent = {
event: string
@@ -74,12 +74,13 @@ function createHarness(options?: {
}): Harness {
const io = new FakeServer()
const terminalSocket = new FakeSocket('terminal-socket')
terminalSocket.data.namespace = 'default'
const terminalRegistry = new TerminalRegistry({ idleTimeoutMs: 0 })
const cliNamespace = io.of('/cli')
registerTerminalHandlers(terminalSocket as unknown as Socket, {
io: io as unknown as Server,
getSession: () => ({ active: options?.sessionActive ?? true }),
registerTerminalHandlers(terminalSocket as unknown as SocketWithData, {
io: io as unknown as SocketServer,
getSession: () => ({ active: options?.sessionActive ?? true, namespace: 'default' }),
terminalRegistry,
maxTerminalsPerSocket: options?.maxTerminalsPerSocket ?? 4,
maxTerminalsPerSession: options?.maxTerminalsPerSession ?? 4
@@ -89,6 +90,7 @@ function createHarness(options?: {
}
function connectCliSocket(cliNamespace: FakeNamespace, cliSocket: FakeSocket, sessionId: string): void {
cliSocket.data.namespace = 'default'
cliNamespace.sockets.set(cliSocket.id, cliSocket)
const roomId = `session:${sessionId}`
const room = cliNamespace.adapter.rooms.get(roomId) ?? new Set<string>()
server/src/socket/handlers/terminal.ts
+12 -10
View File
@@ -1,6 +1,6 @@
import type { Server, Socket } from 'socket.io'
import { z } from 'zod'
import type { TerminalRegistry, TerminalRegistryEntry } from '../terminalRegistry'
import type { SocketServer, SocketWithData } from '../socketTypes'
const terminalCreateSchema = z.object({
sessionId: z.string().min(1),
@@ -25,16 +25,17 @@ const terminalCloseSchema = z.object({
})
export type TerminalHandlersDeps = {
io: Server
getSession: (sessionId: string) => { active: boolean } | null
io: SocketServer
getSession: (sessionId: string) => { active: boolean; namespace: string } | null
terminalRegistry: TerminalRegistry
maxTerminalsPerSocket: number
maxTerminalsPerSession: number
}
export function registerTerminalHandlers(socket: Socket, deps: TerminalHandlersDeps): void {
export function registerTerminalHandlers(socket: SocketWithData, deps: TerminalHandlersDeps): void {
const { io, getSession, terminalRegistry, maxTerminalsPerSocket, maxTerminalsPerSession } = deps
const cliNamespace = io.of('/cli')
const namespace = typeof socket.data.namespace === 'string' ? socket.data.namespace : null
const emitTerminalError = (terminalId: string, message: string) => {
socket.emit('terminal:error', { terminalId, message })
@@ -48,9 +49,9 @@ export function registerTerminalHandlers(socket: Socket, deps: TerminalHandlersD
return entry
}
const resolveCliSocket = (entry: TerminalRegistryEntry, reportError: boolean): Socket | null => {
const resolveCliSocket = (entry: TerminalRegistryEntry, reportError: boolean): SocketWithData | null => {
const cliSocket = cliNamespace.sockets.get(entry.cliSocketId)
if (!cliSocket) {
if (!cliSocket || cliSocket.data.namespace !== namespace) {
terminalRegistry.remove(entry.terminalId)
if (reportError) {
emitTerminalError(entry.terminalId, 'CLI disconnected.')
@@ -62,7 +63,7 @@ export function registerTerminalHandlers(socket: Socket, deps: TerminalHandlersD
const emitCloseToCli = (entry: TerminalRegistryEntry): void => {
const cliSocket = cliNamespace.sockets.get(entry.cliSocketId)
if (!cliSocket) {
if (!cliSocket || cliSocket.data.namespace !== namespace) {
return
}
cliSocket.emit('terminal:close', {
@@ -77,8 +78,9 @@ export function registerTerminalHandlers(socket: Socket, deps: TerminalHandlersD
return null
}
for (const socketId of room) {
if (cliNamespace.sockets.has(socketId)) {
return socketId
const cliSocket = cliNamespace.sockets.get(socketId)
if (cliSocket && cliSocket.data.namespace === namespace) {
return cliSocket.id
}
}
return null
@@ -92,7 +94,7 @@ export function registerTerminalHandlers(socket: Socket, deps: TerminalHandlersD
const { sessionId, terminalId, cols, rows } = parsed.data
const session = getSession(sessionId)
if (!session || !session.active) {
if (!namespace || !session || session.namespace !== namespace || !session.active) {
emitTerminalError(terminalId, 'Session is inactive or unavailable.')
return
}
server/src/socket/server.ts
+12 -6
View File
@@ -1,18 +1,21 @@
import { Server as Engine } from '@socket.io/bun-engine'
import { Server } from 'socket.io'
import { Server, type DefaultEventsMap } from 'socket.io'
import { jwtVerify } from 'jose'
import { z } from 'zod'
import type { Store } from '../store'
import { configuration } from '../configuration'
import { safeCompareStrings } from '../utils/crypto'
import { parseAccessToken } from '../utils/accessToken'
import { registerCliHandlers } from './handlers/cli'
import { registerTerminalHandlers } from './handlers/terminal'
import { RpcRegistry } from './rpcRegistry'
import type { SyncEvent } from '../sync/syncEngine'
import { TerminalRegistry } from './terminalRegistry'
import type { SocketData, SocketServer } from './socketTypes'
const jwtPayloadSchema = z.object({
uid: z.number()
uid: z.number(),
ns: z.string()
})
const DEFAULT_IDLE_TIMEOUT_MS = 15 * 60_000
@@ -30,7 +33,7 @@ function resolveEnvNumber(name: string, fallback: number): number {
export type SocketServerDeps = {
store: Store
jwtSecret: Uint8Array
getSession?: (sessionId: string) => { active: boolean } | null
getSession?: (sessionId: string) => { active: boolean; namespace: string } | null
onWebappEvent?: (event: SyncEvent) => void
onSessionAlive?: (payload: { sid: string; time: number; thinking?: boolean; mode?: 'local' | 'remote' }) => void
onSessionEnd?: (payload: { sid: string; time: number }) => void
@@ -38,14 +41,14 @@ export type SocketServerDeps = {
}
export function createSocketServer(deps: SocketServerDeps): {
io: Server
io: SocketServer
engine: Engine
rpcRegistry: RpcRegistry
} {
const corsOrigins = configuration.corsOrigins
const allowAllOrigins = corsOrigins.includes('*')
const io = new Server({
const io = new Server<DefaultEventsMap, DefaultEventsMap, DefaultEventsMap, SocketData>({
cors: {
origin: (origin, callback) => {
if (!origin) {
@@ -94,9 +97,11 @@ export function createSocketServer(deps: SocketServerDeps): {
cliNs.use((socket, next) => {
const auth = socket.handshake.auth as Record<string, unknown> | undefined
const token = typeof auth?.token === 'string' ? auth.token : null
if (!safeCompareStrings(token, configuration.cliApiToken)) {
const parsedToken = token ? parseAccessToken(token) : null
if (!parsedToken || !safeCompareStrings(parsedToken.baseToken, configuration.cliApiToken)) {
return next(new Error('Invalid token'))
}
socket.data.namespace = parsedToken.namespace
next()
})
cliNs.on('connection', (socket) => registerCliHandlers(socket, {
@@ -124,6 +129,7 @@ export function createSocketServer(deps: SocketServerDeps): {
return next(new Error('Invalid token payload'))
}
socket.data.userId = parsed.data.uid
socket.data.namespace = parsed.data.ns
next()
return
} catch {
server/src/socket/socketTypes.ts
+9
View File
@@ -0,0 +1,9 @@
import type { DefaultEventsMap, Server, Socket } from 'socket.io'
export type SocketData = {
namespace?: string
userId?: number
}
export type SocketServer = Server<DefaultEventsMap, DefaultEventsMap, DefaultEventsMap, SocketData>
export type SocketWithData = Socket<DefaultEventsMap, DefaultEventsMap, DefaultEventsMap, SocketData>
server/src/sse/sseManager.test.ts
+66
View File
@@ -0,0 +1,66 @@
import { describe, expect, it } from 'bun:test'
import { SSEManager } from './sseManager'
import type { SyncEvent } from '../sync/syncEngine'
describe('SSEManager namespace filtering', () => {
it('routes events to matching namespace', () => {
const manager = new SSEManager(0)
const receivedAlpha: SyncEvent[] = []
const receivedBeta: SyncEvent[] = []
manager.subscribe({
id: 'alpha',
namespace: 'alpha',
all: true,
send: (event) => {
receivedAlpha.push(event)
},
sendHeartbeat: () => {}
})
manager.subscribe({
id: 'beta',
namespace: 'beta',
all: true,
send: (event) => {
receivedBeta.push(event)
},
sendHeartbeat: () => {}
})
manager.broadcast({ type: 'session-updated', sessionId: 's1', namespace: 'alpha' })
expect(receivedAlpha).toHaveLength(1)
expect(receivedBeta).toHaveLength(0)
})
it('broadcasts connection-changed to all namespaces', () => {
const manager = new SSEManager(0)
const received: Array<{ id: string; event: SyncEvent }> = []
manager.subscribe({
id: 'alpha',
namespace: 'alpha',
all: true,
send: (event) => {
received.push({ id: 'alpha', event })
},
sendHeartbeat: () => {}
})
manager.subscribe({
id: 'beta',
namespace: 'beta',
all: true,
send: (event) => {
received.push({ id: 'beta', event })
},
sendHeartbeat: () => {}
})
manager.broadcast({ type: 'connection-changed', data: { status: 'connected' } })
expect(received).toHaveLength(2)
expect(received.map((entry) => entry.id).sort()).toEqual(['alpha', 'beta'])
})
})
server/src/sse/sseManager.ts
+11
View File
@@ -2,6 +2,7 @@ import type { SyncEvent } from '../sync/syncEngine'
export type SSESubscription = {
id: string
namespace: string
all: boolean
sessionId: string | null
machineId: string | null
@@ -23,6 +24,7 @@ export class SSEManager {
subscribe(options: {
id: string
namespace: string
all?: boolean
sessionId?: string | null
machineId?: string | null
@@ -31,6 +33,7 @@ export class SSEManager {
}): SSESubscription {
const subscription: SSEConnection = {
id: options.id,
namespace: options.namespace,
all: Boolean(options.all),
sessionId: options.sessionId ?? null,
machineId: options.machineId ?? null,
@@ -42,6 +45,7 @@ export class SSEManager {
this.ensureHeartbeat()
return {
id: subscription.id,
namespace: subscription.namespace,
all: subscription.all,
sessionId: subscription.sessionId,
machineId: subscription.machineId
@@ -96,6 +100,13 @@ export class SSEManager {
}
private shouldSend(connection: SSEConnection, event: SyncEvent): boolean {
if (event.type !== 'connection-changed') {
const eventNamespace = event.namespace
if (!eventNamespace || eventNamespace !== connection.namespace) {
return false
}
}
if (event.type === 'message-received') {
return Boolean(event.sessionId && connection.sessionId === event.sessionId)
}
server/src/store/index.ts
+134 -31
View File
@@ -6,6 +6,7 @@ import { randomUUID } from 'node:crypto'
export type StoredSession = {
id: string
tag: string | null
namespace: string
machineId: string | null
createdAt: number
updatedAt: number
@@ -22,6 +23,7 @@ export type StoredSession = {
export type StoredMachine = {
id: string
namespace: string
createdAt: number
updatedAt: number
metadata: unknown | null
@@ -46,6 +48,7 @@ export type StoredUser = {
id: number
platform: string
platformUserId: string
namespace: string
createdAt: number
}
@@ -57,6 +60,7 @@ export type VersionedUpdateResult<T> =
type DbSessionRow = {
id: string
tag: string | null
namespace: string
machine_id: string | null
created_at: number
updated_at: number
@@ -73,6 +77,7 @@ type DbSessionRow = {
type DbMachineRow = {
id: string
namespace: string
created_at: number
updated_at: number
metadata: string | null
@@ -97,6 +102,7 @@ type DbUserRow = {
id: number
platform: string
platform_user_id: string
namespace: string
created_at: number
}
@@ -113,6 +119,7 @@ function toStoredSession(row: DbSessionRow): StoredSession {
return {
id: row.id,
tag: row.tag,
namespace: row.namespace,
machineId: row.machine_id,
createdAt: row.created_at,
updatedAt: row.updated_at,
@@ -131,6 +138,7 @@ function toStoredSession(row: DbSessionRow): StoredSession {
function toStoredMachine(row: DbMachineRow): StoredMachine {
return {
id: row.id,
namespace: row.namespace,
createdAt: row.created_at,
updatedAt: row.updated_at,
metadata: safeJsonParse(row.metadata),
@@ -159,6 +167,7 @@ function toStoredUser(row: DbUserRow): StoredUser {
id: row.id,
platform: row.platform,
platformUserId: row.platform_user_id,
namespace: row.namespace,
createdAt: row.created_at
}
}
@@ -206,6 +215,7 @@ export class Store {
CREATE TABLE IF NOT EXISTS sessions (
id TEXT PRIMARY KEY,
tag TEXT,
namespace TEXT NOT NULL DEFAULT 'default',
machine_id TEXT,
created_at INTEGER NOT NULL,
updated_at INTEGER NOT NULL,
@@ -220,9 +230,11 @@ export class Store {
seq INTEGER DEFAULT 0
);
CREATE INDEX IF NOT EXISTS idx_sessions_tag ON sessions(tag);
CREATE INDEX IF NOT EXISTS idx_sessions_tag_namespace ON sessions(tag, namespace);
CREATE TABLE IF NOT EXISTS machines (
id TEXT PRIMARY KEY,
namespace TEXT NOT NULL DEFAULT 'default',
created_at INTEGER NOT NULL,
updated_at INTEGER NOT NULL,
metadata TEXT,
@@ -233,6 +245,7 @@ export class Store {
active_at INTEGER,
seq INTEGER DEFAULT 0
);
CREATE INDEX IF NOT EXISTS idx_machines_namespace ON machines(namespace);
CREATE TABLE IF NOT EXISTS messages (
id TEXT PRIMARY KEY,
@@ -250,27 +263,44 @@ export class Store {
id INTEGER PRIMARY KEY AUTOINCREMENT,
platform TEXT NOT NULL,
platform_user_id TEXT NOT NULL,
namespace TEXT NOT NULL DEFAULT 'default',
created_at INTEGER NOT NULL,
UNIQUE(platform, platform_user_id)
);
CREATE INDEX IF NOT EXISTS idx_users_platform ON users(platform);
CREATE INDEX IF NOT EXISTS idx_users_platform_namespace ON users(platform, namespace);
`)
const sessionColumns = this.db.prepare('PRAGMA table_info(sessions)').all() as Array<{ name: string }>
const sessionColumnNames = new Set(sessionColumns.map((c) => c.name))
if (!sessionColumnNames.has('namespace')) {
this.db.exec("ALTER TABLE sessions ADD COLUMN namespace TEXT NOT NULL DEFAULT 'default'")
}
if (!sessionColumnNames.has('todos')) {
this.db.exec('ALTER TABLE sessions ADD COLUMN todos TEXT')
}
if (!sessionColumnNames.has('todos_updated_at')) {
this.db.exec('ALTER TABLE sessions ADD COLUMN todos_updated_at INTEGER')
}
const machineColumns = this.db.prepare('PRAGMA table_info(machines)').all() as Array<{ name: string }>
const machineColumnNames = new Set(machineColumns.map((c) => c.name))
if (!machineColumnNames.has('namespace')) {
this.db.exec("ALTER TABLE machines ADD COLUMN namespace TEXT NOT NULL DEFAULT 'default'")
}
const userColumns = this.db.prepare('PRAGMA table_info(users)').all() as Array<{ name: string }>
const userColumnNames = new Set(userColumns.map((c) => c.name))
if (!userColumnNames.has('namespace')) {
this.db.exec("ALTER TABLE users ADD COLUMN namespace TEXT NOT NULL DEFAULT 'default'")
}
}
getOrCreateSession(tag: string, metadata: unknown, agentState: unknown): StoredSession {
getOrCreateSession(tag: string, metadata: unknown, agentState: unknown, namespace: string): StoredSession {
const existing = this.db.prepare(
'SELECT * FROM sessions WHERE tag = ? ORDER BY created_at DESC LIMIT 1'
).get(tag) as DbSessionRow | undefined
'SELECT * FROM sessions WHERE tag = ? AND namespace = ? ORDER BY created_at DESC LIMIT 1'
).get(tag, namespace) as DbSessionRow | undefined
if (existing) {
return toStoredSession(existing)
@@ -284,13 +314,13 @@ export class Store {
this.db.prepare(`
INSERT INTO sessions (
id, tag, machine_id, created_at, updated_at,
id, tag, namespace, machine_id, created_at, updated_at,
metadata, metadata_version,
agent_state, agent_state_version,
todos, todos_updated_at,
active, active_at, seq
) VALUES (
@id, @tag, NULL, @created_at, @updated_at,
@id, @tag, @namespace, NULL, @created_at, @updated_at,
@metadata, 1,
@agent_state, 1,
NULL, NULL,
@@ -299,6 +329,7 @@ export class Store {
`).run({
id,
tag,
namespace,
created_at: now,
updated_at: now,
metadata: metadataJson,
@@ -312,7 +343,12 @@ export class Store {
return row
}
updateSessionMetadata(id: string, metadata: unknown, expectedVersion: number): VersionedUpdateResult<unknown | null> {
updateSessionMetadata(
id: string,
metadata: unknown,
expectedVersion: number,
namespace: string
): VersionedUpdateResult<unknown | null> {
try {
const now = Date.now()
const json = JSON.stringify(metadata)
@@ -322,14 +358,16 @@ export class Store {
metadata_version = metadata_version + 1,
updated_at = @updated_at,
seq = seq + 1
WHERE id = @id AND metadata_version = @expectedVersion
`).run({ id, metadata: json, updated_at: now, expectedVersion })
WHERE id = @id AND namespace = @namespace AND metadata_version = @expectedVersion
`).run({ id, metadata: json, updated_at: now, expectedVersion, namespace })
if (result.changes === 1) {
return { result: 'success', version: expectedVersion + 1, value: metadata }
}
const current = this.db.prepare('SELECT metadata, metadata_version FROM sessions WHERE id = ?').get(id) as
const current = this.db.prepare(
'SELECT metadata, metadata_version FROM sessions WHERE id = ? AND namespace = ?'
).get(id, namespace) as
| { metadata: string | null; metadata_version: number }
| undefined
if (!current) {
@@ -345,7 +383,12 @@ export class Store {
}
}
updateSessionAgentState(id: string, agentState: unknown, expectedVersion: number): VersionedUpdateResult<unknown | null> {
updateSessionAgentState(
id: string,
agentState: unknown,
expectedVersion: number,
namespace: string
): VersionedUpdateResult<unknown | null> {
try {
const now = Date.now()
const json = agentState === null || agentState === undefined ? null : JSON.stringify(agentState)
@@ -355,14 +398,16 @@ export class Store {
agent_state_version = agent_state_version + 1,
updated_at = @updated_at,
seq = seq + 1
WHERE id = @id AND agent_state_version = @expectedVersion
`).run({ id, agent_state: json, updated_at: now, expectedVersion })
WHERE id = @id AND namespace = @namespace AND agent_state_version = @expectedVersion
`).run({ id, agent_state: json, updated_at: now, expectedVersion, namespace })
if (result.changes === 1) {
return { result: 'success', version: expectedVersion + 1, value: agentState === undefined ? null : agentState }
}
const current = this.db.prepare('SELECT agent_state, agent_state_version FROM sessions WHERE id = ?').get(id) as
const current = this.db.prepare(
'SELECT agent_state, agent_state_version FROM sessions WHERE id = ? AND namespace = ?'
).get(id, namespace) as
| { agent_state: string | null; agent_state_version: number }
| undefined
if (!current) {
@@ -378,7 +423,7 @@ export class Store {
}
}
setSessionTodos(id: string, todos: unknown, todosUpdatedAt: number): boolean {
setSessionTodos(id: string, todos: unknown, todosUpdatedAt: number, namespace: string): boolean {
try {
const json = todos === null || todos === undefined ? null : JSON.stringify(todos)
const result = this.db.prepare(`
@@ -387,12 +432,15 @@ export class Store {
todos_updated_at = @todos_updated_at,
updated_at = CASE WHEN updated_at > @updated_at THEN updated_at ELSE @updated_at END,
seq = seq + 1
WHERE id = @id AND (todos_updated_at IS NULL OR todos_updated_at < @todos_updated_at)
WHERE id = @id
AND namespace = @namespace
AND (todos_updated_at IS NULL OR todos_updated_at < @todos_updated_at)
`).run({
id,
todos: json,
todos_updated_at: todosUpdatedAt,
updated_at: todosUpdatedAt
updated_at: todosUpdatedAt,
namespace
})
return result.changes === 1
@@ -406,15 +454,33 @@ export class Store {
return row ? toStoredSession(row) : null
}
getSessionByNamespace(id: string, namespace: string): StoredSession | null {
const row = this.db.prepare(
'SELECT * FROM sessions WHERE id = ? AND namespace = ?'
).get(id, namespace) as DbSessionRow | undefined
return row ? toStoredSession(row) : null
}
getSessions(): StoredSession[] {
const rows = this.db.prepare('SELECT * FROM sessions ORDER BY updated_at DESC').all() as DbSessionRow[]
return rows.map(toStoredSession)
}
getOrCreateMachine(id: string, metadata: unknown, daemonState: unknown): StoredMachine {
getSessionsByNamespace(namespace: string): StoredSession[] {
const rows = this.db.prepare(
'SELECT * FROM sessions WHERE namespace = ? ORDER BY updated_at DESC'
).all(namespace) as DbSessionRow[]
return rows.map(toStoredSession)
}
getOrCreateMachine(id: string, metadata: unknown, daemonState: unknown, namespace: string): StoredMachine {
const existing = this.db.prepare('SELECT * FROM machines WHERE id = ?').get(id) as DbMachineRow | undefined
if (existing) {
return toStoredMachine(existing)
const stored = toStoredMachine(existing)
if (stored.namespace !== namespace) {
throw new Error('Machine namespace mismatch')
}
return stored
}
const now = Date.now()
@@ -423,18 +489,19 @@ export class Store {
this.db.prepare(`
INSERT INTO machines (
id, created_at, updated_at,
id, namespace, created_at, updated_at,
metadata, metadata_version,
daemon_state, daemon_state_version,
active, active_at, seq
) VALUES (
@id, @created_at, @updated_at,
@id, @namespace, @created_at, @updated_at,
@metadata, 1,
@daemon_state, 1,
0, NULL, 0
)
`).run({
id,
namespace,
created_at: now,
updated_at: now,
metadata: metadataJson,
@@ -448,7 +515,12 @@ export class Store {
return row
}
updateMachineMetadata(id: string, metadata: unknown, expectedVersion: number): VersionedUpdateResult<unknown | null> {
updateMachineMetadata(
id: string,
metadata: unknown,
expectedVersion: number,
namespace: string
): VersionedUpdateResult<unknown | null> {
try {
const now = Date.now()
const json = JSON.stringify(metadata)
@@ -458,14 +530,16 @@ export class Store {
metadata_version = metadata_version + 1,
updated_at = @updated_at,
seq = seq + 1
WHERE id = @id AND metadata_version = @expectedVersion
`).run({ id, metadata: json, updated_at: now, expectedVersion })
WHERE id = @id AND namespace = @namespace AND metadata_version = @expectedVersion
`).run({ id, metadata: json, updated_at: now, expectedVersion, namespace })
if (result.changes === 1) {
return { result: 'success', version: expectedVersion + 1, value: metadata }
}
const current = this.db.prepare('SELECT metadata, metadata_version FROM machines WHERE id = ?').get(id) as
const current = this.db.prepare(
'SELECT metadata, metadata_version FROM machines WHERE id = ? AND namespace = ?'
).get(id, namespace) as
| { metadata: string | null; metadata_version: number }
| undefined
if (!current) {
@@ -481,7 +555,12 @@ export class Store {
}
}
updateMachineDaemonState(id: string, daemonState: unknown, expectedVersion: number): VersionedUpdateResult<unknown | null> {
updateMachineDaemonState(
id: string,
daemonState: unknown,
expectedVersion: number,
namespace: string
): VersionedUpdateResult<unknown | null> {
try {
const now = Date.now()
const json = daemonState === null || daemonState === undefined ? null : JSON.stringify(daemonState)
@@ -493,14 +572,16 @@ export class Store {
active = 1,
active_at = @active_at,
seq = seq + 1
WHERE id = @id AND daemon_state_version = @expectedVersion
`).run({ id, daemon_state: json, updated_at: now, active_at: now, expectedVersion })
WHERE id = @id AND namespace = @namespace AND daemon_state_version = @expectedVersion
`).run({ id, daemon_state: json, updated_at: now, active_at: now, expectedVersion, namespace })
if (result.changes === 1) {
return { result: 'success', version: expectedVersion + 1, value: daemonState === undefined ? null : daemonState }
}
const current = this.db.prepare('SELECT daemon_state, daemon_state_version FROM machines WHERE id = ?').get(id) as
const current = this.db.prepare(
'SELECT daemon_state, daemon_state_version FROM machines WHERE id = ? AND namespace = ?'
).get(id, namespace) as
| { daemon_state: string | null; daemon_state_version: number }
| undefined
if (!current) {
@@ -521,11 +602,25 @@ export class Store {
return row ? toStoredMachine(row) : null
}
getMachineByNamespace(id: string, namespace: string): StoredMachine | null {
const row = this.db.prepare(
'SELECT * FROM machines WHERE id = ? AND namespace = ?'
).get(id, namespace) as DbMachineRow | undefined
return row ? toStoredMachine(row) : null
}
getMachines(): StoredMachine[] {
const rows = this.db.prepare('SELECT * FROM machines ORDER BY updated_at DESC').all() as DbMachineRow[]
return rows.map(toStoredMachine)
}
getMachinesByNamespace(namespace: string): StoredMachine[] {
const rows = this.db.prepare(
'SELECT * FROM machines WHERE namespace = ? ORDER BY updated_at DESC'
).all(namespace) as DbMachineRow[]
return rows.map(toStoredMachine)
}
addMessage(sessionId: string, content: unknown, localId?: string): StoredMessage {
const now = Date.now()
@@ -607,17 +702,25 @@ export class Store {
return rows.map(toStoredUser)
}
addUser(platform: string, platformUserId: string): StoredUser {
getUsersByPlatformAndNamespace(platform: string, namespace: string): StoredUser[] {
const rows = this.db.prepare(
'SELECT * FROM users WHERE platform = ? AND namespace = ? ORDER BY created_at ASC'
).all(platform, namespace) as DbUserRow[]
return rows.map(toStoredUser)
}
addUser(platform: string, platformUserId: string, namespace: string): StoredUser {
const now = Date.now()
this.db.prepare(`
INSERT OR IGNORE INTO users (
platform, platform_user_id, created_at
platform, platform_user_id, namespace, created_at
) VALUES (
@platform, @platform_user_id, @created_at
@platform, @platform_user_id, @namespace, @created_at
)
`).run({
platform,
platform_user_id: platformUserId,
namespace,
created_at: now
})
server/src/store/namespace.test.ts
+29
View File
@@ -0,0 +1,29 @@
import { describe, expect, it } from 'bun:test'
import { Store } from './index'
describe('Store namespace filtering', () => {
it('filters sessions by namespace', () => {
const store = new Store(':memory:')
const sessionAlpha = store.getOrCreateSession('tag', { path: '/alpha' }, null, 'alpha')
const sessionBeta = store.getOrCreateSession('tag', { path: '/beta' }, null, 'beta')
const sessionsAlpha = store.getSessionsByNamespace('alpha')
const ids = sessionsAlpha.map((session) => session.id)
expect(ids).toContain(sessionAlpha.id)
expect(ids).not.toContain(sessionBeta.id)
})
it('filters machines by namespace and blocks mismatches', () => {
const store = new Store(':memory:')
const machineAlpha = store.getOrCreateMachine('machine-1', { host: 'alpha' }, null, 'alpha')
store.getOrCreateMachine('machine-2', { host: 'beta' }, null, 'beta')
const machinesAlpha = store.getMachinesByNamespace('alpha')
const ids = machinesAlpha.map((machine) => machine.id)
expect(ids).toContain(machineAlpha.id)
expect(ids).not.toContain('machine-2')
expect(() => store.getOrCreateMachine('machine-1', { host: 'beta' }, null, 'beta')).toThrow()
})
})
server/src/sync/syncEngine.ts
+63 -6
View File
@@ -72,6 +72,7 @@ const machineMetadataSchema = z.object({
export interface Session {
id: string
namespace: string
seq: number
createdAt: number
updatedAt: number
@@ -90,6 +91,7 @@ export interface Session {
export interface Machine {
id: string
namespace: string
seq: number
createdAt: number
updatedAt: number
@@ -147,6 +149,7 @@ export type SyncEventType =
export interface SyncEvent {
type: SyncEventType
namespace?: string
sessionId?: string
machineId?: string
data?: unknown
@@ -202,9 +205,12 @@ export class SyncEngine {
}
private emit(event: SyncEvent): void {
const namespace = this.resolveNamespace(event)
const enrichedEvent = namespace ? { ...event, namespace } : event
for (const listener of this.listeners) {
try {
listener(event)
listener(enrichedEvent)
} catch (error) {
console.error('[SyncEngine] Listener error:', error)
}
@@ -213,12 +219,14 @@ export class SyncEngine {
const webappEvent: SyncEvent = event.type === 'message-received'
? {
type: event.type,
namespace,
sessionId: event.sessionId,
machineId: event.machineId,
message: event.message
}
: {
type: event.type,
namespace,
sessionId: event.sessionId,
machineId: event.machineId
}
@@ -226,6 +234,19 @@ export class SyncEngine {
this.sseManager.broadcast(webappEvent)
}
private resolveNamespace(event: SyncEvent): string | undefined {
if (event.namespace) {
return event.namespace
}
if (event.sessionId) {
return this.sessions.get(event.sessionId)?.namespace
}
if (event.machineId) {
return this.machines.get(event.machineId)?.namespace
}
return undefined
}
getConnectionStatus(): ConnectionStatus {
return this.connectionStatus
}
@@ -234,10 +255,22 @@ export class SyncEngine {
return Array.from(this.sessions.values())
}
getSessionsByNamespace(namespace: string): Session[] {
return this.getSessions().filter((session) => session.namespace === namespace)
}
getSession(sessionId: string): Session | undefined {
return this.sessions.get(sessionId)
}
getSessionByNamespace(sessionId: string, namespace: string): Session | undefined {
const session = this.sessions.get(sessionId)
if (!session || session.namespace !== namespace) {
return undefined
}
return session
}
getActiveSessions(): Session[] {
return this.getSessions().filter(s => s.active)
}
@@ -246,14 +279,30 @@ export class SyncEngine {
return Array.from(this.machines.values())
}
getMachinesByNamespace(namespace: string): Machine[] {
return this.getMachines().filter((machine) => machine.namespace === namespace)
}
getMachine(machineId: string): Machine | undefined {
return this.machines.get(machineId)
}
getMachineByNamespace(machineId: string, namespace: string): Machine | undefined {
const machine = this.machines.get(machineId)
if (!machine || machine.namespace !== namespace) {
return undefined
}
return machine
}
getOnlineMachines(): Machine[] {
return this.getMachines().filter(m => m.active)
}
getOnlineMachinesByNamespace(namespace: string): Machine[] {
return this.getMachinesByNamespace(namespace).filter((machine) => machine.active)
}
getSessionMessages(sessionId: string): DecryptedMessage[] {
return this.sessionMessages.get(sessionId) || []
}
@@ -320,6 +369,12 @@ export class SyncEngine {
return
}
if (event.type === 'message-received' && event.sessionId) {
if (!this.sessions.has(event.sessionId)) {
this.refreshSession(event.sessionId)
}
}
this.emit(event)
}
@@ -453,7 +508,7 @@ export class SyncEngine {
const message = messages[i]
const todos = extractTodoWriteTodosFromMessageContent(message.content)
if (todos) {
const updated = this.store.setSessionTodos(sessionId, todos, message.createdAt)
const updated = this.store.setSessionTodos(sessionId, todos, message.createdAt, stored.namespace)
if (updated) {
stored = this.store.getSession(sessionId) ?? stored
}
@@ -480,6 +535,7 @@ export class SyncEngine {
const session: Session = {
id: stored.id,
namespace: stored.namespace,
seq: stored.seq,
createdAt: stored.createdAt,
updatedAt: stored.updatedAt,
@@ -530,6 +586,7 @@ export class SyncEngine {
const machine: Machine = {
id: stored.id,
namespace: stored.namespace,
seq: stored.seq,
createdAt: stored.createdAt,
updatedAt: stored.updatedAt,
@@ -558,13 +615,13 @@ export class SyncEngine {
}
}
getOrCreateSession(tag: string, metadata: unknown, agentState: unknown): Session {
const stored = this.store.getOrCreateSession(tag, metadata, agentState)
getOrCreateSession(tag: string, metadata: unknown, agentState: unknown, namespace: string): Session {
const stored = this.store.getOrCreateSession(tag, metadata, agentState, namespace)
return this.refreshSession(stored.id) ?? (() => { throw new Error('Failed to load session') })()
}
getOrCreateMachine(id: string, metadata: unknown, daemonState: unknown): Machine {
const stored = this.store.getOrCreateMachine(id, metadata, daemonState)
getOrCreateMachine(id: string, metadata: unknown, daemonState: unknown, namespace: string): Machine {
const stored = this.store.getOrCreateMachine(id, metadata, daemonState, namespace)
return this.refreshMachine(stored.id) ?? (() => { throw new Error('Failed to load machine') })()
}
server/src/telegram/bot.ts
+19 -4
View File
@@ -167,10 +167,17 @@ export class HappyBot {
return
}
const namespace = this.getNamespaceForChatId(ctx.from?.id ?? null)
if (!namespace) {
await ctx.answerCallbackQuery('Telegram account is not bound')
return
}
const data = ctx.callbackQuery.data
const callbackContext: CallbackContext = {
syncEngine: this.syncEngine,
namespace,
answerCallback: async (text?: string) => {
await ctx.answerCallbackQuery(text)
},
@@ -220,8 +227,8 @@ export class HappyBot {
/**
* Get bound Telegram chat IDs from storage.
*/
private getBoundChatIds(): number[] {
const users = this.store.getUsersByPlatform('telegram')
private getBoundChatIds(namespace: string): number[] {
const users = this.store.getUsersByPlatformAndNamespace('telegram', namespace)
const ids = new Set<number>()
for (const user of users) {
const chatId = Number(user.platformUserId)
@@ -232,6 +239,14 @@ export class HappyBot {
return Array.from(ids)
}
private getNamespaceForChatId(chatId: number | null | undefined): string | null {
if (!chatId) {
return null
}
const stored = this.store.getUser('telegram', String(chatId))
return stored?.namespace ?? null
}
/**
* Send a push notification when agent is ready for input.
*/
@@ -259,7 +274,7 @@ export class HappyBot {
const keyboard = new InlineKeyboard()
.webApp('Open Session', url)
const chatIds = this.getBoundChatIds()
const chatIds = this.getBoundChatIds(session.namespace)
if (chatIds.length === 0) {
return
}
@@ -338,7 +353,7 @@ export class HappyBot {
const text = formatSessionNotification(session)
const keyboard = createNotificationKeyboard(session, this.miniAppUrl)
const chatIds = this.getBoundChatIds()
const chatIds = this.getBoundChatIds(session.namespace)
if (chatIds.length === 0) {
return
}
server/src/telegram/callbacks.ts
+2 -1
View File
@@ -20,6 +20,7 @@ export const ACTIONS = {
*/
export interface CallbackContext {
syncEngine: SyncEngine
namespace: string
answerCallback: (text?: string) => Promise<void>
editMessage: (text: string, keyboard?: InlineKeyboard) => Promise<void>
}
@@ -30,7 +31,7 @@ async function getSessionOrAnswer(
sessionPrefix: string,
options?: { requireActive?: boolean }
): Promise<Session | null> {
const session = findSessionByPrefix(syncEngine.getSessions(), sessionPrefix)
const session = findSessionByPrefix(syncEngine.getSessionsByNamespace(ctx.namespace), sessionPrefix)
if (!session) {
await ctx.answerCallback('Session not found')
return null
server/src/utils/accessToken.test.ts
+26
View File
@@ -0,0 +1,26 @@
import { describe, expect, it } from 'bun:test'
import { DEFAULT_NAMESPACE, parseAccessToken } from './accessToken'
describe('parseAccessToken', () => {
it('defaults namespace when missing', () => {
const parsed = parseAccessToken('token')
expect(parsed).toEqual({ baseToken: 'token', namespace: DEFAULT_NAMESPACE })
})
it('parses namespace suffix', () => {
const parsed = parseAccessToken('token:alice')
expect(parsed).toEqual({ baseToken: 'token', namespace: 'alice' })
})
it('rejects empty namespace', () => {
expect(parseAccessToken('token:')).toBeNull()
})
it('rejects missing base token', () => {
expect(parseAccessToken(':alice')).toBeNull()
})
it('rejects whitespace inside namespace', () => {
expect(parseAccessToken('token: alice')).toBeNull()
})
})
server/src/utils/accessToken.ts
+34
View File
@@ -0,0 +1,34 @@
export const DEFAULT_NAMESPACE = 'default'
export type ParsedAccessToken = {
baseToken: string
namespace: string
}
export function parseAccessToken(raw: string): ParsedAccessToken | null {
if (!raw) {
return null
}
const trimmed = raw.trim()
if (!trimmed) {
return null
}
const separatorIndex = trimmed.lastIndexOf(':')
if (separatorIndex === -1) {
return { baseToken: trimmed, namespace: DEFAULT_NAMESPACE }
}
const baseToken = trimmed.slice(0, separatorIndex)
const namespace = trimmed.slice(separatorIndex + 1)
if (!baseToken || !namespace) {
return null
}
if (baseToken.trim() !== baseToken || namespace.trim() !== namespace) {
return null
}
return { baseToken, namespace }
}
server/src/web/middleware/auth.ts
+4 -1
View File
@@ -5,11 +5,13 @@ import { jwtVerify } from 'jose'
export type WebAppEnv = {
Variables: {
userId: number
namespace: string
}
}
const jwtPayloadSchema = z.object({
uid: z.number()
uid: z.number(),
ns: z.string()
})
export function createAuthMiddleware(jwtSecret: Uint8Array): MiddlewareHandler<WebAppEnv> {
@@ -37,6 +39,7 @@ export function createAuthMiddleware(jwtSecret: Uint8Array): MiddlewareHandler<W
}
c.set('userId', parsed.data.uid)
c.set('namespace', parsed.data.ns)
await next()
return
} catch {
server/src/web/routes/auth.ts
+7 -2
View File
@@ -3,6 +3,7 @@ import { SignJWT } from 'jose'
import { z } from 'zod'
import { configuration } from '../../configuration'
import { safeCompareStrings } from '../../utils/crypto'
import { parseAccessToken } from '../../utils/accessToken'
import { validateTelegramInitData } from '../telegramInitData'
import { getOrCreateOwnerId } from '../ownerId'
import type { WebAppEnv } from '../middleware/auth'
@@ -32,14 +33,17 @@ export function createAuthRoutes(jwtSecret: Uint8Array, store: Store): Hono<WebA
let username: string | undefined
let firstName: string | undefined
let lastName: string | undefined
let namespace: string
// Access Token authentication (CLI_API_TOKEN)
if ('accessToken' in parsed.data) {
if (!safeCompareStrings(parsed.data.accessToken, configuration.cliApiToken)) {
const parsedToken = parseAccessToken(parsed.data.accessToken)
if (!parsedToken || !safeCompareStrings(parsedToken.baseToken, configuration.cliApiToken)) {
return c.json({ error: 'Invalid access token' }, 401)
}
userId = await getOrCreateOwnerId()
firstName = 'Web User'
namespace = parsedToken.namespace
} else {
if (!configuration.telegramEnabled || !configuration.telegramBotToken) {
return c.json({ error: 'Telegram authentication is disabled. Configure TELEGRAM_BOT_TOKEN.' }, 503)
@@ -61,9 +65,10 @@ export function createAuthRoutes(jwtSecret: Uint8Array, store: Store): Hono<WebA
username = result.user.username
firstName = result.user.first_name
lastName = result.user.last_name
namespace = storedUser.namespace
}
const token = await new SignJWT({ uid: userId })
const token = await new SignJWT({ uid: userId, ns: namespace })
.setProtectedHeader({ alg: 'HS256' })
.setIssuedAt()
.setExpirationTime('15m')
server/src/web/routes/bind.ts
+10 -3
View File
@@ -3,6 +3,7 @@ import { SignJWT } from 'jose'
import { z } from 'zod'
import { configuration } from '../../configuration'
import { safeCompareStrings } from '../../utils/crypto'
import { parseAccessToken } from '../../utils/accessToken'
import { validateTelegramInitData } from '../telegramInitData'
import { getOrCreateOwnerId } from '../ownerId'
import type { WebAppEnv } from '../middleware/auth'
@@ -23,9 +24,11 @@ export function createBindRoutes(jwtSecret: Uint8Array, store: Store): Hono<WebA
return c.json({ error: 'Invalid body' }, 400)
}
if (!safeCompareStrings(parsed.data.accessToken, configuration.cliApiToken)) {
const parsedToken = parseAccessToken(parsed.data.accessToken)
if (!parsedToken || !safeCompareStrings(parsedToken.baseToken, configuration.cliApiToken)) {
return c.json({ error: 'Invalid access token' }, 401)
}
const namespace = parsedToken.namespace
if (!configuration.telegramEnabled || !configuration.telegramBotToken) {
return c.json({ error: 'Telegram authentication is disabled. Configure TELEGRAM_BOT_TOKEN.' }, 503)
@@ -37,11 +40,15 @@ export function createBindRoutes(jwtSecret: Uint8Array, store: Store): Hono<WebA
}
const telegramUserId = String(result.user.id)
store.addUser('telegram', telegramUserId)
const existingUser = store.getUser('telegram', telegramUserId)
if (existingUser && existingUser.namespace !== namespace) {
return c.json({ error: 'already_bound' }, 409)
}
store.addUser('telegram', telegramUserId, namespace)
const userId = await getOrCreateOwnerId()
const token = await new SignJWT({ uid: userId })
const token = await new SignJWT({ uid: userId, ns: namespace })
.setProtectedHeader({ alg: 'HS256' })
.setIssuedAt()
.setExpirationTime('15m')
server/src/web/routes/cli.ts
+65 -17
View File
@@ -2,7 +2,8 @@ import { Hono } from 'hono'
import { z } from 'zod'
import { configuration } from '../../configuration'
import { safeCompareStrings } from '../../utils/crypto'
import type { SyncEngine } from '../../sync/syncEngine'
import { parseAccessToken } from '../../utils/accessToken'
import type { Machine, Session, SyncEngine } from '../../sync/syncEngine'
const bearerSchema = z.string().regex(/^Bearer\s+(.+)$/i)
@@ -23,8 +24,44 @@ const getMessagesQuerySchema = z.object({
limit: z.coerce.number().int().min(1).max(200).optional()
})
export function createCliRoutes(getSyncEngine: () => SyncEngine | null): Hono {
const app = new Hono()
type CliEnv = {
Variables: {
namespace: string
}
}
function resolveSessionForNamespace(
engine: SyncEngine,
sessionId: string,
namespace: string
): { ok: true; session: Session } | { ok: false; status: 403 | 404; error: string } {
const session = engine.getSessionByNamespace(sessionId, namespace)
if (session) {
return { ok: true, session }
}
if (engine.getSession(sessionId)) {
return { ok: false, status: 403, error: 'Session access denied' }
}
return { ok: false, status: 404, error: 'Session not found' }
}
function resolveMachineForNamespace(
engine: SyncEngine,
machineId: string,
namespace: string
): { ok: true; machine: Machine } | { ok: false; status: 403 | 404; error: string } {
const machine = engine.getMachineByNamespace(machineId, namespace)
if (machine) {
return { ok: true, machine }
}
if (engine.getMachine(machineId)) {
return { ok: false, status: 403, error: 'Machine access denied' }
}
return { ok: false, status: 404, error: 'Machine not found' }
}
export function createCliRoutes(getSyncEngine: () => SyncEngine | null): Hono<CliEnv> {
const app = new Hono<CliEnv>()
app.use('*', async (c, next) => {
const raw = c.req.header('authorization')
@@ -38,10 +75,12 @@ export function createCliRoutes(getSyncEngine: () => SyncEngine | null): Hono {
}
const token = parsed.data.replace(/^Bearer\s+/i, '')
if (!safeCompareStrings(token, configuration.cliApiToken)) {
const parsedToken = parseAccessToken(token)
if (!parsedToken || !safeCompareStrings(parsedToken.baseToken, configuration.cliApiToken)) {
return c.json({ error: 'Invalid token' }, 401)
}
c.set('namespace', parsedToken.namespace)
return await next()
})
@@ -56,7 +95,8 @@ export function createCliRoutes(getSyncEngine: () => SyncEngine | null): Hono {
return c.json({ error: 'Invalid body' }, 400)
}
const session = engine.getOrCreateSession(parsed.data.tag, parsed.data.metadata, parsed.data.agentState ?? null)
const namespace = c.get('namespace')
const session = engine.getOrCreateSession(parsed.data.tag, parsed.data.metadata, parsed.data.agentState ?? null, namespace)
return c.json({ session })
})
@@ -66,11 +106,12 @@ export function createCliRoutes(getSyncEngine: () => SyncEngine | null): Hono {
return c.json({ error: 'Not ready' }, 503)
}
const sessionId = c.req.param('id')
const session = engine.getSession(sessionId)
if (!session) {
return c.json({ error: 'Session not found' }, 404)
const namespace = c.get('namespace')
const resolved = resolveSessionForNamespace(engine, sessionId, namespace)
if (!resolved.ok) {
return c.json({ error: resolved.error }, resolved.status)
}
return c.json({ session })
return c.json({ session: resolved.session })
})
app.get('/sessions/:id/messages', (c) => {
@@ -79,9 +120,10 @@ export function createCliRoutes(getSyncEngine: () => SyncEngine | null): Hono {
return c.json({ error: 'Not ready' }, 503)
}
const sessionId = c.req.param('id')
const session = engine.getSession(sessionId)
if (!session) {
return c.json({ error: 'Session not found' }, 404)
const namespace = c.get('namespace')
const resolved = resolveSessionForNamespace(engine, sessionId, namespace)
if (!resolved.ok) {
return c.json({ error: resolved.error }, resolved.status)
}
const parsed = getMessagesQuerySchema.safeParse(c.req.query())
@@ -105,7 +147,12 @@ export function createCliRoutes(getSyncEngine: () => SyncEngine | null): Hono {
return c.json({ error: 'Invalid body' }, 400)
}
const machine = engine.getOrCreateMachine(parsed.data.id, parsed.data.metadata, parsed.data.daemonState ?? null)
const namespace = c.get('namespace')
const existing = engine.getMachine(parsed.data.id)
if (existing && existing.namespace !== namespace) {
return c.json({ error: 'Machine access denied' }, 403)
}
const machine = engine.getOrCreateMachine(parsed.data.id, parsed.data.metadata, parsed.data.daemonState ?? null, namespace)
return c.json({ machine })
})
@@ -115,11 +162,12 @@ export function createCliRoutes(getSyncEngine: () => SyncEngine | null): Hono {
return c.json({ error: 'Not ready' }, 503)
}
const machineId = c.req.param('id')
const machine = engine.getMachine(machineId)
if (!machine) {
return c.json({ error: 'Machine not found' }, 404)
const namespace = c.get('namespace')
const resolved = resolveMachineForNamespace(engine, machineId, namespace)
if (!resolved.ok) {
return c.json({ error: resolved.error }, resolved.status)
}
return c.json({ machine })
return c.json({ machine: resolved.machine })
})
return app
server/src/web/routes/events.ts
+32 -1
View File
@@ -2,6 +2,7 @@ import { Hono } from 'hono'
import { streamSSE } from 'hono/streaming'
import { randomUUID } from 'node:crypto'
import type { SSEManager } from '../../sse/sseManager'
import type { SyncEngine } from '../../sync/syncEngine'
import type { WebAppEnv } from '../middleware/auth'
function parseOptionalId(value: string | undefined): string | null {
@@ -18,7 +19,10 @@ function parseBoolean(value: string | undefined): boolean {
return value === 'true' || value === '1'
}
export function createEventsRoutes(getSseManager: () => SSEManager | null): Hono<WebAppEnv> {
export function createEventsRoutes(
getSseManager: () => SSEManager | null,
getSyncEngine: () => SyncEngine | null
): Hono<WebAppEnv> {
const app = new Hono<WebAppEnv>()
app.get('/events', (c) => {
@@ -32,10 +36,37 @@ export function createEventsRoutes(getSseManager: () => SSEManager | null): Hono
const sessionId = parseOptionalId(query.sessionId)
const machineId = parseOptionalId(query.machineId)
const subscriptionId = randomUUID()
const namespace = c.get('namespace')
if (sessionId || machineId) {
const engine = getSyncEngine()
if (!engine) {
return c.json({ error: 'Not connected' }, 503)
}
if (sessionId) {
const session = engine.getSession(sessionId)
if (!session) {
return c.json({ error: 'Session not found' }, 404)
}
if (session.namespace !== namespace) {
return c.json({ error: 'Session access denied' }, 403)
}
}
if (machineId) {
const machine = engine.getMachine(machineId)
if (!machine) {
return c.json({ error: 'Machine not found' }, 404)
}
if (machine.namespace !== namespace) {
return c.json({ error: 'Machine access denied' }, 403)
}
}
}
return streamSSE(c, async (stream) => {
manager.subscribe({
id: subscriptionId,
namespace,
all,
sessionId,
machineId,
server/src/web/routes/guards.ts
+20 -1
View File
@@ -1,5 +1,5 @@
import type { Context } from 'hono'
import type { Session, SyncEngine } from '../../sync/syncEngine'
import type { Machine, Session, SyncEngine } from '../../sync/syncEngine'
import type { WebAppEnv } from '../middleware/auth'
export function requireSyncEngine(
@@ -19,10 +19,14 @@ export function requireSession(
sessionId: string,
options?: { requireActive?: boolean }
): Session | Response {
const namespace = c.get('namespace')
const session = engine.getSession(sessionId)
if (!session) {
return c.json({ error: 'Session not found' }, 404)
}
if (session.namespace !== namespace) {
return c.json({ error: 'Session access denied' }, 403)
}
if (options?.requireActive && !session.active) {
return c.json({ error: 'Session is inactive' }, 409)
}
@@ -43,3 +47,18 @@ export function requireSessionFromParam(
return { sessionId, session }
}
export function requireMachine(
c: Context<WebAppEnv>,
engine: SyncEngine,
machineId: string
): Machine | Response {
const namespace = c.get('namespace')
const machine = engine.getMachine(machineId)
if (!machine) {
return c.json({ error: 'Machine not found' }, 404)
}
if (machine.namespace !== namespace) {
return c.json({ error: 'Machine access denied' }, 403)
}
return machine
}
server/src/web/routes/machines.ts
+9 -7
View File
@@ -2,6 +2,7 @@ import { Hono } from 'hono'
import { z } from 'zod'
import type { SyncEngine } from '../../sync/syncEngine'
import type { WebAppEnv } from '../middleware/auth'
import { requireMachine } from './guards'
const spawnBodySchema = z.object({
directory: z.string().min(1),
@@ -24,7 +25,8 @@ export function createMachinesRoutes(getSyncEngine: () => SyncEngine | null): Ho
return c.json({ error: 'Not connected' }, 503)
}
const machines = engine.getOnlineMachines()
const namespace = c.get('namespace')
const machines = engine.getOnlineMachinesByNamespace(namespace)
return c.json({ machines })
})
@@ -35,9 +37,9 @@ export function createMachinesRoutes(getSyncEngine: () => SyncEngine | null): Ho
}
const machineId = c.req.param('id')
const machine = engine.getMachine(machineId)
if (!machine) {
return c.json({ error: 'Machine not found' }, 404)
const machine = requireMachine(c, engine, machineId)
if (machine instanceof Response) {
return machine
}
const body = await c.req.json().catch(() => null)
@@ -64,9 +66,9 @@ export function createMachinesRoutes(getSyncEngine: () => SyncEngine | null): Ho
}
const machineId = c.req.param('id')
const machine = engine.getMachine(machineId)
if (!machine) {
return c.json({ error: 'Machine not found' }, 404)
const machine = requireMachine(c, engine, machineId)
if (machine instanceof Response) {
return machine
}
const body = await c.req.json().catch(() => null)
server/src/web/routes/sessions.ts
+2 -1
View File
@@ -78,7 +78,8 @@ export function createSessionsRoutes(getSyncEngine: () => SyncEngine | null): Ho
const getPendingCount = (s: Session) => s.agentState?.requests ? Object.keys(s.agentState.requests).length : 0
const sessions = engine.getSessions()
const namespace = c.get('namespace')
const sessions = engine.getSessionsByNamespace(namespace)
.sort((a, b) => {
// Active sessions first
if (a.active !== b.active) {
server/src/web/server.ts
+1 -1
View File
@@ -77,7 +77,7 @@ function createWebApp(options: {
app.route('/api', createBindRoutes(options.jwtSecret, options.store))
app.use('/api/*', createAuthMiddleware(options.jwtSecret))
app.route('/api', createEventsRoutes(options.getSseManager))
app.route('/api', createEventsRoutes(options.getSseManager, options.getSyncEngine))
app.route('/api', createSessionsRoutes(options.getSyncEngine))
app.route('/api', createMessagesRoutes(options.getSyncEngine))
app.route('/api', createPermissionsRoutes(options.getSyncEngine))