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

Merge pull request #1 from kr0sh512/refactor

Browse Source 
Refactor
This commit is contained in:
krosh
2026-02-24 22:07:14 +03:00
committed by GitHub
parent c4856cdcb6 71cb0656fe
commit 8e33b67b46
31 changed files with 713 additions and 1328 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.gitignore
+3 -1
View File
@@ -1,3 +1,5 @@
data/
# Byte-compiled / optimized / DLL files
__pycache__/
*.py[codz]
@@ -186,7 +188,7 @@ cython_debug/
# that can be found at https://github.com/github/gitignore/blob/main/Global/VisualStudioCode.gitignore
# and can be added to the global gitignore or merged into this file. However, if you prefer,
# you could uncomment the following to ignore the entire vscode folder
# .vscode/
.vscode/
# Ruff stuff:
.ruff_cache/
.vscode/settings.json
Vendored
-3
View File
@@ -1,3 +0,0 @@
{
"nixEnvSelector.nixFile": "${workspaceFolder}/flake.nix"
}
UML/structure.jpg
BIN
View File
Binary file not shown.
Side by Side

Before

Width:  |  Height:  |  Size: 186 KiB

examples/input.csv
-3
View File
@@ -1,3 +0,0 @@
, prop1 , prop2 , prop3
ob1 , val1 , , val2
obj2 , , val3 , val4
1 prop1 prop2 prop3
2 ob1 val1 val2
3 obj2 val3 val4
examples/template.yaml
-160
View File
@@ -1,160 +0,0 @@
# yaml-language-server: $schema=./topology-schema.json
schema: asvk.topology/1.0
meta:
id: "string"
title: "string"
description: "string"
provider:
name: virtualbox # virtualbox | kvm | container
defaults:
cpu: 1
ram_mb: 512
disk_gb: 8
os_image: "debian-12-cloud.qcow2"
nic_model: virtio
defaults:
routing:
stack: bird # bird | frr | linux
protocols_enabled: []
switch:
impl: linux-bridge # linux-bridge | ovs
firewall:
impl: nftables # nftables | none
mgmt:
ssh_user: "lab"
ssh_key: "~/.ssh/id_rsa.pub"
variables: # for templates
asn: 65000
domain: "lab.local"
networks: # L2 / L3 / VLAN trunks
- id: lan1
type: l2 # l2 | l3
mtu: 1500
- id: lan13
type: l2
- id: trunk1
type: l2
vlan:
mode: trunk
allowed: [4, 8]
- id: mgmt
type: l3
cidr: 192.168.100.0/24
dhcp: false
nodes: # nodes (VM/containers)
- name: R1
role: router # router | switch | host
resources:
cpu: 1
ram_mb: 512
mgmt:
ip: 192.168.100.11/24
gw: 192.168.100.1
net: mgmt
interfaces: # order matters: eth1 ↔ adapter2 and so on
- name: eth1
network: lan13
addresses: ["10.0.13.1/24"]
- name: eth0 # MGMT
network: mgmt
routing:
stack: bird # inherited from defaults, can be omitted
protocols:
static:
routes:
- to: 0.0.0.0/0
via: 10.0.13.3
ospf: # if needed
enabled: false
rip:
enabled: false
services: # servers/daemons of the host
- name: sshd
enabled: true
switching: # L2 config for switch nodes
- node: S1
impl: linux-bridge # linux-bridge | ovs
bridges:
- name: br0
ports:
- if: eth1 # to trunk1
vlan:
mode: trunk
allowed: [4, 8]
- if: eth2 # access → VLAN 4
vlan:
mode: access
vid: 4
- if: eth3 # access → VLAN 8
vlan:
mode: access
vid: 8
svis: # VLAN management interfaces (SVI)
- vid: 4
ifname: br0.4
addresses: ["10.0.4.254/24"]
firewall: # High-level ACL-> to nftables
- node: R2
rules:
- id: allow-ospf
match:
in_if: eth1
ip_proto: ospf
action: accept
- id: telnet-from-R1-to-S1
match:
src: 10.0.13.1/32
dst: 10.0.4.254/32
l4: { proto: tcp, dport: 23 }
action: accept
- id: ftp-from-R3-to-S2
match:
src: 10.0.13.3/32
dst: 10.0.6.254/32
l4: { proto: tcp, dports: [20, 21] }
action: accept
- id: drop-rest
action: drop
profiles: # overlays - patches on top of the base description
- name: "all-frr"
description: "Switch all control-plane to FRR"
patches:
- op: replace
path: /defaults/routing/stack
value: frr
- name: "mix-bird-frr-ospf"
description: "BIRD by default, but OSPF on R2 via FRR"
patches:
- op: add
path: /nodes
where:
name: R2
patch:
routing:
protocols:
ospf:
enabled: true
impl: frr
- name: "switch-ovs"
patches:
- op: replace
path: /defaults/switch/impl
value: ovs
tests: # tests (expected result)
- name: ping-R1-to-R3
from: R1
to: 10.0.13.3
expect: success
- name: ping-VLAN-4-R1-to-R3
from: R1
to: 10.0.4.1
expect: success
examples/topology-schema.json
-607
View File
@@ -1,607 +0,0 @@
{
"$schema": "http://json-schema.org/draft-07/schema#",
"$id": "https://github.com/asvk-topology/schema/1.0",
"title": "ASVK Topology Configuration Schema",
"description": "JSON Schema for ASVK topology configuration files",
"type": "object",
"required": ["schema", "meta", "provider", "nodes"],
"properties": {
"schema": {
"type": "string",
"description": "Schema identifier",
"pattern": "^asvk\\.topology/[0-9]+\\.[0-9]+$",
"examples": ["asvk.topology/1.0"]
},
"meta": {
"type": "object",
"description": "Metadata for the topology configuration",
"required": ["id", "title"],
"properties": {
"id": {
"type": "string",
"description": "Unique identifier for the topology",
"pattern": "^[a-zA-Z0-9_-]+$"
},
"title": {
"type": "string",
"description": "Human-readable name for the topology"
},
"description": {
"type": "string",
"description": "Detailed description of the topology"
}
}
},
"provider": {
"type": "object",
"description": "Provider configuration for VMs/containers",
"required": ["name"],
"properties": {
"name": {
"type": "string",
"description": "Provider name",
"enum": ["virtualbox", "kvm", "container"]
},
"defaults": {
"type": "object",
"description": "Default resource allocation",
"properties": {
"cpu": {
"type": "integer",
"description": "Default CPU cores",
"minimum": 1,
"default": 1
},
"ram_mb": {
"type": "integer",
"description": "Default RAM in MB",
"minimum": 128,
"default": 512
},
"disk_gb": {
"type": "integer",
"description": "Default disk size in GB",
"minimum": 1,
"default": 8
},
"os_image": {
"type": "string",
"description": "Default OS image"
},
"nic_model": {
"type": "string",
"description": "Default NIC model",
"enum": ["virtio", "e1000", "rtl8139"]
},
"promiscuous": {
"type": "string",
"description": "Promiscuous mode setting",
"enum": ["deny", "allow-vms", "allow-all"]
}
}
},
"images": {
"type": "object",
"description": "Custom images for different node types",
"properties": {
"router": { "type": "string" },
"switch": { "type": "string" },
"host": { "type": "string" }
}
}
}
},
"defaults": {
"type": "object",
"description": "Default behavior and software configuration",
"properties": {
"routing": {
"type": "object",
"properties": {
"stack": {
"type": "string",
"enum": ["bird", "frr", "linux"],
"default": "bird"
},
"protocols_enabled": {
"type": "array",
"items": {
"type": "string",
"enum": ["ospf", "rip", "bgp", "static"]
},
"default": []
}
}
},
"switch": {
"type": "object",
"properties": {
"impl": {
"type": "string",
"enum": ["linux-bridge", "ovs"],
"default": "linux-bridge"
}
}
},
"firewall": {
"type": "object",
"properties": {
"impl": {
"type": "string",
"enum": ["nftables", "none"],
"default": "nftables"
}
}
},
"mgmt": {
"type": "object",
"properties": {
"ssh_user": {
"type": "string",
"default": "lab"
},
"ssh_key": {
"type": "string",
"description": "Path to SSH public key"
}
}
}
}
},
"variables": {
"type": "object",
"description": "Global variables that can be referenced in templates",
"additionalProperties": true
},
"networks": {
"type": "array",
"description": "Network definitions (L2 domains, L3 networks, VLAN trunks)",
"items": {
"type": "object",
"required": ["id", "type"],
"properties": {
"id": {
"type": "string",
"description": "Unique network identifier",
"pattern": "^[a-zA-Z0-9_-]+$"
},
"type": {
"type": "string",
"enum": ["l2", "l3"],
"description": "Network type"
},
"mtu": {
"type": "integer",
"description": "MTU for the network",
"minimum": 68,
"maximum": 9000,
"default": 1500
},
"cidr": {
"type": "string",
"description": "CIDR notation for L3 networks",
"pattern": "^([0-9]{1,3}\\.){3}[0-9]{1,3}/([0-9]|[1-2][0-9]|3[0-2])$"
},
"dhcp": {
"type": "boolean",
"description": "Enable DHCP on this network",
"default": false
},
"vlan": {
"type": "object",
"description": "VLAN configuration for trunk ports",
"properties": {
"mode": {
"type": "string",
"enum": ["access", "trunk"]
},
"allowed": {
"type": "array",
"items": {
"type": "integer",
"minimum": 1,
"maximum": 4094
},
"description": "Allowed VLAN IDs on trunk"
},
"vid": {
"type": "integer",
"description": "Native VLAN ID for access ports",
"minimum": 1,
"maximum": 4094
}
}
}
}
}
},
"nodes": {
"type": "array",
"description": "Node definitions (VMs/containers)",
"items": {
"type": "object",
"required": ["name", "role"],
"properties": {
"name": {
"type": "string",
"description": "Node name",
"pattern": "^[a-zA-Z0-9_-]+$"
},
"role": {
"type": "string",
"enum": ["router", "switch", "host"],
"description": "Node role/type"
},
"image": {
"type": "string",
"description": "OS image for the node (can reference provider.images)"
},
"resources": {
"type": "object",
"properties": {
"cpu": {
"type": "integer",
"minimum": 1
},
"ram_mb": {
"type": "integer",
"minimum": 128
},
"disk_gb": {
"type": "integer",
"minimum": 1
}
}
},
"mgmt": {
"type": "object",
"description": "Management network configuration",
"properties": {
"ip": {
"type": "string",
"pattern": "^([0-9]{1,3}\\.){3}[0-9]{1,3}/([0-9]|[1-2][0-9]|3[0-2])$"
},
"gw": {
"type": "string",
"pattern": "^([0-9]{1,3}\\.){3}[0-9]{1,3}$"
},
"net": {
"type": "string"
}
}
},
"interfaces": {
"type": "array",
"description": "Network interfaces (order matters)",
"items": {
"type": "object",
"required": ["name"],
"properties": {
"name": {
"type": "string",
"pattern": "^(eth[0-9]+|lo)$"
},
"network": {
"type": "string"
},
"loopback": {
"type": "boolean",
"description": "Indicates if this is a loopback interface"
},
"addresses": {
"type": "array",
"items": {
"type": "string",
"pattern": "^([0-9]{1,3}\\.){3}[0-9]{1,3}/([0-9]|[1-2][0-9]|3[0-2])$"
}
}
},
"anyOf": [
{
"required": ["network"]
},
{
"required": ["loopback"],
"properties": {
"loopback": { "const": true }
}
}
]
}
},
"routing": {
"type": "object",
"properties": {
"stack": {
"type": "string",
"enum": ["bird", "frr", "linux"]
},
"protocols": {
"type": "object",
"properties": {
"static": {
"type": "object",
"properties": {
"routes": {
"type": "array",
"items": {
"type": "object",
"required": ["to", "via"],
"properties": {
"to": {
"type": "string",
"pattern": "^([0-9]{1,3}\\.){3}[0-9]{1,3}/([0-9]|[1-2][0-9]|3[0-2])|0\\.0\\.0\\.0/0$"
},
"via": {
"type": "string",
"pattern": "^([0-9]{1,3}\\.){3}[0-9]{1,3}$"
}
}
}
}
}
},
"ospf": {
"type": "object",
"properties": {
"enabled": { "type": "boolean", "default": false },
"impl": {
"type": "string",
"enum": ["bird", "frr"]
}
}
},
"rip": {
"type": "object",
"properties": {
"enabled": { "type": "boolean", "default": false }
}
},
"bgp": {
"type": "object",
"properties": {
"enabled": { "type": "boolean", "default": false },
"asn": { "type": "integer" }
}
}
}
}
}
},
"services": {
"type": "array",
"items": {
"type": "object",
"required": ["name"],
"properties": {
"name": { "type": "string" },
"enabled": { "type": "boolean", "default": true }
}
}
}
}
}
},
"switching": {
"type": "array",
"description": "L2 switching configuration",
"items": {
"type": "object",
"required": ["node", "impl"],
"properties": {
"node": {
"type": "string",
"description": "Switch node name"
},
"impl": {
"type": "string",
"enum": ["linux-bridge", "ovs"]
},
"bridges": {
"type": "array",
"items": {
"type": "object",
"required": ["name"],
"properties": {
"name": { "type": "string" },
"ports": {
"type": "array",
"items": {
"type": "object",
"required": ["if"],
"properties": {
"if": {
"type": "string",
"pattern": "^eth[0-9]+$"
},
"vlan": {
"type": "object",
"properties": {
"mode": {
"type": "string",
"enum": ["access", "trunk"]
},
"allowed": {
"type": "array",
"items": {
"type": "integer",
"minimum": 1,
"maximum": 4094
}
},
"vid": {
"type": "integer",
"minimum": 1,
"maximum": 4094
}
}
}
}
}
},
"svis": {
"type": "array",
"description": "Switched Virtual Interfaces (SVI)",
"items": {
"type": "object",
"required": ["vid", "ifname"],
"properties": {
"vid": {
"type": "integer",
"minimum": 1,
"maximum": 4094
},
"ifname": {
"type": "string",
"pattern": "^[a-zA-Z0-9._-]+$"
},
"addresses": {
"type": "array",
"items": {
"type": "string",
"pattern": "^([0-9]{1,3}\\.){3}[0-9]{1,3}/([0-9]|[1-2][0-9]|3[0-2])$"
}
}
}
}
}
}
}
}
}
}
},
"firewall": {
"type": "array",
"description": "Firewall configuration",
"items": {
"type": "object",
"required": ["node", "rules"],
"properties": {
"node": { "type": "string" },
"rules": {
"type": "array",
"items": {
"type": "object",
"required": ["id", "action"],
"properties": {
"id": {
"type": "string",
"pattern": "^[a-zA-Z0-9_-]+$"
},
"match": {
"type": "object",
"properties": {
"in_if": { "type": "string" },
"out_if": { "type": "string" },
"src": {
"type": "string",
"pattern": "^([0-9]{1,3}\\.){3}[0-9]{1,3}(/([0-9]|[1-2][0-9]|3[0-2]))?$"
},
"dst": {
"type": "string",
"pattern": "^([0-9]{1,3}\\.){3}[0-9]{1,3}(/([0-9]|[1-2][0-9]|3[0-2]))?$"
},
"ip_proto": {
"type": "string",
"enum": ["tcp", "udp", "icmp", "ospf", "all"]
},
"l4": {
"type": "object",
"properties": {
"proto": {
"type": "string",
"enum": ["tcp", "udp"]
},
"sport": {
"type": ["integer", "array"],
"items": { "type": "integer" },
"minimum": 1,
"maximum": 65535
},
"dport": {
"type": ["integer", "array"],
"items": { "type": "integer" },
"minimum": 1,
"maximum": 65535
},
"dports": {
"type": ["integer", "array"],
"items": { "type": "integer" },
"minimum": 1,
"maximum": 65535
}
}
}
}
},
"action": {
"type": "string",
"enum": ["accept", "drop", "reject"]
}
}
}
}
}
}
},
"profiles": {
"type": "array",
"description": "Configuration profiles/overlays",
"items": {
"type": "object",
"required": ["name", "patches"],
"properties": {
"name": { "type": "string" },
"description": { "type": "string" },
"patches": {
"type": "array",
"items": {
"type": "object",
"required": ["op", "path"],
"properties": {
"op": {
"type": "string",
"enum": ["add", "replace", "remove"]
},
"path": { "type": "string" },
"value": true,
"where": {
"type": "object",
"description": "Conditional matching for patches"
},
"patch": true
}
}
}
}
}
},
"tests": {
"type": "array",
"description": "Validation tests",
"items": {
"type": "object",
"required": ["name", "from", "to", "expect"],
"properties": {
"name": { "type": "string" },
"from": { "type": "string" },
"to": {
"type": ["string", "object"],
"properties": {
"ip": {
"type": "string",
"pattern": "^([0-9]{1,3}\\.){3}[0-9]{1,3}$"
},
"node": { "type": "string" }
}
},
"expect": {
"type": "string",
"pattern": "^(success|failure|timeout|via .*)$"
}
}
}
}
}
}
flake.nix
+3
View File
@@ -15,6 +15,9 @@
python312
python312Packages.virtualenv
python312Packages.pip
graphviz
d2
];
};
});
host-host.csv
-6
View File
@@ -1,6 +0,0 @@
name,"type","interfaces","ip","endpoints"
PC1,"host","int1","",""
PC2,"host","int2","",""
int1,"interface","","10.0.12.1/24",""
int2,"interface","","10.0.12.2/24",""
link,"wire","","","int1,int2"
1 name type interfaces ip endpoints
2 PC1 host int1
3 PC2 host int2
4 int1 interface 10.0.12.1/24
5 int2 interface 10.0.12.2/24
6 link wire int1,int2
output.yaml
-16
View File
@@ -1,16 +0,0 @@
links:
- endpoints:
- PC1
- PC2
meta:
id: /home/krosh/Documents/Github/network-diagrams-tool/host-host.csv
name: /home/krosh/Documents/Github/network-diagrams-tool/host-host.csv
nodes:
- interfaces:
- ip: 10.0.12.1/24
name: PC1
role: host
- interfaces:
- ip: 10.0.12.2/24
name: PC2
role: host
pyproject.toml
+19
View File
@@ -0,0 +1,19 @@
[project]
name = "netdiag"
version = "0.2.0"
[project.scripts]
netdiag = "netdiag.base:run"
[tool.setuptools]
package-dir = {"" = "src"}
[tool.setuptools.packages.find]
where = ["src"]
[tool.black]
line-length = 88
[tool.isort]
profile = "black"
line_length = 88
requirements.txt
+1
View File
@@ -0,0 +1 @@
pyaml
schema.yaml
-21
View File
@@ -1,21 +0,0 @@
meta:
id: "lab-02-basic-utils"
name: "Basic Network Utilities & Traffic Monitoring"
description: "Two hosts connected directly to each other."
links:
- endpoints: ["PC1", "PC2"]
nodes:
- name: PC1
role: host
interfaces:
- ip: "10.0.12.1/24"
- name: PC2
role: host
interfaces:
- ip: "10.0.12.2/24"
# Optional:
# - "ip link set dev eth1 down"
src/check_correct.py
-22
View File
@@ -1,22 +0,0 @@
from parse import parse_csv_to_structure
import sys
from structures import Host, Wire
def check_correct(data):
assert isinstance(data, dict)
for key, value in data.items():
assert isinstance(key, str)
assert isinstance(value, dict)
# for subkey, subvalue in value.items():
# assert isinstance(subkey, str)
# assert isinstance(subvalue, str)
if __name__ == "__main__":
data = parse_csv_to_structure(sys.argv[1])
if check_correct(data):
print("yay!")
print(data)
src/make_res.py
-81
View File
@@ -1,81 +0,0 @@
from check_correct import check_correct
from parse import parse_csv_to_structure
import sys
from structures import Host, Wire, Interface
import yaml
"""
meta:
id: "lab-02-basic-utils"
name: "Basic Network Utilities & Traffic Monitoring"
description: "Two hosts connected directly to each other."
links:
- endpoints: ["PC1", "PC2"]
nodes:
- name: PC1
role: host
interfaces:
- ip: "10.0.12.1/24"
- name: PC2
role: host
interfaces:
- ip: "10.0.12.2/24"
"""
def make_res(data, name):
res = {
"meta": {
"id": name,
"name": name,
},
"links": [],
"nodes": [],
}
host_int = {}
for key, value in data.items():
if value.get("type") == "host":
value["name"] = key
host = Host(value)
node_entry = {
"name": host.name,
"role": "host",
"interfaces": [
{"ip": Interface(data[iface]).ip_address}
for iface in host.interfaces
],
}
res["nodes"].append(node_entry)
host_int[host.interfaces[0]] = host.name
for key, value in data.items():
if value.get("type") == "wire":
wire = Wire(value)
link_entry = {
"endpoints": [
host_int[wire.endpoints[0]],
host_int[wire.endpoints[1]],
]
}
res["links"].append(link_entry)
return res
if __name__ == "__main__":
data = parse_csv_to_structure(sys.argv[1])
print(data)
if check_correct(data):
print("Data is correct.")
res = make_res(data, sys.argv[1])
print(res)
with open("output.yaml", "w") as f:
yaml.dump(res, f)
src/netdiag/__init__.py
View File
src/netdiag/__main__.py
+3
View File
@@ -0,0 +1,3 @@
from .base import run
raise SystemExit(run())
src/netdiag/args.py
+22
View File
@@ -0,0 +1,22 @@
import argparse
def parse_args(argv=None) -> argparse.Namespace:
parser = argparse.ArgumentParser(
description="Network Diagrams Tool - Generate network diagrams from CSV input"
)
parser.add_argument(
"-i",
"--input",
type=str,
default="data/input/table.csv",
help="Path to the input CSV file (default: data/input/table.csv)",
)
parser.add_argument(
"-o",
"--output",
type=str,
default="data/output",
help="Directory for output files (default: data/output)",
)
return parser.parse_args(args=argv)
src/netdiag/base.py
+32
View File
@@ -0,0 +1,32 @@
import logging
from pathlib import Path
from .args import parse_args
from .output.d2 import generate_d2_diagram
from .output.file_convert import make_yaml
from .output.graphviz import generate_diagram
from .parse import parse_csv
from .parse.convert_raw import (
convert_raw_topology,
)
logging.basicConfig(
level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s"
)
def run(argv: list[str] | None = None) -> None:
args = parse_args(argv)
raw_devices = parse_csv(Path(args.input))
topology = convert_raw_topology(raw_devices)
generate_diagram(topology, Path(args.output) / "diagram.png")
make_yaml(topology, Path(args.output) / "topology.yaml")
# generate_d2_diagram(topology, Path(args.output) / "diagram.d2")
logging.info("All tasks completed successfully.")
if __name__ == "__main__":
run()
src/netdiag/domain/__init__.py
View File
src/netdiag/domain/models.py
+193
View File
@@ -0,0 +1,193 @@
from typing import Any, Dict, List, Optional
# ===========================
class Interface:
name: str
ip_address: Optional[str]
network: Optional[str]
default_gateway: Optional[str]
# ---
device: "Device" # set by Device.add_interface() when the interface is added to a device
def __init__(
self,
name: str,
ip_address: Optional[str] = None,
network: Optional[str] = None,
default_gateway: Optional[str] = None,
):
if not isinstance(name, str):
raise ValueError("Interface 'name' must be a string")
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 default_gateway is not None and not isinstance(default_gateway, str):
raise ValueError("Interface 'default_gateway' must be a string or None")
self.name = name
self.ip_address = ip_address
self.network = network
self.default_gateway = default_gateway
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
# ===========================
class Device:
name: str
role: str = "device"
interfaces: Dict[str, Interface]
def __init__(self, name: str):
if not isinstance(name, str):
raise ValueError("Device 'name' must be a string")
self.name = name
self.interfaces = dict()
def add_interface(self, interface: Interface):
interface.device = (
self # set the device attribute of the interface to this device
)
self.interfaces[interface.name] = interface
def rm_interface(self, interface: Interface):
if interface.name in self.interfaces:
del self.interfaces[interface.name]
# interface.device = None # clear the device attribute of the interface
del interface
else:
raise ValueError(
f"Interface '{interface.name}' not found in device '{self.name}'"
)
def __repr__(self) -> str:
return f"Device(name={self.name}, interfaces={self.interfaces})"
class Host(Device):
role: str = "host"
class Router(Device):
role: str = "router"
class Switch(Device):
role: str = "switch"
# ===========================
class Network:
name: str
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")
if vlan is not None and not isinstance(vlan, str):
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):
raise ValueError("Argument must be an instance of Interface")
if interface in self.interfaces:
raise ValueError(
f"Interface '{interface.name}' already exists in network '{self.name}'"
)
interface.network = (
self.name # set the network attribute of the interface to this network
)
self.interfaces.append(interface)
def rm_interface(self, interface: Interface):
if interface in self.interfaces:
self.interfaces.remove(interface)
interface.network = None # clear the network attribute of the interface
else:
raise ValueError(
f"Interface '{interface.name}' not found in network '{self.name}'"
)
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})"
class Topology:
devices: Dict[str, Device]
networks: Dict[str, Network]
def __init__(self):
self.devices = dict()
self.networks = dict()
def add_device(self, device: Device):
if not isinstance(device, Device):
raise ValueError("Argument must be an instance of Device")
if device.name in self.devices:
raise ValueError(
f"Device with name '{device.name}' already exists in topology"
)
self.devices[device.name] = device
def rm_device(self, device: Device):
if device.name in self.devices:
del self.devices[device.name]
else:
raise ValueError(f"Device with name '{device.name}' not found in topology")
def add_network(self, network: Network):
if not isinstance(network, Network):
raise ValueError("Argument must be an instance of Network")
if network.name in self.networks:
raise ValueError(
f"Network with name '{network.name}' already exists in topology"
)
self.networks[network.name] = network
def rm_network(self, network: Network):
if network.name in self.networks:
del self.networks[network.name]
else:
raise ValueError(
f"Network with name '{network.name}' not found in topology"
)
def __repr__(self) -> str:
return f"Topology(devices={self.devices}, networks={self.networks})"
src/netdiag/domain/validation.py
View File
src/netdiag/output/__init__.py
View File
src/netdiag/output/d2.py
+137
View File
@@ -0,0 +1,137 @@
import shutil
import subprocess
from pathlib import Path
from ..domain.models import Topology
#! WIP
"""
D2 example:
com_left; com_right
VLAN 2: {
PC2: |md
# PC2
10.0.0.2/24
|
PC2.shape: rectangle
}
VLAN 3: {
PC3: |md
# PC3
10.0.0.3/24
|
PC3.shape: rectangle
}
VLAN 4: {
PC1: |md
# PC1
10.0.0.1/24
|
PC1.shape: rectangle
PC4: |md
# PC4
10.0.0.4/24
|
PC4.shape: rectangle
}
com_left -- com_right : {
source-arrowhead.label: eth1
target-arrowhead.label: eth1
}
VLAN 2.PC2 -- com_left : {
source-arrowhead.label: eth1
target-arrowhead.label: eth3
}
VLAN 4.PC1 -- com_left : {
source-arrowhead.label: eth1
target-arrowhead.label: eth2
}
VLAN 4.PC4 -- com_right : {
source-arrowhead.label: eth1
target-arrowhead.label: eth2
}
VLAN 3.PC3 -- com_right : {
source-arrowhead.label: eth1
target-arrowhead.label: eth3
}
"""
# https://d2lang.com/tour/themes/
THEME_NUMBER = 200
# https://icons.terrastruct.com/
icons = {
"router": "https://icons.terrastruct.com/tech%2Frouter.svg",
"host": "https://icons.terrastruct.com/tech%2F065-monitor-4.svg",
"switch": "https://icons.terrastruct.com/tech%2Fswitch.svg",
"device": "https://icons.terrastruct.com/azure%2FCompute%20Service%20Color%2FVM%2FVM-non-azure.svg",
"vlan": "https://icons.terrastruct.com/azure%2FNetworking%20Service%20Color%2FVirtual%20Networks.svg",
}
def _check_d2_installed() -> bool:
return shutil.which("d2") is not None
def generate_d2_diagram(topology: Topology, output_path: Path) -> None:
if not _check_d2_installed():
raise EnvironmentError(
"D2 is not installed or 'd2' command is not found in PATH. Please install D2 to use this feature."
)
diagram_path = Path()
picture_path = Path()
if output_path.is_dir():
diagram_path = output_path / "diagram.d2"
picture_path = output_path / "diagram.png"
else:
diagram_path = output_path.with_suffix(".d2")
picture_path = output_path.with_suffix(".png")
file: list[str] = []
# some logic
# ---
with open(diagram_path, "w", encoding="utf-8") as f:
f.write("\n".join(file))
_generate_picture(diagram_path, picture_path)
def _generate_picture(diagram: Path, output_path: Path) -> None:
res = subprocess.run(
["d2", "validate", str(diagram)],
check=True,
capture_output=True,
)
if res.returncode != 0:
raise RuntimeError(
f"D2 validation failed (code {res.returncode})\n"
f"Stdout: {res.stdout.decode()}\n"
f"Stderr: {res.stderr.decode()}"
)
res = subprocess.run(
["d2", f"--theme={THEME_NUMBER}", str(diagram), str(output_path)],
check=True,
capture_output=True,
)
if res.returncode != 0:
raise RuntimeError(
f"D2 diagram generation failed (code {res.returncode})\n"
f"Stdout: {res.stdout.decode()}\n"
f"Stderr: {res.stderr.decode()}"
)
src/netdiag/output/file_convert.py
+72
View File
@@ -0,0 +1,72 @@
from pathlib import Path
import yaml
from ..domain.models import Topology
def make_yaml(topology: Topology, output_path: Path) -> None:
data = dict()
data["meta"] = {
"id": output_path.name,
"name": output_path.name,
}
data["networks"] = []
for _, network in topology.networks.items():
interfaces_with_device = [
iface for iface in network.interfaces if iface.device is not None
]
if len(interfaces_with_device) >= 2:
data["networks"].append(
{
network.name: [
f"{iface.device.name}.{iface.name}"
for iface in interfaces_with_device
]
}
)
data["nodes"] = []
for _, device in topology.devices.items():
data["nodes"].append(
{
"role": device.role,
"name": device.name,
"interfaces": [
{
interface_name: [
{
"ip": (
interface.ip_address
if interface.ip_address
else None
),
"network": (
interface.network if interface.network else None
),
"gateway": (
interface.default_gateway
if interface.default_gateway
else None
),
}
]
}
for interface_name, interface in device.interfaces.items()
],
}
)
with open(str(output_path), "w", encoding="utf-8") as f:
yaml.safe_dump(
data,
f,
allow_unicode=True,
sort_keys=False,
default_flow_style=False,
indent=2,
)
src/netdiag/output/graphviz.py
+40
View File
@@ -0,0 +1,40 @@
from ..domain.models import Topology
from pathlib import Path
import graphviz
import shutil
def _check_graphviz_installed() -> bool:
return shutil.which("dot") is not None
def generate_diagram(topology: Topology, output_path: Path) -> None:
if not _check_graphviz_installed():
raise EnvironmentError(
"Graphviz is not installed or 'dot' command is not found in PATH. Please install Graphviz to use this feature."
)
dot = graphviz.Graph(name="Network Topology", format="png", engine="neato")
dot.attr(overlap="false", splines="true")
for device in topology.devices.values():
shape = "box"
dot.node(device.name, label=device.name, shape=shape)
for network in topology.networks.values():
interfaces_with_device = [
iface for iface in network.interfaces if iface.device is not None
]
for i, iface_a in enumerate(interfaces_with_device):
for iface_b in interfaces_with_device[i + 1 :]:
label = network.name or ""
dot.edge(iface_a.device.name, iface_b.device.name, label=label)
output_path = (
output_path.with_suffix("") if output_path.suffix == ".png" else output_path
)
dot.render(str(output_path), cleanup=True)
return
src/netdiag/parse/__init__.py
+33
View File
@@ -0,0 +1,33 @@
import csv
import logging
from pathlib import Path
from typing import Any, Dict, List
class RawDevices:
id: int
fields: Dict[str, Any]
def __init__(self, id: int, fields: Dict[str, Any]):
if not isinstance(id, int):
raise ValueError("RawDevices 'id' must be an integer")
if not isinstance(fields, dict):
raise ValueError("RawDevices 'fields' must be a dictionary")
if any(not isinstance(k, str) for k in fields.keys()):
raise ValueError("RawDevices 'fields' keys must be strings")
self.id = id
self.fields = fields
def __repr__(self) -> str:
return f"RawDevices(id={self.id}, fields={self.fields})"
def parse_csv(file_path: Path, delimiter: str = ",") -> List[RawDevices]:
logging.info(f"Parsing CSV file: {file_path}")
with open(file_path, mode="r", encoding="utf-8") as csvfile:
reader = csv.DictReader(csvfile, delimiter=delimiter)
devices = [RawDevices(idx + 1, row) for idx, row in enumerate(reader)]
return devices
src/netdiag/parse/convert_raw.py
+155
View File
@@ -0,0 +1,155 @@
import logging
from ..domain.models import (
Device,
Host,
Interface,
Network,
Router,
Switch,
Topology,
VirtualInterface,
)
from . import RawDevices
name_matching = {
"DEVICE_TYPE": "Role",
"DEVICE_NAME": "Name",
"INTERFACE_NAME": "Interface",
"NETWORK_NAME": "Network",
"VLAN": "VLAN",
"NETWORK_IP": "Network IP",
"SUBNET_MASK": "Mask",
"IP_ADDRESS": "Device IP",
"DEFAULT_GATEWAY": "Default Gateway",
# --- In-cell fields ---
"TRUNK": "trunk",
"HOST": "Host",
"ROUTER": "Router",
"SWITCH": "Switch",
}
def parse_devices(raw_devices: list[RawDevices]) -> list[Device]:
devices = []
for raw_device in raw_devices:
device_type = raw_device.fields.get(name_matching["DEVICE_TYPE"], "").strip()
device_name = raw_device.fields.get(name_matching["DEVICE_NAME"], "").strip()
if not device_type or not device_name:
raise ValueError(
f"Device with ID {raw_device.id} is missing required fields 'DEVICE_TYPE' and 'DEVICE_NAME'"
)
if device_name in [d.name for d in devices]:
continue
if device_type == name_matching["HOST"]:
device = Host(name=device_name)
elif device_type == name_matching["ROUTER"]:
device = Router(name=device_name)
elif device_type == name_matching["SWITCH"]:
device = Switch(name=device_name)
else:
raise ValueError(
f"Device with ID {raw_device.id} has unrecognized DEVICE_TYPE '{device_type}'"
)
devices.append(device)
return devices
def add_interfaces(devices: list[Device], raw_devices: list[RawDevices]) -> None:
for raw_device in raw_devices:
device_name = raw_device.fields.get(name_matching["DEVICE_NAME"], "").strip()
device = next(
(d for d in devices if d.name == device_name), None
) # some other way?
if not device:
raise ValueError(
f"Device with name '{device_name}' not found for interface parsing"
)
interface = Interface(
name=raw_device.fields.get(name_matching["INTERFACE_NAME"], "").strip(),
ip_address=raw_device.fields.get(name_matching["IP_ADDRESS"], "").strip(),
network=raw_device.fields.get(name_matching["NETWORK_NAME"], "").strip(),
default_gateway=raw_device.fields.get(
name_matching["DEFAULT_GATEWAY"], ""
).strip(),
)
device.add_interface(interface)
return
def parse_networks(
devices: list[Device], raw_devices: list[RawDevices]
) -> list[Network]:
networks = []
for raw_device in raw_devices:
network_name = raw_device.fields.get(name_matching["NETWORK_NAME"], "").strip()
if not network_name:
continue
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)
if not network.vlan and vlan:
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)
return networks
def assign_interfaces_to_networks(
networks: list[Network], devices: list[Device]
) -> None:
for network in networks:
for device in devices:
for _, interface in device.interfaces.items():
if interface.network == network.name:
network.add_interface(interface)
return
def convert_raw_topology(raw_devices: list[RawDevices]) -> Topology:
topology = Topology()
logging.info(f"Parsing {len(raw_devices)} raw devices to devices")
devices = parse_devices(raw_devices)
logging.info(f"Parsing interfaces from raw devices")
add_interfaces(devices, raw_devices)
for device in devices:
topology.add_device(device)
logging.info(f"Parsing networks from raw devices")
networks = parse_networks(devices, raw_devices)
assign_interfaces_to_networks(networks, devices)
for network in networks:
topology.networks[network.name] = network
return topology
src/old/structures copy.py
-61
View File
@@ -1,61 +0,0 @@
from dataclasses import dataclass, field
from typing import Dict, Any
@dataclass
class Device:
name: str
model: str
properties: Dict[str, Any] = field(default_factory=dict)
def __post_init__(self):
if not self.name:
raise ValueError("Device name cannot be empty")
if not self.model:
raise ValueError("Device model cannot be empty")
if not isinstance(self.properties, dict):
raise TypeError("properties must be a dict")
@dataclass
class Switch(Device):
def __post_init__(self):
super().__post_init__()
if "ports" not in self.properties:
raise ValueError(f"Switch '{self.name}' must define 'ports'")
if (
not isinstance(self.properties["ports"], int)
or self.properties["ports"] <= 0
):
raise ValueError("'ports' must be a positive integer")
@dataclass
class Router(Device):
def __post_init__(self):
super().__post_init__()
if "interfaces" not in self.properties:
raise ValueError(f"Router '{self.name}' must define 'interfaces'")
if (
not isinstance(self.properties["interfaces"], int)
or self.properties["interfaces"] <= 0
):
raise ValueError("'interfaces' must be a positive integer")
@dataclass
class Host(Device):
def __post_init__(self):
super().__post_init__()
ip = self.properties.get("ip")
if not ip or not isinstance(ip, str):
raise ValueError(f"Host '{self.name}' must contain a valid 'ip' string")
@dataclass
class Wire:
from_device: Device
to_device: Device
src/old/structures.py
-182
View File
@@ -1,182 +0,0 @@
from dataclasses import dataclass, field
from typing import List, Optional, Dict
# ===========================
# Interface Classes
# ===========================
@dataclass
class Interface:
name: str
ip_address: Optional[str] = None
mac_address: Optional[str] = None
speed: Optional[str] = None
duplex: Optional[str] = None
def __init__(self, fields):
self.name = fields.get("name")
self.ip_address = fields.get("ip")
self.mac_address = fields.get("mac")
self.speed = fields.get("speed")
self.duplex = fields.get("duplex")
self.__post_init__()
def __post_init__(self):
if not isinstance(self.name, str) or not self.name:
raise ValueError("Interface 'name' must be a non-empty string")
@dataclass
class VirtualInterface(Interface):
vlan_id: int = 0
parent_physical: Optional[Interface] = None
def __init__(self, fields):
super().__init__(fields)
self.vlan_id = fields.get("vlanID", fields.get("vlan_id", 0))
self.parent_physical = fields.get(
"parent_physical", fields.get("parentPhysical", fields.get("parent"))
)
self.__post_init__()
def __post_init__(self):
super().__post_init__()
if not isinstance(self.vlan_id, int) or self.vlan_id < 0:
raise ValueError(
"VirtualInterface 'vlan_id' must be a non-negative integer"
)
if not isinstance(self.parent_physical, Interface):
raise TypeError(
"VirtualInterface 'parent_physical' must reference an Interface"
)
# ===========================
# Network / Subnet
# ===========================
@dataclass
class Network:
cidr: str
gateway: Optional[str] = None
interfaces: List[Interface] = field(default_factory=list)
def __init__(self, fields):
self.cidr = fields.get("cidr")
self.gateway = fields.get("gateway")
interfaces = fields.get("interfaces", [])
if isinstance(interfaces, str):
interfaces = [iface for iface in interfaces.split(",") if iface]
self.interfaces = interfaces
self.__post_init__()
def __post_init__(self):
if not isinstance(self.cidr, str) or not self.cidr:
raise ValueError("Network 'cidr' must be a non-empty string")
if self.gateway is not None and not isinstance(self.gateway, str):
raise ValueError("Network 'gateway' must be a string")
# ===========================
# Device Base Class (Abstract)
# ===========================
@dataclass
class Device:
name: str
mgmt_ip: Optional[str] = None
interfaces: List[Interface] = field(default_factory=list)
def __init__(self, fields):
self.name = fields.get("name")
self.mgmt_ip = fields.get("mgmtIP")
self.interfaces = [f for f in fields.get("interfaces").split(",") if f]
self.__post_init__()
def __post_init__(self):
if not isinstance(self.name, str) or not self.name:
raise ValueError("Device 'name' must be a non-empty string")
if self.mgmt_ip is not None and not isinstance(self.mgmt_ip, str):
raise ValueError("Device 'mgmtIP' must be a string")
# ===========================
# Host / Router / Switch
# ===========================
@dataclass
class Host(Device):
operating_system: Optional[str] = None
def __init__(self, fields):
super().__init__(fields)
self.operating_system = fields.get("operatingSystem")
self.__post_init__()
def __post_init__(self):
super().__post_init__()
if not isinstance(self.operating_system, str):
raise ValueError("Host 'operatingSystem' must be a string")
@dataclass
class Router(Device):
routing_table: Dict[str, str] = field(default_factory=dict)
routing_protocols: List[str] = field(default_factory=list)
def __init__(self, fields):
super().__init__(fields)
self.routing_table = fields.get("routingTable", {})
self.routing_protocols = fields.get("routingProtocols", [])
self.__post_init__()
def __post_init__(self):
super().__post_init__()
@dataclass
class Switch(Device):
mac_table: Dict[str, str] = field(default_factory=dict)
vlan_database: Dict[int, str] = field(default_factory=dict)
def __init__(self, fields):
super().__init__(fields)
self.mac_table = fields.get("macTable", {})
self.vlan_database = fields.get("vlanDatabase", {})
self.__post_init__()
def __post_init__(self):
super().__post_init__()
# ===========================
# Physical Link (Wire)
# ===========================
@dataclass
class Wire:
id: str
endpoints: List[Interface] # Must contain exactly 2 interfaces
bandwidth: Optional[str] = None
def __init__(self, fields):
self.id = fields.get("name")
self.endpoints = [ep for ep in fields.get("endpoints").split(",") if ep]
self.bandwidth = fields.get("bandwidth")
self.__post_init__()
def __post_init__(self):
if not isinstance(self.id, str) or not self.id:
raise ValueError("Wire 'id' must be a non-empty string")
if len(self.endpoints) != 2:
raise ValueError(
"Wire 'endpoints' must contain exactly 2 Interface objects"
)
if not all(isinstance(i, Interface) for i in self.endpoints):
raise TypeError("Wire 'endpoints' must be Interface objects")
src/parse.py
-31
View File
@@ -1,31 +0,0 @@
import csv
def parse_csv_to_structure(path):
result = {}
with open(path, newline="") as f:
reader = csv.reader(f)
rows = list(reader)
headers = rows[0][1:]
for row in rows[1:]:
obj = row[0].strip()
if not obj:
continue
result[obj] = {}
for header, value in zip(headers, row[1:]):
value = value.strip()
if value:
result[obj][header] = value
return result
if __name__ == "__main__":
data = parse_csv_to_structure("input.csv")
print(data)
src/structures.py
-134
View File
@@ -1,134 +0,0 @@
from dataclasses import dataclass, field
from typing import List, Optional, Dict
# ===========================
# Network / Subnet
# ===========================
@dataclass
class Network:
name: str
vlan_id : Optional[int] = None
ip: Optional[str] = None
subnet_mask: Optional[str] = None
def __init__(self, fields):
self.name = fields.get("name")
self.vlan_id = fields.get("vlan", fields.get("vlan_id"))
self.ip = fields.get("ip")
self.subnet_mask = fields.get("mask", fields.get("subnet_mask"))
self.__post_init__()
def __post_init__(self):
if not isinstance(self.name, str) or not self.name:
raise ValueError("Network 'name' must be a non-empty string")
# ===========================
# Interface Classes
# ===========================
@dataclass
class Interface:
class IP:
def __init__(self, ip_str: str, default_gateway: Optional[str] = None):
self.ip_str = ip_str
self.default_gateway = default_gateway
def __str__(self):
return self.ip_str
name: str
mode: Optional[str] = None
network: Optional[Network] = None
ips: Optional[List[IP]] = None
mac_address: Optional[str] = None
speed: Optional[str] = None
def __init__(self, fields):
self.name = fields.get("name")
ips = fields.get("ips", [])
if isinstance(ips, str):
ips = [self.IP(ip_str=ip.strip()) for ip in ips.split(",") if ip.strip()]
self.ips = ips
# возможно передаём просто field
self.network = Network(fields.get("network", {})) if fields.get("network") else None
self.mac_address = fields.get("mac")
self.speed = fields.get("speed")
self.__post_init__()
def __post_init__(self):
if not isinstance(self.name, str) or not self.name:
raise ValueError("Interface 'name' must be a non-empty string")
# ===========================
# Device Base Class (Abstract)
# ===========================
@dataclass
class Device:
name: str
interfaces: List[Interface] = field(default_factory=list)
def __init__(self, fields):
self.name = fields.get("name")
# тут точно понять, какие поля передаются
self.interfaces = [Interface(iface) for iface in fields.get("interfaces", [])]
self.__post_init__()
def __post_init__(self):
if not isinstance(self.name, str) or not self.name:
raise ValueError("Device 'name' must be a non-empty string")
# ===========================
# Host / Router / Switch
# ===========================
@dataclass
class Host(Device):
# operating_system: Optional[str] = None
def __init__(self, fields):
super().__init__(fields)
# self.operating_system = fields.get("operatingSystem")
self.__post_init__()
def __post_init__(self):
super().__post_init__()
@dataclass
class Router(Device):
# routing_table: Dict[str, str] = field(default_factory=dict)
# routing_protocols: List[str] = field(default_factory=list)
def __init__(self, fields):
super().__init__(fields)
# self.routing_table = fields.get("routingTable", {})
# self.routing_protocols = fields.get("routingProtocols", [])
self.__post_init__()
def __post_init__(self):
super().__post_init__()
@dataclass
class Switch(Device):
# mac_table: Dict[str, str] = field(default_factory=dict)
# vlan_database: Dict[int, str] = field(default_factory=dict)
def __init__(self, fields):
super().__init__(fields)
# self.mac_table = fields.get("macTable", {})
# self.vlan_database = fields.get("vlanDatabase", {})
self.__post_init__()
def __post_init__(self):
super().__post_init__()