Hub stage 2: read-only Technitium-compatible server over several servers
node:http server that serves the console build and a /api/* the console accepts: its own login and session (clusterInitialized, the servers as nodes, hub flag, Dashboard and Logs only), node=<id> forwarded with that server's token, no node or cluster merged (stats, top lists, query logs, apps), every write refused with 403, a short response cache, and an unreachable server reported as hubWarnings. The dev mock takes --port and --scale so two of them differ. Vitest skips .claude/ worktrees. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
DenozordecandClaude Opus 5.5 committed
1 parent
9df2c2b908
commit
5f29a438c9
8 files changed
+980
-7
No files matched your search
+332
@@ -0,0 +1,332 @@
|
||||
/*
|
||||
The hub: a small server that impersonates ONE Technitium DNS Server for the console and
|
||||
answers from several real ones.
|
||||
|
||||
node --experimental-strip-types hub/server.ts
|
||||
|
||||
It serves the console's static build and a read-only, Technitium-compatible `/api/*`.
|
||||
A call aimed at one server (`node=<id>`) is forwarded to it with that server's token; a
|
||||
call with no node (or `node=cluster`) asks every server in parallel and merges the
|
||||
answers (see merge.ts). Anything that would change data is refused.
|
||||
*/
|
||||
import http from 'node:http'
|
||||
import { createReadStream } from 'node:fs'
|
||||
import { stat } from 'node:fs/promises'
|
||||
import path from 'node:path'
|
||||
import { fileURLToPath } from 'node:url'
|
||||
import { loadConfig, HUB_VERSION, type HubConfig, type HubServer } from './config.ts'
|
||||
import { Upstream, UpstreamError, type Params } from './upstream.ts'
|
||||
import { Sessions } from './session.ts'
|
||||
import {
|
||||
mergeQueryLogs,
|
||||
mergeSettled,
|
||||
mergeStats,
|
||||
mergeTopResponses,
|
||||
type DashboardStats,
|
||||
type QueryLogPage,
|
||||
type Settled,
|
||||
type TopEntry,
|
||||
} from './merge.ts'
|
||||
|
||||
type Json = Record<string, unknown>
|
||||
|
||||
const MAX_BODY = 64 * 1024
|
||||
|
||||
/* Paths relative to `/api/`. Anything not listed here is refused with 403. */
|
||||
const READ_ENDPOINTS = new Set([
|
||||
'dashboard/stats/get',
|
||||
'dashboard/stats/getTop',
|
||||
'dashboard/metrics/json',
|
||||
'apps/list',
|
||||
'logs/query',
|
||||
'logs/list',
|
||||
'settings/get',
|
||||
'user/checkForUpdate',
|
||||
'admin/cluster/state',
|
||||
])
|
||||
|
||||
const TYPES: Record<string, string> = {
|
||||
'.html': 'text/html; charset=utf-8',
|
||||
'.js': 'text/javascript; charset=utf-8',
|
||||
'.mjs': 'text/javascript; charset=utf-8',
|
||||
'.css': 'text/css; charset=utf-8',
|
||||
'.woff2': 'font/woff2',
|
||||
'.woff': 'font/woff',
|
||||
'.png': 'image/png',
|
||||
'.ico': 'image/x-icon',
|
||||
'.json': 'application/json; charset=utf-8',
|
||||
'.svg': 'image/svg+xml',
|
||||
'.txt': 'text/plain; charset=utf-8',
|
||||
'.map': 'application/json; charset=utf-8',
|
||||
'.webmanifest': 'application/manifest+json',
|
||||
}
|
||||
|
||||
const HTML_HEADERS = {
|
||||
'content-security-policy': "default-src 'self'; img-src 'self' data:; style-src 'self' 'unsafe-inline'",
|
||||
'x-content-type-options': 'nosniff',
|
||||
}
|
||||
|
||||
function send(res: http.ServerResponse, code: number, body: unknown): void {
|
||||
const text = JSON.stringify(body)
|
||||
res.writeHead(code, {
|
||||
'content-type': 'application/json; charset=utf-8',
|
||||
'cache-control': 'no-store',
|
||||
'content-length': Buffer.byteLength(text),
|
||||
})
|
||||
res.end(text)
|
||||
}
|
||||
|
||||
const fail = (errorMessage: string): Json => ({ status: 'error', errorMessage })
|
||||
|
||||
async function readForm(req: http.IncomingMessage): Promise<URLSearchParams> {
|
||||
if (req.method !== 'POST') return new URLSearchParams()
|
||||
const chunks: Buffer[] = []
|
||||
let size = 0
|
||||
for await (const chunk of req) {
|
||||
size += (chunk as Buffer).length
|
||||
if (size > MAX_BODY) break
|
||||
chunks.push(chunk as Buffer)
|
||||
}
|
||||
const type = String(req.headers['content-type'] ?? '')
|
||||
if (size > MAX_BODY || !type.includes('application/x-www-form-urlencoded')) return new URLSearchParams()
|
||||
return new URLSearchParams(Buffer.concat(chunks).toString('utf8'))
|
||||
}
|
||||
|
||||
function bearer(req: http.IncomingMessage): string | null {
|
||||
const header = String(req.headers.authorization ?? '')
|
||||
return header.startsWith('Bearer ') ? header.slice(7).trim() : null
|
||||
}
|
||||
|
||||
function positive(value: string | undefined, fallback: number): number {
|
||||
const n = Number(value)
|
||||
return Number.isInteger(n) && n > 0 ? n : fallback
|
||||
}
|
||||
|
||||
export function createHub(config: HubConfig): http.Server {
|
||||
const upstream = new Upstream(config)
|
||||
const sessions = new Sessions(config)
|
||||
const byId = new Map(config.servers.map((s) => [s.id, s]))
|
||||
|
||||
/* Ask every server; the ones that fail become warnings (see mergeSettled). */
|
||||
async function everywhere<T>(apiPath: string, params: Params): Promise<Settled<T>[]> {
|
||||
const results = await Promise.allSettled(config.servers.map((s) => upstream.call(s, apiPath, params)))
|
||||
return results.map((r, i): Settled<T> => {
|
||||
const node = config.servers[i].id
|
||||
return r.status === 'fulfilled'
|
||||
? { node, ok: true, data: r.value as T }
|
||||
: { node, ok: false, error: r.reason instanceof Error ? r.reason.message : 'failed' }
|
||||
})
|
||||
}
|
||||
|
||||
function ok(response: unknown, warnings: string[] = []): Json {
|
||||
return warnings.length > 0 ? { status: 'ok', response, hubWarnings: warnings } : { status: 'ok', response }
|
||||
}
|
||||
|
||||
async function single(server: HubServer, apiPath: string, params: Params): Promise<Json> {
|
||||
try {
|
||||
return ok(await upstream.call(server, apiPath, params))
|
||||
} catch (e) {
|
||||
if (e instanceof UpstreamError) {
|
||||
return fail(e.kind === 'unreachable' ? `${server.id} did not answer: ${e.message}` : `${server.id}: ${e.message}`)
|
||||
}
|
||||
throw e
|
||||
}
|
||||
}
|
||||
|
||||
async function aggregate(apiPath: string, params: Params): Promise<Json> {
|
||||
try {
|
||||
switch (apiPath) {
|
||||
case 'dashboard/stats/get': {
|
||||
const { data, warnings } = mergeSettled(await everywhere<DashboardStats>(apiPath, params))
|
||||
return ok(mergeStats(data), warnings)
|
||||
}
|
||||
case 'dashboard/stats/getTop': {
|
||||
const { data, warnings } = mergeSettled(await everywhere<Record<string, TopEntry[]>>(apiPath, params))
|
||||
const limit = params.limit === undefined ? undefined : positive(params.limit, 1000)
|
||||
return ok(mergeTopResponses(data, limit), warnings)
|
||||
}
|
||||
case 'logs/query': {
|
||||
const pageNumber = positive(params.pageNumber, 1)
|
||||
const entriesPerPage = positive(params.entriesPerPage, 25)
|
||||
// Page N of the merged list is cut from the head of every server's list.
|
||||
const asked = { ...params, pageNumber: '1', entriesPerPage: String(pageNumber * entriesPerPage) }
|
||||
const settled = await everywhere<QueryLogPage>(apiPath, asked)
|
||||
const { data, warnings } = mergeSettled(settled)
|
||||
// Only the servers that answered, in server order, so the i-th page is the i-th survivor's.
|
||||
const answered = settled.filter((r) => r.ok).map((r) => r.node)
|
||||
const pages = data.map((page, i) => ({ node: answered[i], page }))
|
||||
return ok(mergeQueryLogs(pages, { pageNumber, entriesPerPage, descending: params.descendingOrder !== 'false' }), warnings)
|
||||
}
|
||||
case 'apps/list': {
|
||||
const { data, warnings } = mergeSettled(await everywhere<{ apps?: { name: string }[] }>(apiPath, params))
|
||||
const seen = new Map<string, { name: string }>()
|
||||
for (const r of data) for (const app of r.apps ?? []) if (!seen.has(app.name)) seen.set(app.name, app)
|
||||
return ok({ ...data[0], apps: [...seen.values()] }, warnings)
|
||||
}
|
||||
default:
|
||||
// Everything else read-only is answered by the first server.
|
||||
return await single(config.servers[0], apiPath, params)
|
||||
}
|
||||
} catch (e) {
|
||||
if (e instanceof Error && !(e instanceof UpstreamError)) return fail(e.message)
|
||||
throw e
|
||||
}
|
||||
}
|
||||
|
||||
async function api(req: http.IncomingMessage, res: http.ServerResponse, apiPath: string, url: URL): Promise<void> {
|
||||
const form = await readForm(req)
|
||||
const params: Params = {}
|
||||
for (const [k, v] of url.searchParams) params[k] = v
|
||||
for (const [k, v] of form) params[k] = v
|
||||
|
||||
// Public and session endpoints.
|
||||
if (apiPath === 'status') {
|
||||
send(res, 200, { status: 'ok', hasDefaultCredentials: false, ssoEnabled: false })
|
||||
return
|
||||
}
|
||||
if (apiPath === 'user/login') {
|
||||
const token = sessions.login(params.user ?? '', params.pass ?? '')
|
||||
send(res, 200, token === null ? fail(`Invalid username or password for user: ${params.user ?? ''}`) : sessions.body(token))
|
||||
return
|
||||
}
|
||||
|
||||
const token = bearer(req) ?? params.token ?? null
|
||||
if (!sessions.touch(token)) {
|
||||
send(res, 200, { status: 'invalid-token', errorMessage: 'Invalid token or session expired.' })
|
||||
return
|
||||
}
|
||||
if (apiPath === 'user/session/get') {
|
||||
send(res, 200, sessions.body(token as string))
|
||||
return
|
||||
}
|
||||
if (apiPath === 'user/logout') {
|
||||
sessions.logout(token as string)
|
||||
send(res, 200, { status: 'ok' })
|
||||
return
|
||||
}
|
||||
if (!READ_ENDPOINTS.has(apiPath)) {
|
||||
send(res, 403, fail('The hub is read-only.'))
|
||||
return
|
||||
}
|
||||
if (apiPath === 'user/checkForUpdate') {
|
||||
send(res, 200, ok({ updateAvailable: false }))
|
||||
return
|
||||
}
|
||||
if (apiPath === 'admin/cluster/state') {
|
||||
send(
|
||||
res,
|
||||
200,
|
||||
ok({
|
||||
version: HUB_VERSION,
|
||||
dnsServerDomain: config.domain,
|
||||
clusterInitialized: true,
|
||||
clusterDomain: config.domain,
|
||||
clusterNodes: config.servers.map((s, i) => ({
|
||||
id: i + 1,
|
||||
name: s.id,
|
||||
url: s.url,
|
||||
ipAddresses: [],
|
||||
type: i === 0 ? 'Primary' : 'Secondary',
|
||||
state: 'Connected',
|
||||
})),
|
||||
}),
|
||||
)
|
||||
return
|
||||
}
|
||||
|
||||
// What the real servers get: the caller's params minus the hub's own.
|
||||
const { node = '', token: _token, ...forward } = params
|
||||
const server = byId.get(node)
|
||||
if (server !== undefined) {
|
||||
send(res, 200, await single(server, apiPath, forward))
|
||||
return
|
||||
}
|
||||
if (node !== '' && node !== 'cluster') {
|
||||
send(res, 200, fail(`Unknown node: ${node}`))
|
||||
return
|
||||
}
|
||||
send(res, 200, await aggregate(apiPath, forward))
|
||||
}
|
||||
|
||||
async function serveStatic(req: http.IncomingMessage, res: http.ServerResponse, url: URL): Promise<void> {
|
||||
if (req.method !== 'GET' && req.method !== 'HEAD') {
|
||||
res.writeHead(405, { allow: 'GET, HEAD' }).end()
|
||||
return
|
||||
}
|
||||
const root = path.resolve(config.staticDir)
|
||||
let pathname: string
|
||||
try {
|
||||
pathname = decodeURIComponent(url.pathname)
|
||||
} catch {
|
||||
res.writeHead(400).end()
|
||||
return
|
||||
}
|
||||
const target = path.resolve(root, '.' + pathname)
|
||||
if (pathname.includes('\0') || (target !== root && !target.startsWith(root + path.sep))) {
|
||||
res.writeHead(403).end()
|
||||
return
|
||||
}
|
||||
|
||||
let file = target
|
||||
let info = await stat(target).catch(() => null)
|
||||
if (info?.isDirectory()) {
|
||||
const index = path.join(target, 'index.html')
|
||||
if ((await stat(index).catch(() => null))?.isFile() !== true) info = null
|
||||
else if (!pathname.endsWith('/')) {
|
||||
// As Technitium: `/x` -> `/x/` when the folder has an index.
|
||||
res.writeHead(301, { location: `${url.pathname}/${url.search}` }).end()
|
||||
return
|
||||
} else {
|
||||
file = index
|
||||
info = await stat(index)
|
||||
}
|
||||
}
|
||||
if (info === null || !info.isFile()) {
|
||||
res.writeHead(404, { 'content-type': 'text/plain; charset=utf-8' }).end('Not found')
|
||||
return
|
||||
}
|
||||
|
||||
const ext = path.extname(file).toLowerCase()
|
||||
const headers: Record<string, string | number> = {
|
||||
'content-type': TYPES[ext] ?? 'application/octet-stream',
|
||||
'content-length': info.size,
|
||||
}
|
||||
if (ext === '.html') Object.assign(headers, HTML_HEADERS, { 'cache-control': 'no-cache' })
|
||||
res.writeHead(200, headers)
|
||||
if (req.method === 'HEAD') {
|
||||
res.end()
|
||||
return
|
||||
}
|
||||
createReadStream(file).on('error', () => res.destroy()).pipe(res)
|
||||
}
|
||||
|
||||
return http.createServer((req, res) => {
|
||||
void (async () => {
|
||||
try {
|
||||
const url = new URL(req.url ?? '/', 'http://hub')
|
||||
// The console may sit at a sub path: any `<prefix>/api/<path>` is the API.
|
||||
const match = /^(?:.*\/)?api\/(.+?)\/?$/.exec(url.pathname)
|
||||
if (match !== null) await api(req, res, match[1], url)
|
||||
else await serveStatic(req, res, url)
|
||||
} catch (e) {
|
||||
console.error('hub error:', e instanceof Error ? e.message : e)
|
||||
if (!res.headersSent) send(res, 500, fail('Internal hub error.'))
|
||||
else res.destroy()
|
||||
}
|
||||
})()
|
||||
})
|
||||
}
|
||||
|
||||
/* Run directly (not imported by the tests). */
|
||||
if (process.argv[1] !== undefined && path.resolve(process.argv[1]) === fileURLToPath(import.meta.url)) {
|
||||
let config: HubConfig
|
||||
try {
|
||||
config = loadConfig()
|
||||
} catch (e) {
|
||||
console.error(`hub: ${e instanceof Error ? e.message : String(e)}`)
|
||||
process.exit(1)
|
||||
}
|
||||
createHub(config).listen(config.port, () => {
|
||||
console.log(`Hub ${HUB_VERSION} on :${config.port}, servers: ${config.servers.map((s) => s.id).join(', ')}`)
|
||||
})
|
||||
}
|
||||
Reference in new issue
Block a user