mihir2007/Cyber-Risk
0
1"""2graph_engine.py3---------------4Advanced Attack Graph and Lateral-Movement Engine built with NetworkX.5 6Features:7- Dynamically constructs attack graph topology by querying `NetworkEdge` and8 `Asset` models from PostgreSQL.9- Performs weighted Dijkstra shortest-path analysis from perimeter ingress10 nodes (source="INTERNET") to critical enterprise crown jewels.11- Calculates Path Exposure Coefficients (PEC) sensitive to hop distance,12 edge traversal costs, and active security controls (e.g. network micro-segmentation,13 Zero Trust, and WAF rules).14- Serializes graph topology to `{ "nodes": [...], "edges": [...] }` format15 compatible with D3.js, Cytoscape.js, Vis.js, and React Flow.16"""17 18from __future__ import annotations19 20from dataclasses import dataclass21from typing import Any, Optional, Union22 23import networkx as nx24from sqlalchemy.orm import Session25 26from models import Asset, DataLineageFlow, NetworkEdge, SecurityControl27from schemas import DataLineageFlowRead28 29# Fallback reference backbone used when database has not yet been seeded30FALLBACK_BACKBONE_EDGES: list[tuple[str, str, float, str, bool]] = [31 ("INTERNET", "upi-gateway-prod-01", 1.2, "HTTPS", False),32 ("INTERNET", "upi-gateway-prod-02", 1.2, "HTTPS", False),33 ("INTERNET", "mobile-banking-gw-01", 1.1, "HTTPS", False),34 ("INTERNET", "staff-portal-01", 1.5, "HTTPS", False),35 ("INTERNET", "staff-portal-02", 1.5, "HTTPS", False),36 ("staff-portal-01", "call-center-crm-01", 1.8, "HTTPS", False),37 ("staff-portal-02", "erp-hr-01", 2.0, "HTTPS", False),38 ("upi-gateway-prod-01", "trading-platform-01", 2.2, "INTERNAL_RPC", False),39 ("upi-gateway-prod-02", "trading-platform-01", 2.2, "INTERNAL_RPC", False),40 ("mobile-banking-gw-01", "trading-platform-01", 2.0, "INTERNAL_RPC", False),41 ("trading-platform-01", "core-bank-switch-01", 2.5, "INTERNAL_RPC", True),42 ("trading-platform-01", "core-bank-switch-02", 2.5, "INTERNAL_RPC", True),43 ("trading-platform-01", "cust-db-primary", 2.0, "SQL_NET", True),44 ("core-bank-switch-01", "cust-db-primary", 1.5, "INTERNAL_RPC", True),45 ("core-bank-switch-02", "cust-db-primary", 1.5, "INTERNAL_RPC", True),46 ("cust-db-primary", "cust-db-replica", 1.2, "DB_REPLICATION", True),47 ("cust-db-primary", "analytics-integration-gw", 1.8, "INTERNAL_RPC", True),48 ("analytics-integration-gw", "third-party-marketing-sync", 0.9, "HTTPS_DATA_EXPORT", False),49 ("call-center-crm-01", "cust-db-replica", 3.0, "HTTPS", True),50 ("erp-finance-01", "cust-db-primary", 2.8, "SQL_NET", True),51 ("erp-hr-01", "erp-finance-01", 2.4, "INTERNAL_RPC", False),52 ("office-wifi-controller", "branch-pos-north-12", 1.0, "WIFI_INTERNAL", False),53 ("office-wifi-controller", "branch-pos-south-07", 1.0, "WIFI_INTERNAL", False),54 ("office-wifi-controller", "branch-pos-west-03", 1.0, "WIFI_INTERNAL", False),55 ("branch-pos-north-12", "call-center-crm-01", 2.5, "VPN", False),56 ("branch-pos-south-07", "call-center-crm-01", 2.5, "VPN", False),57 ("branch-pos-west-03", "call-center-crm-01", 2.5, "VPN", False),58 ("core-bank-switch-01", "dr-site-core-switch", 3.5, "ASYNC_MIRROR", True),59 ("core-bank-switch-02", "dr-site-core-switch", 3.5, "ASYNC_MIRROR", True),60]61 62MAX_REASONABLE_HOPS: int = 863MAX_REASONABLE_DISTANCE: float = 20.064MIN_PEC: float = 0.0565MAX_PEC: float = 1.0066 67 68@dataclass(frozen=True)69class AssetGraphExposure:70 """Quantitative risk exposure metrics derived from attack graph topology."""71 72 asset_id: int73 hostname: str74 hops_from_internet: int75 path_exposure_coefficient: float # Normalized in [0.05, 1.00], higher = more exposed76 dijkstra_distance: float = 0.077 shortest_path: tuple[str, ...] = ()78 79 80@dataclass81class AssetLineageProfile:82 """Quantitative data provenance & indirect leak liability profile."""83 84 asset_id: int85 hostname: str86 is_unauthorized_sink: bool = False87 is_shadow_leak_origin: bool = False88 inherited_pii_count: int = 089 inherited_fin_count: int = 090 origin_hostname: Optional[str] = None91 intermediary_hostname: Optional[str] = None92 detection_status: str = "Legitimate Flow"93 leak_root_cause_identified: bool = False94 95 96 97class AttackGraphEngine:98 """99 Constructs, analyzes, and visualizes the enterprise attack graph.100 101 Dynamically loads topology from PostgreSQL `NetworkEdge` and `Asset` models102 and computes Dijkstra shortest attack paths, taking active security103 controls and network segmentation into account.104 """105 106 def __init__(107 self,108 assets: list[Asset] | None = None,109 edges: list[NetworkEdge] | None = None,110 session: Session | None = None,111 active_controls: list[Union[SecurityControl, str]] | None = None,112 ) -> None:113 """114 Initializes the Attack Graph Engine.115 116 Args:117 assets: Optional pre-loaded Asset instances.118 edges: Optional pre-loaded NetworkEdge instances.119 session: Optional active SQLAlchemy session to query assets/edges from PostgreSQL.120 active_controls: List of active SecurityControl instances or control codes.121 """122 self.session = session123 self._active_control_codes = self._parse_active_controls(active_controls)124 self.assets = self._load_assets(assets)125 self.edges = self._load_edges(edges)126 127 self.asset_by_hostname: dict[str, Asset] = {a.hostname: a for a in self.assets}128 self.asset_by_id: dict[int, Asset] = {a.id: a for a in self.assets}129 130 self.graph: nx.DiGraph = nx.DiGraph()131 self._build_graph()132 133 @staticmethod134 def _parse_active_controls(135 controls: list[Union[SecurityControl, str]] | None,136 ) -> set[str]:137 """Extracts active control codes from strings or SecurityControl instances."""138 if not controls:139 return set()140 codes: set[str] = set()141 for item in controls:142 if isinstance(item, str):143 codes.add(item.strip())144 elif isinstance(item, SecurityControl):145 if item.is_active:146 codes.add(item.code.strip())147 return codes148 149 def _load_assets(self, assets: list[Asset] | None) -> list[Asset]:150 """Loads assets from arguments or queries them from PostgreSQL."""151 if assets is not None:152 return assets153 if self.session is not None:154 try:155 return self.session.query(Asset).all()156 except Exception:157 pass158 # Attempt fallback via local session if available159 try:160 from database import SessionLocal161 with SessionLocal() as db:162 return db.query(Asset).all()163 except Exception:164 return []165 166 def _load_edges(self, edges: list[NetworkEdge] | None) -> list[NetworkEdge]:167 """Loads network edges from arguments or queries them from PostgreSQL."""168 if edges is not None and len(edges) > 0:169 return edges170 if self.session is not None:171 try:172 db_edges = self.session.query(NetworkEdge).all()173 if db_edges:174 return db_edges175 except Exception:176 pass177 try:178 from database import SessionLocal179 with SessionLocal() as db:180 db_edges = db.query(NetworkEdge).all()181 if db_edges:182 return db_edges183 except Exception:184 pass185 186 # Fallback to synthesized reference edges if database table is empty187 fallback_objects: list[NetworkEdge] = []188 for idx, item in enumerate(FALLBACK_BACKBONE_EDGES, start=1):189 src, tgt, w, proto, seg = item[:5]190 auth = True191 consent = True192 ftype = "Direct API"193 if src == "cust-db-primary" and tgt == "analytics-integration-gw":194 auth = True195 consent = True196 ftype = "Sub-Processor Delegation"197 elif src == "analytics-integration-gw" and tgt == "third-party-marketing-sync":198 auth = False199 consent = False200 ftype = "Unauthorized Delegation"201 202 fallback_objects.append(203 NetworkEdge(204 id=idx,205 source_node=src,206 target_node=tgt,207 weight=w,208 protocol=proto,209 is_segmented=seg,210 is_authorized_flow=auth,211 consent_recorded=consent,212 flow_type=ftype,213 )214 )215 return fallback_objects216 217 def _calculate_effective_edge_weight(self, edge: NetworkEdge) -> float:218 """219 Calculates the effective traversal cost of an edge based on its baseline220 difficulty and active security controls (e.g. segmentation, ZTNA, WAF, Consent Enforcement).221 """222 weight = float(edge.weight)223 224 # Inherent segmentation penalty225 if edge.is_segmented:226 weight *= 1.5227 228 # Active control multipliers229 if "CTRL_MICROSEG" in self._active_control_codes and edge.is_segmented:230 # Micro-segmentation drastically raises attacker effort to traverse segmented boundaries231 weight *= 3.0232 233 if "CTRL_ZERO_TRUST" in self._active_control_codes:234 # Zero Trust enforces continuous identity verification at every lateral hop235 weight *= 2.0236 237 if "CTRL_WAF" in self._active_control_codes and edge.source_node == "INTERNET":238 # Perimeter WAF filtering on inbound entry points239 weight *= 2.5240 241 if "CTRL_PAM" in self._active_control_codes and edge.protocol in ("SSH", "INTERNAL_RPC"):242 # Privileged access management hardens administrative and RPC conduits243 weight *= 1.8244 245 # Unauthorized sub-processor delegation without consent enforcement246 is_unauthorized = (not getattr(edge, "is_authorized_flow", True)) or (not getattr(edge, "consent_recorded", True))247 if is_unauthorized:248 if "CTRL_CONSENT_ENFORCEMENT" not in self._active_control_codes:249 # Heavily penalized traversal costs on unauthorized edges when CTRL_CONSENT_ENFORCEMENT is absent250 weight *= 4.0251 else:252 # Active consent verifier governs and blocks unauthorized pivots253 weight *= 1.2254 255 if "CTRL_EGRESS_DLP" in self._active_control_codes and (256 getattr(edge, "flow_type", "") in ("Unauthorized Delegation", "Sub-Processor Delegation")257 or "EXPORT" in getattr(edge, "protocol", "").upper()258 ):259 weight *= 2.5260 261 return max(weight, 0.1)262 263 def _build_graph(self) -> None:264 """Constructs the NetworkX directed graph populated with node and edge attributes."""265 self.graph.clear()266 267 # Add all known assets as nodes with risk metadata268 for asset in self.assets:269 self.graph.add_node(270 asset.hostname,271 asset_id=asset.id,272 hostname=asset.hostname,273 asset_type=asset.asset_type,274 tier=asset.tier,275 risk_level=asset.tier,276 business_unit=asset.business_unit,277 revenue_per_minute=asset.revenue_per_minute,278 )279 280 # Add edges and ensure referenced nodes (e.g. 'INTERNET') exist281 for idx, edge in enumerate(self.edges, start=1):282 for node_name in (edge.source_node, edge.target_node):283 if not self.graph.has_node(node_name):284 is_internet = node_name == "INTERNET"285 self.graph.add_node(286 node_name,287 asset_id=None,288 hostname=node_name,289 asset_type="External Gateway" if is_internet else "Internal Switch",290 tier="Perimeter" if is_internet else "Infrastructure",291 risk_level="External" if is_internet else "Medium",292 business_unit="Network Operations",293 revenue_per_minute=0.0,294 )295 296 effective_weight = self._calculate_effective_edge_weight(edge)297 edge_id = f"edge-{edge.id or idx}"298 self.graph.add_edge(299 edge.source_node,300 edge.target_node,301 id=edge_id,302 weight=effective_weight,303 base_weight=float(edge.weight),304 protocol=edge.protocol,305 is_segmented=bool(edge.is_segmented),306 is_authorized_flow=bool(getattr(edge, "is_authorized_flow", True)),307 consent_recorded=bool(getattr(edge, "consent_recorded", True)),308 flow_type=str(getattr(edge, "flow_type", "Direct API")),309 data_tags=getattr(edge, "data_tags", None),310 )311 312 def compute_dijkstra_shortest_paths(313 self, source: str = "INTERNET"314 ) -> dict[str, tuple[list[str], float]]:315 """316 Computes weighted Dijkstra shortest paths and cumulative costs from a source317 node to all reachable nodes in the graph.318 319 Returns:320 Dictionary mapping target node names to (path_list, total_weight).321 """322 if not self.graph.has_node(source):323 return {}324 325 try:326 lengths = nx.single_source_dijkstra_path_length(327 self.graph, source=source, weight="weight"328 )329 paths = nx.single_source_dijkstra_path(330 self.graph, source=source, weight="weight"331 )332 return {node: (paths[node], float(lengths[node])) for node in paths}333 except Exception:334 return {}335 336 def compute_exposure(self, source: str = "INTERNET") -> dict[int, AssetGraphExposure]:337 """338 Calculates path exposure coefficients (PEC) for all enterprise assets339 based on hop count, weighted Dijkstra transit costs, and active segmentation controls.340 341 Higher PEC (up to 1.0) means higher exploitability / easier reachability.342 Lower PEC reflects deeper isolation and active defense controls.343 """344 dijkstra_data = self.compute_dijkstra_shortest_paths(source=source)345 results: dict[int, AssetGraphExposure] = {}346 347 for asset in self.assets:348 if asset.hostname in dijkstra_data:349 path, distance = dijkstra_data[asset.hostname]350 hops = max(len(path) - 1, 1)351 352 # Attenuation formula combining hop count and Dijkstra traversal cost353 # Baseline 1-hop distance of 1.0 produces PEC = 1.0354 effective_cost = 1.0 + 0.4 * (hops - 1) + 0.6 * max(distance - 1.0, 0.0)355 pec = 1.0 / effective_cost356 pec = max(MIN_PEC, min(MAX_PEC, pec))357 else:358 # Unreachable from perimeter359 hops = MAX_REASONABLE_HOPS360 distance = MAX_REASONABLE_DISTANCE361 path = []362 pec = MIN_PEC363 364 results[asset.id] = AssetGraphExposure(365 asset_id=asset.id,366 hostname=asset.hostname,367 hops_from_internet=hops,368 path_exposure_coefficient=round(pec, 4),369 dijkstra_distance=round(distance, 4),370 shortest_path=tuple(path),371 )372 373 return results374 375 def shortest_path_to_crown_jewels(376 self, source: str = "INTERNET"377 ) -> dict[int, dict[str, Any]]:378 """379 Returns Dijkstra shortest attack paths from the perimeter to all Critical-tier380 crown-jewel assets, providing actionable lateral movement trails.381 """382 dijkstra_data = self.compute_dijkstra_shortest_paths(source=source)383 results: dict[int, dict[str, Any]] = {}384 385 for asset in self.assets:386 if asset.tier != "Critical":387 continue388 389 if asset.hostname in dijkstra_data:390 path, distance = dijkstra_data[asset.hostname]391 results[asset.id] = {392 "hostname": asset.hostname,393 "tier": asset.tier,394 "hops": max(len(path) - 1, 1),395 "dijkstra_distance": round(distance, 3),396 "path": path,397 }398 else:399 results[asset.id] = {400 "hostname": asset.hostname,401 "tier": asset.tier,402 "hops": MAX_REASONABLE_HOPS,403 "dijkstra_distance": MAX_REASONABLE_DISTANCE,404 "path": [],405 }406 407 return results408 409 def trace_transitive_data_lineage(410 self,411 ) -> tuple[list[DataLineageFlowRead], dict[int, AssetLineageProfile]]:412 """413 Performs breadth-first data provenance tracing and taint propagation across414 the enterprise graph topology to detect unauthorized sub-processors and415 indirect data leakage (A -> B -> C).416 417 Identifies paths where Origin A transfers sensitive records (PII / Financial)418 to Intermediary B, and B delegates or transmits to Sink C without recorded419 authorization or data principal consent.420 421 Returns:422 - List of DataLineageFlowRead instances detailing detected flows.423 - Mapping from asset_id to AssetLineageProfile with inherited record liabilities.424 """425 lineage_profiles: dict[int, AssetLineageProfile] = {426 asset.id: AssetLineageProfile(asset_id=asset.id, hostname=asset.hostname)427 for asset in self.assets428 }429 430 flow_reads: list[DataLineageFlowRead] = []431 flow_id_counter = 1432 433 # Check for pre-existing DataLineageFlow records in DB session if available434 if self.session is not None:435 try:436 db_flows = self.session.query(DataLineageFlow).all()437 if db_flows:438 for df in db_flows:439 origin_host = self.asset_by_id.get(df.origin_asset_id)440 inter_host = (441 self.asset_by_id.get(df.intermediary_asset_id)442 if df.intermediary_asset_id443 else None444 )445 dest_host = self.asset_by_id.get(df.destination_asset_id)446 447 origin_name = origin_host.hostname if origin_host else f"asset-{df.origin_asset_id}"448 inter_name = inter_host.hostname if inter_host else None449 dest_name = dest_host.hostname if dest_host else f"asset-{df.destination_asset_id}"450 451 is_leak = (452 (not df.is_authorized)453 or (not df.has_user_consent)454 or ("Unauthorized" in df.detection_status)455 )456 457 flow_reads.append(458 DataLineageFlowRead(459 id=df.id,460 origin_hostname=origin_name,461 intermediary_hostname=inter_name,462 destination_hostname=dest_name,463 is_authorized=df.is_authorized,464 has_user_consent=df.has_user_consent,465 records_exposed_pii=df.records_exposed_pii,466 records_exposed_financial=df.records_exposed_financial,467 detection_status=df.detection_status,468 leak_root_cause_identified=is_leak,469 )470 )471 472 if is_leak and df.destination_asset_id in lineage_profiles:473 dest_prof = lineage_profiles[df.destination_asset_id]474 dest_prof.is_unauthorized_sink = True475 dest_prof.is_shadow_leak_origin = True476 dest_prof.inherited_pii_count += df.records_exposed_pii477 dest_prof.inherited_fin_count += df.records_exposed_financial478 dest_prof.origin_hostname = origin_name479 dest_prof.intermediary_hostname = inter_name480 dest_prof.detection_status = df.detection_status481 dest_prof.leak_root_cause_identified = True482 483 if flow_reads:484 return flow_reads, lineage_profiles485 except Exception:486 pass487 488 # Breadth-first graph traversal for data lineage & taint propagation489 origin_assets = [490 a for a in self.assets491 if (a.pii_records_count > 0 or a.financial_records_count > 0)492 ]493 494 seen_flows: set[tuple[str, Optional[str], str]] = set()495 496 for origin in origin_assets:497 if not self.graph.has_node(origin.hostname):498 continue499 500 for intermediary_name in self.graph.successors(origin.hostname):501 edge_ab_data = self.graph.get_edge_data(origin.hostname, intermediary_name) or {}502 ab_authorized = edge_ab_data.get("is_authorized_flow", True)503 ab_consent = edge_ab_data.get("consent_recorded", True)504 505 b_successors = [506 succ for succ in self.graph.successors(intermediary_name)507 if succ != origin.hostname508 ]509 510 if not b_successors:511 flow_key = (origin.hostname, None, intermediary_name)512 if flow_key not in seen_flows:513 seen_flows.add(flow_key)514 is_auth = ab_authorized and ab_consent515 flow_reads.append(516 DataLineageFlowRead(517 id=flow_id_counter,518 origin_hostname=origin.hostname,519 intermediary_hostname=None,520 destination_hostname=intermediary_name,521 is_authorized=is_auth,522 has_user_consent=ab_consent,523 records_exposed_pii=origin.pii_records_count,524 records_exposed_financial=origin.financial_records_count,525 detection_status="Legitimate Flow" if is_auth else "Unauthorized Sub-Processing",526 leak_root_cause_identified=not is_auth,527 )528 )529 flow_id_counter += 1530 continue531 532 for sink_name in b_successors:533 flow_key = (origin.hostname, intermediary_name, sink_name)534 if flow_key in seen_flows:535 continue536 seen_flows.add(flow_key)537 538 edge_bc_data = self.graph.get_edge_data(intermediary_name, sink_name) or {}539 bc_authorized = edge_bc_data.get("is_authorized_flow", True)540 bc_consent = edge_bc_data.get("consent_recorded", True)541 bc_flow_type = edge_bc_data.get("flow_type", "Direct API")542 543 is_unauthorized_leak = (544 not bc_authorized545 or not bc_consent546 or "Unauthorized" in bc_flow_type547 or not ab_authorized548 )549 550 status = (551 "Unauthorized Sub-Processing"552 if is_unauthorized_leak553 else "Legitimate Flow"554 )555 556 flow_reads.append(557 DataLineageFlowRead(558 id=flow_id_counter,559 origin_hostname=origin.hostname,560 intermediary_hostname=intermediary_name,561 destination_hostname=sink_name,562 is_authorized=not is_unauthorized_leak,563 has_user_consent=bc_consent and ab_consent,564 records_exposed_pii=origin.pii_records_count,565 records_exposed_financial=origin.financial_records_count,566 detection_status=status,567 leak_root_cause_identified=is_unauthorized_leak,568 )569 )570 flow_id_counter += 1571 572 if is_unauthorized_leak:573 if self.graph.has_node(sink_name):574 self.graph.nodes[sink_name]["is_shadow_leak_origin"] = True575 self.graph.nodes[sink_name]["is_unauthorized_sink"] = True576 self.graph.nodes[sink_name]["taint_origin"] = origin.hostname577 578 if self.graph.has_edge(intermediary_name, sink_name):579 self.graph.edges[intermediary_name, sink_name]["is_high_risk_lateral_vector"] = True580 581 sink_asset = self.asset_by_hostname.get(sink_name)582 if sink_asset and sink_asset.id in lineage_profiles:583 sp = lineage_profiles[sink_asset.id]584 sp.is_unauthorized_sink = True585 sp.is_shadow_leak_origin = True586 sp.inherited_pii_count += origin.pii_records_count587 sp.inherited_fin_count += origin.financial_records_count588 sp.origin_hostname = origin.hostname589 sp.intermediary_hostname = intermediary_name590 sp.detection_status = "Unauthorized Sub-Processing"591 sp.leak_root_cause_identified = True592 593 return flow_reads, lineage_profiles594 595 def get_graph_visual_json(self, source: str = "INTERNET") -> dict[str, list[dict[str, Any]]]:596 """597 Serializes graph topology into a unified JSON structure compatible with598 D3.js, Cytoscape.js, Vis.js, and React Flow.599 600 Includes node risk tiers, asset classifications, path exposure coefficients,601 edge traversal weights, protocols, segmentation status, and lineage annotations.602 """603 self.trace_transitive_data_lineage()604 exposure_map = self.compute_exposure(source=source)605 exposure_by_host = {606 self.asset_by_id[aid].hostname: exp607 for aid, exp in exposure_map.items()608 if aid in self.asset_by_id609 }610 611 nodes: list[dict[str, Any]] = []612 for node_id, attrs in self.graph.nodes(data=True):613 hostname = attrs.get("hostname", node_id)614 exp = exposure_by_host.get(hostname)615 616 hops = exp.hops_from_internet if exp else (0 if node_id == source else MAX_REASONABLE_HOPS)617 pec = exp.path_exposure_coefficient if exp else (1.0 if node_id == source else MIN_PEC)618 distance = exp.dijkstra_distance if exp else 0.0619 620 node_data = {621 "id": node_id,622 "label": hostname,623 "asset_id": attrs.get("asset_id"),624 "tier": attrs.get("tier", "Infrastructure"),625 "asset_type": attrs.get("asset_type", "Host"),626 "risk_level": attrs.get("risk_level", "Medium"),627 "business_unit": attrs.get("business_unit", ""),628 "revenue_per_minute": attrs.get("revenue_per_minute", 0.0),629 "hops_from_internet": hops,630 "path_exposure_coefficient": pec,631 "dijkstra_distance": distance,632 "is_shadow_leak_origin": attrs.get("is_shadow_leak_origin", False),633 "is_unauthorized_sink": attrs.get("is_unauthorized_sink", False),634 "taint_origin": attrs.get("taint_origin"),635 }636 637 # Both flat attributes and nested 'data' payload for cross-library compatibility638 nodes.append({639 **node_data,640 "data": node_data,641 })642 643 edges: list[dict[str, Any]] = []644 for src, tgt, attrs in self.graph.edges(data=True):645 edge_id = attrs.get("id", f"{src}->{tgt}")646 weight = attrs.get("weight", 1.0)647 base_weight = attrs.get("base_weight", weight)648 protocol = attrs.get("protocol", "HTTPS")649 is_segmented = attrs.get("is_segmented", False)650 651 edge_data = {652 "id": edge_id,653 "source": src,654 "target": tgt,655 "weight": round(weight, 3),656 "base_weight": round(base_weight, 3),657 "cost": round(weight, 3),658 "protocol": protocol,659 "is_segmented": is_segmented,660 "is_authorized_flow": attrs.get("is_authorized_flow", True),661 "consent_recorded": attrs.get("consent_recorded", True),662 "flow_type": attrs.get("flow_type", "Direct API"),663 "data_tags": attrs.get("data_tags"),664 "is_high_risk_lateral_vector": attrs.get("is_high_risk_lateral_vector", False),665 }666 667 edges.append({668 **edge_data,669 "data": edge_data,670 })671 672 return {673 "nodes": nodes,674 "edges": edges,675 }676 677 678def build_exposure_map(679 assets: list[Asset],680 edges: list[NetworkEdge] | None = None,681 session: Session | None = None,682 active_controls: list[Union[SecurityControl, str]] | None = None,683) -> dict[int, AssetGraphExposure]:684 """685 Convenience function that builds the attack graph and computes686 the exposure map for the supplied enterprise assets.687 """688 engine = AttackGraphEngine(689 assets=assets,690 edges=edges,691 session=session,692 active_controls=active_controls,693 )694 return engine.compute_exposure()695