Open History to see the last commit
1import * as Clock from "effect/Clock";
2import * as Context from "effect/Context";
3import * as SecureRandom from "effect/crypto/SecureRandom";
4import * as DateTime from "effect/DateTime";
5import * as Effect from "effect/Effect";
6import * as Layer from "effect/Layer";
7import * as Option from "effect/Option";
8import * as Result from "effect/Result";
9import * as Schema from "effect/Schema";
10import * as Stream from "effect/Stream";
11import * as HttpClient from "effect/http/HttpClient";
12import type * as HttpServerRequest from "effect/http/HttpServerRequest";
13import * as HttpServerResponse from "effect/http/HttpServerResponse";
14import * as HttpApiBuilder from "effect/http-api/HttpApiBuilder";
15import { EnvironmentId } from "@t3tools/contracts";
16import { RelayApi, type RelayHookDeliveryProofPayload } from "@t3tools/contracts/relay";
17import {
18 normalizeRelayIssuer,
19 RELAY_HOOK_DELIVERY_HEADER,
20 RELAY_HOOK_DELIVERY_TYP,
21} from "@t3tools/shared/relayJwt";
22
23import * as RelayConfiguration from "../Config.ts";
24import * as CloudMintSigner from "../auth/CloudMintSigner.ts";
25import { MANAGED_ENDPOINT_KEY_PATTERN } from "../deploymentConfig.ts";
26import * as HeldHooks from "./HeldHooks.ts";
27import * as HookInbox from "./HookInbox.ts";
28import { sendUpstream, TUNNEL_OFFLINE_STATUS } from "./upstream.ts";
29
30const RELAY_HOOK_PATH_PREFIX = "/v1/hooks/";
31export const RELAY_HOOK_MAX_BODY_BYTES = 1_048_576;
32export const RELAY_HOOK_RATE_LIMIT = { limit: 60, periodSeconds: 60 } as const;
33/** URL budgets alone are bypassable by minting URLs; cap each endpoint too. */
34export const RELAY_HOOK_ENDPOINT_RATE_LIMIT = { limit: 600, periodSeconds: 60 } as const;
35/** No tunnel origin (530) or local server unavailable (502/503/504) means the environment did not take the request. */
36const ENVIRONMENT_UNREACHABLE_STATUSES: ReadonlySet<number> = new Set([
37 502,
38 503,
39 504,
40 TUNNEL_OFFLINE_STATUS,
41]);
42
43const DROPPED_REQUEST_HEADERS = new Set([
44 "host",
45 "connection",
46 "keep-alive",
47 "transfer-encoding",
48 "te",
49 "upgrade",
50 "content-length",
51 "cookie",
52 "x-real-ip",
53 // Only the relay may set this; a sender could otherwise collide delivery ids.
54 "x-t3-relay-delivery-id",
55 "x-t3-relay-received-at",
56 RELAY_HOOK_DELIVERY_HEADER,
57 // The environment trusts trace context only from the relay, which sets its own.
58 "traceparent",
59 "tracestate",
60 "b3",
61]);
62const DROPPED_REQUEST_HEADER_PREFIXES = ["proxy-", "cf-", "x-forwarded-", "x-b3-"];
63
64export const isRelayHookPath = (url: string): boolean => url.startsWith(RELAY_HOOK_PATH_PREFIX);
65
66export const redactRelayHookUrl = (url: string): string => {
67 const path = url.split("?", 1)[0] ?? url;
68 const segments = path.split("/");
69 if (segments.length >= 6) {
70 segments[5] = "redacted";
71 }
72 return segments.join("/");
73};
74
75/** Budget decoded tokens so equivalent spellings share limits and wrong tokens cannot exhaust a real sender's budget. */
76const hookBudgetKey = (hook: {
77 readonly endpointKey: string;
78 readonly hookId: string;
79 readonly token: string;
80}) =>
81 Effect.promise(() =>
82 crypto.subtle.digest(
83 "SHA-256",
84 // Length-prefixed, so no segment contents can make two keys collide.
85 new TextEncoder().encode(
86 [hook.endpointKey, hook.hookId, hook.token]
87 .map((part) => `${part.length}:${part}`)
88 .join(""),
89 ),
90 ),
91 ).pipe(
92 Effect.map((digest) =>
93 Array.from(new Uint8Array(digest), (byte) => byte.toString(16).padStart(2, "0")).join(""),
94 ),
95 );
96
97export class HookRateLimiter extends Context.Service<
98 HookRateLimiter,
99 {
100 readonly allowHook: (key: string) => Effect.Effect<boolean>;
101 readonly allowEndpoint: (endpointKey: string) => Effect.Effect<boolean>;
102 }
103>()("t3code-relay/hooks/HookForwarder/HookRateLimiter") {}
104
105export class HookForwarder extends Context.Service<
106 HookForwarder,
107 {
108 readonly handle: (
109 request: HttpServerRequest.HttpServerRequest,
110 ) => Effect.Effect<HttpServerResponse.HttpServerResponse>;
111 }
112>()("t3code-relay/hooks/HookForwarder") {}
113
114class HookBodyTooLarge extends Schema.TaggedError<HookBodyTooLarge>()("HookBodyTooLarge", {}) {}
115
116const errorResponse = (status: number, error: string, headers?: Record<string, string>) =>
117 HttpServerResponse.jsonUnsafe({ error }, { status, ...(headers ? { headers } : {}) });
118
119const hookNotFound = () => errorResponse(404, "hook_not_found");
120
121/** Longer than the inbox holds a request, so a held delivery's proof still verifies. */
122const DELIVERY_PROOF_LIFETIME_SECONDS = 25 * 60 * 60;
123
124// HEAD reaches GET routing but must be refused.
125const FORWARDED_METHODS = new Set(["GET", "POST", "PUT", "PATCH"]);
126
127// Environment responses use the relay origin; never let their bodies render or run there.
128const SANDBOXED_RESPONSE_HEADERS = {
129 "x-content-type-options": "nosniff",
130 "content-security-policy": "sandbox; default-src 'none'",
131} as const;
132
133function parseHookPath(url: string) {
134 const queryIndex = url.indexOf("?");
135 const path = queryIndex === -1 ? url : url.slice(0, queryIndex);
136 const search = queryIndex === -1 ? "" : url.slice(queryIndex);
137 const segments = path.split("/");
138 if (segments.length !== 6) return null;
139 const [, , , endpointKey, rawHookId, rawToken] = segments;
140 if (!endpointKey || !MANAGED_ENDPOINT_KEY_PATTERN.test(endpointKey) || !rawHookId || !rawToken) {
141 return null;
142 }
143 try {
144 return {
145 endpointKey,
146 hookId: decodeURIComponent(rawHookId),
147 token: decodeURIComponent(rawToken),
148 // Forward the encoded segments byte-for-byte; the environment decodes them.
149 rawHookId,
150 rawToken,
151 search,
152 };
153 } catch {
154 return null;
155 }
156}
157
158function forwardedHeaders(headers: Readonly<Record<string, string>>): Record<string, string> {
159 const result: Record<string, string> = {};
160 // Headers the sender names in Connection are hop-by-hop too (RFC 9110 7.6.1).
161 const connectionValue =
162 Object.entries(headers).find(([name]) => name.toLowerCase() === "connection")?.[1] ?? "";
163 const nominated = new Set(
164 connectionValue
165 .split(",")
166 .map((name) => name.trim().toLowerCase())
167 .filter(Boolean),
168 );
169 for (const name in headers) {
170 const lower = name.toLowerCase();
171 if (
172 DROPPED_REQUEST_HEADERS.has(lower) ||
173 nominated.has(lower) ||
174 DROPPED_REQUEST_HEADER_PREFIXES.some((prefix) => lower.startsWith(prefix))
175 ) {
176 continue;
177 }
178 const value = headers[name];
179 if (value !== undefined) result[lower] = value;
180 }
181 return result;
182}
183
184const hasNoBody = (request: HttpServerRequest.HttpServerRequest) =>
185 request.source instanceof Request && request.source.body === null;
186
187const readCappedBody = (request: HttpServerRequest.HttpServerRequest) =>
188 Effect.suspend(() => {
189 if (hasNoBody(request)) {
190 return Effect.succeed(new Uint8Array(0));
191 }
192 const chunks: Array<Uint8Array> = [];
193 let total = 0;
194 return request.stream.pipe(
195 Stream.runForEach((chunk) => {
196 total += chunk.length;
197 if (total > RELAY_HOOK_MAX_BODY_BYTES) {
198 return Effect.fail(new HookBodyTooLarge());
199 }
200 chunks.push(chunk);
201 return Effect.void;
202 }),
203 Effect.map(() => {
204 const body = new Uint8Array(total);
205 let offset = 0;
206 for (const chunk of chunks) {
207 body.set(chunk, offset);
208 offset += chunk.length;
209 }
210 return body;
211 }),
212 );
213 });
214
215const make = Effect.gen(function* () {
216 const heldHooks = yield* HeldHooks.HeldHooks;
217 const settings = yield* RelayConfiguration.RelayConfiguration;
218 const httpClient = yield* HttpClient.HttpClient;
219 const rateLimiter = yield* HookRateLimiter;
220 const inbox = yield* HookInbox.HookInbox;
221 const random = yield* SecureRandom.SecureRandom;
222 const mint = yield* CloudMintSigner.CloudMintSigner;
223
224 const signDeliveryProof = Effect.fn("HookForwarder.signDeliveryProof")(
225 function* (input: {
226 readonly environmentId: string;
227 readonly deliveryId: string;
228 readonly receivedAt: string;
229 readonly hookId: string;
230 readonly jti: string;
231 }) {
232 const now = Math.floor((yield* Clock.currentTimeMillis) / 1_000);
233 return yield* mint.sign({
234 typ: RELAY_HOOK_DELIVERY_TYP,
235 payload: {
236 iss: normalizeRelayIssuer(settings.relayIssuer),
237 aud: `t3-env:${input.environmentId}`,
238 sub: input.environmentId,
239 jti: input.jti,
240 iat: now,
241 exp: now + DELIVERY_PROOF_LIFETIME_SECONDS,
242 environmentId: EnvironmentId.make(input.environmentId),
243 deliveryId: input.deliveryId,
244 receivedAt: input.receivedAt,
245 hookId: input.hookId,
246 } satisfies RelayHookDeliveryProofPayload,
247 });
248 },
249 (effect, _input) => effect.pipe(Effect.orDie),
250 );
251
252 const handle = Effect.fn("relay.hooks.forward")(function* (
253 request: HttpServerRequest.HttpServerRequest,
254 ) {
255 const outcome = (value: string) => Effect.annotateCurrentSpan({ "relay.hook.outcome": value });
256 if (!FORWARDED_METHODS.has(request.method)) {
257 yield* outcome("method_not_allowed");
258 return errorResponse(405, "method_not_allowed", { allow: "GET, POST, PUT, PATCH" });
259 }
260 const receivedAt = DateTime.formatIso(yield* DateTime.now);
261 const parsed = parseHookPath(request.url);
262 if (!parsed) {
263 yield* outcome("invalid_path");
264 return hookNotFound();
265 }
266 yield* Effect.annotateCurrentSpan({
267 "relay.hook.endpoint_key": parsed.endpointKey,
268 "relay.hook_id": parsed.hookId,
269 });
270 const endpointAllowed = yield* rateLimiter.allowEndpoint(parsed.endpointKey);
271 const hookAllowed =
272 endpointAllowed && (yield* rateLimiter.allowHook(yield* hookBudgetKey(parsed)));
273 if (!endpointAllowed || !hookAllowed) {
274 yield* Effect.annotateCurrentSpan({
275 "relay.hook.rate_limit": endpointAllowed ? "hook" : "endpoint",
276 });
277 yield* outcome("rate_limited");
278 return errorResponse(429, "rate_limited", {
279 "retry-after": String(RELAY_HOOK_RATE_LIMIT.periodSeconds),
280 });
281 }
282 const declaredLength = Number(request.headers["content-length"] ?? "0");
283 if (Number.isFinite(declaredLength) && declaredLength > RELAY_HOOK_MAX_BODY_BYTES) {
284 yield* outcome("payload_too_large");
285 return errorResponse(413, "payload_too_large");
286 }
287
288 const endpoint = yield* heldHooks.resolveEndpoint(parsed.endpointKey).pipe(
289 Effect.catch((error) =>
290 Effect.logWarning("Failed to resolve hook endpoint", {
291 endpointKey: parsed.endpointKey,
292 errorTag: error._tag,
293 }).pipe(Effect.as(null)),
294 ),
295 );
296 if (!endpoint) {
297 yield* outcome("not_found");
298 return hookNotFound();
299 }
300 yield* Effect.annotateCurrentSpan({
301 "relay.environment_id": endpoint.environmentId,
302 "relay.hook.hold_while_offline": endpoint.holdWhileOffline,
303 });
304
305 const body =
306 request.method === "GET"
307 ? Result.succeed(new Uint8Array(0))
308 : yield* readCappedBody(request).pipe(Effect.result);
309 if (Result.isFailure(body)) {
310 if (body.failure._tag === "HookBodyTooLarge") {
311 yield* outcome("payload_too_large");
312 return errorResponse(413, "payload_too_large");
313 }
314 yield* outcome("invalid_body");
315 return errorResponse(400, "invalid_body");
316 }
317
318 // Reuse one delivery ID so inbox retries cannot rerun requests that reached the environment before timeout.
319 yield* Effect.annotateCurrentSpan({ "relay.hook.body_bytes": body.success.byteLength });
320 const deliveryId = yield* random.uuidV4();
321 // Sign once before storing held requests so the inbox needs no signing key.
322 const proof = yield* signDeliveryProof({
323 environmentId: endpoint.environmentId,
324 deliveryId,
325 receivedAt,
326 hookId: parsed.hookId,
327 jti: yield* random.uuidV4(),
328 });
329 const hook = {
330 id: deliveryId,
331 receivedAt,
332 method: request.method,
333 rawHookId: parsed.rawHookId,
334 rawToken: parsed.rawToken,
335 hookKey: parsed.hookId,
336 query: parsed.search.replace(/^\?/, ""),
337 headers: { ...forwardedHeaders(request.headers), [RELAY_HOOK_DELIVERY_HEADER]: proof },
338 body: body.success,
339 };
340 // Keep offline outcomes on the enclosing relay.hooks.forward span.
341 const holdOrFail = Effect.fnUntraced(function* (status: 503 | 504, error: string) {
342 if (!endpoint.holdWhileOffline) {
343 yield* outcome(error);
344 return errorResponse(status, error);
345 }
346 const stored = yield* inbox
347 .hold({
348 endpointKey: parsed.endpointKey,
349 baseUrl: endpoint.httpBaseUrl,
350 hook,
351 })
352 .pipe(
353 Effect.catch((cause) =>
354 Effect.logWarning("Could not hold webhook request", {
355 environmentId: endpoint.environmentId,
356 errorTag: cause._tag,
357 }).pipe(Effect.as(null)),
358 ),
359 );
360 if (stored === null) {
361 yield* outcome(error);
362 return errorResponse(status, error);
363 }
364 if (!stored) {
365 yield* outcome("inbox_full");
366 return errorResponse(503, "inbox_full");
367 }
368 yield* outcome("held");
369 return HttpServerResponse.jsonUnsafe({ queued: true }, { status: 202 });
370 });
371
372 const upstream = yield* sendUpstream(endpoint.httpBaseUrl, hook).pipe(
373 Effect.provideService(HttpClient.HttpClient, httpClient),
374 Effect.result,
375 );
376 if (Result.isFailure(upstream)) {
377 yield* Effect.annotateCurrentSpan({ "relay.hook.upstream_error": upstream.failure._tag });
378 return yield* holdOrFail(503, "environment_unavailable");
379 }
380 if (Option.isNone(upstream.success)) {
381 return yield* holdOrFail(504, "environment_timeout");
382 }
383 const response = upstream.success.value;
384 if (ENVIRONMENT_UNREACHABLE_STATUSES.has(response.status)) {
385 yield* Effect.annotateCurrentSpan({ "relay.hook.upstream_status": response.status });
386 return yield* holdOrFail(503, "environment_unavailable");
387 }
388 yield* Effect.annotateCurrentSpan({
389 "relay.hook.outcome": "forwarded",
390 "relay.hook.upstream_status": response.status,
391 ...(response.outcome === undefined
392 ? {}
393 : { "relay.hook.upstream_outcome": response.outcome }),
394 });
395 const headers = {
396 ...SANDBOXED_RESPONSE_HEADERS,
397 ...(response.contentType ? { "content-type": response.contentType } : {}),
398 };
399 if (response.body.length === 0) {
400 return HttpServerResponse.empty({ status: response.status, headers });
401 }
402 return HttpServerResponse.uint8Array(response.body, {
403 status: response.status,
404 headers,
405 ...(response.contentType ? { contentType: response.contentType } : {}),
406 });
407 });
408
409 return HookForwarder.of({ handle });
410});
411
412export const layer = Layer.effect(HookForwarder, make);
413
414export const layerApi = HttpApiBuilder.group(
415 RelayApi,
416 "hooks",
417 Effect.fnUntraced(function* (handlers) {
418 const forwarder = yield* HookForwarder;
419 const forward = ({ request }: { readonly request: HttpServerRequest.HttpServerRequest }) =>
420 forwarder.handle(request);
421 return handlers
422 .handleRaw("forwardPost", forward)
423 .handleRaw("forwardPut", forward)
424 .handleRaw("forwardPatch", forward)
425 .handleRaw("forwardGet", forward);
426 }),
427);