Reachability API: forward reachability (given a source, find all reachable sinks)

[rahlk]

Part of #170 (source↔sink reachability API).

Goal

Given a source callable, return the sinks it can reach through the call graph. This is the forward direction of source↔sink matching.

Proposed API

On the analysis façade / backend contract:

def get_reachable_sinks(
    source: str,
    sinks: Iterable[str] | None = None,   # candidate sink signatures; None ⇒ any reachable node
    *,
    max_depth: int | None = None,         # hop bound; None ⇒ unbounded
    include_external: bool = True,        # include External (phantom/library) targets
) -> Dict[str, int]:                      # reachable sink signature -> shortest hop distance
    ...

Returning a {sink: distance} map (rather than a bare set) keeps it cheap to rank/triage; a convenience returning just the set is fine too.

Semantics

All t such that source -CALLS*1..max_depth-> t, intersected with sinks when provided.

Backing implementations

• Neo4j:

  MATCH (s:Symbol {signature:$src})-[:CALLS*1..]->(t:Symbol)
  WHERE $sinks IS NULL OR t.signature IN $sinks
  RETURN DISTINCT t.signature AS sink, min(length( ... )) AS dist

  (use a bounded *1..$depth when max_depth is set; include_external=false adds AND NOT t:External).
• In-memory (NetworkX): nx.descendants(G, source) (optionally nx.single_source_shortest_path_length(G, source, cutoff=max_depth) to get distances), intersected with sinks.

Example use

From an entrypoint/route handler, list the dangerous library calls it can reach (node:child_process.exec, fs.writeFile, …) — i.e. the attack surface of that entry.

Acceptance criteria

• Method on the backend contract + façade, implemented for the Neo4j backend and the in-memory backend, with matching semantics.
• sinks=None, explicit sinks set, max_depth, and include_external all honored.
• Distances (or at least membership) verified against a known fixture where the source→sink paths are hand-checked.
• Truncation (depth bound hit) is observable, not silent.