1/**
2 * Builds server-side handlers for the event-log remote protocol.
3 *
4 * Transport modules use these handlers to expose an event journal to remote
5 * replicas. The handlers run the hello/authenticate challenge flow, attach the
6 * authenticated `EventLog.Identity` to later requests, accept single or chunked
7 * writes, and stream changes back as single or chunked messages.
8 *
9 * @stability unstable
10 * @since 4.0.0
11 */
12import type { NonEmptyReadonlyArray } from "../Array.ts"
13import * as Arr from "../Array.ts"
14import * as Cache from "../Cache.ts"
15import * as Context from "../Context.ts"
16import * as Data from "../Data.ts"
17import * as Effect from "../Effect.ts"
18import * as Equal from "../Equal.ts"
19import * as Hash from "../Hash.ts"
20import * as Layer from "../Layer.ts"
21import * as Option from "../Option.ts"
22import * as Redacted from "../Redacted.ts"
23import type * as Rpc from "../rpc/Rpc.ts"
24import type * as RpcGroup from "../rpc/RpcGroup.ts"
25import * as Stream from "../Stream.ts"
26import type { RemoteId } from "./EventJournal.ts"
27import * as EventLog from "./EventLog.ts"
28import {
29 ChunkedMessage,
30 EventLogAuthentication,
31 EventLogProtocolError,
32 EventLogRemoteRpcs,
33 HelloResponse,
34 SingleMessage,
35 type StoreId
36} from "./EventLogMessage.ts"
37import * as EventLogSessionAuth from "./EventLogSessionAuth.ts"
38
39/**
40 * Provides RPC authentication middleware that reads the authenticated
41 * `EventLog.Identity` from client annotations.
42 *
43 * **Details**
44 *
45 * Requests without an identity fail with a forbidden `EventLogProtocolError`.
46 *
47 * @stability unstable
48 * @category layers
49 * @since 4.0.0
50 */
51export const layerAuthMiddleware: Layer.Layer<
52 EventLogAuthentication
53> = Layer.succeed(EventLogAuthentication, (effect, { client, rpc }) => {
54 const identity = Context.getOrUndefined(client.annotations, EventLog.Identity)
55 if (identity) return Effect.provideService(effect, EventLog.Identity, identity)
56 return Effect.fail(
57 new EventLogProtocolError({
58 requestTag: rpc._tag,
59 publicKey: undefined,
60 code: "Forbidden",
61 message: "Unauthenticated request"
62 })
63 )
64})
65
66/**
67 * Creates the shared RPC handlers for the event-log remote protocol.
68 *
69 * **Details**
70 *
71 * The layer manages hello challenges, verifies session authentication, reassembles
72 * chunked writes, delegates write and change handling to the supplied callbacks,
73 * and frames large change payloads into chunks.
74 *
75 * @stability unstable
76 * @category layers
77 * @since 4.0.0
78 */
79export const layerRpcHandlers = (options: {
80 readonly remoteId: RemoteId
81 readonly authorizeIdentity?: (publicKey: string) => Effect.Effect<void, EventLogProtocolError>
82 readonly getOrCreateSessionAuthBinding: (
83 publicKey: string,
84 signingPublicKey: Uint8Array<ArrayBuffer>
85 ) => Effect.Effect<Uint8Array<ArrayBuffer>>
86 readonly onWrite: (
87 data: Uint8Array<ArrayBuffer>,
88 authenticatedPublicKeys: ReadonlySet<string>
89 ) => Effect.Effect<void, EventLogProtocolError>
90 readonly changes: (options: {
91 readonly publicKey: string
92 readonly storeId: StoreId
93 readonly startSequence: number
94 }) => Stream.Stream<Uint8Array<ArrayBuffer>, unknown>
95}): Layer.Layer<
96 Rpc.ToHandler<RpcGroup.Rpcs<typeof EventLogRemoteRpcs>> | EventLogAuthentication
97> =>
98 EventLogRemoteRpcs.toLayer(Effect.gen(function*() {
99 const clientChallenges = yield* Cache.make({
100 lookup: (_clientId: number) => Effect.orDie(EventLogSessionAuth.makeSessionAuthChallenge),
101 capacity: Number.MAX_SAFE_INTEGER,
102 timeToLive: EventLogSessionAuth.SessionAuthChallengeTimeToLiveMillis
103 })
104 let chunkedIdCounter = 0
105
106 const persistedSigningPublicKeys = yield* Cache.make({
107 lookup: (key: SessionAuthCacheKey) =>
108 options.getOrCreateSessionAuthBinding(key.publicKey, key.signingPublicKey).pipe(
109 Effect.catchCause((_) =>
110 Effect.fail(
111 new EventLogProtocolError({
112 requestTag: "Authenticate",
113 publicKey: key.publicKey,
114 code: "Forbidden",
115 message: "Session auth binding lookup failed"
116 })
117 )
118 )
119 ),
120 capacity: 4096
121 })
122
123 return EventLogRemoteRpcs.of({
124 "EventLog.Hello": Effect.fnUntraced(function*(_, { client }) {
125 const challenge = yield* Cache.get(clientChallenges, client.id)
126 return new HelloResponse({
127 remoteId: options.remoteId,
128 challenge
129 })
130 }),
131 "EventLog.Authenticate": Effect.fnUntraced(function*(request, { client }) {
132 const challenge = Option.getOrNull(yield* Cache.getOption(clientChallenges, client.id))
133 if (!challenge) {
134 return yield* new EventLogProtocolError({
135 requestTag: "Authenticate",
136 publicKey: request.publicKey,
137 code: "Forbidden",
138 message: "Session auth challenge has expired"
139 })
140 }
141 yield* Cache.invalidate(clientChallenges, client.id)
142 if (options.authorizeIdentity) {
143 yield* options.authorizeIdentity(request.publicKey)
144 }
145 const signingPublicKey = yield* Cache.get(
146 persistedSigningPublicKeys,
147 new SessionAuthCacheKey({
148 publicKey: request.publicKey,
149 signingPublicKey: request.signingPublicKey
150 })
151 )
152 const verified = yield* EventLogSessionAuth.verifySessionAuthenticateRequest({
153 remoteId: options.remoteId,
154 challenge,
155 publicKey: request.publicKey,
156 signingPublicKey,
157 signature: request.signature,
158 algorithm: request.algorithm
159 }).pipe(
160 Effect.catch(() => Effect.succeed(false))
161 )
162
163 if (!verified) {
164 return yield* new EventLogProtocolError({
165 requestTag: "Authenticate",
166 publicKey: request.publicKey,
167 code: "Forbidden",
168 message: "Session auth signature verification failed"
169 })
170 }
171
172 const authenticatedIdentities = new Set(
173 Context.getOrUndefined(client.annotations, AuthenticatedIdentities)
174 )
175 authenticatedIdentities.add(request.publicKey)
176 void client
177 .annotate(EventLog.Identity, {
178 publicKey: request.publicKey,
179 privateKey: constEmptyPrivateKey
180 })
181 .annotate(AuthenticatedIdentities, authenticatedIdentities)
182 .annotate(ChunkedMessageState, new Map())
183 }),
184 "EventLog.WriteSingle": Effect.fnUntraced(function*(request, { client }) {
185 yield* options.onWrite(
186 request.data,
187 Context.getOrUndefined(client.annotations, AuthenticatedIdentities) ?? new Set()
188 )
189 }),
190 "EventLog.WriteChunked": Effect.fnUntraced(function*(request, { client }) {
191 const state = Context.get(client.annotations, ChunkedMessageState)
192 const data = ChunkedMessage.join(state, request)
193 if (!data) return
194 yield* options.onWrite(
195 data,
196 Context.getOrUndefined(client.annotations, AuthenticatedIdentities) ?? new Set()
197 )
198 }),
199 "EventLog.Changes": (request, { client }) => {
200 const authenticatedIdentities = Context.getOrUndefined(client.annotations, AuthenticatedIdentities)
201 if (!authenticatedIdentities?.has(request.publicKey)) {
202 return Stream.fail(
203 new EventLogProtocolError({
204 requestTag: "Changes",
205 publicKey: request.publicKey,
206 code: "Forbidden",
207 message: "Identity is not authenticated"
208 })
209 )
210 }
211 return options.changes({
212 publicKey: request.publicKey,
213 storeId: request.storeId,
214 startSequence: request.startSequence
215 }).pipe(
216 Stream.mapArray(Arr.flatMap((data): NonEmptyReadonlyArray<SingleMessage | ChunkedMessage> => {
217 if (data.byteLength <= ChunkedMessage.chunkSize) {
218 return [new SingleMessage({ data })]
219 }
220 return ChunkedMessage.split(chunkedIdCounter++, data)
221 })),
222 Stream.catchCause((_) =>
223 Stream.fail(
224 new EventLogProtocolError({
225 requestTag: "Changes",
226 publicKey: request.publicKey,
227 code: "InternalServerError",
228 message: "Decoding failure"
229 })
230 )
231 )
232 )
233 }
234 })
235 })).pipe(
236 Layer.merge(layerAuthMiddleware)
237 )
238
239/**
240 * Annotation that stores partial `ChunkedMessage` data while chunked writes are
241 * being reassembled.
242 *
243 * **When to use**
244 *
245 * Use to keep per-client chunk assembly state while handling chunked event-log
246 * writes.
247 *
248 * @stability unstable
249 * @category services
250 * @since 4.0.0
251 */
252export class ChunkedMessageState extends Context.Reference<
253 Map<number, {
254 readonly parts: Array<Uint8Array>
255 count: number
256 bytes: number
257 }>
258>("effect/eventlog/EventLogServer/ChunkedMessageState", {
259 defaultValue: () => new Map()
260}) {}
261
262/**
263 * Annotation containing the public keys authenticated on an RPC connection.
264 *
265 * @stability unstable
266 * @category services
267 * @since 4.0.0
268 */
269export class AuthenticatedIdentities extends Context.Service<AuthenticatedIdentities, Set<string>>()(
270 "effect/eventlog/EventLogServer/AuthenticatedIdentities"
271) {}
272
273class SessionAuthCacheKey extends Data.Class<{
274 readonly publicKey: string
275 readonly signingPublicKey: Uint8Array<ArrayBuffer>
276}> {
277 [Equal.symbol](that: SessionAuthCacheKey) {
278 return this.publicKey === that.publicKey
279 }
280 [Hash.symbol]() {
281 return Hash.string(this.publicKey)
282 }
283}
284
285const constEmptyPrivateKey = Redacted.make(new Uint8Array(32))