From 6cb8e9e38540b5e2954a6304b61b848fca70909e Mon Sep 17 00:00:00 2001 From: yummybomb <19238148+yummybomb@users.noreply.github.com> Date: Mon, 20 Jul 2026 20:59:38 +0000 Subject: [PATCH 1/6] feat: add complete audit log downloads --- src/index.ts | 9 ++ src/lib/audit-log-download.ts | 184 +++++++++++++++++++++++++++ src/resources/audit-logs.ts | 39 ++++++ tests/lib/audit-log-download.test.ts | 140 ++++++++++++++++++++ 4 files changed, 372 insertions(+) create mode 100644 src/lib/audit-log-download.ts create mode 100644 tests/lib/audit-log-download.test.ts diff --git a/src/index.ts b/src/index.ts index 81aa3351..03e12f5c 100644 --- a/src/index.ts +++ b/src/index.ts @@ -6,6 +6,15 @@ export { type Uploadable, toFile } from './core/uploads'; export { APIPromise } from './core/api-promise'; export { Kernel, type ClientOptions } from './client'; export { type BrowserFetchInit } from './lib/browser-fetch'; +export { + AuditLogDownloadError, + type AuditLogDownloadDestination, + type AuditLogDownloadOptions, + type AuditLogDownloadParams, + type AuditLogDownloadProgress, + type AuditLogDownloadResult, + type AuditLogDownloadWriteResult, +} from './lib/audit-log-download'; export { BrowserRouteCache, type BrowserRoute } from './lib/browser-routing'; export { PagePromise } from './core/pagination'; export { diff --git a/src/lib/audit-log-download.ts b/src/lib/audit-log-download.ts new file mode 100644 index 00000000..90124239 --- /dev/null +++ b/src/lib/audit-log-download.ts @@ -0,0 +1,184 @@ +import { APIConnectionError, APIError, APIUserAbortError, KernelError } from '../core/error'; +import type { RequestOptions } from '../internal/request-options'; +import type { AuditLogExportChunkParams } from '../resources/audit-logs'; + +const DOWNLOAD_ATTEMPTS = 7; +const MAX_RETRY_DELAY_MS = 8_000; + +export class AuditLogDownloadError extends KernelError {} + +export type AuditLogDownloadParams = Omit; + +export interface AuditLogDownloadResult { + bytesWritten: number; + chunks: number; + rows: number; +} + +export interface AuditLogDownloadProgress extends AuditLogDownloadResult { + chunkRows: number; +} + +export type AuditLogDownloadWriteResult = void | number | { bytesWritten: number }; + +export interface AuditLogDownloadDestination { + write(chunk: Uint8Array): AuditLogDownloadWriteResult | Promise; +} + +export interface AuditLogDownloadOptions + extends Omit< + RequestOptions, + 'method' | 'path' | 'query' | 'body' | 'maxRetries' | 'stream' | '__binaryResponse' | '__streamClass' + > { + onProgress?(progress: AuditLogDownloadProgress): void | Promise; +} + +type FetchChunk = (query: AuditLogExportChunkParams, options?: RequestOptions) => Promise; + +export async function downloadAuditLogs( + fetchChunk: FetchChunk, + query: AuditLogDownloadParams, + destination: AuditLogDownloadDestination, + options: AuditLogDownloadOptions = {}, +): Promise { + if (!destination || typeof destination.write !== 'function') { + throw new TypeError('audit log download destination must provide write()'); + } + + const { onProgress, ...requestOptions } = options; + let cursor: string | undefined; + const result: AuditLogDownloadResult = { bytesWritten: 0, chunks: 0, rows: 0 }; + + while (true) { + const chunk = await fetchVerifiedChunk(fetchChunk, cursor ? { ...query, cursor } : query, { + ...requestOptions, + maxRetries: 0, + }); + const { nextCursor, hasMore, rows } = parseChunkHeaders(chunk.headers, cursor); + await writeChunk(destination, chunk.body); + + cursor = nextCursor; + result.bytesWritten += chunk.body.byteLength; + result.chunks += 1; + result.rows += rows; + if (onProgress) { + await onProgress({ ...result, chunkRows: rows }); + } + if (!hasMore) { + return result; + } + } +} + +async function fetchVerifiedChunk( + fetchChunk: FetchChunk, + query: AuditLogExportChunkParams, + options: RequestOptions, +): Promise<{ body: Uint8Array; headers: Headers }> { + for (let attempt = 1; ; attempt += 1) { + try { + const response = await fetchChunk(query, options); + const body = new Uint8Array(await response.arrayBuffer()); + const expected = response.headers.get('x-content-sha256'); + if (!expected) { + throw new AuditLogDownloadError('response missing X-Content-Sha256 header'); + } + const actual = await sha256Hex(body); + if (actual !== expected) { + throw new AuditLogDownloadError( + `audit log chunk checksum mismatch (got ${actual}, want ${expected})`, + ); + } + return { body, headers: response.headers }; + } catch (error) { + if (attempt === DOWNLOAD_ATTEMPTS || !isRetryable(error, options.signal)) { + throw error; + } + await retryDelay(attempt, options.signal); + } + } +} + +function parseChunkHeaders( + headers: Headers, + currentCursor: string | undefined, +): { rows: number; nextCursor: string | undefined; hasMore: boolean } { + const hasMoreValue = headers.get('x-has-more'); + if (hasMoreValue !== 'true' && hasMoreValue !== 'false') { + throw new AuditLogDownloadError('response missing or invalid X-Has-More header'); + } + const hasMore = hasMoreValue === 'true'; + + const rowCount = headers.get('x-row-count'); + const rows = rowCount === null ? NaN : Number(rowCount); + if (!Number.isSafeInteger(rows) || rows < 0) { + throw new AuditLogDownloadError('response missing or invalid X-Row-Count header'); + } + + const nextCursor = headers.get('x-next-cursor') || undefined; + if (hasMore && (!nextCursor || nextCursor === currentCursor)) { + throw new AuditLogDownloadError('response has invalid X-Next-Cursor header'); + } + if (!hasMore && nextCursor) { + throw new AuditLogDownloadError('response returned a cursor after the final chunk'); + } + return { rows, nextCursor, hasMore }; +} + +async function sha256Hex(body: Uint8Array): Promise { + const subtle = globalThis.crypto?.subtle ?? (await import('node:crypto')).webcrypto.subtle; + const digest = await subtle.digest('SHA-256', body); + return Array.from(new Uint8Array(digest), (byte) => byte.toString(16).padStart(2, '0')).join(''); +} + +function isRetryable(error: unknown, signal: AbortSignal | null | undefined): boolean { + if (signal?.aborted || error instanceof APIUserAbortError) { + return false; + } + if (error instanceof APIError && !(error instanceof APIConnectionError)) { + return error.status === 429 || (error.status !== undefined && error.status >= 500); + } + return true; +} + +async function retryDelay(attempt: number, signal: AbortSignal | null | undefined): Promise { + const delay = Math.min(1_000 * 2 ** (attempt - 1), MAX_RETRY_DELAY_MS); + await new Promise((resolve, reject) => { + if (signal?.aborted) { + reject(new APIUserAbortError()); + return; + } + const onAbort = () => { + clearTimeout(timer); + reject(new APIUserAbortError()); + }; + const timer = setTimeout(() => { + signal?.removeEventListener('abort', onAbort); + resolve(); + }, delay); + signal?.addEventListener('abort', onAbort, { once: true }); + }); +} + +async function writeChunk(destination: AuditLogDownloadDestination, body: Uint8Array): Promise { + let offset = 0; + while (offset < body.byteLength) { + const result = await destination.write(offset === 0 ? body : body.subarray(offset)); + if (typeof result === 'number') { + if (result <= 0 || result > body.byteLength - offset) { + throw new AuditLogDownloadError('audit log download destination performed a short write'); + } + offset += result; + continue; + } + if (result && typeof result === 'object' && 'bytesWritten' in result) { + const bytesWritten = (result as { bytesWritten: number }).bytesWritten; + if (bytesWritten <= 0 || bytesWritten > body.byteLength - offset) { + throw new AuditLogDownloadError('audit log download destination performed a short write'); + } + offset += bytesWritten; + continue; + } + return; + } +} diff --git a/src/resources/audit-logs.ts b/src/resources/audit-logs.ts index 9367dc9a..2100ac94 100644 --- a/src/resources/audit-logs.ts +++ b/src/resources/audit-logs.ts @@ -5,6 +5,22 @@ import { APIPromise } from '../core/api-promise'; import { PagePromise, PageTokenPagination, type PageTokenPaginationParams } from '../core/pagination'; import { buildHeaders } from '../internal/headers'; import { RequestOptions } from '../internal/request-options'; +import { + downloadAuditLogs, + type AuditLogDownloadDestination, + type AuditLogDownloadOptions, + type AuditLogDownloadParams, + type AuditLogDownloadProgress, + type AuditLogDownloadResult, +} from '../lib/audit-log-download'; + +export type { + AuditLogDownloadDestination, + AuditLogDownloadOptions, + AuditLogDownloadParams, + AuditLogDownloadProgress, + AuditLogDownloadResult, +} from '../lib/audit-log-download'; /** * Read audit log records for the authenticated organization. @@ -34,6 +50,24 @@ export class AuditLogs extends APIResource { __binaryResponse: true, }); } + + /** + * Download a complete audit log export to a writable destination. The SDK + * verifies every chunk and retries transient transfer failures. It does not + * close the destination. + */ + download( + query: AuditLogDownloadParams, + destination: AuditLogDownloadDestination, + options?: AuditLogDownloadOptions, + ): Promise { + return downloadAuditLogs( + (chunkQuery, chunkOptions) => this.exportChunk(chunkQuery, chunkOptions), + query, + destination, + options, + ); + } } export type AuditLogEntriesPageTokenPagination = PageTokenPagination; @@ -205,5 +239,10 @@ export declare namespace AuditLogs { type AuditLogEntriesPageTokenPagination as AuditLogEntriesPageTokenPagination, type AuditLogListParams as AuditLogListParams, type AuditLogExportChunkParams as AuditLogExportChunkParams, + type AuditLogDownloadDestination as AuditLogDownloadDestination, + type AuditLogDownloadOptions as AuditLogDownloadOptions, + type AuditLogDownloadParams as AuditLogDownloadParams, + type AuditLogDownloadProgress as AuditLogDownloadProgress, + type AuditLogDownloadResult as AuditLogDownloadResult, }; } diff --git a/tests/lib/audit-log-download.test.ts b/tests/lib/audit-log-download.test.ts new file mode 100644 index 00000000..63dbab69 --- /dev/null +++ b/tests/lib/audit-log-download.test.ts @@ -0,0 +1,140 @@ +import { createHash } from 'node:crypto'; +import { open, mkdtemp, readFile, rm } from 'node:fs/promises'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; + +import Kernel from '@onkernel/sdk'; + +const checksum = (body: string) => createHash('sha256').update(body).digest('hex'); + +const chunkResponse = (body: string, rows: number, hasMore: boolean, nextCursor?: string) => + new Response(body, { + headers: { + 'content-type': 'application/octet-stream', + 'x-content-sha256': checksum(body), + 'x-has-more': String(hasMore), + 'x-row-count': String(rows), + ...(nextCursor ? { 'x-next-cursor': nextCursor } : {}), + }, + }); + +const requestURL = (input: string | URL | Request) => + typeof input === 'string' ? input + : input instanceof URL ? input.toString() + : input.url; + +describe('audit log download', () => { + test('writes verified chunks and reports progress', async () => { + const cursors: Array = []; + const client = new Kernel({ + apiKey: 'test', + baseURL: 'https://api.example', + fetch: async (input) => { + const url = new URL(requestURL(input)); + cursors.push(url.searchParams.get('cursor')); + return cursors.length === 1 ? + chunkResponse('first', 2, true, 'next') + : chunkResponse('second', 1, false); + }, + }); + const chunks: Uint8Array[] = []; + const progress: Array<{ bytesWritten: number; chunks: number; rows: number; chunkRows: number }> = []; + + const result = await client.auditLogs.download( + { + start: '2026-06-01T00:00:00Z', + end: '2026-06-02T00:00:00Z', + format: 'jsonl.gz', + }, + { + write(chunk) { + chunks.push(chunk); + }, + }, + { + onProgress(update) { + progress.push(update); + }, + }, + ); + + expect(cursors).toEqual([null, 'next']); + expect(Buffer.concat(chunks).toString()).toBe('firstsecond'); + expect(result).toEqual({ bytesWritten: 11, chunks: 2, rows: 3 }); + expect(progress).toEqual([ + { bytesWritten: 5, chunks: 1, rows: 2, chunkRows: 2 }, + { bytesWritten: 11, chunks: 2, rows: 3, chunkRows: 1 }, + ]); + }); + + test('writes to a Node file handle', async () => { + const client = new Kernel({ + apiKey: 'test', + baseURL: 'https://api.example', + fetch: async () => chunkResponse('chunk', 1, false), + }); + const directory = await mkdtemp(join(tmpdir(), 'kernel-audit-logs-')); + const path = join(directory, 'audit-logs.jsonl.gz'); + + try { + const file = await open(path, 'w'); + try { + await client.auditLogs.download({ start: '2026-06-01T00:00:00Z', end: '2026-06-02T00:00:00Z' }, file); + } finally { + await file.close(); + } + expect(await readFile(path, 'utf8')).toBe('chunk'); + } finally { + await rm(directory, { recursive: true }); + } + }); + + test('rejects an invalid next cursor before writing', async () => { + const client = new Kernel({ + apiKey: 'test', + baseURL: 'https://api.example', + fetch: async () => chunkResponse('chunk', 1, true), + }); + const write = jest.fn(); + + await expect( + client.auditLogs.download({ start: '2026-06-01T00:00:00Z', end: '2026-06-02T00:00:00Z' }, { write }), + ).rejects.toThrow('response has invalid X-Next-Cursor header'); + expect(write).not.toHaveBeenCalled(); + }); + + test('retries a checksum mismatch without writing the bad chunk', async () => { + let attempts = 0; + const client = new Kernel({ + apiKey: 'test', + baseURL: 'https://api.example', + fetch: async () => { + attempts += 1; + if (attempts === 1) { + return new Response('bad', { + headers: { + 'x-content-sha256': checksum('good'), + 'x-has-more': 'false', + 'x-row-count': '1', + }, + }); + } + return chunkResponse('good', 1, false); + }, + }); + const chunks: Uint8Array[] = []; + + const download = client.auditLogs.download( + { start: '2026-06-01T00:00:00Z', end: '2026-06-02T00:00:00Z' }, + { + write(chunk) { + chunks.push(chunk); + }, + }, + ); + await download; + + expect(attempts).toBe(2); + expect(Buffer.concat(chunks).toString()).toBe('good'); + }); +}); From c1ea2a8dc70f8a57b74699bff659e1122542091e Mon Sep 17 00:00:00 2001 From: yummybomb <19238148+yummybomb@users.noreply.github.com> Date: Tue, 21 Jul 2026 14:44:45 +0000 Subject: [PATCH 2/6] Harden audit log downloads --- src/lib/audit-log-download.ts | 51 ++++++++++--------- src/resources/audit-logs.ts | 4 +- tests/lib/audit-log-download.test.ts | 74 ++++++++++++++++++++++++++++ 3 files changed, 104 insertions(+), 25 deletions(-) diff --git a/src/lib/audit-log-download.ts b/src/lib/audit-log-download.ts index 90124239..cbfdaad1 100644 --- a/src/lib/audit-log-download.ts +++ b/src/lib/audit-log-download.ts @@ -1,8 +1,8 @@ -import { APIConnectionError, APIError, APIUserAbortError, KernelError } from '../core/error'; +import { APIUserAbortError, KernelError } from '../core/error'; import type { RequestOptions } from '../internal/request-options'; import type { AuditLogExportChunkParams } from '../resources/audit-logs'; -const DOWNLOAD_ATTEMPTS = 7; +const DEFAULT_MAX_TRANSFER_RETRIES = 6; const MAX_RETRY_DELAY_MS = 8_000; export class AuditLogDownloadError extends KernelError {} @@ -28,9 +28,10 @@ export interface AuditLogDownloadDestination { export interface AuditLogDownloadOptions extends Omit< RequestOptions, - 'method' | 'path' | 'query' | 'body' | 'maxRetries' | 'stream' | '__binaryResponse' | '__streamClass' + 'method' | 'path' | 'query' | 'body' | 'stream' | '__binaryResponse' | '__streamClass' > { onProgress?(progress: AuditLogDownloadProgress): void | Promise; + maxTransferRetries?: number; } type FetchChunk = (query: AuditLogExportChunkParams, options?: RequestOptions) => Promise; @@ -45,16 +46,28 @@ export async function downloadAuditLogs( throw new TypeError('audit log download destination must provide write()'); } - const { onProgress, ...requestOptions } = options; + const { onProgress, maxTransferRetries = DEFAULT_MAX_TRANSFER_RETRIES, ...requestOptions } = options; + if (!Number.isInteger(maxTransferRetries) || maxTransferRetries < 0) { + throw new TypeError('maxTransferRetries must be a non-negative integer'); + } let cursor: string | undefined; const result: AuditLogDownloadResult = { bytesWritten: 0, chunks: 0, rows: 0 }; + const seenCursors = new Set(); while (true) { - const chunk = await fetchVerifiedChunk(fetchChunk, cursor ? { ...query, cursor } : query, { - ...requestOptions, - maxRetries: 0, - }); + const chunk = await fetchVerifiedChunk( + fetchChunk, + cursor ? { ...query, cursor } : query, + requestOptions, + maxTransferRetries, + ); const { nextCursor, hasMore, rows } = parseChunkHeaders(chunk.headers, cursor); + if (hasMore && nextCursor) { + if (seenCursors.has(nextCursor)) { + throw new AuditLogDownloadError('response repeated X-Next-Cursor header'); + } + seenCursors.add(nextCursor); + } await writeChunk(destination, chunk.body); cursor = nextCursor; @@ -74,10 +87,11 @@ async function fetchVerifiedChunk( fetchChunk: FetchChunk, query: AuditLogExportChunkParams, options: RequestOptions, + maxTransferRetries: number, ): Promise<{ body: Uint8Array; headers: Headers }> { - for (let attempt = 1; ; attempt += 1) { + for (let retries = 0; ; retries += 1) { + const response = await fetchChunk(query, options); try { - const response = await fetchChunk(query, options); const body = new Uint8Array(await response.arrayBuffer()); const expected = response.headers.get('x-content-sha256'); if (!expected) { @@ -91,10 +105,10 @@ async function fetchVerifiedChunk( } return { body, headers: response.headers }; } catch (error) { - if (attempt === DOWNLOAD_ATTEMPTS || !isRetryable(error, options.signal)) { + if (retries === maxTransferRetries || options.signal?.aborted || error instanceof APIUserAbortError) { throw error; } - await retryDelay(attempt, options.signal); + await retryDelay(retries + 1, options.signal); } } } @@ -126,21 +140,10 @@ function parseChunkHeaders( } async function sha256Hex(body: Uint8Array): Promise { - const subtle = globalThis.crypto?.subtle ?? (await import('node:crypto')).webcrypto.subtle; - const digest = await subtle.digest('SHA-256', body); + const digest = await globalThis.crypto.subtle.digest('SHA-256', body); return Array.from(new Uint8Array(digest), (byte) => byte.toString(16).padStart(2, '0')).join(''); } -function isRetryable(error: unknown, signal: AbortSignal | null | undefined): boolean { - if (signal?.aborted || error instanceof APIUserAbortError) { - return false; - } - if (error instanceof APIError && !(error instanceof APIConnectionError)) { - return error.status === 429 || (error.status !== undefined && error.status >= 500); - } - return true; -} - async function retryDelay(attempt: number, signal: AbortSignal | null | undefined): Promise { const delay = Math.min(1_000 * 2 ** (attempt - 1), MAX_RETRY_DELAY_MS); await new Promise((resolve, reject) => { diff --git a/src/resources/audit-logs.ts b/src/resources/audit-logs.ts index 2100ac94..3f8ca3d2 100644 --- a/src/resources/audit-logs.ts +++ b/src/resources/audit-logs.ts @@ -54,7 +54,9 @@ export class AuditLogs extends APIResource { /** * Download a complete audit log export to a writable destination. The SDK * verifies every chunk and retries transient transfer failures. It does not - * close the destination. + * close the destination. If the download fails, the destination may contain + * a partial export; use a temporary file and atomic rename when the completed + * export must be published atomically. */ download( query: AuditLogDownloadParams, diff --git a/tests/lib/audit-log-download.test.ts b/tests/lib/audit-log-download.test.ts index 63dbab69..869950f8 100644 --- a/tests/lib/audit-log-download.test.ts +++ b/tests/lib/audit-log-download.test.ts @@ -137,4 +137,78 @@ describe('audit log download', () => { expect(attempts).toBe(2); expect(Buffer.concat(chunks).toString()).toBe('good'); }); + + test('respects request retry options', async () => { + let attempts = 0; + const client = new Kernel({ + apiKey: 'test', + baseURL: 'https://api.example', + fetch: async () => { + attempts += 1; + return new Response(JSON.stringify({ message: 'temporary failure' }), { + status: 500, + headers: { 'content-type': 'application/json', 'x-should-retry': 'false' }, + }); + }, + }); + + await expect( + client.auditLogs.download( + { start: '2026-06-01T00:00:00Z', end: '2026-06-02T00:00:00Z' }, + { write() {} }, + { maxRetries: 0 }, + ), + ).rejects.toThrow('500'); + expect(attempts).toBe(1); + }); + + test('respects transfer retry options', async () => { + let attempts = 0; + const client = new Kernel({ + apiKey: 'test', + baseURL: 'https://api.example', + fetch: async () => { + attempts += 1; + return new Response('bad', { + headers: { + 'x-content-sha256': checksum('good'), + 'x-has-more': 'false', + 'x-row-count': '1', + }, + }); + }, + }); + + await expect( + client.auditLogs.download( + { start: '2026-06-01T00:00:00Z', end: '2026-06-02T00:00:00Z' }, + { write() {} }, + { maxTransferRetries: 0 }, + ), + ).rejects.toThrow('checksum mismatch'); + expect(attempts).toBe(1); + }); + + test('rejects a cursor cycle before writing duplicate data', async () => { + let attempts = 0; + const client = new Kernel({ + apiKey: 'test', + baseURL: 'https://api.example', + fetch: async () => { + attempts += 1; + if (attempts === 1) return chunkResponse('first', 1, true, 'a'); + if (attempts === 2) return chunkResponse('second', 1, true, 'b'); + return chunkResponse('duplicate', 1, true, 'a'); + }, + }); + const chunks: Uint8Array[] = []; + + await expect( + client.auditLogs.download( + { start: '2026-06-01T00:00:00Z', end: '2026-06-02T00:00:00Z' }, + { write: (chunk) => void chunks.push(chunk) }, + ), + ).rejects.toThrow('response repeated X-Next-Cursor header'); + expect(Buffer.concat(chunks).toString()).toBe('firstsecond'); + }); }); From 7ed06467c9d9996bd455415b9ba0075e90339bdb Mon Sep 17 00:00:00 2001 From: yummybomb <19238148+yummybomb@users.noreply.github.com> Date: Tue, 21 Jul 2026 15:31:47 +0000 Subject: [PATCH 3/6] Harden audit log download validation --- src/lib/audit-log-download.ts | 25 +++++++++++-------- tests/lib/audit-log-download.test.ts | 36 ++++++++++++++++++++++++++++ 2 files changed, 51 insertions(+), 10 deletions(-) diff --git a/src/lib/audit-log-download.ts b/src/lib/audit-log-download.ts index cbfdaad1..9d46d4d1 100644 --- a/src/lib/audit-log-download.ts +++ b/src/lib/audit-log-download.ts @@ -3,6 +3,7 @@ import type { RequestOptions } from '../internal/request-options'; import type { AuditLogExportChunkParams } from '../resources/audit-logs'; const DEFAULT_MAX_TRANSFER_RETRIES = 6; +const MAX_CHUNK_ROWS = 50_000; const MAX_RETRY_DELAY_MS = 8_000; export class AuditLogDownloadError extends KernelError {} @@ -124,8 +125,11 @@ function parseChunkHeaders( const hasMore = hasMoreValue === 'true'; const rowCount = headers.get('x-row-count'); - const rows = rowCount === null ? NaN : Number(rowCount); - if (!Number.isSafeInteger(rows) || rows < 0) { + if (rowCount === null || !/^[0-9]+$/.test(rowCount)) { + throw new AuditLogDownloadError('response missing or invalid X-Row-Count header'); + } + const rows = Number(rowCount); + if (!Number.isSafeInteger(rows) || rows > MAX_CHUNK_ROWS) { throw new AuditLogDownloadError('response missing or invalid X-Row-Count header'); } @@ -168,20 +172,21 @@ async function writeChunk(destination: AuditLogDownloadDestination, body: Uint8A while (offset < body.byteLength) { const result = await destination.write(offset === 0 ? body : body.subarray(offset)); if (typeof result === 'number') { - if (result <= 0 || result > body.byteLength - offset) { - throw new AuditLogDownloadError('audit log download destination performed a short write'); - } - offset += result; + offset += validateWriteCount(result, body.byteLength - offset); continue; } if (result && typeof result === 'object' && 'bytesWritten' in result) { const bytesWritten = (result as { bytesWritten: number }).bytesWritten; - if (bytesWritten <= 0 || bytesWritten > body.byteLength - offset) { - throw new AuditLogDownloadError('audit log download destination performed a short write'); - } - offset += bytesWritten; + offset += validateWriteCount(bytesWritten, body.byteLength - offset); continue; } return; } } + +function validateWriteCount(value: number, remaining: number): number { + if (!Number.isSafeInteger(value) || value <= 0 || value > remaining) { + throw new AuditLogDownloadError('audit log download destination performed a short write'); + } + return value; +} diff --git a/tests/lib/audit-log-download.test.ts b/tests/lib/audit-log-download.test.ts index 869950f8..b8b56871 100644 --- a/tests/lib/audit-log-download.test.ts +++ b/tests/lib/audit-log-download.test.ts @@ -103,6 +103,42 @@ describe('audit log download', () => { expect(write).not.toHaveBeenCalled(); }); + test.each(['', '1.0', '50001'])('rejects invalid row count %p', async (rowCount) => { + const client = new Kernel({ + apiKey: 'test', + baseURL: 'https://api.example', + fetch: async () => { + const response = chunkResponse('chunk', 1, false); + response.headers.set('x-row-count', rowCount); + return response; + }, + }); + const write = jest.fn(); + + await expect( + client.auditLogs.download({ start: '2026-06-01T00:00:00Z', end: '2026-06-02T00:00:00Z' }, { write }), + ).rejects.toThrow('response missing or invalid X-Row-Count header'); + expect(write).not.toHaveBeenCalled(); + }); + + test.each([NaN, 0.5, { bytesWritten: NaN }, { bytesWritten: 0.5 }])( + 'rejects invalid destination write result %p', + async (writeResult) => { + const client = new Kernel({ + apiKey: 'test', + baseURL: 'https://api.example', + fetch: async () => chunkResponse('chunk', 1, false), + }); + + await expect( + client.auditLogs.download( + { start: '2026-06-01T00:00:00Z', end: '2026-06-02T00:00:00Z' }, + { write: () => writeResult }, + ), + ).rejects.toThrow('audit log download destination performed a short write'); + }, + ); + test('retries a checksum mismatch without writing the bad chunk', async () => { let attempts = 0; const client = new Kernel({ From c1686bb580a839811c5a22aaff79e214190a60b9 Mon Sep 17 00:00:00 2001 From: yummybomb <19238148+yummybomb@users.noreply.github.com> Date: Tue, 21 Jul 2026 16:13:20 +0000 Subject: [PATCH 4/6] Handle audit download cancellation and timeouts --- src/lib/audit-log-download.ts | 44 +++++++++++++++++++-- src/resources/audit-logs.ts | 1 + tests/lib/audit-log-download.test.ts | 57 +++++++++++++++++++++++++++- 3 files changed, 98 insertions(+), 4 deletions(-) diff --git a/src/lib/audit-log-download.ts b/src/lib/audit-log-download.ts index 9d46d4d1..8bd39dc0 100644 --- a/src/lib/audit-log-download.ts +++ b/src/lib/audit-log-download.ts @@ -1,4 +1,4 @@ -import { APIUserAbortError, KernelError } from '../core/error'; +import { APIConnectionTimeoutError, APIUserAbortError, KernelError } from '../core/error'; import type { RequestOptions } from '../internal/request-options'; import type { AuditLogExportChunkParams } from '../resources/audit-logs'; @@ -41,6 +41,7 @@ export async function downloadAuditLogs( fetchChunk: FetchChunk, query: AuditLogDownloadParams, destination: AuditLogDownloadDestination, + defaultTimeout: number, options: AuditLogDownloadOptions = {}, ): Promise { if (!destination || typeof destination.write !== 'function') { @@ -51,6 +52,7 @@ export async function downloadAuditLogs( if (!Number.isInteger(maxTransferRetries) || maxTransferRetries < 0) { throw new TypeError('maxTransferRetries must be a non-negative integer'); } + const timeout = requestOptions.timeout ?? defaultTimeout; let cursor: string | undefined; const result: AuditLogDownloadResult = { bytesWritten: 0, chunks: 0, rows: 0 }; const seenCursors = new Set(); @@ -61,6 +63,7 @@ export async function downloadAuditLogs( cursor ? { ...query, cursor } : query, requestOptions, maxTransferRetries, + timeout, ); const { nextCursor, hasMore, rows } = parseChunkHeaders(chunk.headers, cursor); if (hasMore && nextCursor) { @@ -89,9 +92,30 @@ async function fetchVerifiedChunk( query: AuditLogExportChunkParams, options: RequestOptions, maxTransferRetries: number, + timeout: number, ): Promise<{ body: Uint8Array; headers: Headers }> { for (let retries = 0; ; retries += 1) { - const response = await fetchChunk(query, options); + const controller = new AbortController(); + const onAbort = () => controller.abort(); + if (options.signal?.aborted) { + controller.abort(); + } else { + options.signal?.addEventListener('abort', onAbort, { once: true }); + } + + let response: Response; + try { + response = await fetchChunk(query, { ...options, signal: controller.signal }); + } catch (error) { + options.signal?.removeEventListener('abort', onAbort); + throw error; + } + + let bodyTimedOut = false; + const timer = setTimeout(() => { + bodyTimedOut = true; + controller.abort(); + }, timeout); try { const body = new Uint8Array(await response.arrayBuffer()); const expected = response.headers.get('x-content-sha256'); @@ -99,6 +123,12 @@ async function fetchVerifiedChunk( throw new AuditLogDownloadError('response missing X-Content-Sha256 header'); } const actual = await sha256Hex(body); + if (bodyTimedOut) { + throw new APIConnectionTimeoutError(); + } + if (options.signal?.aborted) { + throw new APIUserAbortError(); + } if (actual !== expected) { throw new AuditLogDownloadError( `audit log chunk checksum mismatch (got ${actual}, want ${expected})`, @@ -106,10 +136,18 @@ async function fetchVerifiedChunk( } return { body, headers: response.headers }; } catch (error) { - if (retries === maxTransferRetries || options.signal?.aborted || error instanceof APIUserAbortError) { + if (bodyTimedOut) { + error = new APIConnectionTimeoutError(); + } else if (options.signal?.aborted && !(error instanceof APIUserAbortError)) { + error = new APIUserAbortError(); + } + if (retries === maxTransferRetries || error instanceof APIUserAbortError) { throw error; } await retryDelay(retries + 1, options.signal); + } finally { + clearTimeout(timer); + options.signal?.removeEventListener('abort', onAbort); } } } diff --git a/src/resources/audit-logs.ts b/src/resources/audit-logs.ts index 3f8ca3d2..66ca8ae8 100644 --- a/src/resources/audit-logs.ts +++ b/src/resources/audit-logs.ts @@ -67,6 +67,7 @@ export class AuditLogs extends APIResource { (chunkQuery, chunkOptions) => this.exportChunk(chunkQuery, chunkOptions), query, destination, + this._client.timeout, options, ); } diff --git a/tests/lib/audit-log-download.test.ts b/tests/lib/audit-log-download.test.ts index b8b56871..e62ab68e 100644 --- a/tests/lib/audit-log-download.test.ts +++ b/tests/lib/audit-log-download.test.ts @@ -3,7 +3,7 @@ import { open, mkdtemp, readFile, rm } from 'node:fs/promises'; import { tmpdir } from 'node:os'; import { join } from 'node:path'; -import Kernel from '@onkernel/sdk'; +import Kernel, { APIConnectionTimeoutError, APIUserAbortError } from '@onkernel/sdk'; const checksum = (body: string) => createHash('sha256').update(body).digest('hex'); @@ -23,6 +23,27 @@ const requestURL = (input: string | URL | Request) => : input instanceof URL ? input.toString() : input.url; +const stalledChunkResponse = (signal: AbortSignal) => + new Response( + new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode('partial')); + signal.addEventListener( + 'abort', + () => controller.error(new DOMException('This operation was aborted', 'AbortError')), + { once: true }, + ); + }, + }), + { + headers: { + 'x-content-sha256': checksum('partial'), + 'x-has-more': 'false', + 'x-row-count': '1', + }, + }, + ); + describe('audit log download', () => { test('writes verified chunks and reports progress', async () => { const cursors: Array = []; @@ -225,6 +246,40 @@ describe('audit log download', () => { expect(attempts).toBe(1); }); + test('times out a stalled response body', async () => { + const client = new Kernel({ + apiKey: 'test', + baseURL: 'https://api.example', + fetch: async (_input, init) => stalledChunkResponse(init?.signal as AbortSignal), + }); + + await expect( + client.auditLogs.download( + { start: '2026-06-01T00:00:00Z', end: '2026-06-02T00:00:00Z' }, + { write() {} }, + { timeout: 10, maxTransferRetries: 0 }, + ), + ).rejects.toBeInstanceOf(APIConnectionTimeoutError); + }); + + test('maps cancellation during a response body read to APIUserAbortError', async () => { + const controller = new AbortController(); + const client = new Kernel({ + apiKey: 'test', + baseURL: 'https://api.example', + fetch: async (_input, init) => stalledChunkResponse(init?.signal as AbortSignal), + }); + + const download = client.auditLogs.download( + { start: '2026-06-01T00:00:00Z', end: '2026-06-02T00:00:00Z' }, + { write() {} }, + { signal: controller.signal, maxTransferRetries: 0 }, + ); + setTimeout(() => controller.abort(), 0); + + await expect(download).rejects.toBeInstanceOf(APIUserAbortError); + }); + test('rejects a cursor cycle before writing duplicate data', async () => { let attempts = 0; const client = new Kernel({ From 0779ce9179537f39354ff9333a93c885ded2cc40 Mon Sep 17 00:00:00 2001 From: yummybomb <19238148+yummybomb@users.noreply.github.com> Date: Tue, 21 Jul 2026 16:23:15 +0000 Subject: [PATCH 5/6] Limit timeout to audit body reads --- src/lib/audit-log-download.ts | 16 ++++++++++------ tests/lib/audit-log-download.test.ts | 27 +++++++++++++++++++++++++++ 2 files changed, 37 insertions(+), 6 deletions(-) diff --git a/src/lib/audit-log-download.ts b/src/lib/audit-log-download.ts index 8bd39dc0..f556865b 100644 --- a/src/lib/audit-log-download.ts +++ b/src/lib/audit-log-download.ts @@ -118,14 +118,18 @@ async function fetchVerifiedChunk( }, timeout); try { const body = new Uint8Array(await response.arrayBuffer()); + clearTimeout(timer); + if (options.signal?.aborted) { + throw new APIUserAbortError(); + } + if (bodyTimedOut) { + throw new APIConnectionTimeoutError(); + } const expected = response.headers.get('x-content-sha256'); if (!expected) { throw new AuditLogDownloadError('response missing X-Content-Sha256 header'); } const actual = await sha256Hex(body); - if (bodyTimedOut) { - throw new APIConnectionTimeoutError(); - } if (options.signal?.aborted) { throw new APIUserAbortError(); } @@ -136,10 +140,10 @@ async function fetchVerifiedChunk( } return { body, headers: response.headers }; } catch (error) { - if (bodyTimedOut) { - error = new APIConnectionTimeoutError(); - } else if (options.signal?.aborted && !(error instanceof APIUserAbortError)) { + if (options.signal?.aborted && !(error instanceof APIUserAbortError)) { error = new APIUserAbortError(); + } else if (bodyTimedOut) { + error = new APIConnectionTimeoutError(); } if (retries === maxTransferRetries || error instanceof APIUserAbortError) { throw error; diff --git a/tests/lib/audit-log-download.test.ts b/tests/lib/audit-log-download.test.ts index e62ab68e..2692427b 100644 --- a/tests/lib/audit-log-download.test.ts +++ b/tests/lib/audit-log-download.test.ts @@ -262,6 +262,33 @@ describe('audit log download', () => { ).rejects.toBeInstanceOf(APIConnectionTimeoutError); }); + test('does not apply the body timeout to checksum verification', async () => { + const client = new Kernel({ + apiKey: 'test', + baseURL: 'https://api.example', + fetch: async () => chunkResponse('chunk', 1, false), + }); + const digest = globalThis.crypto.subtle.digest.bind(globalThis.crypto.subtle); + const digestSpy = jest + .spyOn(globalThis.crypto.subtle, 'digest') + .mockImplementation(async (algorithm, data) => { + await new Promise((resolve) => setTimeout(resolve, 20)); + return digest(algorithm, data); + }); + + try { + await expect( + client.auditLogs.download( + { start: '2026-06-01T00:00:00Z', end: '2026-06-02T00:00:00Z' }, + { write() {} }, + { timeout: 10, maxTransferRetries: 0 }, + ), + ).resolves.toEqual({ bytesWritten: 5, chunks: 1, rows: 1 }); + } finally { + digestSpy.mockRestore(); + } + }); + test('maps cancellation during a response body read to APIUserAbortError', async () => { const controller = new AbortController(); const client = new Kernel({ From ec64cbd0343b907db268c125f2a3a32e3541892b Mon Sep 17 00:00:00 2001 From: yummybomb <19238148+yummybomb@users.noreply.github.com> Date: Tue, 21 Jul 2026 18:27:20 +0000 Subject: [PATCH 6/6] Default audit log downloads to gzip --- src/lib/audit-log-download.ts | 2 +- src/resources/audit-logs.ts | 10 +++++----- tests/lib/audit-log-download.test.ts | 4 +++- 3 files changed, 9 insertions(+), 7 deletions(-) diff --git a/src/lib/audit-log-download.ts b/src/lib/audit-log-download.ts index f556865b..b92383d8 100644 --- a/src/lib/audit-log-download.ts +++ b/src/lib/audit-log-download.ts @@ -8,7 +8,7 @@ const MAX_RETRY_DELAY_MS = 8_000; export class AuditLogDownloadError extends KernelError {} -export type AuditLogDownloadParams = Omit; +export type AuditLogDownloadParams = Omit; export interface AuditLogDownloadResult { bytesWritten: number; diff --git a/src/resources/audit-logs.ts b/src/resources/audit-logs.ts index 66ca8ae8..ca7dd082 100644 --- a/src/resources/audit-logs.ts +++ b/src/resources/audit-logs.ts @@ -52,11 +52,11 @@ export class AuditLogs extends APIResource { } /** - * Download a complete audit log export to a writable destination. The SDK - * verifies every chunk and retries transient transfer failures. It does not - * close the destination. If the download fails, the destination may contain - * a partial export; use a temporary file and atomic rename when the completed - * export must be published atomically. + * Download a complete gzip-compressed JSON Lines audit log export to a writable + * destination. The SDK verifies every chunk and retries transient transfer + * failures. It does not close the destination. If the download fails, the + * destination may contain a partial export; use a temporary file and atomic + * rename when the completed export must be published atomically. */ download( query: AuditLogDownloadParams, diff --git a/tests/lib/audit-log-download.test.ts b/tests/lib/audit-log-download.test.ts index 2692427b..45946861 100644 --- a/tests/lib/audit-log-download.test.ts +++ b/tests/lib/audit-log-download.test.ts @@ -47,12 +47,14 @@ const stalledChunkResponse = (signal: AbortSignal) => describe('audit log download', () => { test('writes verified chunks and reports progress', async () => { const cursors: Array = []; + const formats: Array = []; const client = new Kernel({ apiKey: 'test', baseURL: 'https://api.example', fetch: async (input) => { const url = new URL(requestURL(input)); cursors.push(url.searchParams.get('cursor')); + formats.push(url.searchParams.get('format')); return cursors.length === 1 ? chunkResponse('first', 2, true, 'next') : chunkResponse('second', 1, false); @@ -65,7 +67,6 @@ describe('audit log download', () => { { start: '2026-06-01T00:00:00Z', end: '2026-06-02T00:00:00Z', - format: 'jsonl.gz', }, { write(chunk) { @@ -80,6 +81,7 @@ describe('audit log download', () => { ); expect(cursors).toEqual([null, 'next']); + expect(formats).toEqual([null, null]); expect(Buffer.concat(chunks).toString()).toBe('firstsecond'); expect(result).toEqual({ bytesWritten: 11, chunks: 2, rows: 3 }); expect(progress).toEqual([