搬瓦工 KiwiVM API 入门指南：如何用只读接口安全监控流量与 VPS 运行状态

curl -s -X POST \
  -d "veid=YOUR_VEID&api_key=YOUR_API_KEY" \
  https://api.64clouds.com/v1/getServiceInfo

{
  "vm_type": "kvm",
  "hostname": "my-vps",
  "node_ip": "...",
  "plan_monthly_data": 1073741824000,
  "data_counter": 85899345920,
  "data_next_reset": 1730419200,
  "ip_addresses": ["..."],
  "error": 0
}

#!/usr/bin/env python3
"""
BandwagonHost KiwiVM API 生产级只读巡检工具
遵循最小权限与防御性编码规范：仅调用 getLiveServiceInfo 与 getServiceInfo 只读端点。
"""

import datetime
import getpass
import json
import os
import sys
import urllib.error
import urllib.parse
import urllib.request
from typing import Any, Dict, Tuple


EXPECTED_HOST = "api.64clouds.com"
READONLY_ENDPOINT = "/v1/getLiveServiceInfo"


def get_credentials() -> Tuple[str, str]:
    """
    安全提取 API 凭据：
    优先从环境变量获取，无环境变量时提示交互式掩码输入。
    """
    veid = os.environ.get("KIWIVM_VEID", "").strip()
    api_key = os.environ.get("KIWIVM_API_KEY", "").strip()

    if not veid:
        veid = input("请输入 KiwiVM VEID (纯数字): ").strip()
    if not api_key:
        api_key = getpass.getpass("请输入 KiwiVM API Key (输入隐藏): ").strip()

    if not veid.isdigit():
        sys.exit("安全校验失败: VEID 必须为纯数字格式。")
    if not api_key:
        sys.exit("安全校验失败: API Key 不能为空。")

    return veid, api_key


def format_bytes(num_bytes: int) -> str:
    """将字节大小格式化为易读的二进制单位 (GiB / MiB)"""
    gib = num_bytes / (1024**3)
    if gib >= 1.0:
        return f"{gib:.2f} GiB"
    mib = num_bytes / (1024**2)
    return f"{mib:.2f} MiB"


def format_timestamp(ts: int) -> str:
    """将 Unix 时间戳转换为本地易读时间字符串"""
    if not ts or ts <= 0:
        return "N/A"
    try:
        dt = datetime.datetime.fromtimestamp(ts, tz=datetime.timezone.utc)
        return dt.strftime("%Y-%m-%d %H:%M:%S UTC")
    except (OverflowError, OSError, ValueError):
        return str(ts)


def query_kiwivm_metrics(veid: str, api_key: str) -> Dict[str, Any]:
    """使用严格的 HTTPS POST 表单调用只读端点"""
    url = f"https://{EXPECTED_HOST}{READONLY_ENDPOINT}"

    # 校验 URL 结构，防范异常篡改
    parsed = urllib.parse.urlsplit(url)
    if parsed.scheme != "https" or parsed.netloc != EXPECTED_HOST or parsed.path != READONLY_ENDPOINT:
        sys.exit("安全阻断: 检测到非法的请求目标或路径变更。")

    # 参数封装在 POST 请求体内
    payload = urllib.parse.urlencode({"veid": veid, "api_key": api_key}).encode("utf-8")
    headers = {
        "User-Agent": "BWH-Ops-Inspector/1.0",
        "Content-Type": "application/x-www-form-urlencoded",
    }

    req = urllib.request.Request(url, data=payload, headers=headers, method="POST")

    try:
        with urllib.request.urlopen(req, timeout=15) as resp:
            if resp.status != 200:
                sys.exit(f"接口返回非预期的 HTTP 状态码: {resp.status}")
            body_text = resp.read().decode("utf-8")
            data = json.loads(body_text)
    except urllib.error.HTTPError as err:
        sys.exit(f"网络请求失败 (HTTP {err.code}): {err.reason}")
    except urllib.error.URLError as err:
        sys.exit(f"网络连接异常: {err.reason}")
    except json.JSONDecodeError:
        sys.exit("接口响应内容解析失败，非合法 JSON 格式。")

    # 校验 KiwiVM 服务端业务错误码
    if data.get("error") != 0:
        err_msg = data.get("message", "未知错误")
        sys.exit(f"KiwiVM 接口调用失败 (代码 {data.get('error')}): {err_msg}")

    return data


def print_inspection_report(data: Dict[str, Any]) -> None:
    """排版并输出只读状态报告"""
    print("\n" + "=" * 52)
    print("       搬瓦工 VPS 宿主机只读健康巡检报告")
    print("=" * 52)

    vm_type = data.get("vm_type", "kvm").upper()
    node_name = data.get("node_name", "N/A")
    load_avg = data.get("load_average", "N/A")

    print(f"架构平台与节点 : {vm_type} (宿主机: {node_name})")
    print(f"系统平均负载   : {load_avg}")

    # 内存核算 (以字节为基准)
    mem_total = data.get("plan_ram", 0)
    mem_available = data.get("mem_available_bytes", 0)
    if mem_total > 0:
        mem_used = max(0, mem_total - mem_available)
        mem_pct = (mem_used / mem_total) * 100
        print(f"内存占用状况   : {format_bytes(mem_used)} / {format_bytes(mem_total)} ({mem_pct:.1f}%)")

    # 流量额度与重置周期
    data_counter = data.get("data_counter", 0)
    monthly_quota = data.get("plan_monthly_data", 0)
    next_reset = data.get("data_next_reset", 0)

    if monthly_quota > 0:
        usage_ratio = (data_counter / monthly_quota) * 100
        remaining = max(0, monthly_quota - data_counter)
        print(f"本月已用流量   : {format_bytes(data_counter)} / {format_bytes(monthly_quota)} ({usage_ratio:.2f}%)")
        print(f"本月剩余流量   : {format_bytes(remaining)}")
    else:
        print(f"本月已用流量   : {format_bytes(data_counter)}")

    print(f"下次重置周期   : {format_timestamp(next_reset)}")
    print("=" * 52 + "\n")


if __name__ == "__main__":
    v_id, a_key = get_credentials()
    metrics = query_kiwivm_metrics(v_id, a_key)
    print_inspection_report(metrics)

# 通过环境变量静默传入凭据（避免在 Shell 历史中保存长命令）
export KIWIVM_VEID="1234567"
export KIWIVM_API_KEY="your_secret_api_key_here"

python3 kiwivm_monitor.py