429 lines
16 KiB
Python
Executable file
429 lines
16 KiB
Python
Executable file
#!/usr/bin/env python3
|
|
"""Produce a deterministic structural summary of a Brick schema-v4 graph."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import math
|
|
import sys
|
|
from collections import Counter, defaultdict
|
|
from pathlib import Path
|
|
from typing import Any, NoReturn
|
|
|
|
DEPENDENCY_KINDS = {"calls", "imports", "inherits"}
|
|
NODE_KINDS = {"repository", "directory", "file", "class", "function", "external", "finding"}
|
|
EDGE_KINDS = {"contains", "defines", "calls", "imports", "inherits", "flags"}
|
|
EVIDENCE_LANES = {"syntactic", "symbolic", "imported", "heuristic"}
|
|
P4_METRIC_FIELDS = {
|
|
"parser_state_count",
|
|
"terminal_state_count",
|
|
"parser_transition_count",
|
|
"parser_cycle_count",
|
|
"cyclic_parser_state_count",
|
|
"unreachable_state_count",
|
|
"control_apply_node_count",
|
|
"control_apply_relationship_count",
|
|
"control_apply_cycle_count",
|
|
"cyclic_control_apply_node_count",
|
|
"unresolved_dependency_count",
|
|
}
|
|
|
|
|
|
def fail(message: str) -> NoReturn:
|
|
print(f"architecture_summary: {message}", file=sys.stderr)
|
|
raise SystemExit(2)
|
|
|
|
|
|
def reject_nonfinite_constant(constant: str) -> NoReturn:
|
|
raise ValueError(f"non-finite JSON number {constant}")
|
|
|
|
|
|
def is_normalized_absolute_root(value: str) -> bool:
|
|
if not value or "\\" in value:
|
|
return False
|
|
if value.startswith("//"):
|
|
remainder = value[2:]
|
|
components = remainder.split("/")
|
|
if (
|
|
len(components) < 2
|
|
or not components[0]
|
|
or not components[1]
|
|
or any(character in components[0] + components[1] for character in "?:")
|
|
):
|
|
return False
|
|
elif value.startswith("/"):
|
|
remainder = value[1:]
|
|
elif (
|
|
len(value) >= 3
|
|
and value[0].isascii()
|
|
and value[0].isalpha()
|
|
and value[1:3] == ":/"
|
|
):
|
|
remainder = value[3:]
|
|
else:
|
|
return False
|
|
return all(
|
|
component not in {".", ".."} and (component != "" or remainder == "")
|
|
for component in remainder.split("/")
|
|
)
|
|
|
|
|
|
def is_normalized_relative_path(value: str, allow_dot: bool = False) -> bool:
|
|
if allow_dot and value == ".":
|
|
return True
|
|
if not value or value.startswith("/") or "\\" in value:
|
|
return False
|
|
if len(value) >= 2 and value[0].isascii() and value[0].isalpha() and value[1] == ":":
|
|
return False
|
|
return all(component not in {"", ".", ".."} for component in value.split("/"))
|
|
|
|
|
|
def is_valid_confidence(value: Any) -> bool:
|
|
if value is None or isinstance(value, str):
|
|
return True
|
|
if isinstance(value, bool) or not isinstance(value, (int, float)):
|
|
return False
|
|
try:
|
|
return math.isfinite(float(value))
|
|
except OverflowError:
|
|
return False
|
|
|
|
|
|
def validate_graph_records(value: dict[str, Any]) -> None:
|
|
nodes = value["nodes"]
|
|
edges = value["edges"]
|
|
node_ids: set[str] = set()
|
|
repository_count = 0
|
|
for node in nodes:
|
|
if not isinstance(node, dict):
|
|
fail("every node must be an object")
|
|
node_id = node.get("id")
|
|
kind = node.get("kind")
|
|
if not isinstance(node_id, str) or not node_id:
|
|
fail("every node must have a non-empty string id")
|
|
if node_id in node_ids:
|
|
fail(f"duplicate node ID {node_id}")
|
|
node_ids.add(node_id)
|
|
if kind not in NODE_KINDS:
|
|
fail(f"unknown node kind {kind!r}")
|
|
if kind == "repository":
|
|
repository_count += 1
|
|
if repository_count != 1:
|
|
fail(f"graph must contain exactly one repository node; found {repository_count}")
|
|
|
|
edge_ids: set[str] = set()
|
|
for edge in edges:
|
|
if not isinstance(edge, dict):
|
|
fail("every edge must be an object")
|
|
edge_id = edge.get("id")
|
|
source = edge.get("source")
|
|
target = edge.get("target")
|
|
kind = edge.get("kind")
|
|
if not isinstance(edge_id, str) or not edge_id:
|
|
fail("every edge must have a non-empty string id")
|
|
if edge_id in edge_ids:
|
|
fail(f"duplicate edge ID {edge_id}")
|
|
if edge_id in node_ids:
|
|
fail(f"entity ID {edge_id} is shared by a node and edge")
|
|
edge_ids.add(edge_id)
|
|
if kind not in EDGE_KINDS:
|
|
fail(f"unknown edge kind {kind!r}")
|
|
if source not in node_ids or target not in node_ids:
|
|
fail(f"edge {edge_id} has an unknown endpoint")
|
|
evidence_records = edge.get("evidence")
|
|
if not isinstance(evidence_records, list) or not evidence_records:
|
|
fail(f"edge {edge_id} must contain non-empty evidence")
|
|
for evidence in evidence_records:
|
|
if not isinstance(evidence, dict):
|
|
fail(f"edge {edge_id} evidence must be an object")
|
|
if evidence.get("lane") not in EVIDENCE_LANES:
|
|
fail(f"edge {edge_id} has an unknown evidence lane")
|
|
if not isinstance(evidence.get("backend"), str):
|
|
fail(f"edge {edge_id} evidence backend must be a string")
|
|
if not isinstance(evidence.get("provenance"), str):
|
|
fail(f"edge {edge_id} evidence provenance must be a string")
|
|
confidence = evidence.get("confidence")
|
|
if not is_valid_confidence(confidence):
|
|
fail(f"edge {edge_id} evidence confidence has an invalid type")
|
|
|
|
|
|
def load_graph(path: Path) -> dict[str, Any]:
|
|
try:
|
|
value = json.loads(
|
|
path.read_text(encoding="utf-8"),
|
|
parse_constant=reject_nonfinite_constant,
|
|
)
|
|
except OSError as error:
|
|
fail(f"cannot read {path}: {error}")
|
|
except (json.JSONDecodeError, ValueError) as error:
|
|
fail(f"invalid JSON in {path}: {error}")
|
|
if not isinstance(value, dict):
|
|
fail("top-level JSON value must be an object")
|
|
if value.get("schema_version") != 4:
|
|
fail(f"unsupported schema_version {value.get('schema_version')!r}; expected 4")
|
|
if not isinstance(value.get("generated_by"), str) or not isinstance(value.get("root"), str):
|
|
fail("schema-v4 graph must contain string generated_by and root fields")
|
|
if not is_normalized_absolute_root(value["root"]):
|
|
fail("root must be a normalized portable absolute path")
|
|
scope = value.get("scope")
|
|
if not isinstance(scope, dict) or not isinstance(scope.get("path"), str):
|
|
fail("schema-v4 graph must contain scan scope.path")
|
|
if not is_normalized_relative_path(scope["path"], allow_dot=True):
|
|
fail("scope.path must be normalized and repository-relative")
|
|
if scope.get("p4_frontend") not in {"lexical", "tree-sitter-p4-pilot"}:
|
|
fail(f"unsupported P4 frontend in scan scope: {scope.get('p4_frontend')!r}")
|
|
excluded = scope.get("excluded_directories", [])
|
|
if not isinstance(excluded, list) or not all(
|
|
isinstance(directory, str) for directory in excluded
|
|
):
|
|
fail("scope.excluded_directories must be an array of strings")
|
|
if excluded != sorted(set(excluded)) or any(
|
|
not is_normalized_relative_path(directory) or "/" in directory
|
|
for directory in excluded
|
|
):
|
|
fail("scope.excluded_directories must contain sorted unique path components")
|
|
if not isinstance(value.get("nodes"), list) or not isinstance(
|
|
value.get("edges"), list
|
|
):
|
|
fail("graph must contain nodes and edges arrays")
|
|
metrics = value.get("metrics")
|
|
p4_metrics = metrics.get("p4") if isinstance(metrics, dict) else None
|
|
if not isinstance(p4_metrics, dict) or set(p4_metrics) != P4_METRIC_FIELDS:
|
|
fail("schema-v4 graph must contain the complete metrics.p4 contract")
|
|
if any(type(metric) is not int for metric in p4_metrics.values()):
|
|
fail("metrics.p4 values must be integers")
|
|
validate_graph_records(value)
|
|
return value
|
|
|
|
|
|
def text_field(value: Any, field: str) -> str | None:
|
|
result = value.get(field) if isinstance(value, dict) else None
|
|
return result if isinstance(result, str) else None
|
|
|
|
|
|
def evidence_value(value: Any, field: str) -> str:
|
|
result = value.get(field) if isinstance(value, dict) else None
|
|
if result is None:
|
|
return "unknown"
|
|
if isinstance(result, float):
|
|
return f"{result:.3f}".rstrip("0").rstrip(".")
|
|
return str(result)
|
|
|
|
|
|
def strongly_connected_components(
|
|
vertices: list[str], adjacency: dict[str, list[str]]
|
|
) -> list[list[str]]:
|
|
index = 0
|
|
indices: dict[str, int] = {}
|
|
lowlinks: dict[str, int] = {}
|
|
stack: list[str] = []
|
|
on_stack: set[str] = set()
|
|
components: list[list[str]] = []
|
|
|
|
def visit(vertex: str) -> None:
|
|
nonlocal index
|
|
indices[vertex] = index
|
|
lowlinks[vertex] = index
|
|
index += 1
|
|
stack.append(vertex)
|
|
on_stack.add(vertex)
|
|
|
|
for neighbor in adjacency.get(vertex, []):
|
|
if neighbor not in indices:
|
|
visit(neighbor)
|
|
lowlinks[vertex] = min(lowlinks[vertex], lowlinks[neighbor])
|
|
elif neighbor in on_stack:
|
|
lowlinks[vertex] = min(lowlinks[vertex], indices[neighbor])
|
|
|
|
if lowlinks[vertex] == indices[vertex]:
|
|
component: list[str] = []
|
|
while True:
|
|
member = stack.pop()
|
|
on_stack.remove(member)
|
|
component.append(member)
|
|
if member == vertex:
|
|
break
|
|
components.append(sorted(component))
|
|
|
|
for vertex in sorted(vertices):
|
|
if vertex not in indices:
|
|
visit(vertex)
|
|
return sorted(components, key=lambda item: (-len(item), item))
|
|
|
|
|
|
def ranked_counts(
|
|
counts: Counter[str], labels: dict[str, str], limit: int
|
|
) -> list[dict[str, Any]]:
|
|
rows = [
|
|
{"id": node_id, "label": labels.get(node_id, node_id), "count": count}
|
|
for node_id, count in counts.items()
|
|
]
|
|
return sorted(rows, key=lambda row: (-row["count"], row["id"]))[:limit]
|
|
|
|
|
|
def node_reference(node_id: str, nodes: dict[str, dict[str, Any]]) -> dict[str, Any]:
|
|
node = nodes.get(node_id, {})
|
|
result: dict[str, Any] = {
|
|
"id": node_id,
|
|
"name": text_field(node, "name") or node_id,
|
|
"kind": text_field(node, "kind") or "unknown",
|
|
}
|
|
path = text_field(node, "path")
|
|
if path is not None:
|
|
result["path"] = path
|
|
line = node.get("line")
|
|
if isinstance(line, int):
|
|
result["line"] = line
|
|
return result
|
|
|
|
|
|
def summarize(graph: dict[str, Any], limit: int) -> dict[str, Any]:
|
|
nodes = [node for node in graph["nodes"] if isinstance(node, dict)]
|
|
edges = [edge for edge in graph["edges"] if isinstance(edge, dict)]
|
|
node_by_id = {
|
|
node_id: node
|
|
for node in nodes
|
|
if (node_id := text_field(node, "id")) is not None
|
|
}
|
|
labels = {
|
|
node_id: text_field(node, "name") or node_id
|
|
for node_id, node in node_by_id.items()
|
|
}
|
|
|
|
node_kinds = Counter(text_field(node, "kind") or "unknown" for node in nodes)
|
|
edge_kinds = Counter(text_field(edge, "kind") or "unknown" for edge in edges)
|
|
evidence_lanes: Counter[str] = Counter()
|
|
evidence_backends: Counter[str] = Counter()
|
|
evidence_confidence: Counter[str] = Counter()
|
|
|
|
adjacency: dict[str, list[str]] = defaultdict(list)
|
|
fan_in: Counter[str] = Counter()
|
|
fan_out: Counter[str] = Counter()
|
|
self_loops: set[str] = set()
|
|
dependency_edges = 0
|
|
unresolved_targets: Counter[str] = Counter()
|
|
|
|
for edge in edges:
|
|
evidence_records = edge.get("evidence")
|
|
if isinstance(evidence_records, list):
|
|
for evidence in evidence_records:
|
|
if isinstance(evidence, dict):
|
|
evidence_lanes[evidence_value(evidence, "lane")] += 1
|
|
evidence_backends[evidence_value(evidence, "backend")] += 1
|
|
evidence_confidence[evidence_value(evidence, "confidence")] += 1
|
|
|
|
kind = text_field(edge, "kind")
|
|
source = text_field(edge, "source")
|
|
target = text_field(edge, "target")
|
|
if kind not in DEPENDENCY_KINDS or source is None or target is None:
|
|
continue
|
|
dependency_edges += 1
|
|
adjacency[source].append(target)
|
|
fan_out[source] += 1
|
|
fan_in[target] += 1
|
|
if source == target:
|
|
self_loops.add(source)
|
|
target_node = node_by_id.get(target)
|
|
if target_node is not None and text_field(target_node, "kind") == "external":
|
|
unresolved_targets[target] += 1
|
|
|
|
vertices = sorted(
|
|
set(adjacency)
|
|
| {neighbor for neighbors in adjacency.values() for neighbor in neighbors}
|
|
)
|
|
components = strongly_connected_components(vertices, adjacency)
|
|
cyclic_components = [
|
|
component
|
|
for component in components
|
|
if len(component) > 1 or component[0] in self_loops
|
|
]
|
|
|
|
findings = [node for node in nodes if text_field(node, "kind") == "finding"]
|
|
finding_backends: Counter[str] = Counter()
|
|
finding_confidence: Counter[str] = Counter()
|
|
finding_rules: Counter[str] = Counter()
|
|
for finding in findings:
|
|
metadata = finding.get("metadata") or finding.get("attributes")
|
|
provenance = finding.get("provenance")
|
|
if isinstance(metadata, dict):
|
|
finding_rules[
|
|
text_field(metadata, "rule_id")
|
|
or text_field(metadata, "rule")
|
|
or "unknown"
|
|
] += 1
|
|
finding_confidence[evidence_value(metadata, "confidence")] += 1
|
|
finding_backends[evidence_value(metadata, "backend")] += 1
|
|
elif isinstance(provenance, dict):
|
|
finding_backends[text_field(provenance, "backend") or "unknown"] += 1
|
|
|
|
unresolved = [
|
|
{
|
|
"id": node_id,
|
|
"label": labels.get(node_id, node_id),
|
|
"incoming_dependencies": count,
|
|
}
|
|
for node_id, count in unresolved_targets.items()
|
|
]
|
|
unresolved.sort(key=lambda row: (-row["incoming_dependencies"], row["id"]))
|
|
|
|
return {
|
|
"schema_version": graph.get("schema_version"),
|
|
"generated_by": graph.get("generated_by"),
|
|
"root": graph.get("root"),
|
|
"scope": graph.get("scope"),
|
|
"diagnostics": graph.get("diagnostics", []),
|
|
"inventory": {
|
|
"nodes": len(nodes),
|
|
"edges": len(edges),
|
|
"node_kinds": dict(sorted(node_kinds.items())),
|
|
"edge_kinds": dict(sorted(edge_kinds.items())),
|
|
},
|
|
"evidence": {
|
|
"lanes": dict(sorted(evidence_lanes.items())),
|
|
"backends": dict(sorted(evidence_backends.items())),
|
|
"confidence": dict(sorted(evidence_confidence.items())),
|
|
},
|
|
"dependencies": {
|
|
"edge_count": dependency_edges,
|
|
"cyclic_component_count": len(cyclic_components),
|
|
"cyclic_components": [
|
|
[node_reference(node_id, node_by_id) for node_id in component]
|
|
for component in cyclic_components[:limit]
|
|
],
|
|
"top_fan_in": ranked_counts(fan_in, labels, limit),
|
|
"top_fan_out": ranked_counts(fan_out, labels, limit),
|
|
"unresolved_target_count": len(unresolved_targets),
|
|
"top_unresolved_targets": unresolved[:limit],
|
|
},
|
|
"findings": {
|
|
"count": len(findings),
|
|
"backends": dict(sorted(finding_backends.items())),
|
|
"confidence": dict(sorted(finding_confidence.items())),
|
|
"rules": dict(sorted(finding_rules.items())),
|
|
},
|
|
"metrics": graph.get("metrics"),
|
|
}
|
|
|
|
|
|
def parse_args() -> argparse.Namespace:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("graph", type=Path, help="Brick schema-v4 JSON artifact")
|
|
parser.add_argument(
|
|
"--limit", type=int, default=10, help="maximum ranked entries (default: 10)"
|
|
)
|
|
args = parser.parse_args()
|
|
if args.limit <= 0:
|
|
parser.error("--limit must be positive")
|
|
return args
|
|
|
|
|
|
def main() -> None:
|
|
args = parse_args()
|
|
graph = load_graph(args.graph)
|
|
json.dump(summarize(graph, args.limit), sys.stdout, indent=2, sort_keys=True)
|
|
sys.stdout.write("\n")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|