1import { requestUrl } from "../requestUrl.ts";
2// Loopback-only hubs expose shell execution or unauthenticated actions.
3// Proxy with environment read scope for streams and operate scope for input and tuning.
4import { AuthOrchestrationReadScope, AuthOrchestrationOperateScope } from "@t3tools/contracts";
5import * as Effect from "effect/Effect";
6import * as Layer from "effect/Layer";
7import * as Option from "effect/Option";
8import {
9 HttpClient,
10 HttpClientRequest,
11 HttpMethod,
12 HttpRouter,
13 HttpServerRequest,
14 HttpServerResponse,
15} from "effect/http";
16import * as Socket from "effect/socket/Socket";
17import * as NodeSocket from "@effect/platform-node/NodeSocket";
18
19import { authenticateMediaRequest } from "../auth/http.ts";
20import * as DeviceService from "./DeviceService.ts";
21import { repeatStillFrame } from "./mjpegStillFrame.ts";
22
23const ALLOWED_PATHS: ReadonlyArray<RegExp> = [
24 /^\/api\/devices$/,
25 /^\/vendor\/serve-sim\/api$/,
26 /^\/vendor\/serve-sim\/api\/screenshot$/,
27 /^\/vendor\/serve-sim\/api\/event-log(\/events)?$/,
28 /^\/vendor\/serve-sim\/helper\/[^/]+\/(stream\.mjpeg|stream\.avcc|config|health|ax|foreground)$/,
29 /^\/vendor\/serve-sim\/helper\/[^/]+\/panel\/(1|3)\/stream\.avcc$/,
30 /^\/vendor\/serve-sim\/helper\/[^/]+\/webrtc\/(offer|close)$/,
31 /^\/vendor\/serve-sim\/appstate$/,
32 /^\/vendor\/serve-emu\/api$/,
33 /^\/vendor\/serve-emu\/api\/(devices|screenshot|stream-mode|stream-settings|accessibility|fold)$/,
34 /^\/vendor\/serve-emu\/health$/,
35];
36
37// Paths that take methods besides GET; one missing from ALLOWED_PATHS takes no GET.
38const MUTABLE_PATHS: ReadonlyArray<RegExp> = [
39 /^\/vendor\/serve-sim\/api\/screenshot$/,
40 // WebRTC signalling: opening and closing a view of the stream, so read scope.
41 /^\/vendor\/serve-sim\/helper\/[^/]+\/webrtc\/(offer|close)$/,
42 /^\/vendor\/serve-emu\/webrtc\/(offer|close)$/,
43 /^\/vendor\/serve-emu\/api\/(screenshot|stream-mode|stream-settings)$/,
44 /^\/vendor\/serve-emu\/api\/fold$/,
45];
46
47const ALLOWED_WS_PATHS: ReadonlyArray<RegExp> = [
48 /^\/api\/devices\/ws$/,
49 /^\/vendor\/serve-sim\/helper\/ws$/,
50 /^\/vendor\/serve-emu\/ws$/,
51];
52
53const DROPPED_REQUEST_HEADERS = new Set([
54 "host",
55 "connection",
56 "upgrade",
57 "sec-websocket-key",
58 "sec-websocket-version",
59 "sec-websocket-extensions",
60 "sec-websocket-protocol",
61 "cookie",
62 "authorization",
63 "dpop",
64 "content-length",
65 "accept-encoding",
66 // The client's HTTPS hop makes the hub reject our loopback Origin over T3 Connect.
67 "x-forwarded-proto",
68]);
69
70const isWebSocketUpgrade = (request: HttpServerRequest.HttpServerRequest) =>
71 request.headers.upgrade?.toLowerCase() === "websocket";
72
73const forwardHeaders = (request: HttpServerRequest.HttpServerRequest, origin: string) => {
74 const headers: Record<string, string> = {};
75 for (const [name, value] of Object.entries(request.headers)) {
76 if (DROPPED_REQUEST_HEADERS.has(name) || value === undefined) continue;
77 headers[name] = value;
78 }
79 // serve-emu refuses mutations whose Origin differs from the request origin.
80 if (request.headers.origin !== undefined) headers.origin = origin;
81 return headers;
82};
83
84// Effect drops its listeners before closing a socket that is still connecting, and `ws` then
85// emits an `error` nobody handles, which exits the server.
86export const upstreamWebSocketConstructor = Layer.effect(
87 Socket.WebSocketConstructor,
88 Effect.gen(function* () {
89 const make = yield* Socket.WebSocketConstructor;
90 return (url, options) => {
91 const socket = make(url, options);
92 socket.addEventListener("error", () => undefined);
93 return socket;
94 };
95 }),
96).pipe(Layer.provide(NodeSocket.layerWebSocketConstructor));
97
98const proxyWebSocket = Effect.fn("DeviceHubProxy.proxyWebSocket")(function* (
99 request: HttpServerRequest.HttpServerRequest,
100 upstreamUrl: string,
101 protocols: ReadonlyArray<string> = [],
102 headers: Readonly<Record<string, string>> = {},
103) {
104 const client = yield* request.upgrade;
105 const upstream = yield* Socket.makeWebSocket(upstreamUrl, {
106 openTimeout: "10 seconds",
107 ...(protocols.length > 0 ? { protocols: [...protocols] } : {}),
108 ...(Object.keys(headers).length > 0 ? { headers: { ...headers } } : {}),
109 }).pipe(Effect.provide(upstreamWebSocketConstructor));
110 yield* Effect.scoped(
111 Effect.gen(function* () {
112 const writeToClient = yield* client.writer;
113 const writeToUpstream = yield* upstream.writer;
114 return yield* Effect.raceFirst(
115 pumpFrames(upstream, writeToClient),
116 pumpFrames(client, writeToUpstream),
117 );
118 }),
119 ).pipe(Effect.ignoreCause);
120 return HttpServerResponse.empty();
121});
122
123const pumpFrames =
124 /* Frame pumping runs continuously; a span per frame stream adds no diagnostic value. */
125 Effect.fnUntraced(function* (source: Socket.Socket, sink: Socket.Writer) {
126 const { pull } = yield* source.reader;
127 while (true) {
128 yield* sink.writeAll(yield* pull);
129 }
130 });
131
132const proxyHttp = Effect.fn("DeviceHubProxy.proxyHttp")(function* (
133 request: HttpServerRequest.HttpServerRequest,
134 upstreamUrl: string,
135 extraHeaders: Readonly<Record<string, string>> = {},
136) {
137 const httpClient = HttpClient.withScope(yield* HttpClient.HttpClient);
138 const method = request.method;
139 if (!HttpMethod.isHttpMethod(method)) {
140 return HttpServerResponse.text("Method Not Allowed", { status: 405 });
141 }
142 const upstreamRequest = HttpClientRequest.make(method)(upstreamUrl).pipe(
143 HttpClientRequest.setHeaders({
144 ...forwardHeaders(request, new URL(upstreamUrl).origin),
145 ...extraHeaders,
146 }),
147 method === "GET" || method === "HEAD"
148 ? (self) => self
149 : // A bare stream body is sent as octet-stream; serve-sim refuses WebRTC
150 // offers that are not application/json.
151 HttpClientRequest.bodyStream(request.stream, {
152 contentType: request.headers["content-type"],
153 }),
154 );
155 const response = yield* httpClient.execute(upstreamRequest);
156 const headers: Record<string, string> = {};
157 for (const [name, value] of Object.entries(response.headers)) {
158 // A cookie from an upstream such as an EAS session's preview host would land on T3's origin.
159 if (
160 name === "content-encoding" ||
161 name === "transfer-encoding" ||
162 name === "connection" ||
163 name === "set-cookie"
164 ) {
165 continue;
166 }
167 if (value !== undefined) headers[name] = value;
168 }
169 // Long-lived MJPEG and AVCC responses must not be buffered by compression.
170 headers["cache-control"] = "no-store, no-transform";
171 const body = headers["content-type"]?.startsWith("multipart/x-mixed-replace")
172 ? repeatStillFrame(response.stream)
173 : response.stream;
174 return HttpServerResponse.stream(body, {
175 status: response.status,
176 headers,
177 ...(headers["content-type"] ? { contentType: headers["content-type"] } : {}),
178 });
179});
180
181const handler = Effect.gen(function* () {
182 const request = yield* HttpServerRequest.HttpServerRequest;
183 const url = requestUrl(request);
184 if (Option.isNone(url)) {
185 return HttpServerResponse.text("Bad Request", { status: 400 });
186 }
187 const hubPath = url.value.pathname.slice(DeviceService.DEVICE_HUB_ROUTE_PREFIX.length) || "/";
188 const upgrade = isWebSocketUpgrade(request);
189 const matches = (paths: ReadonlyArray<RegExp>) => paths.some((pattern) => pattern.test(hubPath));
190 const allowed = upgrade
191 ? matches(ALLOWED_WS_PATHS)
192 : matches(ALLOWED_PATHS) || matches(MUTABLE_PATHS);
193 if (!allowed) {
194 return HttpServerResponse.text("Not Found", { status: 404 });
195 }
196 const readOnly = request.method === "GET" || request.method === "HEAD";
197 if (!upgrade && !matches(readOnly ? ALLOWED_PATHS : MUTABLE_PATHS)) {
198 return HttpServerResponse.text("Method Not Allowed", { status: 405 });
199 }
200 const controlsDevice =
201 (upgrade && hubPath !== "/api/devices/ws") ||
202 (!readOnly && /\/api\/(stream-(mode|settings)|fold)$/.test(hubPath));
203 yield* authenticateMediaRequest(
204 controlsDevice ? AuthOrchestrationOperateScope : AuthOrchestrationReadScope,
205 );
206 const devices = yield* DeviceService.DeviceService;
207 const ready = yield* devices.currentReadiness(url.value.searchParams.get("hostId") ?? undefined);
208 if (!ready) {
209 return HttpServerResponse.text("Device hub is not running", { status: 503 });
210 }
211 // Consume the auth ticket here; never forward it to the hub.
212 const upstreamSearch = new URLSearchParams(url.value.search);
213 upstreamSearch.delete("wsTicket");
214 upstreamSearch.delete("hostId");
215 const search = upstreamSearch.size > 0 ? `?${upstreamSearch.toString()}` : "";
216 const screenshotDevice = upstreamSearch.get("device");
217 if (
218 screenshotDevice &&
219 ["/vendor/serve-sim/api/screenshot", "/vendor/serve-emu/api/screenshot"].includes(hubPath)
220 ) {
221 const { png } = yield* devices.screenshot({ hostId: ready.hostId, deviceId: screenshotDevice });
222 return HttpServerResponse.uint8Array(png, {
223 contentType: "image/png",
224 headers: { "cache-control": "no-store, no-transform" },
225 });
226 }
227 // Hosts with several upstreams (one per cloud session) route and authenticate requests themselves.
228 const routed = ready.upstream ? yield* ready.upstream(hubPath, upstreamSearch) : undefined;
229 if (routed === null) {
230 return HttpServerResponse.text("Not Found", { status: 404 });
231 }
232 const upstreamUrl = routed?.url ?? `${ready.hub.origin}${hubPath}${search}`;
233 if (upgrade) {
234 return yield* proxyWebSocket(
235 request,
236 upstreamUrl.replace(/^http/, "ws"),
237 routed?.protocols,
238 routed?.headers,
239 );
240 }
241 return yield* proxyHttp(request, upstreamUrl, routed?.headers);
242});
243
244export const layer = HttpRouter.add("*", `${DeviceService.DEVICE_HUB_ROUTE_PREFIX}/*`, handler);