/* Calls to the real Technitium servers: `GET /api/?&token=`. Successful answers are cached for a few seconds, so a dashboard that refreshes every minute (and several people looking at it) does not load the servers. */ import type { HubConfig, HubServer } from './config.ts' /* `unreachable`: no answer (network, timeout, not JSON). `rejected`: the server answered with a status other than `ok`. The hub never relays the server's `invalid-token`, or the console would sign the hub's own user out because a hub token expired. */ export class UpstreamError extends Error { kind: 'unreachable' | 'rejected' constructor(kind: 'unreachable' | 'rejected', message: string) { super(message) this.name = 'UpstreamError' this.kind = kind } } export type Params = Record /* What `raw` hands back: the headers worth passing on and the body as a stream. */ export interface RawAnswer { contentType: string disposition?: string body: ReadableStream | null /* Stops the download (the caller went away). */ abort: () => void } interface Cached { until: number response: unknown } const MAX_CACHED = 500 /* An app download or upload can take a while. */ const WRITE_TIMEOUT_MS = 60_000 export class Upstream { #config: HubConfig #fetch: typeof fetch #cache = new Map() #now: () => number constructor(config: HubConfig, fetchImpl: typeof fetch = fetch, now: () => number = Date.now) { this.#config = config this.#fetch = fetchImpl this.#now = now } /** The `response` of the server's envelope, or an `UpstreamError`. */ async call(server: HubServer, path: string, params: Params = {}, token: string = server.token): Promise { const sorted = Object.entries(params).sort(([a], [b]) => (a < b ? -1 : a > b ? 1 : 0)) const key = `${server.id}\n${path}\n${new URLSearchParams(sorted).toString()}` const hit = this.#cache.get(key) if (hit !== undefined && hit.until > this.#now()) return hit.response const response = await this.#exec(server, path, new URLSearchParams(sorted), token, { method: 'GET' }, this.#config.timeoutMs) if (this.#config.cacheSeconds > 0) { if (this.#cache.size >= MAX_CACHED) this.#prune() this.#cache.set(key, { until: this.#now() + this.#config.cacheSeconds * 1000, response }) } return response } /* A write: POST with the server's ADMIN token and never the read one, never cached. Whatever the outcome, what the hub remembers of this server is dropped afterwards, so the next read shows the state the write left. The parameters travel in a form body (`application/x-www-form-urlencoded`) and only `token` stays in the query, so values (a record's data, a TSIG secret) do not end up in a proxy's access log. The server reads query or form alike. A `body` that is already a form gets the parameters added to it. Two bodies cannot be a form, so there the parameters stay in the query: a `FormData` (a multipart upload, the app zip or a zone file) and a string (a zone file as `text/plain`). */ async write(server: HubServer, path: string, params: Params = {}, body?: FormData | URLSearchParams | string): Promise { if (server.adminToken === undefined) throw new UpstreamError('rejected', 'no admin token configured') let query = new URLSearchParams() const init: RequestInit = { method: 'POST' } if (body === undefined || body instanceof URLSearchParams) { const form = new URLSearchParams(body) for (const [k, v] of Object.entries(params)) form.set(k, v) init.body = form } else { query = new URLSearchParams(params) init.body = body if (typeof body === 'string') init.headers = { 'content-type': 'text/plain; charset=utf-8' } } try { return await this.#exec(server, path, query, server.adminToken, init, Math.max(this.#config.timeoutMs, WRITE_TIMEOUT_MS)) } finally { this.invalidate(server.id) } } /* A file the server answers with (a zone export, a log, a block list): read with the READ token, never cached. A JSON answer is the usual envelope, so an error is raised as for any call; any other answer is handed back as it is, to be streamed to the caller. The timeout covers the wait for the headers only: a big log may take longer to arrive. */ async raw(server: HubServer, path: string, params: Params = {}): Promise { const query = new URLSearchParams(params) query.set('token', server.token) const controller = new AbortController() const timer = setTimeout(() => controller.abort(), this.#config.timeoutMs) let res: Response try { res = await this.#fetch(`${server.url}/api/${path}?${query.toString()}`, { method: 'GET', signal: controller.signal }) } catch (e) { throw new UpstreamError('unreachable', this.#describe(e, [server.token], this.#config.timeoutMs)) } finally { clearTimeout(timer) } const type = res.headers.get('content-type') ?? 'application/octet-stream' if (/json/i.test(type)) { let envelope: { status?: string; response?: unknown; errorMessage?: string } try { envelope = (await res.json()) as typeof envelope } catch (e) { throw new UpstreamError('unreachable', this.#describe(e, [server.token], this.#config.timeoutMs)) } throw envelope?.status === 'ok' ? new UpstreamError('rejected', 'it sent JSON where a file was expected') : this.#rejection(envelope) } if (!res.ok) throw new UpstreamError('rejected', `status ${res.status}`) return { contentType: type, disposition: res.headers.get('content-disposition') ?? undefined, body: res.body, abort: () => controller.abort(), } } /** Forget every cached answer of one server. */ invalidate(serverId: string): void { const prefix = `${serverId}\n` for (const key of this.#cache.keys()) if (key.startsWith(prefix)) this.#cache.delete(key) } async #exec(server: HubServer, path: string, query: URLSearchParams, token: string, init: RequestInit, timeoutMs: number): Promise { query.set('token', token) const controller = new AbortController() const timer = setTimeout(() => controller.abort(), timeoutMs) let envelope: { status?: string; response?: unknown; errorMessage?: string } try { const res = await this.#fetch(`${server.url}/api/${path}?${query.toString()}`, { ...init, signal: controller.signal, headers: { accept: 'application/json', ...(init.headers as Record | undefined) }, }) envelope = (await res.json()) as typeof envelope } catch (e) { throw new UpstreamError('unreachable', this.#describe(e, [server.token, token], timeoutMs)) } finally { clearTimeout(timer) } if (envelope?.status !== 'ok') throw this.#rejection(envelope) return envelope.response } /* The error for an envelope that is not `ok`. */ #rejection(envelope: { status?: string; errorMessage?: string } | undefined): UpstreamError { const why = envelope?.status === 'invalid-token' ? 'its API token was refused' : (envelope?.errorMessage ?? `status ${String(envelope?.status)}`) return new UpstreamError('rejected', why) } #prune(): void { const now = this.#now() for (const [key, entry] of this.#cache) if (entry.until <= now) this.#cache.delete(key) // Still full of live entries: drop the oldest inserted. while (this.#cache.size >= MAX_CACHED) this.#cache.delete(this.#cache.keys().next().value as string) } /* A short reason without anything that could carry the token. */ #describe(e: unknown, secrets: string[], timeoutMs: number): string { if (e instanceof Error && e.name === 'AbortError') return `timed out after ${timeoutMs} ms` const cause = e instanceof Error ? (e.cause as { code?: unknown } | undefined) : undefined const text = typeof cause?.code === 'string' ? cause.code : e instanceof Error ? e.message : 'request failed' return secrets.reduce((out, secret) => (secret === '' ? out : out.split(secret).join('***')), text) } }