Feat add protocol traffic engineering

backend/app.py

@@ -67,6 +67,7 @@
 from routers.rip import rip as rip_router
 from routers.ripng import ripng as ripng_router
 from routers.rpki import rpki as rpki_router
+from routers.traffic_engineering import traffic_engineering as te_router
 from routers import version as version_router
 from routers import events as events_router
 from routers.events import start_poller, stop_poller
@@ -354,6 +355,7 @@ async def get_permissions(request: Request) -> dict:
 app.include_router(rip_router.router)
 app.include_router(ripng_router.router)
 app.include_router(rpki_router.router)
+app.include_router(te_router.router)
 app.include_router(version_router.router)
 app.include_router(events_router.router)
 

backend/rbac_permissions.py

@@ -79,6 +79,7 @@ class FeatureGroup(str, Enum):
     SEGMENT_ROUTING = "SEGMENT_ROUTING"
     NHRP = "NHRP"
     RPKI = "RPKI"
+    TRAFFIC_ENGINEERING = "TRAFFIC_ENGINEERING"
 
     # Routing Policies
     ROUTING_POLICIES = "ROUTING_POLICIES"
@@ -173,6 +174,7 @@ class BuiltInRole(str, Enum):
         FeatureGroup.SEGMENT_ROUTING: PermissionLevel.WRITE,
         FeatureGroup.NHRP: PermissionLevel.WRITE,
         FeatureGroup.RPKI: PermissionLevel.WRITE,
+        FeatureGroup.TRAFFIC_ENGINEERING: PermissionLevel.WRITE,
         FeatureGroup.ROUTING_POLICIES: PermissionLevel.WRITE,
         FeatureGroup.ACCESS_LIST: PermissionLevel.WRITE,
         FeatureGroup.PREFIX_LIST: PermissionLevel.WRITE,
@@ -241,6 +243,7 @@ class BuiltInRole(str, Enum):
         FeatureGroup.SEGMENT_ROUTING: PermissionLevel.WRITE,
         FeatureGroup.NHRP: PermissionLevel.WRITE,
         FeatureGroup.RPKI: PermissionLevel.WRITE,
+        FeatureGroup.TRAFFIC_ENGINEERING: PermissionLevel.WRITE,
         FeatureGroup.ROUTING_POLICIES: PermissionLevel.WRITE,
         FeatureGroup.ACCESS_LIST: PermissionLevel.WRITE,
         FeatureGroup.PREFIX_LIST: PermissionLevel.WRITE,
@@ -310,6 +313,7 @@ class BuiltInRole(str, Enum):
         FeatureGroup.SEGMENT_ROUTING: PermissionLevel.READ,
         FeatureGroup.NHRP: PermissionLevel.READ,
         FeatureGroup.RPKI: PermissionLevel.READ,
+        FeatureGroup.TRAFFIC_ENGINEERING: PermissionLevel.READ,
         FeatureGroup.ROUTING_POLICIES: PermissionLevel.READ,
         FeatureGroup.ACCESS_LIST: PermissionLevel.READ,
         FeatureGroup.PREFIX_LIST: PermissionLevel.READ,
@@ -419,6 +423,7 @@ async def get_user_permissions(
                 FeatureGroup.SEGMENT_ROUTING,
                 FeatureGroup.NHRP,
                 FeatureGroup.RPKI,
+                FeatureGroup.TRAFFIC_ENGINEERING,
                 FeatureGroup.ROUTING_POLICIES,
                 FeatureGroup.ACCESS_LIST,
                 FeatureGroup.PREFIX_LIST,
@@ -506,6 +511,7 @@ async def get_user_permissions(
                 FeatureGroup.SEGMENT_ROUTING,
                 FeatureGroup.NHRP,
                 FeatureGroup.RPKI,
+                FeatureGroup.TRAFFIC_ENGINEERING,
                 FeatureGroup.ROUTING_POLICIES,
                 FeatureGroup.ACCESS_LIST,
                 FeatureGroup.PREFIX_LIST,
@@ -703,7 +709,7 @@ def _apply_parent_child_permissions(permissions: Dict[FeatureGroup, PermissionLe
     - VPN → IPSEC, WIREGUARD
     - ROUTING → UNICAST_PROTOCOLS
     - UNICAST_PROTOCOLS → BGP, OSPF, OSPFv3, ISIS, OPENFABRIC, RIP, RIPng, BABEL
-    - ROUTING_INFRASTRUCTURE → BFD, MPLS, SEGMENT_ROUTING, NHRP, RPKI
+    - ROUTING_INFRASTRUCTURE → BFD, MPLS, SEGMENT_ROUTING, NHRP, RPKI, TRAFFIC_ENGINEERING
     - ROUTING_POLICIES → ACCESS_LIST, PREFIX_LIST, ROUTE_POLICY, ROUTE_MAP, LOCAL_ROUTE,
                          BGP_AS_PATH, BGP_COMMUNITY, BGP_EXTENDED_COMMUNITY, BGP_LARGE_COMMUNITY
     - MULTICAST → IGMP_PROXY, PIM, PIM6
@@ -779,6 +785,7 @@ def _apply_parent_child_permissions(permissions: Dict[FeatureGroup, PermissionLe
             FeatureGroup.SEGMENT_ROUTING,
             FeatureGroup.NHRP,
             FeatureGroup.RPKI,
+            FeatureGroup.TRAFFIC_ENGINEERING,
         ]
         for child in infrastructure_children:
             current = permissions.get(child, PermissionLevel.NONE)

backend/routers/traffic_engineering/__init__.py

@@ -0,0 +1,3 @@
+from .traffic_engineering import router
+
+__all__ = ["router"]

backend/routers/traffic_engineering/traffic_engineering.py

@@ -0,0 +1,228 @@
+"""Traffic Engineering Protocol Router.
+
+API endpoints for managing VyOS Traffic Engineering configuration.
+Traffic Engineering is only supported on VyOS 1.5+; on 1.4 the capabilities
+endpoint advertises the feature as unsupported and the config/batch endpoints
+return early with empty data.
+"""
+
+from fastapi import APIRouter, HTTPException, Request
+from starlette.concurrency import run_in_threadpool
+from pydantic import BaseModel, Field
+from typing import List, Optional, Dict, Any
+from session_vyos_service import get_session_vyos_service
+from vyos_builders import TrafficEngineeringBatchBuilder
+from fastapi_permissions import require_read_permission, require_write_permission
+from rbac_permissions import FeatureGroup
+import inspect
+import logging
+
+logger = logging.getLogger(__name__)
+
+router = APIRouter(prefix="/vyos/traffic-engineering", tags=["traffic-engineering"])
+
+
+# ============================================================================
+# Pydantic Models
+# ============================================================================
+
+
+class AdminGroup(BaseModel):
+    name: str
+    bit_position: Optional[int] = None
+
+
+class TeInterface(BaseModel):
+    name: str
+    admin_groups: List[str] = []
+    max_bandwidth: Optional[int] = None
+    max_reservable_bandwidth: Optional[int] = None
+    metric: Optional[int] = None
+
+
+class TrafficEngineeringConfig(BaseModel):
+    admin_groups: List[AdminGroup] = []
+    interfaces: List[TeInterface] = []
+
+
+class TrafficEngineeringBatchOperation(BaseModel):
+    op: str = Field(..., description="Builder method name")
+    value: Optional[str] = Field(None, description="Comma-separated arguments")
+
+
+class TrafficEngineeringBatchRequest(BaseModel):
+    operations: List[TrafficEngineeringBatchOperation]
+
+
+class VyOSResponse(BaseModel):
+    success: bool
+    data: Optional[Dict[str, Any]] = None
+    error: Optional[str] = None
+
+
+# ============================================================================
+# Endpoint 1: Capabilities
+# ============================================================================
+
+
+@router.get("/capabilities")
+async def get_te_capabilities(request: Request):
+    """Return Traffic Engineering feature capabilities based on the device VyOS version."""
+    await require_read_permission(request, FeatureGroup.TRAFFIC_ENGINEERING)
+
+    try:
+        service = get_session_vyos_service(request)
+        builder = TrafficEngineeringBatchBuilder(version=service.get_version())
+        capabilities = builder.get_capabilities()
+
+        if hasattr(request.state, "instance") and request.state.instance:
+            capabilities["instance_name"] = request.state.instance.get("name")
+            capabilities["instance_id"] = request.state.instance.get("id")
+
+        return capabilities
+    except Exception:
+        logger.exception("Unhandled error in get_te_capabilities")
+        raise HTTPException(status_code=500, detail="Internal server error")
+
+
+# ============================================================================
+# Endpoint 2: Config
+# ============================================================================
+
+
+@router.get("/config", response_model=TrafficEngineeringConfig)
+async def get_te_config(http_request: Request, refresh: bool = False):
+    """Return the full Traffic Engineering configuration in a normalized format."""
+    await require_read_permission(http_request, FeatureGroup.TRAFFIC_ENGINEERING)
+
+    try:
+        service = get_session_vyos_service(http_request)
+        version = service.get_version()
+
+        if "1.4" in version:
+            return TrafficEngineeringConfig()
+
+        full_config = await run_in_threadpool(service.get_full_config, refresh=refresh)
+        te_raw = full_config.get("protocols", {}).get("traffic-engineering", {})
+
+        if not te_raw:
+            return TrafficEngineeringConfig()
+
+        return TrafficEngineeringConfig(
+            admin_groups=_parse_admin_groups(te_raw.get("admin-group", {})),
+            interfaces=_parse_interfaces(te_raw.get("interface", {})),
+        )
+    except Exception:
+        logger.exception("Unhandled error in get_te_config")
+        raise HTTPException(status_code=500, detail="Internal server error")
+
+
+# ============================================================================
+# Config Parsers
+# ============================================================================
+
+
+def _safe_int(value) -> Optional[int]:
+    if value is None:
+        return None
+    try:
+        return int(value)
+    except (ValueError, TypeError):
+        return None
+
+
+def _parse_admin_groups(raw: dict) -> List[AdminGroup]:
+    if not raw:
+        return []
+
+    groups = []
+    for name, cfg in raw.items():
+        if cfg is None:
+            cfg = {}
+        groups.append(AdminGroup(
+            name=name,
+            bit_position=_safe_int(cfg.get("bit-position")),
+        ))
+    return groups
+
+
+def _parse_interfaces(raw: dict) -> List[TeInterface]:
+    if not raw:
+        return []
+
+    interfaces = []
+    for iface_name, cfg in raw.items():
+        if cfg is None:
+            cfg = {}
+
+        # admin-group is multi-value: may be a list or a single string
+        raw_groups = cfg.get("admin-group", [])
+        if isinstance(raw_groups, str):
+            admin_groups = [raw_groups]
+        elif isinstance(raw_groups, list):
+            admin_groups = raw_groups
+        else:
+            admin_groups = list(raw_groups) if raw_groups else []
+
+        interfaces.append(TeInterface(
+            name=iface_name,
+            admin_groups=admin_groups,
+            max_bandwidth=_safe_int(cfg.get("max-bandwidth")),
+            max_reservable_bandwidth=_safe_int(cfg.get("max-reservable-bandwidth")),
+            metric=_safe_int(cfg.get("metric")),
+        ))
+    return interfaces
+
+
+# ============================================================================
+# Endpoint 3: Batch Operations
+# ============================================================================
+
+
+@router.post("/batch", response_model=VyOSResponse)
+async def te_batch_configure(http_request: Request, body: TrafficEngineeringBatchRequest):
+    """Execute a batch of Traffic Engineering configuration operations atomically."""
+    await require_write_permission(http_request, FeatureGroup.TRAFFIC_ENGINEERING)
+
+    try:
+        service = get_session_vyos_service(http_request)
+
+        if "1.4" in service.get_version():
+            raise HTTPException(
+                status_code=400,
+                detail="Traffic Engineering is not supported on VyOS 1.4",
+            )
+
+        builder = TrafficEngineeringBatchBuilder(version=service.get_version())
+
+        for operation in body.operations:
+            method = getattr(builder, operation.op)
+            sig = inspect.signature(method)
+            params = [p for p in sig.parameters.keys() if p != "self"]
+
+            if len(params) == 0:
+                method()
+            elif len(params) == 1:
+                if operation.value is not None:
+                    method(operation.value)
+            elif len(params) == 2 and operation.value is not None:
+                values = operation.value.split(",", 1)
+                if len(values) == 2:
+                    method(values[0], values[1])
+                else:
+                    method(operation.value, "")
+
+        response = service.execute_batch(builder)
+
+        return VyOSResponse(
+            success=response.status == 200,
+            data={"message": "Traffic Engineering configuration updated"},
+            error=response.error if response.error else None,
+        )
+    except HTTPException:
+        raise
+    except AttributeError as e:
+        raise HTTPException(status_code=400, detail=f"Unknown operation: {str(e)}")
+    except Exception:
+        logger.exception("Unhandled error in te_batch_configure")
+        raise HTTPException(status_code=500, detail="Internal server error")

backend/vyos_builders/__init__.py

@@ -40,6 +40,7 @@
 from .rip import RipBatchBuilder
 from .ripng import RipNgBatchBuilder
 from .rpki import RpkiBatchBuilder
+from .traffic_engineering import TrafficEngineeringBatchBuilder
 
 # Directly use the self-contained builders
 EthernetBatchBuilder = EthernetInterfaceBuilderMixin
@@ -101,6 +102,7 @@
     "RipBatchBuilder",
     "RipNgBatchBuilder",
     "RpkiBatchBuilder",
+    "TrafficEngineeringBatchBuilder",
     "BondingBatchBuilder",
     "GeneveBatchBuilder",
     "InputBatchBuilder",

backend/vyos_builders/traffic_engineering/__init__.py

@@ -0,0 +1,4 @@
+"""Traffic Engineering Protocol Builder Package."""
+from .traffic_engineering_batch_builder import TrafficEngineeringBatchBuilder
+
+__all__ = ["TrafficEngineeringBatchBuilder"]