From b070e52d3e6b49c98117249e5b3f0d46edce173a Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Mon, 5 Oct 2026 15:14:51 -0700 Subject: [PATCH] feat(relay): forward webhook requests to the environment's tunnel (#15086) Co-authored-by: Claude Opus 5.5 (1M context) --- docs/internals/t3-connect.md | 6 + .../AgentActivityPublisher.test.ts | 1 + .../src/agentActivity/FcmDeliveries.test.ts | 1 + .../agentActivity/MobileRegistrations.test.ts | 1 + .../environments/EnvironmentConnector.test.ts | 1 + .../src/environments/EnvironmentConnector.ts | 181 ++++--- .../environments/EnvironmentLinker.test.ts | 1 + .../src/environments/EnvironmentLinks.ts | 72 +++ infra/relay/src/hooks/HookForwarder.test.ts | 462 ++++++++++++++++++ infra/relay/src/hooks/HookForwarder.ts | 367 ++++++++++++++ infra/relay/src/http/Api.test.ts | 1 + infra/relay/src/http/Api.ts | 56 ++- infra/relay/src/worker.ts | 31 +- packages/contracts/src/relay.ts | 23 + 14 files changed, 1118 insertions(+), 86 deletions(-) create mode 100644 infra/relay/src/hooks/HookForwarder.test.ts create mode 100644 infra/relay/src/hooks/HookForwarder.ts diff --git a/docs/internals/t3-connect.md b/docs/internals/t3-connect.md index 367d9116e92e42a0e53a7d19a185199a121e25cf..a857bd95819dcf92229bccd39fd8e0d401aae311 100644 --- a/docs/internals/t3-connect.md +++ b/docs/internals/t3-connect.md @@ -4,6 +4,12 @@ T3 Connect uses Clerk for cloud identity. The relay manages environment links, credentials for reaching environments, and managed tunnel allocations. After bootstrap, clients send application traffic through the environment's tunnel hostname; the relay Worker does not proxy their HTTP or WebSocket sessions. +The one exception is automation webhooks: the relay forwards +`/v1/hooks/:environmentId/:hookId/:token` statelessly to the environment's +tunnel so senders get a stable URL. It stores nothing, keeps bodies and tokens +out of its traces, and leaves token and signature verification to the +environment +([forwarder](../../infra/relay/src/hooks/HookForwarder.ts)). Clerk, deployment, and native authentication setup live in the [Connect setup runbook](../operations/connect-setup.md). diff --git a/infra/relay/src/agentActivity/AgentActivityPublisher.test.ts b/infra/relay/src/agentActivity/AgentActivityPublisher.test.ts index 7d76ec513a5a44fbaeeb435125a859c1740c5e2f..5f2c9463d4dac144cc49526092ddc09db2bd8f8b 100644 --- a/infra/relay/src/agentActivity/AgentActivityPublisher.test.ts +++ b/infra/relay/src/agentActivity/AgentActivityPublisher.test.ts @@ -94,6 +94,7 @@ function makeEnvironmentLinks( ]), listForUser: () => Effect.succeed([]), getForUser: () => Effect.succeed(null), + findActiveManagedForEnvironment: () => Effect.succeed([]), revokeForUser: () => Effect.succeed(false), ...overrides, }; diff --git a/infra/relay/src/agentActivity/FcmDeliveries.test.ts b/infra/relay/src/agentActivity/FcmDeliveries.test.ts index b42a4348a2c05d3505399cf9ee6f8b50eb547087..ee903db8f3825e9fd5e090d67c82f617294ff0ab 100644 --- a/infra/relay/src/agentActivity/FcmDeliveries.test.ts +++ b/infra/relay/src/agentActivity/FcmDeliveries.test.ts @@ -171,6 +171,7 @@ function harness() { : [], ), listForUser: () => Effect.succeed([]), + findActiveManagedForEnvironment: () => Effect.succeed([]), revokeForUser: () => Effect.succeed(false), getForUser: (input) => Effect.sync(() => diff --git a/infra/relay/src/agentActivity/MobileRegistrations.test.ts b/infra/relay/src/agentActivity/MobileRegistrations.test.ts index 41b32ef99db75d7a4f16bb541e1ef93dec6dd35d..57c5d6e55928b95119ec695fdbe6b7fb86abbaf1 100644 --- a/infra/relay/src/agentActivity/MobileRegistrations.test.ts +++ b/infra/relay/src/agentActivity/MobileRegistrations.test.ts @@ -116,6 +116,7 @@ function makeEnvironmentLinks( ]), listForUser: () => Effect.succeed([]), getForUser: () => Effect.succeed(null), + findActiveManagedForEnvironment: () => Effect.succeed([]), revokeForUser: () => Effect.succeed(false), ...overrides, }; diff --git a/infra/relay/src/environments/EnvironmentConnector.test.ts b/infra/relay/src/environments/EnvironmentConnector.test.ts index df1873529c2a38e6ff8389b05049da7ec0555f10..1a088b829d98bd6e75d04bcb70b7d570be062969 100644 --- a/infra/relay/src/environments/EnvironmentConnector.test.ts +++ b/infra/relay/src/environments/EnvironmentConnector.test.ts @@ -229,6 +229,7 @@ function makeLinks( environmentPublicKey: environmentKeyPair.publicKey, ...overrides, }), + findActiveManagedForEnvironment: () => Effect.succeed([]), revokeForUser: () => Effect.succeed(false), }; } diff --git a/infra/relay/src/environments/EnvironmentConnector.ts b/infra/relay/src/environments/EnvironmentConnector.ts index d4b975d5438ad152007107882bc0fe7186c277aa..f2edf1bc722786c0f6b5dd19089bd4e8b99d943d 100644 --- a/infra/relay/src/environments/EnvironmentConnector.ts +++ b/infra/relay/src/environments/EnvironmentConnector.ts @@ -16,6 +16,7 @@ import { RelayEnvironmentConnectNotAuthorizedReason, type RelayEnvironmentConnectResponse, type RelayEnvironmentStatusResponse, + type RelayManagedEndpoint, } from "@t3tools/contracts/relay"; import { normalizeRelayIssuer, @@ -35,6 +36,7 @@ import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; import * as Redacted from "effect/Redacted"; +import * as Result from "effect/Result"; import * as Schema from "effect/Schema"; import * as FetchHttpClient from "effect/http/FetchHttpClient"; import * as HttpClient from "effect/http/HttpClient"; @@ -185,7 +187,7 @@ const currentTraceId = Effect.currentSpan.pipe( Effect.orElseSucceed(() => "unavailable"), ); -const withoutRedirects = (effect: Effect.Effect) => +export const withoutRedirects = (effect: Effect.Effect) => effect.pipe(Effect.provideService(FetchHttpClient.RequestInit, { redirect: "manual" })); const verifyWithEnvironmentKeys = Effect.fnUntraced(function* (input: { @@ -288,6 +290,92 @@ function verifyEnvironmentHealthResponse(input: { ); } +export interface ManagedEndpointValidationFailure { + readonly reason: Exclude< + RelayEnvironmentConnectNotAuthorizedReason, + "client_proof_key_thumbprint_missing" | "environment_link_not_found" + >; + /** Diagnostic span attributes; never contains secrets. */ + readonly attributes?: Record; +} + +/** + * Checks that a link's relay-managed endpoint is backed by a ready allocation + * on the configured base domain and still matches what the link recorded. + */ +export function validateManagedEndpoint(input: { + readonly link: EnvironmentLinks.RelayLinkedEnvironmentRecord; + readonly allocation: ManagedEndpointAllocations.ManagedEndpointAllocation | null; + readonly baseDomain: string | undefined; +}): Result.Result { + const { link, allocation, baseDomain } = input; + if (link.endpoint.providerKind !== "cloudflare_tunnel") { + return Result.fail({ + reason: "endpoint_provider_not_managed", + attributes: { "relay.authorization.endpoint_provider_kind": link.endpoint.providerKind }, + }); + } + if (!allocation) { + return Result.fail({ reason: "managed_endpoint_allocation_not_found" }); + } + const allocationAttributes = { + "relay.authorization.allocation_hostname": allocation.hostname, + "relay.authorization.allocation_has_ready_at": allocation.readyAt !== null, + "relay.authorization.allocation_has_tunnel_id": allocation.tunnelId !== null, + "relay.authorization.allocation_has_dns_record_id": allocation.dnsRecordId !== null, + }; + if (!baseDomain) { + return Result.fail({ + reason: "managed_endpoint_base_domain_not_configured", + attributes: allocationAttributes, + }); + } + if ( + allocation.readyAt === null || + allocation.tunnelId === null || + allocation.dnsRecordId === null + ) { + return Result.fail({ + reason: "managed_endpoint_allocation_not_ready", + attributes: allocationAttributes, + }); + } + if (!isManagedEndpointHostname(allocation.hostname, baseDomain)) { + return Result.fail({ + reason: "managed_endpoint_hostname_invalid", + attributes: { + ...allocationAttributes, + "relay.authorization.managed_endpoint_base_domain": baseDomain, + }, + }); + } + const endpoint = ManagedEndpointAllocations.resolveReadyManagedEndpoint({ + allocation, + baseDomain, + }); + if ( + endpoint === null || + endpoint.httpBaseUrl !== link.endpoint.httpBaseUrl || + endpoint.wsBaseUrl !== link.endpoint.wsBaseUrl + ) { + return Result.fail({ + reason: "managed_endpoint_mismatch", + attributes: { + ...allocationAttributes, + "relay.authorization.linked_http_base_url": link.endpoint.httpBaseUrl, + "relay.authorization.linked_ws_base_url": link.endpoint.wsBaseUrl, + ...(endpoint + ? { + "relay.authorization.resolved_http_base_url": endpoint.httpBaseUrl, + "relay.authorization.resolved_ws_base_url": endpoint.wsBaseUrl, + } + : {}), + }, + }); + } + return Result.succeed(endpoint); +} + const make = Effect.gen(function* () { const links = yield* EnvironmentLinks.EnvironmentLinks; const allocations = yield* ManagedEndpointAllocations.ManagedEndpointAllocations; @@ -305,89 +393,22 @@ const make = Effect.gen(function* () { readonly link: EnvironmentLinks.RelayLinkedEnvironmentRecord; readonly allocation: ManagedEndpointAllocations.ManagedEndpointAllocation | null; }) { - if (input.link.endpoint.providerKind !== "cloudflare_tunnel") { - yield* Effect.annotateCurrentSpan({ - "relay.authorization.endpoint_provider_kind": input.link.endpoint.providerKind, - }); - return yield* new EnvironmentConnectNotAuthorized({ - environmentId: input.link.environmentId, - operation: input.operation, - reason: "endpoint_provider_not_managed", - }); - } - if (!input.allocation) { - return yield* new EnvironmentConnectNotAuthorized({ - environmentId: input.link.environmentId, - operation: input.operation, - reason: "managed_endpoint_allocation_not_found", - }); - } - const allocationAttributes = { - "relay.authorization.allocation_hostname": input.allocation.hostname, - "relay.authorization.allocation_has_ready_at": input.allocation.readyAt !== null, - "relay.authorization.allocation_has_tunnel_id": input.allocation.tunnelId !== null, - "relay.authorization.allocation_has_dns_record_id": input.allocation.dnsRecordId !== null, - } as const; - if (!settings.managedEndpointBaseDomain) { - yield* Effect.annotateCurrentSpan(allocationAttributes); - return yield* new EnvironmentConnectNotAuthorized({ - environmentId: input.link.environmentId, - operation: input.operation, - reason: "managed_endpoint_base_domain_not_configured", - }); - } - if ( - input.allocation.readyAt === null || - input.allocation.tunnelId === null || - input.allocation.dnsRecordId === null - ) { - yield* Effect.annotateCurrentSpan(allocationAttributes); - return yield* new EnvironmentConnectNotAuthorized({ - environmentId: input.link.environmentId, - operation: input.operation, - reason: "managed_endpoint_allocation_not_ready", - }); - } - if ( - !isManagedEndpointHostname(input.allocation.hostname, settings.managedEndpointBaseDomain) - ) { - yield* Effect.annotateCurrentSpan({ - ...allocationAttributes, - "relay.authorization.managed_endpoint_base_domain": settings.managedEndpointBaseDomain, - }); - return yield* new EnvironmentConnectNotAuthorized({ - environmentId: input.link.environmentId, - operation: input.operation, - reason: "managed_endpoint_hostname_invalid", - }); - } - const endpoint = ManagedEndpointAllocations.resolveReadyManagedEndpoint({ + const result = validateManagedEndpoint({ + link: input.link, allocation: input.allocation, baseDomain: settings.managedEndpointBaseDomain, }); - if ( - endpoint === null || - endpoint.httpBaseUrl !== input.link.endpoint.httpBaseUrl || - endpoint.wsBaseUrl !== input.link.endpoint.wsBaseUrl - ) { - yield* Effect.annotateCurrentSpan({ - ...allocationAttributes, - "relay.authorization.linked_http_base_url": input.link.endpoint.httpBaseUrl, - "relay.authorization.linked_ws_base_url": input.link.endpoint.wsBaseUrl, - ...(endpoint - ? { - "relay.authorization.resolved_http_base_url": endpoint.httpBaseUrl, - "relay.authorization.resolved_ws_base_url": endpoint.wsBaseUrl, - } - : {}), - }); - return yield* new EnvironmentConnectNotAuthorized({ - environmentId: input.link.environmentId, - operation: input.operation, - reason: "managed_endpoint_mismatch", - }); + if (Result.isSuccess(result)) { + return result.success; } - return endpoint; + if (result.failure.attributes) { + yield* Effect.annotateCurrentSpan(result.failure.attributes); + } + return yield* new EnvironmentConnectNotAuthorized({ + environmentId: input.link.environmentId, + operation: input.operation, + reason: result.failure.reason, + }); }, ); diff --git a/infra/relay/src/environments/EnvironmentLinker.test.ts b/infra/relay/src/environments/EnvironmentLinker.test.ts index 67928c25f09c3c7eae941805dbfd2867a79e3d60..912ffc30f70b9d3a3dc40607d8ce57635acac971 100644 --- a/infra/relay/src/environments/EnvironmentLinker.test.ts +++ b/infra/relay/src/environments/EnvironmentLinker.test.ts @@ -126,6 +126,7 @@ function testLayer(input?: { listDeliveryUsersForEnvironment: () => Effect.succeed([]), listForUser: () => Effect.succeed([]), getForUser: () => Effect.succeed(null), + findActiveManagedForEnvironment: () => Effect.succeed([]), revokeForUser: () => Effect.succeed(false), }), Layer.succeed(EnvironmentCredentials.EnvironmentCredentials, { diff --git a/infra/relay/src/environments/EnvironmentLinks.ts b/infra/relay/src/environments/EnvironmentLinks.ts index f41794d60538551d889d0112d3b36e7c9e18b070..c6832a8025f304b145c77f03340ca0a095f22e7e 100644 --- a/infra/relay/src/environments/EnvironmentLinks.ts +++ b/infra/relay/src/environments/EnvironmentLinks.ts @@ -75,6 +75,18 @@ export class EnvironmentLinkLookupPersistenceError extends Schema.TaggedError()( + "EnvironmentLinkEnvironmentLookupPersistenceError", + { + environmentId: Schema.String, + cause: Schema.Defect(), + }, +) { + override get message(): string { + return `Failed to look up active managed links for environment '${this.environmentId}'`; + } +} + export class EnvironmentLinkRevokePersistenceError extends Schema.TaggedError()( "EnvironmentLinkRevokePersistenceError", { @@ -114,6 +126,13 @@ export class EnvironmentLinks extends Context.Service< readonly userId: string; readonly environmentId: string; }) => Effect.Effect; + /** Active relay-managed links for an environment, across all users (webhook forwarding). */ + readonly findActiveManagedForEnvironment: (input: { + readonly environmentId: string; + }) => Effect.Effect< + ReadonlyArray, + EnvironmentLinkEnvironmentLookupPersistenceError + >; readonly revokeForUser: (input: { readonly userId: string; readonly environmentId: string; @@ -322,6 +341,59 @@ const make = Effect.gen(function* () { ); }), + findActiveManagedForEnvironment: Effect.fn( + "relay.environment_links.find_active_managed_for_environment", + )(function* (input) { + yield* Effect.annotateCurrentSpan({ "relay.environment_id": input.environmentId }); + return yield* db + .select({ + userId: relayEnvironmentLinks.userId, + environmentId: relayEnvironmentLinks.environmentId, + environmentLabel: relayEnvironmentLinks.environmentLabel, + environmentPublicKey: relayEnvironmentLinks.environmentPublicKey, + endpointHttpBaseUrl: relayEnvironmentLinks.endpointHttpBaseUrl, + endpointWsBaseUrl: relayEnvironmentLinks.endpointWsBaseUrl, + endpointProviderKind: relayEnvironmentLinks.endpointProviderKind, + createdAt: relayEnvironmentLinks.createdAt, + }) + .from(relayEnvironmentLinks) + .where( + and( + eq(relayEnvironmentLinks.environmentId, input.environmentId), + isNull(relayEnvironmentLinks.revokedAt), + eq(relayEnvironmentLinks.endpointProviderKind, "cloudflare_tunnel"), + eq(relayEnvironmentLinks.managedTunnelsEnabled, true), + ), + ) + // One row per user who linked this environment; every row is checked + // until one has a ready endpoint, so none may be cut off. + .pipe( + Effect.map((rows) => + rows.map((row) => ({ + userId: row.userId, + environmentId: row.environmentId as RelayClientEnvironmentRecord["environmentId"], + label: + row.environmentLabel.trim().length > 0 ? row.environmentLabel : row.environmentId, + endpoint: { + httpBaseUrl: row.endpointHttpBaseUrl, + wsBaseUrl: row.endpointWsBaseUrl, + providerKind: + row.endpointProviderKind as RelayClientEnvironmentRecord["endpoint"]["providerKind"], + }, + environmentPublicKey: row.environmentPublicKey, + linkedAt: row.createdAt, + })), + ), + Effect.mapError( + (cause) => + new EnvironmentLinkEnvironmentLookupPersistenceError({ + environmentId: input.environmentId, + cause, + }), + ), + ); + }), + revokeForUser: Effect.fn("relay.environment_links.revoke_for_user")(function* (input) { yield* Effect.annotateCurrentSpan({ "relay.environment_id": input.environmentId, diff --git a/infra/relay/src/hooks/HookForwarder.test.ts b/infra/relay/src/hooks/HookForwarder.test.ts new file mode 100644 index 0000000000000000000000000000000000000000..6c4383d10df3c5da65a92c793a98c5ae2819f1ff --- /dev/null +++ b/infra/relay/src/hooks/HookForwarder.test.ts @@ -0,0 +1,462 @@ +import * as NodeHttpPlatform from "@effect/platform-node/NodeHttpPlatform"; +import * as NodeServices from "@effect/platform-node/NodeServices"; +import { describe, expect, it } from "@effect/vitest"; +import { RelayApi } from "@t3tools/contracts/relay"; +import * as Deferred from "effect/Deferred"; +import * as Duration from "effect/Duration"; +import * as Effect from "effect/Effect"; +import * as Fiber from "effect/Fiber"; +import * as Layer from "effect/Layer"; +import * as Redacted from "effect/Redacted"; +import * as TestClock from "effect/testing/TestClock"; +import * as Tracer from "effect/Tracer"; +import * as HttpClient from "effect/http/HttpClient"; +import * as HttpClientError from "effect/http/HttpClientError"; +import type * as HttpClientRequest from "effect/http/HttpClientRequest"; +import * as HttpClientResponse from "effect/http/HttpClientResponse"; +import * as Etag from "effect/http/Etag"; +import * as HttpRouter from "effect/http/HttpRouter"; +import * as HttpApi from "effect/http-api/HttpApi"; +import * as HttpApiBuilder from "effect/http-api/HttpApiBuilder"; +import * as HttpServerRequest from "effect/http/HttpServerRequest"; +import * as HttpServerResponse from "effect/http/HttpServerResponse"; + +import * as RelayConfiguration from "../Config.ts"; +import * as EnvironmentLinks from "../environments/EnvironmentLinks.ts"; +import * as ManagedEndpointAllocations from "../environments/ManagedEndpointAllocations.ts"; +import { + RELAY_HTTP_ROUTER_CONFIG, + relayCors, + relayNotFoundRoute, + traceRelayHttpRequestWith, +} from "../http/Api.ts"; +import * as HookForwarder from "./HookForwarder.ts"; + +const settings: RelayConfiguration.RelayConfiguration["Service"] = { + relayIssuer: "https://relay.example.test", + apns: null, + clerkSecretKey: Redacted.make("clerk-secret-key"), + clerkPublishableKey: "pk_test_test", + clerkJwtAudience: "t3-code-relay", + apnsDeliveryJobSigningSecret: Redacted.make("apns-delivery-secret"), + cloudMintPrivateKey: Redacted.make("cloud-mint-private-key"), + cloudMintPublicKey: "cloud-mint-public-key", + managedEndpointBaseDomain: "example.test", + managedEndpointNamespace: undefined, +}; + +const environmentId = "env-hook"; +const readyAllocation: ManagedEndpointAllocations.ManagedEndpointAllocation = { + userId: "user_1", + environmentId, + hostname: "env.example.test", + tunnelId: "tunnel-id", + tunnelName: "tunnel-name", + dnsRecordId: "dns-record-id", + readyAt: "2026-05-25T00:00:00.000Z", + origin: { localHttpHost: "127.0.0.1", localHttpPort: 3773 }, + updatedAt: "2026-05-25T00:00:00.000Z", + generation: 1, +}; + +const managedLink = { + userId: "user_1", + environmentId: environmentId as never, + label: "Hook env", + endpoint: { + httpBaseUrl: "https://env.example.test/", + wsBaseUrl: "wss://env.example.test/ws", + providerKind: "cloudflare_tunnel" as const, + }, + environmentPublicKey: "public-key", + linkedAt: "2026-05-25T00:00:00.000Z", +}; + +interface Harness { + readonly execute?: ( + request: HttpClientRequest.HttpClientRequest, + ) => Effect.Effect; + readonly links?: ReadonlyArray; + readonly allocation?: ManagedEndpointAllocations.ManagedEndpointAllocation | null; + readonly allow?: (key: string) => boolean; +} + +function makeHarness(options: Harness = {}) { + const sent: Array = []; + const rateLimitKeys: Array = []; + const execute = + options.execute ?? + ((request: HttpClientRequest.HttpClientRequest) => + Effect.succeed( + HttpClientResponse.fromWeb( + request, + new Response("ok", { status: 200, headers: { "content-type": "text/plain" } }), + ), + )); + const forwarderLayer = HookForwarder.layer.pipe( + Layer.provide( + Layer.mergeAll( + Layer.succeed(RelayConfiguration.RelayConfiguration, settings), + Layer.mock(EnvironmentLinks.EnvironmentLinks, { + findActiveManagedForEnvironment: (input) => + Effect.succeed( + input.environmentId === environmentId ? (options.links ?? [managedLink]) : [], + ), + }), + Layer.mock(ManagedEndpointAllocations.ManagedEndpointAllocations, { + get: () => + Effect.succeed(options.allocation === undefined ? readyAllocation : options.allocation), + }), + Layer.succeed( + HttpClient.HttpClient, + HttpClient.make((request) => { + sent.push(request); + return execute(request); + }), + ), + Layer.succeed(HookForwarder.HookRateLimiter, { + allow: (key) => + Effect.sync(() => { + rateLimitKeys.push(key); + return options.allow ? options.allow(key) : true; + }), + }), + ), + ), + ); + const httpEffect = HttpRouter.toHttpEffect( + Layer.mergeAll( + HttpApiBuilder.layer(HttpApi.make("RelayApi").add(RelayApi.groups.hooks)).pipe( + Layer.provide(HookForwarder.hooksApi.pipe(Layer.provide(forwarderLayer))), + Layer.provide([NodeServices.layer, NodeHttpPlatform.layer, Etag.layerWeak]), + ), + relayNotFoundRoute, + relayCors, + ), + ).pipe(Effect.provideService(HttpRouter.RouterConfig, RELAY_HTTP_ROUTER_CONFIG)); + const send = (request: Request) => + Effect.gen(function* () { + const handler = yield* httpEffect; + return yield* handler.pipe( + Effect.provideService( + HttpServerRequest.HttpServerRequest, + HttpServerRequest.fromWeb(request), + ), + ); + }); + return { sent, rateLimitKeys, send, httpEffect }; +} + +const hookUrl = (path = "hook-1/secret-token", query = "") => + `https://relay.test/v1/hooks/${environmentId}/${path}${query}`; + +const readBody = (response: HttpServerResponse.HttpServerResponse) => + Effect.promise(() => HttpServerResponse.toWeb(response).arrayBuffer()).pipe( + Effect.map((buffer) => new Uint8Array(buffer)), + ); +const readJson = (response: HttpServerResponse.HttpServerResponse) => + Effect.promise(() => HttpServerResponse.toWeb(response).json()); + +const requestBytes = (request: HttpClientRequest.HttpClientRequest) => + request.body._tag === "Uint8Array" ? request.body.body : new Uint8Array(0); + +describe("HookForwarder", () => { + it.effect("forwards the exact body bytes, query and filtered headers", () => + Effect.gen(function* () { + const harness = makeHarness(); + const body = new Uint8Array([0, 255, 10, 13, 0x7b, 0x22, 0xc3, 0x28]); + const response = yield* harness.send( + new Request(hookUrl("hook-1/tok%2Fen", "?a=1&b=two%20words"), { + method: "POST", + headers: { + "content-type": "application/octet-stream", + authorization: "Bearer sender-secret", + "x-hub-signature-256": "sha256=abc", + cookie: "session=1", + "cf-connecting-ip": "1.2.3.4", + "x-forwarded-for": "1.2.3.4", + "x-real-ip": "1.2.3.4", + "proxy-authorization": "Basic x", + connection: "keep-alive", + }, + body, + }), + ); + expect(response.status).toBe(200); + expect(harness.sent).toHaveLength(1); + const sent = harness.sent[0]!; + expect(sent.method).toBe("POST"); + expect(sent.url).toBe("https://env.example.test/api/hooks/hook-1/tok%2Fen?a=1&b=two%20words"); + expect(Array.from(requestBytes(sent))).toEqual(Array.from(body)); + expect(sent.headers.authorization).toBe("Bearer sender-secret"); + expect(sent.headers["x-hub-signature-256"]).toBe("sha256=abc"); + expect(sent.headers["content-type"]).toBe("application/octet-stream"); + for (const dropped of [ + "cookie", + "cf-connecting-ip", + "x-forwarded-for", + "x-real-ip", + "proxy-authorization", + "connection", + "host", + ]) { + expect(sent.headers[dropped]).toBeUndefined(); + } + // Recomputed from the forwarded bytes, not copied from the sender. + expect(sent.headers["content-length"]).toBe(String(body.length)); + // The budget key is a hash, so the token never reaches the limiter. + expect(harness.rateLimitKeys).toHaveLength(1); + expect(harness.rateLimitKeys[0]).toMatch(/^[0-9a-f]{64}$/); + }), + ); + + it.effect("passes upstream status, body and content-type through", () => + Effect.gen(function* () { + const harness = makeHarness({ + execute: (request) => + Effect.succeed( + HttpClientResponse.fromWeb( + request, + new Response('{"error":"bad_signature"}', { + status: 401, + headers: { "content-type": "application/json", "set-cookie": "x=1" }, + }), + ), + ), + }); + const response = yield* harness.send(new Request(hookUrl(), { method: "GET" })); + expect(response.status).toBe(401); + expect(response.headers["content-type"]).toBe("application/json"); + expect(response.headers["set-cookie"]).toBeUndefined(); + expect(response.headers["access-control-allow-origin"]).toBeUndefined(); + expect(new TextDecoder().decode(yield* readBody(response))).toBe('{"error":"bad_signature"}'); + expect(harness.sent[0]?.method).toBe("GET"); + }), + ); + + it.effect("does not follow or relay upstream redirects", () => + Effect.gen(function* () { + const harness = makeHarness({ + execute: (request) => + Effect.succeed( + HttpClientResponse.fromWeb( + request, + new Response(null, { + status: 302, + headers: { location: "https://internal.example/" }, + }), + ), + ), + }); + const response = yield* harness.send(new Request(hookUrl(), { method: "POST", body: "{}" })); + expect(response.status).toBe(302); + expect(response.headers.location).toBeUndefined(); + expect(harness.sent).toHaveLength(1); + }), + ); + + it.effect("rejects bodies over 1 MiB by content-length and while reading", () => + Effect.gen(function* () { + const harness = makeHarness(); + const declared = yield* harness.send( + new Request(hookUrl(), { + method: "POST", + headers: { "content-length": String(HookForwarder.RELAY_HOOK_MAX_BODY_BYTES + 1) }, + body: "x", + }), + ); + expect(declared.status).toBe(413); + + const oversized = new Uint8Array(HookForwarder.RELAY_HOOK_MAX_BODY_BYTES + 1); + const stream = new ReadableStream({ + start(controller) { + controller.enqueue(oversized.subarray(0, 600_000)); + controller.enqueue(oversized.subarray(600_000)); + controller.close(); + }, + }); + const streamed = yield* harness.send( + new Request(hookUrl(), { method: "POST", body: stream, duplex: "half" } as RequestInit), + ); + expect(streamed.status).toBe(413); + expect(yield* readJson(streamed)).toEqual({ error: "payload_too_large" }); + expect(harness.sent).toHaveLength(0); + }), + ); + + it.effect("returns 404 for unknown environments and unready endpoints", () => + Effect.gen(function* () { + const unknown = makeHarness(); + const response = yield* unknown.send( + new Request("https://relay.test/v1/hooks/other-env/hook-1/token", { method: "POST" }), + ); + expect(response.status).toBe(404); + expect(yield* readJson(response)).toEqual({ error: "hook_not_found" }); + + const unready = makeHarness({ allocation: { ...readyAllocation, readyAt: null } }); + const unreadyResponse = yield* unready.send(new Request(hookUrl(), { method: "POST" })); + expect(unreadyResponse.status).toBe(404); + expect(unready.sent).toHaveLength(0); + }), + ); + + it.effect("maps tunnel-offline and network failures to 503", () => + Effect.gen(function* () { + const offline = makeHarness({ + execute: (request) => + Effect.succeed(HttpClientResponse.fromWeb(request, new Response("", { status: 530 }))), + }); + const offlineResponse = yield* offline.send(new Request(hookUrl(), { method: "POST" })); + expect(offlineResponse.status).toBe(503); + expect(yield* readJson(offlineResponse)).toEqual({ error: "environment_unavailable" }); + + const network = makeHarness({ + execute: (request) => + Effect.fail( + new HttpClientError.HttpClientError({ + reason: new HttpClientError.TransportError({ request, cause: new Error("reset") }), + }), + ), + }); + const networkResponse = yield* network.send(new Request(hookUrl(), { method: "POST" })); + expect(networkResponse.status).toBe(503); + }), + ); + + it.effect("does not relay an upstream response larger than the cap", () => + Effect.gen(function* () { + const harness = makeHarness({ + execute: (request) => + Effect.succeed( + HttpClientResponse.fromWeb( + request, + new Response(new Uint8Array(64 * 1024 + 1), { status: 200 }), + ), + ), + }); + const response = yield* harness.send(new Request(hookUrl(), { method: "POST" })); + expect(response.status).toBe(503); + expect(yield* readJson(response)).toEqual({ error: "environment_unavailable" }); + }), + ); + + it.effect("maps an upstream timeout to 504", () => + Effect.gen(function* () { + // Advance the clock only once the request is waiting on the environment; + // hashing the budget key before that step is asynchronous. + const reachedUpstream = yield* Deferred.make(); + const harness = makeHarness({ + execute: () => + Deferred.succeed(reachedUpstream, undefined).pipe(Effect.andThen(Effect.never)), + }); + const fiber = yield* harness + .send(new Request(hookUrl(), { method: "POST", body: "{}" })) + .pipe(Effect.forkChild); + yield* Deferred.await(reachedUpstream); + yield* TestClock.adjust(Duration.millis(HookForwarder.RELAY_HOOK_UPSTREAM_TIMEOUT_MS)); + const response = yield* Fiber.join(fiber); + expect(response.status).toBe(504); + expect(yield* readJson(response)).toEqual({ error: "environment_timeout" }); + }), + ); + + it.effect("answers OPTIONS without a CORS preflight or forwarding", () => + Effect.gen(function* () { + const harness = makeHarness(); + const response = yield* harness.send(new Request(hookUrl(), { method: "OPTIONS" })); + // No hook endpoint accepts OPTIONS, so it falls through to the 404 route. + expect(response.status).toBe(404); + expect(response.headers["access-control-allow-origin"]).toBeUndefined(); + expect(response.headers["access-control-allow-methods"]).toBeUndefined(); + expect(harness.sent).toHaveLength(0); + }), + ); + + it.effect("drops headers the sender names in Connection", () => + Effect.gen(function* () { + const harness = makeHarness(); + yield* harness.send( + new Request(hookUrl(), { + method: "POST", + body: "{}", + headers: { connection: "x-hop, keep-alive", "x-hop": "1", "x-keep": "2" }, + }), + ); + const sent: Readonly> = harness.sent[0]?.headers ?? {}; + expect(sent["x-hop"]).toBeUndefined(); + expect(sent["x-keep"]).toBe("2"); + }), + ); + + it.effect("gives requests with a wrong token their own budget", () => + Effect.gen(function* () { + const harness = makeHarness(); + yield* harness.send( + new Request(hookUrl("hook-1/real-token"), { method: "POST", body: "{}" }), + ); + yield* harness.send(new Request(hookUrl("hook-1/guessed"), { method: "POST", body: "{}" })); + expect(new Set(harness.rateLimitKeys).size).toBe(2); + }), + ); + + it.effect("gives two spellings of one token a single budget", () => + Effect.gen(function* () { + const harness = makeHarness(); + yield* harness.send(new Request(hookUrl("hook-1/token"), { method: "POST", body: "{}" })); + yield* harness.send(new Request(hookUrl("hook-1/%74oken"), { method: "POST", body: "{}" })); + expect(new Set(harness.rateLimitKeys).size).toBe(1); + }), + ); + + it.effect("returns 429 when the hook's budget is spent", () => + Effect.gen(function* () { + const harness = makeHarness({ allow: () => false }); + const response = yield* harness.send(new Request(hookUrl(), { method: "POST" })); + expect(response.status).toBe(429); + expect(harness.sent).toHaveLength(0); + }), + ); + + it.effect("never records the token in the server span", () => + Effect.gen(function* () { + const spans: Array = []; + const tracer = Tracer.make({ + span: (options) => { + const span = new Tracer.NativeSpan(options); + spans.push(span); + return span; + }, + }); + const harness = makeHarness(); + const handler = yield* harness.httpEffect; + const response = yield* traceRelayHttpRequestWith( + handler, + Layer.succeed(Tracer.Tracer, tracer), + ).pipe( + Effect.provideService( + HttpServerRequest.HttpServerRequest, + HttpServerRequest.fromWeb( + new Request(hookUrl("hook-1/super-secret-token", "?sig=also-secret"), { + method: "POST", + headers: { "x-gitlab-token": "header-secret" }, + body: "{}", + }), + ), + ), + ); + expect(response.status).toBe(200); + expect(harness.sent[0]?.url).toContain("super-secret-token"); + yield* Effect.yieldNow; + const serialized = spans + .flatMap((span) => [span.name, ...Array.from(span.attributes.values(), String)]) + .join("\n"); + expect(serialized).not.toContain("super-secret-token"); + expect(serialized).not.toContain("also-secret"); + expect(serialized).not.toContain("header-secret"); + const server = spans.find((span) => span.kind === "server"); + expect(server?.attributes.get("url.path")).toBe( + `/v1/hooks/${environmentId}/hook-1/`, + ); + }), + ); +}); diff --git a/infra/relay/src/hooks/HookForwarder.ts b/infra/relay/src/hooks/HookForwarder.ts new file mode 100644 index 0000000000000000000000000000000000000000..7f33a7d2513f195d56e9cdbff75be14cb6e52a35 --- /dev/null +++ b/infra/relay/src/hooks/HookForwarder.ts @@ -0,0 +1,367 @@ +import * as Context from "effect/Context"; +import * as Duration from "effect/Duration"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; +import * as Result from "effect/Result"; +import * as Schema from "effect/Schema"; +import * as Stream from "effect/Stream"; +import * as HttpClient from "effect/http/HttpClient"; +import * as HttpClientRequest from "effect/http/HttpClientRequest"; +import type * as HttpClientResponse from "effect/http/HttpClientResponse"; +import type * as HttpServerRequest from "effect/http/HttpServerRequest"; +import * as HttpServerResponse from "effect/http/HttpServerResponse"; +import * as HttpApiBuilder from "effect/http-api/HttpApiBuilder"; +import { RelayApi } from "@t3tools/contracts/relay"; + +import * as RelayConfiguration from "../Config.ts"; +import { validateManagedEndpoint, withoutRedirects } from "../environments/EnvironmentConnector.ts"; +import * as EnvironmentLinks from "../environments/EnvironmentLinks.ts"; +import * as ManagedEndpointAllocations from "../environments/ManagedEndpointAllocations.ts"; + +export const RELAY_HOOK_PATH_PREFIX = "/v1/hooks/"; +export const RELAY_HOOK_MAX_BODY_BYTES = 1_048_576; +export const RELAY_HOOK_UPSTREAM_TIMEOUT_MS = 8_000; +export const RELAY_HOOK_RATE_LIMIT = { limit: 60, periodSeconds: 60 } as const; + +const DROPPED_REQUEST_HEADERS = new Set([ + "host", + "connection", + "keep-alive", + "transfer-encoding", + "te", + "upgrade", + "content-length", + "cookie", + "x-real-ip", +]); +const DROPPED_REQUEST_HEADER_PREFIXES = ["proxy-", "cf-", "x-forwarded-"]; +// Cloudflare answers 530 when the tunnel for a hostname has no connected origin. +const TUNNEL_OFFLINE_STATUS = 530; +/** The environment answers with a small JSON status; anything past this is cut off. */ +const MAX_RESPONSE_BYTES = 64 * 1024; + +export const isRelayHookPath = (url: string): boolean => url.startsWith(RELAY_HOOK_PATH_PREFIX); + +/** Replaces the hook token (and any query) so traces and logs never record the secret. */ +export const redactRelayHookUrl = (url: string): string => { + const path = url.split("?", 1)[0] ?? url; + const segments = path.split("/"); + if (segments.length >= 6) { + segments[5] = ""; + } + return segments.join("/"); +}; + +/** + * Request budget for public hook forwarding, keyed by a hash of the hook URL + * (environment, hook and token). Requests with a wrong token get their own + * budget, so they cannot use up a real sender's; the environment rejects them. + * Built from decoded segments, because the environment decodes them too: two + * spellings of one token (`token`, `%74oken`) must share one budget. + */ +const hookBudgetKey = (hook: { + readonly environmentId: string; + readonly hookId: string; + readonly token: string; +}) => + Effect.promise(() => + crypto.subtle.digest( + "SHA-256", + // Length-prefixed, so no segment contents can make two keys collide. + new TextEncoder().encode( + [hook.environmentId, hook.hookId, hook.token] + .map((part) => `${part.length}:${part}`) + .join(""), + ), + ), + ).pipe( + Effect.map((digest) => + Array.from(new Uint8Array(digest), (byte) => byte.toString(16).padStart(2, "0")).join(""), + ), + ); + +export class HookRateLimiter extends Context.Service< + HookRateLimiter, + { readonly allow: (key: string) => Effect.Effect } +>()("t3code-relay/hooks/HookForwarder/HookRateLimiter") {} + +export class HookForwarder extends Context.Service< + HookForwarder, + { + readonly handle: ( + request: HttpServerRequest.HttpServerRequest, + ) => Effect.Effect; + } +>()("t3code-relay/hooks/HookForwarder") {} + +class HookBodyTooLarge extends Schema.TaggedError()("HookBodyTooLarge", {}) {} +class ResponseTooLarge extends Schema.TaggedError()("ResponseTooLarge", {}) {} + +const errorResponse = (status: number, error: string, headers?: Record) => + HttpServerResponse.jsonUnsafe({ error }, { status, ...(headers ? { headers } : {}) }); + +const hookNotFound = () => errorResponse(404, "hook_not_found"); + +function parseHookPath(url: string) { + const queryIndex = url.indexOf("?"); + const path = queryIndex === -1 ? url : url.slice(0, queryIndex); + const search = queryIndex === -1 ? "" : url.slice(queryIndex); + const segments = path.split("/"); + // ["", "v1", "hooks", environmentId, hookId, token] + if (segments.length !== 6) return null; + const [, , , rawEnvironmentId, rawHookId, rawToken] = segments; + if (!rawEnvironmentId || !rawHookId || !rawToken) return null; + try { + return { + environmentId: decodeURIComponent(rawEnvironmentId), + hookId: decodeURIComponent(rawHookId), + token: decodeURIComponent(rawToken), + // Forward the encoded segments byte-for-byte; the environment decodes them. + rawHookId, + rawToken, + search, + }; + } catch { + return null; + } +} + +function forwardedHeaders(headers: Readonly>): Record { + const result: Record = {}; + // Headers the sender names in Connection are hop-by-hop too (RFC 9110 7.6.1). + const connectionValue = + Object.entries(headers).find(([name]) => name.toLowerCase() === "connection")?.[1] ?? ""; + const nominated = new Set( + connectionValue + .split(",") + .map((name) => name.trim().toLowerCase()) + .filter(Boolean), + ); + for (const name in headers) { + const lower = name.toLowerCase(); + if ( + DROPPED_REQUEST_HEADERS.has(lower) || + nominated.has(lower) || + DROPPED_REQUEST_HEADER_PREFIXES.some((prefix) => lower.startsWith(prefix)) + ) { + continue; + } + const value = headers[name]; + if (value !== undefined) result[lower] = value; + } + return result; +} + +const hasNoBody = (request: HttpServerRequest.HttpServerRequest) => + request.source instanceof Request && request.source.body === null; + +/** Collects a byte stream, failing with `tooLarge` once it passes `maxBytes` rather than buffering it all. */ +const collectCapped = ( + stream: Stream.Stream, + maxBytes: number, + tooLarge: () => E2, +) => + Effect.suspend(() => { + const chunks: Array = []; + let total = 0; + return stream.pipe( + Stream.runForEach((chunk) => { + total += chunk.length; + if (total > maxBytes) { + return Effect.fail(tooLarge()); + } + chunks.push(chunk); + return Effect.void; + }), + Effect.map(() => { + const body = new Uint8Array(total); + let offset = 0; + for (const chunk of chunks) { + body.set(chunk, offset); + offset += chunk.length; + } + return body; + }), + ); + }); + +const readCappedBody = (request: HttpServerRequest.HttpServerRequest) => + hasNoBody(request) + ? Effect.succeed(new Uint8Array(0)) + : collectCapped(request.stream, RELAY_HOOK_MAX_BODY_BYTES, () => new HookBodyTooLarge()); + +const readCappedResponse = (response: HttpClientResponse.HttpClientResponse) => + collectCapped(response.stream, MAX_RESPONSE_BYTES, () => new ResponseTooLarge()).pipe( + // A response without a body, such as a redirect, has no stream at all. + Effect.catchIf( + (error) => error._tag === "HttpClientError" && error.reason._tag === "EmptyBodyError", + () => Effect.succeed(new Uint8Array(0)), + ), + ); + +const make = Effect.gen(function* () { + const links = yield* EnvironmentLinks.EnvironmentLinks; + const allocations = yield* ManagedEndpointAllocations.ManagedEndpointAllocations; + const settings = yield* RelayConfiguration.RelayConfiguration; + const httpClient = yield* HttpClient.HttpClient; + const rateLimiter = yield* HookRateLimiter; + + const resolveEndpoint = Effect.fn("relay.hooks.resolve_endpoint")(function* ( + environmentId: string, + ) { + const candidates = yield* links.findActiveManagedForEnvironment({ environmentId }); + for (const link of candidates) { + const allocation = yield* allocations.get({ userId: link.userId, environmentId }); + const result = validateManagedEndpoint({ + link, + allocation, + baseDomain: settings.managedEndpointBaseDomain, + }); + if (Result.isSuccess(result)) { + return result.success; + } + } + return null; + }); + + const handle = Effect.fn("relay.hooks.forward")(function* ( + request: HttpServerRequest.HttpServerRequest, + ) { + const outcome = (value: string) => Effect.annotateCurrentSpan({ "relay.hook.outcome": value }); + const parsed = parseHookPath(request.url); + if (!parsed) { + yield* outcome("invalid_path"); + return hookNotFound(); + } + yield* Effect.annotateCurrentSpan({ + "relay.environment_id": parsed.environmentId, + "relay.hook_id": parsed.hookId, + }); + if (!(yield* rateLimiter.allow(yield* hookBudgetKey(parsed)))) { + yield* outcome("rate_limited"); + return errorResponse(429, "rate_limited", { + "retry-after": String(RELAY_HOOK_RATE_LIMIT.periodSeconds), + }); + } + const declaredLength = Number(request.headers["content-length"] ?? "0"); + if (Number.isFinite(declaredLength) && declaredLength > RELAY_HOOK_MAX_BODY_BYTES) { + yield* outcome("payload_too_large"); + return errorResponse(413, "payload_too_large"); + } + + const endpoint = yield* resolveEndpoint(parsed.environmentId).pipe( + Effect.catch((error) => + Effect.logWarning("Failed to resolve hook endpoint", { + environmentId: parsed.environmentId, + errorTag: error._tag, + }).pipe(Effect.as(null)), + ), + ); + if (!endpoint) { + yield* outcome("not_found"); + return hookNotFound(); + } + + const body = + request.method === "GET" + ? Result.succeed(new Uint8Array(0)) + : yield* readCappedBody(request).pipe(Effect.result); + if (Result.isFailure(body)) { + if (body.failure._tag === "HookBodyTooLarge") { + yield* outcome("payload_too_large"); + return errorResponse(413, "payload_too_large"); + } + yield* outcome("invalid_body"); + return errorResponse(400, "invalid_body"); + } + + const baseUrl = endpoint.httpBaseUrl.endsWith("/") + ? endpoint.httpBaseUrl + : `${endpoint.httpBaseUrl}/`; + const headers = forwardedHeaders(request.headers); + let upstreamRequest = HttpClientRequest.make( + request.method as "GET" | "POST" | "PUT" | "PATCH", + )(`${baseUrl}api/hooks/${parsed.rawHookId}/${parsed.rawToken}${parsed.search}`, { headers }); + if (request.method !== "GET") { + upstreamRequest = HttpClientRequest.bodyUint8Array( + upstreamRequest, + body.success, + headers["content-type"], + ); + } + + const upstream = yield* httpClient.execute(upstreamRequest).pipe( + Effect.flatMap((response) => + readCappedResponse(response).pipe( + Effect.map((body) => ({ + status: response.status, + contentType: response.headers["content-type"], + body, + })), + ), + ), + withoutRedirects, + // The client span would record url.full, which carries the token. + Effect.provideService(HttpClient.TracerDisabledWhen, () => true), + Effect.timeoutOption(Duration.millis(RELAY_HOOK_UPSTREAM_TIMEOUT_MS)), + Effect.result, + ); + if (Result.isFailure(upstream)) { + yield* outcome("environment_unavailable"); + return errorResponse(503, "environment_unavailable"); + } + if (Option.isNone(upstream.success)) { + yield* outcome("environment_timeout"); + return errorResponse(504, "environment_timeout"); + } + const response = upstream.success.value; + if (response.status === TUNNEL_OFFLINE_STATUS) { + yield* outcome("environment_unavailable"); + return errorResponse(503, "environment_unavailable"); + } + yield* Effect.annotateCurrentSpan({ + "relay.hook.outcome": "forwarded", + "relay.hook.upstream_status": response.status, + }); + // Only content-type is passed through: no location (redirects are never + // followed or relayed), no cookies, no upstream infrastructure headers. + const contentTypeHeaders = response.contentType + ? { "content-type": response.contentType } + : undefined; + if (response.body.length === 0) { + return HttpServerResponse.empty({ + status: response.status, + ...(contentTypeHeaders ? { headers: contentTypeHeaders } : {}), + }); + } + return HttpServerResponse.uint8Array(response.body, { + status: response.status, + ...(response.contentType ? { contentType: response.contentType } : {}), + }); + }); + + return HookForwarder.of({ handle }); +}); + +export const layer = Layer.effect(HookForwarder, make); + +/** + * Implements the RelayApi `hooks` group. The endpoints are raw: the forwarder + * re-reads the encoded path segments from the request so the token and hook + * id reach the environment byte for byte, and streams the body itself. + */ +export const hooksApi = HttpApiBuilder.group( + RelayApi, + "hooks", + Effect.fnUntraced(function* (handlers) { + const forwarder = yield* HookForwarder; + const forward = ({ request }: { readonly request: HttpServerRequest.HttpServerRequest }) => + forwarder.handle(request); + return handlers + .handleRaw("forwardPost", forward) + .handleRaw("forwardPut", forward) + .handleRaw("forwardPatch", forward) + .handleRaw("forwardGet", forward); + }), +); diff --git a/infra/relay/src/http/Api.test.ts b/infra/relay/src/http/Api.test.ts index 6e0db9cd144cc2ca995629684843af8321724b19..6220c27c4b30af1e45b3a285644d583b25d94871 100644 --- a/infra/relay/src/http/Api.test.ts +++ b/infra/relay/src/http/Api.test.ts @@ -332,6 +332,7 @@ function relayUnlinkTestLayer(input?: { listDeliveryUsersForEnvironment: () => Effect.die("unused listDeliveryUsersForEnvironment"), listForUser: () => Effect.die("unused listForUser"), getForUser: input?.getForUser ?? (() => Effect.succeed(null)), + findActiveManagedForEnvironment: () => Effect.succeed([]), revokeForUser: input?.revokeForUser ?? (() => Effect.succeed(false)), }), ), diff --git a/infra/relay/src/http/Api.ts b/infra/relay/src/http/Api.ts index ef2f31ea6599ab79eaaf22e83bf45be07d23d133..5c3b1fe5cfdebc2b9821dfc5113a53d273811d85 100644 --- a/infra/relay/src/http/Api.ts +++ b/infra/relay/src/http/Api.ts @@ -11,6 +11,7 @@ import * as Record from "effect/Record"; import * as Redacted from "effect/Redacted"; import * as Schema from "effect/Schema"; import * as Tracer from "effect/Tracer"; +import * as Headers from "effect/http/Headers"; import * as HttpEffect from "effect/http/HttpEffect"; import * as HttpMiddleware from "effect/http/HttpMiddleware"; import * as HttpRouter from "effect/http/HttpRouter"; @@ -75,6 +76,7 @@ import * as ManagedEndpointAllocations from "../environments/ManagedEndpointAllo import * as EnvironmentPublishSignatures from "../environments/EnvironmentPublishSignatures.ts"; import * as MobileRegistrations from "../agentActivity/MobileRegistrations.ts"; import { withSpanAttributes } from "../observability.ts"; +import { isRelayHookPath, redactRelayHookUrl } from "../hooks/HookForwarder.ts"; import * as RelayDb from "../db.ts"; // Delegated thread IDs carry escaped command provenance and exceed the router's @@ -150,6 +152,10 @@ export const relayCors = HttpRouter.middleware( >, ) { const request = yield* HttpServerRequest.HttpServerRequest; + // Public webhook forwarding is server-to-server: no preflight, no CORS grants. + if (isRelayHookPath(request.url)) { + return yield* httpEffect; + } if (request.method === "OPTIONS") { return HttpServerResponse.empty({ status: 204, @@ -222,10 +228,45 @@ export const traceRelayHttpRequest = ( HttpServerRequest.HttpServerRequest | R >, ) => - // HttpMiddleware finalizes its span on the dispatcher; do not close a request-scoped exporter first. - HttpMiddleware.tracer( - appendRelayTraceContextResponseHeader.pipe(Effect.andThen(relayRequestDeadline(httpEffect))), - ).pipe(Effect.ensuring(Effect.yieldNow)); + Effect.gen(function* () { + const request = yield* HttpServerRequest.HttpServerRequest; + const traced = appendRelayTraceContextResponseHeader.pipe( + Effect.andThen(relayRequestDeadline(httpEffect)), + ); + if (!isRelayHookPath(request.url)) { + // HttpMiddleware finalizes its span on the dispatcher; do not close a request-scoped exporter first. + return yield* HttpMiddleware.tracer(traced).pipe(Effect.ensuring(Effect.yieldNow)); + } + // Hook URLs carry a secret token: the tracer and deadline log see a redacted + // request, while the route itself still receives the original. + const redacted = request.modify({ url: redactRelayHookUrl(request.url) }); + return yield* HttpMiddleware.tracer( + appendRelayTraceContextResponseHeader.pipe( + Effect.andThen( + relayRequestDeadline( + httpEffect.pipe(Effect.provideService(HttpServerRequest.HttpServerRequest, request)), + ), + ), + ), + ).pipe( + Effect.provideService(HttpServerRequest.HttpServerRequest, redacted), + Effect.ensuring(Effect.yieldNow), + ); + }); + +// Webhook senders put shared secrets and signatures in headers such as +// x-hub-signature-256, stripe-signature, x-gitlab-token and x-webhook-key. +const webhookHeaderRedactionLayer = Layer.effect( + Headers.CurrentRedactedNames, + Effect.map(Headers.CurrentRedactedNames, (names) => [ + ...names, + /signature/i, + /token/i, + /secret/i, + /key/i, + /auth/i, + ]), +); export const traceRelayHttpRequestWith = ( httpEffect: Effect.Effect< @@ -236,7 +277,12 @@ export const traceRelayHttpRequestWith = ( tracerLayer: Layer.Layer, ) => traceRelayHttpRequest(httpEffect).pipe( - Effect.provide(Layer.merge(tracerLayer, httpHeaderRedactionLayer)), + Effect.provide( + Layer.merge( + tracerLayer, + webhookHeaderRedactionLayer.pipe(Layer.provide(httpHeaderRedactionLayer)), + ), + ), ); export const withoutCapturedParentSpan = ( diff --git a/infra/relay/src/worker.ts b/infra/relay/src/worker.ts index 4571a707db2fba4b44ca5c80d3f996d58f0c957d..cc03abd5bee8f7af9e4e81d29806103f2956e015 100644 --- a/infra/relay/src/worker.ts +++ b/infra/relay/src/worker.ts @@ -73,6 +73,7 @@ import * as ManagedEndpointProvider from "./environments/ManagedEndpointProvider import * as ManagedEndpointReaper from "./environments/ManagedEndpointReaper.ts"; import * as ManagedTunnelLimits from "./environments/ManagedTunnelLimits.ts"; import * as MobileRegistrations from "./agentActivity/MobileRegistrations.ts"; +import * as HookForwarder from "./hooks/HookForwarder.ts"; const webcryptoLayer = Layer.succeed( Crypto.Crypto, @@ -189,6 +190,13 @@ export const ApiLive = Api.make( const managedEndpointDnsBinding = yield* Cloudflare.DNS.ReadWriteDns(managedEndpointZone); const managedEndpointZoneName = yield* managedEndpointZone.name; const managedEndpointCleanupMode = yield* RelayConfiguration.managedEndpointCleanupModeConfig; + const hookRateLimit = yield* Cloudflare.RateLimit("HOOK_RATE_LIMIT", { + namespaceId: 1001, + simple: { + limit: HookForwarder.RELAY_HOOK_RATE_LIMIT.limit, + period: HookForwarder.RELAY_HOOK_RATE_LIMIT.periodSeconds, + }, + }); // // 3. Runtime layers and app construction @@ -282,7 +290,27 @@ export const ApiLive = Api.make( Layer.provideMerge(webcryptoLayer), ); - const appLayer = relayApiLayer.pipe( + // Fails open: a limiter outage must not drop webhooks the environment would accept. + const hookRateLimiterLayer = Layer.succeed(HookForwarder.HookRateLimiter, { + allow: (key) => + hookRateLimit.limit({ key }).pipe( + Effect.map((result) => result.success), + Effect.provideService(Alchemy.RuntimeContext, alchemyRuntimeContext), + Effect.catch((error) => + Effect.logWarning("Hook rate limiter unavailable", { error: error.message }).pipe( + Effect.as(true), + ), + ), + ), + }); + + const appLayer = Layer.merge( + relayApiLayer, + HookForwarder.hooksApi.pipe( + Layer.provide(HookForwarder.layer), + Layer.provide(hookRateLimiterLayer), + ), + ).pipe( Layer.provideMerge(relayClientAuthLayer), Layer.provideMerge(relayDpopClientAuthLayer), Layer.provideMerge(relayEnvironmentAuthLayer), @@ -397,6 +425,7 @@ export const ApiLive = Api.make( Layer.provideMerge(Cloudflare.Queues.EventSourceLive), Layer.provideMerge(Cloudflare.Tunnel.ReadWriteTunnelBinding), Layer.provideMerge(Cloudflare.DNS.ReadWriteDnsHttp), + Layer.provideMerge(Cloudflare.Workers.RateLimitBinding), ), ), ), diff --git a/packages/contracts/src/relay.ts b/packages/contracts/src/relay.ts index 832c5405ba64b19a329e8ddedbdde25be53842f5..95d680bac24630db34b95aa37b2f2bd89a0cfca7 100644 --- a/packages/contracts/src/relay.ts +++ b/packages/contracts/src/relay.ts @@ -1175,6 +1175,28 @@ const RelayServerGroup = HttpApiGroup.make("server") .annotate(OpenApi.Description, "Environment-authenticated activity publication.") .middleware(RelayEnvironmentAuth); +/** + * Public, stateless webhook forwarding to an environment's managed tunnel. + * Unauthenticated: the token in the path is the environment's credential, and + * the relay only routes and forwards. Raw, so the body reaches the + * environment byte for byte and signatures still verify there. + */ +const RelayHookParams = Schema.Struct({ + environmentId: Schema.String, + hookId: Schema.String, + token: Schema.String, +}); +const RELAY_HOOK_PATH = "/v1/hooks/:environmentId/:hookId/:token"; +const relayHookEndpoint = { params: RelayHookParams } as const; +const RelayHooksGroup = HttpApiGroup.make("hooks") + .add( + HttpApiEndpoint.post("forwardPost", RELAY_HOOK_PATH, relayHookEndpoint), + HttpApiEndpoint.put("forwardPut", RELAY_HOOK_PATH, relayHookEndpoint), + HttpApiEndpoint.patch("forwardPatch", RELAY_HOOK_PATH, relayHookEndpoint), + HttpApiEndpoint.get("forwardGet", RELAY_HOOK_PATH, relayHookEndpoint), + ) + .annotate(OpenApi.Description, "Forward webhook requests to an environment."); + export const RelayApi = HttpApi.make("RelayApi") .add( RelayHealthGroup, @@ -1184,6 +1206,7 @@ export const RelayApi = HttpApi.make("RelayApi") RelayTokenGroup, RelayDpopClientGroup, RelayServerGroup, + RelayHooksGroup, ) .annotate(OpenApi.Title, "T3 Code Relay API") .annotate(OpenApi.Version, "1.0.0") -- 2.54.0