metro-warden/ui/tabs/routes.py

"""
Routes Tab — displays the system routing table.

Reads routes from the OS on mount and on manual refresh.
Uses ``ip route show`` on Linux or ``netstat -rn`` as a fallback.
"""

from __future__ import annotations

import asyncio
import shutil
import subprocess
import re
from typing import List, Dict

from textual.app import ComposeResult
from textual.binding import Binding
from textual.widget import Widget
from textual.widgets import DataTable, Footer, Label, Static


def _parse_ip_route() -> List[Dict]:
    """Parse ``ip route show`` output into a list of route dicts."""
    routes: List[Dict] = []
    try:
        result = subprocess.run(
            ["ip", "route", "show"],
            capture_output=True,
            text=True,
            timeout=5,
        )
        for line in result.stdout.splitlines():
            line = line.strip()
            if not line:
                continue
            parts = line.split()
            dest = parts[0] if parts else "?"
            gateway = ""
            iface = ""
            metric = ""
            proto = ""
            scope = ""
            src = ""
            i = 1
            while i < len(parts):
                token = parts[i]
                if token == "via" and i + 1 < len(parts):
                    gateway = parts[i + 1]; i += 2
                elif token == "dev" and i + 1 < len(parts):
                    iface = parts[i + 1]; i += 2
                elif token == "metric" and i + 1 < len(parts):
                    metric = parts[i + 1]; i += 2
                elif token == "proto" and i + 1 < len(parts):
                    proto = parts[i + 1]; i += 2
                elif token == "scope" and i + 1 < len(parts):
                    scope = parts[i + 1]; i += 2
                elif token == "src" and i + 1 < len(parts):
                    src = parts[i + 1]; i += 2
                else:
                    i += 1
            routes.append({
                "destination": dest,
                "gateway": gateway or "—",
                "interface": iface or "—",
                "metric": metric or "—",
                "proto": proto or "—",
                "scope": scope or "—",
                "src": src or "—",
            })
    except (subprocess.TimeoutExpired, FileNotFoundError):
        pass
    return routes


def _parse_netstat_route() -> List[Dict]:
    """Fallback: parse ``netstat -rn`` output."""
    routes: List[Dict] = []
    try:
        result = subprocess.run(
            ["netstat", "-rn"],
            capture_output=True,
            text=True,
            timeout=5,
        )
        lines = result.stdout.splitlines()
        # Skip header lines
        data_start = False
        for line in lines:
            if re.match(r"^(Destination|0\.0\.0\.0|default)", line):
                data_start = True
            if not data_start:
                continue
            parts = line.split()
            if len(parts) >= 4:
                routes.append({
                    "destination": parts[0],
                    "gateway": parts[1] if len(parts) > 1 else "—",
                    "interface": parts[-1],
                    "metric": "—",
                    "proto": "—",
                    "scope": "—",
                    "src": "—",
                })
    except (subprocess.TimeoutExpired, FileNotFoundError):
        pass
    return routes


def _get_routes() -> List[Dict]:
    if shutil.which("ip"):
        routes = _parse_ip_route()
        if routes:
            return routes
    return _parse_netstat_route()


class RoutesTab(Widget):
    """Routing table tab — refreshes on mount and on 'r' keypress."""

    DEFAULT_CSS = """
    RoutesTab {
        layout: vertical;
        height: 1fr;
        padding: 1 2;
    }
    RoutesTab .tab-title {
        color: $primary;
        text-style: bold;
        height: 1;
        margin-bottom: 1;
    }
    RoutesTab DataTable {
        height: 1fr;
        border: round $primary;
    }
    RoutesTab .hint {
        height: 1;
        color: $text-muted;
        margin-top: 1;
    }
    """

    BINDINGS = [Binding("r", "refresh_routes", "Refresh")]

    COLUMNS = [
        ("Destination", 20),
        ("Gateway", 16),
        ("Interface", 12),
        ("Proto", 8),
        ("Scope", 8),
        ("Metric", 8),
        ("Src", 16),
    ]

    def compose(self) -> ComposeResult:
        yield Static("//  ROUTES  —  Routing Table", classes="tab-title")
        yield DataTable(id="routes-table", zebra_stripes=True, cursor_type="row")
        yield Static("Press [bold]r[/bold] to refresh", classes="hint")

    def on_mount(self) -> None:
        table = self.query_one("#routes-table", DataTable)
        for col_label, width in self.COLUMNS:
            table.add_column(col_label, width=width)
        self.run_worker(self._load_routes(), exclusive=True)

    async def _load_routes(self) -> None:
        routes = await asyncio.to_thread(_get_routes)
        self._populate_table(routes)

    def _populate_table(self, routes: List[Dict]) -> None:
        table = self.query_one("#routes-table", DataTable)
        table.clear()
        for r in routes:
            table.add_row(
                r["destination"],
                r["gateway"],
                r["interface"],
                r["proto"],
                r["scope"],
                r["metric"],
                r["src"],
            )

    def action_refresh_routes(self) -> None:
        self.run_worker(self._load_routes(), exclusive=True)