214 lines
8.8 KiB
Python
214 lines
8.8 KiB
Python
"""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()
|