1import * as securitylake from "@distilled.cloud/aws/securitylake";
2import * as Data from "effect/Data";
3import * as Effect from "effect/Effect";
4import * as Stream from "effect/Stream";
5import { isResolved } from "../../Diff.ts";
6import * as Provider from "../../Provider.ts";
7import { Resource } from "../../Resource.ts";
8import type { Providers } from "../Providers.ts";
9
10/**
11 * A natively supported AWS log source that Security Lake can collect.
12 */
13export type AwsLogSourceName = securitylake.AwsLogSourceName;
14
15export interface AwsLogSourceProps {
16 /**
17 * The AWS service to collect logs from (e.g. `ROUTE53`, `VPC_FLOW`,
18 * `CLOUD_TRAIL_MGMT`, `SH_FINDINGS`, `LAMBDA_EXECUTION`, `S3_DATA`,
19 * `EKS_AUDIT`, `WAF`). Changing this replaces the log source.
20 */
21 sourceName: AwsLogSourceName;
22
23 /**
24 * The version of the source schema to collect. Changing this replaces the
25 * log source.
26 * @default - the latest version supported by Security Lake
27 */
28 sourceVersion?: string;
29
30 /**
31 * The Regions to collect this source in. Security Lake must already be
32 * enabled (via `SecurityLake.DataLake`) in each Region.
33 */
34 regions: string[];
35
36 /**
37 * The AWS account IDs to collect this source from.
38 * @default - the current account
39 */
40 accounts?: string[];
41}
42
43/** @resource */
44export interface AwsLogSource extends Resource<
45 "AWS.SecurityLake.AwsLogSource",
46 AwsLogSourceProps,
47 {
48 /** The AWS service the source collects logs from. */
49 sourceName: string;
50 /** The resolved source schema version. */
51 sourceVersion: string | undefined;
52 /** The Regions the source is enabled in. */
53 regions: string[];
54 /** The account IDs the source is collected from (undefined = current). */
55 accounts: string[] | undefined;
56 },
57 never,
58 Providers
59> {}
60
61/**
62 * A natively supported AWS log source (Route 53, VPC Flow Logs, CloudTrail,
63 * Security Hub findings, ...) enabled for collection into the Security Lake
64 * data lake. Requires `SecurityLake.DataLake` to already be enabled in every
65 * configured Region.
66 *
67 * ### Collecting AWS logs
68 * **Example:** Route 53 resolver query logs
69 * ```typescript
70 * const lake = yield* SecurityLake.DataLake("Lake", {
71 * configurations: [{ region: "us-west-2" }],
72 * metaStoreManagerRoleArn: metastoreRole.roleArn,
73 * });
74 * const route53 = yield* SecurityLake.AwsLogSource("Route53Logs", {
75 * sourceName: "ROUTE53",
76 * regions: lake.regions,
77 * });
78 * ```
79 *
80 * **Example:** VPC Flow Logs from specific accounts
81 * ```typescript
82 * const vpcFlow = yield* SecurityLake.AwsLogSource("VpcFlow", {
83 * sourceName: "VPC_FLOW",
84 * sourceVersion: "2.0",
85 * regions: ["us-west-2", "us-east-1"],
86 * accounts: ["123456789012"],
87 * });
88 * ```
89 */
90const AwsLogSourceResource = Resource<AwsLogSource>(
91 "AWS.SecurityLake.AwsLogSource",
92);
93
94export { AwsLogSourceResource as AwsLogSource };
95
96/**
97 * Security Lake reported per-account failures when enabling or disabling an
98 * AWS log source.
99 */
100export class AwsLogSourceOperationFailed extends Data.TaggedError(
101 "AwsLogSourceOperationFailed",
102)<{
103 readonly operation: "create" | "delete";
104 readonly sourceName: string;
105 readonly failedAccounts: string[];
106}> {}
107
108const failIfPartialFailure = (
109 operation: "create" | "delete",
110 sourceName: string,
111 failed: string[] | undefined,
112) =>
113 failed !== undefined && failed.length > 0
114 ? Effect.fail(
115 new AwsLogSourceOperationFailed({
116 operation,
117 sourceName,
118 failedAccounts: [...failed],
119 }),
120 )
121 : Effect.void;
122
123export const AwsLogSourceProvider = () =>
124 Provider.effect(
125 AwsLogSourceResource,
126 Effect.gen(function* () {
127 // Observed state: the set of Regions in which this source (matched by
128 // name, and version when pinned) is currently collecting, plus the
129 // resolved versions AWS reports.
130 const observeSource = Effect.fn(function* (
131 sourceName: string,
132 sourceVersion: string | undefined,
133 ) {
134 const entries = yield* securitylake.listLogSources
135 .items({
136 sources: [
137 {
138 awsLogSource: {
139 sourceName,
140 ...(sourceVersion !== undefined ? { sourceVersion } : {}),
141 },
142 },
143 ],
144 })
145 .pipe(Stream.runCollect);
146 const regions = new Set<string>();
147 let resolvedVersion: string | undefined;
148 for (const entry of entries) {
149 const matched = (entry.sources ?? []).find(
150 (source) =>
151 source.awsLogSource?.sourceName === sourceName &&
152 (sourceVersion === undefined ||
153 source.awsLogSource.sourceVersion === sourceVersion),
154 );
155 if (matched && entry.region !== undefined) {
156 regions.add(entry.region);
157 resolvedVersion ??= matched.awsLogSource?.sourceVersion;
158 }
159 }
160 return { regions, resolvedVersion };
161 });
162
163 const sourceConfig = (
164 sourceName: string,
165 sourceVersion: string | undefined,
166 regions: string[],
167 accounts: string[] | undefined,
168 ): securitylake.AwsLogSourceConfiguration => ({
169 sourceName,
170 regions,
171 ...(sourceVersion !== undefined ? { sourceVersion } : {}),
172 ...(accounts !== undefined ? { accounts } : {}),
173 });
174
175 return {
176 read: Effect.fn(function* ({ olds, output }) {
177 const sourceName = output?.sourceName ?? olds?.sourceName;
178 if (sourceName === undefined) return undefined;
179 const sourceVersion = output?.sourceVersion ?? olds?.sourceVersion;
180 // An account that never onboarded Security Lake rejects
181 // listLogSources — that means "no log source", not a failure.
182 const observed = yield* observeSource(sourceName, sourceVersion).pipe(
183 Effect.catchTag(
184 [
185 "AccessDeniedException",
186 "ResourceNotFoundException",
187 "UnauthorizedException",
188 ],
189 () =>
190 Effect.succeed({
191 regions: new Set<string>(),
192 resolvedVersion: undefined as string | undefined,
193 }),
194 ),
195 );
196 if (observed.regions.size === 0) return undefined;
197 // Log sources carry no tags, so ownership can't be distinguished —
198 // read reports observed state directly.
199 return {
200 sourceName,
201 sourceVersion: observed.resolvedVersion ?? sourceVersion,
202 regions: [...observed.regions].sort(),
203 accounts: output?.accounts ?? olds?.accounts,
204 };
205 }),
206
207 // Enumerate every enabled AWS log source, aggregated by
208 // (sourceName, sourceVersion) across Regions.
209 list: () =>
210 Effect.gen(function* () {
211 const entries = yield* securitylake.listLogSources.items({}).pipe(
212 Stream.runCollect,
213 // An account that never onboarded Security Lake has no data
214 // lake to list sources for.
215 Effect.catchTag(
216 [
217 "AccessDeniedException",
218 "ResourceNotFoundException",
219 "UnauthorizedException",
220 ],
221 () => Effect.succeed([] as securitylake.LogSource[]),
222 ),
223 );
224 const grouped = new Map<
225 string,
226 {
227 sourceName: string;
228 sourceVersion: string | undefined;
229 regions: Set<string>;
230 }
231 >();
232 for (const entry of entries) {
233 for (const source of entry.sources ?? []) {
234 const aws = source.awsLogSource;
235 if (aws?.sourceName === undefined || entry.region === undefined)
236 continue;
237 const key = `${aws.sourceName}@${aws.sourceVersion ?? ""}`;
238 const group = grouped.get(key) ?? {
239 sourceName: aws.sourceName,
240 sourceVersion: aws.sourceVersion,
241 regions: new Set<string>(),
242 };
243 group.regions.add(entry.region);
244 grouped.set(key, group);
245 }
246 }
247 return [...grouped.values()].map((group) => ({
248 sourceName: group.sourceName,
249 sourceVersion: group.sourceVersion,
250 regions: [...group.regions].sort(),
251 accounts: undefined,
252 }));
253 }),
254
255 // The source identity (name + pinned version) is create-only — a
256 // change tears down collection for the old source and enables the new.
257 diff: Effect.fn(function* ({ news, olds }) {
258 if (!isResolved(news)) return undefined;
259 if (
260 news.sourceName !== olds.sourceName ||
261 news.sourceVersion !== olds.sourceVersion
262 ) {
263 return { action: "replace" } as const;
264 }
265 }),
266
267 reconcile: Effect.fn(function* ({ news, output, session }) {
268 const desiredRegions = news.regions;
269 const previousAccounts = output?.accounts;
270
271 // 1. OBSERVE — which Regions already collect this source.
272 const observed = yield* observeSource(
273 news.sourceName,
274 news.sourceVersion,
275 );
276
277 // 2. ENSURE — enable collection in Regions where it is missing.
278 const missingRegions = desiredRegions.filter(
279 (region) => !observed.regions.has(region),
280 );
281 if (missingRegions.length > 0) {
282 const response = yield* securitylake.createAwsLogSource({
283 sources: [
284 sourceConfig(
285 news.sourceName,
286 news.sourceVersion,
287 missingRegions,
288 news.accounts,
289 ),
290 ],
291 });
292 yield* failIfPartialFailure(
293 "create",
294 news.sourceName,
295 response.failed ? [...response.failed] : undefined,
296 );
297 }
298
299 // 3. SYNC accounts — enable for newly added accounts everywhere,
300 // disable for accounts no longer desired.
301 const addedAccounts = (news.accounts ?? []).filter(
302 (account) => !(previousAccounts ?? []).includes(account),
303 );
304 const alreadyEnabledRegions = desiredRegions.filter((region) =>
305 observed.regions.has(region),
306 );
307 if (
308 addedAccounts.length > 0 &&
309 previousAccounts !== undefined &&
310 alreadyEnabledRegions.length > 0
311 ) {
312 const response = yield* securitylake.createAwsLogSource({
313 sources: [
314 sourceConfig(
315 news.sourceName,
316 news.sourceVersion,
317 alreadyEnabledRegions,
318 addedAccounts,
319 ),
320 ],
321 });
322 yield* failIfPartialFailure(
323 "create",
324 news.sourceName,
325 response.failed ? [...response.failed] : undefined,
326 );
327 }
328 const removedAccounts = (previousAccounts ?? []).filter(
329 (account) =>
330 news.accounts !== undefined && !news.accounts.includes(account),
331 );
332 if (removedAccounts.length > 0 && desiredRegions.length > 0) {
333 const response = yield* securitylake.deleteAwsLogSource({
334 sources: [
335 sourceConfig(
336 news.sourceName,
337 news.sourceVersion,
338 desiredRegions,
339 removedAccounts,
340 ),
341 ],
342 });
343 yield* failIfPartialFailure(
344 "delete",
345 news.sourceName,
346 response.failed ? [...response.failed] : undefined,
347 );
348 }
349
350 // 3b. SYNC Region removals — only Regions this resource previously
351 // managed (recorded in output) are disabled.
352 const staleRegions = (output?.regions ?? []).filter(
353 (managed) =>
354 !desiredRegions.includes(managed) &&
355 observed.regions.has(managed),
356 );
357 if (staleRegions.length > 0) {
358 yield* securitylake.deleteAwsLogSource({
359 sources: [
360 sourceConfig(
361 news.sourceName,
362 news.sourceVersion,
363 staleRegions,
364 previousAccounts ?? news.accounts,
365 ),
366 ],
367 });
368 }
369
370 // 4. RETURN fresh attributes.
371 const final = yield* observeSource(
372 news.sourceName,
373 news.sourceVersion,
374 );
375 yield* session.note(news.sourceName);
376 return {
377 sourceName: news.sourceName,
378 sourceVersion: final.resolvedVersion ?? news.sourceVersion,
379 regions:
380 final.regions.size > 0
381 ? [...final.regions].sort()
382 : [...desiredRegions].sort(),
383 accounts: news.accounts,
384 };
385 }),
386
387 delete: Effect.fn(function* ({ output }) {
388 if (output.regions.length === 0) return;
389 yield* securitylake
390 .deleteAwsLogSource({
391 sources: [
392 {
393 sourceName: output.sourceName,
394 regions: output.regions,
395 ...(output.sourceVersion !== undefined
396 ? { sourceVersion: output.sourceVersion }
397 : {}),
398 ...(output.accounts !== undefined
399 ? { accounts: output.accounts }
400 : {}),
401 },
402 ],
403 })
404 .pipe(
405 // Gone already, or the data lake itself was offboarded first
406 // (which stops all collection and makes log-source APIs reject
407 // with UnauthorizedException/AccessDeniedException).
408 Effect.catchTag(
409 [
410 "AccessDeniedException",
411 "ResourceNotFoundException",
412 "UnauthorizedException",
413 ],
414 () => Effect.void,
415 ),
416 );
417 }),
418 };
419 }),
420 );