stackpilot/backend/services/device_service.py

"""Host device detection (USB, serial/TTY, DRI) and host filesystem browser."""
from __future__ import annotations

import glob
import os
from dataclasses import asdict, dataclass

from config import settings


@dataclass
class HostDevice:
    path: str
    kind: str  # "usb" | "tty" | "dri" | "other"
    name: str = ""

    def to_dict(self) -> dict:
        return asdict(self)


def _read(path: str) -> str:
    try:
        with open(path, "r", encoding="utf-8", errors="replace") as fh:
            return fh.read().strip()
    except OSError:
        return ""


def _usb_devices() -> list[HostDevice]:
    devices: list[HostDevice] = []
    # /dev/bus/usb/<bus>/<dev>
    for path in sorted(glob.glob("/dev/bus/usb/*/*")):
        name = ""
        # Best-effort name lookup via sysfs is non-trivial to map; leave generic.
        devices.append(HostDevice(path=path, kind="usb", name=name or "USB device"))
    return devices


def _usb_names_from_sysfs() -> list[HostDevice]:
    """Richer USB list from sysfs with product/manufacturer strings."""
    out: list[HostDevice] = []
    for dev in sorted(glob.glob("/sys/bus/usb/devices/*")):
        busnum = _read(os.path.join(dev, "busnum"))
        devnum = _read(os.path.join(dev, "devnum"))
        if not busnum or not devnum:
            continue
        product = _read(os.path.join(dev, "product"))
        manufacturer = _read(os.path.join(dev, "manufacturer"))
        label = " ".join(p for p in (manufacturer, product) if p) or "USB device"
        path = f"/dev/bus/usb/{int(busnum):03d}/{int(devnum):03d}"
        out.append(HostDevice(path=path, kind="usb", name=label))
    return out


def _tty_devices() -> list[HostDevice]:
    devices: list[HostDevice] = []
    patterns = ["/dev/ttyUSB*", "/dev/ttyACM*", "/dev/ttyAMA*", "/dev/serial/by-id/*"]
    for pattern in patterns:
        for path in sorted(glob.glob(pattern)):
            base = os.path.basename(path)
            driver = _read(f"/sys/class/tty/{base}/device/driver/module/name") or ""
            devices.append(
                HostDevice(path=path, kind="tty", name=driver or "Serial device")
            )
    return devices


def _dri_devices() -> list[HostDevice]:
    return [
        HostDevice(path=p, kind="dri", name="GPU render node")
        for p in sorted(glob.glob("/dev/dri/*"))
    ]


def detect_devices() -> dict:
    usb = _usb_names_from_sysfs() or _usb_devices()
    return {
        "usb": [d.to_dict() for d in usb],
        "tty": [d.to_dict() for d in _tty_devices()],
        "dri": [d.to_dict() for d in _dri_devices()],
    }


# --------------------------------------------------------------------------- #
# Host filesystem browser (sandboxed)
# --------------------------------------------------------------------------- #


class BrowseError(Exception):
    pass


def _real_root(path: str) -> str:
    """Map a logical host path into the container view (HOST_ROOT_PREFIX).

    Refuses anything that resolves inside StackPilot's own ``DATA_DIR``. That
    directory holds ``stackpilot.db`` — users, password hashes and
    backup-destination credentials — and the API deliberately never hands those
    out (destination secrets come back masked). Without this the file browser
    would be a way around that, for admins too. Note this only bites when ``HOST_ROOT_PREFIX`` is empty: with a
    prefix set, no logical path can reach the container's own ``/data`` at all.
    """
    prefix = settings.HOST_ROOT_PREFIX.rstrip("/")
    real = prefix + path if prefix else path
    data_dir = os.path.normpath(settings.DATA_DIR)
    norm = os.path.normpath(real)
    if norm == data_dir or norm.startswith(data_dir + os.sep):
        raise BrowseError("Path is inside StackPilot's own data directory")
    return real


def _is_allowed(path: str) -> bool:
    norm = os.path.normpath(path)
    for root in settings.ALLOWED_BROWSE_ROOTS:
        root = os.path.normpath(root)
        if root == "/" or norm == root or norm.startswith(root + os.sep):
            return True
    return False


def _toward_roots(path: str) -> list[dict]:
    """For a path *above* the allowed roots: the folders below it that lead to
    one (``/`` -> ``mnt``, ``opt``; ``/srv`` -> ``data`` for ``/srv/data``).

    The browser opens at ``/``, which the default roots do not include, and the
    breadcrumbs and "up" lead there too — so the page used to open on an error
    with no way into any root at all.
    """
    prefix = "/" if path == "/" else path + os.sep
    names: set[str] = set()
    for root in settings.ALLOWED_BROWSE_ROOTS:
        root = os.path.normpath(root)
        if not root.startswith(prefix):
            continue
        try:
            if not os.path.isdir(_real_root(root)):
                continue
        except BrowseError:
            continue
        names.add(root[len(prefix):].split(os.sep, 1)[0])
    entries = []
    for name in sorted(names, key=str.lower):
        try:
            st = os.stat(_real_root(os.path.join(path, name)))
        except (OSError, BrowseError):
            continue
        entries.append(
            {
                "name": name,
                "type": "dir",
                "size": None,
                "permissions": oct(st.st_mode & 0o777),
                "mtime": st.st_mtime,
                "symlink": False,
            }
        )
    return entries


def browse(path: str = "/", show_hidden: bool = False) -> dict:
    path = os.path.normpath(path or "/")
    if not path.startswith("/"):
        raise BrowseError("Path must be absolute")
    if not _is_allowed(path):
        entries = _toward_roots(path)
        if not entries:
            raise BrowseError("Path is outside the allowed browse roots")
        return {
            "path": path,
            "parent": os.path.dirname(path) if path != "/" else None,
            "roots": settings.ALLOWED_BROWSE_ROOTS,
            "entries": entries,
            # Only a way through: nothing can be created or changed here.
            "virtual": True,
        }

    real = _real_root(path)
    if not os.path.isdir(real):
        raise BrowseError(f"Not a directory: {path}")

    entries = []
    try:
        names = os.listdir(real)
    except PermissionError as exc:
        raise BrowseError(f"Permission denied: {path}") from exc

    for name in sorted(names):
        if not show_hidden and name.startswith("."):
            continue
        full_real = os.path.join(real, name)
        try:
            st = os.lstat(full_real)
            is_dir = os.path.isdir(full_real)
            entries.append(
                {
                    "name": name,
                    "type": "dir" if is_dir else "file",
                    "size": st.st_size if not is_dir else None,
                    "permissions": oct(st.st_mode & 0o777),
                    "mtime": st.st_mtime,
                    "symlink": os.path.islink(full_real),
                }
            )
        except OSError:
            continue

    # Sort: dirs first, then files, both alphabetical.
    entries.sort(key=lambda e: (e["type"] != "dir", e["name"].lower()))
    parent = os.path.dirname(path) if path != "/" else None
    return {
        "path": path,
        "parent": parent,
        "roots": settings.ALLOWED_BROWSE_ROOTS,
        "entries": entries,
    }