diff --git a/.gitignore b/.gitignore index b7faf40..4414f4f 100644 --- a/.gitignore +++ b/.gitignore @@ -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/ diff --git a/.vscode/settings.json b/.vscode/settings.json deleted file mode 100644 index 69b361b..0000000 --- a/.vscode/settings.json +++ /dev/null @@ -1,3 +0,0 @@ -{ - "nixEnvSelector.nixFile": "${workspaceFolder}/flake.nix" -} \ No newline at end of file diff --git a/UML/structure.jpg b/UML/structure.jpg deleted file mode 100644 index 8f33bdb..0000000 Binary files a/UML/structure.jpg and /dev/null differ diff --git a/examples/input.csv b/examples/input.csv deleted file mode 100644 index 21a7b21..0000000 --- a/examples/input.csv +++ /dev/null @@ -1,3 +0,0 @@ - , prop1 , prop2 , prop3 -ob1 , val1 , , val2 -obj2 , , val3 , val4 \ No newline at end of file diff --git a/examples/template.yaml b/examples/template.yaml deleted file mode 100644 index e08905b..0000000 --- a/examples/template.yaml +++ /dev/null @@ -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 diff --git a/examples/topology-schema.json b/examples/topology-schema.json deleted file mode 100644 index 4baf645..0000000 --- a/examples/topology-schema.json +++ /dev/null @@ -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 .*)$" - } - } - } - } - } -} diff --git a/flake.nix b/flake.nix index 07fa9b8..a33b4fc 100644 --- a/flake.nix +++ b/flake.nix @@ -15,6 +15,9 @@ python312 python312Packages.virtualenv python312Packages.pip + + graphviz + d2 ]; }; }); diff --git a/host-host.csv b/host-host.csv deleted file mode 100644 index 94a6413..0000000 --- a/host-host.csv +++ /dev/null @@ -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" \ No newline at end of file diff --git a/output.yaml b/output.yaml deleted file mode 100644 index 220ed9a..0000000 --- a/output.yaml +++ /dev/null @@ -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 diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..6758f0a --- /dev/null +++ b/pyproject.toml @@ -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 diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..39e5bd8 --- /dev/null +++ b/requirements.txt @@ -0,0 +1 @@ +pyaml \ No newline at end of file diff --git a/schema.yaml b/schema.yaml deleted file mode 100644 index e47ba5b..0000000 --- a/schema.yaml +++ /dev/null @@ -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" diff --git a/src/check_correct.py b/src/check_correct.py deleted file mode 100755 index f417406..0000000 --- a/src/check_correct.py +++ /dev/null @@ -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) diff --git a/src/make_res.py b/src/make_res.py deleted file mode 100644 index ee4f06d..0000000 --- a/src/make_res.py +++ /dev/null @@ -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) diff --git a/src/netdiag/__init__.py b/src/netdiag/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/netdiag/__main__.py b/src/netdiag/__main__.py new file mode 100644 index 0000000..8654f3f --- /dev/null +++ b/src/netdiag/__main__.py @@ -0,0 +1,3 @@ +from .base import run + +raise SystemExit(run()) diff --git a/src/netdiag/args.py b/src/netdiag/args.py new file mode 100644 index 0000000..14e407c --- /dev/null +++ b/src/netdiag/args.py @@ -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) diff --git a/src/netdiag/base.py b/src/netdiag/base.py new file mode 100644 index 0000000..8aca63c --- /dev/null +++ b/src/netdiag/base.py @@ -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() diff --git a/src/netdiag/domain/__init__.py b/src/netdiag/domain/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/netdiag/domain/models.py b/src/netdiag/domain/models.py new file mode 100644 index 0000000..35e3c0c --- /dev/null +++ b/src/netdiag/domain/models.py @@ -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})" diff --git a/src/netdiag/domain/validation.py b/src/netdiag/domain/validation.py new file mode 100644 index 0000000..e69de29 diff --git a/src/netdiag/output/__init__.py b/src/netdiag/output/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/netdiag/output/d2.py b/src/netdiag/output/d2.py new file mode 100644 index 0000000..3225c71 --- /dev/null +++ b/src/netdiag/output/d2.py @@ -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()}" + ) diff --git a/src/netdiag/output/file_convert.py b/src/netdiag/output/file_convert.py new file mode 100644 index 0000000..e142111 --- /dev/null +++ b/src/netdiag/output/file_convert.py @@ -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, + ) diff --git a/src/netdiag/output/graphviz.py b/src/netdiag/output/graphviz.py new file mode 100644 index 0000000..fea6f11 --- /dev/null +++ b/src/netdiag/output/graphviz.py @@ -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 diff --git a/src/netdiag/parse/__init__.py b/src/netdiag/parse/__init__.py new file mode 100644 index 0000000..f316b0e --- /dev/null +++ b/src/netdiag/parse/__init__.py @@ -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 diff --git a/src/netdiag/parse/convert_raw.py b/src/netdiag/parse/convert_raw.py new file mode 100644 index 0000000..bbb4476 --- /dev/null +++ b/src/netdiag/parse/convert_raw.py @@ -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 diff --git a/src/old/structures copy.py b/src/old/structures copy.py deleted file mode 100644 index 48df349..0000000 --- a/src/old/structures copy.py +++ /dev/null @@ -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 diff --git a/src/old/structures.py b/src/old/structures.py deleted file mode 100644 index 8b239b6..0000000 --- a/src/old/structures.py +++ /dev/null @@ -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") diff --git a/src/parse.py b/src/parse.py deleted file mode 100644 index ea579a7..0000000 --- a/src/parse.py +++ /dev/null @@ -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) diff --git a/src/structures.py b/src/structures.py deleted file mode 100644 index dea9981..0000000 --- a/src/structures.py +++ /dev/null @@ -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__() -