diff --git a/src/network/BufferedNetworkBody.ts b/src/network/BufferedNetworkBody.ts new file mode 100644 index 0000000..e0f4f45 --- /dev/null +++ b/src/network/BufferedNetworkBody.ts @@ -0,0 +1,92 @@ +import { + assertNetworkOperationActive, + CAPABILITY_TRANSPORT_ERROR_CODES, + createCapabilityTransportError, + sanitizeCapabilityTransportError, + type NetworkResourceLabel, +} from "./NetworkErrors"; + +export const MAX_BUFFERED_NETWORK_RESPONSE_BYTES = 12 * 1024 * 1024; +export const MAX_BUFFERED_NETWORK_CHUNKS = 16_384; + +const TYPED_ARRAY_PROTOTYPE = Object.getPrototypeOf(Uint8Array.prototype) as object; +const TYPED_ARRAY_LENGTH_GETTER = Object.getOwnPropertyDescriptor( + TYPED_ARRAY_PROTOTYPE, + "length", +)?.get; +const TYPED_ARRAY_BYTE_LENGTH_GETTER = Object.getOwnPropertyDescriptor( + TYPED_ARRAY_PROTOTYPE, + "byteLength", +)?.get; + +function getByteChunkLength(value: unknown): number | undefined { + if ( + !ArrayBuffer.isView(value) || + !TYPED_ARRAY_LENGTH_GETTER || + !TYPED_ARRAY_BYTE_LENGTH_GETTER + ) { + return undefined; + } + try { + const length = Reflect.apply(TYPED_ARRAY_LENGTH_GETTER, value, []) as unknown; + const byteLength = Reflect.apply(TYPED_ARRAY_BYTE_LENGTH_GETTER, value, []) as unknown; + return Number.isSafeInteger(length) && length === byteLength + ? (byteLength as number) + : undefined; + } catch { + return undefined; + } +} + +function growBuffer( + buffer: Uint8Array, + required: number, + limit: number, +): Uint8Array { + if (buffer.byteLength >= required) return buffer; + let capacity = Math.max(1024, buffer.byteLength); + while (capacity < required) capacity = Math.min(limit, Math.max(required, capacity * 2)); + const grown = new Uint8Array(capacity); + grown.set(buffer); + return grown; +} + +export async function consumeBufferedNetworkBody( + body: AsyncIterable, + maxResponseBytes: number, + resourceLabel: NetworkResourceLabel, + signal: AbortSignal, +): Promise { + let buffer = new Uint8Array(0); + let totalBytes = 0; + let chunkCount = 0; + try { + for await (const chunk of body) { + assertNetworkOperationActive(signal, resourceLabel); + chunkCount += 1; + const chunkLength = getByteChunkLength(chunk); + if (chunkLength === undefined) { + throw createCapabilityTransportError( + CAPABILITY_TRANSPORT_ERROR_CODES.adapterResponseInvalid, + resourceLabel, + ); + } + if ( + chunkCount > MAX_BUFFERED_NETWORK_CHUNKS || + chunkLength > maxResponseBytes - totalBytes + ) { + throw createCapabilityTransportError( + CAPABILITY_TRANSPORT_ERROR_CODES.responseTooLarge, + resourceLabel, + ); + } + if (chunkLength === 0) continue; + buffer = growBuffer(buffer, totalBytes + chunkLength, maxResponseBytes); + buffer.set(chunk as unknown as ArrayLike, totalBytes); + totalBytes += chunkLength; + } + return buffer.subarray(0, totalBytes); + } catch (error) { + throw sanitizeCapabilityTransportError(error, resourceLabel); + } +} diff --git a/src/network/CapabilityScopedTransport.test.ts b/src/network/CapabilityScopedTransport.test.ts new file mode 100644 index 0000000..008df3b --- /dev/null +++ b/src/network/CapabilityScopedTransport.test.ts @@ -0,0 +1,794 @@ +import { describe, expect, it, vi } from "vitest"; +import { + CAPABILITY_TRANSPORT_ERROR_CODES, + CapabilityScopedTransport, + CapabilityTransportError, + createCapabilityScopedNetworkRuntime, + MAX_BUFFERED_NETWORK_CHUNKS, + MAX_NETWORK_REDIRECTS, + MAX_NETWORK_RESOLVED_ADDRESSES, + NETWORK_BUFFERED_RESOURCE_POLICIES, + type NetworkNameResolver, + type PinnedNetworkHopAdapter, + type PinnedNetworkHopRequest, + type PinnedNetworkHopResponse, +} from "./CapabilityScopedTransport"; +import { + NetworkCapabilityError, + createNetworkCapabilityAuthority, + type NetworkCapability, + type NetworkCapabilityScope, +} from "./NetworkCapability"; +import { NetworkDisposedError } from "./NetworkErrors"; +import { NetworkScheduler } from "./NetworkScheduler"; +import type { EpisodeHandle, FeedHandle } from "src/security/resourceHandles"; + +const feedId = `podnotes-feed-${"1a".repeat(32)}` as FeedHandle; +const episodeId = `podnotes-episode-${"2b".repeat(32)}` as EpisodeHandle; +const subscriptionScope = Object.freeze({ resourceKind: "subscription", feedId } as const); +const feedArtworkScope = Object.freeze({ resourceKind: "feed-artwork", feedId } as const); +const streamScope = Object.freeze({ + resourceKind: "episode-stream", + feedId, + episodeId, +} as const); +const encoder = new TextEncoder(); + +interface Deferred { + readonly promise: Promise; + resolve(value: T): void; + reject(error: unknown): void; +} + +function deferred(): Deferred { + let resolve!: (value: T) => void; + let reject!: (error: unknown) => void; + const promise = new Promise((resolvePromise, rejectPromise) => { + resolve = resolvePromise; + reject = rejectPromise; + }); + return { promise, resolve, reject }; +} + +async function* bytesBody(...chunks: Array): AsyncIterable { + for (const chunk of chunks) yield typeof chunk === "string" ? encoder.encode(chunk) : chunk; +} + +const publicNameResolver: NetworkNameResolver = () => ["93.184.216.34"]; + +function hopResponse( + request: PinnedNetworkHopRequest, + options: { + status?: number; + location?: string; + connectedAddress?: string; + body?: AsyncIterable; + close?: () => void | PromiseLike; + } = {}, +): PinnedNetworkHopResponse { + return { + status: options.status ?? 200, + body: options.body ?? bytesBody("ok"), + connectedAddress: options.connectedAddress ?? request.connectAddress, + close: options.close ?? (() => undefined), + ...(options.location === undefined ? {} : { location: options.location }), + }; +} + +function setup( + scope: Scope, + adapter: PinnedNetworkHopAdapter, + options: { + target?: string; + privateOrigins?: readonly string[]; + nameResolver?: NetworkNameResolver; + scheduler?: NetworkScheduler; + } = {}, +) { + const authority = createNetworkCapabilityAuthority(); + const capability = authority.issuer.issue( + scope, + options.target ?? "https://feeds.example.com/feed.xml", + options.privateOrigins, + ); + const scheduler = options.scheduler ?? new NetworkScheduler(); + const nameResolver = options.nameResolver ?? publicNameResolver; + const transport = new CapabilityScopedTransport( + authority.resolver, + nameResolver, + adapter, + scheduler, + ); + return { authority, capability, nameResolver, scheduler, transport }; +} + +function expectSafeTransportError( + error: unknown, + code: (typeof CAPABILITY_TRANSPORT_ERROR_CODES)[keyof typeof CAPABILITY_TRANSPORT_ERROR_CODES], + sentinel?: string, +): void { + expect(error).toBeInstanceOf(CapabilityTransportError); + expect((error as CapabilityTransportError).code).toBe(code); + expect((error as CapabilityTransportError).resourceLabel).toBe("podcast subscription"); + expect(Object.isFrozen(error)).toBe(true); + expect(error).not.toHaveProperty("cause"); + if (sentinel) { + expect(String(error)).not.toContain(sentinel); + expect(JSON.stringify(error)).not.toContain(sentinel); + } +} + +async function flushOperations(): Promise { + await Promise.resolve(); + await Promise.resolve(); + await Promise.resolve(); +} + +describe("CapabilityScopedTransport pinned request boundary", () => { + it("pins one resolved address and exposes only structured request data to the adapter", async () => { + const target = "HTTPS://Feeds.Example.COM:443/audio%2fpart?sig=TOP-SECRET%2BVALUE"; + const source = encoder.encode("payload"); + let observed: PinnedNetworkHopRequest | undefined; + const nameResolver = vi.fn(() => ["93.184.216.34"]); + const adapter = vi.fn((request) => { + observed = request; + return hopResponse(request, { body: bytesBody(source) }); + }); + const { transport, capability } = setup(subscriptionScope, adapter, { + target, + nameResolver, + }); + + const response = await transport.getBytes(capability, subscriptionScope); + + expect(nameResolver).toHaveBeenCalledWith("feeds.example.com", expect.any(AbortSignal)); + expect(observed).toEqual({ + protocol: "https:", + connectAddress: "93.184.216.34", + port: 443, + serverName: "feeds.example.com", + hostHeader: "feeds.example.com", + requestTarget: "/audio%2fpart?sig=TOP-SECRET%2BVALUE", + method: "GET", + credentials: "omit", + redirect: "manual", + signal: expect.any(AbortSignal), + }); + expect(observed).not.toHaveProperty("target"); + expect(Object.isFrozen(observed)).toBe(true); + expect(Object.keys(transport)).toEqual([]); + expect(Object.isFrozen(transport)).toBe(true); + expect(new TextDecoder().decode(response.bytes)).toBe("payload"); + expect(Object.isFrozen(response)).toBe(true); + expect(JSON.stringify(response)).not.toContain(target); + source.fill(0); + expect(new TextDecoder().decode(response.bytes)).toBe("payload"); + }); + + it("uses an IP literal directly, requires its exact private grant, and skips DNS", async () => { + const nameResolver = vi.fn(); + const adapter = vi.fn((request) => hopResponse(request)); + const target = "http://192.168.1.10:8080/feed"; + const blocked = setup(subscriptionScope, adapter, { target, nameResolver }); + const blockedError = await blocked.transport + .getBytes(blocked.capability, subscriptionScope) + .catch((caught: unknown) => caught); + expectSafeTransportError(blockedError, CAPABILITY_TRANSPORT_ERROR_CODES.targetRejected); + expect(adapter).not.toHaveBeenCalled(); + + const allowed = setup(subscriptionScope, adapter, { + target, + nameResolver, + privateOrigins: ["http://192.168.1.10:8080"], + }); + const response = await allowed.transport.getBytes(allowed.capability, subscriptionScope); + expect(response.status).toBe(200); + expect(new TextDecoder().decode(response.bytes)).toBe("ok"); + expect(nameResolver).not.toHaveBeenCalled(); + expect(adapter.mock.calls[0][0].connectAddress).toBe("192.168.1.10"); + expect(adapter.mock.calls[0][0]).not.toHaveProperty("serverName"); + }); + + it("selects only an authorized DNS answer and permits private DNS only by exact origin", async () => { + const addresses = ["192.168.1.20", "93.184.216.34"]; + const publicAdapter = vi.fn((request) => hopResponse(request)); + const publicSetup = setup(subscriptionScope, publicAdapter, { + nameResolver: () => addresses, + }); + await publicSetup.transport.getBytes(publicSetup.capability, subscriptionScope); + expect(publicAdapter.mock.calls[0][0].connectAddress).toBe("93.184.216.34"); + + const privateAdapter = vi.fn((request) => hopResponse(request)); + const privateSetup = setup(subscriptionScope, privateAdapter, { + target: "https://private.example/feed", + privateOrigins: ["https://private.example"], + nameResolver: () => addresses, + }); + await privateSetup.transport.getBytes(privateSetup.capability, subscriptionScope); + expect(privateAdapter.mock.calls[0][0].connectAddress).toBe("192.168.1.20"); + }); + + it("falls back to the next pinned DNS address under the same operation deadline", async () => { + const adapter = vi.fn((request) => { + if (request.connectAddress === "2001:4860::1") { + throw new Error("unreachable first address"); + } + return hopResponse(request); + }); + const { transport, capability } = setup(subscriptionScope, adapter, { + nameResolver: () => ["2001:4860::1", "93.184.216.34"], + }); + + await expect(transport.getBytes(capability, subscriptionScope)).resolves.toMatchObject({ + status: 200, + }); + expect(adapter.mock.calls.map(([request]) => request.connectAddress)).toEqual([ + "2001:4860::1", + "93.184.216.34", + ]); + }); + + it("verifies the socket's actual remote address, including mapped IPv4", async () => { + const mappedAdapter = vi.fn((request) => + hopResponse(request, { connectedAddress: "::ffff:93.184.216.34" }), + ); + const mapped = setup(subscriptionScope, mappedAdapter); + await expect( + mapped.transport.getBytes(mapped.capability, subscriptionScope), + ).resolves.toMatchObject({ + status: 200, + }); + + const close = vi.fn(); + const mismatchedAdapter = vi.fn((request) => + hopResponse(request, { connectedAddress: "169.254.169.254", close }), + ); + const mismatched = setup(subscriptionScope, mismatchedAdapter); + const error = await mismatched.transport + .getBytes(mismatched.capability, subscriptionScope) + .catch((caught: unknown) => caught); + expectSafeTransportError(error, CAPABILITY_TRANSPORT_ERROR_CODES.adapterResponseInvalid); + expect(close).toHaveBeenCalledOnce(); + }); + + it.each([ + ["empty", []], + ["hostname", ["private.example"]], + ["duplicate", ["8.8.8.8", "::ffff:8.8.8.8"]], + [ + "over-limit", + Array.from( + { length: MAX_NETWORK_RESOLVED_ADDRESSES + 1 }, + (_, index) => `8.8.8.${index + 1}`, + ), + ], + ] as const)("rejects a malformed %s DNS answer", async (_name, addresses) => { + const adapter = vi.fn(); + const { transport, capability } = setup(subscriptionScope, adapter, { + nameResolver: () => addresses, + }); + const error = await transport + .getBytes(capability, subscriptionScope) + .catch((caught: unknown) => caught); + expectSafeTransportError(error, CAPABILITY_TRANSPORT_ERROR_CODES.addressRejected); + expect(adapter).not.toHaveBeenCalled(); + }); + + it("sanitizes resolver failures and hostile array traps", async () => { + const sentinel = "https://private.example/feed?token=DNS-SECRET"; + const adapters = [ + () => { + throw new Error(sentinel); + }, + () => + new Proxy(["93.184.216.34"], { + ownKeys: () => { + throw new Error(sentinel); + }, + }), + ] satisfies NetworkNameResolver[]; + + for (const nameResolver of adapters) { + const adapter = vi.fn(); + const { transport, capability } = setup(subscriptionScope, adapter, { nameResolver }); + const error = await transport + .getBytes(capability, subscriptionScope) + .catch((caught: unknown) => caught); + expectSafeTransportError( + error, + CAPABILITY_TRANSPORT_ERROR_CODES.addressRejected, + sentinel, + ); + } + }); +}); + +describe("CapabilityScopedTransport redirects and bounded responses", () => { + it("resolves every redirect through a newly pinned hop", async () => { + const absolute = "HTTPS://CDN.Example.COM:443/audio%2fpart?sig=A%2BB"; + const requests: PinnedNetworkHopRequest[] = []; + const closes = [vi.fn(), vi.fn(), vi.fn()]; + const adapter = vi.fn((request) => { + requests.push(request); + if (requests.length === 1) { + return hopResponse(request, { status: 302, location: absolute, close: closes[0] }); + } + if (requests.length === 2) { + return hopResponse(request, { + status: 307, + location: "../final?token=still-private", + close: closes[1], + }); + } + return hopResponse(request, { body: bytesBody("done"), close: closes[2] }); + }); + const { transport, capability } = setup(subscriptionScope, adapter); + + const response = await transport.getBytes(capability, subscriptionScope); + + expect(requests.map(({ serverName }) => serverName)).toEqual([ + "feeds.example.com", + "cdn.example.com", + "cdn.example.com", + ]); + expect(requests.map(({ requestTarget }) => requestTarget)).toEqual([ + "/feed.xml", + "/audio%2fpart?sig=A%2BB", + "/final?token=still-private", + ]); + expect(new TextDecoder().decode(response.bytes)).toBe("done"); + expect(closes.every((close) => close.mock.calls.length === 1)).toBe(true); + }); + + it.each([ + ["https downgrade", "http://public.example/final"], + ["private redirect", "https://127.0.0.1/private"], + ["credential redirect", "https://user:pass@public.example/final"], + ["unsupported redirect", "file:///etc/passwd"], + ["fragment redirect", "#private-fragment"], + ["backslash redirect", "\\private.example\\feed"], + ] as const)("rejects a %s before opening the redirected hop", async (_name, location) => { + const close = vi.fn(); + const adapter = vi.fn((request) => + hopResponse(request, { status: 302, location, close }), + ); + const { transport, capability } = setup(subscriptionScope, adapter); + const error = await transport + .getBytes(capability, subscriptionScope) + .catch((caught: unknown) => caught); + expect(error).toBeInstanceOf(CapabilityTransportError); + expect(adapter).toHaveBeenCalledOnce(); + expect(close).toHaveBeenCalledOnce(); + }); + + it("allows a private redirect only through an exact pre-issued origin grant", async () => { + const requests: PinnedNetworkHopRequest[] = []; + const adapter = vi.fn((request) => { + requests.push(request); + return requests.length === 1 + ? hopResponse(request, { status: 302, location: "http://127.0.0.1:8080/feed" }) + : hopResponse(request); + }); + const { transport, capability } = setup(subscriptionScope, adapter, { + target: "http://public.example/feed", + privateOrigins: ["http://127.0.0.1:8080"], + }); + + await expect(transport.getBytes(capability, subscriptionScope)).resolves.toMatchObject({ + status: 200, + }); + expect(requests[1].connectAddress).toBe("127.0.0.1"); + }); + + it("rejects redirect loops and chains beyond the fixed limit", async () => { + const loopAdapter = vi.fn((request) => + hopResponse(request, { status: 302, location: "https://feeds.example.com/feed.xml" }), + ); + const loop = setup(subscriptionScope, loopAdapter); + const loopError = await loop.transport + .getBytes(loop.capability, subscriptionScope) + .catch((caught: unknown) => caught); + expectSafeTransportError(loopError, CAPABILITY_TRANSPORT_ERROR_CODES.redirectRejected); + expect(loopAdapter).toHaveBeenCalledOnce(); + + let hop = 0; + const chainAdapter = vi.fn((request) => { + hop += 1; + return hopResponse(request, { + status: 302, + location: `https://redirect-${hop}.example/feed`, + }); + }); + const chain = setup(subscriptionScope, chainAdapter); + const chainError = await chain.transport + .getBytes(chain.capability, subscriptionScope) + .catch((caught: unknown) => caught); + expectSafeTransportError(chainError, CAPABILITY_TRANSPORT_ERROR_CODES.redirectLimit); + expect(chainAdapter).toHaveBeenCalledTimes(MAX_NETWORK_REDIRECTS + 1); + }); + + it("requires the capability's fixed status policy and closes the hop", async () => { + const close = vi.fn(); + const adapter = vi.fn((request) => + hopResponse(request, { status: 404, body: bytesBody("secret"), close }), + ); + const { transport, capability } = setup(subscriptionScope, adapter); + const error = await transport + .getBytes(capability, subscriptionScope) + .catch((caught: unknown) => caught); + expectSafeTransportError(error, CAPABILITY_TRANSPORT_ERROR_CODES.statusRejected); + expect(close).toHaveBeenCalledOnce(); + }); + + it("enforces the capability's byte ceiling incrementally", async () => { + const limit = NETWORK_BUFFERED_RESOURCE_POLICIES.subscription.maxResponseBytes; + let iteratorClosed = false; + async function* oversizedBody(): AsyncIterable { + try { + yield new Uint8Array(limit); + yield new Uint8Array(1); + } finally { + iteratorClosed = true; + } + } + const close = vi.fn(); + const adapter = vi.fn((request) => + hopResponse(request, { body: oversizedBody(), close }), + ); + const { transport, capability } = setup(subscriptionScope, adapter); + const error = await transport + .getBytes(capability, subscriptionScope) + .catch((caught: unknown) => caught); + expectSafeTransportError(error, CAPABILITY_TRANSPORT_ERROR_CODES.responseTooLarge); + expect(iteratorClosed).toBe(true); + expect(close).toHaveBeenCalledOnce(); + }); + + it("bounds zero-length chunk work independently of byte count", async () => { + async function* excessiveChunks(): AsyncIterable { + for (let index = 0; index <= MAX_BUFFERED_NETWORK_CHUNKS; index += 1) { + yield new Uint8Array(0); + } + } + const adapter = vi.fn((request) => + hopResponse(request, { body: excessiveChunks() }), + ); + const { transport, capability } = setup(subscriptionScope, adapter); + const error = await transport + .getBytes(capability, subscriptionScope) + .catch((caught: unknown) => caught); + expectSafeTransportError(error, CAPABILITY_TRANSPORT_ERROR_CODES.responseTooLarge); + }); + + it("rejects non-byte views and sanitizes adapter, body, and close failures", async () => { + const sentinel = "https://user:pass@private.example/audio?token=TOP-SECRET"; + const cases: PinnedNetworkHopAdapter[] = [ + (request) => + hopResponse(request, { + body: bytesBody(new Uint16Array([1]) as unknown as Uint8Array), + }), + () => { + throw new Error(sentinel); + }, + (request) => + hopResponse(request, { + body: { + [Symbol.asyncIterator]: () => ({ + next: () => Promise.reject(new Error(sentinel)), + }), + }, + }), + (request) => + hopResponse(request, { + close: () => { + throw new Error(sentinel); + }, + }), + ]; + + for (const adapter of cases) { + const { transport, capability } = setup(subscriptionScope, adapter); + const error = await transport + .getBytes(capability, subscriptionScope) + .catch((caught: unknown) => caught); + expect(error).toBeInstanceOf(CapabilityTransportError); + expect(String(error)).not.toContain(sentinel); + expect(JSON.stringify(error)).not.toContain(sentinel); + } + }); + + it("closes malformed adapter responses without invoking accessor fields", async () => { + let statusRead = false; + const close = vi.fn(); + const adapter: PinnedNetworkHopAdapter = (request) => { + const response = { + body: bytesBody("ok"), + connectedAddress: request.connectAddress, + close, + } as Record; + Object.defineProperty(response, "status", { + enumerable: true, + get: () => { + statusRead = true; + return 200; + }, + }); + return response as unknown as PinnedNetworkHopResponse; + }; + const { transport, capability } = setup(subscriptionScope, adapter); + const error = await transport + .getBytes(capability, subscriptionScope) + .catch((caught: unknown) => caught); + expectSafeTransportError(error, CAPABILITY_TRANSPORT_ERROR_CODES.adapterResponseInvalid); + expect(statusRead).toBe(false); + expect(close).toHaveBeenCalledOnce(); + }); + + it("preserves the adapter response receiver when invoking close", async () => { + let responseObject: PinnedNetworkHopResponse | undefined; + let usedResponseReceiver = false; + const adapter: PinnedNetworkHopAdapter = (request) => { + responseObject = { + status: 200, + body: bytesBody("ok"), + connectedAddress: request.connectAddress, + close(): void { + usedResponseReceiver = this === responseObject; + }, + }; + return responseObject; + }; + const { transport, capability } = setup(subscriptionScope, adapter); + + await transport.getBytes(capability, subscriptionScope); + expect(usedResponseReceiver).toBe(true); + }); +}); + +describe("CapabilityScopedTransport policy, revocation, and lifecycle", () => { + it("derives immutable buffering and lane policy from the resource kind", async () => { + expect(Object.isFrozen(NETWORK_BUFFERED_RESOURCE_POLICIES)).toBe(true); + expect(Object.isFrozen(NETWORK_BUFFERED_RESOURCE_POLICIES.subscription)).toBe(true); + expect( + Object.isFrozen(NETWORK_BUFFERED_RESOURCE_POLICIES.subscription.acceptedStatuses), + ).toBe(true); + + const gate = deferred(); + const adapter = vi.fn(() => gate.promise); + const artwork = setup(feedArtworkScope, adapter); + const pending = artwork.transport.getBytes(artwork.capability, feedArtworkScope); + expect(artwork.transport.getLaneSnapshot("media")).toMatchObject({ active: 1 }); + expect(artwork.transport.getLaneSnapshot("metadata")).toMatchObject({ active: 0 }); + await flushOperations(); + const request = adapter.mock.calls[0][0]; + gate.resolve(hopResponse(request)); + await pending; + }); + + it("rejects buffered episode-stream use before DNS or adapter work", async () => { + const nameResolver = vi.fn(); + const adapter = vi.fn(); + const { transport, capability } = setup(streamScope, adapter, { nameResolver }); + const error = await transport + .getBytes(capability, streamScope) + .catch((caught: unknown) => caught); + expect(error).toBeInstanceOf(CapabilityTransportError); + expect((error as CapabilityTransportError).code).toBe( + CAPABILITY_TRANSPORT_ERROR_CODES.resourceRejected, + ); + expect(nameResolver).not.toHaveBeenCalled(); + expect(adapter).not.toHaveBeenCalled(); + }); + + it("revalidates a queued capability before network work starts", async () => { + const firstGate = deferred(); + let calls = 0; + const adapter = vi.fn((request) => { + calls += 1; + return calls === 1 ? firstGate.promise : hopResponse(request); + }); + const scheduler = new NetworkScheduler({ + laneLimits: { metadata: { maxActive: 1, maxQueued: 1 } }, + }); + const authority = createNetworkCapabilityAuthority(); + const firstCapability = authority.issuer.issue( + subscriptionScope, + "https://first.example/feed", + ); + const secondCapability = authority.issuer.issue( + subscriptionScope, + "https://second.example/feed", + ); + const transport = new CapabilityScopedTransport( + authority.resolver, + publicNameResolver, + adapter, + scheduler, + ); + const first = transport.getBytes(firstCapability, subscriptionScope); + const second = transport.getBytes(secondCapability, subscriptionScope); + expect(transport.getLaneSnapshot("metadata")).toMatchObject({ active: 1, queued: 1 }); + expect(authority.issuer.revoke(secondCapability)).toBe(true); + + await flushOperations(); + firstGate.resolve(hopResponse(adapter.mock.calls[0][0])); + await first; + await expect(second).rejects.toBeInstanceOf(NetworkCapabilityError); + expect(adapter).toHaveBeenCalledOnce(); + }); + + it("revocation aborts and closes already-active capability work", async () => { + const nextGate = deferred>(); + const close = vi.fn(() => nextGate.reject(new Error("closed"))); + const body: AsyncIterable = { + [Symbol.asyncIterator]: () => ({ next: () => nextGate.promise }), + }; + const adapter = vi.fn((request) => + hopResponse(request, { body, close }), + ); + const active = setup(subscriptionScope, adapter); + const pending = active.transport.getBytes(active.capability, subscriptionScope); + await flushOperations(); + + expect(active.authority.issuer.revoke(active.capability)).toBe(true); + expect(close).toHaveBeenCalledOnce(); + await expect(pending).rejects.toBeInstanceOf(CapabilityTransportError); + await flushOperations(); + expect(active.transport.getLaneSnapshot("metadata").active).toBe(0); + }); + + it("does not return buffered bytes when revocation occurs during asynchronous close", async () => { + const closeGate = deferred(); + const close = vi.fn(() => closeGate.promise); + const adapter = vi.fn((request) => + hopResponse(request, { body: bytesBody("complete"), close }), + ); + const active = setup(subscriptionScope, adapter); + const pending = active.transport.getBytes(active.capability, subscriptionScope); + await vi.waitFor(() => expect(close).toHaveBeenCalledOnce()); + + expect(active.authority.issuer.revoke(active.capability)).toBe(true); + closeGate.resolve(); + await expect(pending).rejects.toBeInstanceOf(CapabilityTransportError); + }); + + it("rejects forged tokens, cross-scope use, and forged resolver facets", async () => { + const adapter = vi.fn(); + const { authority, transport, capability, scheduler } = setup(subscriptionScope, adapter); + const forged = Object.freeze(Object.create(null)) as NetworkCapability; + await expect(transport.getBytes(forged, subscriptionScope)).rejects.toBeInstanceOf( + NetworkCapabilityError, + ); + await expect( + transport.getBytes( + capability as unknown as NetworkCapability, + streamScope, + ), + ).rejects.toBeInstanceOf(NetworkCapabilityError); + expect( + () => + new CapabilityScopedTransport( + { resolve: authority.resolver.resolve }, + publicNameResolver, + adapter, + scheduler, + ), + ).toThrow("authentic network capability resolver"); + expect(adapter).not.toHaveBeenCalled(); + }); + + it("requires every trusted composition dependency explicitly", () => { + const authority = createNetworkCapabilityAuthority(); + const scheduler = new NetworkScheduler(); + const adapter = vi.fn(); + expect( + () => + new CapabilityScopedTransport( + authority.resolver, + null as unknown as NetworkNameResolver, + adapter, + scheduler, + ), + ).toThrow(TypeError); + expect( + () => + new CapabilityScopedTransport( + authority.resolver, + publicNameResolver, + adapter, + undefined as unknown as NetworkScheduler, + ), + ).toThrow("shared network scheduler"); + }); + + it("does not export an executable raw pinned-hop client", async () => { + const hopBoundary = await import("./PinnedNetworkHop"); + const { transport } = setup(subscriptionScope, vi.fn()); + + expect(hopBoundary).not.toHaveProperty("PinnedNetworkHopClient"); + expect(Object.getOwnPropertyNames(Object.getPrototypeOf(transport)).sort()).toEqual([ + "constructor", + "dispose", + "getBytes", + "getLaneSnapshot", + "getSnapshot", + "isDisposed", + ]); + }); + + it("composition runtime never distributes the raw-target resolver", async () => { + const adapter = vi.fn((request) => hopResponse(request)); + const runtime = createCapabilityScopedNetworkRuntime( + publicNameResolver, + adapter, + new NetworkScheduler(), + ); + const capability = runtime.issuer.issue( + subscriptionScope, + "https://feeds.example.com/feed.xml?token=hidden", + ); + + expect(Object.keys(runtime).sort()).toEqual(["dispose", "issuer", "transport"]); + expect(runtime).not.toHaveProperty("resolver"); + await expect( + runtime.transport.getBytes(capability, subscriptionScope), + ).resolves.toMatchObject({ + status: 200, + }); + runtime.dispose(); + expect(runtime.transport.isDisposed()).toBe(true); + }); + + it("aborts and closes active response work exactly once on disposal", async () => { + const nextGate = deferred>(); + const close = vi.fn(() => { + nextGate.reject(new Error("closed")); + }); + const body: AsyncIterable = { + [Symbol.asyncIterator]: () => ({ next: () => nextGate.promise }), + }; + let signal: AbortSignal | undefined; + const adapter = vi.fn((request) => { + signal = request.signal; + return hopResponse(request, { body, close }); + }); + const { transport, capability } = setup(subscriptionScope, adapter); + const pending = transport.getBytes(capability, subscriptionScope); + await flushOperations(); + + transport.dispose(); + expect(signal?.aborted).toBe(true); + expect(close).toHaveBeenCalledOnce(); + await expect(pending).rejects.toBeInstanceOf(NetworkDisposedError); + await flushOperations(); + expect(close).toHaveBeenCalledOnce(); + }); + + it("closes a late adapter response without consuming it after disposal", async () => { + const adapterGate = deferred(); + const close = vi.fn(); + let request: PinnedNetworkHopRequest | undefined; + let bodyStarted = false; + const body: AsyncIterable = { + [Symbol.asyncIterator]: () => { + bodyStarted = true; + return { next: () => Promise.resolve({ done: true, value: undefined }) }; + }, + }; + const adapter = vi.fn((nextRequest) => { + request = nextRequest; + return adapterGate.promise; + }); + const { transport, capability } = setup(subscriptionScope, adapter); + const pending = transport.getBytes(capability, subscriptionScope); + await flushOperations(); + transport.dispose(); + await expect(pending).rejects.toBeInstanceOf(NetworkDisposedError); + + if (!request) throw new Error("adapter request was not captured"); + adapterGate.resolve(hopResponse(request, { body, close })); + await flushOperations(); + expect(close).toHaveBeenCalledOnce(); + expect(bodyStarted).toBe(false); + }); +}); diff --git a/src/network/CapabilityScopedTransport.ts b/src/network/CapabilityScopedTransport.ts new file mode 100644 index 0000000..eccbbe1 --- /dev/null +++ b/src/network/CapabilityScopedTransport.ts @@ -0,0 +1,539 @@ +import { + MAX_BUFFERED_NETWORK_RESPONSE_BYTES, + consumeBufferedNetworkBody, +} from "./BufferedNetworkBody"; +import { + NetworkCapabilityError, + createNetworkCapabilityAuthority, + isNetworkCapabilityResolver, + type NetworkCapability, + type NetworkCapabilityIssuer, + type NetworkCapabilityResolution, + type NetworkCapabilityResolver, + type NetworkCapabilityScope, + type NetworkResourceKind, +} from "./NetworkCapability"; +import { + classifyResolvedAddress, + isResolvedAddressLiteral, + MAX_NETWORK_TARGET_BYTES, + parseTargetPolicyView, + type NetworkProtocol, + type TargetPolicyView, +} from "./LiteralTargetClassifier"; +import { + assertNetworkOperationActive, + CAPABILITY_TRANSPORT_ERROR_CODES, + createCapabilityTransportError, + createNetworkResourceLabel, + sanitizeCapabilityTransportError, + type CapabilityTransportError, + type NetworkResourceLabel, +} from "./NetworkErrors"; +import { + closeInvalidHopResponse, + manageHopResponse, + snapshotHopResponse, + snapshotResolvedAddresses, + targetRequestParts, + type NetworkNameResolver, + type PinnedNetworkHop, + type PinnedNetworkHopAdapter, +} from "./PinnedNetworkHop"; +import type { + NetworkLane, + NetworkLaneSnapshot, + NetworkScheduler, + NetworkSchedulerSession, + NetworkSchedulerSnapshot, +} from "./NetworkScheduler"; + +export { + MAX_BUFFERED_NETWORK_CHUNKS, + MAX_BUFFERED_NETWORK_RESPONSE_BYTES, +} from "./BufferedNetworkBody"; +export { + CAPABILITY_TRANSPORT_ERROR_CODES, + CapabilityTransportError, + type CapabilityTransportErrorCode, +} from "./NetworkErrors"; +export { + MAX_NETWORK_RESOLVED_ADDRESSES, + type NetworkNameResolver, + type PinnedNetworkHopAdapter, + type PinnedNetworkHopRequest, + type PinnedNetworkHopResponse, +} from "./PinnedNetworkHop"; + +export const MAX_NETWORK_REDIRECTS = 5; + +export interface BufferedNetworkResourcePolicy { + readonly lane: NetworkLane; + readonly acceptedStatuses: readonly number[]; + /** Total enqueue-to-completion deadline. */ + readonly timeoutMs: number; + readonly maxResponseBytes: number; +} + +function bufferedPolicy( + lane: NetworkLane, + timeoutMs: number, + maxResponseBytes: number, +): BufferedNetworkResourcePolicy { + return Object.freeze({ + lane, + acceptedStatuses: Object.freeze([200]), + timeoutMs, + maxResponseBytes, + }); +} + +/** + * Buffering, status, deadline, and lane policy is fixed by capability purpose. + * Episode streams deliberately have no buffered operation. + */ +export const NETWORK_BUFFERED_RESOURCE_POLICIES = Object.freeze({ + subscription: bufferedPolicy("metadata", 30_000, 8 * 1024 * 1024), + "feed-artwork": bufferedPolicy("media", 30_000, MAX_BUFFERED_NETWORK_RESPONSE_BYTES), + site: bufferedPolicy("metadata", 30_000, 4 * 1024 * 1024), + "episode-stream": null, + "episode-chapters": bufferedPolicy("metadata", 20_000, 4 * 1024 * 1024), + "episode-artwork": bufferedPolicy("media", 30_000, MAX_BUFFERED_NETWORK_RESPONSE_BYTES), + "episode-item-link": bufferedPolicy("metadata", 30_000, 4 * 1024 * 1024), +} as const satisfies Readonly>); + +export interface CapabilityBytesResponse { + readonly status: number; + readonly bytes: Uint8Array; +} + +type TransportOutcome = + | { readonly ok: true; readonly value: T } + | { + readonly ok: false; + readonly error: CapabilityTransportError | NetworkCapabilityError; + }; + +const REDIRECT_STATUSES = new Set([301, 302, 303, 307, 308]); + +function createCapabilityResolutionError(): NetworkCapabilityError { + return Object.freeze(new NetworkCapabilityError("resolve")); +} + +function evaluateTarget( + target: string, + privateOrigins: ReadonlySet, + resourceLabel: NetworkResourceLabel, + previousProtocol?: NetworkProtocol, +): TargetPolicyView { + const policy = parseTargetPolicyView(target); + if ( + !policy.ok || + policy.value.hasCredentials || + (policy.value.hostClassification === "non-public-or-special-literal" && + !privateOrigins.has(policy.value.normalizedOrigin)) + ) { + throw createCapabilityTransportError( + CAPABILITY_TRANSPORT_ERROR_CODES.targetRejected, + resourceLabel, + ); + } + if (previousProtocol === "https:" && policy.value.protocol === "http:") { + throw createCapabilityTransportError( + CAPABILITY_TRANSPORT_ERROR_CODES.redirectRejected, + resourceLabel, + ); + } + return policy.value; +} + +function hasUnsafeRelativeLocationCodePoint(value: string): boolean { + for (let index = 0; index < value.length; index += 1) { + const code = value.charCodeAt(index); + if (code <= 0x20 || (code >= 0x7f && code <= 0x9f)) return true; + if (code >= 0xd800 && code <= 0xdbff) { + const next = value.charCodeAt(index + 1); + if (next < 0xdc00 || next > 0xdfff) return true; + index += 1; + continue; + } + if (code >= 0xdc00 && code <= 0xdfff) return true; + if ( + code === 0x061c || + (code >= 0x200b && code <= 0x200f) || + (code >= 0x2028 && code <= 0x202e) || + (code >= 0x2066 && code <= 0x2069) || + code === 0xfeff + ) { + return true; + } + } + return false; +} + +function resolveRedirectTarget( + location: string | undefined, + currentTarget: string, + resourceLabel: NetworkResourceLabel, +): string { + if ( + location === undefined || + location.length === 0 || + location.length > MAX_NETWORK_TARGET_BYTES || + location !== location.trim() || + location.includes("\\") || + hasUnsafeRelativeLocationCodePoint(location) || + new TextEncoder().encode(location).byteLength > MAX_NETWORK_TARGET_BYTES + ) { + throw createCapabilityTransportError( + CAPABILITY_TRANSPORT_ERROR_CODES.redirectRejected, + resourceLabel, + ); + } + if (/^[a-z][a-z0-9+.-]*:/i.test(location)) return location; + try { + return new URL(location, currentTarget).href; + } catch { + throw createCapabilityTransportError( + CAPABILITY_TRANSPORT_ERROR_CODES.redirectRejected, + resourceLabel, + ); + } +} + +function combineAbortSignals(...signals: readonly AbortSignal[]): { + readonly signal: AbortSignal; + dispose(): void; +} { + const controller = new AbortController(); + const abort = (): void => controller.abort(); + for (const signal of signals) signal.addEventListener("abort", abort, { once: true }); + if (signals.some((signal) => signal.aborted)) abort(); + return Object.freeze({ + signal: controller.signal, + dispose(): void { + for (const signal of signals) signal.removeEventListener("abort", abort); + }, + }); +} + +export class CapabilityScopedTransport { + readonly #capabilityResolver: NetworkCapabilityResolver; + readonly #nameResolver: NetworkNameResolver; + readonly #adapter: PinnedNetworkHopAdapter; + readonly #session: NetworkSchedulerSession; + + constructor( + capabilityResolver: NetworkCapabilityResolver, + nameResolver: NetworkNameResolver, + adapter: PinnedNetworkHopAdapter, + scheduler: NetworkScheduler, + ) { + if (!isNetworkCapabilityResolver(capabilityResolver)) { + throw new TypeError("An authentic network capability resolver is required."); + } + if (!scheduler || typeof scheduler.createSession !== "function") { + throw new TypeError("A shared network scheduler is required."); + } + if (typeof nameResolver !== "function") { + throw new TypeError("A network name resolver is required."); + } + if (typeof adapter !== "function") { + throw new TypeError("A pinned network hop adapter is required."); + } + this.#capabilityResolver = capabilityResolver; + this.#nameResolver = nameResolver; + this.#adapter = adapter; + this.#session = scheduler.createSession(); + Object.freeze(this); + } + + async getBytes( + capability: NetworkCapability, + expectedScope: Scope, + ): Promise { + let initialResolution: NetworkCapabilityResolution; + try { + initialResolution = this.#capabilityResolver.resolve(capability, expectedScope); + } catch { + throw createCapabilityResolutionError(); + } + const validatedScope = initialResolution.scope as Scope; + const resourceLabel = createNetworkResourceLabel(initialResolution.resourceLabel); + const policy = NETWORK_BUFFERED_RESOURCE_POLICIES[validatedScope.resourceKind]; + if (!policy) { + throw createCapabilityTransportError( + CAPABILITY_TRANSPORT_ERROR_CODES.resourceRejected, + resourceLabel, + ); + } + + const outcome = await this.#session.schedule>({ + lane: policy.lane, + resourceLabel, + timeoutMs: policy.timeoutMs, + operation: async (signal) => { + let currentResolution: NetworkCapabilityResolution; + try { + currentResolution = this.#capabilityResolver.resolve( + capability, + validatedScope, + ); + } catch { + return { ok: false, error: createCapabilityResolutionError() }; + } + const combinedSignal = combineAbortSignals( + signal, + currentResolution.revocationSignal, + currentResolution.authoritySignal, + ); + try { + const value = await this.#execute( + currentResolution.target, + new Set(currentResolution.privateOrigins), + policy, + resourceLabel, + combinedSignal.signal, + ); + return { ok: true, value }; + } catch (error) { + return { + ok: false, + error: sanitizeCapabilityTransportError(error, resourceLabel), + }; + } finally { + combinedSignal.dispose(); + } + }, + }); + + if (!outcome.ok) throw outcome.error; + return outcome.value; + } + + dispose(): void { + this.#session.dispose(); + } + + isDisposed(): boolean { + return this.#session.isDisposed(); + } + + getLaneSnapshot(lane: NetworkLane): NetworkLaneSnapshot { + return this.#session.getLaneSnapshot(lane); + } + + getSnapshot(): NetworkSchedulerSnapshot { + return this.#session.getSnapshot(); + } + + async #selectConnectAddresses( + policy: TargetPolicyView, + privateOrigins: ReadonlySet, + resourceLabel: NetworkResourceLabel, + signal: AbortSignal, + ): Promise { + if (isResolvedAddressLiteral(policy.normalizedHostname)) { + return Object.freeze([policy.normalizedHostname]); + } + + let rawAddresses: unknown; + try { + rawAddresses = await this.#nameResolver(policy.normalizedHostname, signal); + } catch { + throw createCapabilityTransportError( + CAPABILITY_TRANSPORT_ERROR_CODES.addressRejected, + resourceLabel, + ); + } + assertNetworkOperationActive(signal, resourceLabel); + const addresses = snapshotResolvedAddresses(rawAddresses, resourceLabel); + const grantsPrivateAddress = privateOrigins.has(policy.normalizedOrigin); + const selected = addresses.filter( + (address) => + classifyResolvedAddress(address) === "ordinary-public-literal" || + grantsPrivateAddress, + ); + if (selected.length === 0) { + throw createCapabilityTransportError( + CAPABILITY_TRANSPORT_ERROR_CODES.addressRejected, + resourceLabel, + ); + } + return Object.freeze(selected); + } + + async #openHop( + target: string, + policy: TargetPolicyView, + connectAddresses: readonly string[], + resourceLabel: NetworkResourceLabel, + signal: AbortSignal, + ): Promise { + const parts = targetRequestParts(target, policy); + for (const connectAddress of connectAddresses) { + assertNetworkOperationActive(signal, resourceLabel); + const request = Object.freeze({ + protocol: policy.protocol, + connectAddress, + port: policy.port, + ...(isResolvedAddressLiteral(policy.normalizedHostname) + ? {} + : { serverName: policy.normalizedHostname }), + hostHeader: parts.hostHeader, + requestTarget: parts.requestTarget, + method: "GET" as const, + credentials: "omit" as const, + redirect: "manual" as const, + signal, + }); + + let response: unknown; + try { + response = await this.#adapter(request); + } catch { + assertNetworkOperationActive(signal, resourceLabel); + continue; + } + let prepared: ReturnType; + try { + prepared = snapshotHopResponse(response, connectAddress, resourceLabel); + } catch (error) { + await closeInvalidHopResponse(response); + throw error; + } + const managed = manageHopResponse(prepared, signal); + if (signal.aborted) { + await managed.close(); + assertNetworkOperationActive(signal, resourceLabel); + } + return managed; + } + throw createCapabilityTransportError( + CAPABILITY_TRANSPORT_ERROR_CODES.operationFailed, + resourceLabel, + ); + } + + async #execute( + initialTarget: string, + privateOrigins: ReadonlySet, + resourcePolicy: BufferedNetworkResourcePolicy, + resourceLabel: NetworkResourceLabel, + signal: AbortSignal, + ): Promise { + let target = initialTarget; + let previousProtocol: NetworkProtocol | undefined; + const visitedTargets = new Set(); + + for (let redirectCount = 0; ; redirectCount += 1) { + assertNetworkOperationActive(signal, resourceLabel); + const targetPolicy = evaluateTarget( + target, + privateOrigins, + resourceLabel, + previousProtocol, + ); + if (visitedTargets.has(target)) { + throw createCapabilityTransportError( + CAPABILITY_TRANSPORT_ERROR_CODES.redirectRejected, + resourceLabel, + ); + } + visitedTargets.add(target); + const connectAddresses = await this.#selectConnectAddresses( + targetPolicy, + privateOrigins, + resourceLabel, + signal, + ); + assertNetworkOperationActive(signal, resourceLabel); + + const response = await this.#openHop( + target, + targetPolicy, + connectAddresses, + resourceLabel, + signal, + ); + let bodyResult: Uint8Array | undefined; + let hopError: unknown; + try { + assertNetworkOperationActive(signal, resourceLabel); + if (REDIRECT_STATUSES.has(response.status)) { + if (redirectCount >= MAX_NETWORK_REDIRECTS) { + throw createCapabilityTransportError( + CAPABILITY_TRANSPORT_ERROR_CODES.redirectLimit, + resourceLabel, + ); + } + target = resolveRedirectTarget(response.location, target, resourceLabel); + previousProtocol = targetPolicy.protocol; + } else if (!resourcePolicy.acceptedStatuses.includes(response.status)) { + throw createCapabilityTransportError( + CAPABILITY_TRANSPORT_ERROR_CODES.statusRejected, + resourceLabel, + ); + } else { + bodyResult = await consumeBufferedNetworkBody( + response.body, + resourcePolicy.maxResponseBytes, + resourceLabel, + signal, + ); + } + } catch (error) { + hopError = error; + } + + try { + await response.close(); + } catch { + if (!hopError) { + hopError = createCapabilityTransportError( + CAPABILITY_TRANSPORT_ERROR_CODES.operationFailed, + resourceLabel, + ); + } + } + if (hopError) throw hopError; + assertNetworkOperationActive(signal, resourceLabel); + if (bodyResult) return Object.freeze({ status: response.status, bytes: bodyResult }); + } + } +} + +export interface CapabilityScopedNetworkRuntime { + readonly issuer: NetworkCapabilityIssuer; + readonly transport: CapabilityScopedTransport; + dispose(): void; +} + +/** + * Application composition entry point. It traps the raw-target resolver inside + * the transport and distributes only issuance and purpose-scoped operations. + */ +export function createCapabilityScopedNetworkRuntime( + nameResolver: NetworkNameResolver, + adapter: PinnedNetworkHopAdapter, + scheduler: NetworkScheduler, +): CapabilityScopedNetworkRuntime { + const authority = createNetworkCapabilityAuthority(); + const transport = new CapabilityScopedTransport( + authority.resolver, + nameResolver, + adapter, + scheduler, + ); + let disposed = false; + return Object.freeze({ + issuer: authority.issuer, + transport, + dispose(): void { + if (disposed) return; + disposed = true; + transport.dispose(); + authority.dispose(); + }, + }); +} diff --git a/src/network/LiteralTargetClassifier.test.ts b/src/network/LiteralTargetClassifier.test.ts new file mode 100644 index 0000000..1eefa81 --- /dev/null +++ b/src/network/LiteralTargetClassifier.test.ts @@ -0,0 +1,218 @@ +import { describe, expect, it } from "vitest"; +import { + MAX_NETWORK_TARGET_BYTES, + MAX_RESOLVED_ADDRESS_TEXT_LENGTH, + classifyResolvedAddress, + isResolvedAddressLiteral, + parseTargetPolicyView, + resolvedAddressesEqual, +} from "./LiteralTargetClassifier"; + +describe("literal network target classification", () => { + it.each([ + "0.0.0.0", + "10.0.0.1", + "100.64.0.1", + "127.0.0.1", + "169.254.169.254", + "172.31.255.255", + "192.0.0.9", + "192.0.2.1", + "192.31.196.1", + "192.52.193.1", + "192.88.99.1", + "192.168.1.1", + "192.175.48.1", + "198.18.0.1", + "198.51.100.1", + "203.0.113.1", + "224.0.0.1", + "255.255.255.255", + ])("fails closed for a special IPv4 address: %s", (address) => { + expect(classifyResolvedAddress(address)).toBe("non-public-or-special-literal"); + }); + + it.each(["1.1.1.1", "8.8.8.8", "93.184.216.34", "223.255.255.254"])( + "accepts an ordinary public IPv4 address: %s", + (address) => { + expect(classifyResolvedAddress(address)).toBe("ordinary-public-literal"); + }, + ); + + it("compares socket addresses canonically without accepting non-addresses", () => { + expect(resolvedAddressesEqual("2001:4860::1", "2001:4860:0:0:0:0:0:1")).toBe(true); + expect(resolvedAddressesEqual("8.8.8.8", "::ffff:8.8.8.8")).toBe(true); + expect(resolvedAddressesEqual("8.8.8.8", "8.8.4.4")).toBe(false); + expect(resolvedAddressesEqual("private.example", "private.example")).toBe(false); + }); + + it.each([ + "::", + "::1", + "64:ff9b::c000:201", + "64:ff9b:1::1", + "100::1", + "100:0:0:1::1", + "2001::1", + "2001:db8::1", + "2002::1", + "2620:4f:8000::1", + "3fff::1", + "5f00::1", + "fc00::1", + "fe80::1", + "fec0::1", + "ff02::1", + "4000::1", + ])("fails closed for a special or non-global IPv6 address: %s", (address) => { + expect(classifyResolvedAddress(address)).toBe("non-public-or-special-literal"); + }); + + it.each(["2001:4860:4860::8888", "2606:4700:4700::1111"])( + "accepts an ordinary public IPv6 address: %s", + (address) => { + expect(classifyResolvedAddress(address)).toBe("ordinary-public-literal"); + }, + ); + + it("classifies IPv4-mapped IPv6 addresses by their embedded address", () => { + expect(classifyResolvedAddress("::ffff:127.0.0.1")).toBe("non-public-or-special-literal"); + expect(classifyResolvedAddress("::ffff:8.8.8.8")).toBe("ordinary-public-literal"); + }); + + it.each([ + "", + " 8.8.8.8", + "999.1.1.1", + "8.8.8", + "example.com", + "fe80::1%en0", + "2001:::1", + "[2001:db8::1", + "[2001:4860::1]", + "32.1.72.96::", + "2001:4860:32.1.72.96::", + "8".repeat(MAX_RESOLVED_ADDRESS_TEXT_LENGTH + 1), + ])("treats a non-address resolver result as non-public: %s", (address) => { + expect(classifyResolvedAddress(address)).toBe("non-public-or-special-literal"); + expect(isResolvedAddressLiteral(address)).toBe(false); + }); + + it.each(["8.8.8.8", "127.0.0.1", "2001:4860::1", "::1", "::ffff:8.8.8.8"])( + "recognizes a strict resolver address spelling: %s", + (address) => { + expect(isResolvedAddressLiteral(address)).toBe(true); + }, + ); + + it.each([ + ["https://127.0.0.1/feed", "non-public-or-special-literal"], + ["https://2130706433/feed", "non-public-or-special-literal"], + ["https://0x7f000001/feed", "non-public-or-special-literal"], + ["https://0177.0.0.1/feed", "non-public-or-special-literal"], + ["https://192.168.1/feed", "non-public-or-special-literal"], + ["https://8.8.8.8/feed", "ordinary-public-literal"], + ["https://[::1]/feed", "non-public-or-special-literal"], + ["https://[2606:4700:4700::1111]/feed", "ordinary-public-literal"], + ["https://feeds.example.com/feed", "hostname-requires-resolution"], + ] as const)("classifies a parsed target host: %s", (target, classification) => { + const result = parseTargetPolicyView(target); + expect(result.ok).toBe(true); + if (result.ok) expect(result.value.hostClassification).toBe(classification); + }); + + it.each([ + "http://localhost/feed", + "https://api.localhost/feed", + "https://printer.local/feed", + "https://service.internal/feed", + "https://router.home.arpa/feed", + "https://single-label/feed", + ])("fails closed for a locally scoped hostname: %s", (target) => { + const result = parseTargetPolicyView(target); + expect(result.ok).toBe(true); + if (result.ok) { + expect(result.value.hostClassification).toBe("non-public-or-special-literal"); + } + }); + + it("produces a canonical policy view without returning a transmission target", () => { + const result = parseTargetPolicyView("HTTPS://User:Pass@ExAmPle.COM.:443/a%2Fb?sig=A%2BB"); + expect(result).toEqual({ + ok: true, + value: { + protocol: "https:", + normalizedHostname: "example.com", + normalizedOrigin: "https://example.com", + port: 443, + hasCredentials: true, + hostClassification: "hostname-requires-resolution", + }, + }); + if (result.ok) { + expect(Object.isFrozen(result.value)).toBe(true); + expect(result.value).not.toHaveProperty("target"); + expect(JSON.stringify(result.value)).not.toContain("sig=A"); + } + }); + + it("canonicalizes IPv6 origins and preserves non-default ports in policy only", () => { + const result = parseTargetPolicyView("http://[2001:4860:4860::8888]:8080/feed"); + expect(result).toEqual({ + ok: true, + value: { + protocol: "http:", + normalizedHostname: "2001:4860:4860::8888", + normalizedOrigin: "http://[2001:4860:4860::8888]:8080", + port: 8080, + hasCredentials: false, + hostClassification: "ordinary-public-literal", + }, + }); + }); + + it.each([ + [undefined, "empty"], + ["", "empty"], + [" https://example.com", "empty"], + ["https://example.com ", "empty"], + ["https://exa\n mple.com", "unsafe-unicode"], + [`https://exa${String.fromCharCode(0xd800)}mple.com`, "unsafe-unicode"], + ["mailto:owner@example.com", "unsupported-scheme"], + ["file:///etc/passwd", "unsupported-scheme"], + ["ftp://example.com/feed", "unsupported-scheme"], + ["https://example.com/#fragment", "malformed"], + ["https:\\example.com\\feed", "malformed"], + ["http:example.com/feed", "malformed"], + ["http:/example.com/feed", "malformed"], + ["http:///example.com/feed", "malformed"], + ["https:////example.com/feed", "malformed"], + ["not a target", "unsafe-unicode"], + ["https://", "malformed"], + ] as const)("rejects an invalid target without echoing it: %s", (target, reason) => { + const result = parseTargetPolicyView(target); + expect(result).toEqual({ ok: false, reason }); + if (typeof target === "string" && target.length > 0) { + expect(JSON.stringify(result)).not.toContain(target); + } + }); + + it.each(["https://@example.com/feed", "https://:@example.com/feed"])( + "detects even empty userinfo syntax: %s", + (target) => { + const result = parseTargetPolicyView(target); + expect(result.ok).toBe(true); + if (result.ok) expect(result.value.hasCredentials).toBe(true); + }, + ); + + it("bounds targets by UTF-8 bytes rather than UTF-16 code units", () => { + const prefix = "https://example.com/"; + const oversized = `${prefix}${"x".repeat(MAX_NETWORK_TARGET_BYTES - prefix.length + 1)}`; + expect(parseTargetPolicyView(oversized)).toEqual({ ok: false, reason: "too-large" }); + + const multibyte = `${prefix}${"é".repeat(MAX_NETWORK_TARGET_BYTES / 2)}`; + expect(multibyte.length).toBeLessThan(MAX_NETWORK_TARGET_BYTES); + expect(parseTargetPolicyView(multibyte)).toEqual({ ok: false, reason: "too-large" }); + }); +}); diff --git a/src/network/LiteralTargetClassifier.ts b/src/network/LiteralTargetClassifier.ts new file mode 100644 index 0000000..5bcc0f0 --- /dev/null +++ b/src/network/LiteralTargetClassifier.ts @@ -0,0 +1,391 @@ +export const MAX_NETWORK_TARGET_BYTES = 16 * 1024; +export const MAX_RESOLVED_ADDRESS_TEXT_LENGTH = 64; + +export type HostClassification = + | "ordinary-public-literal" + | "non-public-or-special-literal" + | "hostname-requires-resolution"; + +export type LiteralAddressClassification = Exclude< + HostClassification, + "hostname-requires-resolution" +>; + +export type NetworkProtocol = "http:" | "https:"; + +export interface TargetPolicyView { + readonly protocol: NetworkProtocol; + /** Canonical policy-only hostname. IPv6 brackets and a terminal DNS dot are removed. */ + readonly normalizedHostname: string; + /** Canonical policy-only origin. Never use this value as the transmitted target. */ + readonly normalizedOrigin: string; + readonly port: number; + readonly hasCredentials: boolean; + readonly hostClassification: HostClassification; +} + +export type TargetPolicyFailureReason = + | "empty" + | "too-large" + | "malformed" + | "unsupported-scheme" + | "unsafe-unicode"; + +export type TargetPolicyResult = + | { readonly ok: true; readonly value: TargetPolicyView } + | { readonly ok: false; readonly reason: TargetPolicyFailureReason }; + +interface AddressRange { + readonly network: bigint; + readonly prefixLength: number; +} + +const IPV4_BIT_LENGTH = 32; +const IPV6_BIT_LENGTH = 128; + +const IPV4_BLOCKED_RANGES = [ + ["0.0.0.0", 8], + ["10.0.0.0", 8], + ["100.64.0.0", 10], + ["127.0.0.0", 8], + ["169.254.0.0", 16], + ["172.16.0.0", 12], + ["192.0.0.0", 24], + ["192.0.2.0", 24], + ["192.31.196.0", 24], + ["192.52.193.0", 24], + ["192.88.99.0", 24], + ["192.168.0.0", 16], + ["192.175.48.0", 24], + ["198.18.0.0", 15], + ["198.51.100.0", 24], + ["203.0.113.0", 24], + ["224.0.0.0", 4], + ["240.0.0.0", 4], +] as const; + +const IPV6_BLOCKED_RANGES = [ + ["::", 128], + ["::1", 128], + ["64:ff9b::", 96], + ["64:ff9b:1::", 48], + ["100::", 64], + ["100:0:0:1::", 64], + ["2001::", 23], + ["2001:db8::", 32], + ["2002::", 16], + ["2620:4f:8000::", 48], + ["3fff::", 20], + ["5f00::", 16], + ["fc00::", 7], + ["fe80::", 10], + ["fec0::", 10], + ["ff00::", 8], +] as const; + +function parseCanonicalIpv4(value: string): bigint | undefined { + const parts = value.split("."); + if (parts.length !== 4) return undefined; + + let address = 0n; + for (const part of parts) { + if (!/^(?:0|[1-9][0-9]{0,2})$/.test(part)) return undefined; + const octet = Number(part); + if (octet > 255) return undefined; + address = (address << 8n) | BigInt(octet); + } + return address; +} + +function parseIpv6(value: string): bigint | undefined { + const candidate = value.toLowerCase(); + if (candidate.length === 0 || candidate.includes("%")) return undefined; + + const compressionIndex = candidate.indexOf("::"); + if (compressionIndex !== -1 && candidate.indexOf("::", compressionIndex + 2) !== -1) { + return undefined; + } + + const parseSide = (side: string): string[] | undefined => { + if (side.length === 0) return []; + const pieces = side.split(":"); + if (pieces.some((piece) => piece.length === 0)) return undefined; + return pieces; + }; + + const left = parseSide( + compressionIndex === -1 ? candidate : candidate.slice(0, compressionIndex), + ); + const right = parseSide(compressionIndex === -1 ? "" : candidate.slice(compressionIndex + 2)); + if (!left || !right) return undefined; + + const sideWithLastPiece = right.length > 0 ? right : left; + const lastPiece = sideWithLastPiece[sideWithLastPiece.length - 1]; + if (lastPiece?.includes(".")) { + // An embedded dotted quad is the final 32 bits of an IPv6 spelling. It + // cannot precede a trailing compression marker such as 192.0.2.1::. + if (compressionIndex !== -1 && right.length === 0) return undefined; + const ipv4 = parseCanonicalIpv4(lastPiece); + if (ipv4 === undefined) return undefined; + sideWithLastPiece.splice( + sideWithLastPiece.length - 1, + 1, + Number((ipv4 >> 16n) & 0xffffn).toString(16), + Number(ipv4 & 0xffffn).toString(16), + ); + } + + const combined = [...left, ...right]; + if (combined.some((piece) => piece.includes("."))) return undefined; + if (combined.some((piece) => !/^[0-9a-f]{1,4}$/.test(piece))) return undefined; + if (compressionIndex === -1 && combined.length !== 8) return undefined; + if (compressionIndex !== -1 && combined.length >= 8) return undefined; + + const missing = 8 - combined.length; + const hextets = + compressionIndex === -1 + ? combined + : [...left, ...Array.from({ length: missing }, () => "0"), ...right]; + if (hextets.length !== 8) return undefined; + + let address = 0n; + for (const hextet of hextets) address = (address << 16n) | BigInt(`0x${hextet}`); + return address; +} + +function rangeFromIpv4(network: string, prefixLength: number): AddressRange { + const parsed = parseCanonicalIpv4(network); + if (parsed === undefined) throw new Error("Invalid internal IPv4 range"); + return Object.freeze({ network: parsed, prefixLength }); +} + +function rangeFromIpv6(network: string, prefixLength: number): AddressRange { + const parsed = parseIpv6(network); + if (parsed === undefined) throw new Error("Invalid internal IPv6 range"); + return Object.freeze({ network: parsed, prefixLength }); +} + +const BLOCKED_IPV4 = IPV4_BLOCKED_RANGES.map(([network, prefixLength]) => + rangeFromIpv4(network, prefixLength), +); +const BLOCKED_IPV6 = IPV6_BLOCKED_RANGES.map(([network, prefixLength]) => + rangeFromIpv6(network, prefixLength), +); + +function isInRange(address: bigint, range: AddressRange, bitLength: number): boolean { + const trailingBits = BigInt(bitLength - range.prefixLength); + return address >> trailingBits === range.network >> trailingBits; +} + +function classifyIpv4(address: bigint): LiteralAddressClassification { + return BLOCKED_IPV4.some((range) => isInRange(address, range, IPV4_BIT_LENGTH)) + ? "non-public-or-special-literal" + : "ordinary-public-literal"; +} + +function classifyIpv6(address: bigint): LiteralAddressClassification { + // IPv4-mapped IPv6 addresses inherit the IPv4 classification. + if (address >> 32n === 0xffffn) return classifyIpv4(address & 0xffff_ffffn); + + if (BLOCKED_IPV6.some((range) => isInRange(address, range, IPV6_BIT_LENGTH))) { + return "non-public-or-special-literal"; + } + + // Only the globally allocated 2000::/3 space is accepted by default. Newly + // allocated special ranges outside it therefore fail closed until reviewed. + return address >> 125n === 1n ? "ordinary-public-literal" : "non-public-or-special-literal"; +} + +export function classifyResolvedAddress(address: unknown): LiteralAddressClassification { + if ( + typeof address !== "string" || + address.length === 0 || + address.length > MAX_RESOLVED_ADDRESS_TEXT_LENGTH || + address !== address.trim() + ) { + return "non-public-or-special-literal"; + } + + const ipv4 = parseCanonicalIpv4(address); + if (ipv4 !== undefined) return classifyIpv4(ipv4); + + const ipv6 = parseIpv6(address); + return ipv6 === undefined ? "non-public-or-special-literal" : classifyIpv6(ipv6); +} + +/** True only for the strict address spellings accepted from a resolver adapter. */ +export function isResolvedAddressLiteral(address: unknown): address is string { + if ( + typeof address !== "string" || + address.length === 0 || + address.length > MAX_RESOLVED_ADDRESS_TEXT_LENGTH || + address !== address.trim() + ) { + return false; + } + return parseCanonicalIpv4(address) !== undefined || parseIpv6(address) !== undefined; +} + +function canonicalResolvedAddress( + address: unknown, +): { readonly family: 4 | 6; readonly value: bigint } | undefined { + if ( + typeof address !== "string" || + address.length === 0 || + address.length > MAX_RESOLVED_ADDRESS_TEXT_LENGTH || + address !== address.trim() + ) { + return undefined; + } + const ipv4 = parseCanonicalIpv4(address); + if (ipv4 !== undefined) return { family: 4, value: ipv4 }; + const ipv6 = parseIpv6(address); + if (ipv6 === undefined) return undefined; + if (ipv6 >> 32n === 0xffffn) return { family: 4, value: ipv6 & 0xffff_ffffn }; + return { family: 6, value: ipv6 }; +} + +/** Compares strict resolver spellings, including IPv4-mapped socket addresses. */ +export function resolvedAddressesEqual(left: unknown, right: unknown): boolean { + const leftAddress = canonicalResolvedAddress(left); + const rightAddress = canonicalResolvedAddress(right); + return Boolean( + leftAddress && + rightAddress && + leftAddress.family === rightAddress.family && + leftAddress.value === rightAddress.value, + ); +} + +function containsUnsafeCodePoint(value: string): boolean { + for (let index = 0; index < value.length; index += 1) { + const codeUnit = value.charCodeAt(index); + if (codeUnit >= 0xd800 && codeUnit <= 0xdbff) { + const following = value.charCodeAt(index + 1); + if (following < 0xdc00 || following > 0xdfff) return true; + index += 1; + continue; + } + if (codeUnit >= 0xdc00 && codeUnit <= 0xdfff) return true; + if (codeUnit <= 0x20 || (codeUnit >= 0x7f && codeUnit <= 0x9f)) return true; + if ( + codeUnit === 0x061c || + codeUnit === 0x200b || + codeUnit === 0x200c || + codeUnit === 0x200d || + codeUnit === 0x200e || + codeUnit === 0x200f || + (codeUnit >= 0x2028 && codeUnit <= 0x202e) || + (codeUnit >= 0x2066 && codeUnit <= 0x2069) || + codeUnit === 0xfeff + ) { + return true; + } + } + return false; +} + +function classifyHostname(hostname: string): HostClassification { + const ipv4 = parseCanonicalIpv4(hostname); + if (ipv4 !== undefined) return classifyIpv4(ipv4); + + const ipv6 = parseIpv6(hostname); + if (ipv6 !== undefined) return classifyIpv6(ipv6); + + const labels = hostname.split("."); + if ( + labels.length === 1 || + labels.some((label) => label.length === 0 || label.length > 63) || + hostname.length > 253 || + hostname === "localhost" || + hostname.endsWith(".localhost") || + hostname === "local" || + hostname.endsWith(".local") || + hostname === "internal" || + hostname.endsWith(".internal") || + hostname === "home.arpa" || + hostname.endsWith(".home.arpa") + ) { + return "non-public-or-special-literal"; + } + + return "hostname-requires-resolution"; +} + +function normalizedHostFromUrl(url: URL): string | undefined { + let hostname = url.hostname.toLowerCase(); + if (hostname.startsWith("[") && hostname.endsWith("]")) { + hostname = hostname.slice(1, -1); + } else if (hostname.endsWith(".")) { + hostname = hostname.slice(0, -1); + if (hostname.endsWith(".")) return undefined; + } + return hostname.length === 0 ? undefined : hostname; +} + +function normalizedOrigin(protocol: NetworkProtocol, hostname: string, port: number): string { + const host = hostname.includes(":") ? `[${hostname}]` : hostname; + const defaultPort = protocol === "https:" ? 443 : 80; + return `${protocol}//${host}${port === defaultPort ? "" : `:${port}`}`; +} + +export function parseTargetPolicyView(value: unknown): TargetPolicyResult { + if (typeof value !== "string" || value.length === 0) { + return { ok: false, reason: "empty" }; + } + // Every UTF-8 encoding is at least as large as its UTF-16 code-unit count. + // Reject obvious oversize input before trim, scanning, or allocation. + if (value.length > MAX_NETWORK_TARGET_BYTES) return { ok: false, reason: "too-large" }; + if (value !== value.trim()) return { ok: false, reason: "empty" }; + if (containsUnsafeCodePoint(value)) return { ok: false, reason: "unsafe-unicode" }; + if (new TextEncoder().encode(value).byteLength > MAX_NETWORK_TARGET_BYTES) { + return { ok: false, reason: "too-large" }; + } + // WHATWG special-URL backslash normalization is not safe for a byte-exact + // capability target interpreted later by a different transport. + if (value.includes("\\")) return { ok: false, reason: "malformed" }; + const scheme = /^([a-z][a-z0-9+.-]*):/i.exec(value)?.[1]?.toLowerCase(); + if (scheme !== "http" && scheme !== "https") { + return { ok: false, reason: scheme ? "unsupported-scheme" : "malformed" }; + } + // Require an explicit authority spelling. WHATWG otherwise accepts and + // rewrites forms such as `http:example.com` and `http:///example.com`. + if (!/^https?:\/\/[^/?#]/i.test(value)) return { ok: false, reason: "malformed" }; + + let url: URL; + try { + url = new URL(value); + } catch { + return { ok: false, reason: "malformed" }; + } + + if (url.protocol !== "http:" && url.protocol !== "https:") { + return { ok: false, reason: "unsupported-scheme" }; + } + if (url.hash.length > 0) return { ok: false, reason: "malformed" }; + + const hostname = normalizedHostFromUrl(url); + if (!hostname) return { ok: false, reason: "malformed" }; + const protocol = url.protocol; + const authorityStart = value.indexOf("//") + 2; + const authorityEndCandidate = value.slice(authorityStart).search(/[/?#]/); + const authorityEnd = + authorityEndCandidate === -1 ? value.length : authorityStart + authorityEndCandidate; + const rawAuthority = value.slice(authorityStart, authorityEnd); + const port = url.port.length > 0 ? Number(url.port) : protocol === "https:" ? 443 : 80; + if (!Number.isSafeInteger(port) || port < 1 || port > 65_535) { + return { ok: false, reason: "malformed" }; + } + + const classification = classifyHostname(hostname); + const result: TargetPolicyView = Object.freeze({ + protocol, + normalizedHostname: hostname, + normalizedOrigin: normalizedOrigin(protocol, hostname, port), + port, + hasCredentials: + rawAuthority.includes("@") || url.username.length > 0 || url.password.length > 0, + hostClassification: classification, + }); + return { ok: true, value: result }; +} diff --git a/src/network/NetworkCapability.test.ts b/src/network/NetworkCapability.test.ts new file mode 100644 index 0000000..c8bf05d --- /dev/null +++ b/src/network/NetworkCapability.test.ts @@ -0,0 +1,383 @@ +import { describe, expect, it } from "vitest"; +import { + EPISODE_NETWORK_RESOURCE_KINDS, + FEED_NETWORK_RESOURCE_KINDS, + MAX_NETWORK_PRIVATE_ORIGINS, + MAX_NETWORK_TARGET_BYTES, + NETWORK_RESOURCE_LABELS, + NetworkCapabilityError, + createNetworkCapabilityAuthority, + isNetworkCapabilityResolver, + type NetworkCapability, + type NetworkCapabilityScope, +} from "./NetworkCapability"; +import type { EpisodeHandle, FeedHandle } from "src/security/resourceHandles"; + +const feedId = `podnotes-feed-${"1a".repeat(32)}` as FeedHandle; +const otherFeedId = `podnotes-feed-${"2b".repeat(32)}` as FeedHandle; +const episodeId = `podnotes-episode-${"3c".repeat(32)}` as EpisodeHandle; +const otherEpisodeId = `podnotes-episode-${"4d".repeat(32)}` as EpisodeHandle; + +const subscriptionScope = Object.freeze({ + resourceKind: "subscription", + feedId, +} as const); + +const streamScope = Object.freeze({ + resourceKind: "episode-stream", + feedId, + episodeId, +} as const); + +function expectResolutionFailure( + resolver: ReturnType["resolver"], + token: NetworkCapability, + expectedScope: NetworkCapabilityScope, +): void { + let thrown: unknown; + try { + resolver.resolve(token, expectedScope); + } catch (error) { + thrown = error; + } + expect(thrown).toBeInstanceOf(NetworkCapabilityError); + expect((thrown as Error).message).toBe("Network capability is invalid or unavailable."); +} + +describe("network capability authority", () => { + it.each(FEED_NETWORK_RESOURCE_KINDS)( + "binds the %s resource to its exact feed", + (resourceKind) => { + const authority = createNetworkCapabilityAuthority(); + const scope = { resourceKind, feedId } as const; + const token = authority.issuer.issue(scope, "https://media.example/resource"); + + const resolution = authority.resolver.resolve(token, scope); + expect(resolution).toMatchObject({ + scope, + target: "https://media.example/resource", + privateOrigins: [], + resourceLabel: NETWORK_RESOURCE_LABELS[resourceKind], + }); + expect(resolution.revocationSignal.aborted).toBe(false); + }, + ); + + it.each(EPISODE_NETWORK_RESOURCE_KINDS)( + "binds the %s resource to its exact feed and episode", + (resourceKind) => { + const authority = createNetworkCapabilityAuthority(); + const scope = { resourceKind, feedId, episodeId } as const; + const token = authority.issuer.issue(scope, "https://media.example/resource"); + + const resolution = authority.resolver.resolve(token, scope); + expect(resolution).toMatchObject({ + scope, + target: "https://media.example/resource", + privateOrigins: [], + resourceLabel: NETWORK_RESOURCE_LABELS[resourceKind], + }); + expect(resolution.revocationSignal.aborted).toBe(false); + }, + ); + + it("preserves exact target bytes without URL normalization", () => { + const authority = createNetworkCapabilityAuthority(); + const target = "HTTPS://Example.COM:443/audio%2fpart.mp3?Signature=A%2BB"; + const token = authority.issuer.issue(streamScope, target); + + expect(authority.resolver.resolve(token, streamScope).target).toBe(target); + }); + + it("copies and freezes every piece of hidden resolution authority", () => { + const authority = createNetworkCapabilityAuthority(); + const mutableScope = { + resourceKind: "episode-stream" as const, + feedId, + episodeId, + }; + const privateOrigins = ["https://private.example:8443"]; + const token = authority.issuer.issue( + mutableScope, + "https://private.example:8443/audio.mp3", + privateOrigins, + ); + + mutableScope.feedId = otherFeedId; + mutableScope.episodeId = otherEpisodeId; + privateOrigins[0] = "https://changed.example"; + privateOrigins.push("https://extra.example"); + + const resolution = authority.resolver.resolve(token, streamScope); + expect(resolution.scope).toEqual(streamScope); + expect(resolution.privateOrigins).toEqual(["https://private.example:8443"]); + expect(Object.isFrozen(resolution)).toBe(true); + expect(Object.isFrozen(resolution.scope)).toBe(true); + expect(Object.isFrozen(resolution.privateOrigins)).toBe(true); + expect(() => { + (resolution.privateOrigins as string[]).push("https://mutated.example"); + }).toThrow(); + expect(() => { + Object.assign(resolution.scope, { feedId: otherFeedId }); + }).toThrow(); + }); + + it("uses frozen property-free tokens that reveal nothing when serialized", () => { + const authority = createNetworkCapabilityAuthority(); + const target = "https://secret.example/feed.xml?token=do-not-log"; + const privateOrigin = "https://private.example"; + const token = authority.issuer.issue(subscriptionScope, target, [privateOrigin]); + + expect(Object.isFrozen(token)).toBe(true); + expect(Object.getPrototypeOf(token)).toBeNull(); + expect(Object.keys(token)).toEqual([]); + expect(Object.getOwnPropertyNames(token)).toEqual([]); + expect(Object.getOwnPropertySymbols(token)).toEqual([]); + expect(Reflect.ownKeys(token)).toEqual([]); + expect(JSON.stringify(token)).toBe("{}"); + expect(JSON.stringify(token)).not.toContain(target); + expect(JSON.stringify(token)).not.toContain(privateOrigin); + expect(() => Object.defineProperty(token, "target", { value: target })).toThrow(); + }); + + it("keeps issuer and resolver authority on separate frozen facets", () => { + const authority = createNetworkCapabilityAuthority(); + + expect(Object.isFrozen(authority)).toBe(true); + expect(Object.isFrozen(authority.issuer)).toBe(true); + expect(Object.isFrozen(authority.resolver)).toBe(true); + expect(Object.keys(authority.issuer).sort()).toEqual(["issue", "revoke"]); + expect(Object.keys(authority.resolver)).toEqual(["resolve"]); + expect("resolve" in authority.issuer).toBe(false); + expect("issue" in authority.resolver).toBe(false); + expect(isNetworkCapabilityResolver(authority.resolver)).toBe(true); + expect(isNetworkCapabilityResolver({ resolve: authority.resolver.resolve })).toBe(false); + }); + + it("rejects forged, cloned, deserialized, primitive, and cross-authority tokens", () => { + const first = createNetworkCapabilityAuthority(); + const second = createNetworkCapabilityAuthority(); + const token = first.issuer.issue(subscriptionScope, "https://secret.example/feed.xml"); + const propertyFreeForgery = Object.freeze(Object.create(null) as object); + const spreadClone = { ...token }; + const deserialized = JSON.parse(JSON.stringify(token)) as unknown; + + for (const candidate of [propertyFreeForgery, spreadClone, deserialized, null, "token"]) { + expectResolutionFailure( + first.resolver, + candidate as NetworkCapability, + subscriptionScope, + ); + } + expectResolutionFailure(second.resolver, token, subscriptionScope); + }); + + it("rejects every cross-scope use before exposing resolution data", () => { + const authority = createNetworkCapabilityAuthority(); + const target = "https://secret.example/audio.mp3?credential=hidden"; + const token = authority.issuer.issue(streamScope, target); + const wrongScopes: NetworkCapabilityScope[] = [ + { resourceKind: "episode-stream", feedId: otherFeedId, episodeId }, + { resourceKind: "episode-stream", feedId, episodeId: otherEpisodeId }, + { resourceKind: "episode-chapters", feedId, episodeId }, + { resourceKind: "subscription", feedId }, + ]; + + for (const wrongScope of wrongScopes) { + let thrown: unknown; + try { + authority.resolver.resolve(token, wrongScope); + } catch (error) { + thrown = error; + } + expect(thrown).toBeInstanceOf(NetworkCapabilityError); + expect((thrown as Error).message).not.toContain(target); + } + }); + + it("rejects malformed issue and expected scopes without invoking accessors", () => { + const authority = createNetworkCapabilityAuthority(); + const target = "https://secret.example/feed.xml?credential=hidden"; + let accessorRead = false; + const accessorScope = { + feedId, + } as Record; + Object.defineProperty(accessorScope, "resourceKind", { + enumerable: true, + get: () => { + accessorRead = true; + return "subscription"; + }, + }); + const symbolScope = { ...subscriptionScope, [Symbol("hidden")]: true }; + const prototypeKeyScope = { ...subscriptionScope } as Record; + Object.defineProperty(prototypeKeyScope, "__proto__", { + enumerable: true, + value: { resourceKind: "subscription", feedId }, + }); + const malformedScopes: unknown[] = [ + {}, + [], + { resourceKind: "subscription", feedId: "https://example.com/feed.xml" }, + { resourceKind: "unknown", feedId }, + { resourceKind: "subscription", feedId, episodeId }, + { resourceKind: "episode-stream", feedId }, + { resourceKind: "episode-stream", feedId, episodeId: "episode-title" }, + { ...subscriptionScope, extra: true }, + Object.create({ resourceKind: "subscription", feedId }), + accessorScope, + symbolScope, + prototypeKeyScope, + ]; + + for (const malformedScope of malformedScopes) { + expect(() => + authority.issuer.issue(malformedScope as NetworkCapabilityScope, target), + ).toThrow(NetworkCapabilityError); + } + expect(accessorRead).toBe(false); + + const token = authority.issuer.issue(subscriptionScope, target); + for (const malformedScope of malformedScopes) { + let thrown: unknown; + try { + authority.resolver.resolve(token, malformedScope as NetworkCapabilityScope); + } catch (error) { + thrown = error; + } + expect(thrown).toBeInstanceOf(NetworkCapabilityError); + expect((thrown as Error).message).not.toContain(target); + } + expect(accessorRead).toBe(false); + }); + + it.each([ + "", + " https://example.com/feed.xml", + "https://example.com/feed.xml ", + "ftp://example.com/feed.xml", + "file:///tmp/audio.mp3", + "not a URL", + "http:example.com/feed.xml", + "http:///example.com/feed.xml", + "https://user:password@example.com/feed.xml", + "https://@example.com/feed.xml", + "https://example.com/feed.xml#fragment", + "https:\\example.com\\feed.xml", + "https://example.com/a\u0000b", + "https://example.com/a\u202eb", + `https://example.com/${"a".repeat(MAX_NETWORK_TARGET_BYTES)}`, + ])("rejects an invalid or unbounded target without echoing it: %s", (target) => { + const authority = createNetworkCapabilityAuthority(); + let thrown: unknown; + try { + authority.issuer.issue(subscriptionScope, target); + } catch (error) { + thrown = error; + } + expect(thrown).toBeInstanceOf(NetworkCapabilityError); + if (target.length > 0) expect((thrown as Error).message).not.toContain(target); + }); + + it("counts target limits in UTF-8 bytes", () => { + const authority = createNetworkCapabilityAuthority(); + const oversized = `https://example.com/${"\u{1f680}".repeat(MAX_NETWORK_TARGET_BYTES / 2)}`; + + expect(() => authority.issuer.issue(subscriptionScope, oversized)).toThrow( + NetworkCapabilityError, + ); + }); + + it.each([ + ["trailing slash", ["https://private.example/"]], + ["path", ["https://private.example/path"]], + ["query", ["https://private.example?key=value"]], + ["fragment", ["https://private.example#section"]], + ["credentials", ["https://user:pass@private.example"]], + ["non-http", ["ftp://private.example"]], + ["non-canonical host", ["https://PRIVATE.example"]], + ["duplicate", ["https://private.example", "https://private.example"]], + ])("rejects invalid private origins: %s", (_name, privateOrigins) => { + const authority = createNetworkCapabilityAuthority(); + expect(() => + authority.issuer.issue( + subscriptionScope, + "https://example.com/feed.xml", + privateOrigins, + ), + ).toThrow(NetworkCapabilityError); + }); + + it("rejects sparse, accessor-backed, decorated, and over-limit private-origin arrays", () => { + const authority = createNetworkCapabilityAuthority(); + const sparse = Array.from({ length: 1 }); + delete sparse[0]; + const accessorBacked: string[] = []; + Object.defineProperty(accessorBacked, "0", { + enumerable: true, + get: () => "https://private.example", + }); + Object.defineProperty(accessorBacked, "length", { value: 1 }); + const decorated = ["https://private.example"]; + Object.defineProperty(decorated, "extra", { value: true, enumerable: true }); + const tooMany = Array.from( + { length: MAX_NETWORK_PRIVATE_ORIGINS + 1 }, + (_, index) => `https://private-${index}.example`, + ); + const revoked = Proxy.revocable([], {}); + revoked.revoke(); + + for (const privateOrigins of [sparse, accessorBacked, decorated, tooMany, revoked.proxy]) { + expect(() => + authority.issuer.issue( + subscriptionScope, + "https://example.com/feed.xml", + privateOrigins, + ), + ).toThrow(NetworkCapabilityError); + } + }); + + it("revokes individual tokens and disposes the complete authority", () => { + const authority = createNetworkCapabilityAuthority(); + const first = authority.issuer.issue(subscriptionScope, "https://example.com/feed.xml"); + const second = authority.issuer.issue(streamScope, "https://example.com/audio.mp3"); + const firstSignal = authority.resolver.resolve(first, subscriptionScope).revocationSignal; + const secondResolution = authority.resolver.resolve(second, streamScope); + const secondSignal = secondResolution.revocationSignal; + const authoritySignal = secondResolution.authoritySignal; + + expect(authority.issuer.revoke(first)).toBe(true); + expect(firstSignal.aborted).toBe(true); + expect(secondSignal.aborted).toBe(false); + expect(authoritySignal.aborted).toBe(false); + expect(authority.issuer.revoke(first)).toBe(false); + expect(authority.issuer.revoke(Object.freeze({}))).toBe(false); + expectResolutionFailure(authority.resolver, first, subscriptionScope); + expect(authority.resolver.resolve(second, streamScope).target).toBe( + "https://example.com/audio.mp3", + ); + + authority.dispose(); + expect(secondSignal.aborted).toBe(false); + expect(authoritySignal.aborted).toBe(true); + expectResolutionFailure(authority.resolver, second, streamScope); + expect(authority.issuer.revoke(second)).toBe(false); + expect(() => + authority.issuer.issue(subscriptionScope, "https://example.com/new.xml"), + ).toThrow(NetworkCapabilityError); + }); + + it("never derives resource labels from targets, origins, titles, or handles", () => { + const authority = createNetworkCapabilityAuthority(); + const target = "https://sensitive.example/private-feed-name.xml"; + const origin = "https://sensitive.example"; + const token = authority.issuer.issue(subscriptionScope, target, [origin]); + const { resourceLabel } = authority.resolver.resolve(token, subscriptionScope); + + expect(resourceLabel).toBe("podcast subscription"); + expect(resourceLabel).not.toContain(target); + expect(resourceLabel).not.toContain(origin); + expect(resourceLabel).not.toContain(feedId); + }); +}); diff --git a/src/network/NetworkCapability.ts b/src/network/NetworkCapability.ts new file mode 100644 index 0000000..dc1ee6f --- /dev/null +++ b/src/network/NetworkCapability.ts @@ -0,0 +1,313 @@ +import { + isEpisodeHandle, + isFeedHandle, + type EpisodeHandle, + type FeedHandle, +} from "src/security/resourceHandles"; + +import { parseTargetPolicyView } from "./LiteralTargetClassifier"; + +export { MAX_NETWORK_TARGET_BYTES } from "./LiteralTargetClassifier"; +export const MAX_NETWORK_PRIVATE_ORIGINS = 16; + +export const FEED_NETWORK_RESOURCE_KINDS = ["subscription", "feed-artwork", "site"] as const; +export const EPISODE_NETWORK_RESOURCE_KINDS = [ + "episode-stream", + "episode-chapters", + "episode-artwork", + "episode-item-link", +] as const; + +export type FeedNetworkResourceKind = (typeof FEED_NETWORK_RESOURCE_KINDS)[number]; +export type EpisodeNetworkResourceKind = (typeof EPISODE_NETWORK_RESOURCE_KINDS)[number]; +export type NetworkResourceKind = FeedNetworkResourceKind | EpisodeNetworkResourceKind; + +export interface FeedNetworkCapabilityScope { + readonly resourceKind: FeedNetworkResourceKind; + readonly feedId: FeedHandle; +} + +export interface EpisodeNetworkCapabilityScope { + readonly resourceKind: EpisodeNetworkResourceKind; + readonly feedId: FeedHandle; + readonly episodeId: EpisodeHandle; +} + +export type NetworkCapabilityScope = FeedNetworkCapabilityScope | EpisodeNetworkCapabilityScope; + +export const NETWORK_RESOURCE_LABELS = Object.freeze({ + subscription: "podcast subscription", + "feed-artwork": "podcast artwork", + site: "podcast site", + "episode-stream": "episode audio", + "episode-chapters": "episode chapters", + "episode-artwork": "episode artwork", + "episode-item-link": "episode page", +} as const satisfies Record); + +export type NetworkResourceLabel = (typeof NETWORK_RESOURCE_LABELS)[NetworkResourceKind]; + +declare const networkCapabilityBrand: unique symbol; + +/** + * An opaque authority token. The brand exists only to help TypeScript. Runtime + * tokens are frozen, null-prototype objects with no own properties or symbols. + */ +export interface NetworkCapability { + readonly [networkCapabilityBrand]: Scope; +} + +export interface NetworkCapabilityResolution< + Scope extends NetworkCapabilityScope = NetworkCapabilityScope, +> { + readonly scope: Readonly; + readonly target: string; + readonly privateOrigins: readonly string[]; + readonly resourceLabel: NetworkResourceLabel; + readonly revocationSignal: AbortSignal; + readonly authoritySignal: AbortSignal; +} + +export interface NetworkCapabilityIssuer { + issue( + scope: Scope, + target: string, + privateOrigins?: readonly string[], + ): NetworkCapability; + revoke(capability: unknown): boolean; +} + +export interface NetworkCapabilityResolver { + resolve( + capability: NetworkCapability, + expectedScope: Scope, + ): NetworkCapabilityResolution; +} + +/** + * Construction-only object. Application composition should distribute the + * issuer and resolver facets independently to keep target issuance separate + * from transport resolution. + */ +export interface NetworkCapabilityAuthority { + readonly issuer: NetworkCapabilityIssuer; + readonly resolver: NetworkCapabilityResolver; + dispose(): void; +} + +export class NetworkCapabilityError extends Error { + constructor(operation: "issue" | "resolve") { + super( + operation === "issue" + ? "Invalid network capability request." + : "Network capability is invalid or unavailable.", + ); + this.name = "NetworkCapabilityError"; + } +} + +type UnknownRecord = Record; + +const feedResourceKinds = new Set(FEED_NETWORK_RESOURCE_KINDS); +const episodeResourceKinds = new Set(EPISODE_NETWORK_RESOURCE_KINDS); +const authenticResolvers = new WeakSet(); + +function isBoundedHttpTarget(value: unknown): value is string { + const policy = parseTargetPolicyView(value); + return policy.ok && !policy.value.hasCredentials; +} + +function isExactPrivateOrigin(value: unknown): value is string { + if (!isBoundedHttpTarget(value)) return false; + try { + const parsed = new URL(value); + return ( + !parsed.username && + !parsed.password && + parsed.pathname === "/" && + !parsed.search && + !parsed.hash && + parsed.origin === value + ); + } catch { + return false; + } +} + +function copyPrivateOrigins(value: unknown): readonly string[] | null { + if (value === undefined) return Object.freeze([] as string[]); + try { + if (!Array.isArray(value)) return null; + const length = value.length; + if (length > MAX_NETWORK_PRIVATE_ORIGINS) return null; + const ownKeys = Reflect.ownKeys(value); + if (ownKeys.length !== length + 1 || !ownKeys.includes("length")) return null; + + const origins: string[] = []; + for (let index = 0; index < length; index += 1) { + const key = String(index); + if (!ownKeys.includes(key)) return null; + const descriptor = Object.getOwnPropertyDescriptor(value, key); + if (!descriptor?.enumerable || !("value" in descriptor)) return null; + if (!isExactPrivateOrigin(descriptor.value)) return null; + origins.push(descriptor.value); + } + if (new Set(origins).size !== origins.length) return null; + return Object.freeze(origins); + } catch { + return null; + } +} + +function strictDataRecord(value: unknown): UnknownRecord | null { + try { + if (typeof value !== "object" || value === null || Array.isArray(value)) return null; + const prototype = Object.getPrototypeOf(value); + if (prototype !== Object.prototype && prototype !== null) return null; + + const record = Object.create(null) as UnknownRecord; + for (const key of Reflect.ownKeys(value)) { + if (typeof key !== "string") return null; + const descriptor = Object.getOwnPropertyDescriptor(value, key); + if (!descriptor?.enumerable || !("value" in descriptor)) return null; + record[key] = descriptor.value; + } + return record; + } catch { + return null; + } +} + +function hasExactKeys(record: UnknownRecord, expected: readonly string[]): boolean { + const keys = Object.keys(record); + return keys.length === expected.length && expected.every((key) => keys.includes(key)); +} + +function copyScope(value: unknown): NetworkCapabilityScope | null { + const record = strictDataRecord(value); + if (!record || typeof record.resourceKind !== "string" || !isFeedHandle(record.feedId)) { + return null; + } + + if (feedResourceKinds.has(record.resourceKind)) { + if (!hasExactKeys(record, ["resourceKind", "feedId"])) return null; + return Object.freeze({ + resourceKind: record.resourceKind as FeedNetworkResourceKind, + feedId: record.feedId, + }); + } + + if (episodeResourceKinds.has(record.resourceKind)) { + if ( + !hasExactKeys(record, ["resourceKind", "feedId", "episodeId"]) || + !isEpisodeHandle(record.episodeId) + ) { + return null; + } + return Object.freeze({ + resourceKind: record.resourceKind as EpisodeNetworkResourceKind, + feedId: record.feedId, + episodeId: record.episodeId, + }); + } + + return null; +} + +function scopesMatch(left: NetworkCapabilityScope, right: NetworkCapabilityScope): boolean { + if (left.resourceKind !== right.resourceKind || left.feedId !== right.feedId) return false; + if (episodeResourceKinds.has(left.resourceKind)) { + return "episodeId" in left && "episodeId" in right && left.episodeId === right.episodeId; + } + return !("episodeId" in left) && !("episodeId" in right); +} + +function isObject(value: unknown): value is object { + return (typeof value === "object" && value !== null) || typeof value === "function"; +} + +/** Runtime authentication for the resolver facet issued by this module. */ +export function isNetworkCapabilityResolver(value: unknown): value is NetworkCapabilityResolver { + return isObject(value) && authenticResolvers.has(value); +} + +/** + * Low-level composition primitive. Never distribute the returned aggregate; + * application code must trap its resolver inside the network runtime and expose + * only the issuer plus purpose-scoped operations. + */ +export function createNetworkCapabilityAuthority(): NetworkCapabilityAuthority { + const resolutions = new WeakMap< + object, + { + readonly resolution: NetworkCapabilityResolution; + readonly revocation: AbortController; + } + >(); + const authorityRevocation = new AbortController(); + let disposed = false; + + const issuer: NetworkCapabilityIssuer = Object.freeze({ + issue( + scope: Scope, + target: string, + privateOrigins?: readonly string[], + ): NetworkCapability { + const copiedScope = copyScope(scope); + if (disposed || !copiedScope || !isBoundedHttpTarget(target)) { + throw new NetworkCapabilityError("issue"); + } + const copiedPrivateOrigins = copyPrivateOrigins(privateOrigins); + if (!copiedPrivateOrigins) throw new NetworkCapabilityError("issue"); + + const token = Object.freeze(Object.create(null) as object); + const revocation = new AbortController(); + const resolution = Object.freeze({ + scope: copiedScope, + target, + privateOrigins: copiedPrivateOrigins, + resourceLabel: NETWORK_RESOURCE_LABELS[copiedScope.resourceKind], + revocationSignal: revocation.signal, + authoritySignal: authorityRevocation.signal, + }); + resolutions.set(token, { resolution, revocation }); + return token as NetworkCapability; + }, + revoke(capability: unknown): boolean { + if (disposed || !isObject(capability)) return false; + const entry = resolutions.get(capability); + if (!entry || !resolutions.delete(capability)) return false; + entry.revocation.abort(); + return true; + }, + }); + + const resolver: NetworkCapabilityResolver = Object.freeze({ + resolve( + capability: NetworkCapability, + expectedScope: Scope, + ): NetworkCapabilityResolution { + const copiedExpectedScope = copyScope(expectedScope); + if (!copiedExpectedScope || disposed || !isObject(capability)) { + throw new NetworkCapabilityError("resolve"); + } + + const entry = resolutions.get(capability); + if (!entry || !scopesMatch(entry.resolution.scope, copiedExpectedScope)) { + throw new NetworkCapabilityError("resolve"); + } + return entry.resolution as NetworkCapabilityResolution; + }, + }); + authenticResolvers.add(resolver); + + return Object.freeze({ + issuer, + resolver, + dispose(): void { + if (disposed) return; + disposed = true; + authorityRevocation.abort(); + }, + }); +} diff --git a/src/network/NetworkErrors.test.ts b/src/network/NetworkErrors.test.ts new file mode 100644 index 0000000..dcbb080 --- /dev/null +++ b/src/network/NetworkErrors.test.ts @@ -0,0 +1,141 @@ +import { describe, expect, it } from "vitest"; +import { + CAPABILITY_TRANSPORT_ERROR_CODES, + CapabilityTransportError, + createCapabilityTransportError, + createNetworkResourceLabel, + isNetworkResourceLabel, + MAX_NETWORK_RESOURCE_LABEL_LENGTH, + NETWORK_ERROR_CODES, + NETWORK_STATIC_RESOURCE_LABELS, + NetworkDeadlineError, + NetworkDisposedError, + NetworkInvariantError, + NetworkOperationError, + NetworkQueueFullError, + NetworkSchedulerError, + sanitizeCapabilityTransportError, + type CapabilityTransportErrorCode, + type NetworkResourceLabel, +} from "./NetworkErrors"; + +describe("network resource labels", () => { + it.each([ + "feed metadata", + "episode media", + "provider operation", + `podnotes-feed-${"a".repeat(64)}`, + `podnotes-episode-${"0".repeat(64)}`, + ])("accepts the explicit target-free label %s", (value) => { + expect(isNetworkResourceLabel(value)).toBe(true); + expect(createNetworkResourceLabel(value)).toBe(value); + }); + + it.each([ + "", + "Feed metadata", + "sk-secret", + "private token", + "episode-media", + "provider operation 2", + "https://secret.example/feed?token=raw", + "secret.example", + "localhost:8080", + "feed/path", + "feed\\path", + "feed?token", + "feed#fragment", + "user@example", + "token=value", + "space doubled", + " leading", + "trailing ", + "one two three four five six seven", + "x".repeat(MAX_NETWORK_RESOURCE_LABEL_LENGTH + 1), + `podnotes-feed-${"g".repeat(64)}`, + `podnotes-episode-${"a".repeat(63)}`, + ])("rejects non-label input without echoing it: %s", (value) => { + expect(isNetworkResourceLabel(value)).toBe(false); + let caught: unknown; + try { + createNetworkResourceLabel(value); + } catch (error) { + caught = error; + } + expect(caught).toBeInstanceOf(TypeError); + if (value.length > 0) expect(String(caught)).not.toContain(value); + }); + + it("rejects non-string values", () => { + expect(isNetworkResourceLabel(undefined)).toBe(false); + expect(isNetworkResourceLabel({ toString: () => "feed metadata" })).toBe(false); + }); + + it("keeps the static application-label registry immutable", () => { + expect(Object.isFrozen(NETWORK_STATIC_RESOURCE_LABELS)).toBe(true); + expect(NETWORK_STATIC_RESOURCE_LABELS).toContain("podcast subscription"); + expect(() => + (NETWORK_STATIC_RESOURCE_LABELS as unknown as string[]).push("secret"), + ).toThrow(); + }); +}); + +describe("capability transport error authority", () => { + const label = createNetworkResourceLabel("feed metadata"); + + it("rejects unregistered runtime codes and labels before trusting an error", () => { + expect(() => + createCapabilityTransportError( + "NETWORK_INVENTED" as CapabilityTransportErrorCode, + label, + ), + ).toThrow(TypeError); + expect(() => + createCapabilityTransportError( + CAPABILITY_TRANSPORT_ERROR_CODES.operationFailed, + "https://secret.example/feed?token=raw" as NetworkResourceLabel, + ), + ).toThrow(TypeError); + }); + + it("sanitizes a directly constructed error that never passed the trusted mint", () => { + const forged = new CapabilityTransportError( + CAPABILITY_TRANSPORT_ERROR_CODES.responseTooLarge, + "https://secret.example/feed?token=raw" as NetworkResourceLabel, + ); + + const sanitized = sanitizeCapabilityTransportError(forged, label); + + expect(sanitized).not.toBe(forged); + expect(sanitized.code).toBe(CAPABILITY_TRANSPORT_ERROR_CODES.operationFailed); + expect(sanitized.resourceLabel).toBe(label); + expect(String(sanitized)).not.toContain("secret.example"); + }); +}); + +describe("network scheduler errors", () => { + const label = createNetworkResourceLabel("feed metadata"); + + it.each([ + [NetworkDeadlineError, "NetworkDeadlineError", NETWORK_ERROR_CODES.deadlineExceeded], + [NetworkDisposedError, "NetworkDisposedError", NETWORK_ERROR_CODES.sessionDisposed], + [NetworkQueueFullError, "NetworkQueueFullError", NETWORK_ERROR_CODES.queueFull], + [NetworkOperationError, "NetworkOperationError", NETWORK_ERROR_CODES.operationFailed], + [NetworkInvariantError, "NetworkInvariantError", NETWORK_ERROR_CODES.internalInvariant], + ] as const)("exposes only the stable %s contract", (ErrorType, name, code) => { + const error = new ErrorType(label); + + expect(error).toBeInstanceOf(NetworkSchedulerError); + expect(error.name).toBe(name); + expect(error.code).toBe(code); + expect(error.resourceLabel).toBe(label); + expect(error.message).toBe(`${code}: ${label}`); + expect(error).not.toHaveProperty("cause"); + }); + + it("keeps the stable code table immutable", () => { + expect(Object.isFrozen(NETWORK_ERROR_CODES)).toBe(true); + expect(() => Object.assign(NETWORK_ERROR_CODES, { queueFull: "CHANGED" })).toThrow(); + expect(NETWORK_ERROR_CODES.queueFull).toBe("NETWORK_QUEUE_FULL"); + }); +}); diff --git a/src/network/NetworkErrors.ts b/src/network/NetworkErrors.ts new file mode 100644 index 0000000..4bc695d --- /dev/null +++ b/src/network/NetworkErrors.ts @@ -0,0 +1,180 @@ +const HANDLE_RESOURCE_LABEL_PATTERN = /^podnotes-(?:feed|episode)-[0-9a-f]{64}$/; + +export const NETWORK_STATIC_RESOURCE_LABELS = Object.freeze([ + "podcast subscription", + "podcast artwork", + "podcast site", + "episode audio", + "episode chapters", + "episode artwork", + "episode page", + "feed metadata", + "episode media", + "provider operation", +] as const); + +const staticResourceLabels = new Set(NETWORK_STATIC_RESOURCE_LABELS); + +export const MAX_NETWORK_RESOURCE_LABEL_LENGTH = 96; + +declare const networkResourceLabelBrand: unique symbol; + +/** + * An application-owned description or opaque handle that is safe to expose in + * an error. Raw URLs, origins, paths, response data, and provider messages are + * deliberately outside this type's runtime grammar. + */ +export type NetworkResourceLabel = string & { + readonly [networkResourceLabelBrand]: true; +}; + +export const NETWORK_ERROR_CODES = Object.freeze({ + deadlineExceeded: "NETWORK_DEADLINE_EXCEEDED", + sessionDisposed: "NETWORK_SESSION_DISPOSED", + queueFull: "NETWORK_QUEUE_FULL", + operationFailed: "NETWORK_OPERATION_FAILED", + internalInvariant: "NETWORK_INTERNAL_INVARIANT", +} as const); + +export type NetworkErrorCode = (typeof NETWORK_ERROR_CODES)[keyof typeof NETWORK_ERROR_CODES]; + +export const CAPABILITY_TRANSPORT_ERROR_CODES = Object.freeze({ + resourceRejected: "NETWORK_RESOURCE_REJECTED", + targetRejected: "NETWORK_TARGET_REJECTED", + addressRejected: "NETWORK_ADDRESS_REJECTED", + adapterResponseInvalid: "NETWORK_ADAPTER_RESPONSE_INVALID", + redirectRejected: "NETWORK_REDIRECT_REJECTED", + redirectLimit: "NETWORK_REDIRECT_LIMIT", + statusRejected: "NETWORK_STATUS_REJECTED", + responseTooLarge: "NETWORK_RESPONSE_TOO_LARGE", + operationFailed: "NETWORK_OPERATION_FAILED", +} as const); + +export type CapabilityTransportErrorCode = + (typeof CAPABILITY_TRANSPORT_ERROR_CODES)[keyof typeof CAPABILITY_TRANSPORT_ERROR_CODES]; + +const capabilityTransportErrorCodes = new Set( + Object.values(CAPABILITY_TRANSPORT_ERROR_CODES), +); +const internallyIssuedCapabilityErrors = new WeakSet(); + +export function isNetworkResourceLabel(value: unknown): value is NetworkResourceLabel { + if (typeof value !== "string" || value.length > MAX_NETWORK_RESOURCE_LABEL_LENGTH) { + return false; + } + + return staticResourceLabels.has(value) || HANDLE_RESOURCE_LABEL_PATTERN.test(value); +} + +export function createNetworkResourceLabel(value: string): NetworkResourceLabel { + if (!isNetworkResourceLabel(value)) { + throw new TypeError( + "Network resource labels must be registered application labels or PodNotes handles.", + ); + } + return value; +} + +/** Stable target-free error from the capability-scoped transport boundary. */ +export class CapabilityTransportError extends Error { + constructor( + public readonly code: CapabilityTransportErrorCode, + public readonly resourceLabel: NetworkResourceLabel, + ) { + super(`${code}: ${resourceLabel}`); + this.name = "CapabilityTransportError"; + } +} + +/** @internal Creates errors that may safely cross an adapter boundary. */ +export function createCapabilityTransportError( + code: CapabilityTransportErrorCode, + resourceLabel: NetworkResourceLabel, +): CapabilityTransportError { + if (!capabilityTransportErrorCodes.has(code) || !isNetworkResourceLabel(resourceLabel)) { + throw new TypeError( + "Capability transport errors require a registered code and safe label.", + ); + } + const error = new CapabilityTransportError(code, resourceLabel); + internallyIssuedCapabilityErrors.add(error); + return Object.freeze(error); +} + +/** @internal Drops arbitrary resolver, adapter, and response failure details. */ +export function sanitizeCapabilityTransportError( + error: unknown, + resourceLabel: NetworkResourceLabel, +): CapabilityTransportError { + return typeof error === "object" && + error !== null && + internallyIssuedCapabilityErrors.has(error) + ? (error as CapabilityTransportError) + : createCapabilityTransportError( + CAPABILITY_TRANSPORT_ERROR_CODES.operationFailed, + resourceLabel, + ); +} + +/** @internal Shared cancellation check for bounded network operations. */ +export function assertNetworkOperationActive( + signal: AbortSignal, + resourceLabel: NetworkResourceLabel, +): void { + if (signal.aborted) { + throw createCapabilityTransportError( + CAPABILITY_TRANSPORT_ERROR_CODES.operationFailed, + resourceLabel, + ); + } +} + +/** + * Scheduler errors intentionally expose only a stable code and the explicit + * target-free label supplied by application code. Raw adapter failures are + * never attached as `cause` or interpolated into the message. + */ +export class NetworkSchedulerError extends Error { + constructor( + public readonly code: NetworkErrorCode, + public readonly resourceLabel: NetworkResourceLabel, + ) { + super(`${code}: ${resourceLabel}`); + this.name = "NetworkSchedulerError"; + } +} + +export class NetworkDeadlineError extends NetworkSchedulerError { + constructor(resourceLabel: NetworkResourceLabel) { + super(NETWORK_ERROR_CODES.deadlineExceeded, resourceLabel); + this.name = "NetworkDeadlineError"; + } +} + +export class NetworkDisposedError extends NetworkSchedulerError { + constructor(resourceLabel: NetworkResourceLabel) { + super(NETWORK_ERROR_CODES.sessionDisposed, resourceLabel); + this.name = "NetworkDisposedError"; + } +} + +export class NetworkQueueFullError extends NetworkSchedulerError { + constructor(resourceLabel: NetworkResourceLabel) { + super(NETWORK_ERROR_CODES.queueFull, resourceLabel); + this.name = "NetworkQueueFullError"; + } +} + +export class NetworkOperationError extends NetworkSchedulerError { + constructor(resourceLabel: NetworkResourceLabel) { + super(NETWORK_ERROR_CODES.operationFailed, resourceLabel); + this.name = "NetworkOperationError"; + } +} + +export class NetworkInvariantError extends NetworkSchedulerError { + constructor(resourceLabel: NetworkResourceLabel) { + super(NETWORK_ERROR_CODES.internalInvariant, resourceLabel); + this.name = "NetworkInvariantError"; + } +} diff --git a/src/network/NetworkScheduler.test.ts b/src/network/NetworkScheduler.test.ts new file mode 100644 index 0000000..4500b8b --- /dev/null +++ b/src/network/NetworkScheduler.test.ts @@ -0,0 +1,855 @@ +import { describe, expect, it, vi } from "vitest"; +import { + createNetworkResourceLabel, + NETWORK_ERROR_CODES, + NetworkDeadlineError, + NetworkDisposedError, + NetworkInvariantError, + NetworkOperationError, + NetworkQueueFullError, + type NetworkResourceLabel, +} from "./NetworkErrors"; +import { + DEFAULT_NETWORK_LANE_LIMITS, + MAX_NETWORK_DEADLINE_MS, + NetworkScheduler, + type NetworkClock, + type NetworkLane, + type NetworkSchedulerOptions, + type NetworkTask, +} from "./NetworkScheduler"; + +interface Deferred { + readonly promise: Promise; + resolve(value: T): void; + reject(error: unknown): void; +} + +function deferred(): Deferred { + let resolve!: (value: T) => void; + let reject!: (error: unknown) => void; + const promise = new Promise((resolvePromise, rejectPromise) => { + resolve = resolvePromise; + reject = rejectPromise; + }); + return { promise, resolve, reject }; +} + +class ManualClock implements NetworkClock { + private nextHandle = 1; + private nowMs = 0; + private callbacks = new Map void; dueAt: number; delayMs: number }>(); + + now(): number { + return this.nowMs; + } + + setTimeout(callback: () => void, delayMs: number): number { + const handle = this.nextHandle; + this.nextHandle += 1; + this.callbacks.set(handle, { callback, dueAt: this.nowMs + delayMs, delayMs }); + return handle; + } + + clearTimeout(handle: unknown): void { + this.callbacks.delete(handle as number); + } + + advance(milliseconds: number): void { + this.nowMs += milliseconds; + } + + fire(handle: number): void { + const scheduled = this.callbacks.get(handle); + this.callbacks.delete(handle); + scheduled?.callback(); + } + + fireAtDueTime(handle: number): void { + const scheduled = this.callbacks.get(handle); + if (!scheduled) return; + this.nowMs = Math.max(this.nowMs, scheduled.dueAt); + this.fire(handle); + } + + pendingHandles(): number[] { + return [...this.callbacks.keys()]; + } + + pendingDelays(): number[] { + return [...this.callbacks.values()].map(({ delayMs }) => delayMs); + } +} + +const FEED_LABEL = createNetworkResourceLabel("feed metadata"); +const EPISODE_LABEL = createNetworkResourceLabel("episode media"); +const PROVIDER_LABEL = createNetworkResourceLabel("provider operation"); + +function task( + operation: (signal: AbortSignal) => T | PromiseLike, + overrides: Partial, "operation">> = {}, +): NetworkTask { + return { + lane: "metadata", + resourceLabel: FEED_LABEL, + timeoutMs: 30_000, + operation, + ...overrides, + }; +} + +async function flushScheduler(): Promise { + await Promise.resolve(); + await Promise.resolve(); +} + +describe("NetworkScheduler construction and validation", () => { + it("publishes deeply frozen defaults and immutable snapshots", () => { + expect(Object.isFrozen(DEFAULT_NETWORK_LANE_LIMITS)).toBe(true); + expect(Object.isFrozen(DEFAULT_NETWORK_LANE_LIMITS.metadata)).toBe(true); + + const scheduler = new NetworkScheduler(); + const snapshot = scheduler.getSnapshot(); + expect(snapshot).toEqual({ + metadata: { active: 0, queued: 0, maxActive: 4, maxQueued: 512, failed: false }, + media: { active: 0, queued: 0, maxActive: 2, maxQueued: 32, failed: false }, + provider: { active: 0, queued: 0, maxActive: 2, maxQueued: 16, failed: false }, + }); + expect(Object.isFrozen(snapshot)).toBe(true); + expect(Object.isFrozen(snapshot.metadata)).toBe(true); + expect(() => Object.assign(snapshot.metadata, { maxActive: 99 })).toThrow(); + }); + + it("copies injected limits once and never lets later sessions reconfigure them", () => { + const limits = { + metadata: { maxActive: 1, maxQueued: 3 }, + media: { maxActive: 5, maxQueued: 7 }, + }; + const scheduler = new NetworkScheduler({ laneLimits: limits }); + const first = scheduler.createSession(); + + limits.metadata.maxActive = 100; + limits.metadata.maxQueued = 100; + const second = scheduler.createSession(); + + expect(first.getLaneSnapshot("metadata")).toEqual({ + active: 0, + queued: 0, + maxActive: 1, + maxQueued: 3, + failed: false, + }); + expect(second.getLaneSnapshot("media")).toMatchObject({ maxActive: 5, maxQueued: 7 }); + }); + + it("rejects inherited limit entries and inherited fields", () => { + const inheritedLimits = Object.create({ + metadata: { maxActive: 100, maxQueued: 100 }, + }) as Record; + const metadata = Object.create({ maxActive: 99 }) as { maxQueued: number }; + metadata.maxQueued = 1; + inheritedLimits.media = metadata; + + expect( + () => + new NetworkScheduler({ + laneLimits: inheritedLimits as NetworkSchedulerOptions["laneLimits"], + }), + ).toThrow("Network scheduler options are invalid"); + }); + + it.each([ + { laneLimits: null }, + { laneLimits: { metadata: null } }, + { laneLimits: { metdata: { maxActive: 1 } } }, + { laneLimits: { metadata: { maxActve: 1 } } }, + { laneLimits: { metadata: { maxActive: undefined } } }, + { unknown: true }, + ] as const)("rejects malformed scheduler configuration: %j", (options) => { + expect(() => new NetworkScheduler(options as unknown as NetworkSchedulerOptions)).toThrow( + TypeError, + ); + }); + + it.each([ + ["maxActive", 0], + ["maxActive", -1], + ["maxActive", 1.5], + ["maxActive", Number.NaN], + ["maxActive", Number.POSITIVE_INFINITY], + ["maxQueued", -1], + ["maxQueued", 1.5], + ["maxQueued", Number.NaN], + ["maxQueued", Number.POSITIVE_INFINITY], + ] as const)("rejects invalid lane limit %s=%s", (field, value) => { + expect( + () => + new NetworkScheduler({ + laneLimits: { metadata: { [field]: value } }, + }), + ).toThrow(TypeError); + }); + + it.each(["constructor", "toString", "__proto__", "Metadata", ""])( + "rejects hostile or unknown lane %s", + async (lane) => { + const session = new NetworkScheduler().createSession(); + const operation = vi.fn(); + + await expect( + session.schedule(task(operation, { lane: lane as NetworkLane })), + ).rejects.toThrow("Network task is invalid"); + expect(operation).not.toHaveBeenCalled(); + expect(() => session.getLaneSnapshot(lane as NetworkLane)).toThrow(TypeError); + }, + ); + + it("requires task fields to be own properties", async () => { + const operation = vi.fn(); + const inherited = Object.create({ lane: "metadata" }) as Record; + inherited.resourceLabel = FEED_LABEL; + inherited.timeoutMs = 1_000; + inherited.operation = operation; + const session = new NetworkScheduler().createSession(); + + await expect(session.schedule(inherited as unknown as NetworkTask)).rejects.toThrow( + "Network task is invalid", + ); + expect(operation).not.toHaveBeenCalled(); + }); + + it("rejects accessor and proxy task traps without exposing their errors", async () => { + const sentinel = "https://private.example/feed?token=TASK-SECRET"; + let accessorRead = false; + const accessorTask = task(() => undefined) as unknown as Record; + Object.defineProperty(accessorTask, "lane", { + enumerable: true, + get: () => { + accessorRead = true; + return "metadata"; + }, + }); + const trappedTask = new Proxy( + task(() => undefined), + { + ownKeys: () => { + throw new Error(sentinel); + }, + }, + ); + const session = new NetworkScheduler().createSession(); + + for (const hostile of [accessorTask, trappedTask]) { + const error = await session + .schedule(hostile as unknown as NetworkTask) + .catch((caught: unknown) => caught); + expect(error).toBeInstanceOf(TypeError); + expect(String(error)).toBe("TypeError: Network task is invalid."); + expect(String(error)).not.toContain(sentinel); + } + expect(accessorRead).toBe(false); + }); + + it.each([0, -1, 1.5, Number.NaN, Number.POSITIVE_INFINITY, MAX_NETWORK_DEADLINE_MS + 1])( + "rejects unsafe deadline %s without starting", + async (timeoutMs) => { + const operation = vi.fn(); + const session = new NetworkScheduler().createSession(); + await expect(session.schedule(task(operation, { timeoutMs }))).rejects.toThrow( + TypeError, + ); + expect(operation).not.toHaveBeenCalled(); + }, + ); + + it("accepts the maximum safe timer delay without truncation", async () => { + const clock = new ManualClock(); + const pending = deferred(); + const session = new NetworkScheduler({ clock }).createSession(); + const scheduled = session.schedule( + task(() => pending.promise, { timeoutMs: MAX_NETWORK_DEADLINE_MS }), + ); + + expect(clock.pendingDelays()).toEqual([MAX_NETWORK_DEADLINE_MS]); + session.dispose(); + await expect(scheduled).rejects.toBeInstanceOf(NetworkDisposedError); + pending.resolve(); + await flushScheduler(); + }); + + it("snapshots the injected clock methods", async () => { + const clock = new ManualClock(); + const originalSetTimeout = clock.setTimeout; + const scheduler = new NetworkScheduler({ clock }); + clock.setTimeout = vi.fn(() => { + throw new Error("mutated clock"); + }); + const session = scheduler.createSession(); + + await expect(session.schedule(task(() => "ok"))).resolves.toBe("ok"); + expect(clock.setTimeout).not.toHaveBeenCalled(); + expect(originalSetTimeout).not.toBe(clock.setTimeout); + }); + + it("turns timer setup failure into a stable invariant rejection", async () => { + const operation = vi.fn(); + const clock: NetworkClock = { + now: () => 0, + setTimeout: () => { + throw new Error("timer internals"); + }, + clearTimeout: () => undefined, + }; + const session = new NetworkScheduler({ clock }).createSession(); + + await expect(session.schedule(task(operation))).rejects.toBeInstanceOf( + NetworkInvariantError, + ); + expect(operation).not.toHaveBeenCalled(); + expect(session.getLaneSnapshot("metadata")).toMatchObject({ + active: 0, + queued: 0, + failed: true, + }); + }); + + it("sanitizes an initial clock-read failure", async () => { + const sentinel = "https://private.example/feed?token=CLOCK-SECRET"; + const clock: NetworkClock = { + now: () => { + throw new Error(sentinel); + }, + setTimeout: () => 1, + clearTimeout: () => undefined, + }; + const session = new NetworkScheduler({ clock }).createSession(); + const error = await session + .schedule(task(() => undefined)) + .catch((caught: unknown) => caught); + + expect(error).toBeInstanceOf(NetworkInvariantError); + expect(String(error)).not.toContain(sentinel); + }); + + it("settles and releases capacity when the clock fails during completion", async () => { + let reads = 0; + const clock: NetworkClock = { + now: () => (++reads <= 2 ? 0 : Number.NaN), + setTimeout: () => 1, + clearTimeout: () => undefined, + }; + const operation = vi.fn(() => "value"); + const session = new NetworkScheduler({ clock }).createSession(); + + await expect(session.schedule(task(operation))).rejects.toBeInstanceOf( + NetworkInvariantError, + ); + expect(operation).toHaveBeenCalledOnce(); + await flushScheduler(); + expect(session.getLaneSnapshot("metadata")).toMatchObject({ + active: 0, + queued: 0, + failed: true, + }); + await expect(session.schedule(task(() => "must not run"))).rejects.toBeInstanceOf( + NetworkInvariantError, + ); + }); + + it("fails a lane closed while retaining every permit until its adapter settles", async () => { + let clockFailed = false; + const clock: NetworkClock = { + now: () => (clockFailed ? Number.NaN : 0), + setTimeout: () => 1, + clearTimeout: () => undefined, + }; + const firstGate = deferred(); + const secondGate = deferred(); + const queuedOperation = vi.fn(); + const signals: AbortSignal[] = []; + const scheduler = new NetworkScheduler({ + clock, + laneLimits: { metadata: { maxActive: 2, maxQueued: 1 } }, + }); + const firstSession = scheduler.createSession(); + const secondSession = scheduler.createSession(); + const first = firstSession.schedule( + task((signal) => { + signals.push(signal); + return firstGate.promise; + }), + ); + const second = secondSession.schedule( + task((signal) => { + signals.push(signal); + return secondGate.promise; + }), + ); + const queued = firstSession.schedule(task(queuedOperation)); + const observed = [first, second, queued].map((promise) => + promise.catch((error: unknown) => error), + ); + + clockFailed = true; + firstGate.resolve(); + const errors = await Promise.all(observed); + expect(errors.every((error) => error instanceof NetworkInvariantError)).toBe(true); + expect(signals).toHaveLength(2); + expect(signals.every((signal) => signal.aborted)).toBe(true); + expect(queuedOperation).not.toHaveBeenCalled(); + await flushScheduler(); + expect(scheduler.getLaneSnapshot("metadata")).toMatchObject({ + active: 1, + queued: 0, + failed: true, + }); + + secondGate.resolve(); + await flushScheduler(); + expect(scheduler.getLaneSnapshot("metadata")).toMatchObject({ + active: 0, + queued: 0, + failed: true, + }); + await expect(firstSession.schedule(task(() => undefined))).rejects.toBeInstanceOf( + NetworkInvariantError, + ); + }); + + it("does not strand a caller when timer cleanup throws", async () => { + const clock: NetworkClock = { + now: () => 0, + setTimeout: () => 1, + clearTimeout: () => { + throw new Error("timer cleanup internals"); + }, + }; + const session = new NetworkScheduler({ clock }).createSession(); + + await expect(session.schedule(task(() => "value"))).resolves.toBe("value"); + await flushScheduler(); + expect(session.getLaneSnapshot("metadata")).toMatchObject({ active: 0, queued: 0 }); + }); +}); + +describe("NetworkScheduler lanes and queueing", () => { + it("keeps all three lanes independent", async () => { + const clock = new ManualClock(); + const pending = [deferred(), deferred(), deferred()]; + const started: NetworkLane[] = []; + const scheduler = new NetworkScheduler({ + clock, + laneLimits: { + metadata: { maxActive: 1 }, + media: { maxActive: 1 }, + provider: { maxActive: 1 }, + }, + }); + const session = scheduler.createSession(); + const lanes = ["metadata", "media", "provider"] as const; + + const promises = lanes.map((lane, index) => + session.schedule( + task( + () => { + started.push(lane); + return pending[index].promise; + }, + { + lane, + resourceLabel: + lane === "metadata" + ? FEED_LABEL + : lane === "media" + ? EPISODE_LABEL + : PROVIDER_LABEL, + }, + ), + ), + ); + + expect(started).toEqual(lanes); + expect(scheduler.getSnapshot()).toMatchObject({ + metadata: { active: 1, queued: 0 }, + media: { active: 1, queued: 0 }, + provider: { active: 1, queued: 0 }, + }); + pending.forEach((entry, index) => entry.resolve(String(index))); + await Promise.all(promises); + await flushScheduler(); + }); + + it("runs each lane FIFO across sessions", async () => { + const clock = new ManualClock(); + const scheduler = new NetworkScheduler({ + clock, + laneLimits: { metadata: { maxActive: 1, maxQueued: 3 } }, + }); + const firstSession = scheduler.createSession(); + const secondSession = scheduler.createSession(); + const gates = [deferred(), deferred(), deferred()]; + const started: number[] = []; + const operation = (index: number) => () => { + started.push(index); + return gates[index].promise; + }; + + const first = firstSession.schedule(task(operation(0))); + const second = secondSession.schedule(task(operation(1))); + const third = firstSession.schedule(task(operation(2))); + expect(started).toEqual([0]); + expect(scheduler.getLaneSnapshot("metadata")).toMatchObject({ active: 1, queued: 2 }); + + gates[0].resolve(0); + await first; + await flushScheduler(); + expect(started).toEqual([0, 1]); + gates[1].resolve(1); + await second; + await flushScheduler(); + expect(started).toEqual([0, 1, 2]); + gates[2].resolve(2); + await third; + await flushScheduler(); + expect(scheduler.getLaneSnapshot("metadata")).toMatchObject({ active: 0, queued: 0 }); + }); + + it("bounds each queue, including a zero-queue lane", async () => { + const activeGate = deferred(); + const queuedGate = deferred(); + const oneQueued = new NetworkScheduler({ + clock: new ManualClock(), + laneLimits: { metadata: { maxActive: 1, maxQueued: 1 } }, + }).createSession(); + const active = oneQueued.schedule(task(() => activeGate.promise)); + const queued = oneQueued.schedule(task(() => queuedGate.promise)); + + await expect(oneQueued.schedule(task(() => undefined))).rejects.toBeInstanceOf( + NetworkQueueFullError, + ); + activeGate.resolve(); + await active; + await flushScheduler(); + queuedGate.resolve(); + await queued; + + const noQueueGate = deferred(); + const noQueue = new NetworkScheduler({ + clock: new ManualClock(), + laneLimits: { metadata: { maxActive: 1, maxQueued: 0 } }, + }).createSession(); + const sole = noQueue.schedule(task(() => noQueueGate.promise)); + await expect(noQueue.schedule(task(() => undefined))).rejects.toBeInstanceOf( + NetworkQueueFullError, + ); + noQueueGate.resolve(); + await sole; + }); + + it("defensively copies queued task fields before caller mutation", async () => { + const clock = new ManualClock(); + const scheduler = new NetworkScheduler({ + clock, + laneLimits: { metadata: { maxActive: 1, maxQueued: 1 } }, + }); + const session = scheduler.createSession(); + const firstGate = deferred(); + const originalOperation = vi.fn(() => "original"); + const replacementOperation = vi.fn(() => "replacement"); + const active = session.schedule(task(() => firstGate.promise, { timeoutMs: 60_000 })); + const mutableTask: NetworkTask = { + lane: "metadata", + resourceLabel: FEED_LABEL, + timeoutMs: 30_000, + operation: originalOperation, + }; + const queued = session.schedule(mutableTask); + + const callerOwned = mutableTask as unknown as { + lane: NetworkLane; + resourceLabel: NetworkResourceLabel; + timeoutMs: number; + operation: (signal: AbortSignal) => string; + }; + callerOwned.lane = "media"; + callerOwned.resourceLabel = EPISODE_LABEL; + callerOwned.timeoutMs = 1; + callerOwned.operation = replacementOperation; + firstGate.resolve(); + await active; + await expect(queued).resolves.toBe("original"); + expect(originalOperation).toHaveBeenCalledOnce(); + expect(replacementOperation).not.toHaveBeenCalled(); + expect(scheduler.getLaneSnapshot("media")).toMatchObject({ active: 0, queued: 0 }); + }); +}); + +describe("NetworkScheduler deadlines and cancellation", () => { + it("aborts timed-out active work but holds its permit until adapter settlement", async () => { + const clock = new ManualClock(); + const firstGate = deferred(); + const secondGate = deferred(); + let firstSignal: AbortSignal | undefined; + const secondOperation = vi.fn(() => secondGate.promise); + const scheduler = new NetworkScheduler({ + clock, + laneLimits: { metadata: { maxActive: 1, maxQueued: 1 } }, + }); + const session = scheduler.createSession(); + const timedOut = session.schedule( + task((signal) => { + firstSignal = signal; + return firstGate.promise; + }), + ); + const queued = session.schedule(task(secondOperation, { timeoutMs: 60_000 })); + const [firstDeadline] = clock.pendingHandles(); + + clock.fireAtDueTime(firstDeadline); + await expect(timedOut).rejects.toBeInstanceOf(NetworkDeadlineError); + expect(firstSignal?.aborted).toBe(true); + expect(scheduler.getLaneSnapshot("metadata")).toMatchObject({ active: 1, queued: 1 }); + expect(secondOperation).not.toHaveBeenCalled(); + + firstGate.resolve("late"); + await flushScheduler(); + expect(secondOperation).toHaveBeenCalledOnce(); + secondGate.resolve("next"); + await expect(queued).resolves.toBe("next"); + await flushScheduler(); + expect(scheduler.getLaneSnapshot("metadata").active).toBe(0); + }); + + it("removes queued work at its deadline without creating an AbortController", async () => { + const clock = new ManualClock(); + const activeGate = deferred(); + const queuedOperation = vi.fn(); + const scheduler = new NetworkScheduler({ + clock, + laneLimits: { metadata: { maxActive: 1, maxQueued: 1 } }, + }); + const session = scheduler.createSession(); + const active = session.schedule(task(() => activeGate.promise, { timeoutMs: 60_000 })); + const queued = session.schedule(task(queuedOperation)); + const [, queuedDeadline] = clock.pendingHandles(); + + clock.fireAtDueTime(queuedDeadline); + await expect(queued).rejects.toBeInstanceOf(NetworkDeadlineError); + expect(queuedOperation).not.toHaveBeenCalled(); + expect(scheduler.getLaneSnapshot("metadata")).toMatchObject({ active: 1, queued: 0 }); + activeGate.resolve(); + await active; + await flushScheduler(); + expect(queuedOperation).not.toHaveBeenCalled(); + }); + + it("checks absolute deadlines when timer delivery is late", async () => { + const clock = new ManualClock(); + const activeGate = deferred(); + const queuedOperation = vi.fn(() => "must not run"); + let activeSignal: AbortSignal | undefined; + const scheduler = new NetworkScheduler({ + clock, + laneLimits: { metadata: { maxActive: 1, maxQueued: 1 } }, + }); + const session = scheduler.createSession(); + const active = session.schedule( + task((signal) => { + activeSignal = signal; + return activeGate.promise; + }), + ); + const queued = session.schedule(task(queuedOperation)); + + clock.advance(30_001); + activeGate.resolve("late result"); + await expect(active).rejects.toBeInstanceOf(NetworkDeadlineError); + await expect(queued).rejects.toBeInstanceOf(NetworkDeadlineError); + expect(activeSignal?.aborted).toBe(true); + expect(queuedOperation).not.toHaveBeenCalled(); + await flushScheduler(); + expect(clock.pendingHandles()).toEqual([]); + expect(scheduler.getLaneSnapshot("metadata")).toMatchObject({ active: 0, queued: 0 }); + }); + + it("re-arms an early timer instead of expiring work early", async () => { + const clock = new ManualClock(); + const gate = deferred(); + let signal: AbortSignal | undefined; + const session = new NetworkScheduler({ clock }).createSession(); + const scheduled = session.schedule( + task((activeSignal) => { + signal = activeSignal; + return gate.promise; + }), + ); + const [firstTimer] = clock.pendingHandles(); + + clock.fire(firstTimer); + expect(signal?.aborted).toBe(false); + expect(clock.pendingDelays()).toEqual([30_000]); + gate.resolve("on time"); + await expect(scheduled).resolves.toBe("on time"); + expect(clock.pendingHandles()).toEqual([]); + }); + + it("makes the deadline win when settlement arrives at the exact boundary", async () => { + const clock = new ManualClock(); + const gate = deferred(); + let signal: AbortSignal | undefined; + const session = new NetworkScheduler({ clock }).createSession(); + const scheduled = session.schedule( + task((activeSignal) => { + signal = activeSignal; + return gate.promise; + }), + ); + + clock.advance(30_000); + gate.resolve("boundary"); + await expect(scheduled).rejects.toBeInstanceOf(NetworkDeadlineError); + expect(signal?.aborted).toBe(true); + }); + + it("creates a distinct AbortController for every active operation", async () => { + const gates = [deferred(), deferred()]; + const signals: AbortSignal[] = []; + const scheduler = new NetworkScheduler({ + clock: new ManualClock(), + laneLimits: { metadata: { maxActive: 2 } }, + }); + const session = scheduler.createSession(); + const promises = gates.map((gate) => + session.schedule( + task((signal) => { + signals.push(signal); + return gate.promise; + }), + ), + ); + + expect(signals).toHaveLength(2); + expect(signals[0]).not.toBe(signals[1]); + session.dispose(); + expect(signals.every((signal) => signal.aborted)).toBe(true); + await Promise.all(promises.map((promise) => promise.catch((error: unknown) => error))); + gates.forEach((gate) => gate.resolve()); + await flushScheduler(); + }); + + it("releases one permit exactly once after abort-driven rejection", async () => { + const clock = new ManualClock(); + const adapter = deferred(); + const scheduler = new NetworkScheduler({ + clock, + laneLimits: { metadata: { maxActive: 1, maxQueued: 1 } }, + }); + const session = scheduler.createSession(); + const first = session.schedule( + task((signal) => { + signal.addEventListener("abort", () => adapter.reject(new Error("aborted")), { + once: true, + }); + return adapter.promise; + }), + ); + const next = session.schedule(task(() => "next", { timeoutMs: 60_000 })); + const [deadline] = clock.pendingHandles(); + + clock.fireAtDueTime(deadline); + await expect(first).rejects.toBeInstanceOf(NetworkDeadlineError); + await expect(next).resolves.toBe("next"); + await flushScheduler(); + expect(scheduler.getLaneSnapshot("metadata")).toMatchObject({ active: 0, queued: 0 }); + }); +}); + +describe("NetworkScheduler disposal and error boundaries", () => { + it("disposes only one owner while preserving shared capacity and FIFO", async () => { + const clock = new ManualClock(); + const oldGate = deferred(); + let oldSignal: AbortSignal | undefined; + const otherOperation = vi.fn(() => "other"); + const removedOperation = vi.fn(() => "removed"); + const scheduler = new NetworkScheduler({ + clock, + laneLimits: { metadata: { maxActive: 1, maxQueued: 3 } }, + }); + const oldSession = scheduler.createSession(); + const otherSession = scheduler.createSession(); + const active = oldSession.schedule( + task((signal) => { + oldSignal = signal; + return oldGate.promise; + }), + ); + const other = otherSession.schedule(task(otherOperation, { timeoutMs: 60_000 })); + const removed = oldSession.schedule(task(removedOperation, { timeoutMs: 60_000 })); + + oldSession.dispose(); + oldSession.dispose(); + expect(oldSession.isDisposed()).toBe(true); + await expect(active).rejects.toBeInstanceOf(NetworkDisposedError); + await expect(removed).rejects.toBeInstanceOf(NetworkDisposedError); + expect(oldSignal?.aborted).toBe(true); + expect(removedOperation).not.toHaveBeenCalled(); + expect(otherOperation).not.toHaveBeenCalled(); + expect(scheduler.getLaneSnapshot("metadata")).toMatchObject({ active: 1, queued: 1 }); + await expect(oldSession.schedule(task(() => "new"))).rejects.toBeInstanceOf( + NetworkDisposedError, + ); + + oldGate.resolve(); + await expect(other).resolves.toBe("other"); + expect(otherOperation).toHaveBeenCalledOnce(); + await flushScheduler(); + expect(scheduler.getLaneSnapshot("metadata")).toMatchObject({ active: 0, queued: 0 }); + }); + + it.each([ + [ + "sync", + () => { + throw new Error("https://secret.example/feed?token=raw response body"); + }, + ], + ["async", () => Promise.reject(new Error("provider key sk-secret at https://api.example"))], + ] as const)( + "sanitizes %s adapter failures without retaining a raw cause", + async (_kind, operation) => { + const session = new NetworkScheduler({ clock: new ManualClock() }).createSession(); + const error = await session + .schedule(task(operation)) + .catch((caught: unknown) => caught); + + expect(error).toBeInstanceOf(NetworkOperationError); + expect(error).toMatchObject({ + code: NETWORK_ERROR_CODES.operationFailed, + resourceLabel: FEED_LABEL, + }); + expect(String(error)).toBe( + `NetworkOperationError: NETWORK_OPERATION_FAILED: ${FEED_LABEL}`, + ); + expect(JSON.stringify(error)).not.toContain("secret"); + expect(error).not.toHaveProperty("cause"); + }, + ); + + it("keeps timeout errors stable after ignored late success and failure", async () => { + const clock = new ManualClock(); + const gate = deferred(); + const scheduler = new NetworkScheduler({ clock }); + const session = scheduler.createSession(); + const scheduled = session.schedule(task(() => gate.promise)); + const observed = scheduled.catch((error: unknown) => error); + const [deadline] = clock.pendingHandles(); + + clock.fireAtDueTime(deadline); + const error = await observed; + expect(error).toBeInstanceOf(NetworkDeadlineError); + gate.resolve("late secret response"); + await flushScheduler(); + expect(error).toMatchObject({ code: NETWORK_ERROR_CODES.deadlineExceeded }); + expect(scheduler.getLaneSnapshot("metadata").active).toBe(0); + }); +}); diff --git a/src/network/NetworkScheduler.ts b/src/network/NetworkScheduler.ts new file mode 100644 index 0000000..69f88f9 --- /dev/null +++ b/src/network/NetworkScheduler.ts @@ -0,0 +1,631 @@ +import { + createNetworkResourceLabel, + NetworkDeadlineError, + NetworkDisposedError, + NetworkInvariantError, + NetworkOperationError, + NetworkQueueFullError, + type NetworkResourceLabel, +} from "./NetworkErrors"; + +export const NETWORK_LANES = Object.freeze(["metadata", "media", "provider"] as const); +export type NetworkLane = (typeof NETWORK_LANES)[number]; + +export interface NetworkLaneLimit { + readonly maxActive: number; + readonly maxQueued: number; +} + +export type NetworkLaneLimits = Readonly>; + +export const DEFAULT_NETWORK_LANE_LIMITS: NetworkLaneLimits = Object.freeze({ + metadata: Object.freeze({ maxActive: 4, maxQueued: 512 }), + media: Object.freeze({ maxActive: 2, maxQueued: 32 }), + provider: Object.freeze({ maxActive: 2, maxQueued: 16 }), +}); + +/** The largest delay that browsers and Node schedule without 32-bit overflow. */ +export const MAX_NETWORK_DEADLINE_MS = 2_147_483_647; + +export interface NetworkClock { + now(): number; + setTimeout(callback: () => void, delayMs: number): unknown; + clearTimeout(handle: unknown): void; +} + +export interface NetworkSchedulerOptions { + readonly clock?: NetworkClock; + readonly laneLimits?: Partial>>; +} + +export interface NetworkTask { + readonly lane: NetworkLane; + readonly resourceLabel: NetworkResourceLabel; + readonly timeoutMs: number; + readonly operation: (signal: AbortSignal) => T | PromiseLike; +} + +export interface NetworkLaneSnapshot { + readonly active: number; + readonly queued: number; + readonly maxActive: number; + readonly maxQueued: number; + readonly failed: boolean; +} + +export type NetworkSchedulerSnapshot = Readonly>; + +interface Deferred { + readonly promise: Promise; + resolve(value: T): void; + reject(error: unknown): void; +} + +interface SessionState { + disposed: boolean; + readonly activeJobs: Set; +} + +interface PreparedTask { + readonly lane: NetworkLane; + readonly resourceLabel: NetworkResourceLabel; + readonly timeoutMs: number; + readonly operation: (signal: AbortSignal) => unknown | PromiseLike; +} + +interface ScheduledJob { + readonly owner: SessionState; + readonly task: PreparedTask; + readonly deferred: Deferred; + readonly deadlineAt: number; + callerSettled: boolean; + deadlineTimerArmed: boolean; + permitState: "none" | "held" | "released"; + deadlineTimer?: unknown; + abortController?: AbortController; +} + +interface LaneState { + active: number; + failed: boolean; + readonly activeJobs: Set; + readonly queue: ScheduledJob[]; + readonly limit: NetworkLaneLimit; +} + +type NetworkLaneLimitOverrides = Partial>>; + +const defaultClock: NetworkClock = { + now: () => performance.now(), + setTimeout: (callback, delayMs) => window.setTimeout(callback, delayMs), + clearTimeout: (handle) => window.clearTimeout(handle as number), +}; + +const OWN = Object.prototype.hasOwnProperty; +const NETWORK_LANE_MEMBERSHIP = Object.freeze({ + metadata: true, + media: true, + provider: true, +}); + +/** + * A shared, transport-independent capacity scheduler. Construct exactly one for + * a plugin generation and create one session per lifecycle owner. Disposing a + * session cannot reset permits still held by another session or by an adapter + * that ignored cancellation. + */ +export class NetworkScheduler { + private readonly clock: NetworkClock; + private readonly lanes: Record; + + constructor(options: NetworkSchedulerOptions = {}) { + const { clock, laneLimits } = snapshotSchedulerOptions(options); + this.clock = snapshotClock(clock ?? defaultClock); + this.lanes = { + metadata: createLaneState("metadata", laneLimits), + media: createLaneState("media", laneLimits), + provider: createLaneState("provider", laneLimits), + }; + } + + createSession(): NetworkSchedulerSession { + return new NetworkSchedulerSession(this, { + disposed: false, + activeJobs: new Set(), + }); + } + + getLaneSnapshot(lane: NetworkLane): NetworkLaneSnapshot { + const laneName = validateLane(lane); + const state = this.lanes[laneName]; + return Object.freeze({ + active: state.active, + queued: state.queue.length, + maxActive: state.limit.maxActive, + maxQueued: state.limit.maxQueued, + failed: state.failed, + }); + } + + getSnapshot(): NetworkSchedulerSnapshot { + return Object.freeze({ + metadata: this.getLaneSnapshot("metadata"), + media: this.getLaneSnapshot("media"), + provider: this.getLaneSnapshot("provider"), + }); + } + + /** @internal Called only by sessions created by this scheduler. */ + schedule(owner: SessionState, options: NetworkTask): Promise { + let task: PreparedTask; + try { + task = prepareTask(options); + } catch (error) { + return Promise.reject(error); + } + + if (owner.disposed) { + return Promise.reject(new NetworkDisposedError(task.resourceLabel)); + } + + const lane = this.lanes[task.lane]; + if (lane.failed) { + return Promise.reject(new NetworkInvariantError(task.resourceLabel)); + } + if (lane.active >= lane.limit.maxActive && lane.queue.length >= lane.limit.maxQueued) { + return Promise.reject(new NetworkQueueFullError(task.resourceLabel)); + } + + let now: number; + try { + now = readClock(this.clock); + } catch { + return Promise.reject(new NetworkInvariantError(task.resourceLabel)); + } + + const job: ScheduledJob = { + owner, + task, + deferred: createDeferred(), + deadlineAt: now + task.timeoutMs, + callerSettled: false, + deadlineTimerArmed: false, + permitState: "none", + }; + if (!this.armDeadline(job, task.timeoutMs)) { + this.failLane(job); + return job.deferred.promise as Promise; + } + + if (lane.active < lane.limit.maxActive) { + this.start(job); + } else { + lane.queue.push(job); + } + + return job.deferred.promise as Promise; + } + + /** @internal Called only by sessions created by this scheduler. */ + disposeSession(owner: SessionState): void { + if (owner.disposed) return; + owner.disposed = true; + + for (const lane of Object.values(this.lanes)) { + for (let index = lane.queue.length - 1; index >= 0; index -= 1) { + const job = lane.queue[index]; + if (job.owner !== owner) continue; + lane.queue.splice(index, 1); + this.rejectCaller(job, new NetworkDisposedError(job.task.resourceLabel)); + } + } + + for (const job of owner.activeJobs) { + this.abort(job); + this.rejectCaller(job, new NetworkDisposedError(job.task.resourceLabel)); + } + } + + private start(job: ScheduledJob): void { + if (job.owner.disposed) { + this.rejectCaller(job, new NetworkDisposedError(job.task.resourceLabel)); + return; + } + const lane = this.lanes[job.task.lane]; + if (lane.failed) { + this.rejectCaller(job, new NetworkInvariantError(job.task.resourceLabel)); + return; + } + + let now: number; + try { + now = readClock(this.clock); + } catch { + this.failLane(job); + return; + } + if (now >= job.deadlineAt) { + this.expire(job); + return; + } + + lane.active += 1; + job.owner.activeJobs.add(job); + lane.activeJobs.add(job); + job.permitState = "held"; + job.abortController = new AbortController(); + + let adapterPromise: Promise; + try { + adapterPromise = Promise.resolve(job.task.operation(job.abortController.signal)); + } catch { + adapterPromise = Promise.reject(); + } + + void adapterPromise + .then( + (value) => this.complete(job, value), + () => this.fail(job), + ) + .finally(() => this.release(job)) + .catch(() => this.failLane(job)); + } + + private complete(job: ScheduledJob, value: unknown): void { + if (job.callerSettled) return; + let now: number; + try { + now = readClock(this.clock); + } catch { + this.failLane(job); + return; + } + if (now >= job.deadlineAt) { + this.expire(job); + return; + } + this.resolveCaller(job, value); + } + + private fail(job: ScheduledJob): void { + if (job.callerSettled) return; + let now: number; + try { + now = readClock(this.clock); + } catch { + this.failLane(job); + return; + } + if (now >= job.deadlineAt) { + this.expire(job); + return; + } + this.rejectCaller(job, new NetworkOperationError(job.task.resourceLabel)); + } + + private expire(job: ScheduledJob): void { + if (job.callerSettled) return; + + let now: number; + try { + now = readClock(this.clock); + } catch { + this.failLane(job); + return; + } + if (now < job.deadlineAt) { + if (!this.armDeadline(job, job.deadlineAt - now)) { + this.failLane(job); + } + return; + } + + this.removeQueuedOrAbort(job); + this.rejectCaller(job, new NetworkDeadlineError(job.task.resourceLabel)); + } + + private release(job: ScheduledJob): void { + const lane = this.lanes[job.task.lane]; + if (lane.failed) { + this.releaseFailedLanePermit(job); + return; + } + if ( + job.permitState !== "held" || + !job.owner.activeJobs.has(job) || + !lane.activeJobs.has(job) || + lane.active <= 0 + ) { + this.failLane(job); + return; + } + job.owner.activeJobs.delete(job); + lane.activeJobs.delete(job); + job.permitState = "released"; + lane.active -= 1; + this.clearDeadline(job); + this.drain(job.task.lane); + } + + private drain(laneName: NetworkLane): void { + const lane = this.lanes[laneName]; + while (!lane.failed && lane.active < lane.limit.maxActive && lane.queue.length > 0) { + const next = lane.queue.shift(); + if (next) this.start(next); + } + } + + private armDeadline(job: ScheduledJob, delayMs: number): boolean { + this.clearDeadline(job); + try { + job.deadlineTimer = this.clock.setTimeout( + () => this.expire(job), + Math.min(MAX_NETWORK_DEADLINE_MS, Math.max(0, delayMs)), + ); + job.deadlineTimerArmed = true; + return true; + } catch { + job.deadlineTimer = undefined; + job.deadlineTimerArmed = false; + return false; + } + } + + private clearDeadline(job: ScheduledJob): void { + if (!job.deadlineTimerArmed) return; + const handle = job.deadlineTimer; + job.deadlineTimerArmed = false; + job.deadlineTimer = undefined; + try { + this.clock.clearTimeout(handle); + } catch { + // Timer cleanup failure cannot strand or re-settle the caller. + } + } + + private abort(job: ScheduledJob): void { + if (!job.abortController || job.abortController.signal.aborted) return; + job.abortController.abort(); + } + + private resolveCaller(job: ScheduledJob, value: unknown): void { + if (job.callerSettled) return; + job.callerSettled = true; + this.clearDeadline(job); + job.deferred.resolve(value); + } + + private rejectCaller(job: ScheduledJob, error: unknown): void { + if (job.callerSettled) return; + job.callerSettled = true; + this.clearDeadline(job); + job.deferred.reject(error); + } + + private removeQueuedOrAbort(job: ScheduledJob): void { + if (job.abortController) { + this.abort(job); + return; + } + const queue = this.lanes[job.task.lane].queue; + const queueIndex = queue.indexOf(job); + if (queueIndex !== -1) queue.splice(queueIndex, 1); + } + + private failLane(job: ScheduledJob): void { + try { + const lane = this.lanes[job.task.lane]; + lane.failed = true; + for (const active of lane.activeJobs) { + this.abort(active); + this.rejectCaller(active, new NetworkInvariantError(active.task.resourceLabel)); + this.clearDeadline(active); + } + for (const queued of lane.queue.splice(0)) { + this.rejectCaller(queued, new NetworkInvariantError(queued.task.resourceLabel)); + } + this.rejectCaller(job, new NetworkInvariantError(job.task.resourceLabel)); + this.clearDeadline(job); + } catch { + // The lane remains failed and the terminal promise chain stays observed. + } + } + + private releaseFailedLanePermit(job: ScheduledJob): void { + if (job.permitState !== "held") return; + job.permitState = "released"; + job.owner.activeJobs.delete(job); + const lane = this.lanes[job.task.lane]; + lane.activeJobs.delete(job); + if (lane.active > 0) lane.active -= 1; + this.clearDeadline(job); + } +} + +export class NetworkSchedulerSession { + constructor( + private readonly scheduler: NetworkScheduler, + private readonly state: SessionState, + ) {} + + schedule(task: NetworkTask): Promise { + return this.scheduler.schedule(this.state, task); + } + + dispose(): void { + this.scheduler.disposeSession(this.state); + } + + isDisposed(): boolean { + return this.state.disposed; + } + + getLaneSnapshot(lane: NetworkLane): NetworkLaneSnapshot { + return this.scheduler.getLaneSnapshot(lane); + } + + getSnapshot(): NetworkSchedulerSnapshot { + return this.scheduler.getSnapshot(); + } +} + +function prepareTask(options: NetworkTask): PreparedTask { + try { + const record = strictPlainDataRecord(options, [ + "lane", + "resourceLabel", + "timeoutMs", + "operation", + ]); + for (const property of ["lane", "resourceLabel", "timeoutMs", "operation"] as const) { + if (!OWN.call(record, property)) throw new Error(); + } + + const lane = validateLane(record.lane); + const resourceLabel = createNetworkResourceLabel(record.resourceLabel as string); + const timeoutMs = record.timeoutMs; + if ( + !Number.isSafeInteger(timeoutMs) || + (timeoutMs as number) <= 0 || + (timeoutMs as number) > MAX_NETWORK_DEADLINE_MS + ) { + throw new Error(); + } + const operation = record.operation; + if (typeof operation !== "function") throw new Error(); + + return Object.freeze({ + lane, + resourceLabel, + timeoutMs: timeoutMs as number, + operation: operation as (signal: AbortSignal) => unknown | PromiseLike, + }); + } catch { + throw new TypeError("Network task is invalid."); + } +} + +function createLaneState(lane: NetworkLane, overrides: NetworkLaneLimitOverrides): LaneState { + const defaults = DEFAULT_NETWORK_LANE_LIMITS[lane]; + const override = overrides[lane]; + const maxActive = + override && OWN.call(override, "maxActive") ? override.maxActive : defaults.maxActive; + const maxQueued = + override && OWN.call(override, "maxQueued") ? override.maxQueued : defaults.maxQueued; + + validatePositiveInteger(maxActive, `${lane}.maxActive`); + validateNonNegativeInteger(maxQueued, `${lane}.maxQueued`); + + return { + active: 0, + failed: false, + activeJobs: new Set(), + queue: [], + limit: Object.freeze({ maxActive, maxQueued }), + }; +} + +function snapshotSchedulerOptions(options: unknown): { + readonly clock?: NetworkClock; + readonly laneLimits: NetworkLaneLimitOverrides; +} { + try { + const record = strictPlainDataRecord(options, ["clock", "laneLimits"]); + const clock = OWN.call(record, "clock") ? record.clock : undefined; + const laneLimits = snapshotLaneLimitOverrides( + OWN.call(record, "laneLimits") ? record.laneLimits : undefined, + ); + return Object.freeze({ + ...(clock === undefined ? {} : { clock: clock as NetworkClock }), + laneLimits, + }); + } catch { + throw new TypeError("Network scheduler options are invalid."); + } +} + +function snapshotLaneLimitOverrides(value: unknown): NetworkLaneLimitOverrides { + if (value === undefined) return Object.freeze({}); + const record = strictPlainDataRecord(value, NETWORK_LANES); + const snapshot: Partial>> = {}; + for (const lane of NETWORK_LANES) { + if (!OWN.call(record, lane)) continue; + const limit = strictPlainDataRecord(record[lane], ["maxActive", "maxQueued"]); + const copied: { maxActive?: number; maxQueued?: number } = {}; + if (OWN.call(limit, "maxActive")) copied.maxActive = limit.maxActive as number; + if (OWN.call(limit, "maxQueued")) copied.maxQueued = limit.maxQueued as number; + snapshot[lane] = Object.freeze(copied); + } + return Object.freeze(snapshot); +} + +function strictPlainDataRecord( + value: unknown, + allowedKeys: readonly string[], +): Record { + if (typeof value !== "object" || value === null || Array.isArray(value)) throw new Error(); + const prototype = Object.getPrototypeOf(value); + if (prototype !== Object.prototype && prototype !== null) throw new Error(); + const record = Object.create(null) as Record; + for (const key of Reflect.ownKeys(value)) { + if (typeof key !== "string" || !allowedKeys.includes(key)) throw new Error(); + const descriptor = Object.getOwnPropertyDescriptor(value, key); + if (!descriptor?.enumerable || !("value" in descriptor)) throw new Error(); + record[key] = descriptor.value; + } + return record; +} + +function validateLane(value: unknown): NetworkLane { + if (typeof value !== "string" || !OWN.call(NETWORK_LANE_MEMBERSHIP, value)) { + throw new TypeError("Network lane must be metadata, media, or provider."); + } + return value as NetworkLane; +} + +function snapshotClock(clock: NetworkClock): NetworkClock { + if ( + !clock || + typeof clock.now !== "function" || + typeof clock.setTimeout !== "function" || + typeof clock.clearTimeout !== "function" + ) { + throw new TypeError("Network scheduler clock is invalid."); + } + + return Object.freeze({ + now: clock.now.bind(clock), + setTimeout: clock.setTimeout.bind(clock), + clearTimeout: clock.clearTimeout.bind(clock), + }); +} + +function readClock(clock: NetworkClock): number { + const now = clock.now(); + if (!Number.isFinite(now)) { + throw new TypeError("Network scheduler clock returned an invalid time."); + } + return now; +} + +function createDeferred(): Deferred { + let resolve!: (value: T) => void; + let reject!: (error: unknown) => void; + const promise = new Promise((resolvePromise, rejectPromise) => { + resolve = resolvePromise; + reject = rejectPromise; + }); + return { promise, resolve, reject }; +} + +function validatePositiveInteger(value: unknown, name: string): asserts value is number { + if (!Number.isSafeInteger(value) || (value as number) <= 0) { + throw new TypeError(`${name} must be a positive integer.`); + } +} + +function validateNonNegativeInteger(value: unknown, name: string): asserts value is number { + if (!Number.isSafeInteger(value) || (value as number) < 0) { + throw new TypeError(`${name} must be a non-negative integer.`); + } +} diff --git a/src/network/PinnedNetworkHop.ts b/src/network/PinnedNetworkHop.ts new file mode 100644 index 0000000..2119d93 --- /dev/null +++ b/src/network/PinnedNetworkHop.ts @@ -0,0 +1,251 @@ +import { + isResolvedAddressLiteral, + resolvedAddressesEqual, + type NetworkProtocol, + type TargetPolicyView, +} from "./LiteralTargetClassifier"; +import { + CAPABILITY_TRANSPORT_ERROR_CODES, + createCapabilityTransportError, + type NetworkResourceLabel, +} from "./NetworkErrors"; + +export const MAX_NETWORK_RESOLVED_ADDRESSES = 16; + +/** + * Trusted DNS primitive. The core strictly validates and chooses returned + * addresses. The resolver must observe abort, synchronously initiate + * cancellation, and settle promptly so cancelled work cannot retain a permit. + */ +export type NetworkNameResolver = ( + hostname: string, + signal: AbortSignal, +) => readonly string[] | PromiseLike; + +/** + * A pinned request that cannot trigger a second hostname lookup by itself. + * + * The adapter is part of the trusted network boundary. It must connect directly + * to `connectAddress`, use `serverName` only for TLS SNI, send `hostHeader` and + * `requestTarget` without ambient credentials, disable redirects, report the + * socket's actual remote address, and make `close` synchronously initiate + * shutdown, settle promptly, and promptly settle pending body iteration. HTTPS + * must use normal CA-chain validation and verify the certificate against + * `serverName`, or against the IP SAN for a literal target. Insecure TLS + * overrides are prohibited. Before returning a response, the adapter must + * observe abort, synchronously cancel its work, and settle promptly. Platforms + * that cannot satisfy this contract must fail closed. + */ +export interface PinnedNetworkHopRequest { + readonly protocol: NetworkProtocol; + readonly connectAddress: string; + readonly port: number; + readonly serverName?: string; + readonly hostHeader: string; + /** Exact path and query characters derived from the capability target. */ + readonly requestTarget: string; + readonly method: "GET"; + readonly credentials: "omit"; + readonly redirect: "manual"; + readonly signal: AbortSignal; +} + +export interface PinnedNetworkHopResponse { + readonly status: number; + readonly location?: string; + readonly body: AsyncIterable; + /** Actual remote address read from the connected socket. */ + readonly connectedAddress: string; + /** Must synchronously begin shutdown and promptly settle close plus body work. */ + readonly close: () => void | PromiseLike; +} + +export type PinnedNetworkHopAdapter = ( + request: PinnedNetworkHopRequest, +) => PinnedNetworkHopResponse | PromiseLike; + +export interface PinnedNetworkHop { + readonly status: number; + readonly location?: string; + readonly body: AsyncIterable; + readonly close: () => Promise; +} + +const missingProperty = Symbol("missing-property"); + +function isObject(value: unknown): value is object { + return (typeof value === "object" && value !== null) || typeof value === "function"; +} + +function getOwnDataValue(record: object, property: string): unknown | typeof missingProperty { + const descriptor = Object.getOwnPropertyDescriptor(record, property); + return descriptor && "value" in descriptor ? descriptor.value : missingProperty; +} + +/** @internal Pure validation helper for CapabilityScopedTransport. */ +export function snapshotResolvedAddresses( + value: unknown, + resourceLabel: NetworkResourceLabel, +): readonly string[] { + try { + if ( + !Array.isArray(value) || + value.length === 0 || + value.length > MAX_NETWORK_RESOLVED_ADDRESSES + ) { + throw new Error(); + } + const keys = Reflect.ownKeys(value); + if (keys.length !== value.length + 1 || !keys.includes("length")) throw new Error(); + + const addresses: string[] = []; + for (let index = 0; index < value.length; index += 1) { + const address = getOwnDataValue(value, String(index)); + if ( + address === missingProperty || + !isResolvedAddressLiteral(address) || + addresses.some((existing) => resolvedAddressesEqual(existing, address)) + ) { + throw new Error(); + } + addresses.push(address); + } + return Object.freeze(addresses); + } catch { + throw createCapabilityTransportError( + CAPABILITY_TRANSPORT_ERROR_CODES.addressRejected, + resourceLabel, + ); + } +} + +/** @internal Pure validation helper for CapabilityScopedTransport. */ +export function snapshotHopResponse( + value: unknown, + expectedAddress: string, + resourceLabel: NetworkResourceLabel, +): Omit & { readonly close: () => void | PromiseLike } { + try { + if (typeof value !== "object" || value === null || Array.isArray(value)) throw new Error(); + const keys = Reflect.ownKeys(value); + const required = ["status", "body", "connectedAddress", "close"]; + if ( + keys.some((key) => typeof key !== "string") || + keys.length < required.length || + keys.length > required.length + 1 || + required.some((key) => !keys.includes(key)) || + (keys.length === required.length + 1 && !keys.includes("location")) + ) { + throw new Error(); + } + + const status = getOwnDataValue(value, "status"); + const body = getOwnDataValue(value, "body"); + const connectedAddress = getOwnDataValue(value, "connectedAddress"); + const close = getOwnDataValue(value, "close"); + const location = keys.includes("location") ? getOwnDataValue(value, "location") : undefined; + if ( + !Number.isInteger(status) || + (status as number) < 100 || + (status as number) > 599 || + !isObject(body) || + typeof (body as AsyncIterable)[Symbol.asyncIterator] !== "function" || + !resolvedAddressesEqual(connectedAddress, expectedAddress) || + typeof close !== "function" || + (location !== undefined && typeof location !== "string") + ) { + throw new Error(); + } + + return Object.freeze({ + status: status as number, + body: body as AsyncIterable, + close: (): void | PromiseLike => + Reflect.apply( + close as (...arguments_: unknown[]) => unknown, + value, + [], + ) as void | PromiseLike, + ...(location === undefined ? {} : { location }), + }); + } catch { + throw createCapabilityTransportError( + CAPABILITY_TRANSPORT_ERROR_CODES.adapterResponseInvalid, + resourceLabel, + ); + } +} + +/** @internal Pure request derivation helper for CapabilityScopedTransport. */ +export function targetRequestParts( + target: string, + policy: TargetPolicyView, +): { readonly hostHeader: string; readonly requestTarget: string } { + const authorityStart = target.indexOf("//") + 2; + const remainder = target.slice(authorityStart); + const separatorIndex = remainder.search(/[/?]/); + const requestTarget = + separatorIndex === -1 + ? "/" + : remainder[separatorIndex] === "?" + ? `/${remainder.slice(separatorIndex)}` + : remainder.slice(separatorIndex); + const host = policy.normalizedHostname.includes(":") + ? `[${policy.normalizedHostname}]` + : policy.normalizedHostname; + const defaultPort = policy.protocol === "https:" ? 443 : 80; + return Object.freeze({ + hostHeader: `${host}${policy.port === defaultPort ? "" : `:${policy.port}`}`, + requestTarget, + }); +} + +/** @internal One-shot response lifecycle helper for CapabilityScopedTransport. */ +export function manageHopResponse( + response: ReturnType, + signal: AbortSignal, +): PinnedNetworkHop { + let closePromise: Promise | undefined; + const closeOnce = (): Promise => { + if (!closePromise) { + let resolveClose!: () => void; + let rejectClose!: (error: unknown) => void; + closePromise = new Promise((resolve, reject) => { + resolveClose = resolve; + rejectClose = reject; + }); + try { + Promise.resolve(response.close()).then(resolveClose, rejectClose); + } catch (error) { + rejectClose(error); + } + } + return closePromise; + }; + const onAbort = (): void => { + void closeOnce().catch(() => undefined); + }; + signal.addEventListener("abort", onAbort, { once: true }); + if (signal.aborted) onAbort(); + + return Object.freeze({ + status: response.status, + body: response.body, + ...(response.location === undefined ? {} : { location: response.location }), + close: async (): Promise => { + signal.removeEventListener("abort", onAbort); + await closeOnce(); + }, + }); +} + +/** @internal Best-effort cleanup helper for malformed adapter responses. */ +export async function closeInvalidHopResponse(value: unknown): Promise { + if (!isObject(value)) return; + try { + const close = getOwnDataValue(value, "close"); + if (typeof close === "function") await Reflect.apply(close, value, []); + } catch { + // The invalid adapter response remains the only caller-visible failure. + } +}