"""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// 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, agent tokens and backup-destination credentials — and the API deliberately never hands those out (``AgentRead.token_set`` is a bool, 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 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): raise BrowseError("Path is outside the allowed browse roots") 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, }