Cite observed traffic on hops and widen risk:plan's shift-left gate

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 <address>" 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

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 <address>" 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

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: [] }]);

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<vo
         const riskInput = riskInputFromPath(path.nodes);
         const { score, severity } = computeRiskScore(riskInput);
         const likelihood = pathLikelihood(riskInput, coverage.level);
+        const observedTraffic = observedTrafficFromPath(path.nodes);
         return {
           ...path,
           score,
           severity,
           likelihood: likelihood.value,
           likelihoodFactors: likelihood.factors,
+          ...(observedTraffic.length > 0 ? { observedTraffic } : {}),
         };
       })
       .sort((a, b) => b.score - a.score || b.likelihood - a.likelihood);

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);
+        }
       }
     }
   }

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');
+  });
 });

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<string, { S?: string; N?: string }> }).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<string, TrafficObservation>
-): { trafficObserved: true; lastSeen?: string } | undefined {
+): { trafficObserved: true; lastSeen?: string; endpoints?: ObservedEndpoint[] } | undefined {
   let latest: string | undefined;
   let observed = false;
+  const endpoints = new Map<string, ObservedEndpoint>();
   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 } : {}),
+  };
 }