feat(server): support a custom CA bundle for the backend connection

apps/cli/e2e/server/tls.e2e-spec.ts

@@ -0,0 +1,106 @@
+import { readFileSync } from 'node:fs';
+import * as https from 'node:https';
+import { join } from 'node:path';
+import request from 'supertest';
+
+import { ADCServer } from '../../src/server';
+
+const readCert = (fileName: string) =>
+  readFileSync(join(__dirname, '../assets/tls/', fileName), 'utf-8');
+
+describe('Server - Backend TLS', () => {
+  let server: ADCServer;
+  // a backend whose certificate is signed by a CA the system does not trust
+  let backend: https.Server;
+  let backendURL: string;
+
+  const send = (path: string, opts: Record<string, unknown>) =>
+    request(server.TEST_ONLY_getExpress())
+      .put(path)
+      .send({
+        task: {
+          opts: {
+            backend: 'apisix',
+            server: backendURL,
+            token: 'mock',
+            cacheKey: 'default',
+            ...opts,
+          },
+          config: {},
+        },
+      });
+
+  beforeAll(async () => {
+    server = new ADCServer({
+      listen: new URL('http://127.0.0.1:3000'),
+      listenStatus: 3001,
+    });
+    backend = https.createServer(
+      { cert: readCert('server.cer'), key: readCert('server.key') },
+      (_, res) => (res.writeHead(404), res.end('{}')),
+    );
+    await new Promise<void>((resolve, reject) => {
+      const onError = (err: Error) => reject(err);
+      backend.once('error', onError);
+      backend.listen(0, '127.0.0.1', () => {
+        backend.off('error', onError);
+        const address = backend.address();
+        if (!address || typeof address === 'string')
+          return reject(new Error('backend did not bind a TCP port'));
+        // the test certificate is issued for localhost
+        backendURL = `https://localhost:${address.port}`;
+        resolve();
+      });
+    });
+  });
+
+  afterAll(async () => {
+    await new Promise<void>((resolve) => backend.close(() => resolve()));
+  });
+
+  // once the handshake succeeds the request reaches the backend, and each
+  // endpoint fails on what it finds there instead of on the certificate
+  describe.each([
+    { path: '/sync', reachedBackend: /status code 404/ },
+    { path: '/validate', reachedBackend: /Validate is not supported/ },
+  ])('$path', ({ path, reachedBackend }) => {
+    it('rejects an untrusted certificate', async () => {
+      const { body } = await send(path, {});
+
+      expect(body.message).toMatch(/unable to verify the first certificate/);
+    });
+
+    it('accepts the certificate when its CA is provided', async () => {
+      const { body } = await send(path, { caCert: readCert('ca.cer') });
+
+      expect(body.message).toMatch(reachedBackend);
+    });
+
+    it('ignores the CA bundle when verification is off', async () => {
+      const { body } = await send(path, {
+        tlsSkipVerify: true,
+        caCert: readCert('ca.cer'),
+      });
+
+      expect(body.message).toMatch(reachedBackend);
+    });
+
+    it.each([
+      ['not PEM at all', 'not-a-certificate'],
+      ['a header with no certificate', '-----BEGIN CERTIFICATE-----'],
+      [
+        'an unparseable body',
+        '-----BEGIN CERTIFICATE-----\nAAAA\n-----END CERTIFICATE-----',
+      ],
+      [
+        'one good and one broken certificate',
+        `${readCert('ca.cer')}\n-----BEGIN CERTIFICATE-----\nAAAA\n-----END CERTIFICATE-----`,
+      ],
+    ])('rejects a CA bundle that is %s', async (_, caCert) => {
+      const { status, body } = await send(path, { caCert });
+
+      expect(status).toEqual(400);
+      expect(body.message).toMatch(/caCert must be a PEM-encoded certificate/);
+    });
+  });
+});

apps/cli/src/server/agents.spec.ts

@@ -0,0 +1,63 @@
+import { MAX_CA_CERT_AGENTS, resolveHttpsAgent } from './agents';
+
+const CA_A = `-----BEGIN CERTIFICATE-----\nAAAA\n-----END CERTIFICATE-----`;
+const CA_B = `-----BEGIN CERTIFICATE-----\nBBBB\n-----END CERTIFICATE-----`;
+
+describe('Server - HTTPS agents', () => {
+  it('verifies against the system trust store by default', () => {
+    const agent = resolveHttpsAgent({});
+    expect(agent.options.rejectUnauthorized).toEqual(true);
+    expect(agent.options.ca).toBeUndefined();
+  });
+
+  it('skips verification when tlsSkipVerify is set', () => {
+    const agent = resolveHttpsAgent({ tlsSkipVerify: true });
+    expect(agent.options.rejectUnauthorized).toEqual(false);
+  });
+
+  it('verifies against the given CA bundle', () => {
+    const agent = resolveHttpsAgent({ caCert: CA_A });
+    expect(agent.options.rejectUnauthorized).toEqual(true);
+    expect(agent.options.ca).toEqual(CA_A);
+  });
+
+  it('reuses one agent per CA bundle, so sockets stay pooled', () => {
+    expect(resolveHttpsAgent({ caCert: CA_A })).toBe(
+      resolveHttpsAgent({ caCert: CA_A }),
+    );
+    expect(resolveHttpsAgent({ caCert: CA_A })).not.toBe(
+      resolveHttpsAgent({ caCert: CA_B }),
+    );
+    expect(resolveHttpsAgent({ caCert: CA_A })).not.toBe(resolveHttpsAgent({}));
+  });
+
+  it('reuses one agent across equivalent bundles', () => {
+    expect(resolveHttpsAgent({ caCert: `  ${CA_A}\n` })).toBe(
+      resolveHttpsAgent({ caCert: CA_A }),
+    );
+    expect(resolveHttpsAgent({ caCert: '   ' })).toBe(resolveHttpsAgent({}));
+  });
+
+  it('caps the cache, evicting the least recently used bundle', () => {
+    const bundle = (i: number) =>
+      `-----BEGIN CERTIFICATE-----\ncap-${i}\n-----END CERTIFICATE-----`;
+
+    const first = resolveHttpsAgent({ caCert: bundle(0) });
+    const second = resolveHttpsAgent({ caCert: bundle(1) });
+
+    // fill the cache, keeping the first bundle in use so the second ages out
+    for (let i = 2; i < MAX_CA_CERT_AGENTS + 2; i++) {
+      resolveHttpsAgent({ caCert: bundle(i) });
+      resolveHttpsAgent({ caCert: bundle(0) });
+    }
+
+    expect(resolveHttpsAgent({ caCert: bundle(0) })).toBe(first);
+    expect(resolveHttpsAgent({ caCert: bundle(1) })).not.toBe(second);
+  });
+
+  it('tlsSkipVerify takes precedence over the CA bundle', () => {
+    const agent = resolveHttpsAgent({ tlsSkipVerify: true, caCert: CA_A });
+    expect(agent.options.rejectUnauthorized).toEqual(false);
+    expect(agent.options.ca).toBeUndefined();
+  });
+});

apps/cli/src/server/agents.ts

@@ -0,0 +1,69 @@
+import { HttpAgent, HttpOptions, HttpsAgent } from 'agentkeepalive';
+
+// create connection pool
+const keepAlive: HttpOptions = {
+  keepAlive: true,
+  maxSockets: 256, // per host
+  maxFreeSockets: 16, // per host free
+  freeSocketTimeout:
+    parseInt(process.env.ADC_INGRESS_FREE_SOCKET_TIMEOUT ?? '') || 50000, // free socket keepalive for 50 seconds, and if the ADC_INGRESS_FREE_SOCKET_TIMEOUT environment variable is provided, it takes precedence.
+};
+
+export const httpAgent = new HttpAgent(keepAlive);
+
+const httpsAgent = new HttpsAgent({
+  rejectUnauthorized: true,
+  ...keepAlive,
+});
+const httpsInsecureAgent = new HttpsAgent({
+  rejectUnauthorized: false,
+  ...keepAlive,
+});
+
+// one agent per CA bundle, so sockets stay pooled across requests. the key is
+// request input, so the cache is capped and evicts least recently used first.
+// evicted agents are dropped rather than destroyed: in-flight requests finish,
+// and freeSocketTimeout reaps the sockets they leave behind.
+export const MAX_CA_CERT_AGENTS = 32;
+const httpsCACertAgents = new Map<string, HttpsAgent>();
+
+export interface TLSOptions {
+  tlsSkipVerify?: boolean;
+
+  // PEM-encoded CA certificate (or bundle) to verify the backend against,
+  // instead of the system trust store. Ignored when tlsSkipVerify is set.
+  caCert?: string;
+}
+
+//TODO: support mTLS
+export const resolveHttpsAgent = ({
+  tlsSkipVerify,
+  caCert,
+}: TLSOptions): HttpsAgent => {
+  if (tlsSkipVerify) return httpsInsecureAgent;
+
+  const ca = caCert?.trim();
+  if (!ca) return httpsAgent;
+
+  const cached = httpsCACertAgents.get(ca);
+  if (cached) {
+    // re-insert to mark it most recently used
+    httpsCACertAgents.delete(ca);
+    httpsCACertAgents.set(ca, cached);
+    return cached;
+  }
+
+  const agent = new HttpsAgent({
+    rejectUnauthorized: true,
+    ca,
+    ...keepAlive,
+  });
+  httpsCACertAgents.set(ca, agent);
+
+  if (httpsCACertAgents.size > MAX_CA_CERT_AGENTS) {
+    const lruKey = httpsCACertAgents.keys().next().value;
+    if (lruKey !== undefined) httpsCACertAgents.delete(lruKey);
+  }
+
+  return agent;
+};

apps/cli/src/server/schema.ts

@@ -1,6 +1,35 @@
 import * as ADCSDK from '@api7/adc-sdk';
+import { X509Certificate } from 'node:crypto';
 import { z } from 'zod';
 
+const tlsSkipVerify = z.boolean().optional();
+
+const PEM_CERTIFICATE =
+  /-----BEGIN CERTIFICATE-----[\s\S]*?-----END CERTIFICATE-----/g;
+
+// every certificate in the bundle is parsed: X509Certificate on its own reads
+// only the first one, and tls.createSecureContext accepts anything at all.
+const isCertificateBundle = (pem: string) => {
+  const certificates = pem.match(PEM_CERTIFICATE);
+  if (!certificates) return false;
+  try {
+    certificates.forEach((certificate) => new X509Certificate(certificate));
+    return true;
+  } catch {
+    return false;
+  }
+};
+
+// PEM-encoded CA certificate (or bundle) used to verify the backend,
+// instead of the system trust store.
+const caCert = z
+  .string()
+  .min(1)
+  .refine(isCertificateBundle, {
+    error: 'caCert must be a PEM-encoded certificate',
+  })
+  .optional();
+
 const SyncTask = z.strictObject({
   opts: z.looseObject({
     backend: z.string().min(1),
@@ -12,6 +41,8 @@ const SyncTask = z.strictObject({
     labelSelector: z.record(z.string(), z.string()).optional(),
     cacheKey: z.string(),
     bypassCache: z.boolean().optional().default(false),
+    tlsSkipVerify,
+    caCert,
   }),
   config: z.looseObject({}),
 });
@@ -31,6 +62,8 @@ const ValidateTask = z.strictObject({
     excludeResourceType: z.array(z.enum(ADCSDK.ResourceType)).optional(),
     labelSelector: z.record(z.string(), z.string()).optional(),
     cacheKey: z.string(),
+    tlsSkipVerify,
+    caCert,
   }),
   config: z.looseObject({}),
 });

apps/cli/src/server/sync.ts

@@ -1,6 +1,5 @@
 import { Differ } from '@api7/adc-differ';
 import * as ADCSDK from '@api7/adc-sdk';
-import { HttpAgent, HttpOptions, HttpsAgent } from 'agentkeepalive';
 import { AxiosResponse } from 'axios';
 import type { RequestHandler } from 'express';
 import { omit, toString } from 'lodash-es';
@@ -13,29 +12,10 @@ import {
   loadBackend,
 } from '../command/utils';
 import { check } from '../linter';
+import { httpAgent, resolveHttpsAgent } from './agents';
 import { logger } from './logger';
 import { SyncInput, type SyncInputType } from './schema';
 
-// create connection pool
-const keepAlive: HttpOptions = {
-  keepAlive: true,
-  maxSockets: 256, // per host
-  maxFreeSockets: 16, // per host free
-  freeSocketTimeout:
-    parseInt(process.env.ADC_INGRESS_FREE_SOCKET_TIMEOUT ?? '') || 50000, // free socket keepalive for 50 seconds, and if the ADC_INGRESS_FREE_SOCKET_TIMEOUT environment variable is provided, it takes precedence.
-};
-const httpAgent = new HttpAgent(keepAlive);
-
-//TODO: dynamic rejectUnauthorized and support mTLS
-const httpsAgent = new HttpsAgent({
-  rejectUnauthorized: true,
-  ...keepAlive,
-});
-const httpsInsecureAgent = new HttpsAgent({
-  rejectUnauthorized: false,
-  ...keepAlive,
-});
-
 export const syncHandler: RequestHandler<
   unknown,
   unknown,
@@ -74,7 +54,7 @@ export const syncHandler: RequestHandler<
         ? task.opts.server.join(',')
         : task.opts.server) as string,
       httpAgent,
-      httpsAgent: (task.opts as any).tlsSkipVerify ? httpsInsecureAgent : httpsAgent,
+      httpsAgent: resolveHttpsAgent(task.opts),
     });
 
     backend.on('TASK_START', ({ name }) =>

apps/cli/src/server/validate.ts

@@ -1,35 +1,15 @@
 import { Differ } from '@api7/adc-differ';
 import * as ADCSDK from '@api7/adc-sdk';
-import { HttpAgent, HttpOptions, HttpsAgent } from 'agentkeepalive';
 import type { RequestHandler } from 'express';
 import { toString } from 'lodash-es';
 import { lastValueFrom } from 'rxjs';
 
 import { fillLabels, filterResourceType, loadBackend } from '../command/utils';
 import { check } from '../linter';
+import { httpAgent, resolveHttpsAgent } from './agents';
 import { logger } from './logger';
 import { ValidateInput, type ValidateInputType } from './schema';
 
-// create connection pool
-const keepAlive: HttpOptions = {
-  keepAlive: true,
-  maxSockets: 256, // per host
-  maxFreeSockets: 16, // per host free
-  freeSocketTimeout:
-    parseInt(process.env.ADC_INGRESS_FREE_SOCKET_TIMEOUT ?? '') || 50000,
-};
-const httpAgent = new HttpAgent(keepAlive);
-
-//TODO: dynamic rejectUnauthorized and support mTLS
-const httpsAgent = new HttpsAgent({
-  rejectUnauthorized: true,
-  ...keepAlive,
-});
-const httpsInsecureAgent = new HttpsAgent({
-  rejectUnauthorized: false,
-  ...keepAlive,
-});
-
 export const validateHandler: RequestHandler<
   unknown,
   unknown,
@@ -73,7 +53,7 @@ export const validateHandler: RequestHandler<
         ? task.opts.server.join(',')
         : task.opts.server) as string,
       httpAgent,
-      httpsAgent: (task.opts as any).tlsSkipVerify ? httpsInsecureAgent : httpsAgent,
+      httpsAgent: resolveHttpsAgent(task.opts),
     });
 
     backend.on('AXIOS_DEBUG', ({ description, response }) =>