feat: network topology visualisation (#95)

backend/app/api/__init__.py

@@ -898,3 +898,65 @@ def get_segmentation(db: Annotated[Session, Depends(get_db)]) -> SegmentationOut
         ],
         recommendations=result.recommendations,
     )
+
+
+# ── /api/topology ──────────────────────────────────────────────────────────────
+
+
+class TopologyNode(BaseModel):
+    id: int
+    ip_address: str
+    label: str | None
+    hostname: str | None
+    device_type: str
+    highest_severity: str | None  # critical | high | medium | low | None
+    port_count: int
+    security_score: int
+    is_gateway: bool
+
+
+@router.get("/topology", response_model=list[TopologyNode])
+def get_topology(db: Annotated[Session, Depends(get_db)]) -> list[TopologyNode]:
+    """Return all devices enriched for network topology visualisation."""
+    from app.models.device import Device
+
+    devices = (
+        db.execute(select(Device).options(selectinload(Device.risks), selectinload(Device.ports)))
+        .scalars()
+        .all()
+    )
+
+    # Gateway detection: explicit router type, or first device whose IP ends in .1
+    gateway_id: int | None = None
+    for d in devices:
+        if d.device_type == "router":
+            gateway_id = d.id
+            break
+    if gateway_id is None:
+        for d in devices:
+            if d.ip_address and d.ip_address.endswith(".1"):
+                gateway_id = d.id
+                break
+
+    _severity_rank = {"critical": 4, "high": 3, "medium": 2, "low": 1}
+
+    def _highest_severity(d) -> str | None:  # noqa: ANN001 — SQLAlchemy instance
+        severities = [r.severity for r in d.risks if not r.acknowledged_at]
+        if not severities:
+            return None
+        return max(severities, key=lambda s: _severity_rank.get(s, 0))
+
+    return [
+        TopologyNode(
+            id=d.id,
+            ip_address=d.ip_address,
+            label=d.label,
+            hostname=d.hostname,
+            device_type=d.device_type or "unknown",
+            highest_severity=_highest_severity(d),
+            port_count=len(d.ports),
+            security_score=_device_security_score(d),
+            is_gateway=(d.id == gateway_id),
+        )
+        for d in devices
+    ]

backend/tests/test_api.py

@@ -870,3 +870,65 @@ def test_segmentation_mixed_risk_pair_detected(client, db_engine):
     db.delete(srv)
     db.commit()
     db.close()
+
+
+# ── /api/topology ──────────────────────────────────────────────────────────────
+
+
+def test_topology_empty(client):
+    """GET /api/topology returns empty list when no devices."""
+    resp = client.get("/api/topology")
+    assert resp.status_code == 200
+    assert resp.json() == []
+
+
+def test_topology_returns_nodes(client, seeded_db):
+    """GET /api/topology returns one node per device with required fields."""
+    resp = client.get("/api/topology")
+    assert resp.status_code == 200
+    nodes = resp.json()
+    assert len(nodes) >= 1
+    node = nodes[0]
+    for field in ("id", "ip_address", "device_type", "port_count", "security_score", "is_gateway"):
+        assert field in node, f"missing field: {field}"
+
+
+def test_topology_gateway_detection_by_ip(client, db_engine):
+    """GET /api/topology marks device ending in .1 as gateway."""
+    from app.models.device import Device
+
+    S = sessionmaker(bind=db_engine)  # noqa: N806 -- uppercase matches SQLAlchemy Session convention
+    db = S()
+    gw = Device(ip_address="192.168.1.1", device_type="unknown")
+    other = Device(ip_address="192.168.1.50", device_type="workstation")
+    db.add_all([gw, other])
+    db.commit()
+
+    resp = client.get("/api/topology")
+    nodes = {n["ip_address"]: n for n in resp.json()}
+    assert nodes["192.168.1.1"]["is_gateway"] is True
+    assert nodes["192.168.1.50"]["is_gateway"] is False
+
+    db.delete(gw)
+    db.delete(other)
+    db.commit()
+    db.close()
+
+
+def test_topology_gateway_detection_by_type(client, db_engine):
+    """GET /api/topology marks device with device_type=router as gateway."""
+    from app.models.device import Device
+
+    S = sessionmaker(bind=db_engine)  # noqa: N806 -- uppercase matches SQLAlchemy Session convention
+    db = S()
+    router = Device(ip_address="10.0.0.254", device_type="router")
+    db.add(router)
+    db.commit()
+
+    resp = client.get("/api/topology")
+    nodes = {n["ip_address"]: n for n in resp.json()}
+    assert nodes["10.0.0.254"]["is_gateway"] is True
+
+    db.delete(router)
+    db.commit()
+    db.close()