diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index 22ca54f5..544f6d3e 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -248,6 +248,20 @@ than collection (data classification, which accounts are collected, the real topology), so findings state what collection would add, and ports no catalogue rule covers yet are marked as such rather than given an invented rule id. +The gate widens beyond Terraform plans: IAM policy documents embedded in the +plan (`aws_iam_policy` / `aws_iam_role_policy`) are parsed with the **same +policy parser the CIEM evaluator runs**, and a policy that newly grants +`iam:PassRole` on `Resource: "*"` fails it as critical (with a before-policy — +plan `change.before` or `--policy-before` — the checks are a true diff: an +unchanged grant is not newly introduced; statement conditions, SCPs and +permission boundaries are not evaluated and findings say so). Standalone +files cover flows outside Terraform: `--policy after.json [--policy-before +before.json]`. `--helm-values values.yaml` flags the chart values that decide +network reach: an ingress annotated internet-facing is high, a plain enabled +ingress is a medium note (the controller's default decides). All inputs share +one report and one exit code, so teams can wire the same optional CI gate to +plan, policy and chart changes. + ## Observed reachability Flow logs are the one signal that says where traffic actually went. The @@ -256,15 +270,23 @@ queries the `vpc_flow_logs` Glue table — partition projection over account/region/day, the standard v5 fields with the account id taken from the partition — for ACCEPT flows per network interface within a bounded window (default 7 days, max 31, at most 5000 interfaces per run) and writes -`TrafficObservations` in DynamoDB (one row per interface, TTL 30 days). The -collector reads that table during EC2 collection with the hub's own -credentials and marks each instance `traffic_observed` (with the latest -observation time); member-account credentials never read it. Reachability -confidence in path likelihood is then **observed AND configured**: traffic -observed on a workload proves reachability even when a restricted read left -the configuration unknown, and the issue page lists the observed-traffic fact -beside the configured ones. Everything is cost-gated: Terraform creates -nothing without `flow_logs_bucket`, the CronJob ships suspended, and a +`TrafficObservations` in DynamoDB (one row per interface, TTL 30 days). A +second, bounded aggregation runs alongside it: the top remote senders +(`srcaddr`) per interface, ranked by bytes via a window function and capped at +5 per interface with a parsed-row cap, stored as a small endpoint list on the +observation row — two Athena scans per run, one extra bound, and a failure in +the endpoint query degrades to observations without citations. The +collector reads the table during EC2 collection with the hub's own +credentials and marks each instance `traffic_observed` (latest observation +time, and the deduplicated top endpoints, biggest senders first, capped at 5, +as `observed_endpoints`); member-account credentials never read it. +Reachability confidence in path likelihood is then **observed AND configured**: +traffic observed on a workload proves reachability even when a restricted read +left the configuration unknown, and the graph, the attack-path explorer and +the issue page cite it — a hop that arrives at an observed workload shows an +"observed traffic from
" chip, and issues store the per-hop citations +(`observedTraffic`) in their metadata. Everything is cost-gated: Terraform +creates nothing without `flow_logs_bucket`, the CronJob ships suspended, and a deployment without the table simply has no observed traffic. ## Threat-intelligence matching diff --git a/CHANGELOG.md b/CHANGELOG.md index 51dbf8f8..8fe3497c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,18 @@ All notable changes to Khalifa. +## [Unreleased] - Observed-traffic hop citations and shift-left v2 + +### Added + +- **Endpoint-level flow-log evidence**: the flow-log analyzer now runs a second, bounded aggregation alongside the per-interface totals — the top remote senders (`srcaddr`) per interface, ranked by bytes with a window function, capped at 5 per interface with a parsed-row cap and two Athena scans per run. The endpoints ride the existing `TrafficObservations` row (TTL unchanged); a failure in the endpoint query degrades to observations without citations, never a failed run. The collector deduplicates the top senders across an instance's interfaces (biggest first, capped at 5) onto `observed_endpoints`. +- **Hops cite observations**: attack paths carry per-hop `observedTraffic` citations (the engine maps each observed vertex to the 1-based hop that arrives at it, tolerantly), the explorer response and issue metadata include them, and the graph draws an "observed traffic from
" chip on the hop's target — so a path shows observed traffic on a specific hop, not just configured exposure. +- **Shift-left v2 in `risk:plan`**: IAM policy documents embedded in the plan (`aws_iam_policy` / `aws_iam_role_policy`) are parsed with the same parser the CIEM evaluator runs; a policy that newly grants `iam:PassRole` on `Resource: "*"` fails the gate as critical, and a wildcard `Action: "*"` is high. With a before-policy (plan `change.before`, or `--policy-before` with the new `--policy`) checks are a true diff — unchanged grants are not re-reported; conditions/SCPs are explicitly not evaluated. New `--helm-values values.yaml` flags an internet-facing ingress as high (plain enabled ingress: a medium note). Standalone policy files cover non-Terraform flows; the README documents the optional CI gate for all three inputs. + +### Notes + +- Still open from the re-review (`~/Documents/Testing/plans/2026-10-27-predictive-completion.md`): the long tail (UEBA breadth, EPSS/KEV enrichment, trust-center publication of projections, per-issue risk window/SLO). + ## [Unreleased] - Freshness tempering across risk ### Changed diff --git a/api-service/src/__tests__/attack-path-search.test.ts b/api-service/src/__tests__/attack-path-search.test.ts index df5c89a3..0f446ad1 100644 --- a/api-service/src/__tests__/attack-path-search.test.ts +++ b/api-service/src/__tests__/attack-path-search.test.ts @@ -233,6 +233,21 @@ describe('GET /attack-paths/search', () => { expect(res.body.coverage).toMatchObject({ level: 'unknown' }); }); + it('cites observed traffic on the hop that arrives at a flagged workload', async () => { + const ec2 = v('i', 'Ec2Instance', { + is_internet_exposed: true, + traffic_observed: true, + observed_endpoints: JSON.stringify([{ address: '203.0.113.7', bytes: 2048, flows: 3 }]), + }); + searchAttackPaths.mockResolvedValue([{ nodes: [v('internet', 'Internet'), ec2], edges: [] }]); + + const res = await call({ from: 'internet', to: 'crown' }); + + expect(res.body.paths[0].observedTraffic).toEqual([ + { hop: 1, vertexId: 'i', endpoints: [{ address: '203.0.113.7', bytes: 2048, flows: 3 }] }, + ]); + }); + it('tempers every path when the newest collection is stale, and says so', async () => { const ec2 = v('i', 'Ec2Instance', { is_internet_exposed: true }); searchAttackPaths.mockResolvedValue([{ nodes: [ec2], edges: [] }]); diff --git a/api-service/src/routes/attack-paths.ts b/api-service/src/routes/attack-paths.ts index b121a0a4..0a5c98f1 100644 --- a/api-service/src/routes/attack-paths.ts +++ b/api-service/src/routes/attack-paths.ts @@ -7,6 +7,7 @@ import { getRuleById, pathLikelihood, riskInputFromPath, + observedTrafficFromPath, } from '@khalifa/risk-engine'; import { getApiNeptuneClient } from '../services/api-dependencies'; import { graphCoverage } from '../services/coverage'; @@ -127,12 +128,14 @@ export async function searchAttackPaths(req: Request, res: Response): Promise 0 ? { observedTraffic } : {}), }; }) .sort((a, b) => b.score - a.score || b.likelihood - a.likelihood); diff --git a/lambdas/collector/index.ts b/lambdas/collector/index.ts index af4df9e3..9d335aec 100644 --- a/lambdas/collector/index.ts +++ b/lambdas/collector/index.ts @@ -1073,6 +1073,10 @@ async function collectEc2( if (observation) { node.properties.traffic_observed = true; if (observation.lastSeen) node.properties.traffic_observed_at = observation.lastSeen; + if (observation.endpoints && observation.endpoints.length > 0) { + // Bounded citation for attack-path hops (address/bytes/flows/last_seen). + node.properties.observed_endpoints = JSON.stringify(observation.endpoints); + } } } } diff --git a/lambdas/collector/src/observations.test.ts b/lambdas/collector/src/observations.test.ts index a418c573..84538734 100644 --- a/lambdas/collector/src/observations.test.ts +++ b/lambdas/collector/src/observations.test.ts @@ -1,4 +1,8 @@ -import { instanceObservation, loadTrafficObservations } from './observations'; +import { + instanceObservation, + loadTrafficObservations, + MAX_INSTANCE_ENDPOINTS, +} from './observations'; function clientReturning(...responses: unknown[]): any { const send = jest.fn(); @@ -34,6 +38,19 @@ describe('loadTrafficObservations', () => { packets: { N: '12' }, bytes: { N: '4096' }, last_seen: { S: '2026-10-19T02:00:00Z' }, + endpoints: { + L: [ + { + M: { + address: { S: '203.0.113.7' }, + bytes: { N: '2048' }, + flows: { N: '2' }, + last_seen: { S: '2026-10-19T01:30:00Z' }, + }, + }, + { M: { address: { S: '198.51.100.9' }, bytes: { N: '1' }, flows: { N: '1' } } }, + ], + }, }, ], }, @@ -50,9 +67,36 @@ describe('loadTrafficObservations', () => { packets: 12, bytes: 4096, lastSeen: '2026-10-19T02:00:00Z', + endpoints: [ + { + address: '203.0.113.7', + bytes: 2048, + flows: 2, + lastSeen: '2026-10-19T01:30:00Z', + }, + { address: '198.51.100.9', bytes: 1, flows: 1 }, + ], }); }); + it('treats a malformed endpoint list as no citations', async () => { + const client = clientReturning({ + Responses: { + TrafficObservations: [ + { + interface_id: { S: 'eni-1' }, + packets: { N: '1' }, + bytes: { N: '1' }, + endpoints: { L: [{ M: { flows: { N: '2' } } }, { S: 'junk' }] }, + }, + ], + }, + }); + + const observations = await loadTrafficObservations(client, 'TrafficObservations', ['eni-1']); + expect(observations.get('eni-1')?.endpoints).toBeUndefined(); + }); + it('caps the lookups per run', async () => { const ids = Array.from({ length: 700 }, (_, index) => `eni-${index}`); const client = clientReturning( @@ -94,4 +138,63 @@ describe('instanceObservation', () => { expect(instanceObservation(['eni-9', undefined], observations)).toBeUndefined(); expect(instanceObservation([], observations)).toBeUndefined(); }); + + it('merges endpoints across interfaces, biggest senders first, capped', () => { + const withEndpoints = new Map([ + [ + 'eni-1', + { + interfaceId: 'eni-1', + packets: 1, + bytes: 2, + endpoints: [ + { address: '203.0.113.7', bytes: 1024, flows: 2 }, + { address: '198.51.100.9', bytes: 512, flows: 1 }, + ], + }, + ], + [ + 'eni-2', + { + interfaceId: 'eni-2', + packets: 1, + bytes: 2, + endpoints: [ + { address: '203.0.113.7', bytes: 4096, flows: 8 }, + { address: '192.0.2.10', bytes: 256, flows: 1 }, + ], + }, + ], + ]); + + const observation = instanceObservation(['eni-1', 'eni-2'], withEndpoints); + + expect(observation?.endpoints).toEqual([ + { address: '203.0.113.7', bytes: 4096, flows: 8 }, + { address: '198.51.100.9', bytes: 512, flows: 1 }, + { address: '192.0.2.10', bytes: 256, flows: 1 }, + ]); + }); + + it('caps the endpoint citations per instance', () => { + const many = new Map([ + [ + 'eni-1', + { + interfaceId: 'eni-1', + packets: 1, + bytes: 1, + endpoints: Array.from({ length: 9 }, (_, index) => ({ + address: `203.0.113.${index}`, + bytes: 100 - index, + flows: 1, + })), + }, + ], + ]); + + const observation = instanceObservation(['eni-1'], many); + expect(observation?.endpoints).toHaveLength(MAX_INSTANCE_ENDPOINTS); + expect(observation?.endpoints?.[0].address).toBe('203.0.113.0'); + }); }); diff --git a/lambdas/collector/src/observations.ts b/lambdas/collector/src/observations.ts index 0524c54e..119a50bc 100644 --- a/lambdas/collector/src/observations.ts +++ b/lambdas/collector/src/observations.ts @@ -3,13 +3,26 @@ import { Logger } from '../../shared/types'; const logger = new Logger('traffic-observations'); +export interface ObservedEndpoint { + address: string; + bytes: number; + flows: number; + lastSeen?: string; +} + export interface TrafficObservation { interfaceId: string; packets: number; bytes: number; lastSeen?: string; + /** The top remote senders the flow-log analyzer aggregated, if it ran with endpoint evidence. */ + endpoints?: ObservedEndpoint[]; } +// Per instance, enough endpoints to cite a hop without bloating the vertex. +export const MAX_INSTANCE_ENDPOINTS = 5; +const MAX_ENDPOINTS_PER_OBSERVATION = 20; + // Per-run bound: chunk into 100-key requests and stop after this many so a // pathological account cannot turn one collection into thousands of reads. export const MAX_OBSERVATION_LOOKUPS = 500; @@ -48,6 +61,7 @@ export async function loadTrafficObservations( packets: Number(item.packets?.N ?? 0), bytes: Number(item.bytes?.N ?? 0), lastSeen: item.last_seen?.S, + ...(parseEndpoints(item.endpoints) ?? {}), }); } } catch (error) { @@ -61,15 +75,40 @@ export async function loadTrafficObservations( return observations; } +// The analyzer's endpoint list on an observation item, read defensively: a +// malformed list means no citations, never a failed collection. +function parseEndpoints(value: unknown): { endpoints?: ObservedEndpoint[] } | undefined { + const raw = (value as { L?: unknown[] } | undefined)?.L; + if (!Array.isArray(raw)) return undefined; + const endpoints: ObservedEndpoint[] = []; + for (const entry of raw.slice(0, MAX_ENDPOINTS_PER_OBSERVATION)) { + const map = (entry as { M?: Record }).M; + const address = map?.address?.S; + if (!address) continue; + const bytes = Number(map?.bytes?.N ?? 0); + const flows = Number(map?.flows?.N ?? 0); + endpoints.push({ + address, + bytes: Number.isFinite(bytes) ? bytes : 0, + flows: Number.isFinite(flows) ? flows : 0, + ...(map?.last_seen?.S ? { lastSeen: map.last_seen.S } : {}), + }); + } + return endpoints.length > 0 ? { endpoints } : undefined; +} + // One instance-level fact derived from its interfaces: traffic was observed, -// and when it was most recently seen. Reachability confidence then becomes -// observed AND configured (see pathLikelihood). +// when it was most recently seen, and which remote endpoints sent it — +// deduplicated across interfaces, biggest senders first, capped. Reachability +// confidence then becomes observed AND configured (see pathLikelihood), and an +// attack path through this instance can cite observed traffic on its hop. export function instanceObservation( interfaceIds: (string | undefined)[], observations: Map -): { trafficObserved: true; lastSeen?: string } | undefined { +): { trafficObserved: true; lastSeen?: string; endpoints?: ObservedEndpoint[] } | undefined { let latest: string | undefined; let observed = false; + const endpoints = new Map(); for (const interfaceId of interfaceIds) { if (!interfaceId) continue; const observation = observations.get(interfaceId); @@ -78,6 +117,18 @@ export function instanceObservation( if (observation.lastSeen && (!latest || observation.lastSeen > latest)) { latest = observation.lastSeen; } + for (const endpoint of observation.endpoints ?? []) { + const existing = endpoints.get(endpoint.address); + if (!existing || endpoint.bytes > existing.bytes) endpoints.set(endpoint.address, endpoint); + } } - return observed ? { trafficObserved: true, ...(latest ? { lastSeen: latest } : {}) } : undefined; + if (!observed) return undefined; + const top = [...endpoints.values()] + .sort((a, b) => b.bytes - a.bytes || a.address.localeCompare(b.address)) + .slice(0, MAX_INSTANCE_ENDPOINTS); + return { + trafficObserved: true, + ...(latest ? { lastSeen: latest } : {}), + ...(top.length > 0 ? { endpoints: top } : {}), + }; } diff --git a/lambdas/flow-log-analyzer/index.test.ts b/lambdas/flow-log-analyzer/index.test.ts index 5901e157..bc366cd6 100644 --- a/lambdas/flow-log-analyzer/index.test.ts +++ b/lambdas/flow-log-analyzer/index.test.ts @@ -4,6 +4,7 @@ import { observationItems, OBSERVATION_TTL_DAYS, } from './index'; +import type { TrafficEndpoint } from './src/queries'; const environment = { ATHENA_DATABASE: 'khalifa_cloudtrail_db', @@ -12,37 +13,86 @@ const environment = { TRAFFIC_OBSERVATIONS_TABLE: 'TrafficObservations', }; -function athena(rows: string[][]) { +function athena( + rows: string[][], + endpointRows: string[][] = [], + options: { endpointFails?: boolean } = {} +) { const send = jest.fn(); - send.mockImplementation(async (command: { constructor: { name: string } }) => { - if (command.constructor.name === 'StartQueryExecutionCommand') { - return { QueryExecutionId: 'exec-1' }; - } - if (command.constructor.name === 'GetQueryExecutionCommand') { - return { QueryExecution: { Status: { State: 'SUCCEEDED' } } }; + let started = 0; + send.mockImplementation( + async (command: { constructor: { name: string }; input: Record }) => { + const name = command.constructor.name; + if (name === 'StartQueryExecutionCommand') { + started += 1; + return { QueryExecutionId: `exec-${started}` }; + } + if (name === 'GetQueryExecutionCommand') { + if (options.endpointFails && command.input.QueryExecutionId === 'exec-2') { + return { + QueryExecution: { Status: { State: 'FAILED', StateChangeReason: 'endpoint scan' } }, + }; + } + return { QueryExecution: { Status: { State: 'SUCCEEDED' } } }; + } + if (command.input.QueryExecutionId === 'exec-2') { + const headers = [ + 'interface_id', + 'account_id', + 'region', + 'srcaddr', + 'packets', + 'bytes', + 'flows', + 'last_seen', + ]; + return { + ResultSet: { + Rows: [ + { Data: headers.map((VarCharValue) => ({ VarCharValue })) }, + ...endpointRows.map((row) => ({ + Data: row.map((VarCharValue) => ({ VarCharValue })), + })), + ], + }, + }; + } + const headers = [ + 'account_id', + 'region', + 'interface_id', + 'packets', + 'bytes', + 'flows', + 'first_seen', + 'last_seen', + ]; + return { + ResultSet: { + Rows: [ + { Data: headers.map((VarCharValue) => ({ VarCharValue })) }, + ...rows.map((row) => ({ Data: row.map((VarCharValue) => ({ VarCharValue })) })), + ], + }, + }; } - const headers = [ - 'account_id', - 'region', - 'interface_id', - 'packets', - 'bytes', - 'flows', - 'first_seen', - 'last_seen', - ]; - return { - ResultSet: { - Rows: [ - { Data: headers.map((VarCharValue) => ({ VarCharValue })) }, - ...rows.map((row) => ({ Data: row.map((VarCharValue) => ({ VarCharValue })) })), - ], - }, - }; - }); + ); return send; } +function endpointRow(interfaceId: string, address: string, bytes = '2048') { + return [ + interfaceId, + '111122223333', + 'us-west-2', + address, + '12', + bytes, + '2', + '2026-10-19T03:00:00Z', + ]; +} + function row(interfaceId: string) { return [ '111122223333', @@ -111,7 +161,14 @@ describe('analyzeFlowLogs', () => { environment ); - expect(result).toEqual({ observations: 2, truncated: false, queryExecutionId: 'exec-1' }); + expect(result).toEqual({ + observations: 2, + endpoints: 0, + truncated: false, + endpointsTruncated: false, + queryExecutionId: 'exec-1', + endpointsQueryExecutionId: 'exec-2', + }); expect(athenaSend.mock.calls[0][0].input.QueryString).toContain( 'khalifa_cloudtrail_db.vpc_flow_logs' ); @@ -171,3 +228,107 @@ describe('analyzeFlowLogs', () => { warn.mockRestore(); }); }); + +describe('endpoint citations', () => { + it('merges top remote senders into the observation items', async () => { + const dynamoSend = jest.fn().mockResolvedValue({}); + const result = await analyzeFlowLogs( + {}, + { + athenaClient: { + send: athena( + [row('eni-1'), row('eni-2')], + [endpointRow('eni-1', '203.0.113.7'), endpointRow('eni-1', '198.51.100.9', '1024')] + ), + } as any, + dynamoClient: { send: dynamoSend } as any, + sleep: jest.fn().mockResolvedValue(undefined), + }, + environment + ); + + expect(result.endpoints).toBe(2); + expect(result.endpointsTruncated).toBe(false); + const items = dynamoSend.mock.calls[0][0].input.RequestItems.TrafficObservations; + expect(items[0].PutRequest.Item.interface_id).toEqual({ S: 'eni-1' }); + expect(items[0].PutRequest.Item.endpoints.L).toEqual([ + { + M: { + address: { S: '203.0.113.7' }, + bytes: { N: '2048' }, + flows: { N: '2' }, + last_seen: { S: '2026-10-19T03:00:00Z' }, + }, + }, + { + M: { + address: { S: '198.51.100.9' }, + bytes: { N: '1024' }, + flows: { N: '2' }, + last_seen: { S: '2026-10-19T03:00:00Z' }, + }, + }, + ]); + expect(items[1].PutRequest.Item.endpoints).toBeUndefined(); + }); + + it('degrades to observations without citations when the endpoint query fails', async () => { + const dynamoSend = jest.fn().mockResolvedValue({}); + const warn = jest.spyOn(console, 'warn').mockImplementation(() => undefined); + + const result = await analyzeFlowLogs( + {}, + { + athenaClient: { + send: athena([row('eni-1')], [], { endpointFails: true }), + } as any, + dynamoClient: { send: dynamoSend } as any, + sleep: jest.fn().mockResolvedValue(undefined), + }, + environment + ); + + warn.mockRestore(); + expect(result.observations).toBe(1); + expect(result.endpoints).toBe(0); + expect(result.endpointsQueryExecutionId).toBeUndefined(); + const items = dynamoSend.mock.calls[0][0].input.RequestItems.TrafficObservations; + expect(items[0].PutRequest.Item.endpoints).toBeUndefined(); + }); + + it('caps citations per interface even when more rows arrive', () => { + const endpoints = new Map([ + [ + 'eni-1', + Array.from({ length: 5 }, (_, index) => ({ + accountId: '111122223333', + region: 'us-west-2', + interfaceId: 'eni-1', + remoteAddress: `203.0.113.${index}`, + packets: 1, + bytes: 1, + flows: 1, + lastSeen: '2026-10-19T03:00:00Z', + })), + ], + ]); + const items = observationItems( + [ + { + accountId: '111122223333', + region: 'us-west-2', + interfaceId: 'eni-1', + packets: 1, + bytes: 1, + flows: 1, + firstSeen: '2026-10-19T01:00:00Z', + lastSeen: '2026-10-19T02:00:00Z', + }, + ], + 7, + new Date('2026-10-19T06:00:00Z'), + endpoints + ); + expect((items[0].endpoints as { L: unknown[] }).L).toHaveLength(5); + }); +}); diff --git a/lambdas/flow-log-analyzer/index.ts b/lambdas/flow-log-analyzer/index.ts index afac7920..c6c22f17 100644 --- a/lambdas/flow-log-analyzer/index.ts +++ b/lambdas/flow-log-analyzer/index.ts @@ -9,11 +9,17 @@ import { marshall } from '@aws-sdk/util-dynamodb'; import type { AttributeValue } from '@aws-sdk/client-dynamodb'; import { Logger } from '../shared/types'; import { + buildEndpointQuery, buildTrafficQuery, DEFAULT_OBSERVATION_DAYS, + ENDPOINTS_PER_INTERFACE, MAX_OBSERVATIONS, + MAX_ENDPOINT_ROWS, + parseEndpointRow, parseTrafficRow, + validateEndpointsPerInterface, validateObservationDays, + type TrafficEndpoint, type TrafficObservation, } from './src/queries'; @@ -21,9 +27,14 @@ import { // written to a DynamoDB cache the collector reads when it collects instances. // Reachability confidence becomes observed AND configured: configured exposure // says where an attacker could reach, an accepted flow says where traffic -// actually went. Optional and cost-gated — the Glue table and the CronJob only -// exist when the operator points them at a flow-log bucket (suspend: true by -// default). +// actually went — and the second, bounded aggregation records *which remote +// endpoints* sent that traffic, so an attack path can cite observed traffic on +// a specific hop. Cost bounds: one extra Athena query over the same window, +// top-ENDPOINTS_PER_INTERFACE remote senders per interface, a parsed-row cap, +// and a failure in the endpoint query degrades to observations without +// endpoints rather than failing the run. Optional and cost-gated — the Glue +// table and the CronJob only exist when the operator points them at a +// flow-log bucket (suspend: true by default). const logger = new Logger('flow-log-analyzer'); const defaultAthenaClient = new AthenaClient({ region: process.env.AWS_REGION || 'us-east-1', @@ -49,6 +60,7 @@ interface AnalyzerConfig { athenaResultsLocation: string; trafficObservationsTable: string; observationDays: number; + endpointsPerInterface: number; } interface Row { @@ -65,8 +77,11 @@ interface AnalyzerDependencies { export interface FlowLogAnalysisResult { observations: number; + endpoints: number; truncated: boolean; + endpointsTruncated: boolean; queryExecutionId: string; + endpointsQueryExecutionId?: string; } export function normalizeAthenaResultsLocation(value: string | undefined): string { @@ -101,6 +116,9 @@ function loadConfig(environment: NodeJS.ProcessEnv): AnalyzerConfig { environment.FLOW_LOG_DAYS ?? DEFAULT_OBSERVATION_DAYS, 'FLOW_LOG_DAYS' ), + endpointsPerInterface: validateEndpointsPerInterface( + Number(environment.ENDPOINTS_PER_INTERFACE ?? ENDPOINTS_PER_INTERFACE) + ), }; } @@ -139,16 +157,81 @@ export async function analyzeFlowLogs( } const observations = parsed.slice(0, MAX_OBSERVATIONS); + const endpointEvidence = await collectEndpointEvidence(config, days, dependencies); + await writeObservations( observations, days, config.trafficObservationsTable, dependencies.dynamoClient, dependencies.sleep, - dependencies.now() + dependencies.now(), + endpointEvidence.endpoints + ); + logger.info( + `Wrote ${observations.length} traffic observations with ${endpointEvidence.count} endpoint citations` ); - logger.info(`Wrote ${observations.length} traffic observations`); - return { observations: observations.length, truncated, queryExecutionId }; + return { + observations: observations.length, + endpoints: endpointEvidence.count, + truncated, + endpointsTruncated: endpointEvidence.truncated, + queryExecutionId, + ...(endpointEvidence.queryExecutionId + ? { endpointsQueryExecutionId: endpointEvidence.queryExecutionId } + : {}), + }; +} + +interface EndpointEvidence { + endpoints: Map; + count: number; + truncated: boolean; + queryExecutionId?: string; +} + +// Bounded and failure-tolerant: the endpoint query is the second Athena scan +// of the run, and when it fails the observations are still written without +// citations — the hop evidence is an enrichment, never a dependency. +async function collectEndpointEvidence( + config: AnalyzerConfig, + days: number, + dependencies: AnalyzerDependencies +): Promise { + const endpoints = new Map(); + try { + const queryExecutionId = await startQuery( + buildEndpointQuery(config.athenaDatabase, days, config.endpointsPerInterface), + config, + dependencies.athenaClient + ); + await waitForQuery( + queryExecutionId, + dependencies.athenaClient, + dependencies.sleep, + dependencies.signal + ); + const rows = await fetchRawRows(queryExecutionId, dependencies.athenaClient); + const parsed = rows.map((row, index) => parseEndpointRow(row.Data || [], index + 2)); + const truncated = parsed.length > MAX_ENDPOINT_ROWS; + if (truncated) { + logger.warn('Endpoint rows truncated', { limit: MAX_ENDPOINT_ROWS, rows: parsed.length }); + } + let count = 0; + for (const endpoint of parsed.slice(0, MAX_ENDPOINT_ROWS)) { + const list = endpoints.get(endpoint.interfaceId) ?? []; + if (list.length >= config.endpointsPerInterface) continue; + list.push(endpoint); + endpoints.set(endpoint.interfaceId, list); + count += 1; + } + return { endpoints, count, truncated, queryExecutionId }; + } catch (error) { + logger.warn('Endpoint aggregation unavailable; writing observations without citations', { + error: error instanceof Error ? error.message : String(error), + }); + return { endpoints, count: 0, truncated: false }; + } } export const handler = async (event: AnalyzerEvent = {}): Promise => @@ -157,25 +240,32 @@ export const handler = async (event: AnalyzerEvent = {}): Promise ): Record[] { const nowSeconds = Math.floor(observedAt.getTime() / 1000); - return observations.map( - (observation) => - marshall({ - interface_id: observation.interfaceId, - account_id: observation.accountId, - region: observation.region, - packets: observation.packets, - bytes: observation.bytes, - flows: observation.flows, - first_seen: observation.firstSeen, - last_seen: observation.lastSeen, - window_days: windowDays, - observed_at: observedAt.toISOString(), - expires_at: nowSeconds + OBSERVATION_TTL_DAYS * 86_400, - }) as Record - ); + return observations.map((observation) => { + const citations = (endpoints?.get(observation.interfaceId) ?? []).map((endpoint) => ({ + address: endpoint.remoteAddress, + bytes: endpoint.bytes, + flows: endpoint.flows, + last_seen: endpoint.lastSeen, + })); + return marshall({ + interface_id: observation.interfaceId, + account_id: observation.accountId, + region: observation.region, + packets: observation.packets, + bytes: observation.bytes, + flows: observation.flows, + first_seen: observation.firstSeen, + last_seen: observation.lastSeen, + window_days: windowDays, + observed_at: observedAt.toISOString(), + expires_at: nowSeconds + OBSERVATION_TTL_DAYS * 86_400, + ...(citations.length > 0 ? { endpoints: citations } : {}), + }) as Record; + }); } async function writeObservations( @@ -184,9 +274,10 @@ async function writeObservations( tableName: string, dynamoClient: DynamoDBClient, sleep: (milliseconds: number) => Promise, - observedAt: Date + observedAt: Date, + endpoints?: Map ): Promise { - const items = observationItems(observations, windowDays, observedAt); + const items = observationItems(observations, windowDays, observedAt, endpoints); for (let index = 0; index < items.length; index += BATCH_SIZE) { const batch = items.slice(index, index + BATCH_SIZE); let attempt = 0; diff --git a/lambdas/flow-log-analyzer/src/queries.test.ts b/lambdas/flow-log-analyzer/src/queries.test.ts index c10700f0..a6bfb10d 100644 --- a/lambdas/flow-log-analyzer/src/queries.test.ts +++ b/lambdas/flow-log-analyzer/src/queries.test.ts @@ -1,7 +1,12 @@ import { + buildEndpointQuery, buildTrafficQuery, + ENDPOINTS_PER_INTERFACE, MAX_OBSERVATION_DAYS, + MAX_ENDPOINT_ROWS, + parseEndpointRow, parseTrafficRow, + validateEndpointsPerInterface, validateObservationDays, } from './queries'; @@ -84,3 +89,76 @@ describe('parseTrafficRow', () => { expect(() => parseTrafficRow(values, 3)).toThrow('Invalid Athena traffic row 3'); }); }); + +describe('buildEndpointQuery', () => { + it('ranks accepted remote senders per interface with the window and cap bounded', () => { + const sql = buildEndpointQuery(DATABASE, 7); + + expect(sql).toContain(`${DATABASE}.vpc_flow_logs`); + expect(sql).toContain("action = 'ACCEPT'"); + expect(sql).toContain('SUM(bytes)'); + expect(sql).toContain('ROW_NUMBER() OVER (PARTITION BY interface_id ORDER BY SUM(bytes) DESC'); + expect(sql).toContain(`endpoint_rank <= ${ENDPOINTS_PER_INTERFACE}`); + expect(sql).toContain("log_date >= date_format(current_date - interval '7' day, '%Y/%m/%d')"); + expect(sql).toContain("srcaddr <> '-'"); + expect(sql).toContain('GROUP BY account_id, region, interface_id, srcaddr'); + expect(sql).not.toContain('${'); + }); + + it('validates the per-interface cap and database identifier', () => { + expect(() => validateEndpointsPerInterface(0)).toThrow( + 'ENDPOINTS_PER_INTERFACE must be an integer between 1 and 20' + ); + expect(() => validateEndpointsPerInterface(21)).toThrow( + 'ENDPOINTS_PER_INTERFACE must be an integer between 1 and 20' + ); + expect(() => buildEndpointQuery('db; DROP TABLE logs', 7)).toThrow( + 'ATHENA_DATABASE must be an unquoted Athena identifier' + ); + }); +}); + +describe('parseEndpointRow', () => { + const data = (values: string[]) => values.map((VarCharValue) => ({ VarCharValue })); + + it('parses a well-formed row', () => { + expect( + parseEndpointRow( + data([ + 'eni-0abc', + '111122223333', + 'us-west-2', + '203.0.113.7', + '42', + '4096', + '2', + '2026-10-19T04:00:00Z', + ]), + 2 + ) + ).toEqual({ + interfaceId: 'eni-0abc', + accountId: '111122223333', + region: 'us-west-2', + remoteAddress: '203.0.113.7', + packets: 42, + bytes: 4096, + flows: 2, + lastSeen: '2026-10-19T04:00:00Z', + }); + }); + + it.each([ + [['eni-1']], + [['eni-1', '111122223333', 'us-west-2', 'not-an-ip', '1', '1', '1', '2026-10-19T04:00:00Z']], + [['eni-1', '123', 'us-west-2', '203.0.113.7', '1', '1', '1', '2026-10-19T04:00:00Z']], + ])('rejects malformed row %#', (values) => { + expect(() => parseEndpointRow(data(values as string[]), 3)).toThrow( + /Invalid Athena endpoint row/ + ); + }); + + it('bounds what the analyzer will parse', () => { + expect(MAX_ENDPOINT_ROWS).toBeGreaterThan(MAX_OBSERVATION_DAYS); + }); +}); diff --git a/lambdas/flow-log-analyzer/src/queries.ts b/lambdas/flow-log-analyzer/src/queries.ts index 14e6ebc5..76182742 100644 --- a/lambdas/flow-log-analyzer/src/queries.ts +++ b/lambdas/flow-log-analyzer/src/queries.ts @@ -109,3 +109,92 @@ function timestamp(value: string | undefined, name: string, rowNumber: number): } return normalized; } + +export interface TrafficEndpoint { + accountId: string; + region: string; + interfaceId: string; + remoteAddress: string; + packets: number; + bytes: number; + flows: number; + lastSeen: string; +} + +// Cost bounds for the endpoint aggregation: one extra Athena query over the +// same window, top-N remote endpoints per interface (a window-function rank), +// and a row cap on what is parsed. The write stays one attribute list per +// interface, so the DynamoDB item size grows by a few hundred bytes at most. +export const ENDPOINTS_PER_INTERFACE = 5; +export const MAX_ENDPOINT_ROWS = 25_000; + +export function validateEndpointsPerInterface(value: number): number { + if (!Number.isSafeInteger(value) || value < 1 || value > 20) { + throw new Error('ENDPOINTS_PER_INTERFACE must be an integer between 1 and 20'); + } + return value; +} + +// Which remote endpoints (srcaddr — the ingress source) actually sent accepted +// traffic to each interface, ranked by bytes and capped per interface. The +// endpoint evidence lets an attack path cite observed traffic on a hop, not +// just "this instance saw traffic somewhere". +export function buildEndpointQuery( + database: string, + days: number, + perInterface: number = ENDPOINTS_PER_INTERFACE +): string { + assertAthenaDatabase(database); + const window = validateObservationDays(days); + const rank = validateEndpointsPerInterface(perInterface); + return ` + SELECT interface_id, account_id, region, srcaddr, packets, bytes, flows, last_seen + FROM ( + SELECT + account_id, + region, + interface_id, + srcaddr, + SUM(packets) AS packets, + SUM(bytes) AS bytes, + COUNT(*) AS flows, + date_format(from_unixtime(MAX(end)), '%Y-%m-%dT%H:%i:%sZ') AS last_seen, + ROW_NUMBER() OVER (PARTITION BY interface_id ORDER BY SUM(bytes) DESC, srcaddr ASC) AS endpoint_rank + FROM ${database}.vpc_flow_logs + WHERE log_date >= date_format(current_date - interval '${window}' day, '%Y/%m/%d') + AND action = 'ACCEPT' + AND interface_id IS NOT NULL + AND interface_id <> '-' + AND srcaddr IS NOT NULL + AND srcaddr <> '-' + AND from_unixtime(start) >= current_timestamp - interval '${window}' day + GROUP BY account_id, region, interface_id, srcaddr + ) + WHERE endpoint_rank <= ${rank} + ORDER BY interface_id, endpoint_rank + `; +} + +export function parseEndpointRow( + data: { VarCharValue?: string }[], + rowNumber: number +): TrafficEndpoint { + if (!data || data.length !== 8) { + throw new Error(`Invalid Athena endpoint row ${rowNumber}: expected 8 columns`); + } + const interfaceId = required(data[0].VarCharValue, 'interface_id', rowNumber); + const accountId = required(data[1].VarCharValue, 'account_id', rowNumber); + const region = required(data[2].VarCharValue, 'region', rowNumber); + const remoteAddress = required(data[3].VarCharValue, 'srcaddr', rowNumber); + if (!/^[0-9A-Fa-f.:]{3,45}$/.test(remoteAddress)) { + throw new Error(`Invalid Athena endpoint row ${rowNumber}: srcaddr must be an IP address`); + } + const packets = safeCount(data[4].VarCharValue, 'packets', rowNumber); + const bytes = safeCount(data[5].VarCharValue, 'bytes', rowNumber); + const flows = safeCount(data[6].VarCharValue, 'flows', rowNumber); + const lastSeen = timestamp(data[7].VarCharValue, 'last_seen', rowNumber); + if (!/^\d{12}$/.test(accountId)) { + throw new Error(`Invalid Athena endpoint row ${rowNumber}: account_id must be 12 digits`); + } + return { accountId, region, interfaceId, remoteAddress, packets, bytes, flows, lastSeen }; +} diff --git a/package-lock.json b/package-lock.json index 1d514bc0..39193416 100644 --- a/package-lock.json +++ b/package-lock.json @@ -16,10 +16,12 @@ ], "devDependencies": { "@types/jest": "^29.5.14", + "@types/js-yaml": "^4.0.9", "esbuild": "0.28.2", "eslint": "^8.56.0", "eslint-plugin-prettier": "^5.5.6", "jest": "^29.7.0", + "js-yaml": "^4.3.2", "prettier": "^3.2.5", "ts-jest": "^29.4.11", "tsx": "^4.19.0", @@ -4543,6 +4545,13 @@ "pretty-format": "^29.0.0" } }, + "node_modules/@types/js-yaml": { + "version": "4.0.9", + "resolved": "https://registry.npmjs.org/@types/js-yaml/-/js-yaml-4.0.9.tgz", + "integrity": "sha512-k4MGaQl5TGo/iipqb2UDG2UwjXziSWkh0uysQelTlJpX1qGlpUZYm8PnO4DxG1qBomtJUdYJ6qR6xdIah10JLg==", + "dev": true, + "license": "MIT" + }, "node_modules/@types/jsdom": { "version": "20.0.1", "resolved": "https://registry.npmjs.org/@types/jsdom/-/jsdom-20.0.1.tgz", diff --git a/package.json b/package.json index 534268f3..1561353d 100644 --- a/package.json +++ b/package.json @@ -48,14 +48,16 @@ }, "devDependencies": { "@types/jest": "^29.5.14", + "@types/js-yaml": "^4.0.9", "esbuild": "0.28.2", "eslint": "^8.56.0", "eslint-plugin-prettier": "^5.5.6", "jest": "^29.7.0", + "js-yaml": "^4.3.2", "prettier": "^3.2.5", "ts-jest": "^29.4.11", - "typescript": "^5.3.2", - "tsx": "^4.19.0" + "tsx": "^4.19.0", + "typescript": "^5.3.2" }, "overrides": { "qs": "^6.16.0", diff --git a/packages/risk-engine/src/index.ts b/packages/risk-engine/src/index.ts index b937533f..8c60b153 100644 --- a/packages/risk-engine/src/index.ts +++ b/packages/risk-engine/src/index.ts @@ -51,6 +51,13 @@ export { export { globMatch } from './glob'; export { extractRulePath, type RulePath } from './rule-path'; export { pathLikelihood, type PathLikelihood, type PathLikelihoodFactors } from './likelihood'; +export { + MAX_CITED_ENDPOINTS, + observedTrafficFromPath, + parseObservedEndpoints, + type ObservedEndpoint, + type ObservedTrafficCitation, +} from './observedTraffic'; export { COVERAGE_AGING_HOURS, COVERAGE_FACTORS, diff --git a/packages/risk-engine/src/observedTraffic.test.ts b/packages/risk-engine/src/observedTraffic.test.ts new file mode 100644 index 00000000..233c1779 --- /dev/null +++ b/packages/risk-engine/src/observedTraffic.test.ts @@ -0,0 +1,66 @@ +import { observedTrafficFromPath, parseObservedEndpoints } from './observedTraffic'; +import type { GraphVertex } from './types'; + +const vertex = (id: string, properties: Record = {}): GraphVertex => + ({ id, label: 'Ec2Instance', properties }) as GraphVertex; + +describe('parseObservedEndpoints', () => { + it('parses a JSON string or array, keeping known fields', () => { + expect( + parseObservedEndpoints( + JSON.stringify([ + { address: '203.0.113.7', bytes: 2048, flows: 3, lastSeen: '2026-10-19T02:00:00Z' }, + { address: '198.51.100.9' }, + ]) + ) + ).toEqual([ + { address: '203.0.113.7', bytes: 2048, flows: 3, lastSeen: '2026-10-19T02:00:00Z' }, + { address: '198.51.100.9' }, + ]); + }); + + it('is empty for junk, oversized lists or entries without an address', () => { + expect(parseObservedEndpoints('not json')).toEqual([]); + expect(parseObservedEndpoints(undefined)).toEqual([]); + expect(parseObservedEndpoints(JSON.stringify([{ bytes: 1 }, { address: '' }]))).toEqual([]); + expect( + parseObservedEndpoints( + Array.from({ length: 9 }, (_, index) => ({ address: `203.0.113.${index}` })) + ) + ).toHaveLength(5); + }); +}); + +describe('observedTrafficFromPath', () => { + it('cites the hop that arrives at an observed vertex', () => { + const citations = observedTrafficFromPath([ + vertex('internet'), + vertex('i-1', { + traffic_observed: true, + observed_endpoints: JSON.stringify([{ address: '203.0.113.7', bytes: 2048 }]), + }), + vertex('bucket'), + ]); + + expect(citations).toEqual([ + { + hop: 1, + vertexId: 'i-1', + endpoints: [{ address: '203.0.113.7', bytes: 2048 }], + }, + ]); + }); + + it('cites a hop even without endpoint detail, and ignores the entry vertex', () => { + const citations = observedTrafficFromPath([ + vertex('internet', { traffic_observed: true }), + vertex('i-1', { traffic_observed: true }), + ]); + + expect(citations).toEqual([{ hop: 1, vertexId: 'i-1', endpoints: [] }]); + }); + + it('is empty for paths with no observed traffic', () => { + expect(observedTrafficFromPath([vertex('internet'), vertex('i-1')])).toEqual([]); + }); +}); diff --git a/packages/risk-engine/src/observedTraffic.ts b/packages/risk-engine/src/observedTraffic.ts new file mode 100644 index 00000000..12087b79 --- /dev/null +++ b/packages/risk-engine/src/observedTraffic.ts @@ -0,0 +1,68 @@ +import type { GraphVertex } from './types'; + +// Observed-traffic citations: the flow-log analyzer aggregates the remote +// endpoints that actually sent accepted traffic to an interface, and the +// collector writes the top senders onto the workload vertex +// (observed_endpoints, JSON, bounded). A path hop that arrives at such a +// vertex can cite the observation instead of only claiming configured +// exposure. +export interface ObservedEndpoint { + address: string; + bytes?: number; + flows?: number; + lastSeen?: string; +} + +export interface ObservedTrafficCitation { + /** 1-based hop number: the edge that arrives at this vertex. */ + hop: number; + vertexId: string; + endpoints: ObservedEndpoint[]; +} + +export const MAX_CITED_ENDPOINTS = 5; + +// Tolerant by construction: a malformed or oversized property means no +// citations, never a failed path. +export function parseObservedEndpoints(value: unknown): ObservedEndpoint[] { + let parsed: unknown = value; + if (typeof value === 'string') { + try { + parsed = JSON.parse(value); + } catch { + return []; + } + } + if (!Array.isArray(parsed)) return []; + const endpoints: ObservedEndpoint[] = []; + for (const entry of parsed.slice(0, MAX_CITED_ENDPOINTS)) { + if (!entry || typeof entry !== 'object') continue; + const record = entry as Record; + const address = typeof record.address === 'string' ? record.address : undefined; + if (!address) continue; + endpoints.push({ + address, + ...(typeof record.bytes === 'number' && Number.isFinite(record.bytes) + ? { bytes: record.bytes } + : {}), + ...(typeof record.flows === 'number' && Number.isFinite(record.flows) + ? { flows: record.flows } + : {}), + ...(typeof record.lastSeen === 'string' ? { lastSeen: record.lastSeen } : {}), + }); + } + return endpoints; +} + +// One citation per hop whose target vertex has observed traffic. The entry +// vertex (index 0) is not a hop: nothing connects to it inside the path. +export function observedTrafficFromPath(path: GraphVertex[]): ObservedTrafficCitation[] { + const citations: ObservedTrafficCitation[] = []; + for (let index = 1; index < path.length; index++) { + const vertex = path[index]; + if (vertex.properties?.traffic_observed !== true) continue; + const endpoints = parseObservedEndpoints(vertex.properties?.observed_endpoints); + citations.push({ hop: index, vertexId: vertex.id, endpoints }); + } + return citations; +} diff --git a/packages/risk-engine/src/runner.ts b/packages/risk-engine/src/runner.ts index e684b861..632c6265 100644 --- a/packages/risk-engine/src/runner.ts +++ b/packages/risk-engine/src/runner.ts @@ -17,6 +17,7 @@ import type { import { extractRulePath, type RulePath } from './rule-path'; import { coverageLevelForAge, type CoverageLevel } from './coverage'; import { pathLikelihood } from './likelihood'; +import { observedTrafficFromPath } from './observedTraffic'; import { pathIndicators, ThreatIndicatorStore, type ThreatIndicator } from './threat-intel'; import { techniquesForRule } from './attack-techniques'; @@ -453,6 +454,7 @@ export class RiskRuleRunner { const scoreResult = computeRiskScore(riskInput); const likelihood = pathLikelihood(riskInput, this.coverageLevel); const independentFinding = this.independentFindingFor(rule, resources); + const observedTraffic = observedTrafficFromPath(path.vertices); return { id: this.issueStore.generateIssueId(rule.id, identity), @@ -477,6 +479,7 @@ export class RiskRuleRunner { ...(threatIntel.length > 0 ? { threatIntelMatches: threatIntel } : {}), ...(independentFinding !== undefined ? { independentFinding } : {}), ...(this.coverageLevel ? { coverageLevel: this.coverageLevel } : {}), + ...(observedTraffic.length > 0 ? { observedTraffic } : {}), attackTechniques: techniquesForRule(rule.id), }, }; @@ -509,6 +512,7 @@ export class RiskRuleRunner { const riskInput = this.riskInputWithThreatIntel(path, threatIntel); const scoreResult = computeRiskScore(riskInput); const likelihood = pathLikelihood(riskInput, this.coverageLevel); + const observedTraffic = observedTrafficFromPath(path.vertices); const updatedAt = new Date().toISOString(); const refreshed: Issue = { @@ -531,6 +535,7 @@ export class RiskRuleRunner { matchCount, lastRefreshedAt: updatedAt, ...(this.coverageLevel ? { coverageLevel: this.coverageLevel } : {}), + ...(observedTraffic.length > 0 ? { observedTraffic } : {}), attackTechniques: techniquesForRule(rule.id), }, threatIntel, diff --git a/scripts/risk-plan/README.md b/scripts/risk-plan/README.md index c0cf7795..dd316068 100644 --- a/scripts/risk-plan/README.md +++ b/scripts/risk-plan/README.md @@ -9,6 +9,11 @@ terraform show -json plan.out > plan.json npm run risk:plan -- plan.json # fails (exit 1) on anything >= high npm run risk:plan -- plan.json --min-severity critical npm run risk:plan -- plan.json --json # machine-readable + +# IAM policy diff (outside Terraform too): +npm run risk:plan -- --policy after.json --policy-before before.json +# Helm values for the chart: +npm run risk:plan -- --helm-values values.yaml ``` Exit codes: `0` no finding at or above the minimum severity, `1` the gate @@ -33,6 +38,15 @@ collected, the real topology), so findings say what collection would add. The catalogue is local (`catalogue.ts`) so the script needs no built engine at runtime; `catalogue.test.ts` pins it to the engine's names and severities. +IAM policies are parsed with the same parser the CIEM evaluator runs +(`lambdas/policy-evaluator/policy-parser.ts`), and with a before document +(plan `change.before`, or `--policy-before` for standalone files) the checks +are a diff: a grant that already existed is not newly introduced. Statement +conditions, SCPs and permission boundaries are **not** evaluated; findings +whose statement has conditions say so. `--helm-values` reads the chart values +with the same YAML a `helm install` would; only ingress that decides network +reach is judged today. + Optional CI gate (a plan artifact has to come from the pipeline that runs `terraform plan`): @@ -42,4 +56,7 @@ Optional CI gate (a plan artifact has to come from the pipeline that runs terraform -chdir=env plan -out plan.out terraform -chdir=env show -json plan.out > plan.json npm run risk:plan -- plan.json --min-severity high + +- name: Risk-check the chart values + run: npm run risk:plan -- --helm-values deploy/helm/khalifa/values.yaml ``` diff --git a/scripts/risk-plan/analyzers.test.ts b/scripts/risk-plan/analyzers.test.ts index 218b3875..bfcf2dd4 100644 --- a/scripts/risk-plan/analyzers.test.ts +++ b/scripts/risk-plan/analyzers.test.ts @@ -2,7 +2,7 @@ import { analyzePlan } from './analyzers'; import type { PlannedResource } from './plan'; function res(type: string, address: string, after: Record): PlannedResource { - return { address, type, name: address, actions: ['create'], after }; + return { address, type, name: address, actions: ['create'], after, before: {} }; } const policy = (document: unknown) => JSON.stringify(document); @@ -91,6 +91,7 @@ describe('security group ingress', () => { name: 'gone', actions: ['delete'], after: {}, + before: {}, }, ]) ).toEqual([]); diff --git a/scripts/risk-plan/analyzers.ts b/scripts/risk-plan/analyzers.ts index ad3898d7..10d078e0 100644 --- a/scripts/risk-plan/analyzers.ts +++ b/scripts/risk-plan/analyzers.ts @@ -1,4 +1,5 @@ import { RULE_CATALOGUE, SEVERITY_RANK, type PlanSeverity } from './catalogue'; +import { policyFindingsForResource } from './iam-policies'; import type { PlannedResource } from './plan'; export interface PlanFinding { @@ -14,7 +15,7 @@ export interface PlanFinding { const WORLD_CIDRS = new Set(['0.0.0.0/0', '::/0']); -interface RawFinding { +export interface RawFinding { ruleId: string | null; address: string; detail: string; @@ -58,6 +59,10 @@ export function analyzePlan(resources: PlannedResource[]): PlanFinding[] { case 'aws_iam_role': raw.push(...roleTrustFindings(resource)); break; + case 'aws_iam_policy': + case 'aws_iam_role_policy': + raw.push(...policyFindingsForResource(resource)); + break; case 'aws_secretsmanager_secret_policy': raw.push(...secretPolicyFindings(resource)); break; @@ -80,6 +85,12 @@ export function analyzePlan(resources: PlannedResource[]): PlanFinding[] { } } + return finalizeFindings(raw); +} + +// One finding per resource and rule, with the details joined: a security group +// with three world-open rules is one thing to fix, not three. +export function finalizeFindings(raw: RawFinding[]): PlanFinding[] { const merged = new Map(); for (const finding of raw) { const key = `${finding.address}\0${finding.ruleId ?? finding.fallbackName}`; @@ -99,7 +110,11 @@ export function analyzePlan(resources: PlannedResource[]): PlanFinding[] { }); } - return [...merged.values()].sort( + return sortFindings([...merged.values()]); +} + +export function sortFindings(findings: PlanFinding[]): PlanFinding[] { + return [...findings].sort( (a, b) => SEVERITY_RANK[b.severity] - SEVERITY_RANK[a.severity] || a.address.localeCompare(b.address) ); diff --git a/scripts/risk-plan/helm-values.ts b/scripts/risk-plan/helm-values.ts new file mode 100644 index 00000000..8b4ce3b8 --- /dev/null +++ b/scripts/risk-plan/helm-values.ts @@ -0,0 +1,52 @@ +import { load } from 'js-yaml'; +import type { RawFinding } from './analyzers'; + +// Shift-left for Helm values: the values file decides how the chart's API/UI +// reach the network, and the one value that changes that materially is +// ingress. An internet-facing scheme (however the controller spells it in its +// annotations) is the deploy-time precursor of the public-endpoint exposure +// the live rules would flag; a plain ingress is a lesser note because the +// controller's default decides. +export class HelmValuesParseError extends Error {} + +export function helmValuesFindings(valuesText: string): RawFinding[] { + let values: unknown; + try { + values = load(valuesText); + } catch { + throw new HelmValuesParseError('--helm-values must be a valid YAML values file'); + } + if (!values || typeof values !== 'object') return []; + + const ingress = (values as { ingress?: unknown }).ingress; + if (!ingress || typeof ingress !== 'object') return []; + const block = ingress as { enabled?: unknown; annotations?: unknown }; + if (block.enabled !== true) return []; + + const annotations = JSON.stringify(block.annotations ?? {}); + const internetFacing = annotations.includes('internet-facing'); + if (internetFacing) { + return [ + { + ruleId: null, + fallbackName: 'Internet-Facing Ingress Declared for the API/UI', + fallbackSeverity: 'high', + address: 'helm:ingress', + detail: + 'ingress annotations request an internet-facing load balancer for the API/UI; authentication (Cognito, Access) still applies, but the endpoint becomes reachable from the internet', + predictedPath: 'Internet → ingress → khalifa-api / khalifa-ui', + }, + ]; + } + return [ + { + ruleId: null, + fallbackName: 'Ingress Declared for the API/UI', + fallbackSeverity: 'medium', + address: 'helm:ingress', + detail: + "ingress is enabled; whether it is internet-facing depends on your controller's default — set an explicit private scheme if the API/UI should stay internal", + predictedPath: 'Ingress → khalifa-api / khalifa-ui', + }, + ]; +} diff --git a/scripts/risk-plan/iam-policies.ts b/scripts/risk-plan/iam-policies.ts new file mode 100644 index 00000000..88957d35 --- /dev/null +++ b/scripts/risk-plan/iam-policies.ts @@ -0,0 +1,117 @@ +import { + isActionMatched, + parsePolicyDocument, + type PolicyDocument, +} from '../../lambdas/policy-evaluator/policy-parser'; +import type { ParsedStatement } from '../../lambdas/policy-evaluator/types'; +import type { RawFinding } from './analyzers'; +import type { PlannedResource } from './plan'; + +// Shift-left for IAM policies: a policy document in a plan (aws_iam_policy / +// aws_iam_role_policy) or a standalone before/after pair is parsed with the +// same policy parser the CIEM evaluator runs, and grants that the catalogue's +// rules would eventually fire on are reported before they deploy. With a +// before document the check is a diff: a grant that already existed is not +// newly introduced. Conditions (iam:PassedToService, org boundaries, SCPs) +// are NOT evaluated — findings say so. + +export class PolicyParseError extends Error {} + +const PASS_ROLE = 'iam:PassRole'; + +function safeParse(documentJson: string): PolicyDocument | undefined { + try { + return parsePolicyDocument(documentJson); + } catch { + return undefined; + } +} + +function allowsAction(statement: ParsedStatement, action: string): boolean { + return ( + statement.effect === 'Allow' && + statement.actions.some((pattern) => isActionMatched(pattern, action)) + ); +} + +function allowsOnAllResources(statement: ParsedStatement, action: string): boolean { + return allowsAction(statement, action) && statement.resources.includes('*'); +} + +function grantDetail(statement: ParsedStatement, base: string): string { + const conditions = statement.conditions && Object.keys(statement.conditions).length > 0; + return conditions ? `${base} (statement conditions are not evaluated by plan analysis)` : base; +} + +function wildcardActionStatement(document: PolicyDocument): ParsedStatement | undefined { + return document.Statement.find( + (statement) => statement.effect === 'Allow' && statement.actions.includes('*') + ); +} + +// The analyzer core, shared by plan-embedded documents and standalone files. +export function policyDocumentFindings( + afterJson: string, + beforeJson: string | undefined, + address: string +): RawFinding[] { + const after = safeParse(afterJson); + if (!after) return []; + const before = beforeJson ? safeParse(beforeJson) : undefined; + const findings: RawFinding[] = []; + + const passRole = after.Statement.find((statement) => allowsOnAllResources(statement, PASS_ROLE)); + const passRoleBefore = before?.Statement.some((statement) => + allowsOnAllResources(statement, PASS_ROLE) + ); + if (passRole && !passRoleBefore) { + findings.push({ + ruleId: null, + fallbackName: 'IAM Policy Grants iam:PassRole on All Resources', + fallbackSeverity: 'critical', + address, + detail: grantDetail( + passRole, + 'allows iam:PassRole on Resource "*" — whoever can use this policy can hand any role, admin ones included, to compute they launch (RULE-034 fires once an admin role is in the reachable path)' + ), + predictedPath: 'Identity → iam:PassRole(*) → compute it launches → role', + }); + } + + const wildcard = wildcardActionStatement(after); + const wildcardBefore = before ? wildcardActionStatement(before) !== undefined : false; + if (wildcard && !wildcardBefore) { + findings.push({ + ruleId: null, + fallbackName: 'IAM Policy Grants Wildcard Actions', + fallbackSeverity: 'high', + address, + detail: grantDetail(wildcard, 'grants Action "*": every API at once, including IAM changes'), + predictedPath: 'Identity → * actions → any collected resource', + }); + } + + return findings; +} + +// Plan-embedded policies: `after.policy` is a JSON string on aws_iam_policy +// and aws_iam_role_policy; before gives true diff semantics on updates. +export function policyFindingsForResource(resource: PlannedResource): RawFinding[] { + const after = resource.after.policy; + if (typeof after !== 'string') return []; + const before = typeof resource.before.policy === 'string' ? resource.before.policy : undefined; + return policyDocumentFindings(after, before, resource.address); +} + +// Standalone files (for flows outside Terraform, e.g. a diff of +// `aws iam get-policy-version` before and after). A malformed document is the +// operator's error, not a silent skip. +export function standalonePolicyFindings(afterText: string, beforeText?: string): RawFinding[] { + if (safeParse(afterText) === undefined) { + throw new PolicyParseError('--policy must be a valid IAM policy document (JSON)'); + } + if (beforeText !== undefined && safeParse(beforeText) === undefined) { + throw new PolicyParseError('--policy-before must be a valid IAM policy document (JSON)'); + } + return policyDocumentFindings(afterText, beforeText, 'policy'); +} diff --git a/scripts/risk-plan/index.test.ts b/scripts/risk-plan/index.test.ts index 7a0dc08e..ad543ab5 100644 --- a/scripts/risk-plan/index.test.ts +++ b/scripts/risk-plan/index.test.ts @@ -89,10 +89,13 @@ describe('parseArgs', () => { }); it.each([ - [[], 'A plan file path is required'], + [[], 'A plan file, --policy or --helm-values input is required'], [['--min-severity', 'potato', 'plan.json'], '--min-severity must be'], [['--verbose', 'plan.json'], 'Unknown option: --verbose'], [['a.json', 'b.json'], 'Unexpected argument: b.json'], + [['--policy'], '--policy needs a file path'], + [['--policy-before', 'before.json'], '--policy-before only applies together with --policy'], + [['--helm-values'], '--helm-values needs a file path'], ])('rejects %j', (argv, message) => { const parsed = parseArgs(argv as string[]); expect(parsed.error).toContain(message); @@ -129,3 +132,98 @@ describe('main', () => { expect(err.join('\n')).toContain('not valid JSON'); }); }); + +describe('shift-left inputs', () => { + const passRolePolicy = JSON.stringify({ + Version: '2012-10-17', + Statement: [ + { Effect: 'Allow', Action: ['iam:PassRole'], Resource: '*' }, + { Effect: 'Allow', Action: ['s3:GetObject'], Resource: 'arn:aws:s3:::data/*' }, + ], + }); + const scopedBefore = JSON.stringify({ + Version: '2012-10-17', + Statement: [ + { + Effect: 'Allow', + Action: ['iam:PassRole'], + Resource: 'arn:aws:iam::111122223333:role/app', + }, + ], + }); + + it('fails the gate on an IAM policy diff that newly grants iam:PassRole on *', () => { + const result = runRiskPlan({ policyText: passRolePolicy, policyBeforeText: scopedBefore }); + + expect(result.exitCode).toBe(1); + expect(result.output).toContain('IAM Policy Grants iam:PassRole on All Resources'); + expect(result.output).toContain('Identity → iam:PassRole(*) → compute it launches → role'); + expect(result.report.failing).toHaveLength(1); + }); + + it('reports nothing new when the grant already existed', () => { + const alreadyWide = JSON.stringify({ + Version: '2012-10-17', + Statement: [{ Effect: 'Allow', Action: ['iam:PassRole'], Resource: '*' }], + }); + + const result = runRiskPlan({ policyText: passRolePolicy, policyBeforeText: alreadyWide }); + expect(result.exitCode).toBe(0); + expect(result.report.findings).toEqual([]); + }); + + it('scans policies embedded in a plan, diffing before against after', () => { + const result = runRiskPlan({ + planText: planText([ + change('aws_iam_policy.wide', 'aws_iam_policy', { policy: passRolePolicy }), + ]), + }); + + expect(result.exitCode).toBe(1); + expect(result.output).toContain('aws_iam_policy.wide'); + + const diffed = runRiskPlan({ + planText: JSON.stringify({ + resource_changes: [ + { + address: 'aws_iam_policy.wide', + mode: 'managed', + type: 'aws_iam_policy', + name: 'wide', + change: { + actions: ['update'], + before: { policy: passRolePolicy }, + after: { policy: passRolePolicy }, + }, + }, + ], + }), + }); + expect(diffed.report.findings).toEqual([]); + }); + + it('flags an internet-facing Helm ingress and lets a plain one pass a high gate', () => { + const flagged = runRiskPlan({ + helmValuesText: + 'ingress:\n enabled: true\n annotations:\n alb.ingress.kubernetes.io/scheme: internet-facing\n', + }); + expect(flagged.exitCode).toBe(1); + expect(flagged.output).toContain('Internet-Facing Ingress Declared for the API/UI'); + + const note = runRiskPlan({ helmValuesText: 'ingress:\n enabled: true\n' }); + expect(note.exitCode).toBe(0); + expect(note.output).toContain('Ingress Declared for the API/UI'); + + const disabled = runRiskPlan({ helmValuesText: 'ingress:\n enabled: false\n' }); + expect(disabled.report.findings).toEqual([]); + }); + + it('treats a malformed policy or values file as a usage error', () => { + expect(() => runRiskPlan({ policyText: 'not json' })).toThrow( + '--policy must be a valid IAM policy document (JSON)' + ); + expect(() => runRiskPlan({ helmValuesText: ':::: not yaml ::::' })).toThrow( + '--helm-values must be a valid YAML values file' + ); + }); +}); diff --git a/scripts/risk-plan/index.ts b/scripts/risk-plan/index.ts index bf5b2d09..0e7dd15d 100644 --- a/scripts/risk-plan/index.ts +++ b/scripts/risk-plan/index.ts @@ -1,6 +1,8 @@ import { readFileSync } from 'node:fs'; -import { analyzePlan } from './analyzers'; +import { analyzePlan, finalizeFindings, sortFindings, type PlanFinding } from './analyzers'; import { isSeverity, type PlanSeverity } from './catalogue'; +import { HelmValuesParseError, helmValuesFindings } from './helm-values'; +import { PolicyParseError, standalonePolicyFindings } from './iam-policies'; import { PlanParseError, parsePlan } from './plan'; import { buildReport, formatReport, reportJson, type PlanReport } from './report'; @@ -12,17 +14,37 @@ export interface RiskPlanResult { } export interface RiskPlanOptions { - planText: string; + planText?: string; + /** A standalone IAM policy document (the proposed `after`), for flows outside Terraform. */ + policyText?: string; + /** The policy document as it exists today: with it, only newly granted risk is reported. */ + policyBeforeText?: string; + /** A Helm values file for the chart: ingress that reaches the internet is flagged. */ + helmValuesText?: string; minSeverity?: PlanSeverity; json?: boolean; } -// The CI gate's core: parse a `terraform show -json` document, predict which -// catalogue rules the planned changes would feed, and fail when any reach the -// minimum severity. Read-only — nothing here touches AWS or the graph. +// The CI gate's core: parse a `terraform show -json` document and/or standalone +// IAM policy / Helm values inputs, predict which catalogue rules the changes +// would feed (IAM policies are parsed with the same parser the CIEM evaluator +// runs; with a before-policy the checks are a diff), and fail when any reach +// the minimum severity. Read-only — nothing here touches AWS or the graph. export function runRiskPlan(options: RiskPlanOptions): RiskPlanResult { const minSeverity = options.minSeverity ?? 'high'; - const report = buildReport(analyzePlan(parsePlan(options.planText)), minSeverity); + const findings: PlanFinding[] = []; + if (options.planText !== undefined) { + findings.push(...analyzePlan(parsePlan(options.planText))); + } + if (options.policyText !== undefined) { + findings.push( + ...finalizeFindings(standalonePolicyFindings(options.policyText, options.policyBeforeText)) + ); + } + if (options.helmValuesText !== undefined) { + findings.push(...finalizeFindings(helmValuesFindings(options.helmValuesText))); + } + const report = buildReport(sortFindings(findings), minSeverity); return { report, output: options.json ? reportJson(report) : formatReport(report), @@ -32,13 +54,18 @@ export function runRiskPlan(options: RiskPlanOptions): RiskPlanResult { export interface ParsedArgs { planPath?: string; + policyPath?: string; + policyBeforePath?: string; + helmValuesPath?: string; minSeverity: PlanSeverity; json: boolean; error?: string; } const USAGE = - 'Usage: npm run risk:plan -- [--min-severity low|medium|high|critical] [--json]\n' + + 'Usage: npm run risk:plan -- [--policy after.json] [--policy-before before.json]\n' + + ' [--helm-values values.yaml] [--min-severity low|medium|high|critical] [--json]\n' + + 'At least one of , --policy or --helm-values is required.\n' + 'Produce the plan file with: terraform plan -out plan.out && terraform show -json plan.out > plan.json'; export function parseArgs(argv: string[]): ParsedArgs { @@ -47,6 +74,23 @@ export function parseArgs(argv: string[]): ParsedArgs { const arg = argv[index]; if (arg === '--json') { parsed.json = true; + } else if (arg === '--policy') { + const value = argv[++index]; + if (!value || value.startsWith('-')) + return { ...parsed, error: '--policy needs a file path' }; + parsed.policyPath = value; + } else if (arg === '--policy-before') { + const value = argv[++index]; + if (!value || value.startsWith('-')) { + return { ...parsed, error: '--policy-before needs a file path' }; + } + parsed.policyBeforePath = value; + } else if (arg === '--helm-values') { + const value = argv[++index]; + if (!value || value.startsWith('-')) { + return { ...parsed, error: '--helm-values needs a file path' }; + } + parsed.helmValuesPath = value; } else if (arg === '--min-severity') { const value = argv[++index]; if (!value || !isSeverity(value)) { @@ -67,7 +111,12 @@ export function parseArgs(argv: string[]): ParsedArgs { return { ...parsed, error: `Unexpected argument: ${arg}` }; } } - if (!parsed.planPath) return { ...parsed, error: 'A plan file path is required' }; + if (parsed.policyBeforePath && !parsed.policyPath) { + return { ...parsed, error: '--policy-before only applies together with --policy' }; + } + if (!parsed.planPath && !parsed.policyPath && !parsed.helmValuesPath) { + return { ...parsed, error: 'A plan file, --policy or --helm-values input is required' }; + } return parsed; } @@ -87,14 +136,27 @@ export function main(argv: string[], io: MainIo): number { } try { const result = runRiskPlan({ - planText: io.readFile(parsed.planPath!, 'utf8'), + ...(parsed.planPath !== undefined ? { planText: io.readFile(parsed.planPath, 'utf8') } : {}), + ...(parsed.policyPath !== undefined + ? { policyText: io.readFile(parsed.policyPath, 'utf8') } + : {}), + ...(parsed.policyBeforePath !== undefined + ? { policyBeforeText: io.readFile(parsed.policyBeforePath, 'utf8') } + : {}), + ...(parsed.helmValuesPath !== undefined + ? { helmValuesText: io.readFile(parsed.helmValuesPath, 'utf8') } + : {}), minSeverity: parsed.minSeverity, json: parsed.json, }); io.out(result.output); return result.exitCode; } catch (error) { - if (error instanceof PlanParseError) { + if ( + error instanceof PlanParseError || + error instanceof PolicyParseError || + error instanceof HelmValuesParseError + ) { io.err(error.message); return 2; } diff --git a/scripts/risk-plan/plan.ts b/scripts/risk-plan/plan.ts index 8b5c5055..d06ec2f0 100644 --- a/scripts/risk-plan/plan.ts +++ b/scripts/risk-plan/plan.ts @@ -8,11 +8,14 @@ export interface PlannedResource { actions: string[]; /** Known `after` values; a key is absent when the plan cannot know it. */ after: Record; + /** Known `before` values, for diff semantics (an unchanged grant is not new). */ + before: Record; } interface ChangeShape { actions?: unknown; after?: unknown; + before?: unknown; after_unknown?: unknown; } @@ -57,6 +60,7 @@ export function parsePlan(text: string): PlannedResource[] { name: typeof change.name === 'string' ? change.name : change.address, actions, after: knownValues((change.change as ChangeShape | undefined)?.after), + before: knownValues((change.change as ChangeShape | undefined)?.before), }); }); return resources; diff --git a/ui/lib/attack-graph.test.ts b/ui/lib/attack-graph.test.ts index 84f15320..b10edc3f 100644 --- a/ui/lib/attack-graph.test.ts +++ b/ui/lib/attack-graph.test.ts @@ -45,6 +45,27 @@ const bucket = v('arn:aws:s3:::data', 'S3Bucket', { }); describe('riskFlags', () => { + it('cites observed traffic with the first remote sender when present', () => { + const observed = { + id: 'arn:aws:ec2:us-west-2:111122223333:instance/i-1', + label: 'Ec2Instance', + properties: { + traffic_observed: true, + observed_endpoints: JSON.stringify([{ address: '203.0.113.7', bytes: 2048, flows: 3 }]), + }, + } as GraphVertex; + expect(riskFlags(observed)).toEqual([ + { level: 'warn', text: 'Observed traffic from 203.0.113.7' }, + ]); + + const withoutDetail = { + id: 'arn:aws:ec2:us-west-2:111122223333:instance/i-2', + label: 'Ec2Instance', + properties: { traffic_observed: true, observed_endpoints: 'junk' }, + } as GraphVertex; + expect(riskFlags(withoutDetail)).toEqual([{ level: 'warn', text: 'Observed traffic' }]); + }); + it('flags only properties the collector records', () => { expect(riskFlags(ec2)).toEqual([{ level: 'risk', text: 'Internet exposed' }]); expect(riskFlags(v('b', 'S3Bucket', { data_classification: 'public' }))).toEqual([]); diff --git a/ui/lib/attack-graph.ts b/ui/lib/attack-graph.ts index d06a6411..12a39709 100644 --- a/ui/lib/attack-graph.ts +++ b/ui/lib/attack-graph.ts @@ -106,6 +106,22 @@ function stageOf(vertex: GraphVertex): Stage { return STAGE_BY_LABEL[vertex.label] ?? 'compute'; } +// The first remote sender the flow-log analyzer tagged on this vertex, for a +// citation chip; junk or absence is simply no address. +function firstObservedEndpoint(value: unknown): string | undefined { + let parsed: unknown = value; + if (typeof value === 'string') { + try { + parsed = JSON.parse(value); + } catch { + return undefined; + } + } + if (!Array.isArray(parsed)) return undefined; + const first = parsed[0] as { address?: unknown } | undefined; + return first && typeof first.address === 'string' && first.address ? first.address : undefined; +} + // Flags come only from properties the collector writes; nothing is inferred. export function riskFlags(vertex: GraphVertex): RiskFlag[] { const p = vertex.properties ?? {}; @@ -124,6 +140,10 @@ export function riskFlags(vertex: GraphVertex): RiskFlag[] { if (p.password_enabled === true && p.mfa_enabled === false) add('risk', 'Console without MFA'); if (typeof p.key_age_days === 'number' && p.key_age_days > 90) add('warn', 'Key older than 90 days'); + if (p.traffic_observed === true) { + const endpoint = firstObservedEndpoint(p.observed_endpoints); + add('warn', endpoint ? `Observed traffic from ${endpoint}` : 'Observed traffic'); + } if (p.data_classification === 'secret') add('risk', 'Secret data'); if (p.data_classification === 'restricted') add('risk', 'Restricted data'); if (p.crown_jewel === true) add('risk', 'Crown jewel'); diff --git a/ui/types/index.ts b/ui/types/index.ts index 60f7ea40..13e668d6 100644 --- a/ui/types/index.ts +++ b/ui/types/index.ts @@ -194,6 +194,20 @@ export interface CoverageStatement { note: string; } +export interface ObservedEndpoint { + address: string; + bytes?: number; + flows?: number; + lastSeen?: string; +} + +export interface ObservedTrafficCitation { + /** 1-based hop number: the edge that arrives at this vertex. */ + hop: number; + vertexId: string; + endpoints: ObservedEndpoint[]; +} + export interface ScoredPath { nodes: GraphVertex[]; edges: GraphEdge[]; @@ -202,6 +216,8 @@ export interface ScoredPath { /** Explainable 0..1 path likelihood (product of the factors). */ likelihood: number; likelihoodFactors: PathLikelihoodFactors; + /** Hops whose target workload actually saw accepted traffic (flow logs). */ + observedTraffic?: ObservedTrafficCitation[]; } export interface AttackPathSimulationEdge {