diff --git a/.projenrc.ts b/.projenrc.ts index d599b69bc..32fc4ed86 100644 --- a/.projenrc.ts +++ b/.projenrc.ts @@ -1455,6 +1455,10 @@ new BundleCli(cli, { test: 'bin/cdk --version', entryPoints: [ 'lib/index.js', + // The detached telemetry sender. A separate entry point so that it stands on its own in the + // published package (where `dependencies` are stripped), which is what lets it use the real + // `proxy-agent` instead of hand-rolling proxy support out of Node built-ins. + 'lib/cli/telemetry/sender-bundle.js', ], minifyWhitespace: true, }); diff --git a/packages/@aws-cdk-testing/cli-integ/lib/telemetry-endpoint.ts b/packages/@aws-cdk-testing/cli-integ/lib/telemetry-endpoint.ts new file mode 100644 index 000000000..f68158fa8 --- /dev/null +++ b/packages/@aws-cdk-testing/cli-integ/lib/telemetry-endpoint.ts @@ -0,0 +1,189 @@ +import { promises as fs } from 'fs'; +import * as net from 'net'; +import * as os from 'os'; +import * as path from 'path'; +import * as mockttp from 'mockttp'; +import { sleep } from './aws'; + +/** + * A local stand-in for the telemetry endpoint. + * + * Built on the same mockttp machinery as `startProxyServer`, which matters for one specific reason: + * mockttp mints a leaf certificate for the requested host signed by the CA we hand it, so the CLI + * can be pointed at `caBundlePath` with `--ca-bundle-path` and the delivery will actually complete a + * TLS handshake. A bare self-signed certificate would fail hostname verification instead. + * + * Dispose it in a `finally` block. + */ +export interface TelemetryEndpoint { + /** + * URL to point `TELEMETRY_ENDPOINT` at. + */ + readonly url: string; + + /** + * Path to the CA certificate that signs this endpoint's certificate. + * + * Pass through `NODE_EXTRA_CA_CERTS` so the CLI, and the detached sender it spawns, will trust it. + * + * Deliberately NOT `--ca-bundle-path` or `AWS_CA_BUNDLE`: those REPLACE the trust store for the + * whole CLI, so the SDK's own calls to public AWS endpoints stop verifying. `STS.GetCallerIdentity` + * then fails to find an issuer, the default account never resolves, and the app exits non-zero + * before any telemetry assertion is reached. `NODE_EXTRA_CA_CERTS` adds to the store instead. + */ + readonly caBundlePath: string; + + /** + * Every telemetry batch this endpoint has received so far. + */ + batches(): Promise; + + /** + * Wait for at least one batch to arrive. + * + * Delivery happens in a detached child that outlives the CLI, so tests have to poll rather than + * assert immediately after the command returns. + * + * @returns the first batch, or undefined if none arrived in time + */ + waitForBatch(timeoutMs?: number): Promise; + + dispose(): Promise; +} + +/** + * A batch of events as the endpoint received it. + */ +export interface TelemetryBatch { + readonly events: Array>; +} + +/** + * Options for `startTelemetryEndpoint`. + */ +export interface TelemetryEndpointOptions { + /** + * Status code to answer with. + * + * @default 200 + */ + readonly statusCode?: number; + + /** + * Where to put the generated certificate directory. + * + * @default the OS temp directory + */ + readonly certDirRoot?: string; +} + +export async function startTelemetryEndpoint(options: TelemetryEndpointOptions = {}): Promise { + const certDir = await fs.mkdtemp(path.join(options.certDirRoot ?? os.tmpdir(), 'cdk-telemetry-')); + const certPath = path.join(certDir, 'cert.pem'); + const keyPath = path.join(certDir, 'key.pem'); + + const { key, cert } = await mockttp.generateCACertificate(); + await fs.writeFile(keyPath, key); + await fs.writeFile(certPath, cert); + + const server = mockttp.getLocal({ https: { keyPath, certPath } }); + const endpoint = await server.forPost('/metrics').thenReply( + options.statusCode ?? 200, + JSON.stringify({ ok: true }), + { 'content-type': 'application/json' }, + ); + + // No port argument: mockttp picks a free one itself. Naming a port -- even a random one out of a + // range -- collides once suites run in parallel, and there is no retry to recover from it. + await server.start(); + + const batches = async (): Promise => { + const requests = await endpoint.getSeenRequests(); + return requests.map((req) => JSON.parse(req.body.buffer.toString('utf-8')) as TelemetryBatch); + }; + + return { + // `localhost` rather than 127.0.0.1: the certificate mockttp mints covers the hostname, and this + // is the name the sender will verify against. + url: `https://localhost:${server.port}/metrics`, + caBundlePath: certPath, + batches, + waitForBatch: (timeoutMs = 30_000) => waitFor(async () => (await batches())[0], timeoutMs), + async dispose() { + await server.stop(); + await fs.rm(certDir, { recursive: true, force: true }); + }, + }; +} + +/** + * A TCP listener that accepts connections and then never answers. + * + * Stands in for an endpoint that hangs, which is how we tell "the CLI did not wait for delivery" + * apart from "delivery happened to be fast". + * + * Dispose it in a `finally` block. + */ +export interface BlackHoleEndpoint { + /** + * URL to point `TELEMETRY_ENDPOINT` at. + */ + readonly url: string; + + /** + * How many connections have been accepted. + * + * A non-zero count is what proves delivery was actually attempted rather than skipped. + */ + connectionCount(): number; + + /** + * Wait for at least one connection to arrive. + */ + waitForConnection(timeoutMs?: number): Promise; + + dispose(): Promise; +} + +export async function startBlackHoleEndpoint(): Promise { + const sockets: net.Socket[] = []; + let connections = 0; + + const server = net.createServer((socket) => { + connections += 1; + sockets.push(socket); + }); + await new Promise((ok) => server.listen(0, '127.0.0.1', ok)); + const port = (server.address() as net.AddressInfo).port; + + return { + url: `https://127.0.0.1:${port}/metrics`, + connectionCount: () => connections, + waitForConnection: (timeoutMs = 30_000) => waitFor(async () => connections > 0 || undefined, timeoutMs).then((x) => x === true), + async dispose() { + for (const socket of sockets) { + socket.destroy(); + } + await new Promise((ok) => server.close(() => ok())); + }, + }; +} + +/** + * Poll `fn` until it returns something truthy, or give up after `timeoutMs`. + * + * Deliberately not `eventually` from `./eventually`: that one retries until a call stops THROWING and + * rethrows on give-up, whereas both callers here want "returned nothing within the deadline" to be a + * plain undefined they can assert on. + */ +export async function waitFor(fn: () => Promise, timeoutMs: number): Promise { + const deadline = Date.now() + timeoutMs; + while (Date.now() < deadline) { + const result = await fn(); + if (result) { + return result; + } + await sleep(500); + } + return undefined; +} diff --git a/packages/@aws-cdk-testing/cli-integ/tests/telemetry-integ-tests/cdk-cli-telemetry-disable-sends-no-data.integtest.ts b/packages/@aws-cdk-testing/cli-integ/tests/telemetry-integ-tests/cdk-cli-telemetry-disable-sends-no-data.integtest.ts index 8104ed518..9a87169cd 100644 --- a/packages/@aws-cdk-testing/cli-integ/tests/telemetry-integ-tests/cdk-cli-telemetry-disable-sends-no-data.integtest.ts +++ b/packages/@aws-cdk-testing/cli-integ/tests/telemetry-integ-tests/cdk-cli-telemetry-disable-sends-no-data.integtest.ts @@ -5,10 +5,10 @@ integTest( withDefaultFixture(async (fixture) => { const output = await fixture.cdk(['cli-telemetry', '--disable'], { verboseLevel: 3 }); - // Check the trace that telemetry was not executed successfully - expect(output).not.toContain('Telemetry Sent Successfully'); + // Check the trace that telemetry was never handed to a sender + expect(output).not.toContain('Telemetry dispatched'); // Check the trace that endpoint telemetry was never connected - expect(output).toContain('Endpoint Telemetry NOT connected'); + expect(output).toContain('Telemetry disabled'); }), ); diff --git a/packages/@aws-cdk-testing/cli-integ/tests/telemetry-integ-tests/cdk-deploy-telemetry.integtest.ts b/packages/@aws-cdk-testing/cli-integ/tests/telemetry-integ-tests/cdk-deploy-telemetry.integtest.ts index 2ddab0fd2..52ff0438e 100644 --- a/packages/@aws-cdk-testing/cli-integ/tests/telemetry-integ-tests/cdk-deploy-telemetry.integtest.ts +++ b/packages/@aws-cdk-testing/cli-integ/tests/telemetry-integ-tests/cdk-deploy-telemetry.integtest.ts @@ -13,8 +13,8 @@ integTest( verboseLevel: 3, // trace mode }); - // Check the trace that telemetry was executed successfully - expect(deployOutput).toContain('Telemetry Sent Successfully'); + // Check the trace that telemetry was handed to the detached sender that delivers it + expect(deployOutput).toContain('Telemetry dispatched'); const json = fs.readJSONSync(telemetryFile); expect(json).toEqual([ diff --git a/packages/@aws-cdk-testing/cli-integ/tests/telemetry-integ-tests/cdk-hotswap-telemetry.integtest.ts b/packages/@aws-cdk-testing/cli-integ/tests/telemetry-integ-tests/cdk-hotswap-telemetry.integtest.ts index 2a0e24790..362c3d92a 100644 --- a/packages/@aws-cdk-testing/cli-integ/tests/telemetry-integ-tests/cdk-hotswap-telemetry.integtest.ts +++ b/packages/@aws-cdk-testing/cli-integ/tests/telemetry-integ-tests/cdk-hotswap-telemetry.integtest.ts @@ -23,8 +23,8 @@ integTest( modEnv: { DYNAMIC_LAMBDA_PROPERTY_VALUE: 'updated' }, }); - // Check the trace that telemetry was executed successfully - expect(deployOutput).toContain('Telemetry Sent Successfully'); + // Check the trace that telemetry was handed to the detached sender that delivers it + expect(deployOutput).toContain('Telemetry dispatched'); const json = fs.readJSONSync(telemetryFile); expect(json).toEqual( diff --git a/packages/@aws-cdk-testing/cli-integ/tests/telemetry-integ-tests/cdk-synth-guessagent.integtest.ts b/packages/@aws-cdk-testing/cli-integ/tests/telemetry-integ-tests/cdk-synth-guessagent.integtest.ts index 8fbc2f111..18f91e22e 100644 --- a/packages/@aws-cdk-testing/cli-integ/tests/telemetry-integ-tests/cdk-synth-guessagent.integtest.ts +++ b/packages/@aws-cdk-testing/cli-integ/tests/telemetry-integ-tests/cdk-synth-guessagent.integtest.ts @@ -17,8 +17,8 @@ integTest( }, // trace mode ); - // Check the trace that telemetry was executed successfully - expect(synthOutput).toContain('Telemetry Sent Successfully'); + // Check the trace that telemetry was handed to the detached sender that delivers it + expect(synthOutput).toContain('Telemetry dispatched'); const json = fs.readJSONSync(telemetryFile); expect(json).toEqual([ diff --git a/packages/@aws-cdk-testing/cli-integ/tests/telemetry-integ-tests/cdk-synth-telemetry-with-errors.integtest.ts b/packages/@aws-cdk-testing/cli-integ/tests/telemetry-integ-tests/cdk-synth-telemetry-with-errors.integtest.ts index 2c12d39ac..d5c72f7fc 100644 --- a/packages/@aws-cdk-testing/cli-integ/tests/telemetry-integ-tests/cdk-synth-telemetry-with-errors.integtest.ts +++ b/packages/@aws-cdk-testing/cli-integ/tests/telemetry-integ-tests/cdk-synth-telemetry-with-errors.integtest.ts @@ -17,8 +17,8 @@ integTest( expect(output).toContain('This is an error'); - // Check the trace that telemetry was executed successfully despite error in synth - expect(output).toContain('Telemetry Sent Successfully'); + // Check the trace that telemetry was handed to the detached sender despite the error in synth + expect(output).toContain('Telemetry dispatched'); const json = fs.readJSONSync(telemetryFile); expect(json).toEqual([ diff --git a/packages/@aws-cdk-testing/cli-integ/tests/telemetry-integ-tests/cdk-synth-telemetry.integtest.ts b/packages/@aws-cdk-testing/cli-integ/tests/telemetry-integ-tests/cdk-synth-telemetry.integtest.ts index 3cce5306a..2737c2a16 100644 --- a/packages/@aws-cdk-testing/cli-integ/tests/telemetry-integ-tests/cdk-synth-telemetry.integtest.ts +++ b/packages/@aws-cdk-testing/cli-integ/tests/telemetry-integ-tests/cdk-synth-telemetry.integtest.ts @@ -13,8 +13,8 @@ integTest( { verboseLevel: 3 }, // trace mode ); - // Check the trace that telemetry was executed successfully - expect(synthOutput).toContain('Telemetry Sent Successfully'); + // Check the trace that telemetry was handed to the detached sender that delivers it + expect(synthOutput).toContain('Telemetry dispatched'); const json = fs.readJSONSync(telemetryFile); expect(json).toEqual([ diff --git a/packages/@aws-cdk-testing/cli-integ/tests/telemetry-integ-tests/cdk-telemetry-disable-command-posts-nothing.integtest.ts b/packages/@aws-cdk-testing/cli-integ/tests/telemetry-integ-tests/cdk-telemetry-disable-command-posts-nothing.integtest.ts new file mode 100644 index 000000000..c1a74cce1 --- /dev/null +++ b/packages/@aws-cdk-testing/cli-integ/tests/telemetry-integ-tests/cdk-telemetry-disable-command-posts-nothing.integtest.ts @@ -0,0 +1,51 @@ +import { TELEMETRY_QUIET_PERIOD_MS } from './constants'; +import { integTest, sleep, withDefaultFixture } from '../../lib'; +import { startTelemetryEndpoint } from '../../lib/telemetry-endpoint'; + +/** + * Opting out via the persisted setting has to actually stop the data leaving the machine. + * + * `cli-telemetry --disable` writes to the CDK context rather than reading an environment variable, so + * it reaches the same decision by a different route than + * `cdk-telemetry-disabled-posts-nothing`. Proven the same way: a real local endpoint, and nothing + * POSTed to it by the CLI or by the detached child that outlives it. + * + * The endpoint's CA is still supplied, via `NODE_EXTRA_CA_CERTS`, even though nothing should reach it: + * without a trusted CA "nothing arrived" would also be true of an ENABLED run whose TLS handshake + * simply failed, and the test would pass for the wrong reason. Not `--ca-bundle-path`, which REPLACES + * the trust store and breaks the SDK's own calls to public AWS endpoints. + */ +integTest( + 'cli-telemetry --disable posts nothing to the endpoint', + withDefaultFixture(async (fixture) => { + const endpoint = await startTelemetryEndpoint({ certDirRoot: fixture.integTestDir }); + try { + await fixture.cdk(['cli-telemetry', '--disable'], { + modEnv: { + CDK_HOME: fixture.integTestDir, + TELEMETRY_ENDPOINT: endpoint.url, + }, + }); + + const output = await fixture.cdkSynth({ + options: [ + fixture.fullStackName('test-1'), + ], + modEnv: { + CDK_HOME: fixture.integTestDir, + TELEMETRY_ENDPOINT: endpoint.url, + NODE_EXTRA_CA_CERTS: endpoint.caBundlePath, + }, + verboseLevel: 3, // trace + }); + + expect(output).toContain('Telemetry disabled'); + + await sleep(TELEMETRY_QUIET_PERIOD_MS); + + expect(await endpoint.batches()).toEqual([]); + } finally { + await endpoint.dispose(); + } + }), +); diff --git a/packages/@aws-cdk-testing/cli-integ/tests/telemetry-integ-tests/cdk-telemetry-disabled-posts-nothing.integtest.ts b/packages/@aws-cdk-testing/cli-integ/tests/telemetry-integ-tests/cdk-telemetry-disabled-posts-nothing.integtest.ts new file mode 100644 index 000000000..3b9e65000 --- /dev/null +++ b/packages/@aws-cdk-testing/cli-integ/tests/telemetry-integ-tests/cdk-telemetry-disabled-posts-nothing.integtest.ts @@ -0,0 +1,45 @@ +import { TELEMETRY_QUIET_PERIOD_MS } from './constants'; +import { integTest, sleep, withDefaultFixture } from '../../lib'; +import { startTelemetryEndpoint } from '../../lib/telemetry-endpoint'; + +/** + * Opting out via the environment has to actually stop the data leaving the machine. + * + * The other disable tests assert on the CLI's own trace output, which only proves the sink was never + * constructed. This points `TELEMETRY_ENDPOINT` at a real local server and proves nothing is POSTed + * to it -- including by the detached child, which outlives the CLI and would therefore not show up in + * its output at all. + * + * The endpoint's CA is still supplied, via `NODE_EXTRA_CA_CERTS`, even though nothing should reach it: + * without a trusted CA "nothing arrived" would also be true of an ENABLED run whose TLS handshake + * simply failed, and the test would pass for the wrong reason. Not `--ca-bundle-path`, which REPLACES + * the trust store and breaks the SDK's own calls to public AWS endpoints. + */ +integTest( + 'CDK_DISABLE_CLI_TELEMETRY posts nothing to the endpoint', + withDefaultFixture(async (fixture) => { + const endpoint = await startTelemetryEndpoint({ certDirRoot: fixture.integTestDir }); + try { + const output = await fixture.cdkSynth({ + options: [ + fixture.fullStackName('test-1'), + ], + modEnv: { + CDK_HOME: fixture.integTestDir, + TELEMETRY_ENDPOINT: endpoint.url, + NODE_EXTRA_CA_CERTS: endpoint.caBundlePath, + CDK_DISABLE_CLI_TELEMETRY: 'true', + }, + verboseLevel: 3, // trace + }); + + expect(output).toContain('Telemetry disabled'); + + await sleep(TELEMETRY_QUIET_PERIOD_MS); + + expect(await endpoint.batches()).toEqual([]); + } finally { + await endpoint.dispose(); + } + }), +); diff --git a/packages/@aws-cdk-testing/cli-integ/tests/telemetry-integ-tests/cdk-telemetry-does-not-block-exit.integtest.ts b/packages/@aws-cdk-testing/cli-integ/tests/telemetry-integ-tests/cdk-telemetry-does-not-block-exit.integtest.ts new file mode 100644 index 000000000..88c234b92 --- /dev/null +++ b/packages/@aws-cdk-testing/cli-integ/tests/telemetry-integ-tests/cdk-telemetry-does-not-block-exit.integtest.ts @@ -0,0 +1,102 @@ +import { integTest, withDefaultFixture } from '../../lib'; +import { startBlackHoleEndpoint } from '../../lib/telemetry-endpoint'; + +/** + * The detached sender's own network budget (`NETWORK_TIMEOUT_MS` in `lib/cli/telemetry/sender.ts`). + * + * This is the regression signature: a CLI that went back to waiting for delivery would block for this + * long against a black hole, so the overhead budget only has to stay comfortably underneath it. + */ +const SENDER_NETWORK_BUDGET_MS = 10_000; + +/** + * Smallest overhead we are willing to call a regression, however fast the machine is. + * + * Process spawn and interpreter startup are not free, and on a fast machine half the baseline is less + * than that noise. + */ +const OVERHEAD_FLOOR_MS = 2_000; + +/** + * Share of the baseline synth time we allow as overhead. + * + * Relative rather than absolute because a loaded CI machine varies run-to-run by a large fraction of + * the run's own duration; a fixed millisecond budget turns that noise into a failure. + */ +const OVERHEAD_FRACTION = 0.5; + +/** + * Largest overhead we are willing to call noise, whatever the baseline. + * + * INVARIANT: must stay comfortably below `SENDER_NETWORK_BUDGET_MS`. Letting the budget grow with an + * arbitrarily slow baseline would eventually exceed it, and the test would pass no matter what. + */ +const OVERHEAD_CEILING_MS = SENDER_NETWORK_BUDGET_MS / 2; + +/** + * How much slower the telemetry run may be than the baseline before we call it a regression. + */ +function overheadBudgetMs(baselineMs: number): number { + return Math.min(Math.max(baselineMs * OVERHEAD_FRACTION, OVERHEAD_FLOOR_MS), OVERHEAD_CEILING_MS); +} + +/** + * How many times to run each variant. The fastest run of each is compared, which is far less noisy + * than a single sample on a loaded CI machine. + */ +const RUNS = 2; + +/** + * Telemetry is delivered by a detached child, so the CLI must not wait for the POST. + * + * The endpoint is a black hole: it accepts the TCP connection and then never writes a byte, so + * anything waiting on a response hangs until its own timeout. Two things have to be true, and + * checking only one of them is how this test would quietly stop meaning anything: + * + * 1. the black hole received a connection, so delivery really was attempted; and + * 2. the CLI still exited promptly, so it was not the one waiting. + */ +integTest( + 'cdk synth does not wait for the telemetry endpoint', + withDefaultFixture(async (fixture) => { + const blackHole = await startBlackHoleEndpoint(); + + const timeSynth = async (modEnv: Record): Promise => { + const start = Date.now(); + await fixture.cdkSynth({ options: [fixture.fullStackName('test-1')], modEnv }); + return Date.now() - start; + }; + + const fastest = async (modEnv: Record): Promise => { + const timings: number[] = []; + for (let i = 0; i < RUNS; i++) { + timings.push(await timeSynth(modEnv)); + } + return Math.min(...timings); + }; + + try { + // Baseline: the same synth with telemetry switched off entirely. + const disabledMs = await fastest({ CDK_DISABLE_CLI_TELEMETRY: 'true' }); + + // The same synth, with telemetry pointed at the black hole. + const blackHoleMs = await fastest({ + CDK_HOME: fixture.integTestDir, + TELEMETRY_ENDPOINT: blackHole.url, + }); + + const overhead = blackHoleMs - disabledMs; + const budget = overheadBudgetMs(disabledMs); + fixture.log(`fastest synth with telemetry disabled: ${disabledMs}ms, pointed at a black hole: ${blackHoleMs}ms (overhead ${overhead}ms, budget ${budget}ms)`); + + // Half one: something actually tried to deliver. Without this the test would also pass if + // telemetry were silently broken. + expect(await blackHole.waitForConnection()).toBe(true); + + // Half two: whatever is hanging on the black hole, it is not the CLI. + expect(overhead).toBeLessThan(budget); + } finally { + await blackHole.dispose(); + } + }), +); diff --git a/packages/@aws-cdk-testing/cli-integ/tests/telemetry-integ-tests/cdk-telemetry-goes-through-a-proxy.integtest.ts b/packages/@aws-cdk-testing/cli-integ/tests/telemetry-integ-tests/cdk-telemetry-goes-through-a-proxy.integtest.ts new file mode 100644 index 000000000..24df8ea94 --- /dev/null +++ b/packages/@aws-cdk-testing/cli-integ/tests/telemetry-integ-tests/cdk-telemetry-goes-through-a-proxy.integtest.ts @@ -0,0 +1,82 @@ +import * as https from 'node:https'; +import type { AddressInfo } from 'node:net'; +import * as mockttp from 'mockttp'; +import { integTest, withDefaultFixture } from '../../lib'; +import { startProxyServer } from '../../lib/proxy'; +import { waitFor } from '../../lib/telemetry-endpoint'; + +/** + * Telemetry has to keep working for users behind a corporate proxy. + * + * The POST is made by a detached child process, which cannot be handed the parent's `proxy-agent` + * instance, so the proxy URL and the CA bundle path are forwarded to it as plain data and it builds + * its own agent. This proves that hand-off end to end against the same TLS-terminating proxy the + * other proxy tests use, whose certificate is signed by a throwaway CA that is in no system trust + * store. + * + * `TELEMETRY_ENDPOINT` points at a local server, so the test neither needs egress to production nor + * posts real telemetry from CI. What is under test is the CLI -> proxy hop: that the child opened a + * CONNECT tunnel and completed a TLS handshake against a certificate it could only have verified + * using the forwarded CA. The proxy -> endpoint hop is deliberately out of scope (the proxy will not + * trust the local server's self-signed certificate, which does not matter -- the proxy records the + * decrypted request either way). + */ +integTest( + 'telemetry is delivered through a configured proxy', + withDefaultFixture(async (fixture) => { + // Stand-in for the telemetry endpoint. Never actually serves a response to the proxy; it only + // needs to occupy a port so the CONNECT target is real. + const { key, cert } = await mockttp.generateCACertificate(); + const endpointServer = https.createServer({ key, cert }, (_req, res) => { + res.writeHead(200, { 'content-type': 'application/json' }); + res.end('{"ok":true}'); + }); + await new Promise((ok) => endpointServer.listen(0, '127.0.0.1', ok)); + const endpointPort = (endpointServer.address() as AddressInfo).port; + const telemetryEndpoint = `https://localhost:${endpointPort}/metrics`; + + const proxyServer = await startProxyServer(); + try { + const output = await fixture.cdkSynth({ + options: [ + fixture.fullStackName('test-1'), + '--proxy', proxyServer.url, + '--ca-bundle-path', proxyServer.certPath, + ], + modEnv: { + CDK_HOME: fixture.integTestDir, + TELEMETRY_ENDPOINT: telemetryEndpoint, + }, + verboseLevel: 3, // trace + }); + + // The parent reports the hand-off, not the delivery. + expect(output).toContain('Telemetry dispatched'); + + // Delivery happens after the CLI exits, so poll rather than asserting immediately. + const telemetryRequest = await waitFor( + async () => { + const requests = await proxyServer.getSeenRequests(); + return requests.find((req) => req.url.includes(`localhost:${endpointPort}`)); + }, + 30_000, + ); + + expect(telemetryRequest).toBeDefined(); + expect(telemetryRequest!.method).toBe('POST'); + + // The proxy terminates TLS, so we can read the decrypted body and confirm the child sent a + // well-formed batch (and therefore that both the proxy URL and the CA made it across). + const body = JSON.parse(telemetryRequest!.body.buffer.toString('utf-8')); + expect(Array.isArray(body.events)).toBe(true); + expect(body.events.length).toBeGreaterThan(0); + expect(body.events[0]).toEqual(expect.objectContaining({ + identifiers: expect.objectContaining({ sessionId: expect.anything() }), + })); + expect(telemetryRequest!.body.buffer.toString('utf-8')).not.toContain('BEGIN CERTIFICATE'); + } finally { + await proxyServer.stop(); + await new Promise((ok) => endpointServer.close(() => ok())); + } + }), +); diff --git a/packages/@aws-cdk-testing/cli-integ/tests/telemetry-integ-tests/cdk-telemetry-reaches-the-endpoint.integtest.ts b/packages/@aws-cdk-testing/cli-integ/tests/telemetry-integ-tests/cdk-telemetry-reaches-the-endpoint.integtest.ts new file mode 100644 index 000000000..a9debd903 --- /dev/null +++ b/packages/@aws-cdk-testing/cli-integ/tests/telemetry-integ-tests/cdk-telemetry-reaches-the-endpoint.integtest.ts @@ -0,0 +1,57 @@ +import { integTest, withDefaultFixture } from '../../lib'; +import { startTelemetryEndpoint } from '../../lib/telemetry-endpoint'; + +/** + * Telemetry has to actually arrive, not merely be handed off. + * + * The POST is made by a detached child process, so the CLI's own output can only ever say that the + * batch was dispatched. This points `TELEMETRY_ENDPOINT` at a local HTTPS server and waits for the + * request to turn up there, which is the only assertion that covers the whole chain: the temp-file + * hand-off, resolving and spawning the sender, forwarding the CA bundle path, and the POST itself. + * + * The endpoint's certificate is signed by a throwaway CA that is in no system trust store, so + * delivery only succeeds if the sender really trusts that CA. + * + * That CA is supplied through `NODE_EXTRA_CA_CERTS` rather than `--ca-bundle-path`, because the two do + * different things: `--ca-bundle-path` REPLACES the trust store for the whole CLI, which also breaks + * the SDK's own calls to public AWS endpoints (`STS.GetCallerIdentity` fails to find an issuer, the + * default account never resolves, and the app exits before any of this is reached). + * `NODE_EXTRA_CA_CERTS` adds to the store instead, so public roots keep working. The forwarding of + * `caBundlePath` through the payload is covered where it can be asserted in isolation: the + * `reads the CA bundle from the path it was given` sender test, and the proxy integ test. + */ +integTest( + 'telemetry is delivered to the endpoint', + withDefaultFixture(async (fixture) => { + const endpoint = await startTelemetryEndpoint({ certDirRoot: fixture.integTestDir }); + try { + await fixture.cdkSynth({ + options: [ + fixture.fullStackName('test-1'), + ], + modEnv: { + CDK_HOME: fixture.integTestDir, + TELEMETRY_ENDPOINT: endpoint.url, + NODE_EXTRA_CA_CERTS: endpoint.caBundlePath, + }, + verboseLevel: 3, // trace + }); + + // Delivery happens after the CLI exits, so poll rather than asserting immediately. + const batch = await endpoint.waitForBatch(); + + expect(batch).toBeDefined(); + expect(Array.isArray(batch!.events)).toBe(true); + expect(batch!.events.length).toBeGreaterThan(0); + expect(batch!.events[0]).toEqual(expect.objectContaining({ + identifiers: expect.objectContaining({ sessionId: expect.anything() }), + })); + + // The certificate must have travelled as a path, not as bytes in the payload: a real system + // bundle is ~190KB, and inlining it would tie every batch's size to the CA bundle's. + expect(JSON.stringify(batch)).not.toContain('BEGIN CERTIFICATE'); + } finally { + await endpoint.dispose(); + } + }), +); diff --git a/packages/@aws-cdk-testing/cli-integ/tests/telemetry-integ-tests/constants.ts b/packages/@aws-cdk-testing/cli-integ/tests/telemetry-integ-tests/constants.ts index 698b54f89..8dc48c141 100644 --- a/packages/@aws-cdk-testing/cli-integ/tests/telemetry-integ-tests/constants.ts +++ b/packages/@aws-cdk-testing/cli-integ/tests/telemetry-integ-tests/constants.ts @@ -1 +1,12 @@ export const CURRENT_TELEMETRY_VERSION = '2.0'; + +/** + * How long to keep watching a telemetry endpoint after the CLI has exited, when proving that nothing + * was sent. + * + * Delivery is asynchronous and handled by a detached child, so "nothing arrived" is only meaningful + * once we have waited longer than a successful delivery would have taken. The positive test + * (`cdk-telemetry-reaches-the-endpoint`) normally sees its batch within a second or two, so this is + * already several times the observed latency. + */ +export const TELEMETRY_QUIET_PERIOD_MS = 5_000; diff --git a/packages/aws-cdk/.projen/tasks.json b/packages/aws-cdk/.projen/tasks.json index 39dffb291..2ac2ae77a 100644 --- a/packages/aws-cdk/.projen/tasks.json +++ b/packages/aws-cdk/.projen/tasks.json @@ -207,7 +207,7 @@ "exec": "mkdir -p dist/js" }, { - "exec": "node-backpack pack --destination dist/js --allowed-license \"Apache-2.0\" --allowed-license \"MIT\" --allowed-license \"BSD-3-Clause\" --allowed-license \"ISC\" --allowed-license \"BSD-2-Clause\" --allowed-license \"0BSD\" --allowed-license \"MIT OR Apache-2.0\" --dont-attribute '^@aws-cdk/|^@cdklabs/|^cdk-assets$' --test 'bin/cdk --version' --entrypoint 'lib/index.js' --metafile dist/metafile.json" + "exec": "node-backpack pack --destination dist/js --allowed-license \"Apache-2.0\" --allowed-license \"MIT\" --allowed-license \"BSD-3-Clause\" --allowed-license \"ISC\" --allowed-license \"BSD-2-Clause\" --allowed-license \"0BSD\" --allowed-license \"MIT OR Apache-2.0\" --dont-attribute '^@aws-cdk/|^@cdklabs/|^cdk-assets$' --test 'bin/cdk --version' --entrypoint 'lib/index.js' --entrypoint 'lib/cli/telemetry/sender-bundle.js' --metafile dist/metafile.json" } ] }, @@ -222,7 +222,7 @@ "exec": "cp $(node -p 'require.resolve(\"@aws-cdk/aws-service-spec/db.json.gz\")') ./" }, { - "exec": "node-backpack validate --fix --allowed-license \"Apache-2.0\" --allowed-license \"MIT\" --allowed-license \"BSD-3-Clause\" --allowed-license \"ISC\" --allowed-license \"BSD-2-Clause\" --allowed-license \"0BSD\" --allowed-license \"MIT OR Apache-2.0\" --dont-attribute '^@aws-cdk/|^@cdklabs/|^cdk-assets$' --test 'bin/cdk --version' --entrypoint 'lib/index.js'" + "exec": "node-backpack validate --fix --allowed-license \"Apache-2.0\" --allowed-license \"MIT\" --allowed-license \"BSD-3-Clause\" --allowed-license \"ISC\" --allowed-license \"BSD-2-Clause\" --allowed-license \"0BSD\" --allowed-license \"MIT OR Apache-2.0\" --dont-attribute '^@aws-cdk/|^@cdklabs/|^cdk-assets$' --test 'bin/cdk --version' --entrypoint 'lib/index.js' --entrypoint 'lib/cli/telemetry/sender-bundle.js'" } ] }, diff --git a/packages/aws-cdk/README.md b/packages/aws-cdk/README.md index c7ab781b5..17bbf256b 100644 --- a/packages/aws-cdk/README.md +++ b/packages/aws-cdk/README.md @@ -1519,6 +1519,11 @@ that can be set in many different ways (such as `~/.cdk.json`). $ # Check the current status of telemetry $ cdk cli-telemetry --status ``` + +Telemetry is delivered by a short-lived background process, so the CLI exits without waiting for the +network. That also means nothing is reported in the CLI's own output if delivery fails; set +[`CDK_TELEMETRY_SENDER_DEBUG=1`](#environment) to see it. + ### `cdk flags` View and modify your feature flag configurations. @@ -1886,9 +1891,14 @@ in `build` will be executed by the "watch" process before deployment. The following environment variables affect aws-cdk: - `COLUMNS`: When the CLI cannot detect the terminal width (for example, when output is piped or running in CI), this standard variable is used as the rendering width for `cdk diff` tables. If unset, tables render at their natural width. +- `CDK_DISABLE_CLI_TELEMETRY`: If set to `true`, disable CLI telemetry collection (see [`cdk cli-telemetry`](#cdk-cli-telemetry)). - `CDK_DISABLE_VERSION_CHECK`: If set, disable automatic check for newer versions. - `CDK_NEW_BOOTSTRAP`: use the modern bootstrapping stack. - `CDK_ROLE_SESSION_NAME`: customize the session name used when the CLI assumes a role (for example `cdk-hnb659fds-deploy-role`). When unset, the CLI defaults to `aws-cdk-`. Useful for attributing deployments in CloudTrail when running from a CI/CD pipeline. +- `CDK_TELEMETRY_SENDER_DEBUG`: If set to `1`, print diagnostics from telemetry delivery. Telemetry is + sent by a short-lived background process that the CLI does not wait for, so its output is normally + discarded; setting this passes it through to stderr. Only useful when investigating why telemetry is + not arriving. ### Region resolution diff --git a/packages/aws-cdk/lib/cli/cli.ts b/packages/aws-cdk/lib/cli/cli.ts index b8b06353a..346fe7836 100644 --- a/packages/aws-cdk/lib/cli/cli.ts +++ b/packages/aws-cdk/lib/cli/cli.ts @@ -12,7 +12,7 @@ import { CliIoHost } from './io-host'; import { parseCommandLineArguments } from './parse-command-line-arguments'; import { checkForPlatformWarnings } from './platform-warnings'; import { prettyPrintError } from './pretty-print-error'; -import { ProxyAgentProvider } from './proxy-agent'; +import { normalizeNetworkSetting, ProxyAgentProvider } from './proxy-agent'; import { GLOBAL_PLUGIN_HOST } from './singleton-plugin-host'; import { cdkCliErrorName } from './telemetry/error'; import type { ErrorDetails } from './telemetry/schema'; @@ -123,13 +123,14 @@ export async function exec(args: string[], synthesizer?: Synthesizer): Promise 0) { diff --git a/packages/aws-cdk/lib/cli/proxy-agent.ts b/packages/aws-cdk/lib/cli/proxy-agent.ts index cef2823df..5bbb803fa 100644 --- a/packages/aws-cdk/lib/cli/proxy-agent.ts +++ b/packages/aws-cdk/lib/cli/proxy-agent.ts @@ -1,7 +1,7 @@ -import { ToolkitError } from '@aws-cdk/toolkit-lib'; +import * as path from 'node:path'; import * as fs from 'fs-extra'; import { ProxyAgent, proxies } from 'proxy-agent'; -import type { IoHelper } from '../api-private'; +import { ToolkitError } from '../toolkit-error'; /** * Validate a proxy address up front. @@ -30,6 +30,25 @@ export function validateProxyAddress(proxyAddress: string): void { } } +/** + * Coerce a raw network setting into a value the rest of the CLI can rely on. + * + * `Settings.get()` is untyped and surfaces an unset `--proxy` or `--ca-bundle-path` as either + * `undefined` or an empty array, depending on how it was parsed. An empty STRING is a different + * thing: `--proxy ''` means "go direct, ignore the proxy environment variables", so it has to survive + * normalization. Anything that is not a string counts as unconfigured, which is what makes the + * environment the fallback. + * + * Applied at the point these settings enter typed code, for two reasons. An empty array is truthy, so + * it slips past every `if (value)` guard downstream and then fails somewhere unhelpful -- + * `path.resolve([])` throws a `TypeError` that the CA-bundle resolver swallows, silently discarding + * the bundle. And both values now cross a process boundary into the detached telemetry sender, which + * has no access to the settings to re-derive them. + */ +export function normalizeNetworkSetting(raw: unknown): string | undefined { + return typeof raw === 'string' ? raw : undefined; +} + /** * Options for proxy-agent SDKs */ @@ -49,49 +68,103 @@ interface ProxyAgentOptions { readonly caBundlePath?: string; } +/** + * The proxy configuration resolved for this invocation. + */ +export interface ResolvedProxyAgent { + /** + * The agent to pass to anything making HTTPS requests in this process. + */ + readonly agent: ProxyAgent; + + /** + * Absolute path to the resolved CA bundle, if one was configured and exists on disk. + * + * Exposed because `agent` cannot cross a process boundary: the detached telemetry sender builds its + * own and needs to be told which bundle to trust. The path travels rather than the bytes, because a + * system bundle is routinely ~190KB. + * + * @default - no CA bundle was configured, or the configured one does not exist + */ + readonly caBundlePath?: string; +} + +/** + * The part of `IoHelper` that proxy resolution needs. + * + * Structural so the detached telemetry sender, which has no IoHost, can pass a writer that goes to + * stderr instead. A full `IoHelper` satisfies this as-is. + */ +export interface ProxyAgentDiagnostics { + readonly defaults: { + debug(message: string): Promise; + }; +} + export class ProxyAgentProvider { - private readonly ioHelper: IoHelper; + private readonly ioHelper: ProxyAgentDiagnostics; - public constructor(ioHelper: IoHelper) { + public constructor(ioHelper: ProxyAgentDiagnostics) { this.ioHelper = ioHelper; } - public async create(options: ProxyAgentOptions) { - // Only validate when an actual proxy address was configured. When `--proxy` - // is not given the setting is unset (and can surface at runtime as an empty - // string or empty array), in which case we skip validation and let - // ProxyAgent fall back to environment-variable detection. - if (typeof options.proxyAddress === 'string' && options.proxyAddress.length > 0) { - validateProxyAddress(options.proxyAddress); + public async create(options: ProxyAgentOptions): Promise { + const proxyAddress = normalizeNetworkSetting(options.proxyAddress); + + // Only a non-empty address is a proxy to validate. An empty one is a configured "go direct". + if (proxyAddress) { + validateProxyAddress(proxyAddress); } - // Force it to use the proxy provided through the command line. - // Otherwise, let the ProxyAgent auto-detect the proxy using environment variables. - const getProxyForUrl = options.proxyAddress != null - ? () => Promise.resolve(options.proxyAddress!) + // Force it to use the proxy provided through the command line -- including an empty one, which + // `proxy-agent` reads as "no proxy for this URL". Only an unconfigured proxy falls through to + // ProxyAgent's own environment-variable detection. + const getProxyForUrl = proxyAddress !== undefined + ? () => Promise.resolve(proxyAddress) : undefined; - return new ProxyAgent({ - ca: await this.tryGetCACert(options.caBundlePath), - getProxyForUrl, - }); + const caBundlePath = await this.resolveCABundlePath(normalizeNetworkSetting(options.caBundlePath)); + + return { + agent: new ProxyAgent({ + ca: await this.tryReadCABundle(caBundlePath), + getProxyForUrl, + }), + caBundlePath, + }; } - private async tryGetCACert(bundlePath?: string) { - const path = bundlePath || this.caBundlePathFromEnvironment(); - if (path) { - await this.ioHelper.defaults.debug(`Using CA bundle path: ${path}`); - try { - if (!fs.pathExistsSync(path)) { - return undefined; - } - return fs.readFileSync(path, { encoding: 'utf-8' }); - } catch (e: any) { - await this.ioHelper.defaults.debug(String(e)); - return undefined; - } + /** + * Resolve the configured CA bundle to an absolute path, or undefined if there isn't a usable one. + * + * Absolute because the path is handed to the detached sender, which runs from a different cwd. + */ + private async resolveCABundlePath(bundlePath?: string): Promise { + const configured = bundlePath || this.caBundlePathFromEnvironment(); + if (!configured) { + return undefined; + } + + try { + const resolved = path.resolve(configured); + await this.ioHelper.defaults.debug(`Using CA bundle path: ${resolved}`); + return fs.pathExistsSync(resolved) ? resolved : undefined; + } catch (e: any) { + await this.ioHelper.defaults.debug(String(e)); + return undefined; + } + } + + private async tryReadCABundle(bundlePath?: string): Promise { + if (!bundlePath) { + return undefined; + } + try { + return fs.readFileSync(bundlePath, { encoding: 'utf-8' }); + } catch (e: any) { + await this.ioHelper.defaults.debug(String(e)); + return undefined; } - return undefined; } /** diff --git a/packages/aws-cdk/lib/cli/telemetry/post-telemetry.ts b/packages/aws-cdk/lib/cli/telemetry/post-telemetry.ts new file mode 100644 index 000000000..c2aebbe1f --- /dev/null +++ b/packages/aws-cdk/lib/cli/telemetry/post-telemetry.ts @@ -0,0 +1,74 @@ +import type { IncomingMessage } from 'http'; +import type { Agent } from 'https'; +import { request } from 'https'; +import * as tls from 'tls'; +import type { TelemetrySchema } from './schema'; +import { ToolkitError } from '../../toolkit-error'; + +/** + * A batch of telemetry events, as the endpoint expects to receive it. + */ +export interface TelemetryBatch { + readonly events: TelemetrySchema[]; +} + +/** + * Options for a single delivery attempt. + */ +export interface PostTelemetryOptions { + /** + * Agent to make the request through, carrying proxy and CA configuration. + * + * @default - Node's default agent, i.e. a direct connection + */ + readonly agent?: Agent; + + /** + * Abort the attempt if the request has not completed within this many milliseconds. + */ + readonly timeoutMs: number; +} + +/** + * POST a batch of telemetry events, resolving with the endpoint's response. + * + * Rejects if the connection fails or the timeout expires, but NOT on an unsuccessful status code -- + * inspect `statusCode` for that. + */ +export function postTelemetry( + url: URL, + batch: TelemetryBatch, + options: PostTelemetryOptions, +): Promise { + return new Promise((ok, ko) => { + const payload = JSON.stringify(batch); + const req = request({ + hostname: url.hostname, + port: url.port || null, + path: url.pathname + url.search, + method: 'POST', + headers: { + 'content-type': 'application/json', + 'content-length': Buffer.byteLength(payload), + // The only caller makes one request and exits, so a keep-alive socket left in the agent's + // pool would just be something else holding the process open. + 'connection': 'close', + }, + agent: options.agent, + timeout: options.timeoutMs, + // Always pin identity to the destination host. `https-proxy-agent` does the TLS upgrade itself + // without passing that host to `tls.connect`, so for an IP-literal endpoint Node has nothing to + // match against and skips the check entirely -- naming the host explicitly closes that hole and + // pins to the endpoint rather than to whatever the proxy presents. + checkServerIdentity: (_host: string, cert: tls.PeerCertificate) => + tls.checkServerIdentity(url.hostname, cert), + }, ok); + + req.on('error', ko); + req.on('timeout', () => { + req.destroy(new ToolkitError('RequestTimeout', `Timeout after ${options.timeoutMs}ms, aborting request`)); + }); + + req.end(payload); + }); +} diff --git a/packages/aws-cdk/lib/cli/telemetry/sender-bundle.ts b/packages/aws-cdk/lib/cli/telemetry/sender-bundle.ts new file mode 100644 index 000000000..154e63e8b --- /dev/null +++ b/packages/aws-cdk/lib/cli/telemetry/sender-bundle.ts @@ -0,0 +1,86 @@ +import * as fs from 'node:fs'; +import type { TelemetrySenderConfig } from './sender'; +import { isSuccess, sendTelemetry, trace } from './sender'; + +/** + * Entry point for the detached telemetry sender. + * + * A dedicated esbuild entry point (see `BundleCli` in `.projenrc.ts`) so it stands on its own in the + * published package, where `dependencies` are stripped -- which is what lets it use the real + * `proxy-agent`. Spawned detached by `sink/subprocess-sink.ts` with a payload file path as its only + * argument. + */ + +/** + * Backstop for a socket that neither completes nor errors. `unref`ed, so it never keeps the process + * alive by itself; must exceed the sender's own network budget. + */ +const HARD_KILL_MS = 30_000; + +/** + * Read the payload file and delete it either way: it was written for this process alone, so leaving + * it behind would leak one file per invocation. + */ +function takePayload(payloadPath: string): string | undefined { + try { + return fs.readFileSync(payloadPath, 'utf-8'); + } catch (e: any) { + trace(`Could not read payload from ${payloadPath}: ${e?.message}`); + return undefined; + } finally { + try { + fs.unlinkSync(payloadPath); + } catch { + // The OS cleans its own temp directory. + } + } +} + +/** + * Deliver one payload. + * + * The single place delivery outcomes are handled: failures arrive as rejections, and a non-2xx is + * judged here rather than deeper down. + */ +async function deliver(payloadPath: string): Promise { + const raw = takePayload(payloadPath); + if (raw === undefined) { + return; + } + + let cfg: TelemetrySenderConfig; + try { + cfg = JSON.parse(raw) as TelemetrySenderConfig; + } catch (e: any) { + trace(`Malformed payload: ${e?.message}`); + return; + } + + try { + const statusCode = await sendTelemetry(cfg); + const ok = isSuccess(statusCode); + trace(ok ? `Telemetry sent (${statusCode})` : `Telemetry rejected with ${statusCode}`); + } catch (e: any) { + trace(`Telemetry not sent: ${e?.code ?? e?.name ?? 'Error'}: ${e?.message}`); + } +} + +async function main(): Promise { + const payloadPath = process.argv[2]; + if (!payloadPath) { + trace('No payload path was given, nothing to send'); + return; + } + await deliver(payloadPath); +} + +const hardKill = setTimeout(() => process.exit(0), HARD_KILL_MS); +hardKill.unref(); + +// Always exit 0: nobody reads this status, and a non-zero exit would make a failed delivery look +// like a crashed CLI. +const done = () => { + clearTimeout(hardKill); + process.exit(0); +}; +void main().then(done, done); diff --git a/packages/aws-cdk/lib/cli/telemetry/sender.ts b/packages/aws-cdk/lib/cli/telemetry/sender.ts new file mode 100644 index 000000000..be0dc3722 --- /dev/null +++ b/packages/aws-cdk/lib/cli/telemetry/sender.ts @@ -0,0 +1,110 @@ +import * as fs from 'node:fs'; +import type { TelemetryBatch } from './post-telemetry'; +import { postTelemetry } from './post-telemetry'; +import { ToolkitError } from '../../toolkit-error'; +import type { ProxyAgentDiagnostics } from '../proxy-agent'; +import { ProxyAgentProvider } from '../proxy-agent'; + +/** + * Budget for the delivery attempt. + * + * Generous because nothing is waiting on it: this process is detached and the CLI has already exited, + * so the only thing a longer timeout costs is a background process living a little longer. It has to + * cover a proxy handshake plus the POST on a loaded machine. + */ +const NETWORK_TIMEOUT_MS = 10_000; + +/** + * What the parent hands to this process. + */ +export interface TelemetrySenderConfig { + /** + * Absolute URL to POST the telemetry payload to. + */ + readonly endpoint: string; + + /** + * The batch of events to deliver. + */ + readonly body: TelemetryBatch; + + /** + * Proxy to route through. An empty string means "explicitly no proxy", as `--proxy ''` does. + * + * @default - resolved from the inherited proxy environment variables, as in the parent + */ + readonly proxyUrl?: string; + + /** + * Absolute path to a CA bundle to trust. The path, not the contents: a system bundle is ~190KB. + * + * @default - the default Node trust store, plus anything in `NODE_EXTRA_CA_CERTS` + */ + readonly caBundlePath?: string; + + /** + * Budget for the delivery attempt, in milliseconds. + * + * @default 10000 + */ + readonly timeoutMs?: number; +} + +/** + * POST a telemetry payload, routing through a proxy when one applies. + * + * Returns the status code for the caller to judge, and lets failures reject: every outcome is handled + * in one place, in the entry point. + */ +export async function sendTelemetry(cfg: TelemetrySenderConfig): Promise { + if (!cfg?.endpoint) { + throw new ToolkitError('NoEndpoint', 'No telemetry endpoint was given'); + } + + const url = new URL(cfg.endpoint); + + // The provider the CLI itself uses, so the child routes the way the parent would have, including + // SOCKS and PAC proxies and NO_PROXY from the inherited environment. + const { agent } = await new ProxyAgentProvider(senderDiagnostics).create({ + proxyAddress: cfg.proxyUrl, + caBundlePath: cfg.caBundlePath, + }); + + const res = await postTelemetry(url, cfg.body ?? { events: [] }, { + agent, + timeoutMs: cfg.timeoutMs ?? NETWORK_TIMEOUT_MS, + }); + + // Drain, or the socket is never released. + res.resume(); + + return res.statusCode; +} + +export function isSuccess(statusCode: number | undefined): boolean { + return statusCode !== undefined && statusCode >= 200 && statusCode < 300; +} + +/** + * Diagnostics for the detached child, which has no IoHost. + */ +export const senderDiagnostics: ProxyAgentDiagnostics = { + defaults: { + debug: async (message: string) => trace(message), + }, +}; + +/** + * Only visible when the parent was run with `CDK_TELEMETRY_SENDER_DEBUG=1`, which is also what makes + * it pass this process's stderr through. Synchronous because `process.exit` discards buffered writes. + */ +export function trace(message: string): void { + if (process.env.CDK_TELEMETRY_SENDER_DEBUG !== '1') { + return; + } + try { + fs.writeSync(2, `[cdk-telemetry-sender] ${message}\n`); + } catch { + // Diagnostics must never be the reason anything fails. + } +} diff --git a/packages/aws-cdk/lib/cli/telemetry/sink/endpoint-sink.ts b/packages/aws-cdk/lib/cli/telemetry/sink/endpoint-sink.ts deleted file mode 100644 index ed5173d01..000000000 --- a/packages/aws-cdk/lib/cli/telemetry/sink/endpoint-sink.ts +++ /dev/null @@ -1,154 +0,0 @@ -import type { IncomingMessage } from 'http'; -import type { Agent } from 'https'; -import { request } from 'https'; -import { ToolkitError } from '@aws-cdk/toolkit-lib'; -import { NetworkDetector } from '../../../api/network-detector'; -import { IoHelper } from '../../../api-private'; -import type { IIoHost } from '../../io-host'; -import type { TelemetrySchema } from '../schema'; -import type { ITelemetrySink } from './sink-interface'; - -const REQUEST_ATTEMPT_TIMEOUT_MS = 500; - -/** - * Properties for the Endpoint Telemetry Client - */ -export interface EndpointTelemetrySinkProps { - /** - * The external endpoint to hit - */ - readonly endpoint: string; - - /** - * Where messages are going to be sent - */ - readonly ioHost: IIoHost; - - /** - * The agent responsible for making the network requests. - * - * Use this to set up a proxy connection. - * - * @default - Uses the shared global node agent - */ - readonly agent?: Agent; -} - -/** - * The telemetry client that hits an external endpoint. - */ -export class EndpointTelemetrySink implements ITelemetrySink { - private events: TelemetrySchema[] = []; - private endpoint: URL; - private ioHelper: IoHelper; - private agent?: Agent; - - public constructor(props: EndpointTelemetrySinkProps) { - this.endpoint = new URL(props.endpoint); - - if (!this.endpoint.hostname || !this.endpoint.pathname) { - throw new ToolkitError('MalformedEndpoint', `Telemetry Endpoint malformed. Received hostname: ${this.endpoint.hostname}, pathname: ${this.endpoint.pathname}`); - } - - this.ioHelper = IoHelper.fromActionAwareIoHost(props.ioHost); - this.agent = props.agent; - - // Batch events every 30 seconds - setInterval(() => this.flush(), 30000).unref(); - } - - /** - * Add an event to the collection. - */ - public async emit(event: TelemetrySchema): Promise { - try { - this.events.push(event); - } catch (e: any) { - // Never throw errors, just log them via ioHost - await this.ioHelper.defaults.trace(`Failed to add telemetry event: ${e.message}`); - } - } - - public async flush(): Promise { - try { - if (this.events.length === 0) { - return; - } - - const res = await this.https(this.endpoint, { events: this.events }); - - // Clear the events array after successful output - if (res) { - this.events = []; - } - } catch (e: any) { - // Never throw errors, just log them via ioHost - await this.ioHelper.defaults.trace(`Failed to send telemetry event: ${e.message}`); - } - } - - /** - * Returns true if telemetry successfully posted, false otherwise. - */ - private async https( - url: URL, - body: { events: TelemetrySchema[] }, - ): Promise { - // Check connectivity before attempting network request - const hasConnectivity = await NetworkDetector.hasConnectivity(this.agent); - if (!hasConnectivity) { - await this.ioHelper.defaults.trace('No internet connectivity detected, skipping telemetry'); - return false; - } - - try { - const res = await doRequest(url, body, this.agent); - - // Successfully posted - if (res.statusCode && res.statusCode >= 200 && res.statusCode < 300) { - await this.ioHelper.defaults.trace('Telemetry Sent Successfully'); - return true; - } - - await this.ioHelper.defaults.trace(`Telemetry Unsuccessful: POST ${url.hostname}${url.pathname}: ${res.statusCode}:${res.statusMessage}`); - - return false; - } catch (e: any) { - await this.ioHelper.defaults.trace(`Telemetry Error: POST ${url.hostname}${url.pathname}: ${JSON.stringify(e)}`); - return false; - } - } -} - -/** - * A Promisified version of `https.request()` - */ -function doRequest( - url: URL, - data: { events: TelemetrySchema[] }, - agent?: Agent, -) { - return new Promise((ok, ko) => { - const payload: string = JSON.stringify(data); - const req = request({ - hostname: url.hostname, - port: url.port || null, - path: url.pathname, - method: 'POST', - headers: { - 'content-type': 'application/json', - 'content-length': payload.length, - }, - agent, - timeout: REQUEST_ATTEMPT_TIMEOUT_MS, - }, ok); - - req.on('error', ko); - req.on('timeout', () => { - const error = new ToolkitError('RequestTimeout', `Timeout after ${REQUEST_ATTEMPT_TIMEOUT_MS}ms, aborting request`); - req.destroy(error); - }); - - req.end(payload); - }); -} diff --git a/packages/aws-cdk/lib/cli/telemetry/sink/subprocess-sink.ts b/packages/aws-cdk/lib/cli/telemetry/sink/subprocess-sink.ts new file mode 100644 index 000000000..8bfe065d5 --- /dev/null +++ b/packages/aws-cdk/lib/cli/telemetry/sink/subprocess-sink.ts @@ -0,0 +1,235 @@ +import { spawn } from 'node:child_process'; +import { randomUUID } from 'node:crypto'; +import * as fs from 'node:fs'; +import * as os from 'node:os'; +import * as path from 'node:path'; +import { IoHelper } from '../../../api-private'; +import { ToolkitError } from '../../../toolkit-error'; +import type { IIoHost } from '../../io-host'; +import { cliRootDir } from '../../root-dir'; +import type { TelemetryBatch } from '../post-telemetry'; +import type { TelemetrySchema } from '../schema'; +import type { TelemetrySenderConfig } from '../sender'; +import type { ITelemetrySink } from './sink-interface'; + +/** + * The bundled sender entry point, relative to this package's root. + */ +const SENDER_ENTRY_POINT = path.join('lib', 'cli', 'telemetry', 'sender-bundle.js'); + +/** + * Reports a successful hand-off, NOT a successful delivery. Integration tests match on this literal. + */ +export const DISPATCHED_TRACE = 'Telemetry dispatched'; + +/** + * Properties for the subprocess telemetry sink. + */ +export interface SubprocessTelemetrySinkProps { + /** + * The external endpoint to hit + */ + readonly endpoint: string; + + /** + * Where messages are going to be sent + */ + readonly ioHost: IIoHost; + + /** + * Proxy the sender should route through, as configured by `--proxy` or the `proxy` setting. + * + * @default - resolved from the environment by the sender, as `proxy-agent` does for the rest of the CLI + */ + readonly proxyUrl?: string; + + /** + * Absolute path to the CA bundle to trust, as configured by `--ca-bundle-path` or `AWS_CA_BUNDLE`. + * + * @default - only the system trust store + */ + readonly caBundlePath?: string; + + /** + * How to locate the bundled sender entry point. + * + * Injectable so a test can exercise the missing-sender path without reaching into this object's + * privates; returning undefined is what "not on disk" looks like. + * + * @default - looked up relative to this package's root + */ + readonly resolveSender?: () => string | undefined; +} + +/** + * A telemetry sink that delivers events from a detached child process. + * + * Events are written to a temporary file and handed to a child that outlives us, so the CLI can exit + * immediately instead of waiting on the network. Nothing here ever learns whether delivery succeeded. + * + * Deliberately does not check connectivity first: that check would itself be a network call on the + * exit path, which is what this sink exists to avoid. + */ +export class SubprocessTelemetrySink implements ITelemetrySink { + private events: TelemetrySchema[] = []; + private endpoint: URL; + private ioHelper: IoHelper; + private senderPath?: string; + private proxyUrl?: string; + private caBundlePath?: string; + + public constructor(props: SubprocessTelemetrySinkProps) { + this.endpoint = new URL(props.endpoint); + + if (!this.endpoint.hostname || !this.endpoint.pathname) { + throw new ToolkitError('MalformedEndpoint', `Telemetry Endpoint malformed. Received hostname: ${this.endpoint.hostname}, pathname: ${this.endpoint.pathname}`); + } + + this.ioHelper = IoHelper.fromActionAwareIoHost(props.ioHost); + this.senderPath = (props.resolveSender ?? resolveSenderPath)(); + this.proxyUrl = props.proxyUrl; + this.caBundlePath = props.caBundlePath; + + // Batch events every 30 seconds + setInterval(() => this.flush(), 30000).unref(); + } + + /** + * Add an event to the collection. + */ + public async emit(event: TelemetrySchema): Promise { + this.events.push(event); + } + + /** + * Hand whatever has accumulated to a detached sender. + * + * Clears the batch either way: delivery is one-shot, the process that would retry has usually + * exited, and retaining the events would just re-report the failure and regrow the batch every 30s. + * This is the single place delivery failures are handled; `dispatch` reports them by throwing. + */ + public async flush(): Promise { + if (this.events.length === 0) { + return; + } + + const batch = this.events; + this.events = []; + + try { + await this.dispatch(this.endpoint, { events: batch }); + } catch (e: any) { + // Both hand-off failures arrive here: no sender on disk, and a payload write or spawn that + // throws. Nothing retries and no fallback runs, so report how much was lost, not only why. + await this.ioHelper.defaults.trace(`Failed to send telemetry event: ${e.message}. Dropped ${batch.length} event(s).`); + } + } + + /** + * Hand the batch to a detached sender process. + * + * Throws if the batch could not be handed over. + */ + private async dispatch(url: URL, body: TelemetryBatch): Promise { + if (!this.senderPath) { + throw new ToolkitError('SenderNotFound', `Unable to locate the telemetry sender at ${SENDER_ENTRY_POINT}`); + } + + const config: TelemetrySenderConfig = { + endpoint: url.href, + body, + proxyUrl: this.proxyUrl, + caBundlePath: this.caBundlePath, + }; + const payload = JSON.stringify(config); + + // A file rather than the child's stdin: writing to stdin blocks the parent once the payload + // outgrows the OS pipe buffer, which is the wait this sink exists to avoid. + const payloadPath = path.join(os.tmpdir(), `cdk-telemetry-${process.pid}-${randomUUID()}.json`); + + try { + fs.writeFileSync(payloadPath, payload, { encoding: 'utf-8', mode: 0o600 }); + + const child = spawn(process.execPath, [this.senderPath, payloadPath], { + detached: true, + // Pass the child's diagnostics through only when asked; otherwise nothing reads them. + stdio: senderDebugEnabled() ? ['ignore', 'ignore', 'inherit'] : 'ignore', + windowsHide: true, + shell: false, + // Do not hold a reference to the user's working directory; they may want to delete it. + cwd: os.tmpdir(), + }); + + // Fires after the CLI may already have exited, so it cannot go through the IoHost. Still the + // only notification for a spawn that is refused after this method returns. + child.on('error', (e: Error) => { + debugTrace(`failed to spawn sender: ${e.message}`); + tryUnlink(payloadPath); + }); + + // Node reports a refused spawn (ENOENT, EACCES, EMFILE) on that `error` event, which fires + // after this method has already returned -- it does NOT throw here. libuv does leave `pid` + // unset synchronously though, so this is the one point where the failure can still be reported + // as one. Without it the batch is counted as handed off and traced with `pid undefined`. + if (child.pid === undefined) { + throw new ToolkitError('SpawnRefused', 'the sender process was never created'); + } + + child.unref(); + + await this.ioHelper.defaults.trace(`${DISPATCHED_TRACE} (pid ${child.pid}, ${Buffer.byteLength(payload)} bytes)`); + } catch (e: any) { + tryUnlink(payloadPath); + throw new ToolkitError('DispatchFailed', `Spawning a sender for POST ${url.hostname}${url.pathname} failed: ${e.message}`); + } + } +} + +/** + * Locate the bundled sender entry point inside this package. + * + * Walks up to the package root, which works both from `lib/` in source and from the released bundle. + * `process.argv[1]` is deliberately NOT used: it may be the `.bin/cdk` symlink, the `cdk` alias + * package's wrapper, or -- when the CLI is driven programmatically -- somebody else's script. + * + * Returns undefined if the entry point is not on disk, in which case telemetry is skipped. + */ +function resolveSenderPath(): string | undefined { + const root = cliRootDir(false); + if (!root) { + return undefined; + } + + const senderPath = path.join(root, SENDER_ENTRY_POINT); + return fs.existsSync(senderPath) ? senderPath : undefined; +} + +function tryUnlink(filePath: string): void { + try { + fs.unlinkSync(filePath); + } catch { + // Nothing useful to do about it; the OS cleans its own temp directory. + } +} + +/** + * Whether the user asked to see the sender's diagnostics. + */ +function senderDebugEnabled(): boolean { + return process.env.CDK_TELEMETRY_SENDER_DEBUG === '1'; +} + +/** + * Diagnostics for failures that surface after the CLI may already have exited, so they cannot go + * through the IoHost. Gated behind the same variable as the sender's own traces. + */ +function debugTrace(message: string): void { + if (!senderDebugEnabled()) { + return; + } + try { + fs.writeSync(2, `[cdk-telemetry-dispatch] ${message}\n`); + } catch { + // Diagnostics must never be the reason anything fails. + } +} diff --git a/packages/aws-cdk/lib/toolkit-error.ts b/packages/aws-cdk/lib/toolkit-error.ts new file mode 100644 index 000000000..e7c3ba077 --- /dev/null +++ b/packages/aws-cdk/lib/toolkit-error.ts @@ -0,0 +1,12 @@ +/* eslint-disable import/no-relative-packages */ +// Re-exported from its defining module rather than from the `@aws-cdk/toolkit-lib` barrel, so the deep +// path is stated once instead of copied into every file that needs it. +// +// Required for `sender.ts`, `post-telemetry.ts` and `proxy-agent.ts`: those are the detached telemetry +// sender's bundle graph, and the barrel would drag the whole toolkit (~11MB) into it for the sake of +// one error class. The other telemetry files import it for consistency rather than necessity. +// +// `lib/api-private.ts` is not a substitute even though it re-exports from the same package: it also +// exports `deployStack`, `cfnApi`, the change-set describer and the activity printer, so importing it +// would pull the entire deployment path into the sender's graph -- the opposite of the point. +export { ToolkitError } from '../../@aws-cdk/toolkit-lib/lib/toolkit/toolkit-error'; diff --git a/packages/aws-cdk/test/cli/proxy-agent.test.ts b/packages/aws-cdk/test/cli/proxy-agent.test.ts index 06f1efec5..c662d5bc8 100644 --- a/packages/aws-cdk/test/cli/proxy-agent.test.ts +++ b/packages/aws-cdk/test/cli/proxy-agent.test.ts @@ -1,4 +1,5 @@ -import { ProxyAgentProvider, validateProxyAddress } from '../../lib/cli/proxy-agent'; +import * as path from 'node:path'; +import { normalizeNetworkSetting, ProxyAgentProvider, validateProxyAddress } from '../../lib/cli/proxy-agent'; import { TestIoHost } from '../_helpers/io-host'; describe('validateProxyAddress', () => { @@ -47,3 +48,40 @@ describe('ProxyAgentProvider', () => { await expect(provider.create({ proxyAddress })).resolves.toBeDefined(); }); }); + +describe('normalizeNetworkSetting', () => { + test('keeps an empty string, which means "go direct" and is NOT the same as unconfigured', () => { + // The whole point of normalizing: a truthiness check here would turn an explicit `--proxy ''` + // into environment auto-detection, so the CLI and the detached sender would disagree about + // whether a proxy applies. + expect(normalizeNetworkSetting('')).toBe(''); + }); + + test('keeps a configured value unchanged', () => { + expect(normalizeNetworkSetting('http://localhost:1234')).toBe('http://localhost:1234'); + expect(normalizeNetworkSetting('/etc/ssl/certs/ca.pem')).toBe('/etc/ssl/certs/ca.pem'); + }); + + test.each([ + ['undefined', undefined], + ['null', null], + // Settings.get() is untyped and surfaces an unset value as an empty array at runtime. + ['an empty array', []], + ['a populated array', ['http://a', 'http://b']], + ['a number', 8080], + ['an object', { proxy: 'http://localhost:1234' }], + ])('treats %s as unconfigured', (_desc, raw) => { + expect(normalizeNetworkSetting(raw)).toBeUndefined(); + }); + + test('an empty array would otherwise survive a truthiness check and break path.resolve', () => { + // Why this has to happen at the boundary rather than downstream: [] is truthy, so it slips past + // `if (value)` and only fails inside path.resolve, whose TypeError the CA-bundle resolver + // swallows -- silently discarding the bundle. + const raw: unknown = []; + + expect(Boolean(raw)).toBe(true); + expect(() => path.resolve(raw as string)).toThrow(/must be of type string/); + expect(normalizeNetworkSetting(raw)).toBeUndefined(); + }); +}); diff --git a/packages/aws-cdk/test/cli/telemetry/sender.test.ts b/packages/aws-cdk/test/cli/telemetry/sender.test.ts new file mode 100644 index 000000000..8a78836d1 --- /dev/null +++ b/packages/aws-cdk/test/cli/telemetry/sender.test.ts @@ -0,0 +1,674 @@ +/** + * Tests for the detached telemetry sender. + * + * These run the real thing: a real HTTPS server with a certificate signed by a throwaway CA, a real + * HTTP CONNECT proxy, and a real SOCKS5 proxy. Nothing here is mocked -- the sender's whole job is + * transport behaviour, and a mock would not tell us whether it actually works on the wire. + */ +import * as http from 'node:http'; +import * as https from 'node:https'; +import * as net from 'node:net'; +import { cleanupTestCas, generateTestCa, type TestCa } from './test-tls'; +import { isSuccess, sendTelemetry } from '../../../lib/cli/telemetry/sender'; + +jest.setTimeout(30_000); + +/** + * Anything we hold on to purely so that teardown can drop it. + */ +interface Destroyable { + destroy(): void; +} + +/** + * Shut a test server down deterministically. + * + * `server.close()` only resolves once every connection has gone, and a CONNECT tunnel is held open + * by the client (which is an agent with its own pooling policy), so waiting for that would make + * teardown depend on the agent's socket lifetime. Drop the sockets ourselves instead. + */ +function shutdown(server: net.Server, sockets: Destroyable[]): () => Promise { + return () => new Promise((ok) => { + for (const socket of sockets) { + socket.destroy(); + } + server.close(() => ok()); + }); +} + +interface Endpoint { + readonly url: string; + readonly received: Array<{ body: string; headers: http.IncomingHttpHeaders }>; + close(): Promise; +} + +async function startEndpoint(ca: TestCa, options: { statusCode?: number; urlHost?: string } = {}): Promise { + const received: Array<{ body: string; headers: http.IncomingHttpHeaders }> = []; + const sockets: Destroyable[] = []; + const server = https.createServer({ key: ca.serverKey, cert: ca.serverCert }, (req, res) => { + let body = ''; + req.on('data', (c) => (body += c)); + req.on('end', () => { + received.push({ body, headers: req.headers }); + res.writeHead(options.statusCode ?? 200, { 'content-type': 'application/json' }); + res.end('{"ok":true}'); + }); + }); + server.on('connection', (socket) => sockets.push(socket)); + await new Promise((ok) => server.listen(0, '127.0.0.1', ok)); + const port = (server.address() as net.AddressInfo).port; + return { + // Deliberately a hostname by default: the NO_PROXY and CONNECT-target tests below asserts on it. + // Tests that do not care about the hostname pass `urlHost: '127.0.0.1'` to avoid depending on how + // `localhost` resolves. + url: `https://${options.urlHost ?? 'localhost'}:${port}/metrics`, + received, + close: shutdown(server, sockets), + }; +} + +interface Proxy { + readonly url: string; + readonly connects: string[]; + readonly authHeaders: Array; + close(): Promise; +} + +interface ConnectProxyOptions { + readonly requireAuth?: string; + readonly delayConnectResponseMs?: number; +} + +async function startConnectProxy(options: ConnectProxyOptions = {}): Promise { + const connects: string[] = []; + const authHeaders: Array = []; + const sockets: Destroyable[] = []; + const server = http.createServer((_req, res) => { + res.writeHead(400); + res.end('CONNECT only'); + }); + server.on('connection', (socket) => sockets.push(socket)); + + server.on('connect', (req, clientSocket, head) => { + const auth = req.headers['proxy-authorization']; + authHeaders.push(auth); + if (options.requireAuth) { + const expected = `Basic ${Buffer.from(options.requireAuth).toString('base64')}`; + if (auth !== expected) { + clientSocket.write('HTTP/1.1 407 Proxy Authentication Required\r\n\r\n'); + clientSocket.end(); + return; + } + } + connects.push(req.url!); + const [host, port] = req.url!.split(':'); + const upstream = net.connect(Number(port), host, () => { + const established = () => { + clientSocket.write('HTTP/1.1 200 Connection Established\r\n\r\n'); + if (head?.length) { + upstream.write(head); + } + upstream.pipe(clientSocket); + clientSocket.pipe(upstream); + }; + if (options.delayConnectResponseMs) { + setTimeout(established, options.delayConnectResponseMs); + } else { + established(); + } + }); + sockets.push(upstream); + upstream.on('error', () => clientSocket.destroy()); + clientSocket.on('error', () => upstream.destroy()); + }); + + await new Promise((ok) => server.listen(0, '127.0.0.1', ok)); + const port = (server.address() as net.AddressInfo).port; + return { + url: `http://127.0.0.1:${port}`, + connects, + authHeaders, + close: shutdown(server, sockets), + }; +} + +/** + * A real (if minimal) SOCKS5 proxy: no authentication, CONNECT command only. + * + * Exists because SOCKS is the capability the hand-rolled sender could not support and this one can. + * Speaking the actual protocol is the only way to prove that. + */ +async function startSocks5Proxy(): Promise { + const connects: string[] = []; + const sockets: Destroyable[] = []; + const server = net.createServer((client) => { + sockets.push(client); + let stage: 'greeting' | 'request' | 'piping' = 'greeting'; + let buffered = Buffer.alloc(0); + + const onData = (chunk: Buffer) => { + if (stage === 'piping') { + return; + } + buffered = Buffer.concat([buffered, chunk]); + + if (stage === 'greeting') { + // VER | NMETHODS | METHODS... + if (buffered.length < 2 || buffered.length < 2 + buffered[1]) { + return; + } + buffered = buffered.subarray(2 + buffered[1]); + stage = 'request'; + client.write(Buffer.from([0x05, 0x00])); // no authentication required + } + + if (stage === 'request') { + // VER | CMD | RSV | ATYP | ADDR | PORT + if (buffered.length < 4) { + return; + } + const atyp = buffered[3]; + let host: string; + let offset: number; + if (atyp === 0x01) { + if (buffered.length < 10) { + return; + } + host = Array.from(buffered.subarray(4, 8)).join('.'); + offset = 8; + } else if (atyp === 0x03) { + const len = buffered[4]; + if (buffered.length < 5 + len + 2) { + return; + } + host = buffered.subarray(5, 5 + len).toString('utf-8'); + offset = 5 + len; + } else { + client.end(); + return; + } + const port = buffered.readUInt16BE(offset); + connects.push(`${host}:${port}`); + stage = 'piping'; + + const upstream = net.connect(port, host, () => { + // VER | REP=success | RSV | ATYP=IPv4 | BND.ADDR | BND.PORT + client.write(Buffer.from([0x05, 0x00, 0x00, 0x01, 0, 0, 0, 0, 0, 0])); + client.pipe(upstream); + upstream.pipe(client); + }); + sockets.push(upstream); + upstream.on('error', () => client.destroy()); + client.on('error', () => upstream.destroy()); + } + }; + + client.on('data', onData); + client.on('error', () => client.destroy()); + }); + + await new Promise((ok) => server.listen(0, '127.0.0.1', ok)); + const port = (server.address() as net.AddressInfo).port; + return { + url: `socks5://127.0.0.1:${port}`, + connects, + authHeaders: [], + close: shutdown(server, sockets), + }; +} + +/** + * An HTTPS endpoint that completes the handshake, reads the request, and then never answers. + * + * Exercises the sender's request budget: the connection is perfectly healthy, so only the timeout + * can end the attempt. + */ +async function startStalledEndpoint(ca: TestCa): Promise { + const sockets: Destroyable[] = []; + const server = https.createServer({ key: ca.serverKey, cert: ca.serverCert }, () => { + // Deliberately no response. + }); + server.on('connection', (socket) => sockets.push(socket)); + await new Promise((ok) => server.listen(0, '127.0.0.1', ok)); + const port = (server.address() as net.AddressInfo).port; + return { + // Connect by IP, matching the bind address. No test here cares about the hostname, and resolving + // `localhost` to ::1 first -- which Node 18+ does on a dual-stack box -- would ECONNREFUSED + // against a listener bound only to 127.0.0.1. Covered by the certificate's `IP:127.0.0.1` SAN. + url: `https://127.0.0.1:${port}/metrics`, + received: [], + close: shutdown(server, sockets), + }; +} + +const BODY = { events: [{ identifiers: { sessionId: 'test-session' } }] as any }; + +/** + * Assert that delivery failed, and describe how. + * + * Node reports transport problems in `code` (`ECONNREFUSED`, `ERR_TLS_CERT_ALTNAME_INVALID`) while + * our own failures arrive as an error `name`, so tests should not have to know which one carries the + * detail. + */ +function failure(promise: Promise): Promise { + return promise.then( + () => { + throw new Error('expected delivery to fail, but it succeeded'); + }, + (e: any) => `${e?.code ?? ''}|${e?.name ?? ''}|${e?.message ?? ''}`, + ); +} + +describe('sender', () => { + let ca: TestCa; + const savedEnv = { ...process.env }; + + beforeAll(() => { + ca = generateTestCa(); + }); + + afterAll(() => { + cleanupTestCas(); + }); + + afterEach(() => { + // `proxy-agent` reads the proxy environment directly, so tests that exercise auto-detection have + // to mutate it for real. + for (const key of ['HTTPS_PROXY', 'https_proxy', 'HTTP_PROXY', 'http_proxy', 'NO_PROXY', 'no_proxy', 'ALL_PROXY', 'all_proxy']) { + delete process.env[key]; + if (savedEnv[key] !== undefined) { + process.env[key] = savedEnv[key]; + } + } + }); + + describe('direct delivery', () => { + test('POSTs the payload and reports success', async () => { + const endpoint = await startEndpoint(ca); + try { + await expect(sendTelemetry({ endpoint: endpoint.url, body: BODY, caBundlePath: ca.caCertPath, timeoutMs: 5000 })).resolves.toBe(200); + + expect(endpoint.received).toHaveLength(1); + expect(JSON.parse(endpoint.received[0].body)).toEqual(BODY); + expect(endpoint.received[0].headers['content-type']).toBe('application/json'); + } finally { + await endpoint.close(); + } + }); + + test('reports a non-2xx status without treating it as delivered', async () => { + const endpoint = await startEndpoint(ca, { statusCode: 500 }); + try { + const statusCode = await sendTelemetry({ endpoint: endpoint.url, body: BODY, caBundlePath: ca.caCertPath, timeoutMs: 5000 }); + + expect(statusCode).toBe(500); + expect(isSuccess(statusCode)).toBe(false); + } finally { + await endpoint.close(); + } + }); + + test('rejects an untrusted certificate when no CA bundle is supplied', async () => { + const endpoint = await startEndpoint(ca); + try { + await expect(failure(sendTelemetry({ endpoint: endpoint.url, body: BODY, timeoutMs: 5000 }))) + .resolves.toContain('UNABLE_TO_VERIFY_LEAF_SIGNATURE'); + + expect(endpoint.received).toHaveLength(0); + } finally { + await endpoint.close(); + } + }); + + test('reports connection failures by rejecting', async () => { + // Port 1 is reserved and nothing listens on it. + await expect(failure(sendTelemetry({ endpoint: 'https://127.0.0.1:1/metrics', body: BODY, timeoutMs: 2000 }))) + .resolves.toContain('ECONNREFUSED'); + }); + + test('a CA bundle path that does not exist falls back to the system trust store', async () => { + // Rather than crashing or silently trusting everything: the endpoint's certificate is not + // signed by a public root, so this must fail verification. + const endpoint = await startEndpoint(ca); + try { + await expect(failure(sendTelemetry({ + endpoint: endpoint.url, + body: BODY, + caBundlePath: '/definitely/not/a/real/bundle.pem', + timeoutMs: 5000, + }))).resolves.toContain('UNABLE_TO_VERIFY_LEAF_SIGNATURE'); + } finally { + await endpoint.close(); + } + }); + }); + + describe('proxy delivery', () => { + test('tunnels through an http:// proxy with CONNECT', async () => { + const endpoint = await startEndpoint(ca); + const proxy = await startConnectProxy(); + try { + await expect(sendTelemetry({ + endpoint: endpoint.url, + body: BODY, + proxyUrl: proxy.url, + caBundlePath: ca.caCertPath, + timeoutMs: 5000, + })).resolves.toBe(200); + + expect(proxy.connects).toHaveLength(1); + expect(proxy.connects[0]).toMatch(/^localhost:\d+$/); + expect(JSON.parse(endpoint.received[0].body)).toEqual(BODY); + } finally { + await proxy.close(); + await endpoint.close(); + } + }); + + test('sends Basic credentials embedded in the proxy URL', async () => { + const endpoint = await startEndpoint(ca); + const proxy = await startConnectProxy({ requireAuth: 'alice:s3cret' }); + try { + const authed = proxy.url.replace('http://', 'http://alice:s3cret@'); + await expect(sendTelemetry({ + endpoint: endpoint.url, body: BODY, proxyUrl: authed, caBundlePath: ca.caCertPath, timeoutMs: 5000, + })).resolves.toBe(200); + + expect(proxy.authHeaders[0]).toBe(`Basic ${Buffer.from('alice:s3cret').toString('base64')}`); + } finally { + await proxy.close(); + await endpoint.close(); + } + }); + + test('surfaces a proxy 407 as a status code, not as a delivery', async () => { + // `https-proxy-agent` replays a non-200 CONNECT response through the HTTP machinery (and + // destroys the original socket so the request body is never written to the proxy), so this + // arrives as an ordinary status code for the caller to judge. + const endpoint = await startEndpoint(ca); + const proxy = await startConnectProxy({ requireAuth: 'alice:s3cret' }); + try { + const statusCode = await sendTelemetry({ + endpoint: endpoint.url, body: BODY, proxyUrl: proxy.url, caBundlePath: ca.caCertPath, timeoutMs: 5000, + }); + + expect(statusCode).toBe(407); + expect(isSuccess(statusCode)).toBe(false); + expect(endpoint.received).toHaveLength(0); + } finally { + await proxy.close(); + await endpoint.close(); + } + }); + + test('discovers the proxy from the environment when none is configured', async () => { + const endpoint = await startEndpoint(ca); + const proxy = await startConnectProxy(); + try { + process.env.HTTPS_PROXY = proxy.url; + + await expect(sendTelemetry({ endpoint: endpoint.url, body: BODY, caBundlePath: ca.caCertPath, timeoutMs: 5000 })).resolves.toBe(200); + + expect(proxy.connects).toHaveLength(1); + } finally { + await proxy.close(); + await endpoint.close(); + } + }); + + test('honours NO_PROXY and goes direct', async () => { + const endpoint = await startEndpoint(ca); + const proxy = await startConnectProxy(); + try { + process.env.HTTPS_PROXY = proxy.url; + process.env.NO_PROXY = 'localhost'; + + await expect(sendTelemetry({ endpoint: endpoint.url, body: BODY, caBundlePath: ca.caCertPath, timeoutMs: 5000 })).resolves.toBe(200); + + expect(proxy.connects).toHaveLength(0); + } finally { + await proxy.close(); + await endpoint.close(); + } + }); + + test('an explicitly empty proxy means direct, not environment auto-detect', async () => { + // The parent forces whatever `--proxy` was set to, even an empty string, and does not consult + // the environment in that case. The child has to agree, or the two disagree about whether a + // proxy applies. + const endpoint = await startEndpoint(ca); + const proxy = await startConnectProxy(); + try { + process.env.HTTPS_PROXY = proxy.url; + + await expect(sendTelemetry({ endpoint: endpoint.url, body: BODY, caBundlePath: ca.caCertPath, timeoutMs: 5000, proxyUrl: '' })).resolves.toBe(200); + + expect(proxy.connects).toHaveLength(0); + } finally { + await proxy.close(); + await endpoint.close(); + } + }); + + test('tolerates a proxy handshake slower than the in-process 500ms budget', async () => { + // Regression: the sender used to inherit the parent's 500ms exit budget and apply it to EVERY + // step of a proxied send, so a proxy that took longer than that to establish the tunnel was + // silently dropped. That is what broke this path on loaded CI runners. + const endpoint = await startEndpoint(ca); + const proxy = await startConnectProxy({ delayConnectResponseMs: 800 }); + try { + // Deliberately no `timeoutMs`: this exercises the sender's own default budget. + await expect(sendTelemetry({ + endpoint: endpoint.url, + body: BODY, + proxyUrl: proxy.url, + caBundlePath: ca.caCertPath, + })).resolves.toBe(200); + + expect(JSON.parse(endpoint.received[0].body)).toEqual(BODY); + } finally { + await proxy.close(); + await endpoint.close(); + } + }); + + test('gives up on an endpoint that accepts the connection but never responds', async () => { + // The budget was widened, not removed. + const stalled = await startStalledEndpoint(ca); + try { + await expect(failure(sendTelemetry({ + endpoint: stalled.url, + body: BODY, + caBundlePath: ca.caCertPath, + timeoutMs: 300, + }))).resolves.toContain('RequestTimeout'); + } finally { + await stalled.close(); + } + }); + }); + + describe('SOCKS support', () => { + // The reason this sender reuses `proxy-agent` instead of hand-rolling HTTP CONNECT: a + // builtins-only sender cannot speak SOCKS, so it had to skip these users entirely. + // + // These two address the endpoint by IP rather than by name, because `socks5://` (unlike + // `socks5h://`) resolves the destination on THIS side and puts the resulting address in the SOCKS + // request. Given a hostname, what lands there depends on how `localhost` happens to resolve: an + // IPv6-first box sends an ATYP=0x04 address, which `startSocks5Proxy` below does not implement, + // and the connection is closed rather than proxied. A literal IPv4 address is not resolved at all, + // so the request shape is the same everywhere. Covered by the certificate's `IP:127.0.0.1` SAN. + test('delivers through a socks5:// proxy', async () => { + const endpoint = await startEndpoint(ca, { urlHost: '127.0.0.1' }); + const proxy = await startSocks5Proxy(); + try { + await expect(sendTelemetry({ + endpoint: endpoint.url, + body: BODY, + proxyUrl: proxy.url, + caBundlePath: ca.caCertPath, + timeoutMs: 5000, + })).resolves.toBe(200); + + expect(proxy.connects).toHaveLength(1); + expect(JSON.parse(endpoint.received[0].body)).toEqual(BODY); + } finally { + await proxy.close(); + await endpoint.close(); + } + }); + + test('discovers a socks5:// proxy from the environment', async () => { + const endpoint = await startEndpoint(ca, { urlHost: '127.0.0.1' }); + const proxy = await startSocks5Proxy(); + try { + process.env.HTTPS_PROXY = proxy.url; + + await expect(sendTelemetry({ endpoint: endpoint.url, body: BODY, caBundlePath: ca.caCertPath, timeoutMs: 5000 })).resolves.toBe(200); + + expect(proxy.connects).toHaveLength(1); + } finally { + await proxy.close(); + await endpoint.close(); + } + }); + }); + + describe('fails closed', () => { + test('does not fall back to a direct connection when the proxy is unreachable', async () => { + // A proxy is normally mandatory rather than advisory: corporate setups firewall direct egress, + // so bypassing it would be both futile and a policy violation. + const endpoint = await startEndpoint(ca); + try { + await expect(sendTelemetry({ + endpoint: endpoint.url, + body: BODY, + proxyUrl: 'http://127.0.0.1:1', + caBundlePath: ca.caCertPath, + timeoutMs: 5000, + })).rejects.toThrow(); + + expect(endpoint.received).toHaveLength(0); + } finally { + await endpoint.close(); + } + }); + + test('rejects a proxy address with an unsupported protocol', async () => { + const endpoint = await startEndpoint(ca); + try { + await expect(sendTelemetry({ + endpoint: endpoint.url, + body: BODY, + proxyUrl: 'gopher://127.0.0.1:70', + caBundlePath: ca.caCertPath, + timeoutMs: 5000, + })).rejects.toThrow(/Unsupported protocol/); + + expect(endpoint.received).toHaveLength(0); + } finally { + await endpoint.close(); + } + }); + + test('rejects a proxy address with no protocol at all', async () => { + await expect(sendTelemetry({ endpoint: 'https://example.com/m', body: BODY, proxyUrl: ':::not a url', timeoutMs: 500 })) + .rejects.toThrow(/Invalid proxy address/); + }); + + test.each([ + ['a missing endpoint', {}], + ['an empty endpoint', { endpoint: '' }], + ['a malformed endpoint', { endpoint: 'not-a-url' }], + ])('rejects %s', async (_name, cfg) => { + await expect(sendTelemetry(cfg as any)).rejects.toThrow(); + }); + + test('rejects garbage input rather than reporting a phantom send', async () => { + await expect(failure(sendTelemetry(undefined as any))).resolves.toContain('NoEndpoint'); + await expect(failure(sendTelemetry(null as any))).resolves.toContain('NoEndpoint'); + }); + }); + + describe('certificate identity', () => { + // Trusting the signer is not enough -- the certificate also has to cover the host we asked for. + test('rejects a hostname mismatch on the direct path', async () => { + const wrongCa = generateTestCa({ subjectAltName: 'DNS:not-the-endpoint.example.com' }); + const endpoint = await startEndpoint(wrongCa); + try { + await expect(failure(sendTelemetry({ endpoint: endpoint.url, body: BODY, caBundlePath: wrongCa.caCertPath, timeoutMs: 5000 }))) + .resolves.toContain('ERR_TLS_CERT_ALTNAME_INVALID'); + + expect(endpoint.received).toHaveLength(0); + } finally { + await endpoint.close(); + } + }); + + test('rejects a hostname mismatch through a proxy', async () => { + const wrongCa = generateTestCa({ subjectAltName: 'DNS:not-the-endpoint.example.com' }); + const endpoint = await startEndpoint(wrongCa); + const proxy = await startConnectProxy(); + try { + await expect(failure(sendTelemetry({ + endpoint: endpoint.url, + body: BODY, + proxyUrl: proxy.url, + caBundlePath: wrongCa.caCertPath, + timeoutMs: 5000, + }))).resolves.toContain('ERR_TLS_CERT_ALTNAME_INVALID'); + + // The tunnel opened, but the handshake to the endpoint must not have. + expect(proxy.connects).toHaveLength(1); + expect(endpoint.received).toHaveLength(0); + } finally { + await proxy.close(); + await endpoint.close(); + } + }); + + test('rejects an IP-literal endpoint whose certificate omits that IP, through a proxy', async () => { + // The certificate covers DNS:localhost but NOT IP:127.0.0.1, and the proxy is reached as an IP + // too -- so if identity were checked against the proxy's host instead of the destination, this + // would be wrongly accepted. + const localhostOnlyCa = generateTestCa({ subjectAltName: 'DNS:localhost' }); + const endpoint = await startEndpoint(localhostOnlyCa, { urlHost: '127.0.0.1' }); + const proxy = await startConnectProxy(); + try { + await expect(failure(sendTelemetry({ + endpoint: endpoint.url, + body: BODY, + proxyUrl: proxy.url, + caBundlePath: localhostOnlyCa.caCertPath, + timeoutMs: 5000, + }))).resolves.toContain('ERR_TLS_CERT_ALTNAME_INVALID'); + + expect(proxy.connects[0]).toMatch(/^127\.0\.0\.1:\d+$/); + expect(endpoint.received).toHaveLength(0); + } finally { + await proxy.close(); + await endpoint.close(); + } + }); + + test('accepts an IP-literal endpoint whose certificate does cover that IP, through a proxy', async () => { + // The mirror image, so the test above is not just asserting that IP literals never work. + const endpoint = await startEndpoint(ca, { urlHost: '127.0.0.1' }); + const proxy = await startConnectProxy(); + try { + await expect(sendTelemetry({ + endpoint: endpoint.url, + body: BODY, + proxyUrl: proxy.url, + caBundlePath: ca.caCertPath, + timeoutMs: 5000, + })).resolves.toBe(200); + + expect(JSON.parse(endpoint.received[0].body)).toEqual(BODY); + } finally { + await proxy.close(); + await endpoint.close(); + } + }); + }); +}); diff --git a/packages/aws-cdk/test/cli/telemetry/sink/endpoint-sink.test.ts b/packages/aws-cdk/test/cli/telemetry/sink/endpoint-sink.test.ts deleted file mode 100644 index ec024982e..000000000 --- a/packages/aws-cdk/test/cli/telemetry/sink/endpoint-sink.test.ts +++ /dev/null @@ -1,342 +0,0 @@ -import * as https from 'https'; -import { createTestEvent } from './util'; -import { NetworkDetector } from '../../../../lib/api/network-detector'; -import { IoHelper } from '../../../../lib/api-private'; -import { CliIoHost } from '../../../../lib/cli/io-host'; -import { EndpointTelemetrySink } from '../../../../lib/cli/telemetry/sink/endpoint-sink'; - -// Mock the https module -jest.mock('https', () => ({ - request: jest.fn(), -})); - -// Mock NetworkDetector -jest.mock('../../../../lib/api/network-detector', () => ({ - NetworkDetector: { - hasConnectivity: jest.fn(), - }, -})); - -describe('EndpointTelemetrySink', () => { - let ioHost: CliIoHost; - - beforeEach(() => { - jest.resetAllMocks(); - - // Mock NetworkDetector to return true by default for existing tests - (NetworkDetector.hasConnectivity as jest.Mock).mockResolvedValue(true); - - ioHost = CliIoHost.instance(); - }); - - afterEach(() => { - jest.restoreAllMocks(); - }); - - // Helper to create a mock request object with the necessary event handlers - function setupMockRequest() { - // Create a mock response object with a successful status code - const mockResponse = { - statusCode: 200, - statusMessage: 'OK', - }; - - // Create the mock request object - const mockRequest = { - on: jest.fn(), - end: jest.fn(), - setTimeout: jest.fn(), - }; - - // Mock the https.request to return our mockRequest - (https.request as jest.Mock).mockImplementation((_, callback) => { - // If a callback was provided, call it with our mock response - if (callback) { - setTimeout(() => callback(mockResponse), 0); - } - return mockRequest; - }); - - return mockRequest; - } - - test('makes a POST request to the specified endpoint', async () => { - // GIVEN - const mockRequest = setupMockRequest(); - const testEvent = createTestEvent('INVOKE', { foo: 'bar' }); - const client = new EndpointTelemetrySink({ endpoint: 'https://example.com/telemetry', ioHost }); - - // WHEN - await client.emit(testEvent); - await client.flush(); - - // THEN - const expectedPayload = JSON.stringify({ events: [testEvent] }); - expect(https.request).toHaveBeenCalledWith({ - hostname: 'example.com', - port: null, - path: '/telemetry', - method: 'POST', - headers: { - 'content-type': 'application/json', - 'content-length': expectedPayload.length, - }, - agent: undefined, - timeout: 500, - }, expect.anything()); - - expect(mockRequest.end).toHaveBeenCalledWith(expectedPayload); - }); - - test('silently catches request errors', async () => { - // GIVEN - const mockRequest = setupMockRequest(); - const testEvent = createTestEvent('INVOKE'); - const client = new EndpointTelemetrySink({ endpoint: 'https://example.com/telemetry', ioHost }); - - mockRequest.on.mockImplementation((event, callback) => { - if (event === 'error') { - callback(new Error('Network error')); - } - return mockRequest; - }); - - await client.emit(testEvent); - - // THEN - await expect(client.flush()).resolves.not.toThrow(); - }); - - test('multiple events sent as one', async () => { - // GIVEN - const mockRequest = setupMockRequest(); - const testEvent1 = createTestEvent('INVOKE', { foo: 'bar' }); - const testEvent2 = createTestEvent('INVOKE', { foo: 'bazoo' }); - const client = new EndpointTelemetrySink({ endpoint: 'https://example.com/telemetry', ioHost }); - - // WHEN - await client.emit(testEvent1); - await client.emit(testEvent2); - await client.flush(); - - // THEN - const expectedPayload = JSON.stringify({ events: [testEvent1, testEvent2] }); - expect(https.request).toHaveBeenCalledTimes(1); - expect(https.request).toHaveBeenCalledWith({ - hostname: 'example.com', - port: null, - path: '/telemetry', - method: 'POST', - headers: { - 'content-type': 'application/json', - 'content-length': expectedPayload.length, - }, - agent: undefined, - timeout: 500, - }, expect.anything()); - - expect(mockRequest.end).toHaveBeenCalledWith(expectedPayload); - }); - - test('successful flush clears events cache', async () => { - // GIVEN - setupMockRequest(); - const testEvent1 = createTestEvent('INVOKE', { foo: 'bar' }); - const testEvent2 = createTestEvent('INVOKE', { foo: 'bazoo' }); - const client = new EndpointTelemetrySink({ endpoint: 'https://example.com/telemetry', ioHost }); - - // WHEN - await client.emit(testEvent1); - await client.flush(); - await client.emit(testEvent2); - await client.flush(); - - // THEN - const expectedPayload1 = JSON.stringify({ events: [testEvent1] }); - expect(https.request).toHaveBeenCalledTimes(2); - expect(https.request).toHaveBeenCalledWith({ - hostname: 'example.com', - port: null, - path: '/telemetry', - method: 'POST', - headers: { - 'content-type': 'application/json', - 'content-length': expectedPayload1.length, - }, - agent: undefined, - timeout: 500, - }, expect.anything()); - - const expectedPayload2 = JSON.stringify({ events: [testEvent2] }); - expect(https.request).toHaveBeenCalledWith({ - hostname: 'example.com', - port: null, - path: '/telemetry', - method: 'POST', - headers: { - 'content-type': 'application/json', - 'content-length': expectedPayload2.length, - }, - agent: undefined, - timeout: 500, - }, expect.anything()); - }); - - test('failed flush does not clear events cache', async () => { - // GIVEN - const mockRequest = { - on: jest.fn(), - end: jest.fn(), - setTimeout: jest.fn(), - }; - // Mock the https.request to return the first response as 503 - (https.request as jest.Mock).mockImplementationOnce((_, callback) => { - // If a callback was provided, call it with our mock response - if (callback) { - setTimeout(() => callback({ - statusCode: 503, - statusMessage: 'Service Unavailable', - }), 0); - } - return mockRequest; - }).mockImplementation((_, callback) => { - if (callback) { - setTimeout(() => callback({ - statusCode: 200, - statusMessage: 'Success', - }), 0); - } - return mockRequest; - }); - - const testEvent1 = createTestEvent('INVOKE', { foo: 'bar' }); - const testEvent2 = createTestEvent('INVOKE', { foo: 'bazoo' }); - const client = new EndpointTelemetrySink({ endpoint: 'https://example.com/telemetry', ioHost }); - - // WHEN - await client.emit(testEvent1); - - // mocked to fail - await client.flush(); - - await client.emit(testEvent2); - - // mocked to succeed - await client.flush(); - - // THEN - const expectedPayload1 = JSON.stringify({ events: [testEvent1] }); - expect(https.request).toHaveBeenCalledTimes(2); - expect(https.request).toHaveBeenCalledWith({ - hostname: 'example.com', - port: null, - path: '/telemetry', - method: 'POST', - headers: { - 'content-type': 'application/json', - 'content-length': expectedPayload1.length, - }, - agent: undefined, - timeout: 500, - }, expect.anything()); - - const expectedPayload2 = JSON.stringify({ events: [testEvent1, testEvent2] }); - expect(https.request).toHaveBeenCalledWith({ - hostname: 'example.com', - port: null, - path: '/telemetry', - method: 'POST', - headers: { - 'content-type': 'application/json', - 'content-length': expectedPayload2.length, - }, - agent: undefined, - timeout: 500, - }, expect.anything()); - }); - - test('flush is called every 30 seconds', async () => { - // GIVEN - jest.useFakeTimers(); - setupMockRequest(); // Setup the mock request but we don't need the return value - - // Create a spy on setInterval - const setIntervalSpy = jest.spyOn(global, 'setInterval'); - - // Create the client - const client = new EndpointTelemetrySink({ endpoint: 'https://example.com/telemetry', ioHost }); - - // Create a spy on the flush method - const flushSpy = jest.spyOn(client, 'flush'); - - // WHEN - // Advance the timer by 30 seconds - jest.advanceTimersByTime(30000); - - // THEN - // Verify setInterval was called with the correct interval - expect(setIntervalSpy).toHaveBeenCalledWith(expect.any(Function), 30000); - - // Verify flush was called - expect(flushSpy).toHaveBeenCalledTimes(1); - - // Advance the timer by another 30 seconds - jest.advanceTimersByTime(30000); - - // Verify flush was called again - expect(flushSpy).toHaveBeenCalledTimes(2); - - // Clean up - jest.useRealTimers(); - setIntervalSpy.mockRestore(); - }); - - test('handles errors gracefully and logs to trace without throwing', async () => { - // GIVEN - const testEvent = createTestEvent('INVOKE'); - - // Create a mock IoHelper with trace spy - const traceSpy = jest.fn(); - const mockIoHelper = { - defaults: { - trace: traceSpy, - }, - }; - - // Mock IoHelper.fromActionAwareIoHost to return our mock - jest.spyOn(IoHelper, 'fromActionAwareIoHost').mockReturnValue(mockIoHelper as any); - - const client = new EndpointTelemetrySink({ endpoint: 'https://example.com/telemetry', ioHost }); - - // Mock https.request to throw an error - (https.request as jest.Mock).mockImplementation(() => { - throw new Error('Network error'); - }); - - await client.emit(testEvent); - - // WHEN & THEN - flush should not throw even when https.request fails - await expect(client.flush()).resolves.not.toThrow(); - - // Verify that the error was logged to trace - expect(traceSpy).toHaveBeenCalledWith( - expect.stringContaining('Telemetry Error: POST example.com/telemetry:'), - ); - }); - - test('skips request when no connectivity detected', async () => { - // GIVEN - (NetworkDetector.hasConnectivity as jest.Mock).mockResolvedValue(false); - - const testEvent = createTestEvent('INVOKE', { foo: 'bar' }); - const client = new EndpointTelemetrySink({ endpoint: 'https://example.com/telemetry', ioHost }); - - // WHEN - await client.emit(testEvent); - await client.flush(); - - // THEN - expect(NetworkDetector.hasConnectivity).toHaveBeenCalledWith(undefined); - expect(https.request).not.toHaveBeenCalled(); - }); -}); diff --git a/packages/aws-cdk/test/cli/telemetry/sink/exit-while-in-flight.driver.ts b/packages/aws-cdk/test/cli/telemetry/sink/exit-while-in-flight.driver.ts new file mode 100644 index 000000000..d0128dcf9 --- /dev/null +++ b/packages/aws-cdk/test/cli/telemetry/sink/exit-while-in-flight.driver.ts @@ -0,0 +1,31 @@ +import { SubprocessTelemetrySink } from '../../../../lib/cli/telemetry/sink/subprocess-sink'; + +/** + * Driver for the "the CLI exits while delivery is still in flight" test. + * + * Not a test itself -- it is spawned as a separate process, because the property under test is about + * process lifetime and cannot be observed from inside the process doing the work. + * + * Uses the real sink against the endpoint given in argv, prints the dispatched child's pid, and then + * returns. If the sink held the process open (a referenced timer, a pipe waiting to drain, an awaited + * request) this would not exit until the child was finished. + */ +async function main(): Promise { + const endpoint = process.argv[2]; + + const sink = new SubprocessTelemetrySink({ + endpoint, + ioHost: { + notify: async (msg: any) => { + // The sink reports the hand-off, including the pid we need to inspect from outside. + process.stdout.write(`${msg.message}\n`); + }, + requestResponse: async (msg: any) => msg.defaultResponse, + } as any, + }); + + await sink.emit({ identifiers: { sessionId: 'exit-while-in-flight' } } as any); + await sink.flush(); +} + +void main(); diff --git a/packages/aws-cdk/test/cli/telemetry/sink/funnel.test.ts b/packages/aws-cdk/test/cli/telemetry/sink/funnel.test.ts index f07c43d62..b1525afac 100644 --- a/packages/aws-cdk/test/cli/telemetry/sink/funnel.test.ts +++ b/packages/aws-cdk/test/cli/telemetry/sink/funnel.test.ts @@ -1,279 +1,122 @@ -import * as https from 'https'; import * as os from 'os'; import * as path from 'path'; import * as fs from 'fs-extra'; import { createTestEvent } from './util'; -import { NetworkDetector } from '../../../../lib/api/network-detector'; -import { IoHelper } from '../../../../lib/api-private'; import { CliIoHost } from '../../../../lib/cli/io-host'; -import { EndpointTelemetrySink } from '../../../../lib/cli/telemetry/sink/endpoint-sink'; import { FileTelemetrySink } from '../../../../lib/cli/telemetry/sink/file-sink'; import { Funnel } from '../../../../lib/cli/telemetry/sink/funnel'; +import type { ITelemetrySink } from '../../../../lib/cli/telemetry/sink/sink-interface'; -// Mock the https module -jest.mock('https', () => ({ - request: jest.fn(), -})); - -// Mock NetworkDetector -jest.mock('../../../../lib/api/network-detector', () => ({ - NetworkDetector: { - hasConnectivity: jest.fn(), - }, -})); - +/** + * A funnel only fans `emit` and `flush` out to the sinks it was given, so real sinks writing to real + * files are what proves it: each one is independently observable, and a sink that was skipped leaves + * an empty file behind. + */ describe('Funnel', () => { let tempDir: string; - let logFilePath: string; let ioHost: CliIoHost; beforeEach(() => { - jest.resetAllMocks(); - - // Mock NetworkDetector to return true by default for all tests - (NetworkDetector.hasConnectivity as jest.Mock).mockResolvedValue(true); - - // Create a fresh temp directory for each test - tempDir = path.join(os.tmpdir(), `telemetry-test-${Date.now()}-${Math.random().toString(36).substring(2, 9)}`); - fs.mkdirSync(tempDir, { recursive: true }); - logFilePath = path.join(tempDir, 'telemetry.json'); - + tempDir = fs.mkdtempSync(path.join(os.tmpdir(), 'telemetry-funnel-')); ioHost = CliIoHost.instance(); }); afterEach(() => { - // Clean up temp directory after each test - if (fs.existsSync(tempDir)) { - fs.rmdirSync(tempDir, { recursive: true }); - } - - // Restore all mocks - jest.restoreAllMocks(); + fs.rmSync(tempDir, { recursive: true, force: true }); }); - // Helper to create a mock request object with the necessary event handlers - function setupMockRequest() { - // Create a mock response object with a successful status code - const mockResponse = { - statusCode: 200, - statusMessage: 'OK', - }; - - // Create the mock request object - const mockRequest = { - on: jest.fn(), - end: jest.fn(), - setTimeout: jest.fn(), + function fileSink(name: string): { sink: FileTelemetrySink; contents: () => any[] } { + const logFilePath = path.join(tempDir, `${name}.json`); + return { + sink: new FileTelemetrySink({ ioHost, logFilePath }), + contents: () => fs.readJSONSync(logFilePath), }; - - // Mock the https.request to return our mockRequest - (https.request as jest.Mock).mockImplementation((_, callback) => { - // If a callback was provided, call it with our mock response - if (callback) { - setTimeout(() => callback(mockResponse), 0); - } - return mockRequest; - }); - - return mockRequest; } - describe('File and Endpoint', () => { - let fileSink: FileTelemetrySink; - let endpointSink: EndpointTelemetrySink; - const traceSpy = jest.fn(); - - beforeEach(() => { - // Create a mock IoHelper with trace spy - const mockIoHelper = { - defaults: { - trace: traceSpy, - }, - }; - - // Mock IoHelper.fromActionAwareIoHost to return our mock - jest.spyOn(IoHelper, 'fromActionAwareIoHost').mockReturnValue(mockIoHelper as any); - - fileSink = new FileTelemetrySink({ ioHost, logFilePath }); - endpointSink = new EndpointTelemetrySink({ ioHost, endpoint: 'https://example.com/telemetry' }); - }); - - test('saves data to a file', async () => { - // GIVEN - const testEvent = createTestEvent('INVOKE', { context: { foo: true } }); - const client = new Funnel({ sinks: [fileSink, endpointSink] }); - - // WHEN - await client.emit(testEvent); - - // THEN - expect(fs.existsSync(logFilePath)).toBe(true); - const fileJson = fs.readJSONSync(logFilePath, 'utf8'); - expect(fileJson).toEqual([testEvent]); - }); - - test('makes a POST request to the specified endpoint', async () => { - // GIVEN - const mockRequest = setupMockRequest(); - const testEvent = createTestEvent('INVOKE', { foo: 'bar' }); - const client = new Funnel({ sinks: [fileSink, endpointSink] }); + test('emit reaches every sink', async () => { + const first = fileSink('first'); + const second = fileSink('second'); + const event = createTestEvent('INVOKE', { context: { foo: true } }); - // WHEN - await client.emit(testEvent); - await client.flush(); + await new Funnel({ sinks: [first.sink, second.sink] }).emit(event); - // THEN - const expectedPayload = JSON.stringify({ events: [testEvent] }); - expect(https.request).toHaveBeenCalledWith({ - hostname: 'example.com', - port: null, - path: '/telemetry', - method: 'POST', - headers: { - 'content-type': 'application/json', - 'content-length': expectedPayload.length, - }, - agent: undefined, - timeout: 500, - }, expect.anything()); - - expect(mockRequest.end).toHaveBeenCalledWith(expectedPayload); - }); - - test('flush is called every 30 seconds on the endpoint sink only', async () => { - // GIVEN - jest.useFakeTimers(); - setupMockRequest(); - - // Spy on the EndpointTelemetrySink prototype flush method BEFORE creating any instances - const flushSpy = jest.spyOn(EndpointTelemetrySink.prototype, 'flush').mockResolvedValue(); - - // Create a fresh endpoint sink for this test - the setInterval will be set up in constructor - const testEndpointSink = new EndpointTelemetrySink({ ioHost, endpoint: 'https://example.com/telemetry' }); - new Funnel({ sinks: [fileSink, testEndpointSink] }); - - // Reset the spy call count since the constructor might have called flush - flushSpy.mockClear(); - - // WHEN & THEN - // Initially no calls from the interval (the setInterval hasn't fired yet) - expect(flushSpy).toHaveBeenCalledTimes(0); - - // Advance the timer by 30 seconds - this should trigger the first interval flush - jest.advanceTimersByTime(30000); + expect(first.contents()).toEqual([event]); + expect(second.contents()).toEqual([event]); + }); - // Verify flush was called once - expect(flushSpy).toHaveBeenCalledTimes(1); + test('every event reaches every sink, in order', async () => { + const first = fileSink('first'); + const second = fileSink('second'); + const funnel = new Funnel({ sinks: [first.sink, second.sink] }); + const one = createTestEvent('INVOKE', { foo: 'one' }); + const two = createTestEvent('SYNTH', { foo: 'two' }); - // Advance the timer by another 30 seconds - this should trigger the second interval flush - jest.advanceTimersByTime(30000); + await funnel.emit(one); + await funnel.emit(two); - // Verify flush was called again (total of 2 times) - expect(flushSpy).toHaveBeenCalledTimes(2); + expect(first.contents()).toEqual([one, two]); + expect(second.contents()).toEqual([one, two]); + }); - // Clean up - flushSpy.mockRestore(); - jest.useRealTimers(); + test('flush reaches every sink', async () => { + const flushed: string[] = []; + const recording = (name: string): ITelemetrySink => ({ + emit: async () => undefined, + flush: async () => { + flushed.push(name); + }, }); - test('failed flush does not clear events cache', async () => { - // GIVEN - const mockRequest = { - on: jest.fn(), - end: jest.fn(), - setTimeout: jest.fn(), - }; - // Mock the https.request to return the first response as 503 - (https.request as jest.Mock).mockImplementationOnce((_, callback) => { - // If a callback was provided, call it with our mock response - if (callback) { - setTimeout(() => callback({ - statusCode: 503, - statusMessage: 'Service Unavailable', - }), 0); - } - return mockRequest; - }).mockImplementation((_, callback) => { - if (callback) { - setTimeout(() => callback({ - statusCode: 200, - statusMessage: 'Success', - }), 0); - } - return mockRequest; - }); + await new Funnel({ sinks: [recording('a'), recording('b'), recording('c')] }).flush(); - const testEvent1 = createTestEvent('INVOKE', { foo: 'bar' }); - const testEvent2 = createTestEvent('INVOKE', { foo: 'bazoo' }); - const client = new Funnel({ sinks: [fileSink, endpointSink] }); - - // WHEN - await client.emit(testEvent1); + expect(flushed.sort()).toEqual(['a', 'b', 'c']); + }); - // mocked to fail - await client.flush(); + test('a single sink is a valid funnel', async () => { + const only = fileSink('only'); + const event = createTestEvent('INVOKE'); - await client.emit(testEvent2); + const funnel = new Funnel({ sinks: [only.sink] }); + await funnel.emit(event); + await funnel.flush(); - // mocked to succeed - await client.flush(); + expect(only.contents()).toEqual([event]); + }); - // THEN - const expectedPayload1 = JSON.stringify({ events: [testEvent1] }); - expect(https.request).toHaveBeenCalledTimes(2); - expect(https.request).toHaveBeenCalledWith({ - hostname: 'example.com', - port: null, - path: '/telemetry', - method: 'POST', - headers: { - 'content-type': 'application/json', - 'content-length': expectedPayload1.length, - }, - agent: undefined, - timeout: 500, - }, expect.anything()); + test('a funnel with no sinks is inert', async () => { + const funnel = new Funnel({ sinks: [] }); - const expectedPayload2 = JSON.stringify({ events: [testEvent1, testEvent2] }); - expect(https.request).toHaveBeenCalledWith({ - hostname: 'example.com', - port: null, - path: '/telemetry', - method: 'POST', - headers: { - 'content-type': 'application/json', - 'content-length': expectedPayload2.length, - }, - agent: undefined, - timeout: 500, - }, expect.anything()); - }); + await expect(funnel.emit(createTestEvent('INVOKE'))).resolves.toBeUndefined(); + await expect(funnel.flush()).resolves.toBeUndefined(); + }); - test('handles errors gracefully and logs to trace without throwing', async () => { - // GIVEN - const testEvent = createTestEvent('INVOKE'); + test('a throwing sink surfaces, but the other sinks still received the event', async () => { + // The funnel does not isolate failures -- it relies on sinks swallowing their own, which both + // real sinks do. This pins the actual behaviour so a future sink that throws is not a surprise. + const healthy = fileSink('healthy'); + const throwing: ITelemetrySink = { + emit: async () => { + throw new Error('sink is down'); + }, + flush: async () => undefined, + }; + const event = createTestEvent('INVOKE'); - const client = new Funnel({ sinks: [fileSink, endpointSink] }); + await expect(new Funnel({ sinks: [throwing, healthy.sink] }).emit(event)).rejects.toThrow('sink is down'); - // Mock https.request to throw an error - (https.request as jest.Mock).mockImplementation(() => { - throw new Error('Network error'); - }); + expect(healthy.contents()).toEqual([event]); + }); - await client.emit(testEvent); + test('throws when too many sinks are added', () => { + const only = fileSink('only').sink; - // WHEN & THEN - flush should not throw even when https.request fails - await client.flush(); + expect(() => new Funnel({ sinks: [only, only, only, only, only, only] })) + .toThrow(/Funnel class supports a maximum of 5 parallel sinks, got 6 sinks./); + }); - // Verify that the error was lt - // logged to trace - expect(traceSpy).toHaveBeenCalledWith( - expect.stringContaining('Telemetry Error: POST example.com/telemetry:'), - ); - }); + test('accepts the maximum number of sinks', () => { + const only = fileSink('only').sink; - test('throws when too many sinks are added', async () => { - expect(() => new Funnel({ sinks: [fileSink, fileSink, fileSink, fileSink, fileSink, fileSink] })).toThrow(/Funnel class supports a maximum of 5 parallel sinks, got 6 sinks./); - }); + expect(() => new Funnel({ sinks: [only, only, only, only, only] })).not.toThrow(); }); }); diff --git a/packages/aws-cdk/test/cli/telemetry/sink/subprocess-sink.test.ts b/packages/aws-cdk/test/cli/telemetry/sink/subprocess-sink.test.ts new file mode 100644 index 000000000..79c46d746 --- /dev/null +++ b/packages/aws-cdk/test/cli/telemetry/sink/subprocess-sink.test.ts @@ -0,0 +1,648 @@ +import { spawn } from 'node:child_process'; +import type * as childProcess from 'node:child_process'; +import * as fs from 'node:fs'; +import * as https from 'node:https'; +import type * as net from 'node:net'; +import { createServer } from 'node:net'; +import * as os from 'node:os'; +import * as path from 'node:path'; +import { cleanupTestCas, generateTestCa, type TestCa } from '../test-tls'; +import { createTestEvent } from './util'; +import { CliIoHost } from '../../../../lib/cli/io-host'; +import { cliRootDir } from '../../../../lib/cli/root-dir'; +import { DISPATCHED_TRACE, SubprocessTelemetrySink } from '../../../../lib/cli/telemetry/sink/subprocess-sink'; + +// The sink hands the payload to a detached child process rather than making the request itself, so +// this is the boundary to observe. Only `spawn` is replaced -- the rest of the module is still needed +// (the TLS helper shells out to openssl), and the child is exercised for real further down. +jest.mock('node:child_process', () => ({ + ...jest.requireActual('node:child_process'), + spawn: jest.fn(), +})); + +const ENDPOINT = 'https://example.com/telemetry'; + +interface FakeChild { + /** + * `undefined` is how libuv reports a spawn it refused: no throw, no pid. + */ + pid: number | undefined; + on: jest.Mock; + unref: jest.Mock; +} + +let child: FakeChild; + +/** + * The handler the sink registered for the child's `error` event. + * + * Node reports a refused spawn there rather than by throwing, so this is the only way to drive that + * path. + */ +function errorHandler(): (e: Error) => void { + const registered = child.on.mock.calls.filter(([event]) => event === 'error'); + expect(registered).toHaveLength(1); + return registered[0][1]; +} + +/** + * The payload path the sink passed to the child on its most recent spawn, whether or not that spawn + * succeeded. Unlike `dispatched()` this does not read the file, so it survives the failure paths. + */ +function spawnedPayloadPath(): string { + const calls = (spawn as jest.Mock).mock.calls; + expect(calls.length).toBeGreaterThan(0); + return calls[calls.length - 1][1][1]; +} + +/** + * The payload file path the sink passed to the child on its most recent dispatch, and the config it + * wrote there. + */ +function dispatched(): { senderPath: string; payloadPath: string; config: any } { + const calls = (spawn as jest.Mock).mock.calls; + expect(calls.length).toBeGreaterThan(0); + const [, args] = calls[calls.length - 1]; + const [senderPath, payloadPath] = args; + return { senderPath, payloadPath, config: JSON.parse(fs.readFileSync(payloadPath, 'utf-8')) }; +} + +describe('SubprocessTelemetrySink', () => { + let ioHost: CliIoHost; + let traces: string[]; + const written: string[] = []; + + beforeAll(() => { + // The sink only dispatches if it can find the compiled entry point next to the package root. + const compiled = path.join(cliRootDir(), 'lib', 'cli', 'telemetry', 'sender-bundle.js'); + if (!fs.existsSync(compiled)) { + throw new Error(`Expected the compiled telemetry sender at ${compiled}. Run \`npx projen compile\` before these tests.`); + } + }); + + beforeEach(() => { + child = { pid: 4242, on: jest.fn(), unref: jest.fn() }; + (spawn as jest.Mock).mockReturnValue(child); + + ioHost = CliIoHost.instance({ logLevel: 'trace' }, true); + traces = []; + jest.spyOn(ioHost, 'notify').mockImplementation(async (msg) => { + traces.push(msg.message); + }); + }); + + afterEach(() => { + for (const file of written.splice(0)) { + fs.rmSync(file, { force: true }); + } + }); + + afterAll(() => cleanupTestCas()); + + function sink(props: Partial[0]> = {}) { + return new SubprocessTelemetrySink({ ioHost, endpoint: ENDPOINT, ...props }); + } + + describe('hand-off', () => { + test('does not spawn anything at construction time', () => { + sink(); + + expect(spawn as jest.Mock).not.toHaveBeenCalled(); + }); + + test('does not spawn when there are no events', async () => { + await sink().flush(); + + expect(spawn as jest.Mock).not.toHaveBeenCalled(); + }); + + test('writes the payload to a file and passes its path to the bundled sender', async () => { + const client = sink(); + await client.emit(createTestEvent('INVOKE')); + await client.flush(); + + const { senderPath, payloadPath, config } = dispatched(); + written.push(payloadPath); + + expect(senderPath).toBe(path.join(cliRootDir(), 'lib', 'cli', 'telemetry', 'sender-bundle.js')); + expect(payloadPath.startsWith(os.tmpdir())).toBe(true); + expect(config.endpoint).toBe(ENDPOINT); + expect(config.body.events).toHaveLength(1); + }); + + test('runs the sender out of this process', async () => { + // Everything else about the spawn -- detached, unref, discarded stdio -- is only meaningful as + // observable behaviour, which `exits while delivery is still in flight` covers for real. + const client = sink(); + await client.emit(createTestEvent('INVOKE')); + await client.flush(); + + const [command] = (spawn as jest.Mock).mock.calls[0]; + written.push((spawn as jest.Mock).mock.calls[0][1][1]); + + expect(command).toBe(process.execPath); + }); + + test('batches multiple events into a single sender', async () => { + const client = sink(); + await client.emit(createTestEvent('INVOKE')); + await client.emit(createTestEvent('INVOKE')); + await client.flush(); + + const { payloadPath, config } = dispatched(); + written.push(payloadPath); + + expect(config.body.events).toHaveLength(2); + }); + + test('a successful hand-off clears the events cache', async () => { + const client = sink(); + await client.emit(createTestEvent('INVOKE')); + await client.flush(); + written.push((spawn as jest.Mock).mock.calls[0][1][1]); + + await client.flush(); + + expect(spawn as jest.Mock).toHaveBeenCalledTimes(1); + }); + + test('reports the hand-off on the trace channel', async () => { + const client = sink(); + await client.emit(createTestEvent('INVOKE')); + await client.flush(); + written.push((spawn as jest.Mock).mock.calls[0][1][1]); + + expect(traces.some((t) => t.includes(DISPATCHED_TRACE) && t.includes('pid 4242'))).toBe(true); + }); + + test('dispatches without first probing the network', async () => { + // Any connectivity check would itself be a network call on the CLI's exit path, which is the + // thing this sink exists to avoid. + const client = sink(); + await client.emit(createTestEvent('INVOKE')); + await client.flush(); + written.push((spawn as jest.Mock).mock.calls[0][1][1]); + + expect(traces.some((t) => t.includes('connectivity'))).toBe(false); + }); + }); + + describe('network configuration', () => { + test('forwards the CA bundle PATH, never its contents', async () => { + // The invariant: what crosses the process boundary is a path, so the payload's size is + // independent of the CA bundle's. A real system bundle is ~190KB, and inlining it would make + // every batch carry that -- for a value the child can read off disk itself. + const ca = generateTestCa(); + const client = sink({ caBundlePath: ca.caCertPath, proxyUrl: 'http://corp:8080' }); + await client.emit(createTestEvent('INVOKE')); + await client.flush(); + + const { payloadPath, config } = dispatched(); + written.push(payloadPath); + + expect(config.caBundlePath).toBe(ca.caCertPath); + expect(config.proxyUrl).toBe('http://corp:8080'); + + const raw = fs.readFileSync(payloadPath, 'utf-8'); + expect(raw).not.toContain('BEGIN CERTIFICATE'); + expect(raw.length).toBeLessThan(4096); + }); + + test('omits proxy and CA settings when none were configured', async () => { + const client = sink(); + await client.emit(createTestEvent('INVOKE')); + await client.flush(); + + const { payloadPath, config } = dispatched(); + written.push(payloadPath); + + expect(config.proxyUrl).toBeUndefined(); + expect(config.caBundlePath).toBeUndefined(); + }); + + test('an explicitly empty proxy crosses the process boundary as an empty string, not as unset', async () => { + // `--proxy ''` means "go direct, ignore the proxy environment variables". The child inherits + // that environment, so if the empty string were collapsed to unset on the way out the child + // would auto-detect a proxy the parent had been told not to use. + const client = sink({ proxyUrl: '' }); + await client.emit(createTestEvent('INVOKE')); + await client.flush(); + + const { payloadPath, config } = dispatched(); + written.push(payloadPath); + + expect(config.proxyUrl).toBe(''); + expect(Object.keys(config)).toContain('proxyUrl'); + }); + }); + + describe('payload size', () => { + test('hands over a large batch whole, with no size ceiling', async () => { + // The hand-off goes through a file precisely so that batch size is not bounded: a pipe would + // block our own exit once the payload outgrew the OS buffer, which is the wait this sink + // exists to avoid. 64KB is a typical pipe buffer, so exceeding it is the meaningful threshold. + const client = sink(); + for (let i = 0; i < 400; i++) { + await client.emit(createTestEvent('INVOKE')); + } + await client.flush(); + + const { payloadPath, config } = dispatched(); + written.push(payloadPath); + + expect(fs.statSync(payloadPath).size).toBeGreaterThan(65_536); + expect(config.body.events).toHaveLength(400); + expect(traces.some((t) => t.includes('dropped'))).toBe(false); + }); + }); + + describe('failure handling', () => { + test('a refused spawn is reported as a failure, not as a dispatch', async () => { + // Node does NOT throw when it refuses a spawn (ENOENT, EACCES, EMFILE); it reports on the + // child's `error` event, which fires after the hand-off has already returned. What it does do + // synchronously is leave `pid` unset. Without a check for that, this path traced a successful + // dispatch with `pid undefined` and the batch was silently counted as sent. + child.pid = undefined; + + const client = sink(); + await client.emit(createTestEvent('INVOKE')); + await client.emit(createTestEvent('INVOKE')); + await expect(client.flush()).resolves.toBeUndefined(); + + expect(traces.some((t) => t.includes(DISPATCHED_TRACE))).toBe(false); + expect(traces.some((t) => t.includes('Dropped 2 event(s)'))).toBe(true); + expect(fs.existsSync(spawnedPayloadPath())).toBe(false); + }); + + test('a refused spawn does not retain the batch', async () => { + // Delivery is one-shot: the process that would retry has usually exited by now, so retaining + // the batch would only re-report the same failure and regrow it on the next interval. + child.pid = undefined; + + const client = sink(); + await client.emit(createTestEvent('INVOKE')); + await expect(client.flush()).resolves.toBeUndefined(); + expect(traces.filter((t) => t.includes('Dropped'))).toHaveLength(1); + + child.pid = 4242; + await client.flush(); + + expect(spawn as jest.Mock).toHaveBeenCalledTimes(1); + }); + + test("the child's 'error' handler removes the payload file", async () => { + // The residual case: the spawn was accepted synchronously but failed afterwards, by which time + // the CLI may have exited. Nothing else runs, so if this handler does not clean up, every such + // failure leaks a payload file into the temp directory. + const client = sink(); + await client.emit(createTestEvent('INVOKE')); + await client.flush(); + + const payloadPath = spawnedPayloadPath(); + expect(fs.existsSync(payloadPath)).toBe(true); + + errorHandler()(new Error('EACCES: permission denied')); + + expect(fs.existsSync(payloadPath)).toBe(false); + }); + + test('a synchronous throw from spawn is handled too', async () => { + // Defensive: the realistic refusals are asynchronous (see above), but argument validation can + // still throw here, and it must not escape onto the CLI's exit path. + (spawn as jest.Mock).mockImplementation(() => { + throw new Error('EINVAL: invalid argument'); + }); + + const client = sink(); + await client.emit(createTestEvent('INVOKE')); + await expect(client.flush()).resolves.toBeUndefined(); + + expect(traces.filter((t) => t.includes('EINVAL'))).toHaveLength(1); + expect(traces.some((t) => t.includes('Dropped 1 event(s)'))).toBe(true); + expect(fs.existsSync(spawnedPayloadPath())).toBe(false); + }); + + test('logs once and drops the batch when the sender cannot be located', async () => { + const client = sink({ resolveSender: () => undefined }); + await client.emit(createTestEvent('INVOKE')); + + await expect(client.flush()).resolves.toBeUndefined(); + + expect(traces.filter((t) => t.includes('Unable to locate the telemetry sender'))).toHaveLength(1); + expect(traces.some((t) => t.includes('Dropped 1 event(s)'))).toBe(true); + expect(spawn as jest.Mock).not.toHaveBeenCalled(); + + // Not retained: this never starts working mid-process, so retrying every 30s is pure noise. + await client.flush(); + expect(traces.filter((t) => t.includes('Unable to locate the telemetry sender'))).toHaveLength(1); + }); + + test('rejects an endpoint with no host at construction', () => { + expect(() => sink({ endpoint: 'file:///metrics' })).toThrow(/Telemetry Endpoint malformed/); + }); + + test('rejects an unparseable endpoint at construction', () => { + expect(() => sink({ endpoint: 'not-a-url' })).toThrow(/Invalid URL/); + }); + }); + + describe('debug channel', () => { + test('passes the child stderr through when CDK_TELEMETRY_SENDER_DEBUG=1', async () => { + // Otherwise the sender's own traces go to a discarded fd and the one field-debug tool is + // unusable. + process.env.CDK_TELEMETRY_SENDER_DEBUG = '1'; + try { + const client = sink(); + await client.emit(createTestEvent('INVOKE')); + await client.flush(); + + const [, , options] = (spawn as jest.Mock).mock.calls[0]; + written.push((spawn as jest.Mock).mock.calls[0][1][1]); + + expect(options.stdio).toEqual(['ignore', 'ignore', 'inherit']); + } finally { + delete process.env.CDK_TELEMETRY_SENDER_DEBUG; + } + }); + + test('discards the child stdio by default', async () => { + const client = sink(); + await client.emit(createTestEvent('INVOKE')); + await client.flush(); + + const [, , options] = (spawn as jest.Mock).mock.calls[0]; + written.push((spawn as jest.Mock).mock.calls[0][1][1]); + + expect(options.stdio).toBe('ignore'); + }); + }); +}); + +/** + * End-to-end coverage of the entry point itself. + * + * The tests above stop at the process boundary. These run the real `sender-bundle` in a real child + * process against a real HTTPS server, which is the only way to know that the file hand-off, + * cleanup and delivery actually work together. + */ +describe('sender-bundle entry point', () => { + let ca: TestCa; + let cdkHome: string; + + beforeAll(() => { + ca = generateTestCa(); + }); + + beforeEach(() => { + // The child inherits CDK_HOME; point it somewhere disposable so nothing touches the developer's + // real cache. + cdkHome = fs.mkdtempSync(path.join(os.tmpdir(), 'cdk-home-')); + }); + + afterEach(() => { + fs.rmSync(cdkHome, { recursive: true, force: true }); + }); + + afterAll(() => cleanupTestCas()); + + async function startEndpoint(options: { statusCode?: number } = {}): Promise<{ url: string; received: string[]; close(): Promise }> { + const received: string[] = []; + const sockets: Array<{ destroy(): void }> = []; + const server = https.createServer({ key: ca.serverKey, cert: ca.serverCert }, (req, res) => { + let body = ''; + req.on('data', (c) => (body += c)); + req.on('end', () => { + received.push(body); + res.writeHead(options.statusCode ?? 200, { 'content-type': 'application/json' }); + res.end('{"ok":true}'); + }); + }); + server.on('connection', (socket) => sockets.push(socket)); + await new Promise((ok) => server.listen(0, '127.0.0.1', ok)); + const port = (server.address() as net.AddressInfo).port; + return { + // Connect by IP, matching the bind address. Nothing here asserts on the hostname, and resolving + // `localhost` to ::1 first -- which Node 18+ does on a dual-stack box -- would ECONNREFUSED + // against a listener bound only to 127.0.0.1. Covered by the certificate's `IP:127.0.0.1` SAN. + url: `https://127.0.0.1:${port}/metrics`, + received, + close: () => new Promise((ok) => { + for (const socket of sockets) { + socket.destroy(); + } + server.close(() => ok()); + }), + }; + } + + /** + * Run the entry point from source, so this does not depend on a prior build. + */ + function runSender(payloadPath: string): Promise { + const { spawn: realSpawn } = jest.requireActual('node:child_process') as typeof childProcess; + const tsx = path.join(path.dirname(require.resolve('tsx/package.json')), 'dist', 'cli.mjs'); + const entryPoint = path.join(cliRootDir(), 'lib', 'cli', 'telemetry', 'sender-bundle.ts'); + + return new Promise((ok, ko) => { + const proc = realSpawn(process.execPath, [tsx, entryPoint, payloadPath], { + stdio: 'ignore', + env: { ...process.env, CDK_HOME: cdkHome }, + }); + proc.on('error', ko); + proc.on('exit', (code) => ok(code)); + }); + } + + function writePayload(config: unknown): string { + const payloadPath = path.join(os.tmpdir(), `cdk-telemetry-e2e-${Date.now()}-${Math.random().toString(36).slice(2)}.json`); + fs.writeFileSync(payloadPath, JSON.stringify(config)); + return payloadPath; + } + + test('reads the payload file, delivers it, deletes the file, and exits 0', async () => { + const endpoint = await startEndpoint(); + const body = { events: [{ identifiers: { sessionId: 'e2e-session' } }] }; + const payloadPath = writePayload({ + endpoint: endpoint.url, + body, + caBundlePath: ca.caCertPath, + timeoutMs: 10_000, + }); + + try { + const exitCode = await runSender(payloadPath); + + expect(exitCode).toBe(0); + expect(endpoint.received).toHaveLength(1); + expect(JSON.parse(endpoint.received[0])).toEqual(body); + // The child owns the file; nothing else would ever collect it. + expect(fs.existsSync(payloadPath)).toBe(false); + } finally { + fs.rmSync(payloadPath, { force: true }); + await endpoint.close(); + } + }, 60_000); + + test('reads the CA bundle from the path it was given', async () => { + // Proves the path really is enough: the endpoint's certificate is not publicly trusted, so + // delivery only succeeds if the child loaded the bundle off disk itself. + const endpoint = await startEndpoint(); + const withCa = writePayload({ endpoint: endpoint.url, body: { events: [{ n: 1 }] }, caBundlePath: ca.caCertPath, timeoutMs: 10_000 }); + const withoutCa = writePayload({ endpoint: endpoint.url, body: { events: [{ n: 2 }] }, timeoutMs: 10_000 }); + + try { + await expect(runSender(withoutCa)).resolves.toBe(0); + expect(endpoint.received).toHaveLength(0); + + await expect(runSender(withCa)).resolves.toBe(0); + expect(endpoint.received).toHaveLength(1); + } finally { + fs.rmSync(withCa, { force: true }); + fs.rmSync(withoutCa, { force: true }); + await endpoint.close(); + } + }, 60_000); + + test('exits cleanly and removes the file when the payload is unusable', async () => { + const payloadPath = path.join(os.tmpdir(), `cdk-telemetry-e2e-bad-${Date.now()}.json`); + fs.writeFileSync(payloadPath, 'not json at all'); + + try { + await expect(runSender(payloadPath)).resolves.toBe(0); + expect(fs.existsSync(payloadPath)).toBe(false); + } finally { + fs.rmSync(payloadPath, { force: true }); + } + }, 60_000); + + test('exits cleanly when the payload file is missing entirely', async () => { + const missing = path.join(os.tmpdir(), `cdk-telemetry-e2e-missing-${Date.now()}.json`); + + await expect(runSender(missing)).resolves.toBe(0); + }, 60_000); + + describe('a failed delivery is not a failed CLI', () => { + // Nobody waits on this process, but its exit status is still visible to anything watching the + // process tree, and the payload file is nobody else's to collect. Both must hold on the failure + // paths too, or a rejected send starts looking like a crash and leaks a file per invocation. + test('a non-2xx response still exits 0 and removes the payload file', async () => { + const endpoint = await startEndpoint({ statusCode: 500 }); + const payloadPath = writePayload({ endpoint: endpoint.url, body: { events: [{ n: 1 }] }, caBundlePath: ca.caCertPath, timeoutMs: 10_000 }); + + try { + await expect(runSender(payloadPath)).resolves.toBe(0); + expect(endpoint.received).toHaveLength(1); + expect(fs.existsSync(payloadPath)).toBe(false); + } finally { + fs.rmSync(payloadPath, { force: true }); + await endpoint.close(); + } + }, 60_000); + + test('a transport failure still exits 0 and removes the payload file', async () => { + // Port 1 is reserved and nothing listens on it. + const payloadPath = writePayload({ endpoint: 'https://127.0.0.1:1/metrics', body: { events: [{ n: 1 }] }, timeoutMs: 5000 }); + + try { + await expect(runSender(payloadPath)).resolves.toBe(0); + expect(fs.existsSync(payloadPath)).toBe(false); + } finally { + fs.rmSync(payloadPath, { force: true }); + } + }, 60_000); + }); + + test('delivers with a CA bundle much larger than the payload itself', async () => { + // The invariant that lets this work: the payload carries the bundle's PATH, so the child reads a + // ~190KB system bundle (a concatenation of a few hundred certificates) off disk itself and the + // payload stays small. Inlining the certificate would tie every batch's size to the CA bundle's. + const endpoint = await startEndpoint(); + const bundlePath = path.join(cdkHome, 'big-bundle.pem'); + + const single = fs.readFileSync(ca.caCertPath, 'utf-8'); + let bundle = ''; + while (Buffer.byteLength(bundle) < 128 * 1024) { + bundle += single; + } + fs.writeFileSync(bundlePath, bundle); + expect(fs.statSync(bundlePath).size).toBeGreaterThan(65_536); + + const payloadPath = writePayload({ + endpoint: endpoint.url, + body: { events: [{ identifiers: { sessionId: 'big-bundle' } }] }, + caBundlePath: bundlePath, + timeoutMs: 10_000, + }); + + // The invariant itself: a 128KB bundle leaves the payload tiny, because only the path travels. + const payloadSize = fs.statSync(payloadPath).size; + expect(payloadSize).toBeLessThan(4096); + expect(fs.readFileSync(payloadPath, 'utf-8')).not.toContain('BEGIN CERTIFICATE'); + + try { + await expect(runSender(payloadPath)).resolves.toBe(0); + + expect(endpoint.received).toHaveLength(1); + expect(JSON.parse(endpoint.received[0])).toEqual({ events: [{ identifiers: { sessionId: 'big-bundle' } }] }); + } finally { + fs.rmSync(payloadPath, { force: true }); + await endpoint.close(); + } + }, 60_000); +}); + +/** + * The property the whole change exists for: the CLI is gone before delivery finishes. + * + * Cannot be observed from inside the process doing the work, so this runs a driver in a child + * process, points it at an endpoint that accepts the connection and then never answers, and checks + * that the driver exited while the sender it spawned was still running. + */ +describe('exits while delivery is still in flight', () => { + test('the sink does not hold the process open until delivery finishes', async () => { + const { spawn: realSpawn } = jest.requireActual('node:child_process') as typeof childProcess; + + // Accepts the TCP connection and then never writes a byte, so the sender hangs on it until its + // own timeout. That is the window in which the driver has to have exited. + const held: Array<{ destroy(): void }> = []; + const blackHole = createServer((socket) => held.push(socket)); + await new Promise((ok) => blackHole.listen(0, '127.0.0.1', ok)); + const port = (blackHole.address() as net.AddressInfo).port; + + const tsx = path.join(path.dirname(require.resolve('tsx/package.json')), 'dist', 'cli.mjs'); + const driver = path.join(cliRootDir(), 'test', 'cli', 'telemetry', 'sink', 'exit-while-in-flight.driver.ts'); + + let senderPid: number | undefined; + try { + const output = await new Promise((ok, ko) => { + const proc = realSpawn(process.execPath, [tsx, driver, `https://127.0.0.1:${port}/metrics`], { + stdio: ['ignore', 'pipe', 'ignore'], + }); + let stdout = ''; + proc.stdout!.on('data', (chunk) => (stdout += chunk)); + proc.on('error', ko); + // Resolves only once the driver has actually exited. + proc.on('exit', () => ok(stdout)); + }); + + senderPid = Number(output.match(/pid (\d+)/)?.[1]); + expect(senderPid).toBeGreaterThan(0); + + // The driver has exited. If the sender is still alive, delivery was still in flight when it + // did -- which is the whole point of detaching it. + expect(() => process.kill(senderPid!, 0)).not.toThrow(); + } finally { + if (senderPid) { + try { + process.kill(senderPid, 'SIGKILL'); + } catch { + // Already gone. + } + } + for (const socket of held) { + socket.destroy(); + } + await new Promise((ok) => blackHole.close(() => ok())); + } + }, 60_000); +}); diff --git a/packages/aws-cdk/test/cli/telemetry/test-tls.ts b/packages/aws-cdk/test/cli/telemetry/test-tls.ts new file mode 100644 index 000000000..e5f7dfd73 --- /dev/null +++ b/packages/aws-cdk/test/cli/telemetry/test-tls.ts @@ -0,0 +1,123 @@ +import { execFileSync } from 'node:child_process'; +import * as fs from 'node:fs'; +import * as os from 'node:os'; +import * as path from 'node:path'; + +/** + * A throwaway certificate authority plus a leaf certificate for `localhost`. + */ +export interface TestCa { + /** + * PEM contents of the CA certificate, to be passed to the code under test as a trusted root. + */ + readonly caCert: string; + + /** + * Path to the CA certificate on disk. + * + * The sender is configured with a bundle PATH rather than its contents, so this is what most + * tests actually need. Removed by `cleanupTestCas()`. + */ + readonly caCertPath: string; + + /** + * PEM contents of the leaf certificate, for the test server. + */ + readonly serverCert: string; + + /** + * PEM contents of the leaf private key, for the test server. + */ + readonly serverKey: string; +} + +/** + * Options for `generateTestCa`. + */ +export interface TestCaOptions { + /** + * OpenSSL `subjectAltName` value for the leaf certificate. + * + * Override this to mint a certificate that deliberately does NOT cover the host under test, which + * is how the identity-verification tests prove a mismatch is rejected. Note that modern TLS + * ignores the subject CN entirely, so the SAN is the only thing that matters. + * + * @default 'DNS:localhost,IP:127.0.0.1' + */ + readonly subjectAltName?: string; + + /** + * Subject common name for the leaf certificate. + * + * @default 'localhost' + */ + readonly commonName?: string; +} + +/** + * Directories created by `generateTestCa`, so they can all be removed at the end of a suite. + */ +const generatedDirs: string[] = []; + +/** + * Remove every directory created by `generateTestCa`. Call from `afterAll`. + */ +export function cleanupTestCas(): void { + while (generatedDirs.length > 0) { + fs.rmSync(generatedDirs.pop()!, { recursive: true, force: true }); + } +} + +/** + * Mint a fresh CA and leaf certificate for use by a test HTTPS server. + * + * Generated at runtime rather than committed as a fixture: this repository ships no key material, + * and a checked-in private key would be both a bad precedent and something that expires. This is + * the same approach the integration tests take (`mockttp.generateCACertificate`), minus the + * dependency. + * + * The generated files stay on disk -- the code under test is given a bundle path, not its contents -- + * until `cleanupTestCas()` removes them. + * + * Requires `openssl` on PATH, which is present on every platform this package is tested on. + */ +export function generateTestCa(options: TestCaOptions = {}): TestCa { + const subjectAltName = options.subjectAltName ?? 'DNS:localhost,IP:127.0.0.1'; + const commonName = options.commonName ?? 'localhost'; + + // The jest setup chdir's into a deliberately read-only directory, so be explicit about where we + // write. + const dir = fs.mkdtempSync(path.join(os.tmpdir(), 'cdk-telemetry-tls-')); + generatedDirs.push(dir); + + const file = (name: string) => path.join(dir, name); + const openssl = (...args: string[]) => execFileSync('openssl', args, { cwd: dir, stdio: 'pipe' }); + + openssl('req', '-x509', '-newkey', 'rsa:2048', '-sha256', '-days', '3650', '-nodes', + '-keyout', file('ca.key'), '-out', file('ca.crt'), + '-subj', '/CN=CDK Telemetry Test Root CA', + '-addext', 'basicConstraints=critical,CA:TRUE'); + + openssl('req', '-newkey', 'rsa:2048', '-nodes', + '-keyout', file('server.key'), '-out', file('server.csr'), + '-subj', `/CN=${commonName}`); + + fs.writeFileSync(file('server.ext'), [ + `subjectAltName=${subjectAltName}`, + 'basicConstraints=CA:FALSE', + 'extendedKeyUsage=serverAuth', + '', + ].join('\n')); + + openssl('x509', '-req', '-in', file('server.csr'), + '-CA', file('ca.crt'), '-CAkey', file('ca.key'), '-CAcreateserial', + '-out', file('server.crt'), '-days', '3650', '-sha256', + '-extfile', file('server.ext')); + + return { + caCert: fs.readFileSync(file('ca.crt'), 'utf-8'), + caCertPath: file('ca.crt'), + serverCert: fs.readFileSync(file('server.crt'), 'utf-8'), + serverKey: fs.readFileSync(file('server.key'), 'utf-8'), + }; +}