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..84be68557 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,8 +5,8 @@ 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'); 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-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..9fdafffe5 --- /dev/null +++ b/packages/@aws-cdk-testing/cli-integ/tests/telemetry-integ-tests/cdk-telemetry-does-not-block-exit.integtest.ts @@ -0,0 +1,55 @@ +import * as net from 'node:net'; +import type { AddressInfo } from 'node:net'; +import { integTest, withDefaultFixture } from '../../lib'; + +/** + * Telemetry is delivered by a detached child process, so the CLI must not wait for the POST. + * + * The endpoint here is a black hole: a TCP listener that accepts the connection and then never + * writes a byte, so anything talking to it hangs until its own timeout. Before the sender was + * detached, the flush at the end of the invocation blocked on exactly that, which is why this + * asserts on wall-clock time rather than on output. + */ +integTest( + 'cdk synth does not wait for the telemetry endpoint', + withDefaultFixture(async (fixture) => { + const sockets: net.Socket[] = []; + const blackHole = net.createServer((socket) => { + // Accept and hold. Never respond, never close. + sockets.push(socket); + }); + await new Promise((ok) => blackHole.listen(0, '127.0.0.1', ok)); + const port = (blackHole.address() as AddressInfo).port; + + try { + // Baseline: the same synth with telemetry switched off entirely. + const disabledStart = Date.now(); + await fixture.cdkSynth({ + options: [fixture.fullStackName('test-1')], + modEnv: { CDK_DISABLE_CLI_TELEMETRY: 'true' }, + }); + const disabledMs = Date.now() - disabledStart; + + // The same synth, with telemetry pointed at the black hole. + const blackHoleStart = Date.now(); + await fixture.cdkSynth({ + options: [fixture.fullStackName('test-1')], + modEnv: { TELEMETRY_ENDPOINT: `https://127.0.0.1:${port}/metrics` }, + }); + const blackHoleMs = Date.now() - blackHoleStart; + + const overhead = blackHoleMs - disabledMs; + fixture.log(`synth with telemetry disabled: ${disabledMs}ms, pointed at a black hole: ${blackHoleMs}ms (overhead ${overhead}ms)`); + + // The detached sender is what hangs on the black hole, not us. The headroom is generous + // because CI machines are noisy; what this rules out is the CLI blocking on the request + // timeout, which shows up as whole seconds. + expect(overhead).toBeLessThan(2000); + } finally { + for (const socket of sockets) { + socket.destroy(); + } + await new Promise((ok) => blackHole.close(() => ok())); + } + }), +); 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..95445cb2a --- /dev/null +++ b/packages/@aws-cdk-testing/cli-integ/tests/telemetry-integ-tests/cdk-telemetry-goes-through-a-proxy.integtest.ts @@ -0,0 +1,95 @@ +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'; + +/** + * Telemetry has to keep working for users behind a corporate proxy. + * + * This matters more than it looks. The POST is made by a detached child process that has no access + * to the parent's `proxy-agent` instance -- it only has Node built-ins -- so it re-implements HTTP + * CONNECT tunnelling and has to be handed the proxy URL and CA bundle explicitly. This test 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 not in any system trust store. + * + * `TELEMETRY_ENDPOINT` is pointed at a local server rather than the real one, 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() }), + })); + } finally { + await proxyServer.stop(); + await new Promise((ok) => endpointServer.close(() => ok())); + } + }), +); + +/** + * Poll `fn` until it returns something truthy, or give up after `timeoutMs`. + */ +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 new Promise((ok) => setTimeout(ok, 500)); + } + return undefined; +} diff --git a/packages/aws-cdk/bin/cdk b/packages/aws-cdk/bin/cdk index be493e3f8..181538f10 100755 --- a/packages/aws-cdk/bin/cdk +++ b/packages/aws-cdk/bin/cdk @@ -1,4 +1,18 @@ #!/usr/bin/env node +// Publish our own location so the CLI can respawn us as a detached telemetry sender. +// This is the only place that knows it reliably; process.argv[1] may be a .bin symlink, +// the `cdk` alias package's wrapper, or an embedding script. +process.env.CDK_CLI_BIN_PATH = __filename; + +// That detached sender is this same script with a flag. Dispatch before requiring the CLI, +// whose bundle costs ~600ms to load and which the sender does not need. +if (process.env.CDK_TELEMETRY_SENDER === '1') { + require("../lib/cli/telemetry/sender").main(); + // Relies on the CommonJS module wrapper (modules are functions); would be a SyntaxError if this + // file ever became native ESM. + return; +} + // source maps must be enabled before importing files process.setSourceMapsEnabled(true); const { cli } = require("../lib"); diff --git a/packages/aws-cdk/lib/cli/cli.ts b/packages/aws-cdk/lib/cli/cli.ts index 84bb10275..b896c4f78 100644 --- a/packages/aws-cdk/lib/cli/cli.ts +++ b/packages/aws-cdk/lib/cli/cli.ts @@ -116,13 +116,14 @@ export async function exec(args: string[], synthesizer?: Synthesizer): Promise { // 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!) : undefined; - return new ProxyAgent({ - ca: await this.tryGetCACert(options.caBundlePath), - getProxyForUrl, - }); + const caCert = await this.tryGetCACert(options.caBundlePath); + + return { + agent: new ProxyAgent({ + ca: caCert, + getProxyForUrl, + }), + caCert, + }; } private async tryGetCACert(bundlePath?: string) { diff --git a/packages/aws-cdk/lib/cli/telemetry/cli-bin-path.ts b/packages/aws-cdk/lib/cli/telemetry/cli-bin-path.ts new file mode 100644 index 000000000..e53d54c45 --- /dev/null +++ b/packages/aws-cdk/lib/cli/telemetry/cli-bin-path.ts @@ -0,0 +1,46 @@ +import * as fs from 'node:fs'; +import * as path from 'node:path'; +import { cliRootDir } from '../root-dir'; + +/** + * Environment variable through which `bin/cdk` records its own location. + * + * `bin/cdk` is the only place that knows this reliably, so it publishes `__filename` here. + */ +export const CLI_BIN_PATH_ENV = 'CDK_CLI_BIN_PATH'; + +/** + * Locate this CLI's `bin/cdk` script, so that we can respawn ourselves as a telemetry sender. + * + * `process.argv[1]` is deliberately NOT used. Depending on how the CLI was started it points at + * something else entirely: + * + * - installed normally, it is the `node_modules/.bin/cdk` symlink; + * - installed via the `cdk` alias package, it resolves to that package's wrapper, not ours; + * - used programmatically (`require('aws-cdk').cli()`), it is the caller's own script -- respawning + * it would re-run somebody else's program. + * + * So we prefer the path `bin/cdk` published about itself, and fall back to walking up from this + * module to the package root (which works both from `lib/` in source and from the bundle). + * + * Returns undefined if no candidate exists on disk, in which case telemetry is skipped. + */ +export function cliBinPath(env: NodeJS.ProcessEnv = process.env): string | undefined { + const candidates = [ + env[CLI_BIN_PATH_ENV], + packageRelativeBinPath(), + ]; + + for (const candidate of candidates) { + if (candidate && fs.existsSync(candidate)) { + return candidate; + } + } + + return undefined; +} + +function packageRelativeBinPath(): string | undefined { + const root = cliRootDir(false); + return root ? path.join(root, 'bin', 'cdk') : undefined; +} 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..2df7a6c99 --- /dev/null +++ b/packages/aws-cdk/lib/cli/telemetry/sender.ts @@ -0,0 +1,692 @@ +import * as fs from 'node:fs'; +import * as https from 'node:https'; +import * as net from 'node:net'; +import * as tls from 'node:tls'; + +/** + * The detached telemetry sender. + * + * This module is executed in a short-lived, detached child process (see `bin/cdk`, which + * dispatches here when `CDK_TELEMETRY_SENDER=1`). Its only job is to POST a telemetry payload + * that it receives on stdin, and then exit. + * + * IMPORTANT: this file must only import Node built-ins. + * + * The published `aws-cdk` package has *zero* runtime dependencies -- everything is inlined into + * the `lib/index.js` esbuild bundle, and every entry in `dependencies` is rewritten to + * `devDependencies` at pack time. The individually compiled `lib/**\/*.js` files are still shipped, + * but any of them that reaches for an external module (or for a relative module that transitively + * does) will fail with `Cannot find module` when required. Since `bin/cdk` requires this file + * directly -- deliberately *not* going through the bundle, whose load costs ~600ms -- it has to + * stand on its own. + * + * For the same reason this module never throws: `ToolkitError` lives in `@aws-cdk/toolkit-lib`, + * which is not reachable from here. Every failure is swallowed and reported through the return + * value instead. Telemetry must never be able to affect the CLI or leave a lingering process. + */ + +/** + * Budget for each individual network step. + * + * Emphatically NOT the parent's old `REQUEST_ATTEMPT_TIMEOUT_MS` of 500ms. That number existed to + * stop a synchronous POST from holding up the user's prompt; now that the send happens in a + * detached process that nobody waits on, a tight budget buys the user nothing and costs us + * telemetry. It was also applied to *each* step of a proxied send -- connect + CONNECT, then the + * TLS handshake to the endpoint, then the response -- so proxied users had to complete two TLS + * handshakes inside 500ms each and were silently dropped when they could not. On a loaded CI runner + * that is exactly what happened. + * + * The three steps are sequential, so the worst case is 3x this value; keep that comfortably under + * `HARD_KILL_MS` so the ceiling stays a backstop against a genuinely stuck socket rather than + * something that can fire during a slow-but-progressing handshake. 3s per step also matches what + * the rest of the CLI already considers a reasonable background network budget (`NetworkDetector` + * uses 3s in production). + */ +const NETWORK_TIMEOUT_MS = 3_000; + +/** + * Upper bound on the lifetime of this process. + * + * A hung read on stdin, or a TCP connection that neither completes nor errors, would otherwise + * keep a detached process alive indefinitely after the CLI has exited. The timer is `unref`ed so + * it never keeps the process alive by itself, but it still fires if something else does. + * + * Must exceed the worst-case send (3 x `NETWORK_TIMEOUT_MS`) plus reading stdin, which has no + * timeout of its own. Nobody waits on this process -- its stdio is discarded and it is `unref`ed -- + * so a generous ceiling costs the user nothing. + */ +const HARD_KILL_MS = 20_000; + +/** + * Refuse to buffer an unreasonable amount of stdin. + * + * The parent applies its own (much smaller) limit; this is only a backstop. + */ +const MAX_STDIN_BYTES = 1_048_576; + +/** + * Give up if a proxy sends a pathologically large CONNECT response. + */ +const MAX_PROXY_RESPONSE_BYTES = 16_384; + +/** + * Proxy schemes we can tunnel through using only Node built-ins. + * + * `proxy-agent` (used by the CLI itself) additionally supports `socks*` and `pac+*`. Those + * require a real SOCKS implementation and a PAC interpreter respectively, neither of which is + * available here. When we see one we skip the send entirely rather than falling back to a direct + * connection: a proxy is usually mandatory rather than advisory (corporate setups routinely + * firewall direct egress), so bypassing it would be both futile and a policy violation. + */ +const SUPPORTED_PROXY_PROTOCOLS = ['http:', 'https:']; + +/** + * Default ports per scheme, matching `proxy-from-env@1`'s table. + * + * Used when matching `NO_PROXY` entries that carry an explicit port. + */ +const DEFAULT_PORTS: Record = { + ftp: 21, + gopher: 70, + http: 80, + https: 443, + ws: 80, + wss: 443, +}; + +/** + * What the parent process pipes to this process on stdin. + */ +export interface TelemetrySenderConfig { + /** + * Absolute URL to POST the telemetry payload to. + */ + readonly endpoint: string; + + /** + * The telemetry payload. Serialized as-is into the request body. + */ + readonly body: unknown; + + /** + * Proxy to tunnel through, if the user configured one explicitly. + * + * @default - resolved from the inherited proxy environment variables + */ + readonly proxyUrl?: string; + + /** + * Contents (not path) of a CA bundle to trust. + * + * Note that Node's `ca` option REPLACES the default trust set rather than adding to it, so when + * this is present the bundled roots are no longer consulted. That is what we want for a + * TLS-terminating corporate proxy, and it matches what the parent does with the same bytes. + * + * @default - the default Node trust store, plus anything in `NODE_EXTRA_CA_CERTS` + */ + readonly ca?: string; + + /** + * Overrides the inherited `NO_PROXY` environment variable. + * + * @default - the inherited `NO_PROXY`/`no_proxy` + */ + readonly noProxy?: string; + + /** + * Budget for each network step, in milliseconds. + * + * @default 3000 + */ + readonly timeoutMs?: number; +} + +/** + * Outcome of a send attempt. Purely informational -- nothing acts on it except tests and traces. + */ +export interface SendResult { + /** + * Whether the endpoint accepted the payload with a 2xx response. + */ + readonly sent: boolean; + + /** + * How the request was routed, or `skipped` if we never went on the network. + */ + readonly via: 'direct' | 'connect-tunnel' | 'skipped'; + + /** + * HTTP status code, if we got a response at all. + * + * @default - no response was received + */ + readonly statusCode?: number; + + /** + * Why the send did not succeed. + * + * @default - the send succeeded + */ + readonly reason?: string; +} + +/** + * Entry point invoked by `bin/cdk` when `CDK_TELEMETRY_SENDER=1`. + * + * Reads a `TelemetrySenderConfig` as JSON from stdin, attempts one delivery, and always exits 0. + */ +export function main(): void { + const hardKill = setTimeout(() => process.exit(0), HARD_KILL_MS); + hardKill.unref(); + + const finish = () => { + clearTimeout(hardKill); + process.exit(0); + }; + + try { + void readAll(process.stdin, MAX_STDIN_BYTES) + .then((input) => (input === undefined ? undefined : deliver(input))) + .then(finish, finish); + } catch { + finish(); + } +} + +/** + * Read a stream to completion as UTF-8, giving up if it exceeds `maxBytes`. + * + * Chunks are measured and joined as `Buffer`s rather than strings: a string's `length` counts + * UTF-16 code units, so a cap applied to it would let a multi-byte payload through at up to three + * times the intended size. Buffering the raw bytes and decoding once at the end also avoids having + * to reason about multi-byte sequences that straddle a chunk boundary. + * + * Never rejects. Resolves `undefined` if the limit was exceeded or the stream errored, meaning + * "there is nothing here worth sending". + */ +export function readAll(stream: NodeJS.ReadableStream, maxBytes: number): Promise { + return new Promise((ok) => { + const chunks: Buffer[] = []; + let bytes = 0; + let overflowed = false; + + stream.on('data', (chunk: Buffer | string) => { + if (overflowed) { + return; + } + const buf = Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk); + bytes += buf.byteLength; + if (bytes > maxBytes) { + overflowed = true; + chunks.length = 0; + trace(`Input exceeded ${maxBytes} bytes, discarding`); + return; + } + chunks.push(buf); + }); + + stream.on('error', () => ok(undefined)); + stream.on('end', () => ok(overflowed ? undefined : Buffer.concat(chunks).toString('utf-8'))); + }); +} + +/** + * Parse a piped config and attempt delivery. Never rejects. + */ +async function deliver(input: string): Promise { + const result = await parseAndSend(input); + trace(result.sent + ? `Telemetry sent (${result.via}, ${result.statusCode})` + : `Telemetry not sent (${result.via}): ${result.reason}`); + return result; +} + +async function parseAndSend(input: string): Promise { + let cfg: TelemetrySenderConfig; + try { + cfg = JSON.parse(input) as TelemetrySenderConfig; + } catch (e: any) { + return { sent: false, via: 'skipped', reason: `MalformedPayload: ${e?.message}` }; + } + return sendTelemetry(cfg); +} + +/** + * Deliver a telemetry payload, tunnelling through a proxy when one applies. + * + * Never rejects and never throws: every failure is reported through the returned `SendResult`. + */ +export async function sendTelemetry(cfg: TelemetrySenderConfig, env: NodeJS.ProcessEnv = process.env): Promise { + try { + if (!cfg?.endpoint) { + return { sent: false, via: 'skipped', reason: 'NoEndpoint' }; + } + + const url = new URL(cfg.endpoint); + const timeoutMs = cfg.timeoutMs ?? NETWORK_TIMEOUT_MS; + const payload = JSON.stringify(cfg.body ?? {}); + + // `??`, not `||`: the parent forces its configured value whenever `--proxy` (or the `proxy` + // setting) is present at all -- including as an empty string, which means "no proxy" -- and + // never falls back to the environment in that case. Only auto-detect when nothing was + // forwarded, so the child reaches the same decision the parent would. + const proxyUrl = cfg.proxyUrl ?? resolveProxy(cfg.endpoint, proxyEnv(cfg, env)); + if (!proxyUrl) { + return await postDirect(url, payload, cfg.ca, timeoutMs); + } + + let proxy: URL; + try { + proxy = new URL(proxyUrl); + } catch { + return { sent: false, via: 'skipped', reason: `MalformedProxyUrl: ${proxyUrl}` }; + } + + if (!SUPPORTED_PROXY_PROTOCOLS.includes(proxy.protocol)) { + // Fail closed. Do NOT retry directly -- see SUPPORTED_PROXY_PROTOCOLS. + return { sent: false, via: 'skipped', reason: `UnsupportedProxyProtocol: ${proxy.protocol}` }; + } + + return await postViaProxy(url, proxy, payload, cfg.ca, timeoutMs); + } catch (e: any) { + return { sent: false, via: 'skipped', reason: `${e?.name ?? 'Error'}: ${e?.message}` }; + } +} + +/** + * Resolve the proxy to use for `endpoint` from proxy environment variables. + * + * Faithfully re-implements `proxy-from-env@1`, which is what `proxy-agent` falls back to in the + * parent process when the user did not pass `--proxy`. Kept in lockstep by a differential test + * (`test/cli/telemetry/resolve-proxy-parity.test.ts`) that runs both over the same table, so the + * quirks below are deliberate rather than accidental: + * + * - `npm_config_*` variants take precedence over the plain ones; + * - a `NO_PROXY` entry only does suffix matching if it starts with `.` or `*`, otherwise it must + * match the host exactly; + * - IPv6 hosts keep their brackets. + * + * Returns the empty string when no proxy applies. + */ +export function resolveProxy(endpoint: string, env: NodeJS.ProcessEnv): string { + let parsed: URL; + try { + parsed = new URL(endpoint); + } catch { + return ''; + } + + if (!parsed.host || !parsed.protocol) { + return ''; + } + + const protocol = parsed.protocol.split(':', 1)[0]; + // Strip the port off `host` rather than using `hostname`, to keep the brackets around IPv6 + // addresses (which is what NO_PROXY entries are matched against). + const host = parsed.host.replace(/:\d*$/, ''); + const port = parseInt(parsed.port, 10) || DEFAULT_PORTS[protocol] || 0; + + if (!shouldProxy(host, port, env)) { + return ''; + } + + let proxy = + getEnv(env, `npm_config_${protocol}_proxy`) || + getEnv(env, `${protocol}_proxy`) || + getEnv(env, 'npm_config_proxy') || + getEnv(env, 'all_proxy'); + + if (proxy && !proxy.includes('://')) { + // Missing scheme in proxy, default to the requested URL's scheme. + proxy = `${protocol}://${proxy}`; + } + return proxy; +} + +/** + * Apply an explicit `noProxy` override on top of the inherited environment. + */ +function proxyEnv(cfg: TelemetrySenderConfig, env: NodeJS.ProcessEnv): NodeJS.ProcessEnv { + if (cfg.noProxy === undefined) { + return env; + } + return { ...env, NO_PROXY: cfg.noProxy, no_proxy: cfg.noProxy, npm_config_no_proxy: cfg.noProxy }; +} + +function getEnv(env: NodeJS.ProcessEnv, name: string): string { + return env[name.toLowerCase()] || env[name.toUpperCase()] || ''; +} + +/** + * `NO_PROXY` matching, following `proxy-from-env@1`. + */ +function shouldProxy(host: string, port: number, env: NodeJS.ProcessEnv): boolean { + const noProxy = (getEnv(env, 'npm_config_no_proxy') || getEnv(env, 'no_proxy')).toLowerCase(); + if (!noProxy) { + return true; + } + if (noProxy === '*') { + return false; + } + + return noProxy.split(/[,\s]/).every((entry) => { + if (!entry) { + return true; + } + + const withPort = entry.match(/^(.+):(\d+)$/); + let entryHost = withPort ? withPort[1] : entry; + const entryPort = withPort ? parseInt(withPort[2], 10) : 0; + if (entryPort && entryPort !== port) { + return true; + } + + if (!/^[.*]/.test(entryHost)) { + // No wildcard, so this only stops proxying on an exact match. + return host !== entryHost; + } + + if (entryHost.charAt(0) === '*') { + entryHost = entryHost.slice(1); + } + return !host.endsWith(entryHost); + }); +} + +/** + * POST straight to the endpoint. `node:https` applies `ca` natively. + */ +function postDirect(url: URL, payload: string, ca: string | undefined, timeoutMs: number): Promise { + return new Promise((ok) => { + let settled = false; + const done = (result: SendResult) => { + if (!settled) { + settled = true; + ok(result); + } + }; + + const req = https.request({ + hostname: url.hostname, + port: url.port || null, + path: url.pathname, + method: 'POST', + headers: jsonHeaders(payload), + ca, + timeout: timeoutMs, + }, (res) => { + res.resume(); + done({ sent: isSuccess(res.statusCode), via: 'direct', statusCode: res.statusCode, reason: reasonFor(res.statusCode) }); + }); + + req.on('error', (e: any) => done({ sent: false, via: 'direct', reason: `${e?.code ?? e?.name}: ${e?.message}` })); + req.on('timeout', () => { + req.destroy(); + done({ sent: false, via: 'direct', reason: `RequestTimeout after ${timeoutMs}ms` }); + }); + req.end(payload); + }); +} + +/** + * Tunnel to the endpoint with an HTTP CONNECT, then speak HTTPS over the tunnelled socket. + * + * This mirrors what `https-proxy-agent` does for the CLI's other network calls, minus the parts + * we cannot support without external dependencies. + */ +async function postViaProxy(url: URL, proxy: URL, payload: string, ca: string | undefined, timeoutMs: number): Promise { + const port = Number(url.port || 443); + + let tunnel: net.Socket; + try { + tunnel = await openTunnel(proxy, url.hostname, port, ca, timeoutMs); + } catch (e: any) { + return { sent: false, via: 'connect-tunnel', reason: `${e?.code ?? e?.name}: ${e?.message}` }; + } + + let secure: tls.TLSSocket; + try { + secure = await upgradeToTls(tunnel, url.hostname, ca, timeoutMs); + } catch (e: any) { + tunnel.destroy(); + return { sent: false, via: 'connect-tunnel', reason: `${e?.code ?? e?.name}: ${e?.message}` }; + } + + try { + const statusCode = await postOverSocket(secure, hostHeader(url), url.pathname, payload, timeoutMs); + return { sent: isSuccess(statusCode), via: 'connect-tunnel', statusCode, reason: reasonFor(statusCode) }; + } catch (e: any) { + return { sent: false, via: 'connect-tunnel', reason: `${e?.code ?? e?.name}: ${e?.message}` }; + } finally { + secure.destroy(); + } +} + +/** + * Open a CONNECT tunnel through `proxy` to `host:port` and hand back the raw socket. + */ +function openTunnel(proxy: URL, host: string, port: number, ca: string | undefined, timeoutMs: number): Promise { + return new Promise((ok, ko) => { + const proxyHost = (proxy.hostname || '').replace(/^\[|\]$/g, ''); + const proxyPort = Number(proxy.port || (proxy.protocol === 'https:' ? 443 : 80)); + + const socket = proxy.protocol === 'https:' + ? tls.connect({ host: proxyHost, port: proxyPort, servername: sni(proxyHost), ca, ALPNProtocols: ['http/1.1'] }) + : net.connect({ host: proxyHost, port: proxyPort }); + + const timer = setTimeout(() => fail(error('ProxyConnectTimeout', `No CONNECT response after ${timeoutMs}ms`)), timeoutMs); + timer.unref(); + + let buffered = Buffer.alloc(0); + + function cleanup() { + clearTimeout(timer); + socket.removeListener('data', onData); + socket.removeListener('error', fail); + socket.removeListener('close', onClose); + } + + function fail(e: Error) { + cleanup(); + socket.destroy(); + ko(e); + } + + function onClose() { + fail(error('ProxyConnectionClosed', 'Proxy closed the connection before responding')); + } + + function onData(chunk: Buffer) { + buffered = Buffer.concat([buffered, chunk]); + + // Hard cap on what we will buffer before the tunnel is open. Checked unconditionally so it + // still fires when a terminator arrives inside an otherwise oversized chunk. Nothing + // legitimate can be large here: we have not sent a ClientHello yet, so there is no TLS + // traffic to pipeline. + if (buffered.length > MAX_PROXY_RESPONSE_BYTES) { + fail(error('ProxyResponseTooLarge', 'Proxy sent an oversized CONNECT response')); + return; + } + + const headerEnd = buffered.indexOf('\r\n\r\n'); + if (headerEnd === -1) { + return; + } + + const statusLine = buffered.subarray(0, buffered.indexOf('\r\n')).toString('latin1').trim(); + if (!isSuccess(Number(statusLine.split(' ')[1]))) { + fail(error('ProxyConnectFailed', statusLine)); + return; + } + + cleanup(); + + // A proxy may deliver bytes belonging to the tunnel in the same chunk as its response. Put + // them back so the TLS handshake that follows sees them, rather than dropping them. The + // socket must be paused first: removing our listener does not stop it flowing, and + // unshifting into a flowing stream silently discards the data. + const trailing = buffered.subarray(headerEnd + 4); + if (trailing.length > 0) { + socket.pause(); + socket.unshift(trailing); + } + + ok(socket); + } + + socket.on('data', onData); + socket.on('error', fail); + socket.on('close', onClose); + socket.once(proxy.protocol === 'https:' ? 'secureConnect' : 'connect', () => { + socket.write(connectRequest(proxy, host, port)); + }); + }); +} + +/** + * Render the CONNECT request line and headers, including Basic proxy auth when credentials are + * embedded in the proxy URL. + */ +function connectRequest(proxy: URL, host: string, port: number): string { + const target = net.isIPv6(host) ? `[${host}]` : host; + let out = `CONNECT ${target}:${port} HTTP/1.1\r\n`; + out += `Host: ${target}:${port}\r\n`; + out += 'Proxy-Connection: close\r\n'; + if (proxy.username || proxy.password) { + const credentials = `${decodeURIComponent(proxy.username)}:${decodeURIComponent(proxy.password)}`; + out += `Proxy-Authorization: Basic ${Buffer.from(credentials).toString('base64')}\r\n`; + } + return `${out}\r\n`; +} + +/** + * Upgrade an established tunnel to TLS against the *endpoint* (not the proxy). + * + * `host` matters as much as `servername` here, and for a different reason: `servername` drives the + * SNI extension (and is deliberately omitted for IP literals, which may not be sent as SNI), while + * `host` is what Node's `checkServerIdentity` matches the certificate against. With neither set, + * Node falls back to the underlying socket's host -- which on this path is the *proxy* -- so a + * certificate valid for the proxy's name would be accepted for a connection intended for the + * endpoint. Always pass the real destination. + */ +function upgradeToTls(socket: net.Socket, hostname: string, ca: string | undefined, timeoutMs: number): Promise { + return new Promise((ok, ko) => { + const secure = tls.connect({ socket, host: hostname, servername: sni(hostname), ca, ALPNProtocols: ['http/1.1'] }); + const timer = setTimeout(() => { + secure.destroy(); + ko(error('TlsHandshakeTimeout', `TLS handshake did not complete within ${timeoutMs}ms`)); + }, timeoutMs); + timer.unref(); + + secure.once('secureConnect', () => { + clearTimeout(timer); + ok(secure); + }); + secure.once('error', (e: Error) => { + clearTimeout(timer); + ko(e); + }); + }); +} + +/** + * Write a minimal HTTP/1.1 POST over an already-connected socket and read back the status code. + * + * We frame the request by hand because `http.request` cannot be pointed at a pre-existing + * `TLSSocket` without an Agent, and Agents are what we are avoiding here. + */ +function postOverSocket(socket: tls.TLSSocket, host: string, path: string, payload: string, timeoutMs: number): Promise { + return new Promise((ok, ko) => { + const timer = setTimeout(() => ko(error('ResponseTimeout', `No response within ${timeoutMs}ms`)), timeoutMs); + timer.unref(); + + let response = ''; + const onData = (chunk: Buffer) => { + response += chunk.toString('latin1'); + // Symmetric with the CONNECT response cap: bound what we accumulate so a server that never + // terminates its headers cannot grow this without limit inside the timeout window. + if (response.length > MAX_PROXY_RESPONSE_BYTES) { + clearTimeout(timer); + socket.removeListener('data', onData); + ko(error('ResponseTooLarge', 'Endpoint sent an oversized response header')); + return; + } + if (response.includes('\r\n\r\n')) { + clearTimeout(timer); + socket.removeListener('data', onData); + ok(Number(response.split(' ')[1])); + } + }; + + socket.on('data', onData); + socket.once('error', (e: Error) => { + clearTimeout(timer); + ko(e); + }); + + const headers = [ + `POST ${path} HTTP/1.1`, + `Host: ${host}`, + 'content-type: application/json', + `content-length: ${Buffer.byteLength(payload)}`, + 'connection: close', + ].join('\r\n'); + socket.write(`${headers}\r\n\r\n${payload}`); + }); +} + +function jsonHeaders(payload: string): Record { + return { + 'content-type': 'application/json', + 'content-length': Buffer.byteLength(payload), + }; +} + +function hostHeader(url: URL): string { + return url.port ? `${url.hostname}:${url.port}` : url.hostname; +} + +/** + * TLS servername, omitted for IP literals (which must not be sent as SNI). + */ +function sni(host: string): string | undefined { + return net.isIP(host) ? undefined : host; +} + +function isSuccess(statusCode: number | undefined): boolean { + return statusCode !== undefined && statusCode >= 200 && statusCode < 300; +} + +function reasonFor(statusCode: number | undefined): string | undefined { + return isSuccess(statusCode) ? undefined : `UnexpectedStatusCode: ${statusCode}`; +} + +/** + * Build (never throw) a named error. + * + * `ToolkitError` is unavailable here, and a bare `throw` is banned by lint, so failures travel as + * rejections carrying one of these. + */ +function error(name: string, message: string): Error { + const e = new Error(message); + e.name = name; + return e; +} + +/** + * Diagnostics for the detached child, which has no IoHost. + * + * stderr is `ignore`d by the parent, so this is only visible when the sender is run by hand with + * `CDK_TELEMETRY_SENDER_DEBUG=1`. Written synchronously: `process.stderr` is asynchronous when it + * is a pipe, and the `process.exit(0)` that follows would discard a buffered write. + */ +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 index ed5173d01..5baf231f4 100644 --- a/packages/aws-cdk/lib/cli/telemetry/sink/endpoint-sink.ts +++ b/packages/aws-cdk/lib/cli/telemetry/sink/endpoint-sink.ts @@ -1,14 +1,31 @@ -import type { IncomingMessage } from 'http'; -import type { Agent } from 'https'; -import { request } from 'https'; +import { spawn } from 'node:child_process'; +import * as fs from 'node:fs'; +import * as os from 'node:os'; 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; +/** + * Largest payload we are willing to hand to the detached sender. + * + * The payload is written to the child's stdin. Once it exceeds the OS pipe buffer plus libuv's + * own buffering, `write()` no longer completes eagerly and the parent ends up waiting for the + * child to drain -- which is exactly the blocking behaviour the detached sender exists to remove. + * Measured on Linux/Node 20, the parent still exits in ~37ms at 200KB but stalls for seconds at + * 400KB, so 64KB leaves a wide margin. Realistic batches are 3-10KB. + */ +const MAX_DISPATCH_PAYLOAD_BYTES = 65_536; + +/** + * Stable prefix of the trace emitted once a batch has been handed to the sender. + * + * Integration tests match on this literal, so it must not change casually. Note that it reports a + * successful hand-off, not a successful delivery -- by design nobody in this process ever learns + * whether the POST succeeded. + */ +const DISPATCHED_TRACE = 'Telemetry dispatched'; /** * Properties for the Endpoint Telemetry Client @@ -25,23 +42,56 @@ export interface EndpointTelemetrySinkProps { readonly ioHost: IIoHost; /** - * The agent responsible for making the network requests. + * Absolute path to this CLI's `bin/cdk` script, used to respawn ourselves as a telemetry sender. + * + * Without it we cannot dispatch, and telemetry is silently skipped. * - * Use this to set up a proxy connection. + * @default - telemetry is not sent + */ + readonly binCdkPath?: string; + + /** + * Proxy the sender should tunnel through, as configured by `--proxy` or the `proxy` setting. + * + * When absent, the sender falls back to the inherited proxy environment variables, which is the + * same behaviour `proxy-agent` gives the rest of the CLI. + * + * @default - resolved from the environment by the sender + */ + readonly proxyUrl?: string; + + /** + * Contents of the CA bundle to trust, as configured by `--ca-bundle-path` or `AWS_CA_BUNDLE`. * - * @default - Uses the shared global node agent + * @default - only the system trust store */ - readonly agent?: Agent; + readonly caCert?: string; } /** * The telemetry client that hits an external endpoint. + * + * The HTTP POST itself does not happen in this process. Events are handed to a detached child + * process (`bin/cdk` re-invoked with `CDK_TELEMETRY_SENDER=1`) which outlives us, so the CLI can + * exit without waiting on the network. + * + * Deliberately nothing here checks first whether the network is reachable. Any such check is + * itself a network call on the CLI's exit path, which is what this sink exists to avoid. When the + * machine is offline we simply spawn a child that fails and exits: the child has its own timeouts + * and swallows every error, so the cost of being wrong is one short-lived process. + * + * For the same reason this sink imposes no network timeout on the child. The old 500ms per-attempt + * budget existed to keep a synchronous POST from delaying the user's prompt; nothing waits on the + * sender now, so it owns a budget appropriate to actually completing a request (see + * `NETWORK_TIMEOUT_MS` in `../sender`). */ export class EndpointTelemetrySink implements ITelemetrySink { private events: TelemetrySchema[] = []; private endpoint: URL; private ioHelper: IoHelper; - private agent?: Agent; + private binCdkPath?: string; + private proxyUrl?: string; + private caCert?: string; public constructor(props: EndpointTelemetrySinkProps) { this.endpoint = new URL(props.endpoint); @@ -51,7 +101,9 @@ export class EndpointTelemetrySink implements ITelemetrySink { } this.ioHelper = IoHelper.fromActionAwareIoHost(props.ioHost); - this.agent = props.agent; + this.binCdkPath = props.binCdkPath; + this.proxyUrl = props.proxyUrl; + this.caCert = props.caCert; // Batch events every 30 seconds setInterval(() => this.flush(), 30000).unref(); @@ -75,7 +127,7 @@ export class EndpointTelemetrySink implements ITelemetrySink { return; } - const res = await this.https(this.endpoint, { events: this.events }); + const res = await this.dispatch(this.endpoint, { events: this.events }); // Clear the events array after successful output if (res) { @@ -88,67 +140,87 @@ export class EndpointTelemetrySink implements ITelemetrySink { } /** - * Returns true if telemetry successfully posted, false otherwise. + * Hand the batch to a detached sender process. + * + * Returns true if the batch reached a terminal state (either handed off, or dropped because it + * can never be delivered) and should therefore be cleared. Returns false if it is worth + * retrying on the next flush. */ - private async https( + private async dispatch( 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'); + if (!this.binCdkPath) { + await this.ioHelper.defaults.trace('Telemetry not sent: unable to locate the CLI entrypoint to spawn a sender'); 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; - } + const payload = JSON.stringify({ + endpoint: url.href, + body, + proxyUrl: this.proxyUrl, + ca: this.caCert, + }); - await this.ioHelper.defaults.trace(`Telemetry Unsuccessful: POST ${url.hostname}${url.pathname}: ${res.statusCode}:${res.statusMessage}`); + const payloadBytes = Buffer.byteLength(payload); + if (payloadBytes > MAX_DISPATCH_PAYLOAD_BYTES) { + // Writing this much to the child's stdin would block our own exit. Drop the batch; it is + // not going to get smaller on a retry. + await this.ioHelper.defaults.trace(`Telemetry dropped: payload of ${payloadBytes} bytes exceeds ${MAX_DISPATCH_PAYLOAD_BYTES}`); + return true; + } - return false; + try { + const child = spawn(process.execPath, [this.binCdkPath], { + detached: true, + stdio: ['pipe', 'ignore', '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(), + env: { + ...process.env, + CDK_TELEMETRY_SENDER: '1', + }, + }); + + // The child is on its own from here; a spawn failure must not surface anywhere. These fire + // after the CLI may already have exited, so they cannot go through the IoHost -- see + // `debugTrace`. + child.on('error', (e: Error) => { + debugTrace(`failed to spawn sender: ${e.message}`); + }); + child.stdin?.on('error', (e: Error) => { + debugTrace(`failed to write payload to sender: ${e.message}`); + }); + + child.stdin?.end(payload); + child.unref(); + + await this.ioHelper.defaults.trace(`${DISPATCHED_TRACE} (pid ${child.pid}, ${payloadBytes} bytes)`); + return true; } catch (e: any) { - await this.ioHelper.defaults.trace(`Telemetry Error: POST ${url.hostname}${url.pathname}: ${JSON.stringify(e)}`); + await this.ioHelper.defaults.trace(`Telemetry Error: spawning sender for POST ${url.hostname}${url.pathname} failed: ${e.message}`); return false; } } } /** - * A Promisified version of `https.request()` + * Diagnostics for failures that surface after the CLI may already have exited. + * + * The child's `error` events fire asynchronously, potentially once the IoHost is gone and the + * process is on its way out, so they cannot be reported through the normal trace channel. Written + * synchronously to fd 2 for the same reason the sender does it, and gated behind the same variable + * so it is silent unless somebody is deliberately debugging telemetry delivery. */ -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); - }); +function debugTrace(message: string): void { + if (process.env.CDK_TELEMETRY_SENDER_DEBUG !== '1') { + 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/test/cli/telemetry/cli-bin-path.test.ts b/packages/aws-cdk/test/cli/telemetry/cli-bin-path.test.ts new file mode 100644 index 000000000..dc284985b --- /dev/null +++ b/packages/aws-cdk/test/cli/telemetry/cli-bin-path.test.ts @@ -0,0 +1,54 @@ +import * as fs from 'node:fs'; +import * as os from 'node:os'; +import * as path from 'node:path'; +import { CLI_BIN_PATH_ENV, cliBinPath } from '../../../lib/cli/telemetry/cli-bin-path'; + +describe('cliBinPath', () => { + let tempDir: string; + + beforeEach(() => { + tempDir = fs.mkdtempSync(path.join(os.tmpdir(), 'cli-bin-path-')); + }); + + afterEach(() => { + fs.rmSync(tempDir, { recursive: true, force: true }); + }); + + test('prefers the path bin/cdk published about itself', () => { + const binPath = path.join(tempDir, 'cdk'); + fs.writeFileSync(binPath, '#!/usr/bin/env node\n'); + + expect(cliBinPath({ [CLI_BIN_PATH_ENV]: binPath })).toBe(binPath); + }); + + test('ignores the environment variable when it points at nothing', () => { + const result = cliBinPath({ [CLI_BIN_PATH_ENV]: path.join(tempDir, 'does-not-exist') }); + + // Falls back to walking up to this package's own bin/cdk, which does exist in the repo. + expect(result).toBeDefined(); + expect(result!.endsWith(path.join('bin', 'cdk'))).toBe(true); + }); + + test('falls back to the package-relative bin/cdk when the variable is absent', () => { + const result = cliBinPath({}); + + expect(result).toBeDefined(); + expect(fs.existsSync(result!)).toBe(true); + expect(result!.endsWith(path.join('bin', 'cdk'))).toBe(true); + }); + + test('the resolved fallback is this package\'s real entrypoint', () => { + const result = cliBinPath({})!; + + // Sanity check that we resolved the actual CLI entrypoint and not some other file named `cdk`: + // it must contain the sender dispatch guard. + expect(fs.readFileSync(result, 'utf-8')).toContain('CDK_TELEMETRY_SENDER'); + }); + + test('does not use process.argv[1]', () => { + // argv[1] under jest is the jest worker, which must never be respawned as a telemetry sender. + const result = cliBinPath({}); + + expect(result).not.toBe(process.argv[1]); + }); +}); diff --git a/packages/aws-cdk/test/cli/telemetry/resolve-proxy-parity.test.ts b/packages/aws-cdk/test/cli/telemetry/resolve-proxy-parity.test.ts new file mode 100644 index 000000000..10dcbd2b9 --- /dev/null +++ b/packages/aws-cdk/test/cli/telemetry/resolve-proxy-parity.test.ts @@ -0,0 +1,83 @@ +/** + * Differential parity test: our built-ins-only proxy resolution vs. the real thing. + * + * The CLI itself resolves proxies with `proxy-agent`, which delegates to `proxy-from-env` whenever + * the user did not pass `--proxy`. The detached telemetry sender cannot use `proxy-agent` (it has + * no dependencies available), so `resolveProxy` re-implements that logic. This test pins the + * re-implementation to the original by running both over the same table of environments. + * + * Note that we deliberately resolve `proxy-from-env` *through* `proxy-agent` rather than importing + * it directly. A bare import picks up the hoisted copy, which is a different major version with + * different `NO_PROXY` semantics -- comparing against that would make this test worse than + * useless. `proxy-agent` is a real dependency of this package, and this reaches the exact copy it + * uses. + */ +import { resolveProxy } from '../../../lib/cli/telemetry/sender'; + +// eslint-disable-next-line @typescript-eslint/no-require-imports +const realGetProxyForUrl: (url: string) => string = require( + require.resolve('proxy-from-env', { paths: [require.resolve('proxy-agent')] }), +).getProxyForUrl; + +const TELEMETRY_URL = 'https://cdk-cli-telemetry.us-east-1.api.aws/metrics'; + +const CASES: Array<[name: string, url: string, env: Record]> = [ + ['HTTPS_PROXY set', TELEMETRY_URL, { HTTPS_PROXY: 'http://corp:8080' }], + ['lowercase https_proxy', TELEMETRY_URL, { https_proxy: 'http://corp:8080' }], + ['only HTTP_PROXY set (must not apply to https)', TELEMETRY_URL, { HTTP_PROXY: 'http://corp:8080' }], + ['ALL_PROXY', TELEMETRY_URL, { ALL_PROXY: 'http://corp:8080' }], + ['lowercase all_proxy', TELEMETRY_URL, { all_proxy: 'http://corp:8080' }], + ['no proxy variables at all', TELEMETRY_URL, {}], + ['NO_PROXY exact host', TELEMETRY_URL, { HTTPS_PROXY: 'http://corp:8080', NO_PROXY: 'cdk-cli-telemetry.us-east-1.api.aws' }], + ['NO_PROXY domain suffix', TELEMETRY_URL, { HTTPS_PROXY: 'http://corp:8080', NO_PROXY: '.api.aws' }], + ['NO_PROXY suffix without dot', TELEMETRY_URL, { HTTPS_PROXY: 'http://corp:8080', NO_PROXY: 'api.aws' }], + ['NO_PROXY wildcard', TELEMETRY_URL, { HTTPS_PROXY: 'http://corp:8080', NO_PROXY: '*' }], + ['NO_PROXY non-matching', TELEMETRY_URL, { HTTPS_PROXY: 'http://corp:8080', NO_PROXY: 'example.com' }], + ['NO_PROXY comma+space list', TELEMETRY_URL, { HTTPS_PROXY: 'http://corp:8080', NO_PROXY: 'foo.com, .api.aws ,bar.com' }], + ['NO_PROXY host:port match', TELEMETRY_URL, { HTTPS_PROXY: 'http://corp:8080', NO_PROXY: 'cdk-cli-telemetry.us-east-1.api.aws:443' }], + ['NO_PROXY host:port mismatch', TELEMETRY_URL, { HTTPS_PROXY: 'http://corp:8080', NO_PROXY: 'cdk-cli-telemetry.us-east-1.api.aws:8443' }], + ['NO_PROXY empty entries', TELEMETRY_URL, { HTTPS_PROXY: 'http://corp:8080', NO_PROXY: ',, ,' }], + ['scheme-less proxy value', TELEMETRY_URL, { HTTPS_PROXY: 'corp:8080' }], + ['npm_config_https_proxy fallback', TELEMETRY_URL, { npm_config_https_proxy: 'http://corp:8080' }], + ['npm_config_proxy fallback', TELEMETRY_URL, { npm_config_proxy: 'http://corp:8080' }], + ['socks proxy passes through unchanged', TELEMETRY_URL, { HTTPS_PROXY: 'socks5://corp:1080' }], + ['pac proxy passes through unchanged', TELEMETRY_URL, { HTTPS_PROXY: 'pac+http://corp/proxy.pac' }], + ['authenticated proxy url', TELEMETRY_URL, { HTTPS_PROXY: 'http://user:pass@corp:8080' }], + ['explicit non-default port + NO_PROXY host', 'https://localhost:8443/metrics', { HTTPS_PROXY: 'http://corp:8080', NO_PROXY: 'localhost' }], + ['explicit non-default port + NO_PROXY host:port', 'https://localhost:8443/metrics', { HTTPS_PROXY: 'http://corp:8080', NO_PROXY: 'localhost:8443' }], + ['explicit non-default port + wrong NO_PROXY port', 'https://localhost:8443/metrics', { HTTPS_PROXY: 'http://corp:8080', NO_PROXY: 'localhost:9999' }], + ['http endpoint uses HTTP_PROXY', 'http://example.com/x', { HTTP_PROXY: 'http://corp:8080' }], + ['http endpoint ignores HTTPS_PROXY', 'http://example.com/x', { HTTPS_PROXY: 'http://corp:8080' }], + ['uppercase NO_PROXY beats nothing', TELEMETRY_URL, { HTTPS_PROXY: 'http://corp:8080', no_proxy: '.api.aws' }], + ['IPv6 endpoint', 'https://[::1]:8443/x', { HTTPS_PROXY: 'http://corp:8080' }], +]; + +describe('resolveProxy parity with proxy-from-env', () => { + const savedEnv = process.env; + + afterEach(() => { + process.env = savedEnv; + }); + + test.each(CASES)('%s', (_name, url, env) => { + // proxy-from-env reads process.env directly, so swap it for the duration of the call. + process.env = { ...env } as NodeJS.ProcessEnv; + let expected: string; + try { + expected = realGetProxyForUrl(url); + } finally { + process.env = savedEnv; + } + + expect(resolveProxy(url, env)).toEqual(expected); + }); + + test('the reference implementation is the version proxy-agent actually uses', () => { + const resolved = require.resolve('proxy-from-env', { paths: [require.resolve('proxy-agent')] }); + // eslint-disable-next-line @typescript-eslint/no-require-imports + const version = require(`${resolved.slice(0, resolved.lastIndexOf('/'))}/package.json`).version; + + // v2 changed NO_PROXY matching; if this ever bumps, the parity table above must be revisited. + expect(version).toMatch(/^1\./); + }); +}); 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..647ba6141 --- /dev/null +++ b/packages/aws-cdk/test/cli/telemetry/sender.test.ts @@ -0,0 +1,582 @@ +/** + * 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 listener. Nothing here is mocked, because the whole + * point of the sender is that it re-implements transport behaviour that we otherwise get from + * `proxy-agent`, 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 { Readable } from 'node:stream'; +import { generateTestCa, type TestCa } from './test-tls'; +import { readAll, resolveProxy, sendTelemetry } from '../../../lib/cli/telemetry/sender'; + +jest.setTimeout(30_000); + +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 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}'); + }); + }); + await new Promise((ok) => server.listen(0, '127.0.0.1', ok)); + const port = (server.address() as net.AddressInfo).port; + return { + url: `https://${options.urlHost ?? 'localhost'}:${port}/metrics`, + received, + close: () => new Promise((ok) => server.close(() => ok())), + }; +} + +interface Proxy { + readonly url: string; + readonly connects: string[]; + readonly authHeaders: Array; + close(): Promise; +} + +interface ConnectProxyOptions { + readonly requireAuth?: string; + readonly delayConnectResponseMs?: number; + readonly appendAfterConnectResponse?: string; +} + +async function startConnectProxy(options: ConnectProxyOptions = {}): Promise { + const connects: string[] = []; + const authHeaders: Array = []; + const server = http.createServer((_req, res) => { + res.writeHead(400); + res.end('CONNECT only'); + }); + + 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${options.appendAfterConnectResponse ?? ''}`); + if (head?.length) { + upstream.write(head); + } + upstream.pipe(clientSocket); + clientSocket.pipe(upstream); + }; + if (options.delayConnectResponseMs) { + setTimeout(established, options.delayConnectResponseMs); + } else { + established(); + } + }); + 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: () => new Promise((ok) => server.close(() => ok())), + }; +} + +const BODY = { events: [{ identifiers: { sessionId: 'test-session' } }] }; + +describe('sender', () => { + let ca: TestCa; + + beforeAll(() => { + ca = generateTestCa(); + }); + + describe('direct delivery', () => { + test('POSTs the payload and reports success', async () => { + const endpoint = await startEndpoint(ca); + try { + const result = await sendTelemetry({ endpoint: endpoint.url, body: BODY, ca: ca.caCert, timeoutMs: 5000 }, {}); + + expect(result).toEqual({ sent: true, via: 'direct', statusCode: 200, reason: undefined }); + 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 as not sent', async () => { + const endpoint = await startEndpoint(ca, { statusCode: 500 }); + try { + const result = await sendTelemetry({ endpoint: endpoint.url, body: BODY, ca: ca.caCert, timeoutMs: 5000 }, {}); + + expect(result.sent).toBe(false); + expect(result.statusCode).toBe(500); + expect(result.reason).toContain('UnexpectedStatusCode'); + } finally { + await endpoint.close(); + } + }); + + test('rejects an untrusted certificate when no CA is supplied', async () => { + const endpoint = await startEndpoint(ca); + try { + const result = await sendTelemetry({ endpoint: endpoint.url, body: BODY, timeoutMs: 5000 }, {}); + + expect(result.sent).toBe(false); + expect(result.reason).toContain('UNABLE_TO_VERIFY_LEAF_SIGNATURE'); + expect(endpoint.received).toHaveLength(0); + } finally { + await endpoint.close(); + } + }); + + test('reports connection failures without throwing', async () => { + // Port 1 is reserved and nothing listens on it. + const result = await sendTelemetry({ endpoint: 'https://127.0.0.1:1/metrics', body: BODY, timeoutMs: 2000 }, {}); + + expect(result.sent).toBe(false); + expect(result.via).toBe('direct'); + expect(result.reason).toContain('ECONNREFUSED'); + }); + }); + + describe('proxy delivery', () => { + test('tunnels through an http:// proxy with CONNECT', async () => { + const endpoint = await startEndpoint(ca); + const proxy = await startConnectProxy(); + try { + const result = await sendTelemetry({ + endpoint: endpoint.url, + body: BODY, + proxyUrl: proxy.url, + ca: ca.caCert, + timeoutMs: 5000, + }, {}); + + expect(result).toEqual({ sent: true, via: 'connect-tunnel', statusCode: 200, reason: undefined }); + 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@'); + const result = await sendTelemetry({ endpoint: endpoint.url, body: BODY, proxyUrl: authed, ca: ca.caCert, timeoutMs: 5000 }, {}); + + expect(result.sent).toBe(true); + expect(proxy.authHeaders[0]).toBe(`Basic ${Buffer.from('alice:s3cret').toString('base64')}`); + } finally { + await proxy.close(); + await endpoint.close(); + } + }); + + test('surfaces a 407 from the proxy without throwing', async () => { + const endpoint = await startEndpoint(ca); + const proxy = await startConnectProxy({ requireAuth: 'alice:s3cret' }); + try { + const result = await sendTelemetry({ endpoint: endpoint.url, body: BODY, proxyUrl: proxy.url, ca: ca.caCert, timeoutMs: 5000 }, {}); + + expect(result.sent).toBe(false); + expect(result.via).toBe('connect-tunnel'); + expect(result.reason).toContain('407'); + 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 { + const result = await sendTelemetry( + { endpoint: endpoint.url, body: BODY, ca: ca.caCert, timeoutMs: 5000 }, + { HTTPS_PROXY: proxy.url }, + ); + + expect(result.sent).toBe(true); + expect(result.via).toBe('connect-tunnel'); + 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 { + const result = await sendTelemetry( + { endpoint: endpoint.url, body: BODY, ca: ca.caCert, timeoutMs: 5000 }, + { HTTPS_PROXY: proxy.url, NO_PROXY: 'localhost' }, + ); + + expect(result.via).toBe('direct'); + expect(result.sent).toBe(true); + expect(proxy.connects).toHaveLength(0); + } finally { + await proxy.close(); + await endpoint.close(); + } + }); + + test('an explicit noProxy overrides the inherited NO_PROXY', async () => { + const endpoint = await startEndpoint(ca); + const proxy = await startConnectProxy(); + try { + const result = await sendTelemetry( + { endpoint: endpoint.url, body: BODY, ca: ca.caCert, timeoutMs: 5000, noProxy: 'somewhere-else.example.com' }, + { HTTPS_PROXY: proxy.url, NO_PROXY: 'localhost' }, + ); + + expect(result.via).toBe('connect-tunnel'); + expect(proxy.connects).toHaveLength(1); + } finally { + await proxy.close(); + await endpoint.close(); + } + }); + + // 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. + test('tolerates a proxy handshake slower than the old 500ms budget', async () => { + const endpoint = await startEndpoint(ca); + const proxy = await startConnectProxy({ delayConnectResponseMs: 800 }); + try { + // Deliberately no `timeoutMs`: this exercises the sender's own default budget. + const result = await sendTelemetry({ + endpoint: endpoint.url, + body: BODY, + proxyUrl: proxy.url, + ca: ca.caCert, + }, {}); + + expect(result).toEqual({ sent: true, via: 'connect-tunnel', statusCode: 200, reason: undefined }); + expect(JSON.parse(endpoint.received[0].body)).toEqual(BODY); + } finally { + await proxy.close(); + await endpoint.close(); + } + }); + + test('still honours an explicit timeout when the proxy is too slow', async () => { + // The budget was widened, not removed. + const endpoint = await startEndpoint(ca); + const proxy = await startConnectProxy({ delayConnectResponseMs: 1500 }); + try { + const result = await sendTelemetry({ + endpoint: endpoint.url, + body: BODY, + proxyUrl: proxy.url, + ca: ca.caCert, + timeoutMs: 300, + }, {}); + + expect(result.sent).toBe(false); + expect(result.reason).toContain('ProxyConnectTimeout'); + expect(endpoint.received).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 { + const result = await sendTelemetry( + { endpoint: endpoint.url, body: BODY, ca: ca.caCert, timeoutMs: 5000, proxyUrl: '' }, + { HTTPS_PROXY: proxy.url }, + ); + + expect(result.via).toBe('direct'); + expect(result.sent).toBe(true); + expect(proxy.connects).toHaveLength(0); + } finally { + await proxy.close(); + await endpoint.close(); + } + }); + + test('replays bytes a proxy sends in the same chunk as its CONNECT response', async () => { + // A proxy may coalesce tunnel bytes into the same write as `200 Connection Established`. + // Those belong to the TLS stream and must not be dropped. Asserting that is awkward directly, + // so this injects bytes that are NOT valid TLS: if they are replayed the handshake breaks + // (which is what we assert), whereas if they were silently discarded it would succeed. + const endpoint = await startEndpoint(ca); + const proxy = await startConnectProxy({ appendAfterConnectResponse: 'NOT-TLS' }); + try { + const result = await sendTelemetry({ + endpoint: endpoint.url, + body: BODY, + proxyUrl: proxy.url, + ca: ca.caCert, + timeoutMs: 5000, + }, {}); + + expect(result.sent).toBe(false); + expect(result.via).toBe('connect-tunnel'); + expect(endpoint.received).toHaveLength(0); + } finally { + await proxy.close(); + await endpoint.close(); + } + }); + }); + + describe('certificate identity', () => { + // Trusting the signer is not enough -- the certificate also has to cover the host we asked for. + // Only the signer half used to be tested, which let a real gap through on the proxied path: + // `tls.connect` was given no `host`, so for an IP-literal endpoint (where SNI must be omitted) + // Node fell back to the underlying socket's host -- the PROXY -- and happily accepted a + // certificate issued for the proxy's name. + + 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 { + const result = await sendTelemetry({ endpoint: endpoint.url, body: BODY, ca: wrongCa.caCert, timeoutMs: 5000 }, {}); + + expect(result.sent).toBe(false); + expect(result.via).toBe('direct'); + expect(result.reason).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 { + const result = await sendTelemetry({ + endpoint: endpoint.url, + body: BODY, + proxyUrl: proxy.url, + ca: wrongCa.caCert, + timeoutMs: 5000, + }, {}); + + expect(result.sent).toBe(false); + expect(result.via).toBe('connect-tunnel'); + expect(result.reason).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 regression case. The certificate covers DNS:localhost but NOT IP:127.0.0.1, and the + // proxy is reached as `localhost` -- 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 { + const result = await sendTelemetry({ + endpoint: endpoint.url, + body: BODY, + proxyUrl: proxy.url, + ca: localhostOnlyCa.caCert, + timeoutMs: 5000, + }, {}); + + expect(result.sent).toBe(false); + expect(result.via).toBe('connect-tunnel'); + expect(result.reason).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 { + const result = await sendTelemetry({ + endpoint: endpoint.url, + body: BODY, + proxyUrl: proxy.url, + ca: ca.caCert, + timeoutMs: 5000, + }, {}); + + expect(result).toEqual({ sent: true, via: 'connect-tunnel', statusCode: 200, reason: undefined }); + expect(JSON.parse(endpoint.received[0].body)).toEqual(BODY); + } finally { + await proxy.close(); + await endpoint.close(); + } + }); + }); + + describe('fails closed', () => { + // A proxy is normally mandatory rather than advisory: corporate setups firewall direct egress. + // Falling back to a direct connection would be both futile and a policy violation. + test.each([ + 'socks://127.0.0.1:1080', + 'socks4://127.0.0.1:1080', + 'socks5://127.0.0.1:1080', + 'socks5h://127.0.0.1:1080', + 'pac+http://127.0.0.1:8080/proxy.pac', + 'pac+https://127.0.0.1:8080/proxy.pac', + 'pac+file:///etc/proxy.pac', + 'pac+data:application/x-ns-proxy-autoconfig,foo', + ])('skips (never falls back to direct) for %s', async (proxyUrl) => { + const endpoint = await startEndpoint(ca); + try { + const result = await sendTelemetry({ endpoint: endpoint.url, body: BODY, proxyUrl, ca: ca.caCert, timeoutMs: 5000 }, {}); + + expect(result.sent).toBe(false); + expect(result.via).toBe('skipped'); + expect(result.reason).toMatch(/UnsupportedProxyProtocol/); + // The critical assertion: nothing reached the endpoint directly. + expect(endpoint.received).toHaveLength(0); + } finally { + await endpoint.close(); + } + }); + + test('skips a malformed proxy URL', async () => { + const result = await sendTelemetry({ endpoint: 'https://example.com/m', body: BODY, proxyUrl: ':::not a url', timeoutMs: 500 }, {}); + + expect(result).toMatchObject({ sent: false, via: 'skipped' }); + expect(result.reason).toContain('MalformedProxyUrl'); + }); + + test.each([ + ['a missing endpoint', {}], + ['an empty endpoint', { endpoint: '' }], + ['a malformed endpoint', { endpoint: 'not-a-url' }], + ])('skips %s without throwing', async (_name, cfg) => { + const result = await sendTelemetry(cfg as any, {}); + + expect(result.sent).toBe(false); + expect(result.via).toBe('skipped'); + }); + + test('never rejects, even on garbage input', async () => { + await expect(sendTelemetry(undefined as any, {})).resolves.toMatchObject({ sent: false, via: 'skipped' }); + await expect(sendTelemetry(null as any, {})).resolves.toMatchObject({ sent: false, via: 'skipped' }); + }); + }); + + describe('resolveProxy', () => { + test('returns empty string for an unparseable endpoint', () => { + expect(resolveProxy('not a url', { HTTPS_PROXY: 'http://corp:8080' })).toBe(''); + }); + + test('prefixes a scheme-less proxy with the target scheme', () => { + expect(resolveProxy('https://example.com/x', { HTTPS_PROXY: 'corp:8080' })).toBe('https://corp:8080'); + }); + + test('does not use HTTP_PROXY for an https endpoint', () => { + expect(resolveProxy('https://example.com/x', { HTTP_PROXY: 'http://corp:8080' })).toBe(''); + }); + }); + + describe('readAll', () => { + test('joins chunks and decodes as UTF-8', async () => { + const stream = Readable.from([Buffer.from('{"a":'), Buffer.from('1}')]); + + await expect(readAll(stream, 1024)).resolves.toBe('{"a":1}'); + }); + + test('decodes a multi-byte character split across two chunks', async () => { + // '€' is E2 82 AC; feeding it as two chunks would corrupt a naive per-chunk decode. + const euro = Buffer.from('€', 'utf-8'); + const stream = Readable.from([euro.subarray(0, 1), euro.subarray(1)]); + + await expect(readAll(stream, 1024)).resolves.toBe('€'); + }); + + test('measures the cap in bytes, not UTF-16 code units', async () => { + // 10 x '€' is 10 UTF-16 code units but 30 bytes. A cap compared against string `.length` + // would wave this through at a 20 byte limit; it must not. + const payload = Buffer.from('€'.repeat(10), 'utf-8'); + expect(payload.byteLength).toBe(30); + + await expect(readAll(Readable.from([payload]), 20)).resolves.toBeUndefined(); + await expect(readAll(Readable.from([payload]), 30)).resolves.toBe('€'.repeat(10)); + }); + + test('gives up once the running total exceeds the cap', async () => { + const stream = Readable.from([Buffer.alloc(8, 0x61), Buffer.alloc(8, 0x61)]); + + await expect(readAll(stream, 10)).resolves.toBeUndefined(); + }); + + test('accepts a payload exactly at the cap', async () => { + const stream = Readable.from([Buffer.alloc(10, 0x61)]); + + await expect(readAll(stream, 10)).resolves.toBe('a'.repeat(10)); + }); + + test('resolves undefined on a stream error rather than rejecting', async () => { + const stream = new Readable({ + read() { + this.destroy(new Error('EPIPE')); + }, + }); + + await expect(readAll(stream, 1024)).resolves.toBeUndefined(); + }); + + test('tolerates string chunks', async () => { + // Defensive: nothing calls setEncoding today, but a future change must not silently break + // the byte accounting. + const stream = Readable.from(['hello']); + + await expect(readAll(stream, 1024)).resolves.toBe('hello'); + }); + }); +}); 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 index ec024982e..668f6d9de 100644 --- a/packages/aws-cdk/test/cli/telemetry/sink/endpoint-sink.test.ts +++ b/packages/aws-cdk/test/cli/telemetry/sink/endpoint-sink.test.ts @@ -1,30 +1,37 @@ -import * as https from 'https'; +import { spawn } from 'node:child_process'; +import * as os from 'node:os'; 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(), +jest.mock('node:child_process', () => ({ + spawn: jest.fn(), })); -// Mock NetworkDetector -jest.mock('../../../../lib/api/network-detector', () => ({ - NetworkDetector: { - hasConnectivity: jest.fn(), - }, -})); +const BIN_CDK = '/fake/pkg/bin/cdk'; + +interface MockChild { + pid: number; + on: jest.Mock; + unref: jest.Mock; + stdin: { on: jest.Mock; end: jest.Mock }; +} describe('EndpointTelemetrySink', () => { let ioHost: CliIoHost; + let child: MockChild; beforeEach(() => { jest.resetAllMocks(); - // Mock NetworkDetector to return true by default for existing tests - (NetworkDetector.hasConnectivity as jest.Mock).mockResolvedValue(true); + child = { + pid: 4242, + on: jest.fn(), + unref: jest.fn(), + stdin: { on: jest.fn(), end: jest.fn() }, + }; + (spawn as jest.Mock).mockReturnValue(child); ioHost = CliIoHost.instance(); }); @@ -33,310 +40,237 @@ describe('EndpointTelemetrySink', () => { 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', - }; + function sink(props: Partial[0]> = {}) { + return new EndpointTelemetrySink({ + endpoint: 'https://example.com/telemetry', + ioHost, + binCdkPath: BIN_CDK, + ...props, + }); + } - // Create the mock request object - const mockRequest = { - on: jest.fn(), - end: jest.fn(), - setTimeout: jest.fn(), - }; + /** + * The JSON that was piped to the detached sender on the Nth spawn. + */ + function pipedPayload(nth = 0) { + return JSON.parse(child.stdin.end.mock.calls[nth][0]); + } - // 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; + describe('dispatching', () => { + test('does not spawn anything at construction time', () => { + // Constructing a sink must be free of side effects: `startTelemetry` builds one against the + // real production endpoint even in unit tests. + sink(); + + expect(spawn).not.toHaveBeenCalled(); }); - return mockRequest; - } + test('does not spawn when there are no events', async () => { + await sink().flush(); - 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 }); + expect(spawn).not.toHaveBeenCalled(); + }); - // WHEN - await client.emit(testEvent); - await client.flush(); + test('spawns a detached sender and pipes the payload to it', async () => { + const testEvent = createTestEvent('INVOKE', { foo: 'bar' }); + const client = sink(); + + await client.emit(testEvent); + await client.flush(); + + expect(spawn).toHaveBeenCalledTimes(1); + expect(spawn).toHaveBeenCalledWith(process.execPath, [BIN_CDK], expect.objectContaining({ + detached: true, + stdio: ['pipe', 'ignore', 'ignore'], + windowsHide: true, + shell: false, + cwd: os.tmpdir(), + })); + + expect(pipedPayload()).toEqual({ + endpoint: 'https://example.com/telemetry', + body: { events: [testEvent] }, + }); + }); - // 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('does not impose the parent\'s exit budget on the child', async () => { + // The 500ms per-attempt timeout the synchronous POST used was there to protect the user's + // prompt. Nothing waits on the sender now, so forwarding it would only cut off slow (and + // especially proxied) deliveries -- the sender picks its own budget. + const client = sink(); + await client.emit(createTestEvent('INVOKE')); + await client.flush(); - test('silently catches request errors', async () => { - // GIVEN - const mockRequest = setupMockRequest(); - const testEvent = createTestEvent('INVOKE'); - const client = new EndpointTelemetrySink({ endpoint: 'https://example.com/telemetry', ioHost }); + expect(pipedPayload()).not.toHaveProperty('timeoutMs'); + }); - mockRequest.on.mockImplementation((event, callback) => { - if (event === 'error') { - callback(new Error('Network error')); - } - return mockRequest; + test('marks the child as the sender and lets it outlive us', async () => { + const client = sink(); + await client.emit(createTestEvent('INVOKE')); + await client.flush(); + + const options = (spawn as jest.Mock).mock.calls[0][2]; + expect(options.env.CDK_TELEMETRY_SENDER).toBe('1'); + expect(child.unref).toHaveBeenCalledTimes(1); + // A spawn failure must not surface as an unhandled 'error' event. + expect(child.on).toHaveBeenCalledWith('error', expect.any(Function)); + expect(child.stdin.on).toHaveBeenCalledWith('error', expect.any(Function)); }); - await client.emit(testEvent); + test('forwards the proxy and CA configuration the child cannot rediscover', async () => { + const client = sink({ proxyUrl: 'http://corp:8080', caCert: '-----BEGIN CERTIFICATE-----\nxx\n' }); + await client.emit(createTestEvent('INVOKE')); + await client.flush(); - // THEN - await expect(client.flush()).resolves.not.toThrow(); - }); + expect(pipedPayload()).toMatchObject({ + proxyUrl: 'http://corp:8080', + ca: '-----BEGIN CERTIFICATE-----\nxx\n', + }); + }); - 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 }); + test('batches multiple events into a single sender', async () => { + const testEvent1 = createTestEvent('INVOKE', { foo: 'bar' }); + const testEvent2 = createTestEvent('INVOKE', { foo: 'bazoo' }); + const client = sink(); - // WHEN - await client.emit(testEvent1); - await client.emit(testEvent2); - await client.flush(); + 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); - }); + expect(spawn).toHaveBeenCalledTimes(1); + expect(pipedPayload().body).toEqual({ events: [testEvent1, testEvent2] }); + }); - 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 }); + test('successful dispatch clears the events cache', async () => { + const testEvent1 = createTestEvent('INVOKE', { foo: 'bar' }); + const testEvent2 = createTestEvent('INVOKE', { foo: 'bazoo' }); + const client = sink(); - // WHEN - await client.emit(testEvent1); - await client.flush(); - await client.emit(testEvent2); - await client.flush(); + 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()); + expect(spawn).toHaveBeenCalledTimes(2); + expect(pipedPayload(0).body).toEqual({ events: [testEvent1] }); + expect(pipedPayload(1).body).toEqual({ events: [testEvent2] }); + }); }); - 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); + describe('back-pressure guard', () => { + test('drops a payload too large to hand over without blocking our own exit', async () => { + const traceSpy = jest.fn(); + jest.spyOn(IoHelper, 'fromActionAwareIoHost').mockReturnValue({ defaults: { trace: traceSpy } } as any); + + const client = sink(); + // ~200KB of events, comfortably past the 64KB guard. + for (let i = 0; i < 200; i++) { + await client.emit(createTestEvent('INVOKE', { padding: 'x'.repeat(1000) })); } - return mockRequest; - }).mockImplementation((_, callback) => { - if (callback) { - setTimeout(() => callback({ - statusCode: 200, - statusMessage: 'Success', - }), 0); + + await client.flush(); + + expect(spawn).not.toHaveBeenCalled(); + expect(traceSpy).toHaveBeenCalledWith(expect.stringContaining('Telemetry dropped')); + + // The batch is undeliverable, so it must be discarded rather than grown forever. + await client.emit(createTestEvent('INVOKE')); + await client.flush(); + expect(spawn).toHaveBeenCalledTimes(1); + expect(pipedPayload().body.events).toHaveLength(1); + }); + + test('a normal batch is nowhere near the guard', async () => { + const client = sink(); + for (let i = 0; i < 3; i++) { + await client.emit(createTestEvent('INVOKE')); } - return mockRequest; + + await client.flush(); + + expect(spawn).toHaveBeenCalledTimes(1); + expect(Buffer.byteLength(child.stdin.end.mock.calls[0][0])).toBeLessThan(65_536); }); + }); - const testEvent1 = createTestEvent('INVOKE', { foo: 'bar' }); - const testEvent2 = createTestEvent('INVOKE', { foo: 'bazoo' }); - const client = new EndpointTelemetrySink({ endpoint: 'https://example.com/telemetry', ioHost }); + describe('failure handling', () => { + test('dispatches without first probing the network', async () => { + // Any reachability probe would itself be a network call on the CLI's exit path, which is what + // this sink exists to avoid. Offline machines just spawn a child that fails and exits. + const client = sink(); + await client.emit(createTestEvent('INVOKE')); + await client.flush(); - // WHEN - await client.emit(testEvent1); + expect(spawn).toHaveBeenCalledTimes(1); + }); - // mocked to fail - await client.flush(); + test('skips when the CLI entrypoint could not be located', async () => { + const traceSpy = jest.fn(); + jest.spyOn(IoHelper, 'fromActionAwareIoHost').mockReturnValue({ defaults: { trace: traceSpy } } as any); + + const client = sink({ binCdkPath: undefined }); + await client.emit(createTestEvent('INVOKE')); + await client.flush(); + + expect(spawn).not.toHaveBeenCalled(); + expect(traceSpy).toHaveBeenCalledWith(expect.stringContaining('unable to locate the CLI entrypoint')); + }); + + test('swallows a spawn failure, traces it, and retains the events', async () => { + const traceSpy = jest.fn(); + jest.spyOn(IoHelper, 'fromActionAwareIoHost').mockReturnValue({ defaults: { trace: traceSpy } } as any); + (spawn as jest.Mock).mockImplementation(() => { + throw new Error('EMFILE'); + }); + + const client = sink(); + await client.emit(createTestEvent('INVOKE')); + + await expect(client.flush()).resolves.not.toThrow(); + expect(traceSpy).toHaveBeenCalledWith( + expect.stringContaining('Telemetry Error: spawning sender for POST example.com/telemetry'), + ); + + // Retained for a retry. + (spawn as jest.Mock).mockReturnValue(child); + await client.flush(); + expect(child.stdin.end).toHaveBeenCalledTimes(1); + }); - await client.emit(testEvent2); + test('rejects a malformed endpoint at construction', () => { + expect(() => sink({ endpoint: 'not-a-url' })).toThrow(); + }); + }); - // mocked to succeed + test('reports a successful hand-off on the trace channel', async () => { + const traceSpy = jest.fn(); + jest.spyOn(IoHelper, 'fromActionAwareIoHost').mockReturnValue({ defaults: { trace: traceSpy } } as any); + + const client = sink(); + await client.emit(createTestEvent('INVOKE')); 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()); + // Integration tests match on the 'Telemetry dispatched' prefix, so it must survive refactors. + expect(traceSpy).toHaveBeenCalledWith(expect.stringContaining('Telemetry dispatched')); + expect(traceSpy).toHaveBeenCalledWith(expect.stringMatching(/^Telemetry dispatched \(pid 4242, \d+ bytes\)$/)); }); 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 client = sink(); 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/funnel.test.ts b/packages/aws-cdk/test/cli/telemetry/sink/funnel.test.ts index f07c43d62..20415a2e7 100644 --- a/packages/aws-cdk/test/cli/telemetry/sink/funnel.test.ts +++ b/packages/aws-cdk/test/cli/telemetry/sink/funnel.test.ts @@ -1,37 +1,38 @@ -import * as https from 'https'; +import { spawn } from 'node:child_process'; 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'; -// Mock the https module -jest.mock('https', () => ({ - request: jest.fn(), +// The endpoint sink hands the payload to a detached child process rather than making the request +// itself, so this is what has to be intercepted. +jest.mock('node:child_process', () => ({ + spawn: jest.fn(), })); -// Mock NetworkDetector -jest.mock('../../../../lib/api/network-detector', () => ({ - NetworkDetector: { - hasConnectivity: jest.fn(), - }, -})); +const BIN_CDK = '/fake/pkg/bin/cdk'; describe('Funnel', () => { let tempDir: string; let logFilePath: string; let ioHost: CliIoHost; + let child: { pid: number; on: jest.Mock; unref: jest.Mock; stdin: { on: jest.Mock; end: jest.Mock } }; beforeEach(() => { jest.resetAllMocks(); - // Mock NetworkDetector to return true by default for all tests - (NetworkDetector.hasConnectivity as jest.Mock).mockResolvedValue(true); + child = { + pid: 4242, + on: jest.fn(), + unref: jest.fn(), + stdin: { on: jest.fn(), end: jest.fn() }, + }; + (spawn as jest.Mock).mockReturnValue(child); // Create a fresh temp directory for each test tempDir = path.join(os.tmpdir(), `telemetry-test-${Date.now()}-${Math.random().toString(36).substring(2, 9)}`); @@ -51,33 +52,6 @@ describe('Funnel', () => { 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; - } - describe('File and Endpoint', () => { let fileSink: FileTelemetrySink; let endpointSink: EndpointTelemetrySink; @@ -95,9 +69,16 @@ describe('Funnel', () => { jest.spyOn(IoHelper, 'fromActionAwareIoHost').mockReturnValue(mockIoHelper as any); fileSink = new FileTelemetrySink({ ioHost, logFilePath }); - endpointSink = new EndpointTelemetrySink({ ioHost, endpoint: 'https://example.com/telemetry' }); + endpointSink = new EndpointTelemetrySink({ ioHost, endpoint: 'https://example.com/telemetry', binCdkPath: BIN_CDK }); }); + /** + * The JSON that was piped to the detached sender on the Nth spawn. + */ + function pipedPayload(nth = 0) { + return JSON.parse(child.stdin.end.mock.calls[nth][0]); + } + test('saves data to a file', async () => { // GIVEN const testEvent = createTestEvent('INVOKE', { context: { foo: true } }); @@ -107,14 +88,16 @@ describe('Funnel', () => { await client.emit(testEvent); // THEN + // The file sink is deliberately still synchronous: the data must be on disk as soon as + // `emit` resolves, because `--telemetry-file` consumers read it immediately after the CLI + // exits. 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 () => { + test('dispatches the batch to a detached sender', async () => { // GIVEN - const mockRequest = setupMockRequest(); const testEvent = createTestEvent('INVOKE', { foo: 'bar' }); const client = new Funnel({ sinks: [fileSink, endpointSink] }); @@ -123,33 +106,26 @@ describe('Funnel', () => { 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); + expect(spawn).toHaveBeenCalledTimes(1); + expect(spawn).toHaveBeenCalledWith(process.execPath, [BIN_CDK], expect.objectContaining({ + detached: true, + stdio: ['pipe', 'ignore', 'ignore'], + })); + expect(pipedPayload()).toEqual({ + endpoint: 'https://example.com/telemetry', + body: { events: [testEvent] }, + }); }); 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' }); + const testEndpointSink = new EndpointTelemetrySink({ ioHost, endpoint: 'https://example.com/telemetry', binCdkPath: BIN_CDK }); new Funnel({ sinks: [fileSink, testEndpointSink] }); // Reset the spy call count since the constructor might have called flush @@ -177,31 +153,10 @@ describe('Funnel', () => { }); 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; - }); + // GIVEN a first dispatch that cannot be handed off, and a second one that can + (spawn as jest.Mock).mockImplementationOnce(() => { + throw new Error('EAGAIN'); + }).mockImplementation(() => child); const testEvent1 = createTestEvent('INVOKE', { foo: 'bar' }); const testEvent2 = createTestEvent('INVOKE', { foo: 'bazoo' }); @@ -218,35 +173,10 @@ describe('Funnel', () => { // 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()); + // THEN both events are still delivered together on the retry + expect(spawn).toHaveBeenCalledTimes(2); + expect(child.stdin.end).toHaveBeenCalledTimes(1); + expect(pipedPayload().body).toEqual({ events: [testEvent1, testEvent2] }); }); test('handles errors gracefully and logs to trace without throwing', async () => { @@ -255,20 +185,19 @@ describe('Funnel', () => { const client = new Funnel({ sinks: [fileSink, endpointSink] }); - // Mock https.request to throw an error - (https.request as jest.Mock).mockImplementation(() => { - throw new Error('Network error'); + // Spawning the sender fails + (spawn as jest.Mock).mockImplementation(() => { + throw new Error('Spawn error'); }); await client.emit(testEvent); - // WHEN & THEN - flush should not throw even when https.request fails + // WHEN & THEN - flush should not throw even when spawning fails await client.flush(); - // Verify that the error was lt - // logged to trace + // Verify that the error was logged to trace expect(traceSpy).toHaveBeenCalledWith( - expect.stringContaining('Telemetry Error: POST example.com/telemetry:'), + expect.stringContaining('Telemetry Error: spawning sender for POST example.com/telemetry'), ); }); 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..d39b1e52a --- /dev/null +++ b/packages/aws-cdk/test/cli/telemetry/test-tls.ts @@ -0,0 +1,99 @@ +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; + + /** + * 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; +} + +/** + * 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. + * + * 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-')); + try { + 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'), + serverCert: fs.readFileSync(file('server.crt'), 'utf-8'), + serverKey: fs.readFileSync(file('server.key'), 'utf-8'), + }; + } finally { + fs.rmSync(dir, { recursive: true, force: true }); + } +}