1035 lines
54 KiB
JavaScript
1035 lines
54 KiB
JavaScript
import { createServer } from 'node:http'
|
|
import { randomUUID } from 'node:crypto'
|
|
import { chmod, mkdir, readFile, rm } from 'node:fs/promises'
|
|
import { dirname, join } from 'node:path'
|
|
import { UpdateConflictError } from '../../modules/updater/lib/coordinator.mjs'
|
|
import { MAX_SETTINGS_DOCUMENT_BYTES, SettingsDocumentConflictError } from './settings-document.mjs'
|
|
|
|
export const API_PREFIX = '/_dsh_platform/api/v1/'
|
|
export const CONSOLE_PREFIX = '/_dsh_platform/console/'
|
|
const MAX_BODY_BYTES = 16 * 1024
|
|
const MAX_PROXY_BODY_BYTES = 64 * 1024
|
|
const TERMINAL_SESSION_ROUTE = /^terminal\/sessions\/([0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12})$/
|
|
const TERMINAL_STREAM_ROUTE = /^terminal\/sessions\/([0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12})\/stream$/
|
|
const FILE_TASK_ROUTE = /^files\/tasks\/([0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12})$/
|
|
const PROXY_TEST_TASK_ROUTE = /^proxy\/test\/tasks\/([0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12})$/
|
|
const CONSOLE_ASSETS = new Map([
|
|
['', ['public', 'index.html', 'text/html; charset=utf-8']],
|
|
['app.js', ['public', 'app.js', 'text/javascript; charset=utf-8']],
|
|
['theme-init.js', ['public', 'theme-init.js', 'text/javascript; charset=utf-8']],
|
|
['style.css', ['public', 'style.css', 'text/css; charset=utf-8']],
|
|
['vendor/xterm.mjs', ['dependencies', '@xterm/xterm/lib/xterm.mjs', 'text/javascript; charset=utf-8']],
|
|
['vendor/xterm.css', ['dependencies', '@xterm/xterm/css/xterm.css', 'text/css; charset=utf-8']],
|
|
['vendor/addon-fit.mjs', ['dependencies', '@xterm/addon-fit/lib/addon-fit.mjs', 'text/javascript; charset=utf-8']],
|
|
])
|
|
const CONSOLE_HEADERS = Object.freeze({
|
|
'cache-control': 'no-store',
|
|
'content-security-policy': "default-src 'self'; script-src 'self'; style-src 'self' 'unsafe-inline'; connect-src 'self' http: https:; img-src 'self' data:; base-uri 'none'; frame-ancestors 'self'; form-action 'none'",
|
|
'referrer-policy': 'no-referrer',
|
|
'x-content-type-options': 'nosniff',
|
|
})
|
|
|
|
async function jsonBody(request, maxBytes = MAX_BODY_BYTES) {
|
|
const chunks = []
|
|
let size = 0
|
|
for await (const chunk of request) {
|
|
size += chunk.byteLength
|
|
if (size > maxBytes) throw new Error('request body is too large')
|
|
chunks.push(chunk)
|
|
}
|
|
if (size === 0) return {}
|
|
return JSON.parse(Buffer.concat(chunks).toString('utf8'))
|
|
}
|
|
|
|
function send(response, status, value) {
|
|
if (response.headersSent) return response.destroy()
|
|
response.writeHead(status, { 'content-type': 'application/json; charset=utf-8' })
|
|
response.end(`${JSON.stringify(value)}\n`)
|
|
}
|
|
|
|
async function sendConsoleAsset(request, response, pathname, consoleRoots) {
|
|
if (pathname === CONSOLE_PREFIX.slice(0, -1)) {
|
|
response.writeHead(308, { location: CONSOLE_PREFIX, ...CONSOLE_HEADERS })
|
|
response.end()
|
|
return true
|
|
}
|
|
if (!pathname.startsWith(CONSOLE_PREFIX)) return false
|
|
if (!['GET', 'HEAD'].includes(request.method ?? 'GET')) {
|
|
response.writeHead(405, { allow: 'GET, HEAD', ...CONSOLE_HEADERS })
|
|
response.end()
|
|
return true
|
|
}
|
|
const asset = CONSOLE_ASSETS.get(pathname.slice(CONSOLE_PREFIX.length))
|
|
if (asset === undefined) {
|
|
response.writeHead(404, CONSOLE_HEADERS)
|
|
response.end()
|
|
return true
|
|
}
|
|
const body = await readFile(join(consoleRoots[asset[0]], asset[1]))
|
|
response.writeHead(200, {
|
|
...CONSOLE_HEADERS,
|
|
'content-type': asset[2],
|
|
'content-length': String(body.byteLength),
|
|
})
|
|
response.end(request.method === 'HEAD' ? undefined : body)
|
|
return true
|
|
}
|
|
|
|
function event(response, type, value) {
|
|
response.write(`event: ${type}\ndata: ${JSON.stringify(value)}\n\n`)
|
|
}
|
|
|
|
function logOptions(url) {
|
|
const sources = url.searchParams.getAll('source')
|
|
const since = url.searchParams.get('since') ?? undefined
|
|
const taskId = url.searchParams.get('taskId') ?? undefined
|
|
const operation = url.searchParams.get('operation') ?? undefined
|
|
const phase = url.searchParams.get('phase') ?? undefined
|
|
const limitValue = url.searchParams.get('limit')
|
|
return {
|
|
sources: sources.length === 0 ? undefined : sources,
|
|
since,
|
|
taskId,
|
|
operation,
|
|
phase,
|
|
limit: limitValue === null ? 200 : Number(limitValue),
|
|
}
|
|
}
|
|
|
|
export function createManagementServer({
|
|
coordinator,
|
|
logs,
|
|
platformStatus = async () => ({}),
|
|
startDsh = async () => { throw new Error('DSH start is not configured') },
|
|
stopDsh = async () => { throw new Error('DSH stop is not configured') },
|
|
restartDsh = async () => { throw new Error('DSH restart is not configured') },
|
|
resetRuntime = async () => { throw new Error('Runtime reset is not configured') },
|
|
listBundledPlugins = async () => [],
|
|
configureBundledPlugin = async () => { throw new Error('System Plugin management is not configured') },
|
|
recoverBundledPlugin = async () => { throw new Error('System Plugin recovery is not configured') },
|
|
discardBundledPluginChanges = async () => { throw new Error('System Plugin draft management is not configured') },
|
|
listSystemSkills = async () => [],
|
|
configureSystemSkill = async () => { throw new Error('System Skill management is not configured') },
|
|
listUserSkills = async () => ({ schema: 1, revision: 'sha256:empty', skills: [] }),
|
|
validateUserSkillAction = async value => value,
|
|
configureUserSkill = async () => { throw new Error('User Skill management is not configured') },
|
|
listUserPlugins = async () => { throw new Error('User Plugin inventory is not configured') },
|
|
markUserPluginsLoaded = async () => {},
|
|
validateUserPluginActions = async () => { throw new Error('User Plugin recovery is not configured') },
|
|
applyUserPluginActions = async () => { throw new Error('User Plugin recovery is not configured') },
|
|
recoverUserPluginTransaction,
|
|
initialUserPluginTransaction,
|
|
settingsDocument,
|
|
updateAutomaticCheck = async () => { throw new Error('automatic checks are not configured') },
|
|
createManagementOriginTransition = async () => { throw new Error('Management origin transition is not configured') },
|
|
commitManagementOriginTransition = async () => { throw new Error('Management origin transition is not configured') },
|
|
reportManagementOriginTransitionFailure = async () => { throw new Error('Management origin transition reporting is not configured') },
|
|
getAuthenticationSettings = async () => { throw new Error('Authentication settings are not configured') },
|
|
updateAuthenticationSettings = async () => { throw new Error('Authentication settings are not configured') },
|
|
beginTotpEnrollment = async () => { throw new Error('Two-factor enrollment is not configured') },
|
|
confirmTotpEnrollment = async () => { throw new Error('Two-factor enrollment is not configured') },
|
|
cancelTotpEnrollment = async () => { throw new Error('Two-factor enrollment is not configured') },
|
|
beginTotpDisableConfirmation = async () => { throw new Error('Two-factor disable confirmation is not configured') },
|
|
confirmTotpDisable = async () => { throw new Error('Two-factor disable confirmation is not configured') },
|
|
cancelTotpDisableConfirmation = async () => { throw new Error('Two-factor disable confirmation is not configured') },
|
|
revokeAuthenticationSessions = async () => { throw new Error('Authentication session management is not configured') },
|
|
getProxyConfiguration = async () => { throw new Error('outbound proxy configuration is not configured') },
|
|
updateProxyConfiguration = async () => { throw new Error('outbound proxy configuration is not configured') },
|
|
listProxyProviders = async () => ({ schema: 1, source: 'unavailable', providers: [] }),
|
|
startProxyTest = async () => { throw new Error('outbound proxy tests are not configured') },
|
|
getProxyTest = async () => { throw new Error('outbound proxy tests are not configured') },
|
|
cancelProxyTest = async () => { throw new Error('outbound proxy tests are not configured') },
|
|
terminalSessions = {
|
|
create: () => { throw new Error('terminal sessions are not configured') },
|
|
status: () => { throw new Error('terminal sessions are not configured') },
|
|
close: () => { throw new Error('terminal sessions are not configured') },
|
|
upgrade: (_request, socket) => socket.destroy(),
|
|
},
|
|
fileInventory,
|
|
fileTransfers,
|
|
fileTasks,
|
|
fileEditor,
|
|
fileLocations = Object.freeze({ defaultPath: '/workspace', shortcuts: Object.freeze(['/workspace', '/data/dsh', '/data/platform', '/']) }),
|
|
privilegedMutationActive = () => false,
|
|
authorizeInternal = async () => true,
|
|
stateHeartbeatMs = 15_000,
|
|
restartDelayMs = 2_000,
|
|
consoleRoot = join(import.meta.dirname, 'public'),
|
|
consoleDependencyRoot = join(import.meta.dirname, 'node_modules'),
|
|
}) {
|
|
let lifecycleTask
|
|
let runtimeResetTask
|
|
let pluginTask
|
|
let systemSkillTask
|
|
let userSkillTask
|
|
let userPluginTask
|
|
let proxyTestState = Object.freeze({ status: 'idle', taskId: null, currentStage: null, error: null, updatedAt: null })
|
|
let proxyWatchClosed = false
|
|
let lifecycleState = Object.freeze({
|
|
state: 'running', action: null, taskId: null, attempt: 0, maxAttempts: 3,
|
|
error: null, updatedAt: null,
|
|
})
|
|
let runtimeResetState = Object.freeze({ status: 'idle', taskId: null, error: null, updatedAt: null })
|
|
let pluginState = Object.freeze({ status: 'idle', taskId: null, pluginId: null, action: null, error: null, restartRequired: false, updatedAt: null })
|
|
let systemSkillState = Object.freeze({ status: 'idle', taskId: null, skillId: null, action: null, error: null, updatedAt: null })
|
|
let userSkillState = Object.freeze({ status: 'idle', taskId: null, entryId: null, action: null, error: null, updatedAt: null })
|
|
let userPluginState = Object.freeze(initialUserPluginTransaction === undefined
|
|
? { status: 'idle', taskId: null, phase: null, error: null, updatedAt: null }
|
|
: {
|
|
status: initialUserPluginTransaction.phase === 'completed' ? 'success'
|
|
: initialUserPluginTransaction.phase === 'failed' ? 'failed' : 'running',
|
|
taskId: initialUserPluginTransaction.taskId,
|
|
phase: initialUserPluginTransaction.phase,
|
|
error: initialUserPluginTransaction.error,
|
|
updatedAt: initialUserPluginTransaction.updatedAt,
|
|
})
|
|
const userPluginTasks = new Map()
|
|
const pluginTasks = new Map()
|
|
if (userPluginState.taskId !== null) userPluginTasks.set(userPluginState.taskId, userPluginState)
|
|
let server
|
|
coordinator.setMaxListeners?.(0)
|
|
const audit = (message, fields = {}) => logs.diagnostic('audit', message, { stream: 'audit', ...fields })
|
|
const recordAudit = (message, fields = {}) => Promise.resolve().then(() => audit(message, fields)).catch(() => {})
|
|
const refreshLoadedUserPlugins = () => Promise.resolve()
|
|
.then(() => markUserPluginsLoaded())
|
|
.catch(error => recordAudit('user-plugin.loaded-state.capture.failed', { error }))
|
|
|
|
const publishLifecycle = value => {
|
|
lifecycleState = Object.freeze({ ...lifecycleState, ...value, updatedAt: new Date().toISOString() })
|
|
server.emit('management-state', lifecycleState)
|
|
}
|
|
|
|
const currentLifecycle = platformLifecycle => {
|
|
if (lifecycleTask !== undefined || platformLifecycle === undefined) return lifecycleState
|
|
const localTime = Date.parse(lifecycleState.updatedAt ?? '')
|
|
const platformTime = Date.parse(platformLifecycle.updatedAt ?? '')
|
|
return Number.isFinite(platformTime) && (
|
|
!Number.isFinite(localTime)
|
|
|| platformTime > localTime
|
|
|| (platformTime === localTime && platformLifecycle.state !== lifecycleState.state)
|
|
)
|
|
? platformLifecycle
|
|
: lifecycleState
|
|
}
|
|
|
|
const publishRuntimeReset = value => {
|
|
runtimeResetState = Object.freeze({ ...runtimeResetState, ...value, updatedAt: new Date().toISOString() })
|
|
server.emit('management-state', runtimeResetState)
|
|
}
|
|
|
|
const publishPlugin = value => {
|
|
pluginState = Object.freeze({ ...pluginState, ...value, updatedAt: new Date().toISOString() })
|
|
if (pluginState.taskId !== null) {
|
|
pluginTasks.set(pluginState.taskId, pluginState)
|
|
while (pluginTasks.size > 32) pluginTasks.delete(pluginTasks.keys().next().value)
|
|
}
|
|
server.emit('management-state', pluginState)
|
|
}
|
|
|
|
const publishSystemSkill = value => {
|
|
systemSkillState = Object.freeze({ ...systemSkillState, ...value, updatedAt: new Date().toISOString() })
|
|
server.emit('management-state', systemSkillState)
|
|
}
|
|
|
|
const publishUserSkill = value => {
|
|
userSkillState = Object.freeze({ ...userSkillState, ...value, updatedAt: new Date().toISOString() })
|
|
server.emit('management-state', userSkillState)
|
|
}
|
|
|
|
const publishUserPlugin = value => {
|
|
userPluginState = Object.freeze({ ...userPluginState, ...value, updatedAt: new Date().toISOString() })
|
|
if (userPluginState.taskId !== null) {
|
|
userPluginTasks.set(userPluginState.taskId, userPluginState)
|
|
while (userPluginTasks.size > 32) userPluginTasks.delete(userPluginTasks.keys().next().value)
|
|
}
|
|
server.emit('management-state', userPluginState)
|
|
}
|
|
|
|
const publishProxyTest = value => {
|
|
proxyTestState = Object.freeze({ ...value, updatedAt: value.updatedAt ?? new Date().toISOString() })
|
|
server.emit('management-state', { proxyTestOperation: proxyTestState })
|
|
}
|
|
|
|
const watchProxyTest = (taskId, initialState) => {
|
|
void (async () => {
|
|
let lastIdentity = initialState === undefined ? '' : JSON.stringify(initialState)
|
|
const stageStatuses = new Map()
|
|
if (initialState !== undefined) {
|
|
for (const stage of initialState.stages ?? []) {
|
|
if (stage.status === 'pending') continue
|
|
stageStatuses.set(stage.stage, stage.status)
|
|
await recordAudit('proxy.test.stage.changed', {
|
|
taskId,
|
|
stage: stage.stage,
|
|
status: stage.status,
|
|
durationMs: stage.durationMs ?? null,
|
|
errorCode: stage.errorCode ?? null,
|
|
})
|
|
}
|
|
}
|
|
while (!proxyWatchClosed) {
|
|
let state
|
|
try { state = await getProxyTest(taskId) } catch (error) {
|
|
publishProxyTest({
|
|
status: 'failed', taskId, currentStage: null,
|
|
error: error?.proxyError ?? {
|
|
code: error?.code ?? 'PROXY_MANAGER_UNAVAILABLE',
|
|
message: error instanceof Error ? error.message : 'proxy test state is unavailable',
|
|
stage: error?.stage ?? 'test', retryable: error?.retryable === true,
|
|
},
|
|
})
|
|
await recordAudit('proxy.test.failed', { taskId, code: error?.code ?? 'PROXY_MANAGER_UNAVAILABLE' })
|
|
return
|
|
}
|
|
const identity = JSON.stringify(state)
|
|
if (identity !== lastIdentity) {
|
|
lastIdentity = identity
|
|
publishProxyTest(state)
|
|
for (const stage of state.stages ?? []) {
|
|
if (stage.status === 'pending' || stageStatuses.get(stage.stage) === stage.status) continue
|
|
stageStatuses.set(stage.stage, stage.status)
|
|
await recordAudit('proxy.test.stage.changed', {
|
|
taskId,
|
|
stage: stage.stage,
|
|
status: stage.status,
|
|
durationMs: stage.durationMs ?? null,
|
|
errorCode: stage.errorCode ?? null,
|
|
})
|
|
}
|
|
}
|
|
if (state.status !== 'running') {
|
|
await recordAudit(`proxy.test.${state.status}`, {
|
|
taskId, mode: state.mode, errorCode: state.error?.errorCode ?? null,
|
|
})
|
|
return
|
|
}
|
|
await new Promise(resolve => setTimeout(resolve, 250))
|
|
}
|
|
})().catch(error => recordAudit('proxy.test.watch.failed', { error, taskId }))
|
|
}
|
|
|
|
const requireRuntimeIdle = () => {
|
|
if (lifecycleTask !== undefined) throw new UpdateConflictError('a DSH lifecycle operation is already running')
|
|
if (runtimeResetTask !== undefined) throw new UpdateConflictError('the DSH Runtime is already resetting')
|
|
if (pluginTask !== undefined) throw new UpdateConflictError('a System Plugin operation is already running')
|
|
if (systemSkillTask !== undefined) throw new UpdateConflictError('a System Skill operation is already running')
|
|
if (userSkillTask !== undefined) throw new UpdateConflictError('a User Skill operation is already running')
|
|
if (userPluginTask !== undefined) throw new UpdateConflictError('a User Plugin operation is already running')
|
|
if (fileTasks?.hasManagedMutation === true) throw new UpdateConflictError('a managed file operation is already running')
|
|
if (privilegedMutationActive()) throw new UpdateConflictError('a privileged file operation is already running')
|
|
}
|
|
|
|
const startUserPluginAction = async body => {
|
|
if (body === null || typeof body !== 'object' || Array.isArray(body)
|
|
|| Object.keys(body).sort().join(',') !== 'actions,profile,revision'
|
|
|| body.profile !== 'web') throw new Error('User Plugin request is invalid')
|
|
requireRuntimeIdle()
|
|
if (coordinator.hasActiveTask?.() === true) throw new UpdateConflictError('an update task is already running')
|
|
const validated = await validateUserPluginActions({ revision: body.revision, actions: body.actions })
|
|
const platform = await platformStatus()
|
|
const strategy = platform.recoveryMode !== null && platform.recoveryMode !== undefined
|
|
|| ['stopped', 'failed'].includes(platform.dshLifecycle?.state)
|
|
? 'next-start' : 'restart'
|
|
requireRuntimeIdle()
|
|
if (coordinator.hasActiveTask?.() === true) throw new UpdateConflictError('an update task is already running')
|
|
const taskId = randomUUID()
|
|
publishUserPlugin({ status: 'running', taskId, phase: 'validated', strategy: null, error: null })
|
|
userPluginTask = Promise.resolve()
|
|
.then(() => recordAudit('user-plugin.apply.started', { taskId, actions: validated.actions }))
|
|
.then(() => applyUserPluginActions({
|
|
taskId,
|
|
revision: validated.revision,
|
|
actions: validated.actions,
|
|
strategy,
|
|
onProgress: state => publishUserPlugin({ phase: state.phase }),
|
|
}))
|
|
.then(
|
|
async result => {
|
|
await refreshLoadedUserPlugins()
|
|
const completedStrategy = result?.strategy ?? strategy
|
|
await recordAudit('user-plugin.apply.completed', { taskId, strategy: completedStrategy })
|
|
publishUserPlugin({
|
|
status: 'success', taskId, phase: 'completed', strategy: completedStrategy,
|
|
activationError: result?.activationError ?? null, error: null,
|
|
})
|
|
publishPlugin({ restartRequired: false })
|
|
},
|
|
async error => {
|
|
const journal = error?.journal
|
|
await recordAudit('user-plugin.apply.failed', { error, taskId })
|
|
publishUserPlugin({
|
|
status: 'failed', taskId, phase: journal?.phase ?? userPluginState.phase,
|
|
error: error instanceof Error ? error.message : 'User Plugin operation failed',
|
|
})
|
|
},
|
|
)
|
|
.finally(() => { userPluginTask = undefined })
|
|
userPluginTask.catch(() => {})
|
|
return { taskId }
|
|
}
|
|
|
|
const startPluginAction = (pluginId, action, { recovery = false, toggleOnly = false } = {}) => {
|
|
if (typeof pluginId !== 'string' || !/^[a-z0-9][a-z0-9._-]{0,127}$/.test(pluginId)) {
|
|
throw new Error('System Plugin ID is invalid')
|
|
}
|
|
if (!['install', 'uninstall', 'enable', 'disable'].includes(action)) throw new Error('System Plugin action is invalid')
|
|
if (toggleOnly && !['enable', 'disable'].includes(action)) throw new Error('Only System Plugin enable and disable actions are allowed')
|
|
if (recovery && pluginId !== 'platform-management') throw new Error('System Plugin is not a console recovery target')
|
|
requireRuntimeIdle()
|
|
if (coordinator.hasActiveTask?.() === true) throw new UpdateConflictError('an update task is already running')
|
|
const taskId = randomUUID()
|
|
publishPlugin({
|
|
status: 'running', taskId, pluginId, action,
|
|
strategy: null, activationError: null, error: null,
|
|
})
|
|
pluginTask = Promise.resolve()
|
|
.then(() => recordAudit(`system-plugin.${recovery ? 'recovery.' : ''}${action}.started`, { taskId, pluginId }))
|
|
.then(() => recovery ? recoverBundledPlugin(pluginId, action) : configureBundledPlugin(pluginId, action))
|
|
.then(
|
|
async () => {
|
|
const currentPlatformStatus = await platformStatus().catch(() => ({}))
|
|
const unavailable = (currentPlatformStatus?.recoveryMode !== null
|
|
&& currentPlatformStatus?.recoveryMode !== undefined)
|
|
|| ['stopped', 'failed'].includes(currentPlatformStatus?.dshLifecycle?.state)
|
|
const strategy = unavailable ? 'next-start' : 'restart'
|
|
await recordAudit(`system-plugin.${recovery ? 'recovery.' : ''}${action}.completed`, {
|
|
taskId, pluginId, strategy,
|
|
})
|
|
publishPlugin({
|
|
status: 'success', taskId, pluginId, action, strategy,
|
|
activationError: null, error: null, restartRequired: !unavailable,
|
|
})
|
|
},
|
|
async error => {
|
|
const message = error instanceof Error ? error.message : 'System Plugin operation failed'
|
|
await recordAudit(`system-plugin.${recovery ? 'recovery.' : ''}${action}.failed`, { error, taskId, pluginId })
|
|
publishPlugin({ status: 'failed', taskId, pluginId, action, error: message })
|
|
},
|
|
)
|
|
.finally(() => { pluginTask = undefined })
|
|
pluginTask.catch(() => {})
|
|
return { taskId }
|
|
}
|
|
|
|
const startSystemSkillAction = (skillId, action) => {
|
|
if (typeof skillId !== 'string' || !/^[a-z0-9]+(?:-[a-z0-9]+)*$/.test(skillId) || skillId.length > 64) {
|
|
throw new Error('System Skill ID is invalid')
|
|
}
|
|
if (!['install', 'uninstall', 'enable', 'disable'].includes(action)) throw new Error('System Skill action is invalid')
|
|
requireRuntimeIdle()
|
|
if (coordinator.hasActiveTask?.() === true) throw new UpdateConflictError('an update task is already running')
|
|
const taskId = randomUUID()
|
|
publishSystemSkill({ status: 'running', taskId, skillId, action, error: null })
|
|
systemSkillTask = Promise.resolve()
|
|
.then(() => recordAudit(`system-skill.${action}.started`, { taskId, skillId }))
|
|
.then(() => configureSystemSkill(skillId, action))
|
|
.then(
|
|
async () => {
|
|
await recordAudit(`system-skill.${action}.completed`, { taskId, skillId })
|
|
publishSystemSkill({ status: 'success', taskId, skillId, action, error: null })
|
|
},
|
|
async error => {
|
|
const message = error instanceof Error ? error.message : 'System Skill operation failed'
|
|
await recordAudit(`system-skill.${action}.failed`, { error, taskId, skillId })
|
|
publishSystemSkill({ status: 'failed', taskId, skillId, action, error: message })
|
|
},
|
|
)
|
|
.finally(() => { systemSkillTask = undefined })
|
|
systemSkillTask.catch(() => {})
|
|
return { taskId }
|
|
}
|
|
|
|
const startUserSkillAction = async body => {
|
|
if (body === null || typeof body !== 'object' || Array.isArray(body)
|
|
|| Object.keys(body).sort().join(',') !== 'action,entryId,revision') {
|
|
throw new Error('User Skill request is invalid')
|
|
}
|
|
if (typeof body.entryId !== 'string' || !/^sha256:[0-9a-f]{64}$/.test(body.entryId)) throw new Error('User Skill entry ID is invalid')
|
|
if (typeof body.revision !== 'string' || !/^sha256:[0-9a-f]{64}$/.test(body.revision)) throw new Error('User Skill revision is invalid')
|
|
if (!['enable', 'disable', 'delete'].includes(body.action)) throw new Error('User Skill action is invalid')
|
|
requireRuntimeIdle()
|
|
if (coordinator.hasActiveTask?.() === true) throw new UpdateConflictError('an update task is already running')
|
|
await validateUserSkillAction(body)
|
|
requireRuntimeIdle()
|
|
if (coordinator.hasActiveTask?.() === true) throw new UpdateConflictError('an update task is already running')
|
|
const taskId = randomUUID()
|
|
publishUserSkill({ status: 'running', taskId, entryId: body.entryId, action: body.action, error: null })
|
|
userSkillTask = Promise.resolve()
|
|
.then(() => recordAudit(`user-skill.${body.action}.started`, { taskId, entryId: body.entryId }))
|
|
.then(() => configureUserSkill(body))
|
|
.then(
|
|
async () => {
|
|
await recordAudit(`user-skill.${body.action}.completed`, { taskId, entryId: body.entryId })
|
|
publishUserSkill({ status: 'success', taskId, entryId: body.entryId, action: body.action, error: null })
|
|
},
|
|
async error => {
|
|
const message = error instanceof Error ? error.message : 'User Skill operation failed'
|
|
await recordAudit(`user-skill.${body.action}.failed`, { error, taskId, entryId: body.entryId })
|
|
publishUserSkill({ status: 'failed', taskId, entryId: body.entryId, action: body.action, error: message })
|
|
},
|
|
)
|
|
.finally(() => { userSkillTask = undefined })
|
|
userSkillTask.catch(() => {})
|
|
return { taskId }
|
|
}
|
|
|
|
const startLifecycle = action => {
|
|
if (!['start', 'stop', 'restart'].includes(action)) throw new Error('DSH lifecycle action is invalid')
|
|
requireRuntimeIdle()
|
|
if (coordinator.hasActiveTask?.() === true) throw new UpdateConflictError('an update task is already running')
|
|
const taskId = randomUUID()
|
|
const state = action === 'start' ? 'starting' : action === 'stop' ? 'stopping' : 'restarting'
|
|
const operation = action === 'start' ? startDsh : action === 'stop' ? stopDsh : restartDsh
|
|
publishLifecycle({ state, action, taskId, attempt: 0, error: null })
|
|
lifecycleTask = Promise.resolve()
|
|
.then(() => recordAudit(`dsh.${action}.started`, { taskId }))
|
|
.then(() => action === 'restart' && restartDelayMs > 0
|
|
? new Promise(resolve => setTimeout(resolve, restartDelayMs))
|
|
: undefined)
|
|
.then(() => operation())
|
|
.then(
|
|
async () => {
|
|
await recordAudit(`dsh.${action}.completed`, { taskId })
|
|
publishLifecycle({ state: action === 'stop' ? 'stopped' : 'running', action: null, taskId, attempt: 0, error: null })
|
|
if (action !== 'stop') publishPlugin({ restartRequired: false })
|
|
// DSH readiness is the restart completion boundary. Refreshing the
|
|
// user-plugin inventory can be slow on mounted or CI filesystems and
|
|
// must not keep an already-ready Runtime in the restarting state.
|
|
if (action !== 'stop') void refreshLoadedUserPlugins()
|
|
},
|
|
async error => {
|
|
const message = error instanceof Error ? error.message : `DSH ${action} failed`
|
|
await recordAudit(`dsh.${action}.failed`, { error, taskId })
|
|
publishLifecycle({ state: 'failed', action: null, taskId, error: message })
|
|
if (action === 'restart' && pluginState.restartRequired) {
|
|
publishPlugin({ strategy: 'next-start', activationError: message, restartRequired: false })
|
|
}
|
|
},
|
|
)
|
|
.finally(() => { lifecycleTask = undefined })
|
|
lifecycleTask.catch(() => {})
|
|
return { taskId }
|
|
}
|
|
|
|
const startRuntimeReset = () => {
|
|
requireRuntimeIdle()
|
|
if (coordinator.hasActiveTask?.() === true) throw new UpdateConflictError('an update task is already running')
|
|
const taskId = randomUUID()
|
|
publishRuntimeReset({ status: 'resetting', taskId, error: null })
|
|
runtimeResetTask = Promise.resolve()
|
|
.then(() => recordAudit('runtime.reset.started', { taskId }))
|
|
.then(() => resetRuntime())
|
|
.then(
|
|
async () => {
|
|
await refreshLoadedUserPlugins()
|
|
await recordAudit('runtime.reset.completed', { taskId })
|
|
publishRuntimeReset({ status: 'success', taskId, error: null })
|
|
publishPlugin({ restartRequired: false })
|
|
},
|
|
async error => {
|
|
const message = error instanceof Error ? error.message : 'Runtime reset failed'
|
|
await recordAudit('runtime.reset.failed', { error, taskId })
|
|
publishRuntimeReset({ status: 'failed', taskId, error: message })
|
|
},
|
|
)
|
|
.finally(() => { runtimeResetTask = undefined })
|
|
runtimeResetTask.catch(() => {})
|
|
return { taskId }
|
|
}
|
|
|
|
server = createServer(async (request, response) => {
|
|
let pathname = 'invalid-url'
|
|
try {
|
|
const url = new URL(request.url ?? '/', 'http://management.internal')
|
|
pathname = url.pathname
|
|
const internalHealth = pathname === '/_dsh_platform/internal/health'
|
|
if (internalHealth) {
|
|
if (request.method !== 'GET') return send(response, 405, { error: 'method not allowed' })
|
|
return send(response, 200, { ready: true })
|
|
}
|
|
if (!await authorizeInternal(request)) return send(response, 401, {
|
|
error: 'management capability required', code: 'CAPABILITY_REQUIRED',
|
|
})
|
|
if (await sendConsoleAsset(request, response, url.pathname, {
|
|
public: consoleRoot,
|
|
dependencies: consoleDependencyRoot,
|
|
})) return
|
|
if (!url.pathname.startsWith(API_PREFIX)) return send(response, 404, { error: 'not found' })
|
|
const route = url.pathname.slice(API_PREFIX.length)
|
|
if (request.method === 'GET' && route === 'status') {
|
|
const currentPlatformStatus = await platformStatus()
|
|
send(response, 200, {
|
|
...(await coordinator.publicStatus()),
|
|
...currentPlatformStatus,
|
|
dshLifecycle: currentLifecycle(currentPlatformStatus.dshLifecycle),
|
|
runtimeReset: runtimeResetState,
|
|
systemPluginOperation: pluginState,
|
|
systemSkillOperation: systemSkillState,
|
|
userSkillOperation: userSkillState,
|
|
userPluginOperation: userPluginState,
|
|
proxyTestOperation: proxyTestState,
|
|
})
|
|
} else if (request.method === 'GET' && route === 'bundled-plugins') {
|
|
send(response, 200, { plugins: await listBundledPlugins() })
|
|
} else if (request.method === 'GET' && /^bundled-plugins\/task\/[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/.test(route)) {
|
|
const taskId = route.slice('bundled-plugins/task/'.length)
|
|
const task = pluginTasks.get(taskId)
|
|
if (task === undefined) {
|
|
const error = new Error('System Plugin task was not found')
|
|
error.statusCode = 404
|
|
throw error
|
|
}
|
|
send(response, 200, task)
|
|
} else if (request.method === 'GET' && route === 'system-skills') {
|
|
send(response, 200, { skills: await listSystemSkills() })
|
|
} else if (request.method === 'GET' && route === 'user-skills') {
|
|
send(response, 200, await listUserSkills())
|
|
} else if (request.method === 'GET' && route === 'user-plugins') {
|
|
send(response, 200, await listUserPlugins())
|
|
} else if (request.method === 'GET' && route === 'proxy') {
|
|
send(response, 200, await getProxyConfiguration())
|
|
} else if (request.method === 'PUT' && route === 'proxy') {
|
|
const body = await jsonBody(request, MAX_PROXY_BODY_BYTES)
|
|
await recordAudit('proxy.configuration.update.started', {
|
|
baseRevision: typeof body?.baseRevision === 'string' ? body.baseRevision : null,
|
|
})
|
|
try {
|
|
const configured = await updateProxyConfiguration(body)
|
|
await recordAudit('proxy.configuration.update.completed', { revision: configured.revision })
|
|
send(response, 200, configured)
|
|
} catch (error) {
|
|
await recordAudit('proxy.configuration.update.failed', {
|
|
code: error?.code ?? 'PROXY_MANAGER_UNAVAILABLE',
|
|
stage: error?.stage ?? 'unknown',
|
|
})
|
|
throw error
|
|
}
|
|
} else if (request.method === 'GET' && route === 'proxy/provider-inventory') {
|
|
send(response, 200, await listProxyProviders())
|
|
} else if (request.method === 'POST' && route === 'proxy/test') {
|
|
const task = await startProxyTest(await jsonBody(request, MAX_PROXY_BODY_BYTES))
|
|
await recordAudit('proxy.test.started', { taskId: task.taskId })
|
|
publishProxyTest(task)
|
|
watchProxyTest(task.taskId, task)
|
|
send(response, 202, { taskId: task.taskId })
|
|
} else if (request.method === 'GET' && PROXY_TEST_TASK_ROUTE.test(route)) {
|
|
const state = await getProxyTest(PROXY_TEST_TASK_ROUTE.exec(route)[1])
|
|
publishProxyTest(state)
|
|
send(response, 200, state)
|
|
} else if (request.method === 'DELETE' && PROXY_TEST_TASK_ROUTE.test(route)) {
|
|
const state = await cancelProxyTest(PROXY_TEST_TASK_ROUTE.exec(route)[1])
|
|
publishProxyTest(state)
|
|
await recordAudit('proxy.test.cancel-requested', { taskId: state.taskId })
|
|
send(response, 202, state)
|
|
} else if (request.method === 'POST' && route === 'user-plugins/apply') {
|
|
send(response, 202, await startUserPluginAction(await jsonBody(request)))
|
|
} else if (request.method === 'GET' && /^user-plugins\/task\/[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/.test(route)) {
|
|
const taskId = route.slice('user-plugins/task/'.length)
|
|
const task = userPluginTasks.get(taskId)
|
|
if (task === undefined) {
|
|
const error = new Error('User Plugin task was not found')
|
|
error.statusCode = 404
|
|
throw error
|
|
}
|
|
send(response, 200, task)
|
|
} else if (request.method === 'POST' && route === 'terminal/sessions') {
|
|
send(response, 201, terminalSessions.create(await jsonBody(request)))
|
|
} else if (request.method === 'GET' && TERMINAL_SESSION_ROUTE.test(route)) {
|
|
send(response, 200, terminalSessions.status(TERMINAL_SESSION_ROUTE.exec(route)[1]))
|
|
} else if (request.method === 'DELETE' && TERMINAL_SESSION_ROUTE.test(route)) {
|
|
send(response, 200, terminalSessions.close(TERMINAL_SESSION_ROUTE.exec(route)[1]))
|
|
} else if (request.method === 'GET' && route === 'settings-document') {
|
|
if (settingsDocument === undefined) throw new Error('settings document editing is not configured')
|
|
send(response, 200, await settingsDocument.read())
|
|
} else if (request.method === 'PUT' && route === 'settings-document') {
|
|
if (settingsDocument === undefined) throw new Error('settings document editing is not configured')
|
|
const body = await jsonBody(request, MAX_SETTINGS_DOCUMENT_BYTES * 6 + 4096)
|
|
if (body === null || typeof body !== 'object' || Array.isArray(body)
|
|
|| Object.keys(body).some(key => !['content', 'revision'].includes(key))) {
|
|
throw new Error('settings document request is invalid')
|
|
}
|
|
const saved = await settingsDocument.write(body.content, body.revision)
|
|
await audit('settings-document.saved', { revision: saved.revision })
|
|
send(response, 200, saved)
|
|
} else if (request.method === 'GET' && route === 'files/config') {
|
|
send(response, 200, fileLocations)
|
|
} else if (request.method === 'GET' && route === 'files/list') {
|
|
if (fileInventory === undefined) throw new Error('file management is not configured')
|
|
const result = await fileInventory.list(url.searchParams.get('path'), {
|
|
cursor: url.searchParams.get('cursor'), limit: url.searchParams.get('limit'),
|
|
offset: url.searchParams.get('offset'),
|
|
sort: url.searchParams.get('sort') ?? 'name', order: url.searchParams.get('order') ?? 'asc',
|
|
})
|
|
await recordAudit('files.list.completed', { path: result.path, entries: result.entries.length })
|
|
send(response, 200, result)
|
|
} else if (request.method === 'GET' && route === 'files/stat') {
|
|
if (fileInventory === undefined) throw new Error('file management is not configured')
|
|
let result
|
|
try {
|
|
result = await fileInventory.stat(url.searchParams.get('path'))
|
|
} catch (error) {
|
|
if (url.searchParams.get('optional') !== 'true' || error?.statusCode !== 404) throw error
|
|
return send(response, 200, null)
|
|
}
|
|
await recordAudit('files.stat.completed', { path: result.path, type: result.type })
|
|
send(response, 200, result)
|
|
} else if (request.method === 'GET' && route === 'files/content') {
|
|
if (fileInventory === undefined) throw new Error('file management is not configured')
|
|
const result = await fileInventory.content(url.searchParams.get('path'))
|
|
await recordAudit('files.content.completed', { path: result.path, size: result.size })
|
|
send(response, 200, result)
|
|
} else if (request.method === 'PUT' && route === 'files/content') {
|
|
if (fileEditor === undefined) throw new Error('file editing is not configured')
|
|
const body = await jsonBody(request, 2 * 1024 * 1024 + 8192)
|
|
if (body === null || typeof body !== 'object' || Array.isArray(body)
|
|
|| Object.keys(body).some(key => !['path', 'content', 'revision', 'create'].includes(key))) throw new Error('file content request is invalid')
|
|
const saved = await fileEditor.write(body.path, body.content, body.revision, { create: body.create === true })
|
|
await recordAudit('files.content.saved', { path: saved.path, size: saved.size, managed: saved.managed })
|
|
send(response, body.create === true ? 201 : 200, saved)
|
|
} else if (request.method === 'GET' && route === 'files/download') {
|
|
if (fileTransfers === undefined) throw new Error('file transfers are not configured')
|
|
const path = url.searchParams.get('path')
|
|
const transfer = fileTasks?.startTransfer({ operation: 'download', path }, async controls => {
|
|
const download = await fileTransfers.openDownload(path, {
|
|
revision: url.searchParams.get('revision') ?? undefined,
|
|
range: typeof request.headers.range === 'string' ? request.headers.range : undefined,
|
|
signal: controls.signal,
|
|
})
|
|
controls.progress({ processedBytes: 0, totalBytes: Number(download.headers['content-length']) })
|
|
await fileTransfers.sendDownload(response, download, {
|
|
signal: controls.signal,
|
|
onProgress: (processedBytes, totalBytes) => controls.progress({ processedBytes, totalBytes }),
|
|
})
|
|
return download
|
|
})
|
|
if (transfer === undefined) throw new Error('file tasks are not configured')
|
|
await recordAudit('files.download.started', { path, taskId: transfer.task.taskId })
|
|
const download = await transfer.result
|
|
await recordAudit('files.download.completed', { path: download.path, revision: download.revision, taskId: transfer.task.taskId })
|
|
} else if (request.method === 'POST' && route === 'files/upload') {
|
|
if (fileTransfers === undefined) throw new Error('file transfers are not configured')
|
|
const lengthHeader = request.headers['content-length']
|
|
const contentLength = lengthHeader === undefined ? undefined : Number(lengthHeader)
|
|
if (contentLength !== undefined && (!Number.isSafeInteger(contentLength) || contentLength < 0)) throw new Error('Content-Length is invalid')
|
|
const path = url.searchParams.get('path')
|
|
await recordAudit('files.upload.started', { path })
|
|
if (fileTasks === undefined) throw new Error('file tasks are not configured')
|
|
const transfer = fileTasks.startTransfer({ operation: 'upload', path, totalBytes: contentLength ?? null }, controls => fileTransfers.upload(request, path, {
|
|
conflict: url.searchParams.get('conflict') ?? 'reject', contentLength, signal: controls.signal,
|
|
onProgress: (processedBytes, totalBytes) => controls.progress({ processedBytes, totalBytes }),
|
|
}))
|
|
const result = await transfer.result
|
|
await recordAudit('files.upload.completed', { path: result.path, size: result.size })
|
|
send(response, 201, result)
|
|
} else if (request.method === 'POST' && route === 'files/tasks') {
|
|
if (fileTasks === undefined) throw new Error('file tasks are not configured')
|
|
const body = await jsonBody(request)
|
|
if (fileTasks.wouldManage?.(body) === true) requireRuntimeIdle()
|
|
const task = fileTasks.start(body)
|
|
await recordAudit(`files.${task.operation}.started`, { path: task.path, taskId: task.taskId, managed: task.managed })
|
|
send(response, 202, task)
|
|
} else if (request.method === 'GET' && route === 'files/tasks') {
|
|
if (fileTasks === undefined) throw new Error('file tasks are not configured')
|
|
send(response, 200, { tasks: fileTasks.list() })
|
|
} else if (request.method === 'GET' && FILE_TASK_ROUTE.test(route)) {
|
|
if (fileTasks === undefined) throw new Error('file tasks are not configured')
|
|
const taskId = FILE_TASK_ROUTE.exec(route)[1]
|
|
send(response, 200, fileTasks.get(taskId, { cursor: url.searchParams.get('cursor'), limit: url.searchParams.get('limit') }))
|
|
} else if (request.method === 'DELETE' && FILE_TASK_ROUTE.test(route)) {
|
|
if (fileTasks === undefined) throw new Error('file tasks are not configured')
|
|
const task = fileTasks.cancel(FILE_TASK_ROUTE.exec(route)[1])
|
|
await recordAudit('files.task.cancel-requested', { operation: task.operation, taskId: task.taskId })
|
|
send(response, 202, task)
|
|
} else if (request.method === 'POST' && route === 'check') {
|
|
const body = await jsonBody(request)
|
|
const source = body.source ?? 'manual'
|
|
if (!['manual', 'page-open', 'channel-change'].includes(source)) throw new Error('update check source is invalid')
|
|
const target = await coordinator.check(source)
|
|
if (target.busy === true) {
|
|
send(response, 202, { busy: true })
|
|
return
|
|
}
|
|
send(response, 200, target.unavailable === true
|
|
? { available: false, upstream: target.upstream }
|
|
: {
|
|
available: target.updateAvailable ?? true,
|
|
targetSequence: target.value.targetSequence,
|
|
desired: target.value.desired,
|
|
})
|
|
} else if (request.method === 'POST' && route === 'update') {
|
|
requireRuntimeIdle()
|
|
const task = coordinator.startReconcile()
|
|
void task.completion
|
|
.then(
|
|
async () => {
|
|
await refreshLoadedUserPlugins()
|
|
await audit('update.completed', { taskId: task.taskId })
|
|
},
|
|
error => audit('update.failed', { error, taskId: task.taskId }),
|
|
)
|
|
.catch(() => {})
|
|
await audit('update.started', { taskId: task.taskId })
|
|
send(response, 202, { taskId: task.taskId })
|
|
} else if (request.method === 'POST' && ['start-dsh', 'stop-dsh', 'restart-dsh'].includes(route)) {
|
|
const task = startLifecycle(route.slice(0, -4))
|
|
send(response, 202, task)
|
|
} else if (request.method === 'POST' && route === 'runtime/reset') {
|
|
const task = startRuntimeReset()
|
|
send(response, 202, task)
|
|
} else if (request.method === 'POST' && route === 'bundled-plugins/action') {
|
|
const body = await jsonBody(request)
|
|
send(response, 202, startPluginAction(body.id, body.action))
|
|
} else if (request.method === 'POST' && route === 'bundled-plugins/toggle') {
|
|
const body = await jsonBody(request)
|
|
send(response, 202, startPluginAction(body.id, body.action, { toggleOnly: true }))
|
|
} else if (request.method === 'POST' && route === 'bundled-plugins/recovery-action') {
|
|
const body = await jsonBody(request)
|
|
send(response, 202, startPluginAction(body.id, body.action, { recovery: true }))
|
|
} else if (request.method === 'POST' && route === 'bundled-plugins/discard') {
|
|
requireRuntimeIdle()
|
|
if (coordinator.hasActiveTask?.() === true) throw new UpdateConflictError('an update task is already running')
|
|
const result = await discardBundledPluginChanges()
|
|
publishPlugin({ restartRequired: false })
|
|
await audit('system-plugin.changes.discarded')
|
|
send(response, 200, result)
|
|
} else if (request.method === 'POST' && route === 'system-skills/action') {
|
|
const body = await jsonBody(request)
|
|
if (body === null || typeof body !== 'object' || Array.isArray(body)
|
|
|| Object.keys(body).sort().join(',') !== 'action,skillId') {
|
|
throw new Error('System Skill request is invalid')
|
|
}
|
|
send(response, 202, startSystemSkillAction(body.skillId, body.action))
|
|
} else if (request.method === 'POST' && route === 'user-skills/action') {
|
|
send(response, 202, await startUserSkillAction(await jsonBody(request)))
|
|
} else if (request.method === 'PUT' && route === 'channel') {
|
|
const body = await jsonBody(request)
|
|
const value = await coordinator.setChannel(body.channel)
|
|
await audit('update.channel.changed', { updateChannel: value.updateChannel })
|
|
server.emit('management-state', value)
|
|
send(response, 200, value)
|
|
} else if (request.method === 'PUT' && route === 'automatic-check') {
|
|
const value = await updateAutomaticCheck(await jsonBody(request))
|
|
await audit('update.automatic-check.configured', {
|
|
enabled: value.enabled,
|
|
intervalSeconds: value.intervalSeconds,
|
|
notificationsEnabled: value.notificationsEnabled,
|
|
})
|
|
server.emit('management-state', value)
|
|
send(response, 200, value)
|
|
} else if (request.method === 'GET' && route === 'auth-settings') {
|
|
send(response, 200, await getAuthenticationSettings(request))
|
|
} else if (request.method === 'PUT' && route === 'auth-settings') {
|
|
const body = await jsonBody(request)
|
|
const value = await updateAuthenticationSettings(body, request)
|
|
await audit('access.authentication.settings.changed', {})
|
|
send(response, 200, value)
|
|
} else if (request.method === 'POST' && route === 'auth-totp/enrollments') {
|
|
send(response, 201, await beginTotpEnrollment(await jsonBody(request), request))
|
|
} else if (request.method === 'POST' && route === 'auth-totp/enrollments/confirm') {
|
|
send(response, 200, await confirmTotpEnrollment(await jsonBody(request), request))
|
|
} else if (request.method === 'POST' && route === 'auth-totp/enrollments/cancel') {
|
|
send(response, 200, await cancelTotpEnrollment(await jsonBody(request), request))
|
|
} else if (request.method === 'POST' && route === 'auth-totp/disable-confirmations') {
|
|
send(response, 201, await beginTotpDisableConfirmation(await jsonBody(request), request))
|
|
} else if (request.method === 'POST' && route === 'auth-totp/disable-confirmations/confirm') {
|
|
send(response, 200, await confirmTotpDisable(await jsonBody(request), request))
|
|
} else if (request.method === 'POST' && route === 'auth-totp/disable-confirmations/cancel') {
|
|
send(response, 200, await cancelTotpDisableConfirmation(await jsonBody(request), request))
|
|
} else if (request.method === 'POST' && route === 'auth-sessions/revoke') {
|
|
const body = await jsonBody(request)
|
|
const value = await revokeAuthenticationSessions(body, request)
|
|
await audit('access.sessions.revoked', {
|
|
sessionId: body.sessionId ?? null,
|
|
currentSessionRevoked: value.currentSessionRevoked === true,
|
|
})
|
|
send(response, 200, value)
|
|
} else if (request.method === 'POST' && route === 'management-origin/transitions') {
|
|
const body = await jsonBody(request)
|
|
const value = await createManagementOriginTransition(body, request)
|
|
await audit('access.management-transition.created', { mode: body.mode ?? null })
|
|
send(response, 201, value)
|
|
} else if (request.method === 'POST' && route === 'management-origin/transitions/commit') {
|
|
const value = await commitManagementOriginTransition(await jsonBody(request), request)
|
|
await audit('access.management-origin.changed', { mode: value.account?.managementAccess?.mode ?? null })
|
|
send(response, 200, value)
|
|
} else if (request.method === 'POST' && route === 'management-origin/transitions/failure') {
|
|
send(response, 200, await reportManagementOriginTransitionFailure(await jsonBody(request)))
|
|
} else if (request.method === 'POST' && route === 'holds/retry') {
|
|
const body = await jsonBody(request)
|
|
const result = await coordinator.retryHold(body.id)
|
|
await audit('update.hold.retried', { holdId: body.id })
|
|
send(response, 200, result)
|
|
} else if (request.method === 'GET' && route === 'rollback-plan') {
|
|
send(response, 200, { plan: await coordinator.rollbackPlan() })
|
|
} else if (request.method === 'POST' && ['rollback', 'return-stable'].includes(route)) {
|
|
requireRuntimeIdle()
|
|
const body = await jsonBody(request)
|
|
const task = coordinator.startCompleteRollback(body.planId, {
|
|
requireConfirmation: route === 'return-stable',
|
|
confirmDataLoss: body.confirmDataLoss,
|
|
})
|
|
void task.completion
|
|
.then(async () => {
|
|
await refreshLoadedUserPlugins()
|
|
await audit(`${route}.completed`, { taskId: task.taskId })
|
|
})
|
|
.catch(error => audit(`${route}.failed`, { error, taskId: task.taskId }))
|
|
.catch(() => {})
|
|
await audit(`${route}.started`, { taskId: task.taskId })
|
|
send(response, 202, { taskId: task.taskId })
|
|
} else if (request.method === 'GET' && route === 'events') {
|
|
let heartbeat
|
|
response.writeHead(200, {
|
|
'content-type': 'text/event-stream',
|
|
'cache-control': 'no-cache, no-transform',
|
|
connection: 'keep-alive',
|
|
'x-accel-buffering': 'no',
|
|
})
|
|
response.flushHeaders()
|
|
event(response, 'state', await coordinator.state.read())
|
|
const listener = value => event(response, 'state', value)
|
|
coordinator.on('state', listener)
|
|
server.on('management-state', listener)
|
|
heartbeat = setInterval(() => event(response, 'heartbeat', { timestamp: new Date().toISOString() }), stateHeartbeatMs)
|
|
heartbeat.unref()
|
|
response.once('close', () => {
|
|
clearInterval(heartbeat)
|
|
coordinator.off('state', listener)
|
|
server.off('management-state', listener)
|
|
})
|
|
} else if (request.method === 'GET' && route === 'logs') {
|
|
send(response, 200, { entries: await logs.query(logOptions(url)) })
|
|
} else if (request.method === 'GET' && route === 'logs/stream') {
|
|
const options = logOptions(url)
|
|
const selectedSources = options.sources === undefined ? undefined : new Set(options.sources)
|
|
const sent = new Set()
|
|
const sentOrder = []
|
|
const sendLog = value => {
|
|
const identity = JSON.stringify(value)
|
|
if (sent.has(identity)) return
|
|
sent.add(identity)
|
|
sentOrder.push(identity)
|
|
if (sentOrder.length > 10_000) sent.delete(sentOrder.shift())
|
|
event(response, 'log', value)
|
|
}
|
|
const listener = value => {
|
|
if (selectedSources === undefined || selectedSources.has(value.source)) sendLog(value)
|
|
}
|
|
let follower
|
|
let heartbeat
|
|
let closed = false
|
|
const cleanup = () => {
|
|
closed = true
|
|
clearInterval(heartbeat)
|
|
follower?.close()
|
|
logs.off('entry', listener)
|
|
}
|
|
response.once('close', cleanup)
|
|
response.writeHead(200, {
|
|
'content-type': 'text/event-stream',
|
|
'cache-control': 'no-cache, no-transform',
|
|
connection: 'keep-alive',
|
|
'x-accel-buffering': 'no',
|
|
})
|
|
response.flushHeaders()
|
|
event(response, 'heartbeat', { timestamp: new Date().toISOString() })
|
|
event(response, 'sources', await logs.sources())
|
|
for (const entry of await logs.query(options)) sendLog(entry)
|
|
if (closed) return
|
|
logs.on('entry', listener)
|
|
let followFailureReported = false
|
|
follower = await logs.follow(options, sendLog, {
|
|
onError: error => {
|
|
if (followFailureReported) return
|
|
followFailureReported = true
|
|
void logs.diagnostic('log-manager', 'log.follow.failed', { error })
|
|
},
|
|
})
|
|
if (closed) {
|
|
follower.close()
|
|
return
|
|
}
|
|
for (const entry of await logs.query(options)) sendLog(entry)
|
|
heartbeat = setInterval(() => event(response, 'heartbeat', { timestamp: new Date().toISOString() }), 15_000)
|
|
heartbeat.unref()
|
|
if (closed) clearInterval(heartbeat)
|
|
} else send(response, 404, { error: 'not found' })
|
|
} catch (error) {
|
|
const conflict = error instanceof UpdateConflictError || error instanceof SettingsDocumentConflictError
|
|
|| error?.code === 'REVISION_CONFLICT' || error?.statusCode === 409
|
|
await logs.diagnostic('platform-management', 'management.request.failed', {
|
|
error,
|
|
level: conflict ? 'warning' : 'error',
|
|
method: request.method ?? null,
|
|
pathname,
|
|
})
|
|
if (error?.proxyError !== undefined) {
|
|
send(response, error.statusCode ?? 400, { error: error.proxyError })
|
|
} else {
|
|
send(response, conflict ? 409 : (error?.statusCode ?? 400), {
|
|
error: error instanceof Error ? error.message : 'management request failed',
|
|
code: error?.code ?? 'MANAGEMENT_ERROR',
|
|
})
|
|
}
|
|
}
|
|
})
|
|
server.setMaxListeners(0)
|
|
server.on('upgrade', (request, socket, head) => {
|
|
let pathname = 'invalid-url'
|
|
void (async () => {
|
|
pathname = new URL(request.url ?? '/', 'http://management.internal').pathname
|
|
if (!await authorizeInternal(request)) {
|
|
socket.end('HTTP/1.1 401 Unauthorized\r\nConnection: close\r\nContent-Length: 0\r\n\r\n')
|
|
return
|
|
}
|
|
if (!pathname.startsWith(API_PREFIX)) throw new Error('not found')
|
|
const match = TERMINAL_STREAM_ROUTE.exec(pathname.slice(API_PREFIX.length))
|
|
if (match === null) throw new Error('not found')
|
|
terminalSessions.upgrade(request, socket, head, match[1])
|
|
})().catch(error => {
|
|
void logs.diagnostic('platform-management', 'management.terminal-upgrade.failed', {
|
|
error,
|
|
level: error?.statusCode === 404 ? 'warning' : 'error',
|
|
pathname,
|
|
})
|
|
const status = error?.statusCode === 503 ? '503 Service Unavailable' : '404 Not Found'
|
|
if (!socket.destroyed) socket.end(`HTTP/1.1 ${status}\r\nConnection: close\r\nContent-Length: 0\r\n\r\n`)
|
|
})
|
|
})
|
|
server.once('close', () => { proxyWatchClosed = true })
|
|
if (recoverUserPluginTransaction !== undefined) server.once('listening', () => {
|
|
userPluginTask = Promise.resolve()
|
|
.then(() => recoverUserPluginTransaction())
|
|
.then(async state => {
|
|
if (state === undefined) return
|
|
publishUserPlugin({
|
|
status: state.phase === 'completed' ? 'success' : 'failed',
|
|
taskId: state.taskId,
|
|
phase: state.phase,
|
|
error: state.error,
|
|
})
|
|
await audit('user-plugin.recovery.completed', {
|
|
taskId: state.taskId,
|
|
phase: state.phase,
|
|
recoveryResult: state.recoveryResult,
|
|
})
|
|
}, async error => {
|
|
publishUserPlugin({ status: 'failed', phase: 'recovery', error: error instanceof Error ? error.message : String(error) })
|
|
await audit('user-plugin.recovery.failed', { error })
|
|
})
|
|
.finally(() => { userPluginTask = undefined })
|
|
userPluginTask.catch(() => {})
|
|
})
|
|
return server
|
|
}
|
|
|
|
export async function listenManagement(server, socketPath) {
|
|
await mkdir(dirname(socketPath), { recursive: true })
|
|
await rm(socketPath, { force: true })
|
|
await new Promise((resolve, reject) => {
|
|
server.once('error', reject)
|
|
server.listen(socketPath, resolve)
|
|
})
|
|
await chmod(socketPath, 0o600)
|
|
}
|