feat: add network API and netplan configuration models with example usage

backend/src/Models/netplan.py

@@ -34,3 +34,28 @@ class NetworkConfig(BaseModel):
     renderer: Optional[str] = "networkd"
     ethernets: Dict[str, EthernetConfig] = Field(default_factory=dict)
     bridges: Dict[str, BridgeConfig] = Field(default_factory=dict)
+
+
+'''Example usage:
+{
+  "version": 2,
+  "renderer": "networkd",
+  "ethernets": {
+    "eth0": {
+      "dhcp4": false,
+      "addresses": ["192.168.10.20/24"],
+      "gateway4": "192.168.10.1",
+      "nameservers": {
+        "addresses": ["1.1.1.1", "8.8.8.8"]
+      }
+    },
+    "eth1": {},
+    "eth2": {}
+  },
+  "bridges": {
+    "br0": {
+      "interfaces": ["eth1", "eth2"],
+      "dhcp4": true
+    }
+  }
+}'''

backend/src/main.py

@@ -1,11 +1,20 @@
-from asyncio import subprocess
-from http.client import HTTPException
-import src.Models.netplan as netplan
 from fastapi import FastAPI
 from fastapi.middleware.cors import CORSMiddleware
 import os
+import src.Models.netplan as netplan
+import src.network_api as network_api
+from asyncio import subprocess
+from fastapi import HTTPException
 
-app = FastAPI()
+
+app = FastAPI(
+    title="MITM Webserver Backend",
+    description="Backend API for MITM Webserver",
+    version="1.0.0",
+    docs_url="/api/docs",
+    redoc_url="/api/redoc",
+    openapi_url="/api/openapi.json",
+)
 
 app.add_middleware(
     CORSMiddleware,
@@ -15,22 +24,47 @@ app.add_middleware(
     allow_headers=["*"],
 )
 
+
+# ---------------------
+#   Startup / Shutdown
+# ---------------------
+
+
+@app.on_event("shutdown")
+def shutdown_event():
+    network_api.shutdown_network_api()
+
+
+# ---------------------
+#   Basic Endpoints
+# ---------------------
+
 @app.get("/api/hello")
-def read_root():
+def hello():
     return {"message": "Hello from FastAPI 🎉"}
 
-
 @app.get("/api/versions")
-def read_root():
+def versions():
     message = os.popen("python --version").read().strip()
     return {"message": message}
 
 @app.get("/api/nft/ruleset")
-def read_root():
+def nft_ruleset():
     message = os.popen("sudo nft --json list ruleset").read().strip()
     return {"message": message}
 
 
+# ---------------------
+#   Routers
+# ---------------------
+
+app.include_router(network_api.router, prefix="/network", tags=["network"])
+
+
+# ---------------------
+#   Netplan Apply
+# ---------------------
+
 NETPLAN_FILE = "/etc/netplan/99-app.yaml"
 
 @app.post("/netplan/apply")
@@ -41,7 +75,6 @@ def apply_netplan(config: netplan.NetworkConfig):
         with open(NETPLAN_FILE, "w") as f:
             yaml.dump(data, f, sort_keys=False)
 
-        # Safety: rollback if config breaks connectivity
         result = subprocess.run(
             ["netplan", "try", "--timeout", "10"],
             capture_output=True,
@@ -57,4 +90,4 @@ def apply_netplan(config: netplan.NetworkConfig):
         return {"status": "ok", "output": result.stdout}
 
     except Exception as e:
-        raise HTTPException(status_code=500, detail=str(e))
+        raise HTTPException(status_code=500, detail=str(e))

backend/src/network_api.py

@@ -0,0 +1,327 @@
+from fastapi import APIRouter, Depends
+from pydantic import BaseModel, Field
+from typing import List, Optional
+from pyroute2 import IPRoute, NDB
+
+router = APIRouter()
+
+# Globals for lazy initialization
+ip: IPRoute | None = None
+ndb: NDB | None = None
+
+# ------------------------------
+# Pydantic models
+# ------------------------------
+
+class InterfaceAddress(BaseModel):
+    """
+    Represents an IP address assigned to a network interface.
+    """
+    family: str = Field(..., description="IP family: 'ipv4' or 'ipv6'.")
+    address: str = Field(..., description="The IP address assigned to the interface.")
+    prefixlen: int = Field(..., description="Subnet prefix length (e.g., 24 for 255.255.255.0).")
+
+
+class InterfaceInfo(BaseModel):
+    """
+    Represents a network interface with all its properties.
+    """
+    ifindex: int = Field(..., description="Interface index (unique identifier assigned by the kernel).")
+    name: str = Field(..., description="Interface name (e.g., 'eth0', 'enp38s0').")
+    state: str = Field(..., description="Operational state (e.g., 'UP', 'DOWN', 'UNKNOWN').")
+    mac: Optional[str] = Field(None, description="MAC address of the interface, if applicable.")
+    mtu: int = Field(..., description="Maximum Transmission Unit for the interface.")
+    flags: List[str] = Field(..., description="List of interface flags (e.g., ['BROADCAST', 'MULTICAST']).")
+    addresses: List[InterfaceAddress] = Field(..., description="List of IP addresses assigned to the interface.")
+
+class RouteInfo(BaseModel):
+    """
+    Represents a single routing table entry.
+    """
+
+    dst: Optional[str] = Field(
+        None, description="Destination network in CIDR notation (e.g., '192.168.1.0/24'). None means default route."
+    )
+
+    gateway: Optional[str] = Field(
+        None, description="Next-hop gateway IP address for this route. None if the route is directly connected."
+    )
+
+    prefsrc: Optional[str] = Field(
+        None, description="Preferred source IP to use when sending packets via this route."
+    )
+
+    oif: Optional[int] = Field(
+        None, description="Output interface index (ifindex) for this route. Can be used to look up the interface name."
+    )
+
+    ifname: Optional[str] = Field(
+        None, description="Name of the interface corresponding to `oif` (e.g., 'eth0')."
+    )
+
+    table: int = Field(
+        ..., description="Routing table ID (e.g., 254 = main, 255 = local)."
+    )
+
+    proto: Optional[int] = Field(
+        None,
+        description="Protocol of the route (numeric Linux codes, e.g., 2=kernel, 16=static)."
+    )
+
+    scope: Optional[int] = Field(
+        None,
+        description="Scope of the route: 0=global, 253=link, 254=host, 255=nowhere."
+    )
+
+    type: Optional[int] = Field(
+        None,
+        description="Type of the route (numeric code): 1=unicast, 2=local, 3=broadcast, 5=multicast."
+    )
+
+class BridgeInterfaceInfo(BaseModel):
+    ifindex: int = Field(..., description="Interface index of a bridge member")
+    ifname: str = Field(..., description="Interface name of a bridge member")
+    state: Optional[str] = Field(None, description="Operational state of the interface")
+    mtu: Optional[int] = Field(None, description="MTU of the interface")
+
+class BridgeInfo(BaseModel):
+    ifindex: int = Field(..., description="Interface index of the bridge")
+    ifname: str = Field(..., description="Bridge interface name")
+    state: Optional[str] = Field(None, description="Operational state of the bridge")
+    mtu: Optional[int] = Field(None, description="MTU of the bridge")
+    stp_state: Optional[int] = Field(None, description="STP (Spanning Tree Protocol) state of the bridge")
+    members: List[BridgeInterfaceInfo] = Field(default_factory=list, description="List of member interfaces of the bridge")
+
+# ------------------------------
+# Lazy Init Functions
+# ------------------------------
+
+def init_network_api():
+    global ip, ndb
+    if ip is None:
+        ip = IPRoute()
+    if ndb is None:
+        ndb = NDB()
+
+def shutdown_network_api():
+    global ip, ndb
+    if ip:
+        ip.close()
+        ip = None
+    if ndb:
+        ndb.close()
+        ndb = None
+
+def get_iproute():
+    if ip is None:
+        init_network_api()
+    return ip
+
+def get_ndb():
+    if ndb is None:
+        init_network_api()
+    return ndb
+
+# ------------------------------
+# Utility functions
+# ------------------------------
+
+def parse_addresses(addrs):
+    res = []
+    for a in addrs:
+        family = "ipv4" if a.get("family") == 2 else "ipv6"
+        res.append(
+            InterfaceAddress(
+                family=family,
+                address=a.get("address"),
+                prefixlen=a.get("prefixlen"),
+            )
+        )
+    return res
+
+def parse_flags(flags_int: int) -> list[str]:
+    """
+    Converts the integer flags from pyroute2 to human-readable list of strings.
+    """
+    flags_map = {
+        0x1: "UP",
+        0x2: "BROADCAST",
+        0x4: "DEBUG",
+        0x8: "LOOPBACK",
+        0x10: "POINTOPOINT",
+        0x20: "NOTRAILERS",
+        0x40: "RUNNING",
+        0x80: "NOARP",
+        0x100: "PROMISC",
+        0x200: "ALLMULTI",
+        0x400: "MASTER",
+        0x800: "SLAVE",
+        0x1000: "MULTICAST",
+        0x2000: "PORTSEL",
+        0x4000: "AUTOMEDIA",
+        0x8000: "DYNAMIC",
+        0x10000: "LOWER_UP",
+        0x20000: "DORMANT",
+        0x40000: "ECHO",
+    }
+    result = []
+    for bit, name in flags_map.items():
+        if flags_int & bit:
+            result.append(name)
+    return result
+
+# ------------------------------
+# Endpoints
+# ------------------------------
+
+@router.get("/interfaces", response_model=List[InterfaceInfo])
+def get_interfaces(ip: IPRoute = Depends(get_iproute)):
+    result = []
+    links = ip.get_links()
+    addresses = ip.get_addr()
+
+    addr_map = {}
+    for a in addresses:
+        ifindex = a.get("index")
+        addr_map.setdefault(ifindex, []).append(a)
+
+    for link in links:
+        attrs = dict(link["attrs"])
+        ifindex = link["index"]
+        addrs = addr_map.get(ifindex, [])
+
+        result.append(
+            InterfaceInfo(
+                ifindex=ifindex,
+                name=attrs.get("IFLA_IFNAME"),
+                state=attrs.get("IFLA_OPERSTATE", "unknown"),
+                mac=attrs.get("IFLA_ADDRESS"),
+                mtu=attrs.get("IFLA_MTU"),
+                flags=parse_flags(link.get("flags", 0)),
+                addresses=parse_addresses(addrs),
+            )
+        )
+    return result
+
+@router.get("/routes", response_model=List[RouteInfo])
+def get_routes(ip: IPRoute = Depends(get_iproute)):
+    routes = []
+    for r in ip.get_routes():
+        attrs = dict(r["attrs"])
+        dst = attrs.get("RTA_DST")
+        gateway = attrs.get("RTA_GATEWAY")
+        prefsrc = attrs.get("RTA_PREFSRC")
+        oif = r.get("oif")
+        ifname = None
+        if oif is not None:
+            # translate ifindex → name
+            link = ip.get_links(oif)[0]
+            ifname = dict(link["attrs"]).get("IFLA_IFNAME")
+
+        routes.append(
+            RouteInfo(
+                dst=f"{dst}/{r.get('dst_len')}" if dst else None,
+                gateway=gateway,
+                prefsrc=prefsrc,
+                oif=oif,
+                ifname=ifname,
+                table=r.get("table", 254),
+                proto=r.get("proto"),
+                scope=r.get("scope"),
+                type=r.get("type"),
+            )
+        )
+    return routes
+
+
+@router.get("/links", response_model=List[InterfaceInfo])
+def get_raw_links(ip: IPRoute = Depends(get_iproute)):
+    """
+    Returns all interfaces in a clean Pydantic format.
+    This is similar to /interfaces but avoids additional processing if needed.
+    """
+    result = []
+    links = ip.get_links()
+    addresses = ip.get_addr()
+
+    # group addresses by interface index
+    addr_map = {}
+    for a in addresses:
+        ifindex = a.get("index")
+        addr_map.setdefault(ifindex, []).append(a)
+
+    for link in links:
+        attrs = dict(link.get("attrs", []))  # convert list of tuples to dict
+        ifindex = link["index"]
+        addrs = addr_map.get(ifindex, [])
+
+        result.append(
+            InterfaceInfo(
+                ifindex=ifindex,
+                name=attrs.get("IFLA_IFNAME", "unknown"),
+                state=attrs.get("IFLA_OPERSTATE", "unknown"),
+                mac=attrs.get("IFLA_ADDRESS"),
+                mtu=attrs.get("IFLA_MTU", 0),
+                flags=[],  # latest pyroute2 removed ifi_flags, leave empty
+                addresses=parse_addresses(addrs),
+            )
+        )
+
+    return result
+
+@router.get("/bridges", response_model=List[BridgeInfo])
+def get_bridges():
+    """
+    Get all bridge interfaces on the system, including their member interfaces.
+    Returns detailed information:
+    - Bridge index, name, state, MTU
+    - STP state
+    - Member interfaces with index, name, state, and MTU
+    """
+    bridges_list: List[BridgeInfo] = []
+
+    with NDB() as ndb:
+        for br in ndb.interfaces:
+            # Only bridges
+            if getattr(br, "kind", None) == "bridge":
+                members: List[BridgeInterfaceInfo] = []
+                # Find member interfaces
+                for iface in ndb.interfaces:
+                    if getattr(iface, "master", None) == br.index:
+                        members.append(
+                            BridgeInterfaceInfo(
+                                ifindex=iface.index,
+                                ifname=iface.ifname,
+                                state=getattr(iface, "operstate", None),
+                                mtu=getattr(iface, "mtu", None)
+                            )
+                        )
+
+                bridges_list.append(
+                    BridgeInfo(
+                        ifindex=br.index,
+                        ifname=br.ifname,
+                        state=getattr(br, "operstate", None),
+                        mtu=getattr(br, "mtu", None),
+                        stp_state=getattr(br, "stp_state", None),
+                        members=members
+                    )
+                )
+
+    return bridges_list
+
+@router.get("/full-state")
+def full_state(
+    ip: IPRoute = Depends(get_iproute),
+):
+    """
+    Returns the full network state:
+    - Interfaces with IP addresses and flags
+    - Routes
+    - Bridges with member interfaces
+    """
+    return {
+        "interfaces": get_interfaces(ip),
+        "routes": get_routes(ip),
+        "bridges": get_bridges(),  # uses NDB internally
+    }