krosh/network-diagrams-tool
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

add bridge and vlan support

Browse Source
This commit is contained in:
krosh
2026-03-25 15:25:02 +03:00
parent 45198209ab
commit 4c69be99e0
8 changed files with 244 additions and 53 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/netdiag/domain/models.py
+46 -16
View File
@@ -5,10 +5,13 @@ from typing import Any, Dict, List, Optional
class Interface:
name: str
itype: Optional[str]
adapter: Optional[str]
master_interface: Optional[str]
slave_interfaces: Optional[List[str]]
parent_interface: Optional[str]
ip_address: Optional[str]
network: Optional[str]
subnet_mask: Optional[str]
default_gateway: Optional[str]
# ---
device: "Device" # set by Device.add_interface() when the interface is added to a device
@@ -16,38 +19,70 @@ class Interface:
def __init__(
self,
name: str,
itype: Optional[str] = None,
adapter: Optional[str] = None,
master_interface: Optional[str] = None,
slave_interfaces: Optional[List[str]] = None,
parent_interface: Optional[str] = None,
ip_address: Optional[str] = None,
network: Optional[str] = None,
subnet_mask: Optional[str] = None,
default_gateway: Optional[str] = None,
):
if not isinstance(name, str):
raise ValueError("Interface 'name' must be a string")
if itype is not None and not isinstance(itype, str):
raise ValueError("Interface 'itype' must be a string or None")
if ip_address is not None and not isinstance(ip_address, str):
raise ValueError("Interface 'ip_address' must be a string or None")
if network is not None and not isinstance(network, str):
raise ValueError("Interface 'network' must be a string or None")
if subnet_mask is not None and not isinstance(subnet_mask, str):
raise ValueError("Interface 'subnet_mask' must be a string or None")
if default_gateway is not None and not isinstance(default_gateway, str):
raise ValueError("Interface 'default_gateway' must be a string or None")
if adapter is not None and not isinstance(adapter, str):
raise ValueError("Interface 'adapter' must be a string or None")
if master_interface is not None and not isinstance(master_interface, str):
raise ValueError("Interface 'master_interface' must be a string or None")
if slave_interfaces is not None and not isinstance(slave_interfaces, list):
raise ValueError(
"Interface 'slave_interfaces' must be a list of strings or None"
)
if parent_interface is not None and not isinstance(parent_interface, str):
raise ValueError("Interface 'parent_interface' must be a string or None")
itype = None if itype == "" else itype
adapter = None if adapter == "" else adapter
if itype is None and adapter is None:
itype = "virtual"
if adapter is not None:
if itype is not None and itype != "physical":
raise ValueError(
"Interface with an adapter must have 'itype' set to 'physical'"
)
itype = "physical"
if itype not in (None, "physical", "virtual", "bridge", "vlan"):
raise ValueError(
"Interface 'itype' must be one of 'physical', 'virtual', 'bridge', 'vlan', or None"
)
if itype == "bridge" and not slave_interfaces:
raise ValueError("Bridge interfaces must have 'slave_interfaces' defined")
if itype == "vlan" and not parent_interface:
raise ValueError("VLAN interfaces must have 'parent_interface' defined")
self.name = name
self.itype = itype
self.ip_address = ip_address
self.network = network
self.subnet_mask = subnet_mask
self.default_gateway = default_gateway
self.adapter = adapter
self.master_interface = master_interface
self.slave_interfaces = slave_interfaces
self.parent_interface = parent_interface
def __repr__(self) -> str:
return f"Interface(name={self.name}, ip_address={self.ip_address}, network={self.network}, default_gateway={self.default_gateway})"
class VirtualInterface(Interface):
pass
return f"Interface(name={self.name}, itype={self.itype}, ip_address={self.ip_address}, network={self.network}, default_gateway={self.default_gateway}, subnet_mask={self.subnet_mask}, adapter={self.adapter}, slave_interfaces={self.slave_interfaces})"
# ===========================
@@ -105,14 +140,12 @@ class Network:
interfaces: List[Interface]
vlan: Optional[str] # need a proper parse
network_ip: Optional[str]
subnet_mask: Optional[str]
def __init__(
self,
name: str,
vlan: Optional[str] = None,
network_ip: Optional[str] = None,
subnet_mask: Optional[str] = None,
):
if not isinstance(name, str):
raise ValueError("Network 'name' must be a string")
@@ -120,14 +153,11 @@ class Network:
raise ValueError("Network 'vlan' must be a string or None")
if network_ip is not None and not isinstance(network_ip, str):
raise ValueError("Network 'network_ip' must be a string or None")
if subnet_mask is not None and not isinstance(subnet_mask, str):
raise ValueError("Network 'subnet_mask' must be a string or None")
self.name = name
self.interfaces = []
self.vlan = vlan
self.network_ip = network_ip
self.subnet_mask = subnet_mask
def add_interface(self, interface: Interface):
if not isinstance(interface, Interface):
@@ -152,7 +182,7 @@ class Network:
)
def __repr__(self) -> str:
return f"Network(name={self.name}, interfaces={self.interfaces}), vlan={self.vlan}, network_ip={self.network_ip}, subnet_mask={self.subnet_mask})"
return f"Network(name={self.name}, interfaces={self.interfaces}), vlan={self.vlan}, network_ip={self.network_ip})"
class Topology:
src/netdiag/output/d2.py
+1 -2
View File
@@ -2,7 +2,6 @@ import shutil
import subprocess
from ..domain.models import Topology
from ..domain.models import VirtualInterface
from pathlib import Path
from py_d2 import D2Diagram, D2Shape, D2Connection
from py_d2.shape import Shape
@@ -53,7 +52,7 @@ def generate_d2_diagram(topology: Topology, output_path: Path) -> None:
for device in topology.devices.values():
for interface in device.interfaces.values():
if isinstance(interface, VirtualInterface):
if interface.itype != "physical":
continue # Skip virtual interfaces for now
shape = D2Shape(
src/netdiag/output/file_convert.py
+48 -11
View File
@@ -21,12 +21,6 @@ def make_yaml(topology: Topology, output_path: Path) -> None:
if len(interfaces_with_device) >= 2:
data["networks"].append(
# {
# network.name: [
# f"{iface.device.name}.{iface.name}"
# for iface in interfaces_with_device
# ]
# }
{
"name": network.name,
}
@@ -36,24 +30,67 @@ def make_yaml(topology: Topology, output_path: Path) -> None:
for _, device in topology.devices.items():
interfaces = dict()
interfaces["bridges"] = []
interfaces["vlans"] = []
for interface_name, interface in device.interfaces.items():
ip = interface.ip_address if interface.ip_address else None
mask = (
topology.networks[interface.network].subnet_mask
if interface.network
else None
)
mask = interface.subnet_mask if interface.subnet_mask else None
if mask is not None:
mask = mask.split("/")[1] if "/" in mask else mask
ip = f"{interface.ip_address}/{mask}" if interface.ip_address else None
index = None
# check regex Adapter[0-9]+
if interface.adapter:
if (
not interface.adapter.startswith("Adapter")
or not interface.adapter[7:].isdigit()
):
raise ValueError(
f"Invalid adapter name '{interface.adapter}' for interface '{interface_name}' on device '{device.name}'. Adapter name must be in the format 'AdapterX' where X is a number."
)
index = int(interface.adapter[7:])
if interface.itype == "bridge":
interfaces["bridges"].append(
{
"name": interface.name,
"members": interface.slave_interfaces,
"ip": ip,
"network": interface.network if interface.network else None,
"gateway": (
interface.default_gateway
if interface.default_gateway
else None
),
}
)
continue
if interface.itype == "vlan":
interfaces["vlans"].append(
{
"name": interface.name,
"parent": interface.parent_interface,
"ip": ip,
"network": interface.network if interface.network else None,
"gateway": (
interface.default_gateway
if interface.default_gateway
else None
),
}
)
continue
interfaces[interface_name] = {
"ip": ip,
"network": interface.network if interface.network else None,
"gateway": (
interface.default_gateway if interface.default_gateway else None
),
"index": index,
}
data["nodes"].append(
src/netdiag/parse/convert_raw.py
+17 -23
View File
@@ -8,7 +8,6 @@ from ..domain.models import (
Router,
Switch,
Topology,
VirtualInterface,
)
from . import RawDevices
@@ -16,7 +15,10 @@ name_matching = {
"DEVICE_TYPE": "Role",
"DEVICE_NAME": "Name",
"ADAPTER": "Adapter",
"INTERFACE TYPE": "Interface Type",
"MASTER": "Master Interface",
"SLAVES": "Slave Interfaces",
"PARENT": "Parent Interface",
"INTERFACE_NAME": "Interface",
"NETWORK_NAME": "Network",
"VLAN": "VLAN",
@@ -79,24 +81,20 @@ def add_interfaces(devices: list[Device], raw_devices: list[RawDevices]) -> None
f"Device with name '{device_name}' not found for interface parsing"
)
if not raw_device.fields.get(name_matching["ADAPTER"], "").strip():
interface = VirtualInterface(
name=_get_from_field(raw_device.fields, "INTERFACE_NAME"),
master_interface=_get_from_field(raw_device.fields, "MASTER"),
ip_address=_get_from_field(raw_device.fields, "IP_ADDRESS"),
network=_get_from_field(raw_device.fields, "NETWORK_NAME"),
default_gateway=_get_from_field(raw_device.fields, "DEFAULT_GATEWAY"),
)
else:
interface = Interface(
name=_get_from_field(raw_device.fields, "INTERFACE_NAME"),
adapter=_get_from_field(raw_device.fields, "ADAPTER"),
master_interface=_get_from_field(raw_device.fields, "MASTER"),
ip_address=_get_from_field(raw_device.fields, "IP_ADDRESS"),
network=_get_from_field(raw_device.fields, "NETWORK_NAME"),
default_gateway=_get_from_field(raw_device.fields, "DEFAULT_GATEWAY"),
)
interface = Interface(
name=_get_from_field(raw_device.fields, "INTERFACE_NAME"),
itype=_get_from_field(raw_device.fields, "INTERFACE TYPE"),
adapter=_get_from_field(raw_device.fields, "ADAPTER"),
slave_interfaces=(
_get_from_field(raw_device.fields, "SLAVES").split(",")
if _get_from_field(raw_device.fields, "SLAVES")
else None
),
ip_address=_get_from_field(raw_device.fields, "IP_ADDRESS"),
network=_get_from_field(raw_device.fields, "NETWORK_NAME"),
default_gateway=_get_from_field(raw_device.fields, "DEFAULT_GATEWAY"),
subnet_mask=raw_device.fields.get(name_matching["SUBNET_MASK"], "").strip(),
)
device.add_interface(interface)
return
@@ -112,7 +110,6 @@ def parse_networks(raw_devices: list[RawDevices]) -> list[Network]:
vlan = (raw_device.fields.get(name_matching["VLAN"], "").strip(),)
network_ip = (raw_device.fields.get(name_matching["NETWORK_IP"], "").strip(),)
subnet_mask = (raw_device.fields.get(name_matching["SUBNET_MASK"], "").strip(),)
if network_name in [n.name for n in networks]: # update if already exists
network = next(n for n in networks if n.name == network_name)
@@ -121,15 +118,12 @@ def parse_networks(raw_devices: list[RawDevices]) -> list[Network]:
network.vlan = vlan
if not network.network_ip and network_ip:
network.network_ip = network_ip
if not network.subnet_mask and subnet_mask:
network.subnet_mask = subnet_mask
continue
network = Network(
name=network_name,
vlan=raw_device.fields.get(name_matching["VLAN"], "").strip(),
network_ip=raw_device.fields.get(name_matching["NETWORK_IP"], "").strip(),
subnet_mask=raw_device.fields.get(name_matching["SUBNET_MASK"], "").strip(),
)
networks.append(network)