"""Tailscale 管理 API(仅管理员)。""" import re from flask import Blueprint, jsonify, request from core import tailscale_client from core.tailscale_client import TailscaleError from core.logger import get_logger _log = get_logger("web") from web.auth import admin_required bp = Blueprint("tailscale", __name__) @bp.route("/api/tailscale/status") @admin_required def api_tailscale_status(): """Tailscale 管理配置状态(API key / tailnet 是否已配置)。""" c = tailscale_client.tailscale return jsonify({"ok": True, "configured": c.is_configured(), "hint": c.config_hint() if not c.is_configured() else ""}) @bp.route("/api/tailscale/devices") @admin_required def api_tailscale_devices(): """列出 tailnet 全部设备。""" try: return jsonify({"ok": True, "devices": tailscale_client.tailscale.list_devices()}) except TailscaleError as e: return jsonify({"ok": False, "error": str(e)}), 502 @bp.route("/api/tailscale/devices/", methods=["POST"]) @admin_required def api_tailscale_device_update(device_id): """更新设备:按传入字段分发到专属端点(改名 /name、授权 /authorized、密钥 /key)。""" data = request.json or {} c = tailscale_client.tailscale try: if "name" in data: c.set_device_name(device_id, data.get("name")) if "authorized" in data: c.set_device_authorized(device_id, data.get("authorized")) if "key_expiry_disabled" in data: c.set_device_key_expiry(device_id, data.get("key_expiry_disabled")) _log.info(f"Tailscale 设备更新 {device_id}: {data}") return jsonify({"ok": True, "msg": "设备已更新"}) except TailscaleError as e: return jsonify({"ok": False, "error": str(e)}), 502 @bp.route("/api/tailscale/devices//ip", methods=["POST"]) @admin_required def api_tailscale_device_ip(device_id): """设置设备 IPv4 地址(未公开端点,实测可用)。 ⚠ 改 IP 会断开设备当前 tailscale 会话,且平台设备池 serial 随之变化, 改完需同步更新 STF 设备池(分组/任务目标里的旧 IP 会失效)。 """ data = request.json or {} ipv4 = (data.get("ipv4") or "").strip() if not ipv4: return jsonify({"ok": False, "error": "缺少 ipv4"}), 400 try: tailscale_client.tailscale.set_device_ip(device_id, ipv4) _log.info(f"Tailscale 设备 {device_id} IP 已设置为 {ipv4}") return jsonify({"ok": True, "msg": f"设备 IP 已设置为 {ipv4}"}) except TailscaleError as e: return jsonify({"ok": False, "error": str(e)}), 502 @bp.route("/api/tailscale/devices/", methods=["DELETE"]) @admin_required def api_tailscale_device_delete(device_id): """从 tailnet 移除设备。""" try: tailscale_client.tailscale.delete_device(device_id) _log.info(f"Tailscale 设备已移除: {device_id}") return jsonify({"ok": True, "msg": "设备已从 tailnet 移除"}) except TailscaleError as e: return jsonify({"ok": False, "error": str(e)}), 502 @bp.route("/api/tailscale/authkey", methods=["POST"]) @admin_required def api_tailscale_authkey(): """生成设备接入 auth key(key 只显示一次)。""" data = request.json or {} try: # Tailscale 要求 description 仅 ASCII(中文会 400) desc = re.sub(r"[^\x20-\x7e]", "", (data.get("description") or "").strip()) key = tailscale_client.tailscale.create_auth_key( description=desc or "auto_control", reusable=bool(data.get("reusable", False)), ephemeral=bool(data.get("ephemeral", False)), preauthorized=data.get("preauthorized", True), expiry_seconds=int(data.get("expiry_seconds", 3600))) _log.info(f"Tailscale auth key 已生成: {key['id']}") return jsonify({"ok": True, "key": key["key"], "id": key["id"], "expires": key["expires"]}) except TailscaleError as e: return jsonify({"ok": False, "error": str(e)}), 502