feat: 维护页 Tailscale 管理(设备列表/改名/授权/密钥不过期/设置IP/auth key,独立模块 core/tailscale_client.py)+ 密钥迁移 .env(STF_TOKEN/WEB_SECRET_KEY/TAILSCALE_API_KEY 不再入库)+ 修复设备更新端点(/device/{id} 405,改用 /name /authorized /key 专属端点)+ auth key description 仅 ASCII
This commit is contained in:
@@ -0,0 +1,213 @@
|
||||
"""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()
|
||||
Reference in New Issue
Block a user