1import * as NodeHttpServer from "@effect/platform-node/NodeHttpServer";
2import * as NodeServices from "@effect/platform-node/NodeServices";
3import * as NodeSocket from "@effect/platform-node/NodeSocket";
4import { expect, it } from "@effect/vitest";
5import {
6 AuthEnvironmentConnectScope,
7 AuthOrchestrationReadScope,
8 AuthAccessWriteScope,
9 AuthSessionId,
10 EnvironmentId,
11 ORCHESTRATION_PROTOCOL_VERSION,
12 OutboundHubMessage,
13 type OutboundSourceMessage,
14} from "@t3tools/contracts";
15import * as Context from "effect/Context";
16import * as Deferred from "effect/Deferred";
17import * as Effect from "effect/Effect";
18import * as Fiber from "effect/Fiber";
19import * as Layer from "effect/Layer";
20import * as Option from "effect/Option";
21import * as Queue from "effect/Queue";
22import * as Schema from "effect/Schema";
23import * as Socket from "effect/socket/Socket";
24import {
25 FetchHttpClient,
26 HttpClient,
27 HttpClientRequest,
28 HttpRouter,
29 HttpServer,
30 HttpServerRequest,
31 HttpServerResponse,
32} from "effect/http";
33import * as HostedAuth from "../auth/HostedAuth.ts";
34import * as ServerSecretStore from "../auth/ServerSecretStore.ts";
35import * as SessionStore from "../auth/SessionStore.ts";
36import * as OutboundHub from "./OutboundHub.ts";
37import { bridge } from "./OutboundSocket.ts";
38
39const decodeHubMessage = Schema.decodeEffect(Schema.fromJsonString(OutboundHubMessage));
40
41const socketPair = Effect.gen(function* () {
42 const inbound = yield* Queue.unbounded<string | Uint8Array>();
43 const outbound = yield* Queue.unbounded<string | Uint8Array>();
44 const endpoint = (read: typeof inbound, write: typeof outbound) => {
45 const send = (frame: string | Uint8Array | Socket.CloseEvent) =>
46 Socket.isCloseEvent(frame) ? Effect.void : Queue.offer(write, frame).pipe(Effect.asVoid);
47 return Socket.make({
48 reader: Effect.succeed({
49 pull: Queue.take(read).pipe(Effect.map((frame) => [frame] as const)),
50 upgrade: () => Effect.void,
51 }),
52 writer: Effect.succeed({
53 write: send,
54 writeAll: (frames) => Effect.forEach(frames, send, { discard: true }),
55 }),
56 });
57 };
58 return { left: endpoint(inbound, outbound), right: endpoint(outbound, inbound) };
59});
60const session = {
61 sessionId: AuthSessionId.make("device-session"),
62 deviceId: "device-a",
63 subject: "owner",
64 method: "bearer-access-token" as const,
65 scopes: [AuthEnvironmentConnectScope],
66};
67const descriptor = {
68 environmentId: EnvironmentId.make("existing-laptop-id"),
69 label: "Laptop",
70 platform: { os: "linux" as const, arch: "x64" as const },
71 serverVersion: "test",
72 orchestrationProtocolVersion: ORCHESTRATION_PROTOCOL_VERSION,
73 capabilities: { repositoryIdentity: true },
74};
75
76const fixture = Effect.gen(function* () {
77 const data = new Map<string, Uint8Array>();
78 const revoked = yield* Deferred.make<void>();
79 const serviceLayer = OutboundHub.layer.pipe(
80 Layer.provide(
81 Layer.mergeAll(
82 Layer.succeed(ServerSecretStore.ServerSecretStore, {
83 get: (key: string) => Effect.sync(() => Option.fromUndefinedOr(data.get(key))),
84 set: (key: string, value: Uint8Array) =>
85 Effect.sync(() => {
86 data.set(key, value);
87 }),
88 } as unknown as ServerSecretStore.ServerSecretStore["Service"]),
89 Layer.succeed(HostedAuth.HostedAuth, {
90 bindEnvironment: () => Effect.void,
91 } as unknown as HostedAuth.HostedAuth["Service"]),
92 Layer.succeed(SessionStore.SessionStore, {
93 awaitInvalidation: () => Deferred.await(revoked),
94 } as unknown as SessionStore.SessionStore["Service"]),
95 ),
96 ),
97 );
98 const hub = yield* OutboundHub.OutboundHub.pipe(Effect.provide(serviceLayer));
99 const ready = yield* Deferred.make<void>();
100 const requests: Array<OutboundHubMessage> = [];
101 const control = yield* socketPair;
102 const accepted = yield* hub.accept(session, control.left).pipe(Effect.forkScoped);
103 const reader = yield* control.right.reader;
104 const writer = yield* control.right.writer;
105 const send = (message: OutboundSourceMessage) => writer.write(JSON.stringify(message));
106 const source = (port?: number) =>
107 Effect.gen(function* () {
108 yield* send({ type: "register", descriptor });
109 return yield* reader.pull.pipe(
110 Effect.flatMap((frames) =>
111 Effect.forEach(
112 frames,
113 (frame) =>
114 Effect.gen(function* () {
115 const message = yield* decodeHubMessage(
116 typeof frame === "string" ? frame : new TextDecoder().decode(frame),
117 );
118 requests.push(message);
119 if (message.type === "ping") {
120 yield* Deferred.succeed(ready, undefined);
121 return yield* send({ type: "pong" });
122 }
123 if (message.type === "grant")
124 return yield* send({ type: "grant", id: message.id, token: "scoped-leaf-token" });
125 if (port === undefined) return;
126 const channel = yield* socketPair;
127 yield* hub.channel(session, message.id, channel.left).pipe(Effect.forkScoped);
128 yield* Effect.scoped(
129 Effect.gen(function* () {
130 const local = yield* NodeSocket.makeNet({ host: "127.0.0.1", port });
131 return yield* bridge(local, channel.right);
132 }),
133 ).pipe(Effect.ignore, Effect.forkScoped);
134 }),
135 { discard: true },
136 ),
137 ),
138 Effect.forever,
139 );
140 });
141 return {
142 hub,
143 data,
144 accepted,
145 requests,
146 source,
147 ready,
148 send,
149 revoke: Deferred.succeed(revoked, undefined),
150 restart: OutboundHub.OutboundHub.pipe(Effect.provide(serviceLayer)),
151 };
152});
153
154it.live(
155 "keeps enrolled IDs across restart and removes interactive authority on device revocation",
156 () =>
157 Effect.gen(function* () {
158 const f = yield* fixture;
159 yield* f.source().pipe(Effect.forkScoped);
160 yield* Deferred.await(f.ready);
161 const browser = {
162 ...session,
163 sessionId: AuthSessionId.make("browser"),
164 scopes: [AuthOrchestrationReadScope, AuthAccessWriteScope],
165 };
166 const grant = yield* f.hub.connect(descriptor.environmentId, browser);
167 expect(grant.token).toBe("scoped-leaf-token");
168 expect(f.requests.find((message) => message.type === "grant")).toMatchObject({
169 scopes: [AuthOrchestrationReadScope],
170 });
171 yield* f.hub.connect(descriptor.environmentId, browser);
172 expect(f.requests.filter((message) => message.type === "grant")).toHaveLength(1);
173 expect((yield* (yield* f.restart).enrollments)[0]?.descriptor.environmentId).toBe(
174 descriptor.environmentId,
175 );
176 yield* f.revoke;
177 yield* Fiber.join(f.accepted);
178 yield* grant.invalidated;
179 expect(
180 (yield* f.hub.connect(descriptor.environmentId, browser).pipe(Effect.result))._tag,
181 ).toBe("Failure");
182 expect((yield* f.hub.enrollments)[0]?.descriptor.environmentId).toBe(
183 descriptor.environmentId,
184 );
185 }).pipe(Effect.provide(NodeServices.layer)),
186);
187
188it.live("streams HTTP bytes through an outbound channel without a routable laptop URL", () =>
189 Effect.gen(function* () {
190 const routes = HttpRouter.add(
191 "POST",
192 "/echo",
193 Effect.gen(function* () {
194 const request = yield* HttpServerRequest.HttpServerRequest;
195 expect(request.headers.authorization).toBe("Bearer scoped-leaf-token");
196 return HttpServerResponse.uint8Array(new Uint8Array(yield* request.arrayBuffer));
197 }),
198 );
199 const server = yield* Layer.build(
200 HttpRouter.serve(routes, { disableListenLog: true }).pipe(
201 Layer.provideMerge(NodeHttpServer.layerTest),
202 ),
203 );
204 const address = Context.get(server, HttpServer.HttpServer).address;
205 if (address._tag !== "InetAddressV4" && address._tag !== "InetAddressV6")
206 return yield* Effect.die("Missing fixture port");
207 const f = yield* fixture;
208 yield* f.source(address.port).pipe(Effect.forkScoped);
209 yield* Deferred.await(f.ready);
210 const grant = yield* f.hub.connect(descriptor.environmentId, {
211 ...session,
212 scopes: [AuthOrchestrationReadScope],
213 });
214 const bytes = new Uint8Array(524_288).map((_, index) => index % 251);
215 const response = yield* HttpClient.execute(
216 HttpClientRequest.post(`${grant.url}echo`).pipe(
217 HttpClientRequest.bodyUint8Array(bytes),
218 HttpClientRequest.bearerToken(grant.token),
219 ),
220 );
221 expect(new Uint8Array(yield* response.arrayBuffer)).toEqual(bytes);
222 }).pipe(Effect.provide(Layer.merge(NodeServices.layer, FetchHttpClient.layer))),
223);
224
225it.live(
226 "requires a hosted device, a connect grant, and the original session for one-use channels",
227 () =>
228 Effect.gen(function* () {
229 const f = yield* fixture;
230 const socket = yield* socketPair;
231 const { deviceId: _deviceId, ...unbound } = session;
232 expect((yield* f.hub.accept(unbound, socket.left).pipe(Effect.result))._tag).toBe("Failure");
233 expect(
234 (yield* f.hub.accept({ ...session, scopes: [] }, socket.left).pipe(Effect.result))._tag,
235 ).toBe("Failure");
236 expect(
237 (yield* f.hub
238 .channel({ ...session, deviceId: "device-b" }, "unissued", socket.left)
239 .pipe(Effect.result))._tag,
240 ).toBe("Failure");
241 }).pipe(Effect.provide(NodeServices.layer)),
242);