feat(esmole): undici HTTP backend with TLS servername
This commit is contained in:
1 parent
b2a9c3323f
commit
42712d508c
2 files changed
+233
No files matched your search
@@ -0,0 +1,94 @@
|
||||
import type { Backend, BackendTarget } from '@dbmole/core'
|
||||
import { Pool } from 'undici'
|
||||
import type { ConnectionConfig } from '../schema.js'
|
||||
|
||||
export type EsResponse = { status: number; body: unknown }
|
||||
|
||||
export type EsBackend = Backend & {
|
||||
request: (
|
||||
method: string,
|
||||
path: string,
|
||||
opts?: { body?: unknown; params?: Record<string, string | number | boolean> }
|
||||
) => Promise<EsResponse>
|
||||
}
|
||||
|
||||
const authHeader = (cfg: ConnectionConfig): string | undefined => {
|
||||
if (cfg.apiKey) {
|
||||
return 'ApiKey ' + cfg.apiKey
|
||||
}
|
||||
if (cfg.user && cfg.password !== undefined) {
|
||||
return 'Basic ' + Buffer.from(cfg.user + ':' + cfg.password).toString('base64')
|
||||
}
|
||||
return undefined
|
||||
}
|
||||
|
||||
const decode = (status: number, contentType: string | undefined, text: string): EsResponse => {
|
||||
const ct = contentType ?? ''
|
||||
if (
|
||||
ct.includes('application/json') ||
|
||||
text.trimStart().startsWith('{') ||
|
||||
text.trimStart().startsWith('[')
|
||||
) {
|
||||
try {
|
||||
return { status, body: JSON.parse(text) }
|
||||
} catch {
|
||||
return { status, body: text }
|
||||
}
|
||||
}
|
||||
return { status, body: text === '' ? null : text }
|
||||
}
|
||||
|
||||
const buildQuery = (params?: Record<string, string | number | boolean>): string => {
|
||||
if (!params) {
|
||||
return ''
|
||||
}
|
||||
const usp = new URLSearchParams()
|
||||
for (const [k, v] of Object.entries(params)) {
|
||||
usp.set(k, String(v))
|
||||
}
|
||||
const s = usp.toString()
|
||||
return s ? '?' + s : ''
|
||||
}
|
||||
|
||||
export const createEsBackend = (target: BackendTarget<ConnectionConfig>): EsBackend => {
|
||||
const cfg = target.config
|
||||
const origin = cfg.scheme + '://' + target.connectHost + ':' + target.connectPort
|
||||
const pool = new Pool(origin, {
|
||||
connections: 8,
|
||||
connect:
|
||||
cfg.scheme === 'https'
|
||||
? { servername: target.serverName, rejectUnauthorized: cfg.verifyTls }
|
||||
: undefined
|
||||
})
|
||||
const auth = authHeader(cfg)
|
||||
|
||||
const request: EsBackend['request'] = async (method, path, opts) => {
|
||||
const headers: Record<string, string> = {}
|
||||
if (auth) {
|
||||
headers.authorization = auth
|
||||
}
|
||||
let body: string | undefined
|
||||
if (typeof opts?.body === 'string') {
|
||||
body = opts.body
|
||||
headers['content-type'] = 'application/json'
|
||||
} else if (opts?.body !== undefined) {
|
||||
body = JSON.stringify(opts.body)
|
||||
headers['content-type'] = 'application/json'
|
||||
}
|
||||
const p = (path.startsWith('/') ? path : '/' + path) + buildQuery(opts?.params)
|
||||
const res = await pool.request({
|
||||
method: method.toUpperCase() as never,
|
||||
path: p,
|
||||
headers,
|
||||
body
|
||||
})
|
||||
const text = await res.body.text()
|
||||
return decode(res.statusCode, res.headers['content-type'] as string | undefined, text)
|
||||
}
|
||||
|
||||
const dispose = async (): Promise<void> => {
|
||||
await pool.close()
|
||||
}
|
||||
|
||||
return { request, dispose }
|
||||
}
|
||||
@@ -0,0 +1,139 @@
|
||||
import { createServer, type Server } from 'node:http'
|
||||
import type { BackendTarget } from '@dbmole/core'
|
||||
import { afterAll, beforeAll, describe, expect, it } from 'vitest'
|
||||
import { createEsBackend } from '../../../src/es/backend.js'
|
||||
import type { ConnectionConfig } from '../../../src/schema.js'
|
||||
|
||||
let server: Server
|
||||
let port: number
|
||||
const seen: { method?: string; url?: string; auth?: string; body?: string }[] = []
|
||||
|
||||
beforeAll(async () => {
|
||||
server = createServer((req, res) => {
|
||||
let body = ''
|
||||
req.on('data', (c) => {
|
||||
body += c
|
||||
})
|
||||
req.on('end', () => {
|
||||
seen.push({ method: req.method, url: req.url, auth: req.headers.authorization, body })
|
||||
if (req.url?.startsWith('/_fail')) {
|
||||
res.writeHead(404, { 'content-type': 'application/json' })
|
||||
res.end(JSON.stringify({ error: 'index_not_found' }))
|
||||
return
|
||||
}
|
||||
if (req.url?.startsWith('/_text')) {
|
||||
res.writeHead(200, { 'content-type': 'text/plain' })
|
||||
res.end('plain text body')
|
||||
return
|
||||
}
|
||||
if (req.url?.startsWith('/_empty')) {
|
||||
res.writeHead(200, { 'content-type': 'text/plain' })
|
||||
res.end('')
|
||||
return
|
||||
}
|
||||
if (req.url?.startsWith('/_brace')) {
|
||||
// looks like JSON by leading brace but is not parseable
|
||||
res.writeHead(200, { 'content-type': 'text/plain' })
|
||||
res.end('{not json')
|
||||
return
|
||||
}
|
||||
res.writeHead(200, { 'content-type': 'application/json' })
|
||||
res.end(JSON.stringify({ ok: true, echo: body || null }))
|
||||
})
|
||||
})
|
||||
await new Promise<void>((r) => server.listen(0, '127.0.0.1', () => r()))
|
||||
port = (server.address() as { port: number }).port
|
||||
})
|
||||
|
||||
afterAll(async () => {
|
||||
await new Promise<void>((r) => server.close(() => r()))
|
||||
})
|
||||
|
||||
const target = (cfg: Partial<ConnectionConfig> = {}): BackendTarget<ConnectionConfig> => ({
|
||||
config: {
|
||||
name: 'x',
|
||||
host: '127.0.0.1',
|
||||
type: 'elasticsearch',
|
||||
scheme: 'http',
|
||||
verifyTls: false,
|
||||
user: 'u',
|
||||
password: 'p',
|
||||
readonly: false,
|
||||
...cfg
|
||||
} as ConnectionConfig,
|
||||
connectHost: '127.0.0.1',
|
||||
connectPort: port,
|
||||
serverName: '127.0.0.1'
|
||||
})
|
||||
|
||||
describe('createEsBackend', () => {
|
||||
it('sends basic auth and parses JSON', async () => {
|
||||
const b = createEsBackend(target())
|
||||
const res = await b.request('GET', '/_cluster/health')
|
||||
expect(res.status).toBe(200)
|
||||
expect(res.body).toEqual({ ok: true, echo: null })
|
||||
expect(seen.at(-1)?.auth).toMatch(/^Basic /)
|
||||
await b.dispose()
|
||||
})
|
||||
it('sends ApiKey auth when configured', async () => {
|
||||
const b = createEsBackend(target({ user: undefined, password: undefined, apiKey: 'abc==' }))
|
||||
await b.request('GET', '/')
|
||||
expect(seen.at(-1)?.auth).toBe('ApiKey abc==')
|
||||
await b.dispose()
|
||||
})
|
||||
it('returns status+body on 4xx without throwing', async () => {
|
||||
const b = createEsBackend(target())
|
||||
const res = await b.request('GET', '/_fail')
|
||||
expect(res.status).toBe(404)
|
||||
expect(res.body).toEqual({ error: 'index_not_found' })
|
||||
await b.dispose()
|
||||
})
|
||||
it('serializes object bodies as JSON and sends string bodies as-is', async () => {
|
||||
const b = createEsBackend(target())
|
||||
await b.request('POST', '/idx/_search', { body: { query: { match_all: {} } } })
|
||||
expect(JSON.parse(seen.at(-1)?.body ?? '{}')).toEqual({ query: { match_all: {} } })
|
||||
await b.request('POST', '/_bulk', { body: '{"index":{}}\n{"a":1}\n' })
|
||||
expect(seen.at(-1)?.body).toBe('{"index":{}}\n{"a":1}\n')
|
||||
await b.dispose()
|
||||
})
|
||||
it('forwards query params', async () => {
|
||||
const b = createEsBackend(target())
|
||||
await b.request('GET', '/_cat/indices', { params: { format: 'json', v: true } })
|
||||
expect(seen.at(-1)?.url).toContain('format=json')
|
||||
await b.dispose()
|
||||
})
|
||||
it('sends no auth header when neither credentials nor apiKey are present', async () => {
|
||||
const b = createEsBackend(
|
||||
target({ user: undefined, password: undefined, apiKey: undefined })
|
||||
)
|
||||
const res = await b.request('get', 'idx/_search')
|
||||
expect(res.status).toBe(200)
|
||||
expect(seen.at(-1)?.auth).toBeUndefined()
|
||||
// path without leading slash is normalized
|
||||
expect(seen.at(-1)?.url).toBe('/idx/_search')
|
||||
await b.dispose()
|
||||
})
|
||||
it('decodes plain text bodies and empty bodies as null', async () => {
|
||||
const b = createEsBackend(target())
|
||||
const text = await b.request('GET', '/_text')
|
||||
expect(text.body).toBe('plain text body')
|
||||
const empty = await b.request('GET', '/_empty')
|
||||
expect(empty.body).toBeNull()
|
||||
await b.dispose()
|
||||
})
|
||||
it('falls back to raw text when a JSON-looking body fails to parse', async () => {
|
||||
const b = createEsBackend(target())
|
||||
const res = await b.request('GET', '/_brace')
|
||||
expect(res.status).toBe(200)
|
||||
expect(res.body).toBe('{not json')
|
||||
await b.dispose()
|
||||
})
|
||||
it('configures TLS servername for https origins', async () => {
|
||||
// https target: factory must take the connect-config branch without a request
|
||||
const b = createEsBackend(
|
||||
target({ scheme: 'https', verifyTls: true, host: 'es.example.com' })
|
||||
)
|
||||
expect(typeof b.request).toBe('function')
|
||||
await b.dispose()
|
||||
})
|
||||
})
|
||||
Reference in new issue
Block a user