1import * as securitylake from "@distilled.cloud/aws/securitylake";
2import * as Data from "effect/Data";
3import * as Effect from "effect/Effect";
4import * as Option from "effect/Option";
5import * as Stream from "effect/Stream";
6import { Unowned } from "../../AdoptPolicy.ts";
7import { isResolved } from "../../Diff.ts";
8import { createPhysicalName } from "../../PhysicalName.ts";
9import * as Provider from "../../Provider.ts";
10import { Resource } from "../../Resource.ts";
11import { createInternalTags, diffTags, hasAlchemyTags } from "../../Tags.ts";
12import type { Providers } from "../Providers.ts";
13import {
14 readSecurityLakeTags,
15 retryWhileConflict,
16 toTagList,
17} from "./internal.ts";
18
19/**
20 * The AWS identity (principal + external ID) that a subscriber authenticates
21 * with when consuming data from Security Lake.
22 */
23export interface SubscriberIdentity {
24 /** The AWS principal (account ID or service principal) of the subscriber. */
25 principal: string;
26 /** The external ID the subscriber must present when assuming the role. */
27 externalId: string;
28}
29
30/**
31 * A log source a subscriber consumes — either a natively supported AWS source
32 * or a custom source.
33 */
34export type SubscriberSource = securitylake.LogSourceResource;
35
36/**
37 * How the subscriber accesses the data — direct S3 access or Lake Formation
38 * (query) access.
39 */
40export type SubscriberAccessType = "LAKEFORMATION" | "S3";
41
42export interface SubscriberProps {
43 /**
44 * Name of the subscriber. If omitted, a unique physical name is generated
45 * from the app, stage, and logical ID.
46 */
47 subscriberName?: string;
48
49 /**
50 * The AWS identity (principal + external ID) the subscriber uses to access
51 * Security Lake data.
52 */
53 subscriberIdentity: SubscriberIdentity;
54
55 /**
56 * Description of the subscriber.
57 */
58 subscriberDescription?: string;
59
60 /**
61 * The log sources the subscriber consumes, e.g.
62 * `[{ awsLogSource: { sourceName: "ROUTE53", sourceVersion: "2.0" } }]`.
63 */
64 sources: SubscriberSource[];
65
66 /**
67 * How the subscriber accesses the data. Changing this replaces the
68 * subscriber (access type is create-only).
69 * @default ["S3"]
70 */
71 accessTypes?: SubscriberAccessType[];
72
73 /**
74 * Tags applied to the subscriber. Alchemy ownership tags are merged in
75 * automatically.
76 */
77 tags?: Record<string, string>;
78}
79
80/** @resource */
81export interface Subscriber extends Resource<
82 "AWS.SecurityLake.Subscriber",
83 SubscriberProps,
84 {
85 /** Unique ID of the subscriber (UUID). */
86 subscriberId: string;
87 /** ARN of the subscriber. */
88 subscriberArn: string;
89 /** Name of the subscriber. */
90 subscriberName: string;
91 /** Current status (`ACTIVE`, `PENDING`, `READY`, `DEACTIVATED`). */
92 subscriberStatus: string | undefined;
93 /** ARN of the IAM role created for the subscriber to assume. */
94 roleArn: string | undefined;
95 /** ARN of the S3 bucket the subscriber reads from. */
96 s3BucketArn: string | undefined;
97 /** The subscriber's notification endpoint, if one is configured. */
98 subscriberEndpoint: string | undefined;
99 /** ARN of the RAM resource share (Lake Formation access only). */
100 resourceShareArn: string | undefined;
101 /** Name of the RAM resource share (Lake Formation access only). */
102 resourceShareName: string | undefined;
103 },
104 never,
105 Providers
106> {}
107
108/**
109 * A Security Lake subscriber — a consumer (account or service) granted access
110 * to data in the Security Lake data lake for specific log sources.
111 *
112 * ### Creating a Subscriber
113 * **Example:** S3 data-access subscriber
114 * ```typescript
115 * const subscriber = yield* SecurityLake.Subscriber("Analytics", {
116 * subscriberIdentity: {
117 * principal: "123456789012",
118 * externalId: "analytics-external-id",
119 * },
120 * sources: [{ awsLogSource: { sourceName: "ROUTE53", sourceVersion: "2.0" } }],
121 * });
122 * ```
123 *
124 * **Example:** Lake Formation (query) access
125 * ```typescript
126 * const subscriber = yield* SecurityLake.Subscriber("Athena", {
127 * subscriberName: "athena-consumer",
128 * subscriberDescription: "Athena query access to VPC flow logs",
129 * subscriberIdentity: {
130 * principal: "123456789012",
131 * externalId: "athena-external-id",
132 * },
133 * sources: [{ awsLogSource: { sourceName: "VPC_FLOW", sourceVersion: "2.0" } }],
134 * accessTypes: ["LAKEFORMATION"],
135 * tags: { team: "security" },
136 * });
137 * ```
138 */
139const SubscriberResource = Resource<Subscriber>("AWS.SecurityLake.Subscriber");
140
141export { SubscriberResource as Subscriber };
142
143/**
144 * `CreateSubscriber` returned without a subscriber body and the subscriber
145 * could not be re-observed by name.
146 */
147export class SubscriberCreateFailed extends Data.TaggedError(
148 "SubscriberCreateFailed",
149)<{ readonly subscriberName: string }> {}
150
151const buildAttrs = (subscriber: securitylake.SubscriberResource) => ({
152 subscriberId: subscriber.subscriberId,
153 subscriberArn: subscriber.subscriberArn,
154 subscriberName: subscriber.subscriberName,
155 subscriberStatus: subscriber.subscriberStatus,
156 roleArn: subscriber.roleArn,
157 s3BucketArn: subscriber.s3BucketArn,
158 subscriberEndpoint: subscriber.subscriberEndpoint,
159 resourceShareArn: subscriber.resourceShareArn,
160 resourceShareName: subscriber.resourceShareName,
161});
162
163// Desired sources match observed sources when every desired entry appears in
164// the observed list (AWS resolves an omitted sourceVersion, so it is only
165// compared when the desired entry pins one) and the counts agree.
166const sourcesMatch = (
167 desired: readonly securitylake.LogSourceResource[],
168 observed: readonly securitylake.LogSourceResource[],
169) =>
170 desired.length === observed.length &&
171 desired.every((want) => {
172 const wantAws = want.awsLogSource;
173 if (wantAws !== undefined) {
174 return observed.some((have) => {
175 const haveAws = have.awsLogSource;
176 return (
177 haveAws !== undefined &&
178 haveAws.sourceName === wantAws.sourceName &&
179 (wantAws.sourceVersion === undefined ||
180 haveAws.sourceVersion === wantAws.sourceVersion)
181 );
182 });
183 }
184 return observed.some(
185 (have) =>
186 have.customLogSource?.sourceName === want.customLogSource?.sourceName &&
187 (want.customLogSource?.sourceVersion === undefined ||
188 have.customLogSource?.sourceVersion ===
189 want.customLogSource.sourceVersion),
190 );
191 });
192
193export const SubscriberProvider = () =>
194 Provider.effect(
195 SubscriberResource,
196 Effect.gen(function* () {
197 const createName = Effect.fn(function* (
198 id: string,
199 props: { subscriberName?: string },
200 ) {
201 return (
202 props.subscriberName ??
203 (yield* createPhysicalName({ id, maxLength: 64 }))
204 );
205 });
206
207 const getById = (subscriberId: string) =>
208 securitylake.getSubscriber({ subscriberId }).pipe(
209 Effect.map((response) => response.subscriber),
210 Effect.catchTag("ResourceNotFoundException", () =>
211 Effect.succeed(undefined),
212 ),
213 );
214
215 // Subscriber names are unique per account/region — find the live
216 // subscriber carrying our deterministic name when state was lost.
217 const findByName = (subscriberName: string) =>
218 securitylake.listSubscribers.items({}).pipe(
219 Stream.filter(
220 (subscriber) => subscriber.subscriberName === subscriberName,
221 ),
222 Stream.take(1),
223 Stream.runHead,
224 Effect.map(Option.getOrUndefined),
225 );
226
227 return {
228 read: Effect.fn(function* ({ id, olds, output }) {
229 // An account that never onboarded Security Lake rejects subscriber
230 // APIs with UnauthorizedException — that means "no subscriber".
231 const subscriber = yield* (
232 output?.subscriberId
233 ? getById(output.subscriberId)
234 : Effect.flatMap(createName(id, olds ?? {}), findByName)
235 ).pipe(
236 Effect.catchTag("UnauthorizedException", () =>
237 Effect.succeed(undefined),
238 ),
239 );
240 if (subscriber === undefined) return undefined;
241 const attrs = buildAttrs(subscriber);
242 const tags = yield* readSecurityLakeTags(attrs.subscriberArn);
243 return (yield* hasAlchemyTags(id, tags)) ? attrs : Unowned(attrs);
244 }),
245
246 list: () =>
247 securitylake.listSubscribers.items({}).pipe(
248 Stream.runCollect,
249 Effect.map((subscribers) => [...subscribers].map(buildAttrs)),
250 // An account that never onboarded Security Lake has no
251 // subscribers to enumerate.
252 Effect.catchTag(
253 [
254 "AccessDeniedException",
255 "ResourceNotFoundException",
256 "UnauthorizedException",
257 ],
258 () => Effect.succeed([]),
259 ),
260 ),
261
262 // accessTypes is create-only (UpdateSubscriber cannot change it).
263 diff: Effect.fn(function* ({ news, olds }) {
264 if (!isResolved(news)) return undefined;
265 const oldAccess = [...(olds.accessTypes ?? ["S3"])].sort();
266 const newAccess = [...(news.accessTypes ?? ["S3"])].sort();
267 if (
268 oldAccess.length !== newAccess.length ||
269 oldAccess.some((value, index) => value !== newAccess[index])
270 ) {
271 return { action: "replace" } as const;
272 }
273 }),
274
275 reconcile: Effect.fn(function* ({ id, news, output, session }) {
276 const subscriberName = yield* createName(id, news);
277 const internalTags = yield* createInternalTags(id);
278 const desiredTags = { ...news.tags, ...internalTags };
279
280 // 1. OBSERVE — by cached id first, then by deterministic name.
281 let subscriber = output?.subscriberId
282 ? yield* getById(output.subscriberId)
283 : undefined;
284 subscriber ??= yield* findByName(subscriberName);
285
286 // 2. ENSURE — create when missing; a ConflictException means a
287 // concurrent create won the race, so re-observe by name.
288 if (subscriber === undefined) {
289 subscriber = yield* securitylake
290 .createSubscriber({
291 subscriberName,
292 subscriberIdentity: news.subscriberIdentity,
293 subscriberDescription: news.subscriberDescription,
294 sources: news.sources,
295 accessTypes: news.accessTypes,
296 tags: toTagList(desiredTags),
297 })
298 .pipe(
299 Effect.map((response) => response.subscriber),
300 Effect.catchTag("ConflictException", () =>
301 findByName(subscriberName),
302 ),
303 );
304 if (subscriber === undefined) {
305 return yield* Effect.fail(
306 new SubscriberCreateFailed({ subscriberName }),
307 );
308 }
309 } else {
310 // 3. SYNC mutable settings — observed ↔ desired.
311 const changed =
312 subscriber.subscriberName !== subscriberName ||
313 (subscriber.subscriberDescription ?? "") !==
314 (news.subscriberDescription ?? "") ||
315 subscriber.subscriberIdentity.principal !==
316 news.subscriberIdentity.principal ||
317 subscriber.subscriberIdentity.externalId !==
318 news.subscriberIdentity.externalId ||
319 !sourcesMatch(news.sources, subscriber.sources);
320 if (changed) {
321 subscriber = yield* securitylake
322 .updateSubscriber({
323 subscriberId: subscriber.subscriberId,
324 subscriberName,
325 subscriberDescription: news.subscriberDescription,
326 subscriberIdentity: news.subscriberIdentity,
327 sources: news.sources,
328 })
329 .pipe(
330 retryWhileConflict,
331 Effect.map((response) => response.subscriber ?? subscriber!),
332 );
333 }
334
335 // 3b. SYNC tags — diff against OBSERVED cloud tags.
336 const observedTags = yield* readSecurityLakeTags(
337 subscriber.subscriberArn,
338 );
339 const { upsert, removed } = diffTags(observedTags, desiredTags);
340 if (upsert.length > 0) {
341 yield* securitylake.tagResource({
342 resourceArn: subscriber.subscriberArn,
343 tags: upsert.map((t) => ({ key: t.Key, value: t.Value })),
344 });
345 }
346 if (removed.length > 0) {
347 yield* securitylake.untagResource({
348 resourceArn: subscriber.subscriberArn,
349 tagKeys: removed,
350 });
351 }
352 }
353
354 // 4. RETURN fresh attributes.
355 const final = (yield* getById(subscriber.subscriberId)) ?? subscriber;
356 yield* session.note(final.subscriberArn);
357 return buildAttrs(final);
358 }),
359
360 delete: Effect.fn(function* ({ output }) {
361 yield* securitylake
362 .deleteSubscriber({ subscriberId: output.subscriberId })
363 .pipe(
364 retryWhileConflict,
365 // Gone already, or the data lake itself was offboarded first
366 // (which removes all subscribers and makes subscriber APIs
367 // reject with UnauthorizedException).
368 Effect.catchTag(
369 ["ResourceNotFoundException", "UnauthorizedException"],
370 () => Effect.void,
371 ),
372 );
373 }),
374 };
375 }),
376 );