"""Network tools — Tailscale status.""" from __future__ import annotations from datetime import datetime, timezone from ..schemas import TailscaleResult, TailscalePeer from .ssh import run_local, SSHError TAILSCALE_BIN = "/Applications/Tailscale.app/Contents/MacOS/Tailscale" def _now() -> str: return datetime.now(timezone.utc).isoformat() def _parse_tailscale(output: str) -> list[TailscalePeer]: """Parse `tailscale status` output into peer list. Format: IP HOSTNAME USER@ OS STATUS_INFO Status examples: "-" (connected/active), "idle, tx ... rx ...", "offline, last seen ..." """ peers = [] for line in output.strip().splitlines(): parts = line.split() if len(parts) < 4: continue # Skip header-like lines if parts[0].startswith("#") or parts[0] == "IP": continue ip = parts[0] hostname = parts[1] # parts[2] = user@ (skip) os_name = parts[3] if len(parts) > 3 else "" # Remaining is status info status_text = " ".join(parts[4:]) if len(parts) > 4 else "" if "offline" in status_text: status = "offline" elif "idle" in status_text: status = "idle" elif status_text == "-" or status_text == "": status = "active" else: status = "active" peers.append(TailscalePeer( hostname=hostname, ip=ip, status=status, os=os_name, )) return peers async def tailscale_status() -> TailscaleResult: """Get Tailscale network status (runs locally).""" try: stdout, _ = await run_local(f"{TAILSCALE_BIN} status") except SSHError as e: return TailscaleResult( ok=False, checked_at=_now(), error_type=e.error_type, error=str(e), ) peers = _parse_tailscale(stdout) warnings = [] expected_hosts = {"sub-server", "hyungi-macmini", "hyungi-macbookpro"} found_hosts = {p.hostname for p in peers} missing = expected_hosts - found_hosts for h in missing: warnings.append(f"{h} not found in tailnet") return TailscaleResult( ok=True, checked_at=_now(), peers=peers, warnings=warnings, raw=stdout.strip(), )