authorgravatar for julius0216@outlook.comJulius Marminge <julius0216@outlook.com> 2026-10-05 15:14:51-07:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-10-05 15:14:51-07:00
logb070e52d3e6b49c98117249e5b3f0d46edce173a
treed267d6b1427c7815cb5efe9919092b584eb5fa2a
parent7dfb86a32be6ec064dd8988e1a3b44cc046ff958
signature Signed by PGP key B5690EEEBB952194

feat(relay): forward webhook requests to the environment's tunnel (#15086)

Co-authored-by: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

14 files changed, 1118 insertions(+), 86 deletions(-)

docs/internals/t3-connect.md+6
......@@ -4,6 +4,12 @@ T3 Connect uses Clerk for cloud identity. The relay manages environment links,
44credentials for reaching environments, and managed tunnel allocations. After
55bootstrap, clients send application traffic through the environment's tunnel
66hostname; the relay Worker does not proxy their HTTP or WebSocket sessions.
7The one exception is automation webhooks: the relay forwards
8`/v1/hooks/:environmentId/:hookId/:token` statelessly to the environment's
9tunnel so senders get a stable URL. It stores nothing, keeps bodies and tokens
10out of its traces, and leaves token and signature verification to the
11environment
12([forwarder](../../infra/relay/src/hooks/HookForwarder.ts)).
713
814Clerk, deployment, and native authentication setup live in the
915[Connect setup runbook](../operations/connect-setup.md).
infra/relay/src/agentActivity/AgentActivityPublisher.test.ts+1
......@@ -94,6 +94,7 @@ function makeEnvironmentLinks(
9494 ]),
9595 listForUser: () => Effect.succeed([]),
9696 getForUser: () => Effect.succeed(null),
97 findActiveManagedForEnvironment: () => Effect.succeed([]),
9798 revokeForUser: () => Effect.succeed(false),
9899 ...overrides,
99100 };
infra/relay/src/agentActivity/FcmDeliveries.test.ts+1
......@@ -171,6 +171,7 @@ function harness() {
171171 : [],
172172 ),
173173 listForUser: () => Effect.succeed([]),
174 findActiveManagedForEnvironment: () => Effect.succeed([]),
174175 revokeForUser: () => Effect.succeed(false),
175176 getForUser: (input) =>
176177 Effect.sync(() =>
infra/relay/src/agentActivity/MobileRegistrations.test.ts+1
......@@ -116,6 +116,7 @@ function makeEnvironmentLinks(
116116 ]),
117117 listForUser: () => Effect.succeed([]),
118118 getForUser: () => Effect.succeed(null),
119 findActiveManagedForEnvironment: () => Effect.succeed([]),
119120 revokeForUser: () => Effect.succeed(false),
120121 ...overrides,
121122 };
infra/relay/src/environments/EnvironmentConnector.test.ts+1
......@@ -229,6 +229,7 @@ function makeLinks(
229229 environmentPublicKey: environmentKeyPair.publicKey,
230230 ...overrides,
231231 }),
232 findActiveManagedForEnvironment: () => Effect.succeed([]),
232233 revokeForUser: () => Effect.succeed(false),
233234 };
234235}
infra/relay/src/environments/EnvironmentConnector.ts+101-80
......@@ -16,6 +16,7 @@ import {
1616 RelayEnvironmentConnectNotAuthorizedReason,
1717 type RelayEnvironmentConnectResponse,
1818 type RelayEnvironmentStatusResponse,
19 type RelayManagedEndpoint,
1920} from "@t3tools/contracts/relay";
2021import {
2122 normalizeRelayIssuer,
......@@ -35,6 +36,7 @@ import * as Effect from "effect/Effect";
3536import * as Layer from "effect/Layer";
3637import * as Option from "effect/Option";
3738import * as Redacted from "effect/Redacted";
39import * as Result from "effect/Result";
3840import * as Schema from "effect/Schema";
3941import * as FetchHttpClient from "effect/http/FetchHttpClient";
4042import * as HttpClient from "effect/http/HttpClient";
......@@ -185,7 +187,7 @@ const currentTraceId = Effect.currentSpan.pipe(
185187 Effect.orElseSucceed(() => "unavailable"),
186188);
187189
188const withoutRedirects = <A, E, R>(effect: Effect.Effect<A, E, R>) =>
190export const withoutRedirects = <A, E, R>(effect: Effect.Effect<A, E, R>) =>
189191 effect.pipe(Effect.provideService(FetchHttpClient.RequestInit, { redirect: "manual" }));
190192
191193const verifyWithEnvironmentKeys = Effect.fnUntraced(function* <A, E>(input: {
......@@ -288,6 +290,92 @@ function verifyEnvironmentHealthResponse(input: {
288290 );
289291}
290292
293export interface ManagedEndpointValidationFailure {
294 readonly reason: Exclude<
295 RelayEnvironmentConnectNotAuthorizedReason,
296 "client_proof_key_thumbprint_missing" | "environment_link_not_found"
297 >;
298 /** Diagnostic span attributes; never contains secrets. */
299 readonly attributes?: Record<string, string | boolean>;
300}
301
302/**
303 * Checks that a link's relay-managed endpoint is backed by a ready allocation
304 * on the configured base domain and still matches what the link recorded.
305 */
306export function validateManagedEndpoint(input: {
307 readonly link: EnvironmentLinks.RelayLinkedEnvironmentRecord;
308 readonly allocation: ManagedEndpointAllocations.ManagedEndpointAllocation | null;
309 readonly baseDomain: string | undefined;
310}): Result.Result<RelayManagedEndpoint, ManagedEndpointValidationFailure> {
311 const { link, allocation, baseDomain } = input;
312 if (link.endpoint.providerKind !== "cloudflare_tunnel") {
313 return Result.fail({
314 reason: "endpoint_provider_not_managed",
315 attributes: { "relay.authorization.endpoint_provider_kind": link.endpoint.providerKind },
316 });
317 }
318 if (!allocation) {
319 return Result.fail({ reason: "managed_endpoint_allocation_not_found" });
320 }
321 const allocationAttributes = {
322 "relay.authorization.allocation_hostname": allocation.hostname,
323 "relay.authorization.allocation_has_ready_at": allocation.readyAt !== null,
324 "relay.authorization.allocation_has_tunnel_id": allocation.tunnelId !== null,
325 "relay.authorization.allocation_has_dns_record_id": allocation.dnsRecordId !== null,
326 };
327 if (!baseDomain) {
328 return Result.fail({
329 reason: "managed_endpoint_base_domain_not_configured",
330 attributes: allocationAttributes,
331 });
332 }
333 if (
334 allocation.readyAt === null ||
335 allocation.tunnelId === null ||
336 allocation.dnsRecordId === null
337 ) {
338 return Result.fail({
339 reason: "managed_endpoint_allocation_not_ready",
340 attributes: allocationAttributes,
341 });
342 }
343 if (!isManagedEndpointHostname(allocation.hostname, baseDomain)) {
344 return Result.fail({
345 reason: "managed_endpoint_hostname_invalid",
346 attributes: {
347 ...allocationAttributes,
348 "relay.authorization.managed_endpoint_base_domain": baseDomain,
349 },
350 });
351 }
352 const endpoint = ManagedEndpointAllocations.resolveReadyManagedEndpoint({
353 allocation,
354 baseDomain,
355 });
356 if (
357 endpoint === null ||
358 endpoint.httpBaseUrl !== link.endpoint.httpBaseUrl ||
359 endpoint.wsBaseUrl !== link.endpoint.wsBaseUrl
360 ) {
361 return Result.fail({
362 reason: "managed_endpoint_mismatch",
363 attributes: {
364 ...allocationAttributes,
365 "relay.authorization.linked_http_base_url": link.endpoint.httpBaseUrl,
366 "relay.authorization.linked_ws_base_url": link.endpoint.wsBaseUrl,
367 ...(endpoint
368 ? {
369 "relay.authorization.resolved_http_base_url": endpoint.httpBaseUrl,
370 "relay.authorization.resolved_ws_base_url": endpoint.wsBaseUrl,
371 }
372 : {}),
373 },
374 });
375 }
376 return Result.succeed(endpoint);
377}
378
291379const make = Effect.gen(function* () {
292380 const links = yield* EnvironmentLinks.EnvironmentLinks;
293381 const allocations = yield* ManagedEndpointAllocations.ManagedEndpointAllocations;
......@@ -305,89 +393,22 @@ const make = Effect.gen(function* () {
305393 readonly link: EnvironmentLinks.RelayLinkedEnvironmentRecord;
306394 readonly allocation: ManagedEndpointAllocations.ManagedEndpointAllocation | null;
307395 }) {
308 if (input.link.endpoint.providerKind !== "cloudflare_tunnel") {
309 yield* Effect.annotateCurrentSpan({
310 "relay.authorization.endpoint_provider_kind": input.link.endpoint.providerKind,
311 });
312 return yield* new EnvironmentConnectNotAuthorized({
313 environmentId: input.link.environmentId,
314 operation: input.operation,
315 reason: "endpoint_provider_not_managed",
316 });
317 }
318 if (!input.allocation) {
319 return yield* new EnvironmentConnectNotAuthorized({
320 environmentId: input.link.environmentId,
321 operation: input.operation,
322 reason: "managed_endpoint_allocation_not_found",
323 });
324 }
325 const allocationAttributes = {
326 "relay.authorization.allocation_hostname": input.allocation.hostname,
327 "relay.authorization.allocation_has_ready_at": input.allocation.readyAt !== null,
328 "relay.authorization.allocation_has_tunnel_id": input.allocation.tunnelId !== null,
329 "relay.authorization.allocation_has_dns_record_id": input.allocation.dnsRecordId !== null,
330 } as const;
331 if (!settings.managedEndpointBaseDomain) {
332 yield* Effect.annotateCurrentSpan(allocationAttributes);
333 return yield* new EnvironmentConnectNotAuthorized({
334 environmentId: input.link.environmentId,
335 operation: input.operation,
336 reason: "managed_endpoint_base_domain_not_configured",
337 });
338 }
339 if (
340 input.allocation.readyAt === null ||
341 input.allocation.tunnelId === null ||
342 input.allocation.dnsRecordId === null
343 ) {
344 yield* Effect.annotateCurrentSpan(allocationAttributes);
345 return yield* new EnvironmentConnectNotAuthorized({
346 environmentId: input.link.environmentId,
347 operation: input.operation,
348 reason: "managed_endpoint_allocation_not_ready",
349 });
350 }
351 if (
352 !isManagedEndpointHostname(input.allocation.hostname, settings.managedEndpointBaseDomain)
353 ) {
354 yield* Effect.annotateCurrentSpan({
355 ...allocationAttributes,
356 "relay.authorization.managed_endpoint_base_domain": settings.managedEndpointBaseDomain,
357 });
358 return yield* new EnvironmentConnectNotAuthorized({
359 environmentId: input.link.environmentId,
360 operation: input.operation,
361 reason: "managed_endpoint_hostname_invalid",
362 });
363 }
364 const endpoint = ManagedEndpointAllocations.resolveReadyManagedEndpoint({
396 const result = validateManagedEndpoint({
397 link: input.link,
365398 allocation: input.allocation,
366399 baseDomain: settings.managedEndpointBaseDomain,
367400 });
368 if (
369 endpoint === null ||
370 endpoint.httpBaseUrl !== input.link.endpoint.httpBaseUrl ||
371 endpoint.wsBaseUrl !== input.link.endpoint.wsBaseUrl
372 ) {
373 yield* Effect.annotateCurrentSpan({
374 ...allocationAttributes,
375 "relay.authorization.linked_http_base_url": input.link.endpoint.httpBaseUrl,
376 "relay.authorization.linked_ws_base_url": input.link.endpoint.wsBaseUrl,
377 ...(endpoint
378 ? {
379 "relay.authorization.resolved_http_base_url": endpoint.httpBaseUrl,
380 "relay.authorization.resolved_ws_base_url": endpoint.wsBaseUrl,
381 }
382 : {}),
383 });
384 return yield* new EnvironmentConnectNotAuthorized({
385 environmentId: input.link.environmentId,
386 operation: input.operation,
387 reason: "managed_endpoint_mismatch",
388 });
401 if (Result.isSuccess(result)) {
402 return result.success;
389403 }
390 return endpoint;
404 if (result.failure.attributes) {
405 yield* Effect.annotateCurrentSpan(result.failure.attributes);
406 }
407 return yield* new EnvironmentConnectNotAuthorized({
408 environmentId: input.link.environmentId,
409 operation: input.operation,
410 reason: result.failure.reason,
411 });
391412 },
392413 );
393414
infra/relay/src/environments/EnvironmentLinker.test.ts+1
......@@ -126,6 +126,7 @@ function testLayer(input?: {
126126 listDeliveryUsersForEnvironment: () => Effect.succeed([]),
127127 listForUser: () => Effect.succeed([]),
128128 getForUser: () => Effect.succeed(null),
129 findActiveManagedForEnvironment: () => Effect.succeed([]),
129130 revokeForUser: () => Effect.succeed(false),
130131 }),
131132 Layer.succeed(EnvironmentCredentials.EnvironmentCredentials, {
infra/relay/src/environments/EnvironmentLinks.ts+72
......@@ -75,6 +75,18 @@ export class EnvironmentLinkLookupPersistenceError extends Schema.TaggedError<En
7575 }
7676}
7777
78export class EnvironmentLinkEnvironmentLookupPersistenceError extends Schema.TaggedError<EnvironmentLinkEnvironmentLookupPersistenceError>()(
79 "EnvironmentLinkEnvironmentLookupPersistenceError",
80 {
81 environmentId: Schema.String,
82 cause: Schema.Defect(),
83 },
84) {
85 override get message(): string {
86 return `Failed to look up active managed links for environment '${this.environmentId}'`;
87 }
88}
89
7890export class EnvironmentLinkRevokePersistenceError extends Schema.TaggedError<EnvironmentLinkRevokePersistenceError>()(
7991 "EnvironmentLinkRevokePersistenceError",
8092 {
......@@ -114,6 +126,13 @@ export class EnvironmentLinks extends Context.Service<
114126 readonly userId: string;
115127 readonly environmentId: string;
116128 }) => Effect.Effect<RelayLinkedEnvironmentRecord | null, EnvironmentLinkLookupPersistenceError>;
129 /** Active relay-managed links for an environment, across all users (webhook forwarding). */
130 readonly findActiveManagedForEnvironment: (input: {
131 readonly environmentId: string;
132 }) => Effect.Effect<
133 ReadonlyArray<RelayLinkedEnvironmentRecord & { readonly userId: string }>,
134 EnvironmentLinkEnvironmentLookupPersistenceError
135 >;
117136 readonly revokeForUser: (input: {
118137 readonly userId: string;
119138 readonly environmentId: string;
......@@ -322,6 +341,59 @@ const make = Effect.gen(function* () {
322341 );
323342 }),
324343
344 findActiveManagedForEnvironment: Effect.fn(
345 "relay.environment_links.find_active_managed_for_environment",
346 )(function* (input) {
347 yield* Effect.annotateCurrentSpan({ "relay.environment_id": input.environmentId });
348 return yield* db
349 .select({
350 userId: relayEnvironmentLinks.userId,
351 environmentId: relayEnvironmentLinks.environmentId,
352 environmentLabel: relayEnvironmentLinks.environmentLabel,
353 environmentPublicKey: relayEnvironmentLinks.environmentPublicKey,
354 endpointHttpBaseUrl: relayEnvironmentLinks.endpointHttpBaseUrl,
355 endpointWsBaseUrl: relayEnvironmentLinks.endpointWsBaseUrl,
356 endpointProviderKind: relayEnvironmentLinks.endpointProviderKind,
357 createdAt: relayEnvironmentLinks.createdAt,
358 })
359 .from(relayEnvironmentLinks)
360 .where(
361 and(
362 eq(relayEnvironmentLinks.environmentId, input.environmentId),
363 isNull(relayEnvironmentLinks.revokedAt),
364 eq(relayEnvironmentLinks.endpointProviderKind, "cloudflare_tunnel"),
365 eq(relayEnvironmentLinks.managedTunnelsEnabled, true),
366 ),
367 )
368 // One row per user who linked this environment; every row is checked
369 // until one has a ready endpoint, so none may be cut off.
370 .pipe(
371 Effect.map((rows) =>
372 rows.map((row) => ({
373 userId: row.userId,
374 environmentId: row.environmentId as RelayClientEnvironmentRecord["environmentId"],
375 label:
376 row.environmentLabel.trim().length > 0 ? row.environmentLabel : row.environmentId,
377 endpoint: {
378 httpBaseUrl: row.endpointHttpBaseUrl,
379 wsBaseUrl: row.endpointWsBaseUrl,
380 providerKind:
381 row.endpointProviderKind as RelayClientEnvironmentRecord["endpoint"]["providerKind"],
382 },
383 environmentPublicKey: row.environmentPublicKey,
384 linkedAt: row.createdAt,
385 })),
386 ),
387 Effect.mapError(
388 (cause) =>
389 new EnvironmentLinkEnvironmentLookupPersistenceError({
390 environmentId: input.environmentId,
391 cause,
392 }),
393 ),
394 );
395 }),
396
325397 revokeForUser: Effect.fn("relay.environment_links.revoke_for_user")(function* (input) {
326398 yield* Effect.annotateCurrentSpan({
327399 "relay.environment_id": input.environmentId,
infra/relay/src/hooks/HookForwarder.test.ts created+462
......@@ -0,0 +1,462 @@
1import * as NodeHttpPlatform from "@effect/platform-node/NodeHttpPlatform";
2import * as NodeServices from "@effect/platform-node/NodeServices";
3import { describe, expect, it } from "@effect/vitest";
4import { RelayApi } from "@t3tools/contracts/relay";
5import * as Deferred from "effect/Deferred";
6import * as Duration from "effect/Duration";
7import * as Effect from "effect/Effect";
8import * as Fiber from "effect/Fiber";
9import * as Layer from "effect/Layer";
10import * as Redacted from "effect/Redacted";
11import * as TestClock from "effect/testing/TestClock";
12import * as Tracer from "effect/Tracer";
13import * as HttpClient from "effect/http/HttpClient";
14import * as HttpClientError from "effect/http/HttpClientError";
15import type * as HttpClientRequest from "effect/http/HttpClientRequest";
16import * as HttpClientResponse from "effect/http/HttpClientResponse";
17import * as Etag from "effect/http/Etag";
18import * as HttpRouter from "effect/http/HttpRouter";
19import * as HttpApi from "effect/http-api/HttpApi";
20import * as HttpApiBuilder from "effect/http-api/HttpApiBuilder";
21import * as HttpServerRequest from "effect/http/HttpServerRequest";
22import * as HttpServerResponse from "effect/http/HttpServerResponse";
23
24import * as RelayConfiguration from "../Config.ts";
25import * as EnvironmentLinks from "../environments/EnvironmentLinks.ts";
26import * as ManagedEndpointAllocations from "../environments/ManagedEndpointAllocations.ts";
27import {
28 RELAY_HTTP_ROUTER_CONFIG,
29 relayCors,
30 relayNotFoundRoute,
31 traceRelayHttpRequestWith,
32} from "../http/Api.ts";
33import * as HookForwarder from "./HookForwarder.ts";
34
35const settings: RelayConfiguration.RelayConfiguration["Service"] = {
36 relayIssuer: "https://relay.example.test",
37 apns: null,
38 clerkSecretKey: Redacted.make("clerk-secret-key"),
39 clerkPublishableKey: "pk_test_test",
40 clerkJwtAudience: "t3-code-relay",
41 apnsDeliveryJobSigningSecret: Redacted.make("apns-delivery-secret"),
42 cloudMintPrivateKey: Redacted.make("cloud-mint-private-key"),
43 cloudMintPublicKey: "cloud-mint-public-key",
44 managedEndpointBaseDomain: "example.test",
45 managedEndpointNamespace: undefined,
46};
47
48const environmentId = "env-hook";
49const readyAllocation: ManagedEndpointAllocations.ManagedEndpointAllocation = {
50 userId: "user_1",
51 environmentId,
52 hostname: "env.example.test",
53 tunnelId: "tunnel-id",
54 tunnelName: "tunnel-name",
55 dnsRecordId: "dns-record-id",
56 readyAt: "2026-05-25T00:00:00.000Z",
57 origin: { localHttpHost: "127.0.0.1", localHttpPort: 3773 },
58 updatedAt: "2026-05-25T00:00:00.000Z",
59 generation: 1,
60};
61
62const managedLink = {
63 userId: "user_1",
64 environmentId: environmentId as never,
65 label: "Hook env",
66 endpoint: {
67 httpBaseUrl: "https://env.example.test/",
68 wsBaseUrl: "wss://env.example.test/ws",
69 providerKind: "cloudflare_tunnel" as const,
70 },
71 environmentPublicKey: "public-key",
72 linkedAt: "2026-05-25T00:00:00.000Z",
73};
74
75interface Harness {
76 readonly execute?: (
77 request: HttpClientRequest.HttpClientRequest,
78 ) => Effect.Effect<HttpClientResponse.HttpClientResponse, HttpClientError.HttpClientError>;
79 readonly links?: ReadonlyArray<typeof managedLink>;
80 readonly allocation?: ManagedEndpointAllocations.ManagedEndpointAllocation | null;
81 readonly allow?: (key: string) => boolean;
82}
83
84function makeHarness(options: Harness = {}) {
85 const sent: Array<HttpClientRequest.HttpClientRequest> = [];
86 const rateLimitKeys: Array<string> = [];
87 const execute =
88 options.execute ??
89 ((request: HttpClientRequest.HttpClientRequest) =>
90 Effect.succeed(
91 HttpClientResponse.fromWeb(
92 request,
93 new Response("ok", { status: 200, headers: { "content-type": "text/plain" } }),
94 ),
95 ));
96 const forwarderLayer = HookForwarder.layer.pipe(
97 Layer.provide(
98 Layer.mergeAll(
99 Layer.succeed(RelayConfiguration.RelayConfiguration, settings),
100 Layer.mock(EnvironmentLinks.EnvironmentLinks, {
101 findActiveManagedForEnvironment: (input) =>
102 Effect.succeed(
103 input.environmentId === environmentId ? (options.links ?? [managedLink]) : [],
104 ),
105 }),
106 Layer.mock(ManagedEndpointAllocations.ManagedEndpointAllocations, {
107 get: () =>
108 Effect.succeed(options.allocation === undefined ? readyAllocation : options.allocation),
109 }),
110 Layer.succeed(
111 HttpClient.HttpClient,
112 HttpClient.make((request) => {
113 sent.push(request);
114 return execute(request);
115 }),
116 ),
117 Layer.succeed(HookForwarder.HookRateLimiter, {
118 allow: (key) =>
119 Effect.sync(() => {
120 rateLimitKeys.push(key);
121 return options.allow ? options.allow(key) : true;
122 }),
123 }),
124 ),
125 ),
126 );
127 const httpEffect = HttpRouter.toHttpEffect(
128 Layer.mergeAll(
129 HttpApiBuilder.layer(HttpApi.make("RelayApi").add(RelayApi.groups.hooks)).pipe(
130 Layer.provide(HookForwarder.hooksApi.pipe(Layer.provide(forwarderLayer))),
131 Layer.provide([NodeServices.layer, NodeHttpPlatform.layer, Etag.layerWeak]),
132 ),
133 relayNotFoundRoute,
134 relayCors,
135 ),
136 ).pipe(Effect.provideService(HttpRouter.RouterConfig, RELAY_HTTP_ROUTER_CONFIG));
137 const send = (request: Request) =>
138 Effect.gen(function* () {
139 const handler = yield* httpEffect;
140 return yield* handler.pipe(
141 Effect.provideService(
142 HttpServerRequest.HttpServerRequest,
143 HttpServerRequest.fromWeb(request),
144 ),
145 );
146 });
147 return { sent, rateLimitKeys, send, httpEffect };
148}
149
150const hookUrl = (path = "hook-1/secret-token", query = "") =>
151 `https://relay.test/v1/hooks/${environmentId}/${path}${query}`;
152
153const readBody = (response: HttpServerResponse.HttpServerResponse) =>
154 Effect.promise(() => HttpServerResponse.toWeb(response).arrayBuffer()).pipe(
155 Effect.map((buffer) => new Uint8Array(buffer)),
156 );
157const readJson = (response: HttpServerResponse.HttpServerResponse) =>
158 Effect.promise(() => HttpServerResponse.toWeb(response).json());
159
160const requestBytes = (request: HttpClientRequest.HttpClientRequest) =>
161 request.body._tag === "Uint8Array" ? request.body.body : new Uint8Array(0);
162
163describe("HookForwarder", () => {
164 it.effect("forwards the exact body bytes, query and filtered headers", () =>
165 Effect.gen(function* () {
166 const harness = makeHarness();
167 const body = new Uint8Array([0, 255, 10, 13, 0x7b, 0x22, 0xc3, 0x28]);
168 const response = yield* harness.send(
169 new Request(hookUrl("hook-1/tok%2Fen", "?a=1&b=two%20words"), {
170 method: "POST",
171 headers: {
172 "content-type": "application/octet-stream",
173 authorization: "Bearer sender-secret",
174 "x-hub-signature-256": "sha256=abc",
175 cookie: "session=1",
176 "cf-connecting-ip": "1.2.3.4",
177 "x-forwarded-for": "1.2.3.4",
178 "x-real-ip": "1.2.3.4",
179 "proxy-authorization": "Basic x",
180 connection: "keep-alive",
181 },
182 body,
183 }),
184 );
185 expect(response.status).toBe(200);
186 expect(harness.sent).toHaveLength(1);
187 const sent = harness.sent[0]!;
188 expect(sent.method).toBe("POST");
189 expect(sent.url).toBe("https://env.example.test/api/hooks/hook-1/tok%2Fen?a=1&b=two%20words");
190 expect(Array.from(requestBytes(sent))).toEqual(Array.from(body));
191 expect(sent.headers.authorization).toBe("Bearer sender-secret");
192 expect(sent.headers["x-hub-signature-256"]).toBe("sha256=abc");
193 expect(sent.headers["content-type"]).toBe("application/octet-stream");
194 for (const dropped of [
195 "cookie",
196 "cf-connecting-ip",
197 "x-forwarded-for",
198 "x-real-ip",
199 "proxy-authorization",
200 "connection",
201 "host",
202 ]) {
203 expect(sent.headers[dropped]).toBeUndefined();
204 }
205 // Recomputed from the forwarded bytes, not copied from the sender.
206 expect(sent.headers["content-length"]).toBe(String(body.length));
207 // The budget key is a hash, so the token never reaches the limiter.
208 expect(harness.rateLimitKeys).toHaveLength(1);
209 expect(harness.rateLimitKeys[0]).toMatch(/^[0-9a-f]{64}$/);
210 }),
211 );
212
213 it.effect("passes upstream status, body and content-type through", () =>
214 Effect.gen(function* () {
215 const harness = makeHarness({
216 execute: (request) =>
217 Effect.succeed(
218 HttpClientResponse.fromWeb(
219 request,
220 new Response('{"error":"bad_signature"}', {
221 status: 401,
222 headers: { "content-type": "application/json", "set-cookie": "x=1" },
223 }),
224 ),
225 ),
226 });
227 const response = yield* harness.send(new Request(hookUrl(), { method: "GET" }));
228 expect(response.status).toBe(401);
229 expect(response.headers["content-type"]).toBe("application/json");
230 expect(response.headers["set-cookie"]).toBeUndefined();
231 expect(response.headers["access-control-allow-origin"]).toBeUndefined();
232 expect(new TextDecoder().decode(yield* readBody(response))).toBe('{"error":"bad_signature"}');
233 expect(harness.sent[0]?.method).toBe("GET");
234 }),
235 );
236
237 it.effect("does not follow or relay upstream redirects", () =>
238 Effect.gen(function* () {
239 const harness = makeHarness({
240 execute: (request) =>
241 Effect.succeed(
242 HttpClientResponse.fromWeb(
243 request,
244 new Response(null, {
245 status: 302,
246 headers: { location: "https://internal.example/" },
247 }),
248 ),
249 ),
250 });
251 const response = yield* harness.send(new Request(hookUrl(), { method: "POST", body: "{}" }));
252 expect(response.status).toBe(302);
253 expect(response.headers.location).toBeUndefined();
254 expect(harness.sent).toHaveLength(1);
255 }),
256 );
257
258 it.effect("rejects bodies over 1 MiB by content-length and while reading", () =>
259 Effect.gen(function* () {
260 const harness = makeHarness();
261 const declared = yield* harness.send(
262 new Request(hookUrl(), {
263 method: "POST",
264 headers: { "content-length": String(HookForwarder.RELAY_HOOK_MAX_BODY_BYTES + 1) },
265 body: "x",
266 }),
267 );
268 expect(declared.status).toBe(413);
269
270 const oversized = new Uint8Array(HookForwarder.RELAY_HOOK_MAX_BODY_BYTES + 1);
271 const stream = new ReadableStream<Uint8Array>({
272 start(controller) {
273 controller.enqueue(oversized.subarray(0, 600_000));
274 controller.enqueue(oversized.subarray(600_000));
275 controller.close();
276 },
277 });
278 const streamed = yield* harness.send(
279 new Request(hookUrl(), { method: "POST", body: stream, duplex: "half" } as RequestInit),
280 );
281 expect(streamed.status).toBe(413);
282 expect(yield* readJson(streamed)).toEqual({ error: "payload_too_large" });
283 expect(harness.sent).toHaveLength(0);
284 }),
285 );
286
287 it.effect("returns 404 for unknown environments and unready endpoints", () =>
288 Effect.gen(function* () {
289 const unknown = makeHarness();
290 const response = yield* unknown.send(
291 new Request("https://relay.test/v1/hooks/other-env/hook-1/token", { method: "POST" }),
292 );
293 expect(response.status).toBe(404);
294 expect(yield* readJson(response)).toEqual({ error: "hook_not_found" });
295
296 const unready = makeHarness({ allocation: { ...readyAllocation, readyAt: null } });
297 const unreadyResponse = yield* unready.send(new Request(hookUrl(), { method: "POST" }));
298 expect(unreadyResponse.status).toBe(404);
299 expect(unready.sent).toHaveLength(0);
300 }),
301 );
302
303 it.effect("maps tunnel-offline and network failures to 503", () =>
304 Effect.gen(function* () {
305 const offline = makeHarness({
306 execute: (request) =>
307 Effect.succeed(HttpClientResponse.fromWeb(request, new Response("", { status: 530 }))),
308 });
309 const offlineResponse = yield* offline.send(new Request(hookUrl(), { method: "POST" }));
310 expect(offlineResponse.status).toBe(503);
311 expect(yield* readJson(offlineResponse)).toEqual({ error: "environment_unavailable" });
312
313 const network = makeHarness({
314 execute: (request) =>
315 Effect.fail(
316 new HttpClientError.HttpClientError({
317 reason: new HttpClientError.TransportError({ request, cause: new Error("reset") }),
318 }),
319 ),
320 });
321 const networkResponse = yield* network.send(new Request(hookUrl(), { method: "POST" }));
322 expect(networkResponse.status).toBe(503);
323 }),
324 );
325
326 it.effect("does not relay an upstream response larger than the cap", () =>
327 Effect.gen(function* () {
328 const harness = makeHarness({
329 execute: (request) =>
330 Effect.succeed(
331 HttpClientResponse.fromWeb(
332 request,
333 new Response(new Uint8Array(64 * 1024 + 1), { status: 200 }),
334 ),
335 ),
336 });
337 const response = yield* harness.send(new Request(hookUrl(), { method: "POST" }));
338 expect(response.status).toBe(503);
339 expect(yield* readJson(response)).toEqual({ error: "environment_unavailable" });
340 }),
341 );
342
343 it.effect("maps an upstream timeout to 504", () =>
344 Effect.gen(function* () {
345 // Advance the clock only once the request is waiting on the environment;
346 // hashing the budget key before that step is asynchronous.
347 const reachedUpstream = yield* Deferred.make<void>();
348 const harness = makeHarness({
349 execute: () =>
350 Deferred.succeed(reachedUpstream, undefined).pipe(Effect.andThen(Effect.never)),
351 });
352 const fiber = yield* harness
353 .send(new Request(hookUrl(), { method: "POST", body: "{}" }))
354 .pipe(Effect.forkChild);
355 yield* Deferred.await(reachedUpstream);
356 yield* TestClock.adjust(Duration.millis(HookForwarder.RELAY_HOOK_UPSTREAM_TIMEOUT_MS));
357 const response = yield* Fiber.join(fiber);
358 expect(response.status).toBe(504);
359 expect(yield* readJson(response)).toEqual({ error: "environment_timeout" });
360 }),
361 );
362
363 it.effect("answers OPTIONS without a CORS preflight or forwarding", () =>
364 Effect.gen(function* () {
365 const harness = makeHarness();
366 const response = yield* harness.send(new Request(hookUrl(), { method: "OPTIONS" }));
367 // No hook endpoint accepts OPTIONS, so it falls through to the 404 route.
368 expect(response.status).toBe(404);
369 expect(response.headers["access-control-allow-origin"]).toBeUndefined();
370 expect(response.headers["access-control-allow-methods"]).toBeUndefined();
371 expect(harness.sent).toHaveLength(0);
372 }),
373 );
374
375 it.effect("drops headers the sender names in Connection", () =>
376 Effect.gen(function* () {
377 const harness = makeHarness();
378 yield* harness.send(
379 new Request(hookUrl(), {
380 method: "POST",
381 body: "{}",
382 headers: { connection: "x-hop, keep-alive", "x-hop": "1", "x-keep": "2" },
383 }),
384 );
385 const sent: Readonly<Record<string, string>> = harness.sent[0]?.headers ?? {};
386 expect(sent["x-hop"]).toBeUndefined();
387 expect(sent["x-keep"]).toBe("2");
388 }),
389 );
390
391 it.effect("gives requests with a wrong token their own budget", () =>
392 Effect.gen(function* () {
393 const harness = makeHarness();
394 yield* harness.send(
395 new Request(hookUrl("hook-1/real-token"), { method: "POST", body: "{}" }),
396 );
397 yield* harness.send(new Request(hookUrl("hook-1/guessed"), { method: "POST", body: "{}" }));
398 expect(new Set(harness.rateLimitKeys).size).toBe(2);
399 }),
400 );
401
402 it.effect("gives two spellings of one token a single budget", () =>
403 Effect.gen(function* () {
404 const harness = makeHarness();
405 yield* harness.send(new Request(hookUrl("hook-1/token"), { method: "POST", body: "{}" }));
406 yield* harness.send(new Request(hookUrl("hook-1/%74oken"), { method: "POST", body: "{}" }));
407 expect(new Set(harness.rateLimitKeys).size).toBe(1);
408 }),
409 );
410
411 it.effect("returns 429 when the hook's budget is spent", () =>
412 Effect.gen(function* () {
413 const harness = makeHarness({ allow: () => false });
414 const response = yield* harness.send(new Request(hookUrl(), { method: "POST" }));
415 expect(response.status).toBe(429);
416 expect(harness.sent).toHaveLength(0);
417 }),
418 );
419
420 it.effect("never records the token in the server span", () =>
421 Effect.gen(function* () {
422 const spans: Array<Tracer.NativeSpan> = [];
423 const tracer = Tracer.make({
424 span: (options) => {
425 const span = new Tracer.NativeSpan(options);
426 spans.push(span);
427 return span;
428 },
429 });
430 const harness = makeHarness();
431 const handler = yield* harness.httpEffect;
432 const response = yield* traceRelayHttpRequestWith(
433 handler,
434 Layer.succeed(Tracer.Tracer, tracer),
435 ).pipe(
436 Effect.provideService(
437 HttpServerRequest.HttpServerRequest,
438 HttpServerRequest.fromWeb(
439 new Request(hookUrl("hook-1/super-secret-token", "?sig=also-secret"), {
440 method: "POST",
441 headers: { "x-gitlab-token": "header-secret" },
442 body: "{}",
443 }),
444 ),
445 ),
446 );
447 expect(response.status).toBe(200);
448 expect(harness.sent[0]?.url).toContain("super-secret-token");
449 yield* Effect.yieldNow;
450 const serialized = spans
451 .flatMap((span) => [span.name, ...Array.from(span.attributes.values(), String)])
452 .join("\n");
453 expect(serialized).not.toContain("super-secret-token");
454 expect(serialized).not.toContain("also-secret");
455 expect(serialized).not.toContain("header-secret");
456 const server = spans.find((span) => span.kind === "server");
457 expect(server?.attributes.get("url.path")).toBe(
458 `/v1/hooks/${environmentId}/hook-1/<redacted>`,
459 );
460 }),
461 );
462});
infra/relay/src/hooks/HookForwarder.ts created+367
......@@ -0,0 +1,367 @@
1import * as Context from "effect/Context";
2import * as Duration from "effect/Duration";
3import * as Effect from "effect/Effect";
4import * as Layer from "effect/Layer";
5import * as Option from "effect/Option";
6import * as Result from "effect/Result";
7import * as Schema from "effect/Schema";
8import * as Stream from "effect/Stream";
9import * as HttpClient from "effect/http/HttpClient";
10import * as HttpClientRequest from "effect/http/HttpClientRequest";
11import type * as HttpClientResponse from "effect/http/HttpClientResponse";
12import type * as HttpServerRequest from "effect/http/HttpServerRequest";
13import * as HttpServerResponse from "effect/http/HttpServerResponse";
14import * as HttpApiBuilder from "effect/http-api/HttpApiBuilder";
15import { RelayApi } from "@t3tools/contracts/relay";
16
17import * as RelayConfiguration from "../Config.ts";
18import { validateManagedEndpoint, withoutRedirects } from "../environments/EnvironmentConnector.ts";
19import * as EnvironmentLinks from "../environments/EnvironmentLinks.ts";
20import * as ManagedEndpointAllocations from "../environments/ManagedEndpointAllocations.ts";
21
22export const RELAY_HOOK_PATH_PREFIX = "/v1/hooks/";
23export const RELAY_HOOK_MAX_BODY_BYTES = 1_048_576;
24export const RELAY_HOOK_UPSTREAM_TIMEOUT_MS = 8_000;
25export const RELAY_HOOK_RATE_LIMIT = { limit: 60, periodSeconds: 60 } as const;
26
27const DROPPED_REQUEST_HEADERS = new Set([
28 "host",
29 "connection",
30 "keep-alive",
31 "transfer-encoding",
32 "te",
33 "upgrade",
34 "content-length",
35 "cookie",
36 "x-real-ip",
37]);
38const DROPPED_REQUEST_HEADER_PREFIXES = ["proxy-", "cf-", "x-forwarded-"];
39// Cloudflare answers 530 when the tunnel for a hostname has no connected origin.
40const TUNNEL_OFFLINE_STATUS = 530;
41/** The environment answers with a small JSON status; anything past this is cut off. */
42const MAX_RESPONSE_BYTES = 64 * 1024;
43
44export const isRelayHookPath = (url: string): boolean => url.startsWith(RELAY_HOOK_PATH_PREFIX);
45
46/** Replaces the hook token (and any query) so traces and logs never record the secret. */
47export const redactRelayHookUrl = (url: string): string => {
48 const path = url.split("?", 1)[0] ?? url;
49 const segments = path.split("/");
50 if (segments.length >= 6) {
51 segments[5] = "<redacted>";
52 }
53 return segments.join("/");
54};
55
56/**
57 * Request budget for public hook forwarding, keyed by a hash of the hook URL
58 * (environment, hook and token). Requests with a wrong token get their own
59 * budget, so they cannot use up a real sender's; the environment rejects them.
60 * Built from decoded segments, because the environment decodes them too: two
61 * spellings of one token (`token`, `%74oken`) must share one budget.
62 */
63const hookBudgetKey = (hook: {
64 readonly environmentId: string;
65 readonly hookId: string;
66 readonly token: string;
67}) =>
68 Effect.promise(() =>
69 crypto.subtle.digest(
70 "SHA-256",
71 // Length-prefixed, so no segment contents can make two keys collide.
72 new TextEncoder().encode(
73 [hook.environmentId, hook.hookId, hook.token]
74 .map((part) => `${part.length}:${part}`)
75 .join(""),
76 ),
77 ),
78 ).pipe(
79 Effect.map((digest) =>
80 Array.from(new Uint8Array(digest), (byte) => byte.toString(16).padStart(2, "0")).join(""),
81 ),
82 );
83
84export class HookRateLimiter extends Context.Service<
85 HookRateLimiter,
86 { readonly allow: (key: string) => Effect.Effect<boolean> }
87>()("t3code-relay/hooks/HookForwarder/HookRateLimiter") {}
88
89export class HookForwarder extends Context.Service<
90 HookForwarder,
91 {
92 readonly handle: (
93 request: HttpServerRequest.HttpServerRequest,
94 ) => Effect.Effect<HttpServerResponse.HttpServerResponse>;
95 }
96>()("t3code-relay/hooks/HookForwarder") {}
97
98class HookBodyTooLarge extends Schema.TaggedError<HookBodyTooLarge>()("HookBodyTooLarge", {}) {}
99class ResponseTooLarge extends Schema.TaggedError<ResponseTooLarge>()("ResponseTooLarge", {}) {}
100
101const errorResponse = (status: number, error: string, headers?: Record<string, string>) =>
102 HttpServerResponse.jsonUnsafe({ error }, { status, ...(headers ? { headers } : {}) });
103
104const hookNotFound = () => errorResponse(404, "hook_not_found");
105
106function parseHookPath(url: string) {
107 const queryIndex = url.indexOf("?");
108 const path = queryIndex === -1 ? url : url.slice(0, queryIndex);
109 const search = queryIndex === -1 ? "" : url.slice(queryIndex);
110 const segments = path.split("/");
111 // ["", "v1", "hooks", environmentId, hookId, token]
112 if (segments.length !== 6) return null;
113 const [, , , rawEnvironmentId, rawHookId, rawToken] = segments;
114 if (!rawEnvironmentId || !rawHookId || !rawToken) return null;
115 try {
116 return {
117 environmentId: decodeURIComponent(rawEnvironmentId),
118 hookId: decodeURIComponent(rawHookId),
119 token: decodeURIComponent(rawToken),
120 // Forward the encoded segments byte-for-byte; the environment decodes them.
121 rawHookId,
122 rawToken,
123 search,
124 };
125 } catch {
126 return null;
127 }
128}
129
130function forwardedHeaders(headers: Readonly<Record<string, string>>): Record<string, string> {
131 const result: Record<string, string> = {};
132 // Headers the sender names in Connection are hop-by-hop too (RFC 9110 7.6.1).
133 const connectionValue =
134 Object.entries(headers).find(([name]) => name.toLowerCase() === "connection")?.[1] ?? "";
135 const nominated = new Set(
136 connectionValue
137 .split(",")
138 .map((name) => name.trim().toLowerCase())
139 .filter(Boolean),
140 );
141 for (const name in headers) {
142 const lower = name.toLowerCase();
143 if (
144 DROPPED_REQUEST_HEADERS.has(lower) ||
145 nominated.has(lower) ||
146 DROPPED_REQUEST_HEADER_PREFIXES.some((prefix) => lower.startsWith(prefix))
147 ) {
148 continue;
149 }
150 const value = headers[name];
151 if (value !== undefined) result[lower] = value;
152 }
153 return result;
154}
155
156const hasNoBody = (request: HttpServerRequest.HttpServerRequest) =>
157 request.source instanceof Request && request.source.body === null;
158
159/** Collects a byte stream, failing with `tooLarge` once it passes `maxBytes` rather than buffering it all. */
160const collectCapped = <E, R, E2>(
161 stream: Stream.Stream<Uint8Array, E, R>,
162 maxBytes: number,
163 tooLarge: () => E2,
164) =>
165 Effect.suspend(() => {
166 const chunks: Array<Uint8Array> = [];
167 let total = 0;
168 return stream.pipe(
169 Stream.runForEach((chunk) => {
170 total += chunk.length;
171 if (total > maxBytes) {
172 return Effect.fail(tooLarge());
173 }
174 chunks.push(chunk);
175 return Effect.void;
176 }),
177 Effect.map(() => {
178 const body = new Uint8Array(total);
179 let offset = 0;
180 for (const chunk of chunks) {
181 body.set(chunk, offset);
182 offset += chunk.length;
183 }
184 return body;
185 }),
186 );
187 });
188
189const readCappedBody = (request: HttpServerRequest.HttpServerRequest) =>
190 hasNoBody(request)
191 ? Effect.succeed(new Uint8Array(0))
192 : collectCapped(request.stream, RELAY_HOOK_MAX_BODY_BYTES, () => new HookBodyTooLarge());
193
194const readCappedResponse = (response: HttpClientResponse.HttpClientResponse) =>
195 collectCapped(response.stream, MAX_RESPONSE_BYTES, () => new ResponseTooLarge()).pipe(
196 // A response without a body, such as a redirect, has no stream at all.
197 Effect.catchIf(
198 (error) => error._tag === "HttpClientError" && error.reason._tag === "EmptyBodyError",
199 () => Effect.succeed(new Uint8Array(0)),
200 ),
201 );
202
203const make = Effect.gen(function* () {
204 const links = yield* EnvironmentLinks.EnvironmentLinks;
205 const allocations = yield* ManagedEndpointAllocations.ManagedEndpointAllocations;
206 const settings = yield* RelayConfiguration.RelayConfiguration;
207 const httpClient = yield* HttpClient.HttpClient;
208 const rateLimiter = yield* HookRateLimiter;
209
210 const resolveEndpoint = Effect.fn("relay.hooks.resolve_endpoint")(function* (
211 environmentId: string,
212 ) {
213 const candidates = yield* links.findActiveManagedForEnvironment({ environmentId });
214 for (const link of candidates) {
215 const allocation = yield* allocations.get({ userId: link.userId, environmentId });
216 const result = validateManagedEndpoint({
217 link,
218 allocation,
219 baseDomain: settings.managedEndpointBaseDomain,
220 });
221 if (Result.isSuccess(result)) {
222 return result.success;
223 }
224 }
225 return null;
226 });
227
228 const handle = Effect.fn("relay.hooks.forward")(function* (
229 request: HttpServerRequest.HttpServerRequest,
230 ) {
231 const outcome = (value: string) => Effect.annotateCurrentSpan({ "relay.hook.outcome": value });
232 const parsed = parseHookPath(request.url);
233 if (!parsed) {
234 yield* outcome("invalid_path");
235 return hookNotFound();
236 }
237 yield* Effect.annotateCurrentSpan({
238 "relay.environment_id": parsed.environmentId,
239 "relay.hook_id": parsed.hookId,
240 });
241 if (!(yield* rateLimiter.allow(yield* hookBudgetKey(parsed)))) {
242 yield* outcome("rate_limited");
243 return errorResponse(429, "rate_limited", {
244 "retry-after": String(RELAY_HOOK_RATE_LIMIT.periodSeconds),
245 });
246 }
247 const declaredLength = Number(request.headers["content-length"] ?? "0");
248 if (Number.isFinite(declaredLength) && declaredLength > RELAY_HOOK_MAX_BODY_BYTES) {
249 yield* outcome("payload_too_large");
250 return errorResponse(413, "payload_too_large");
251 }
252
253 const endpoint = yield* resolveEndpoint(parsed.environmentId).pipe(
254 Effect.catch((error) =>
255 Effect.logWarning("Failed to resolve hook endpoint", {
256 environmentId: parsed.environmentId,
257 errorTag: error._tag,
258 }).pipe(Effect.as(null)),
259 ),
260 );
261 if (!endpoint) {
262 yield* outcome("not_found");
263 return hookNotFound();
264 }
265
266 const body =
267 request.method === "GET"
268 ? Result.succeed(new Uint8Array(0))
269 : yield* readCappedBody(request).pipe(Effect.result);
270 if (Result.isFailure(body)) {
271 if (body.failure._tag === "HookBodyTooLarge") {
272 yield* outcome("payload_too_large");
273 return errorResponse(413, "payload_too_large");
274 }
275 yield* outcome("invalid_body");
276 return errorResponse(400, "invalid_body");
277 }
278
279 const baseUrl = endpoint.httpBaseUrl.endsWith("/")
280 ? endpoint.httpBaseUrl
281 : `${endpoint.httpBaseUrl}/`;
282 const headers = forwardedHeaders(request.headers);
283 let upstreamRequest = HttpClientRequest.make(
284 request.method as "GET" | "POST" | "PUT" | "PATCH",
285 )(`${baseUrl}api/hooks/${parsed.rawHookId}/${parsed.rawToken}${parsed.search}`, { headers });
286 if (request.method !== "GET") {
287 upstreamRequest = HttpClientRequest.bodyUint8Array(
288 upstreamRequest,
289 body.success,
290 headers["content-type"],
291 );
292 }
293
294 const upstream = yield* httpClient.execute(upstreamRequest).pipe(
295 Effect.flatMap((response) =>
296 readCappedResponse(response).pipe(
297 Effect.map((body) => ({
298 status: response.status,
299 contentType: response.headers["content-type"],
300 body,
301 })),
302 ),
303 ),
304 withoutRedirects,
305 // The client span would record url.full, which carries the token.
306 Effect.provideService(HttpClient.TracerDisabledWhen, () => true),
307 Effect.timeoutOption(Duration.millis(RELAY_HOOK_UPSTREAM_TIMEOUT_MS)),
308 Effect.result,
309 );
310 if (Result.isFailure(upstream)) {
311 yield* outcome("environment_unavailable");
312 return errorResponse(503, "environment_unavailable");
313 }
314 if (Option.isNone(upstream.success)) {
315 yield* outcome("environment_timeout");
316 return errorResponse(504, "environment_timeout");
317 }
318 const response = upstream.success.value;
319 if (response.status === TUNNEL_OFFLINE_STATUS) {
320 yield* outcome("environment_unavailable");
321 return errorResponse(503, "environment_unavailable");
322 }
323 yield* Effect.annotateCurrentSpan({
324 "relay.hook.outcome": "forwarded",
325 "relay.hook.upstream_status": response.status,
326 });
327 // Only content-type is passed through: no location (redirects are never
328 // followed or relayed), no cookies, no upstream infrastructure headers.
329 const contentTypeHeaders = response.contentType
330 ? { "content-type": response.contentType }
331 : undefined;
332 if (response.body.length === 0) {
333 return HttpServerResponse.empty({
334 status: response.status,
335 ...(contentTypeHeaders ? { headers: contentTypeHeaders } : {}),
336 });
337 }
338 return HttpServerResponse.uint8Array(response.body, {
339 status: response.status,
340 ...(response.contentType ? { contentType: response.contentType } : {}),
341 });
342 });
343
344 return HookForwarder.of({ handle });
345});
346
347export const layer = Layer.effect(HookForwarder, make);
348
349/**
350 * Implements the RelayApi `hooks` group. The endpoints are raw: the forwarder
351 * re-reads the encoded path segments from the request so the token and hook
352 * id reach the environment byte for byte, and streams the body itself.
353 */
354export const hooksApi = HttpApiBuilder.group(
355 RelayApi,
356 "hooks",
357 Effect.fnUntraced(function* (handlers) {
358 const forwarder = yield* HookForwarder;
359 const forward = ({ request }: { readonly request: HttpServerRequest.HttpServerRequest }) =>
360 forwarder.handle(request);
361 return handlers
362 .handleRaw("forwardPost", forward)
363 .handleRaw("forwardPut", forward)
364 .handleRaw("forwardPatch", forward)
365 .handleRaw("forwardGet", forward);
366 }),
367);
infra/relay/src/http/Api.test.ts+1
......@@ -332,6 +332,7 @@ function relayUnlinkTestLayer(input?: {
332332 listDeliveryUsersForEnvironment: () => Effect.die("unused listDeliveryUsersForEnvironment"),
333333 listForUser: () => Effect.die("unused listForUser"),
334334 getForUser: input?.getForUser ?? (() => Effect.succeed(null)),
335 findActiveManagedForEnvironment: () => Effect.succeed([]),
335336 revokeForUser: input?.revokeForUser ?? (() => Effect.succeed(false)),
336337 }),
337338 ),
infra/relay/src/http/Api.ts+51-5
......@@ -11,6 +11,7 @@ import * as Record from "effect/Record";
1111import * as Redacted from "effect/Redacted";
1212import * as Schema from "effect/Schema";
1313import * as Tracer from "effect/Tracer";
14import * as Headers from "effect/http/Headers";
1415import * as HttpEffect from "effect/http/HttpEffect";
1516import * as HttpMiddleware from "effect/http/HttpMiddleware";
1617import * as HttpRouter from "effect/http/HttpRouter";
......@@ -75,6 +76,7 @@ import * as ManagedEndpointAllocations from "../environments/ManagedEndpointAllo
7576import * as EnvironmentPublishSignatures from "../environments/EnvironmentPublishSignatures.ts";
7677import * as MobileRegistrations from "../agentActivity/MobileRegistrations.ts";
7778import { withSpanAttributes } from "../observability.ts";
79import { isRelayHookPath, redactRelayHookUrl } from "../hooks/HookForwarder.ts";
7880import * as RelayDb from "../db.ts";
7981
8082// Delegated thread IDs carry escaped command provenance and exceed the router's
......@@ -150,6 +152,10 @@ export const relayCors = HttpRouter.middleware(
150152 >,
151153 ) {
152154 const request = yield* HttpServerRequest.HttpServerRequest;
155 // Public webhook forwarding is server-to-server: no preflight, no CORS grants.
156 if (isRelayHookPath(request.url)) {
157 return yield* httpEffect;
158 }
153159 if (request.method === "OPTIONS") {
154160 return HttpServerResponse.empty({
155161 status: 204,
......@@ -222,10 +228,45 @@ export const traceRelayHttpRequest = <E, R>(
222228 HttpServerRequest.HttpServerRequest | R
223229 >,
224230) =>
225 // HttpMiddleware finalizes its span on the dispatcher; do not close a request-scoped exporter first.
226 HttpMiddleware.tracer(
227 appendRelayTraceContextResponseHeader.pipe(Effect.andThen(relayRequestDeadline(httpEffect))),
228 ).pipe(Effect.ensuring(Effect.yieldNow));
231 Effect.gen(function* () {
232 const request = yield* HttpServerRequest.HttpServerRequest;
233 const traced = appendRelayTraceContextResponseHeader.pipe(
234 Effect.andThen(relayRequestDeadline(httpEffect)),
235 );
236 if (!isRelayHookPath(request.url)) {
237 // HttpMiddleware finalizes its span on the dispatcher; do not close a request-scoped exporter first.
238 return yield* HttpMiddleware.tracer(traced).pipe(Effect.ensuring(Effect.yieldNow));
239 }
240 // Hook URLs carry a secret token: the tracer and deadline log see a redacted
241 // request, while the route itself still receives the original.
242 const redacted = request.modify({ url: redactRelayHookUrl(request.url) });
243 return yield* HttpMiddleware.tracer(
244 appendRelayTraceContextResponseHeader.pipe(
245 Effect.andThen(
246 relayRequestDeadline(
247 httpEffect.pipe(Effect.provideService(HttpServerRequest.HttpServerRequest, request)),
248 ),
249 ),
250 ),
251 ).pipe(
252 Effect.provideService(HttpServerRequest.HttpServerRequest, redacted),
253 Effect.ensuring(Effect.yieldNow),
254 );
255 });
256
257// Webhook senders put shared secrets and signatures in headers such as
258// x-hub-signature-256, stripe-signature, x-gitlab-token and x-webhook-key.
259const webhookHeaderRedactionLayer = Layer.effect(
260 Headers.CurrentRedactedNames,
261 Effect.map(Headers.CurrentRedactedNames, (names) => [
262 ...names,
263 /signature/i,
264 /token/i,
265 /secret/i,
266 /key/i,
267 /auth/i,
268 ]),
269);
229270
230271export const traceRelayHttpRequestWith = <E, R, LayerError, LayerRequirements>(
231272 httpEffect: Effect.Effect<
......@@ -236,7 +277,12 @@ export const traceRelayHttpRequestWith = <E, R, LayerError, LayerRequirements>(
236277 tracerLayer: Layer.Layer<never, LayerError, LayerRequirements>,
237278) =>
238279 traceRelayHttpRequest(httpEffect).pipe(
239 Effect.provide(Layer.merge(tracerLayer, httpHeaderRedactionLayer)),
280 Effect.provide(
281 Layer.merge(
282 tracerLayer,
283 webhookHeaderRedactionLayer.pipe(Layer.provide(httpHeaderRedactionLayer)),
284 ),
285 ),
240286 );
241287
242288export const withoutCapturedParentSpan = <A, E, R>(
infra/relay/src/worker.ts+30-1
......@@ -73,6 +73,7 @@ import * as ManagedEndpointProvider from "./environments/ManagedEndpointProvider
7373import * as ManagedEndpointReaper from "./environments/ManagedEndpointReaper.ts";
7474import * as ManagedTunnelLimits from "./environments/ManagedTunnelLimits.ts";
7575import * as MobileRegistrations from "./agentActivity/MobileRegistrations.ts";
76import * as HookForwarder from "./hooks/HookForwarder.ts";
7677
7778const webcryptoLayer = Layer.succeed(
7879 Crypto.Crypto,
......@@ -189,6 +190,13 @@ export const ApiLive = Api.make(
189190 const managedEndpointDnsBinding = yield* Cloudflare.DNS.ReadWriteDns(managedEndpointZone);
190191 const managedEndpointZoneName = yield* managedEndpointZone.name;
191192 const managedEndpointCleanupMode = yield* RelayConfiguration.managedEndpointCleanupModeConfig;
193 const hookRateLimit = yield* Cloudflare.RateLimit("HOOK_RATE_LIMIT", {
194 namespaceId: 1001,
195 simple: {
196 limit: HookForwarder.RELAY_HOOK_RATE_LIMIT.limit,
197 period: HookForwarder.RELAY_HOOK_RATE_LIMIT.periodSeconds,
198 },
199 });
192200
193201 //
194202 // 3. Runtime layers and app construction
......@@ -282,7 +290,27 @@ export const ApiLive = Api.make(
282290 Layer.provideMerge(webcryptoLayer),
283291 );
284292
285 const appLayer = relayApiLayer.pipe(
293 // Fails open: a limiter outage must not drop webhooks the environment would accept.
294 const hookRateLimiterLayer = Layer.succeed(HookForwarder.HookRateLimiter, {
295 allow: (key) =>
296 hookRateLimit.limit({ key }).pipe(
297 Effect.map((result) => result.success),
298 Effect.provideService(Alchemy.RuntimeContext, alchemyRuntimeContext),
299 Effect.catch((error) =>
300 Effect.logWarning("Hook rate limiter unavailable", { error: error.message }).pipe(
301 Effect.as(true),
302 ),
303 ),
304 ),
305 });
306
307 const appLayer = Layer.merge(
308 relayApiLayer,
309 HookForwarder.hooksApi.pipe(
310 Layer.provide(HookForwarder.layer),
311 Layer.provide(hookRateLimiterLayer),
312 ),
313 ).pipe(
286314 Layer.provideMerge(relayClientAuthLayer),
287315 Layer.provideMerge(relayDpopClientAuthLayer),
288316 Layer.provideMerge(relayEnvironmentAuthLayer),
......@@ -397,6 +425,7 @@ export const ApiLive = Api.make(
397425 Layer.provideMerge(Cloudflare.Queues.EventSourceLive),
398426 Layer.provideMerge(Cloudflare.Tunnel.ReadWriteTunnelBinding),
399427 Layer.provideMerge(Cloudflare.DNS.ReadWriteDnsHttp),
428 Layer.provideMerge(Cloudflare.Workers.RateLimitBinding),
400429 ),
401430 ),
402431 ),
packages/contracts/src/relay.ts+23
......@@ -1175,6 +1175,28 @@ const RelayServerGroup = HttpApiGroup.make("server")
11751175 .annotate(OpenApi.Description, "Environment-authenticated activity publication.")
11761176 .middleware(RelayEnvironmentAuth);
11771177
1178/**
1179 * Public, stateless webhook forwarding to an environment's managed tunnel.
1180 * Unauthenticated: the token in the path is the environment's credential, and
1181 * the relay only routes and forwards. Raw, so the body reaches the
1182 * environment byte for byte and signatures still verify there.
1183 */
1184const RelayHookParams = Schema.Struct({
1185 environmentId: Schema.String,
1186 hookId: Schema.String,
1187 token: Schema.String,
1188});
1189const RELAY_HOOK_PATH = "/v1/hooks/:environmentId/:hookId/:token";
1190const relayHookEndpoint = { params: RelayHookParams } as const;
1191const RelayHooksGroup = HttpApiGroup.make("hooks")
1192 .add(
1193 HttpApiEndpoint.post("forwardPost", RELAY_HOOK_PATH, relayHookEndpoint),
1194 HttpApiEndpoint.put("forwardPut", RELAY_HOOK_PATH, relayHookEndpoint),
1195 HttpApiEndpoint.patch("forwardPatch", RELAY_HOOK_PATH, relayHookEndpoint),
1196 HttpApiEndpoint.get("forwardGet", RELAY_HOOK_PATH, relayHookEndpoint),
1197 )
1198 .annotate(OpenApi.Description, "Forward webhook requests to an environment.");
1199
11781200export const RelayApi = HttpApi.make("RelayApi")
11791201 .add(
11801202 RelayHealthGroup,
......@@ -1184,6 +1206,7 @@ export const RelayApi = HttpApi.make("RelayApi")
11841206 RelayTokenGroup,
11851207 RelayDpopClientGroup,
11861208 RelayServerGroup,
1209 RelayHooksGroup,
11871210 )
11881211 .annotate(OpenApi.Title, "T3 Code Relay API")
11891212 .annotate(OpenApi.Version, "1.0.0")