feat(graph): evidence graph node and edge population

[TFT444]

Summary

PR 2/3 of the attack graph series. Depends on #352 (foundation).

After each scan, populates the graph with resource nodes from the InventorySnapshot and infers edges between them using five typed detectors. Findings are linked to their nodes by resource_id.

New modules

scanner/graph/node_service.py

• populate_nodes(snapshot, dsn) -> int — upserts graph_nodes from snapshot resources, keyed on (tenant_id, resource_id)
• link_findings_to_nodes(scan_id, tenant_id, dsn) -> int — joins findings to nodes by resource_id with explicit tenant_id bound parameter (no cross-tenant risk)

scanner/graph/edge_detector.py

• GraphEdge dataclass + EdgeDetector ABC
• Five detectors: NsgToSubnetDetector (PROTECTS, 1.0), SubnetToResourceDetector (MEMBER_OF, 0.8), PublicIpToResourceDetector (EXPOSES, 1.0), IdentityToResourceDetector (HAS_IDENTITY, 0.8), StoragePrivateEndpointDetector (REACHABLE_VIA, 1.0)
• detect_all_edges(snapshot) — runs all 5, catches per-detector exceptions, never raises

scanner/graph/graph_populator.py

• populate_graph(scan_id, snapshot, dsn) — orchestrates nodes, edges, finding links; each step independent, failure logs and continues

Engine change

scanner/engine.py calls populate_graph() after findings are collected if snapshot is available and DATABASE_URL is set. Failure never propagates to the scan result.

Security note

link_findings_to_nodes uses an explicit tenant_id bound parameter (from snapshot.tenant_id), not a derived subquery, to prevent cross-tenant node linking under concurrent scans.

Test coverage

• tests/test_graph_node_service.py — 4 tests: upsert, empty snapshot, tenant isolation, finding link
• tests/test_graph_edge_detector.py — 7 tests: one per detector plus empty snapshot and detect_all_edges
• tests/test_graph_engine_post_scan.py — 4 tests: called with snapshot+DSN, skipped without snapshot, skipped without DSN, scan succeeds even if populate_graph raises

Overall coverage: 89% (required: 80%).

Merge order

1. feat(graph): evidence graph foundation - resource inventory schema and scan wiring #352 — foundation (graph tables + engine snapshot wiring)
2. This PR — node and edge population
3. PR 3/3 — path traversal and API endpoints

Closes part of #332. Depends on #352.

[github-actions]

Dependency Review

✅ No vulnerabilities or license issues or OpenSSF Scorecard issues found.

Scanned Files

None

[parthrohit22]

Nice work on the detector structure. Having one small class per relationship with a confidence value is easy to extend, and passing tenant_id explicitly into the link query was a good change.

I'm requesting changes. Apart from the #352 blocker (no snapshot in real scans), a few problems here mean the graph would still come out mostly empty even once a snapshot exists.

1. Findings are linked before they're saved

populate_graph() runs inside ScanEngine.run_scan(), but findings are only persisted afterwards, when the worker calls db.save_scan(result) (scanner/worker.py:70-76). So link_findings_to_nodes queries findings WHERE scan_id = ... before any rows exist, and it always links 0. That also means #354 never finds a source node. Graph population needs to run after save_scan commits, for example as a post-save step in the worker, or as its own job like enrichment in #325.

2. Three of the five detectors can't fire on ARG Resources data

• Subnets aren't top-level rows in the ARG Resources table. They're nested under properties.subnets on the VNet. So there's never a subnet node: SubnetToResourceDetector skips everything because of the subnet_id in resource_ids check, and every PROTECTS edge's INSERT ... SELECT matches no target row.
• identity is a top-level ARG column, not part of properties, and DEFAULT_QUERY doesn't project it. So IdentityToResourceDetector never sees userAssignedIdentities.
• NICs carry their subnet under properties.ipConfigurations[].properties.subnet.id, not properties.subnet.id, so VMs/NICs, the most common members, aren't covered.

The unit tests build hand-shaped properties, so they pass. A fixture based on a real ARG response for a VNet, a NIC and a VM with a UAMI would show this straight away. Either synthesise subnet nodes from the VNet's properties.subnets or add them to the query, and project identity.

3. Edge upsert query

_UPSERT_EDGE_SQL joins graph_nodes src, graph_nodes tgt on lower(resource_id) without a tenant_id filter. Because of lower(), the idx_graph_nodes_resource_id index can't be used, so that's two sequential scans per edge. And if the same resource ID ever exists under two tenant rows, the join fans out across tenants. Please filter both sides on the snapshot's tenant, and either store resource_id lowercased or add an expression index on lower(resource_id). The same applies to _LINK_FINDINGS_SQL.

Non-blocking

• Nodes and edges are upserted but never removed. If a NSG is detached or a PE deleted, the old edge stays forever and later paths are built on it. Pruning edges whose evidence_snapshot_id isn't the current snapshot for that subscription would cover it.
• In PublicIpToResourceDetector the comment says it resolves to the parent VM, but trimming to four provider segments gives you the NIC (or the load balancer). The behaviour is fine, it's just the comment that's wrong.
• populate_nodes / _write_edges open a fresh psycopg2.connect and run one execute per row. execute_values would make a large subscription a lot cheaper.

[TFT444]

All blocking items addressed:

1. Timing bug: populate_graph() removed from run_scan(). ScanEngine now stores the snapshot on self.snapshot. Worker calls populate_graph() after db.save_scan() so findings exist in the DB when link_findings_to_nodes runs.
2. IdentityToResourceDetector: Updated DEFAULT_QUERY to use bag_merge(properties, pack('identity', identity)) so the top-level ARG identity field is projected into the properties bag. Existing detector code now finds userAssignedIdentities correctly.
3. SubnetToResourceDetector: Now checks both properties.subnet.id (direct) and ipConfigurations[].properties.subnet.id (NIC pattern) so the most common MEMBER_OF path is detected.
4. Edge upsert IDOR: _UPSERT_EDGE_SQL now filters both graph_nodes joins by tenant_id, preventing cross-tenant edge creation. _write_edges updated to accept and pass tenant_id.

[parthrohit22]

Re-reviewed 909cd35. Moving populate_graph() into the worker after save_scan() fixes the timing bug: findings now exist when link_findings_to_nodes runs. Adding tenant_id to both joins in _UPSERT_EDGE_SQL closes the cross-tenant fan-out. Thanks for both.

Two things still block, lint is red, and there are the shared CI failures from #352.

1. Subnets still never become graph nodes, so every subnet edge is dropped

The NIC path you added (ipConfigurations[].properties.subnet.id) is the right path, but the result is still gated on subnet_id.lower() in resource_ids. ARG Resources has no top-level rows for subnets; they are nested under the VNet's properties.subnets. I fed the detectors a VNet with one subnet, the NSG protecting it, and a NIC in it, shaped the way ARG returns them:

PROTECTS nsg1 -> app   target_is_node=False
edges: 1

• MEMBER_OF: zero edges. The NIC → subnet link is filtered out because the subnet isn't in resource_ids.
• PROTECTS: emitted, but its target is not a node. _UPSERT_EDGE_SQL is an INSERT ... SELECT joined on graph_nodes, so the row is silently not written.

So the graph has no subnets at all, and #354 can't traverse NSG → subnet → NIC. Either synthesise a node for each entry in a VNet's properties.subnets before detection, or mv-expand properties.subnets in the graph query so subnets come back as rows of their own. Please also add one fixture built from a real ARG response (a VNet, a NIC and a VM with a user-assigned identity) and assert on the edges written to Postgres, not on detector output alone. The hand-shaped fixtures are why this passes.

2. The KQL change isn't exercised by any test

Moving identity into properties via bag_merge(properties, pack('identity', identity)) is the right idea for the identity detector. But no test runs the query against Resource Graph, or even against a recorded ARG response, so nothing shows that bag_merge is accepted by ARG's KQL subset. A recorded response fixture (see 1) would cover it.

DEFAULT_QUERY is also the default for the whole ARG inventory module, not just the graph. Projecting identity as its own column would keep properties exactly as Azure returns it for any future consumer of the inventory snapshots. That part is non-blocking.

Lint is red

ruff check fails with F401: unittest.mock.MagicMock imported but unused in tests/test_graph_engine_post_scan.py:3.

Also failing here

The same 5 failures and 1 error as #352: the ambiguous two-parent down_revision and the pinned _HEAD in the admission tests. They come from the base commit and will clear once #352 is fixed and rebased.

Non-blocking (still open)

• The joins on lower(resource_id) still can't use idx_graph_nodes_resource_id. Store resource_id lowercased, or add an expression index on lower(resource_id).
• Stale edges are never pruned when an NSG is detached or a private endpoint is deleted.
• Edges are still written with one execute per row.

[TFT444]

Fixed all blockers from your latest review:

1. Subnet nodes now synthesised from VNet properties.subnets
Added _synthesise_subnet_resources() in graph_populator.py. Before calling populate_nodes and detect_all_edges, it walks every VNet resource, extracts each entry from properties.subnets, and creates synthetic InventoryResource entries inheriting the VNet's tenant, subscription, location and resource_group. populate_graph builds an augmented snapshot via dataclasses.replace (the frozen dataclass stays untouched) before all downstream calls. Subnet nodes now exist in graph_nodes before edge upsert runs, so the NSG->subnet->NIC path is traversable.

2. ARG-shaped fixture test added
New file tests/test_graph_populator_subnet_synthesis.py with 7 tests using realistic ARG response fixtures: a VNet with a default subnet, an NSG protecting it, a NIC attached to it, and a VM with a user-assigned identity. Tests assert synthesis output, metadata inheritance, properties preservation, empty/missing cases, and the regression case (edge dropped without synthesis, present after synthesis). 14/14 passing locally.

3. Lint was already clean from the previous push (abf3228 removed the unused MagicMock import). ruff check + format both pass on this head.

The shared CI failures from #352's down_revision will clear once #352 merges and this branch rebases.

Please re-review when you get a chance @parthrohit22