"""Tailscale API v2 客户端封装（维护页"Tailscale 管理"用）。 能力： - 列出 tailnet 全部设备（名称/主机名/IP/系统/在线/授权/密钥过期状态） - 更新设备：显示名、主机名、授权开关、密钥不过期（keyExpiryDisabled） - 生成设备接入 auth key（可配置 reusable/ephemeral/preauthorized/有效期） - 删除设备 配置（config.py，支持 .env 注入，**不要提交密钥到 git**）： TAILSCALE_API_KEY — 管理后台 → Settings → API Access Tokens 生成的 key（或 OAuth client 的 client_id:secret） TAILSCALE_TAILNET — tailnet 名称或 ID（个人账号一般是登录邮箱前缀，如 1422726308） 注意：设备 IP 由 tailnet 自动分配，API 无法修改；列表里的 addresses 只读展示。 参考：https://tailscale.com/api """ import requests from config import TAILSCALE_API_KEY, TAILSCALE_TAILNET from core.logger import get_logger _log = get_logger("core.tailscale") _API_BASE = "https://api.tailscale.com/api/v2" _TIMEOUT = (3, 15) class TailscaleError(Exception): """Tailscale API 错误（含 HTTP 状态码）。""" def __init__(self, message, code=""): super().__init__(message) self.code = code class TailscaleClient: """Tailscale API v2 客户端。 认证：Basic Auth，API key 作为用户名、空密码； 也可传 OAuth client 的 "client_id:client_secret" 作为 key。 """ def __init__(self, api_key=None, tailnet=None): self.api_key = api_key or TAILSCALE_API_KEY self.tailnet = tailnet or TAILSCALE_TAILNET # ================== 配置状态 ================== def is_configured(self): """是否已配置 API key 与 tailnet。""" return bool(self.api_key and self.tailnet) def config_hint(self): """未配置时的提示文案。""" missing = [] if not self.api_key: missing.append("TAILSCALE_API_KEY（Tailscale 后台 → Settings → API Access Tokens）") if not self.tailnet: missing.append("TAILSCALE_TAILNET（tailnet 名，个人账号一般是邮箱前缀）") return "；".join(missing) # ================== 请求基座 ================== def _headers(self): return {"Authorization": f"Basic {self._basic()}"} def _basic(self): import base64 return base64.b64encode(f"{self.api_key}:".encode()).decode() def _get(self, path): if not self.is_configured(): raise TailscaleError(f"未配置：{self.config_hint()}", "config") try: r = requests.get(f"{_API_BASE}{path}", headers=self._headers(), timeout=_TIMEOUT) except requests.exceptions.ConnectionError as e: raise TailscaleError(f"Tailscale API 不可达: {e}", "network") from e except requests.exceptions.Timeout as e: raise TailscaleError("Tailscale API 请求超时", "network") from e return self._handle(r) def _post(self, path, body): if not self.is_configured(): raise TailscaleError(f"未配置：{self.config_hint()}", "config") try: r = requests.post(f"{_API_BASE}{path}", json=body, headers=self._headers(), timeout=_TIMEOUT) except requests.exceptions.ConnectionError as e: raise TailscaleError(f"Tailscale API 不可达: {e}", "network") from e except requests.exceptions.Timeout as e: raise TailscaleError("Tailscale API 请求超时", "network") from e return self._handle(r) def _handle(self, r): try: data = r.json() except Exception: data = {} if r.status_code == 401: raise TailscaleError("API key 无效或已过期（401）", "auth") if r.status_code == 404: raise TailscaleError("tailnet 不存在或无权访问（404），请检查 TAILSCALE_TAILNET", "notfound") if r.status_code >= 400: msg = data.get("message") or str(data)[:200] _log.error(f"Tailscale API 错误 {r.status_code}: {msg}") raise TailscaleError(f"Tailscale API 错误 {r.status_code}: {msg}", "api") return data # ================== 设备 ================== def list_devices(self): """列出 tailnet 全部设备。 返回 [{id, name, hostname, os, addresses[], authorized, key_expiry_disabled, online, last_seen, ...}]（字段已规整）。 """ data = self._get(f"/tailnet/{self.tailnet}/devices") devices = [] for d in data.get("devices", []): online = d.get("online") devices.append({ "id": d.get("id", ""), "name": d.get("name", ""), "hostname": d.get("hostname", ""), "os": d.get("os", ""), "addresses": d.get("addresses", []), "authorized": bool(d.get("authorized")), "key_expiry_disabled": bool(d.get("keyExpiryDisabled")), # online 三态：True=在线 / False=离线 / None=最近 12 小时内见过但当前未上报（API 返回 null）。 # 注意不能 bool(null)——那会把"最近在线"误显示成"离线"。 "online": online if online is not None else None, "last_seen": d.get("lastSeen", ""), "tags": d.get("tags", []), }) return devices def set_device_name(self, device_id, name): """修改设备显示名（POST /device/{id}/name）。 注意：POST /device/{id} 是 405，改名/授权/密钥各有专属端点。 """ if not name or not str(name).strip(): raise TailscaleError("名称不能为空") self._post(f"/device/{device_id}/name", {"name": str(name).strip()}) return True def set_device_authorized(self, device_id, authorized): """授权/取消授权（POST /device/{id}/authorized）。""" self._post(f"/device/{device_id}/authorized", {"authorized": bool(authorized)}) return True def set_device_key_expiry(self, device_id, disabled): """关闭/恢复设备密钥过期（POST /device/{id}/key）。 disabled=True 即"密钥不过期"（设备不会被定期踢下线）； 恢复过期后按原定过期时间执行，若已过期需重新授权。 """ self._post(f"/device/{device_id}/key", {"keyExpiryDisabled": bool(disabled)}) return True def set_device_ip(self, device_id, ipv4): """为设备设置新的 Tailscale IPv4 地址。 端点 POST /device/{device_id}/ip 实测存在（官方文档未收录，属未公开接口）。 注意： - 修改 IP 会断开该设备当前的 tailscale 会话，几秒后以新 IP 重连 - 平台设备池（STF serial）用的就是 tailnet IP，改完需同步更新 STF 设备池 - IPv6 无公开 API 可改 """ if not ipv4 or not str(ipv4).strip(): raise TailscaleError("IPv4 地址不能为空") self._post(f"/device/{device_id}/ip", {"ipv4": str(ipv4).strip()}) return True def delete_device(self, device_id): """从 tailnet 移除设备（下次设备上线需重新授权）。""" if not self.is_configured(): raise TailscaleError(f"未配置：{self.config_hint()}", "config") try: r = requests.delete(f"{_API_BASE}/device/{device_id}", headers=self._headers(), timeout=_TIMEOUT) except requests.exceptions.ConnectionError as e: raise TailscaleError(f"Tailscale API 不可达: {e}", "network") from e return self._handle(r) # ================== Auth key ================== def create_auth_key(self, description="auto_control", reusable=False, ephemeral=False, preauthorized=True, expiry_seconds=3600): """生成设备接入 auth key。 返回 {id, key, expires}。key 只显示这一次，请立即复制保存。 preauthorized=True：新设备接入自动授权（免去后台手动点授权）。 """ body = { "description": description, "expirySeconds": max(60, int(expiry_seconds)), "capabilities": { "devices": { "create": { "reusable": bool(reusable), "ephemeral": bool(ephemeral), "preauthorized": bool(preauthorized), } } }, } data = self._post(f"/tailnet/{self.tailnet}/keys", body) return { "id": data.get("id", ""), "key": data.get("key", ""), "expires": data.get("expires", ""), } # 模块级单例（与 stf_client 的用法一致） tailscale = TailscaleClient()