第六部分：API接口文档

18. API概述

18.1 接口列表

18.1.1 完整接口清单

本系统提供RESTful风格的API接口，用于VPN服务的管理和监控。

18.1.2 接口分类

管理类接口：

• POST /api/vpn/start
• POST /api/vpn/stop

查询类接口：

• GET /api/vpn/status
• GET /api/system/info

测试类接口：

• GET /api/vpn/test/latency
• GET /api/vpn/test/bandwidth

健康检查：

• GET /health

18.1.3 基础URL

开发环境：

http://localhost:5000

生产环境（本项目）：

http://192.168.1.66:5000

通过VPN访问：

http://10.8.0.1:5000

18.2 认证方式

18.2.1 当前认证机制

本项目版本： 当前版本暂未实现认证机制，所有接口可直接访问。

⚠️ 安全警告：生产环境建议添加认证机制。

18.2.2 推荐认证方案

方案1：API Key认证

客户端请求头：

GET /api/vpn/status HTTP/1.1
Host: 192.168.1.66:5000
X-API-Key: your_api_key_here

服务端验证：

from flask import request, jsonify

API_KEY = "your_secret_api_key"

@app.before_request
def check_api_key():
    if request.endpoint in ['start_vpn', 'stop_vpn']:
        api_key = request.headers.get('X-API-Key')
        if api_key != API_KEY:
            return jsonify({'error': 'Unauthorized'}), 401

方案2：JWT Token认证

客户端登录获取token：

POST /api/auth/login HTTP/1.1
Content-Type: application/json

{
    "username": "admin",
    "password": "password"
}

响应：

{
    "token": "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9..."
}

后续请求携带token：

GET /api/vpn/status HTTP/1.1
Authorization: Bearer eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9...

方案3：IP白名单

仅允许特定IP访问：

ALLOWED_IPS = ['127.0.0.1', '192.168.1.0/24']

@app.before_request
def check_ip():
    if request.endpoint in ['start_vpn', 'stop_vpn']:
        client_ip = request.remote_addr
        if not is_ip_allowed(client_ip, ALLOWED_IPS):
            return jsonify({'error': 'Forbidden'}), 403

18.3 响应格式

18.3.1 成功响应

标准成功响应格式：

{
    "success": true,
    "message": "Operation completed successfully",
    "data": {
        "key": "value"
    }
}

示例1：查询状态成功

{
    "status": "running",
    "interface": "wg0",
    "peers": 1,
    "details": "interface: wg0\n  public key: rUEB..."
}

示例2：启动VPN成功

{
    "success": true,
    "message": "VPN started successfully",
    "details": "interface: wg0\n  listening port: 51820"
}

18.3.2 错误响应

标准错误响应格式：

{
    "success": false,
    "message": "Error description",
    "details": "Detailed error information"
}

示例1：VPN已在运行

{
    "success": false,
    "message": "VPN is already running",
    "details": "interface: wg0 already exists"
}

示例2：权限不足

{
    "error": "Permission denied",
    "message": "Insufficient privileges to execute command"
}

18.3.3 HTTP状态码

状态码使用示例：

# 成功
return jsonify({'status': 'ok'}), 200

# 客户端错误（VPN已运行）
return jsonify({'error': 'Already running'}), 400

# 服务器错误
return jsonify({'error': 'Internal error'}), 500

18.4 错误码定义

18.4.1 错误码列表

18.4.2 错误响应示例

{
    "success": false,
    "error_code": "E1001",
    "message": "VPN is already running",
    "details": "WireGuard interface wg0 already exists"
}

19. 接口详细说明

19.1 GET /api/vpn/status

19.1.1 接口说明

功能： 查询WireGuard VPN当前运行状态

请求方法： GET

请求URL： /api/vpn/status

认证要求： 否

请求参数： 无

19.1.2 请求示例

curl命令：

curl http://192.168.1.66:5000/api/vpn/status

Python requests：

import requests

response = requests.get('http://192.168.1.66:5000/api/vpn/status')
data = response.json()
print(data)

JavaScript fetch：

fetch('http://192.168.1.66:5000/api/vpn/status')
    .then(response => response.json())
    .then(data => console.log(data))
    .catch(error => console.error('Error:', error));

jQuery AJAX：

$.ajax({
    url: 'http://192.168.1.66:5000/api/vpn/status',
    type: 'GET',
    success: function(data) {
        console.log(data);
    },
    error: function(error) {
        console.error('Error:', error);
    }
});

19.1.3 响应示例

VPN运行中：

{
    "status": "running",
    "interface": "wg0",
    "peers": 1,
    "details": "interface: wg0\n  public key: rUEB9pR7DO7qgkbo6Ylerrh/Fvl1XA8RJuALBdlkFkk=\n  private key: (hidden)\n  listening port: 51820\n\npeer: QcDIGh0MyyLJoovsb7hMccZGwMha7aqfa/3cyFtfhE4=\n  endpoint: 192.168.1.100:54231\n  allowed ips: 10.8.0.2/32\n  latest handshake: 45 seconds ago\n  transfer: 1.23 MiB received, 856.78 KiB sent"
}

VPN已停止：

{
    "status": "stopped",
    "interface": "wg0",
    "peers": 0,
    "details": "WireGuard is not running"
}

19.1.4 响应字段说明

19.1.5 解析details字段

提取peer信息：

import re

def parse_peers(details):
    peers = []
    peer_blocks = details.split('peer:')[1:]  # 跳过interface部分
    
    for block in peer_blocks:
        peer = {}
        
        # 提取公钥
        pubkey_match = re.search(r'^([A-Za-z0-9+/=]{44})', block.strip())
        if pubkey_match:
            peer['public_key'] = pubkey_match.group(1)
        
        # 提取endpoint
        endpoint_match = re.search(r'endpoint: ([\d.]+):(\d+)', block)
        if endpoint_match:
            peer['endpoint_ip'] = endpoint_match.group(1)
            peer['endpoint_port'] = endpoint_match.group(2)
        
        # 提取allowed IPs
        allowed_match = re.search(r'allowed ips: ([^\n]+)', block)
        if allowed_match:
            peer['allowed_ips'] = allowed_match.group(1)
        
        # 提取握手时间
        handshake_match = re.search(r'latest handshake: (.+)', block)
        if handshake_match:
            peer['latest_handshake'] = handshake_match.group(1)
        
        # 提取流量
        transfer_match = re.search(r'transfer: ([^,]+), ([^\n]+)', block)
        if transfer_match:
            peer['rx'] = transfer_match.group(1)
            peer['tx'] = transfer_match.group(2)
        
        peers.append(peer)
    
    return peers

19.2 POST /api/vpn/start

19.2.1 接口说明

功能： 启动WireGuard VPN服务

请求方法： POST

请求URL： /api/vpn/start

认证要求： 是（建议）

请求参数： 无

前置条件： VPN必须处于停止状态

19.2.2 请求示例

curl命令：

curl -X POST http://192.168.1.66:5000/api/vpn/start

Python requests：

import requests

response = requests.post('http://192.168.1.66:5000/api/vpn/start')
data = response.json()

if data.get('success'):
    print('VPN启动成功')
else:
    print('VPN启动失败:', data.get('message'))

JavaScript fetch：

fetch('http://192.168.1.66:5000/api/vpn/start', {
    method: 'POST',
    headers: {
        'Content-Type': 'application/json'
    }
})
.then(response => response.json())
.then(data => {
    if (data.success) {
        console.log('VPN started successfully');
    } else {
        console.error('Failed to start VPN:', data.message);
    }
})
.catch(error => console.error('Error:', error));

带认证的请求（API Key）：

curl -X POST http://192.168.1.66:5000/api/vpn/start \
    -H "X-API-Key: your_api_key_here"

19.2.3 响应示例

启动成功：

{
    "success": true,
    "message": "VPN started successfully",
    "details": "interface: wg0\n  public key: rUEB9pR7DO7qgkbo6Ylerrh/Fvl1XA8RJuALBdlkFkk=\n  listening port: 51820"
}

启动失败（已在运行）：

{
    "success": false,
    "message": "VPN is already running",
    "details": "interface: wg0\n  public key: rUEB...\n  listening port: 51820"
}

HTTP状态码： 400

启动失败（其他错误）：

{
    "success": false,
    "message": "Failed to start VPN",
    "details": "Error: RTNETLINK answers: Operation not permitted"
}

HTTP状态码： 500

19.2.4 常见错误

错误1：权限不足

{
    "success": false,
    "message": "Failed to start VPN",
    "details": "Error: Operation not permitted"
}

解决：确保以root权限运行Flask应用

错误2：配置文件错误

{
    "success": false,
    "message": "Failed to start VPN",
    "details": "Configuration parsing error: /etc/wireguard/wg0.conf"
}

解决：检查配置文件语法

错误3：端口已占用

{
    "success": false,
    "message": "Failed to start VPN",
    "details": "Error: Address already in use"
}

解决：检查51820端口是否被占用

19.3 POST /api/vpn/stop

19.3.1 接口说明

功能： 停止WireGuard VPN服务

请求方法： POST

请求URL： /api/vpn/stop

认证要求： 是（建议）

请求参数： 无

前置条件： VPN必须处于运行状态

19.3.2 请求示例

curl命令：

curl -X POST http://192.168.1.66:5000/api/vpn/stop

Python requests：

import requests

response = requests.post('http://192.168.1.66:5000/api/vpn/stop')
data = response.json()

if data.get('success'):
    print('VPN停止成功')
else:
    print('VPN停止失败:', data.get('message'))

JavaScript确认对话框：

if (confirm('确定要停止VPN服务吗？')) {
    fetch('http://192.168.1.66:5000/api/vpn/stop', {
        method: 'POST'
    })
    .then(response => response.json())
    .then(data => {
        if (data.success) {
            alert('VPN已停止');
        } else {
            alert('停止失败: ' + data.message);
        }
    });
}

19.3.3 响应示例

停止成功：

{
    "success": true,
    "message": "VPN stopped successfully",
    "details": ""
}

停止失败（未在运行）：

{
    "success": false,
    "message": "VPN is not running",
    "details": "WireGuard interface not found"
}

HTTP状态码： 400

停止失败（其他错误）：

{
    "success": false,
    "message": "Failed to stop VPN",
    "details": "Error: Device or resource busy"
}

HTTP状态码： 500

19.3.4 注意事项

⚠️ 警告：

1. 停止VPN会断开所有客户端连接
2. 建议在操作前通知用户
3. 生产环境应添加二次确认机制
4. 记录停止操作日志

添加确认机制示例：

@app.route('/api/vpn/stop', methods=['POST'])
def stop_vpn():
    # 检查确认参数
    confirm = request.args.get('confirm', 'false')
    if confirm != 'true':
        return jsonify({
            'success': False,
            'message': 'Confirmation required',
            'hint': 'Add ?confirm=true to the request'
        }), 400
    
    # 执行停止操作
    # ...

使用：

curl -X POST "http://192.168.1.66:5000/api/vpn/stop?confirm=true"

19.4 GET /api/vpn/test/latency

19.4.1 接口说明

功能： 测试VPN隧道延迟

请求方法： GET

请求URL： /api/vpn/test/latency

认证要求： 否

请求参数： 无（固定测试10.8.0.2）

前置条件：

1. VPN必须运行
2. 至少有一个客户端连接（10.8.0.2）

19.4.2 请求示例

curl命令：

curl http://192.168.1.66:5000/api/vpn/test/latency

Python requests：

import requests

response = requests.get('http://192.168.1.66:5000/api/vpn/test/latency')
data = response.json()

if data.get('success'):
    print(f"平均延迟: {data['rtt_avg']}ms")
    print(f"丢包率: {data['packet_loss']}%")
else:
    print(f"测试失败: {data.get('message')}")

实时监控脚本：

import requests
import time

def monitor_latency(interval=5):
    """每隔interval秒测试一次延迟"""
    while True:
        try:
            response = requests.get('http://192.168.1.66:5000/api/vpn/test/latency')
            data = response.json()
            
            if data.get('success'):
                print(f"{time.strftime('%H:%M:%S')} - "
                      f"延迟: {data['rtt_avg']}ms, "
                      f"丢包: {data['packet_loss']}%")
            else:
                print(f"{time.strftime('%H:%M:%S')} - 测试失败")
        
        except Exception as e:
            print(f"{time.strftime('%H:%M:%S')} - 错误: {e}")
        
        time.sleep(interval)

# 运行监控
monitor_latency(interval=5)

19.4.3 响应示例

测试成功：

{
    "success": true,
    "target": "10.8.0.2",
    "packets_sent": 10,
    "packets_received": 10,
    "packet_loss": 0.0,
    "rtt_min": 0.032,
    "rtt_avg": 0.065,
    "rtt_max": 0.173,
    "rtt_mdev": 0.020,
    "unit": "ms"
}

测试失败（目标不可达）：

{
    "success": false,
    "message": "Ping test failed",
    "details": "Destination Host Unreachable"
}

HTTP状态码： 500

测试失败（VPN未运行）：

{
    "success": false,
    "message": "Ping test failed",
    "details": "connect: Network is unreachable"
}

19.4.4 响应字段说明

19.4.5 延迟评估标准

丢包率评估：

19.5 GET /api/vpn/test/bandwidth

19.5.1 接口说明

功能： 测试VPN带宽（基于传输统计估算）

请求方法： GET

请求URL： /api/vpn/test/bandwidth

认证要求： 否

请求参数： 无

前置条件：

1. VPN必须运行
2. 至少有一个客户端连接
3. 有一定数据传输量

⚠️ 注意：此接口通过累计传输统计估算带宽，非实时带宽测试。

19.5.2 请求示例

curl命令：

curl http://192.168.1.66:5000/api/vpn/test/bandwidth

Python requests：

import requests

response = requests.get('http://192.168.1.66:5000/api/vpn/test/bandwidth')
data = response.json()

if data.get('success'):
    print(f"下行带宽: {data['rx_mbps']} Mbps")
    print(f"上行带宽: {data['tx_mbps']} Mbps")
    print(f"总接收: {data['rx_bytes']} 字节")
    print(f"总发送: {data['tx_bytes']} 字节")
else:
    print(f"测试失败: {data.get('message')}")

格式化输出：

def format_bytes(bytes_value):
    """格式化字节数"""
    for unit in ['B', 'KB', 'MB', 'GB', 'TB']:
        if bytes_value < 1024.0:
            return f"{bytes_value:.2f} {unit}"
        bytes_value /= 1024.0
    return f"{bytes_value:.2f} PB"

response = requests.get('http://192.168.1.66:5000/api/vpn/test/bandwidth')
data = response.json()

if data.get('success'):
    print(f"下载: {format_bytes(data['rx_bytes'])} ({data['rx_mbps']} Mbps)")
    print(f"上传: {format_bytes(data['tx_bytes'])} ({data['tx_mbps']} Mbps)")

19.5.3 响应示例

测试成功：

{
    "success": true,
    "rx_bytes": 1290240,
    "tx_bytes": 877568,
    "rx_mbps": 0.17,
    "tx_mbps": 0.12,
    "note": "Bandwidth estimation based on transfer statistics"
}

测试失败（无peer数据）：

{
    "success": false,
    "message": "No peer data available",
    "details": "No connected peers"
}

HTTP状态码： 404

测试失败（无法获取统计）：

{
    "success": false,
    "message": "Failed to get transfer statistics",
    "details": "Unable to read wg0 interface"
}

HTTP状态码： 500

19.5.4 响应字段说明

19.5.5 带宽计算说明

计算公式：

# 假设数据在60秒内传输（time_window）
time_window = 60  # 秒

# 下行带宽（Mbps）
rx_mbps = (rx_bytes * 8) / (time_window * 1000000)

# 上行带宽（Mbps）
tx_mbps = (tx_bytes * 8) / (time_window * 1000000)

示例计算：

rx_bytes = 1290240 字节
time_window = 60 秒

rx_mbps = (1290240 * 8) / (60 * 1000000)
        = 10321920 / 60000000
        = 0.172 Mbps
        ≈ 0.17 Mbps

⚠️ 局限性：

1. 基于累计统计，非实时测试
2. 时间窗口假设可能不准确
3. 适合长期监控，不适合精确测速

改进建议：

使用iperf3进行实时带宽测试：

# 服务器端
iperf3 -s

# 客户端（通过VPN）
iperf3 -c 10.8.0.1

19.6 GET /api/system/info

19.6.1 接口说明

功能： 获取服务器系统信息

请求方法： GET

请求URL： /api/system/info

认证要求： 否

请求参数： 无

19.6.2 请求示例

curl命令：

curl http://192.168.1.66:5000/api/system/info

Python requests：

import requests
from datetime import datetime

response = requests.get('http://192.168.1.66:5000/api/system/info')
data = response.json()

print(f"主机名: {data['hostname']}")
print(f"内核版本: {data['kernel']}")
print(f"运行时间: {data['uptime']}")
print(f"系统负载: {data['load_average']}")
print(f"时间戳: {datetime.fromtimestamp(data['timestamp'])}")

定时获取系统信息：

function updateSystemInfo() {
    fetch('http://192.168.1.66:5000/api/system/info')
        .then(response => response.json())
        .then(data => {
            document.getElementById('hostname').textContent = data.hostname;
            document.getElementById('kernel').textContent = data.kernel;
            document.getElementById('uptime').textContent = data.uptime;
            document.getElementById('load').textContent = data.load_average;
        });
}

// 每30秒更新一次
setInterval(updateSystemInfo, 30000);
updateSystemInfo();  // 立即执行一次

19.6.3 响应示例

成功响应：

{
    "hostname": "eve-ng",
    "kernel": "5.15.0-91-generic",
    "uptime": "up 2 days, 5 hours, 32 minutes",
    "load_average": "0.15 0.18 0.12",
    "timestamp": 1711180800
}

19.6.4 响应字段说明

19.6.5 系统负载解读

load_average格式：

"0.15 0.18 0.12"
  │    │    │
  │    │    └─ 15分钟平均负载
  │    └────── 5分钟平均负载
  └─────────── 1分钟平均负载

负载评估（单核CPU）：

多核CPU调整：

实际负载 = 显示负载 / CPU核心数

示例：
4核CPU，负载2.0
实际负载 = 2.0 / 4 = 0.5（正常）

获取CPU核心数：

# Linux
nproc

# 或
cat /proc/cpuinfo | grep processor | wc -l

19.7 GET /health

19.7.1 接口说明

功能： API健康检查

请求方法： GET

请求URL： /health

认证要求： 否

请求参数： 无

用途：

• 服务存活检测
• 负载均衡器健康检查
• 监控系统探测
• CI/CD部署验证

19.7.2 请求示例

curl命令：

curl http://192.168.1.66:5000/health

健康检查脚本：

#!/bin/bash
# check-api-health.sh

API_URL="http://192.168.1.66:5000/health"

response=$(curl -s -o /dev/null -w "%{http_code}" "$API_URL")

if [ "$response" -eq 200 ]; then
    echo "✓ API is healthy"
    exit 0
else
    echo "✗ API is unhealthy (HTTP $response)"
    exit 1
fi

监控集成（Prometheus）：

from prometheus_client import Gauge
import requests

api_health = Gauge('api_health_status', 'API health status (1=healthy, 0=unhealthy)')

def check_api_health():
    try:
        response = requests.get('http://192.168.1.66:5000/health', timeout=5)
        if response.status_code == 200:
            api_health.set(1)
        else:
            api_health.set(0)
    except:
        api_health.set(0)

19.7.3 响应示例

健康状态：

{
    "status": "healthy",
    "service": "VPN Management API",
    "version": "1.0"
}

HTTP状态码： 200

不健康状态（示例）：

{
    "status": "unhealthy",
    "service": "VPN Management API",
    "version": "1.0",
    "error": "Database connection failed"
}

HTTP状态码： 503

19.7.4 响应字段说明

19.7.5 增强健康检查

检查依赖服务：

@app.route('/health', methods=['GET'])
def health_check():
    checks = {
        'api': True,
        'wireguard': check_wireguard(),
        'database': check_database(),  # 如果有数据库
        'disk_space': check_disk_space()
    }
    
    all_healthy = all(checks.values())
    
    return jsonify({
        'status': 'healthy' if all_healthy else 'unhealthy',
        'service': 'VPN Management API',
        'version': '1.0',
        'checks': checks
    }), 200 if all_healthy else 503

def check_wireguard():
    """检查WireGuard是否可用"""
    result = run_command(['which', 'wg'])
    return result['success']

def check_disk_space():
    """检查磁盘空间"""
    import shutil
    stat = shutil.disk_usage('/')
    free_percent = (stat.free / stat.total) * 100
    return free_percent > 10  # 剩余空间>10%

响应示例：

{
    "status": "healthy",
    "service": "VPN Management API",
    "version": "1.0",
    "checks": {
        "api": true,
        "wireguard": true,
        "database": true,
        "disk_space": true
    }
}

20. API调用示例

20.1 curl命令示例

20.1.1 基础调用

查询VPN状态：

curl http://192.168.1.66:5000/api/vpn/status

启动VPN：

curl -X POST http://192.168.1.66:5000/api/vpn/start

停止VPN：

curl -X POST http://192.168.1.66:5000/api/vpn/stop

延迟测试：

curl http://192.168.1.66:5000/api/vpn/test/latency

带宽测试：

curl http://192.168.1.66:5000/api/vpn/test/bandwidth

系统信息：

curl http://192.168.1.66:5000/api/system/info

健康检查：

curl http://192.168.1.66:5000/health

20.1.2 格式化输出

使用jq格式化JSON：

# 安装jq
sudo apt install jq

# 格式化输出
curl -s http://192.168.1.66:5000/api/vpn/status | jq '.'

# 提取特定字段
curl -s http://192.168.1.66:5000/api/vpn/status | jq '.status'

# 彩色输出
curl -s http://192.168.1.66:5000/api/vpn/status | jq -C '.' | less -R

使用python -m json.tool：

curl -s http://192.168.1.66:5000/api/vpn/status | python3 -m json.tool

20.1.3 错误处理

检查HTTP状态码：

response=$(curl -s -w "\n%{http_code}" http://192.168.1.66:5000/api/vpn/status)
http_code=$(echo "$response" | tail -n1)
body=$(echo "$response" | head -n-1)

if [ "$http_code" -eq 200 ]; then
    echo "成功: $body"
else
    echo "失败 (HTTP $http_code): $body"
fi

超时设置：

# 设置5秒超时
curl --max-time 5 http://192.168.1.66:5000/api/vpn/status

# 连接超时3秒，总超时10秒
curl --connect-timeout 3 --max-time 10 http://192.168.1.66:5000/api/vpn/start

重试机制：

# 失败时重试3次，间隔2秒
curl --retry 3 --retry-delay 2 http://192.168.1.66:5000/api/vpn/status

20.1.4 调试选项

显示详细信息：

# 显示请求和响应头
curl -v http://192.168.1.66:5000/api/vpn/status

# 仅显示响应头
curl -I http://192.168.1.66:5000/api/vpn/status

# 保存响应到文件
curl -o response.json http://192.168.1.66:5000/api/vpn/status

# 静默模式（不显示进度）
curl -s http://192.168.1.66:5000/api/vpn/status

跟踪重定向：

# 跟随重定向
curl -L http://192.168.1.66:5000/api/vpn/status

20.2 Python调用示例

20.2.1 基础调用

import requests
import json

# API基础URL
BASE_URL = 'http://192.168.1.66:5000'

# 查询VPN状态
def get_vpn_status():
    response = requests.get(f'{BASE_URL}/api/vpn/status')
    return response.json()

# 启动VPN
def start_vpn():
    response = requests.post(f'{BASE_URL}/api/vpn/start')
    return response.json()

# 停止VPN
def stop_vpn():
    response = requests.post(f'{BASE_URL}/api/vpn/stop')
    return response.json()

# 延迟测试
def test_latency():
    response = requests.get(f'{BASE_URL}/api/vpn/test/latency')
    return response.json()

# 带宽测试
def test_bandwidth():
    response = requests.get(f'{BASE_URL}/api/vpn/test/bandwidth')
    return response.json()

# 系统信息
def get_system_info():
    response = requests.get(f'{BASE_URL}/api/system/info')
    return response.json()

# 使用示例
if __name__ == '__main__':
    # 查询状态
    status = get_vpn_status()
    print(f"VPN状态: {status['status']}")
    print(f"连接数: {status['peers']}")
    
    # 延迟测试
    latency = test_latency()
    if latency.get('success'):
        print(f"平均延迟: {latency['rtt_avg']}ms")

20.2.2 封装API类

import requests
from typing import Dict, Optional

class WireGuardAPI:
    """WireGuard VPN管理API客户端"""
    
    def __init__(self, base_url: str = 'http://192.168.1.66:5000', 
                 timeout: int = 30, api_key: Optional[str] = None):
        self.base_url = base_url.rstrip('/')
        self.timeout = timeout
        self.session = requests.Session()
        
        # 设置API Key（如果有）
        if api_key:
            self.session.headers.update({'X-API-Key': api_key})
    
    def _request(self, method: str, endpoint: str, **kwargs) -> Dict:
        """发送HTTP请求"""
        url = f"{self.base_url}{endpoint}"
        kwargs.setdefault('timeout', self.timeout)
        
        try:
            response = self.session.request(method, url, **kwargs)
            response.raise_for_status()
            return response.json()
        except requests.exceptions.RequestException as e:
            return {
                'success': False,
                'error': str(e)
            }
    
    def get_status(self) -> Dict:
        """查询VPN状态"""
        return self._request('GET', '/api/vpn/status')
    
    def start(self) -> Dict:
        """启动VPN"""
        return self._request('POST', '/api/vpn/start')
    
    def stop(self) -> Dict:
        """停止VPN"""
        return self._request('POST', '/api/vpn/stop')
    
    def test_latency(self) -> Dict:
        """延迟测试"""
        return self._request('GET', '/api/vpn/test/latency')
    
    def test_bandwidth(self) -> Dict:
        """带宽测试"""
        return self._request('GET', '/api/vpn/test/bandwidth')
    
    def get_system_info(self) -> Dict:
        """获取系统信息"""
        return self._request('GET', '/api/system/info')
    
    def health_check(self) -> Dict:
        """健康检查"""
        return self._request('GET', '/health')
    
    def is_healthy(self) -> bool:
        """检查服务是否健康"""
        result = self.health_check()
        return result.get('status') == 'healthy'
    
    def is_running(self) -> bool:
        """检查VPN是否运行"""
        result = self.get_status()
        return result.get('status') == 'running'

# 使用示例
if __name__ == '__main__':
    # 创建API客户端
    api = WireGuardAPI()
    
    # 健康检查
    if not api.is_healthy():
        print("API服务异常")
        exit(1)
    
    # 查询状态
    if api.is_running():
        print("VPN正在运行")
        
        # 性能测试
        latency = api.test_latency()
        if latency.get('success'):
            print(f"延迟: {latency['rtt_avg']}ms")
            print(f"丢包: {latency['packet_loss']}%")
    else:
        print("VPN已停止")
        
        # 启动VPN
        result = api.start()
        if result.get('success'):
            print("VPN启动成功")
        else:
            print(f"VPN启动失败: {result.get('message')}")

20.2.3 异步调用

import asyncio
import aiohttp
from typing import Dict

class AsyncWireGuardAPI:
    """异步WireGuard API客户端"""
    
    def __init__(self, base_url: str = 'http://192.168.1.66:5000'):
        self.base_url = base_url.rstrip('/')
    
    async def _request(self, method: str, endpoint: str) -> Dict:
        """发送异步HTTP请求"""
        url = f"{self.base_url}{endpoint}"
        
        async with aiohttp.ClientSession() as session:
            async with session.request(method, url) as response:
                return await response.json()
    
    async def get_status(self) -> Dict:
        return await self._request('GET', '/api/vpn/status')
    
    async def start(self) -> Dict:
        return await self._request('POST', '/api/vpn/start')
    
    async def test_latency(self) -> Dict:
        return await self._request('GET', '/api/vpn/test/latency')
    
    async def get_all_info(self) -> Dict:
        """并发获取所有信息"""
        status, latency, system = await asyncio.gather(
            self.get_status(),
            self.test_latency(),
            self._request('GET', '/api/system/info')
        )
        
        return {
            'status': status,
            'latency': latency,
            'system': system
        }

# 使用示例
async def main():
    api = AsyncWireGuardAPI()
    
    # 并发获取所有信息
    info = await api.get_all_info()
    
    print(f"VPN状态: {info['status']['status']}")
    print(f"延迟: {info['latency'].get('rtt_avg', 'N/A')}ms")
    print(f"主机名: {info['system']['hostname']}")

# 运行
asyncio.run(main())

20.3 JavaScript调用示例

20.3.1 原生fetch API

// API基础URL
const BASE_URL = 'http://192.168.1.66:5000';

// 查询VPN状态
async function getVPNStatus() {
    try {
        const response = await fetch(`${BASE_URL}/api/vpn/status`);
        const data = await response.json();
        return data;
    } catch (error) {
        console.error('Error:', error);
        return null;
    }
}

// 启动VPN
async function startVPN() {
    try {
        const response = await fetch(`${BASE_URL}/api/vpn/start`, {
            method: 'POST'
        });
        const data = await response.json();
        return data;
    } catch (error) {
        console.error('Error:', error);
        return null;
    }
}

// 停止VPN
async function stopVPN() {
    try {
        const response = await fetch(`${BASE_URL}/api/vpn/stop`, {
            method: 'POST'
        });
        const data = await response.json();
        return data;
    } catch (error) {
        console.error('Error:', error);
        return null;
    }
}

// 延迟测试
async function testLatency() {
    try {
        const response = await fetch(`${BASE_URL}/api/vpn/test/latency`);
        const data = await response.json();
        return data;
    } catch (error) {
        console.error('Error:', error);
        return null;
    }
}

// 使用示例
async function main() {
    // 查询状态
    const status = await getVPNStatus();
    console.log('VPN状态:', status.status);
    console.log('连接数:', status.peers);
    
    // 延迟测试
    const latency = await testLatency();
    if (latency && latency.success) {
        console.log('平均延迟:', latency.rtt_avg, 'ms');
    }
}

main();

20.3.2 封装API类

class WireGuardAPI {
    constructor(baseURL = 'http://192.168.1.66:5000', apiKey = null) {
        this.baseURL = baseURL;
        this.apiKey = apiKey;
    }
    
    async request(method, endpoint) {
        const url = `${this.baseURL}${endpoint}`;
        const options = {
            method: method,
            headers: {
                'Content-Type': 'application/json'
            }
        };
        
        // 添加API Key（如果有）
        if (this.apiKey) {
            options.headers['X-API-Key'] = this.apiKey;
        }
        
        try {
            const response = await fetch(url, options);
            const data = await response.json();
            
            return {
                success: response.ok,
                status: response.status,
                data: data
            };
        } catch (error) {
            return {
                success: false,
                error: error.message
            };
        }
    }
    
    async getStatus() {
        return await this.request('GET', '/api/vpn/status');
    }
    
    async start() {
        return await this.request('POST', '/api/vpn/start');
    }
    
    async stop() {
        return await this.request('POST', '/api/vpn/stop');
    }
    
    async testLatency() {
        return await this.request('GET', '/api/vpn/test/latency');
    }
    
    async testBandwidth() {
        return await this.request('GET', '/api/vpn/test/bandwidth');
    }
    
    async getSystemInfo() {
        return await this.request('GET', '/api/system/info');
    }
    
    async healthCheck() {
        return await this.request('GET', '/health');
    }
}

// 使用示例
const api = new WireGuardAPI();

// 查询状态
api.getStatus().then(result => {
    if (result.success) {
        console.log('VPN状态:', result.data.status);
    } else {
        console.error('查询失败:', result.error);
    }
});

// 启动VPN（带确认）
async function startVPNWithConfirm() {
    if (confirm('确定要启动VPN吗？')) {
        const result = await api.start();
        if (result.success && result.data.success) {
            alert('VPN启动成功');
        } else {
            alert('VPN启动失败: ' + result.data.message);
        }
    }
}

20.3.3 React Hook示例

import { useState, useEffect } from 'react';

// 自定义Hook：VPN状态管理
function useVPNStatus(interval = 5000) {
    const [status, setStatus] = useState(null);
    const [loading, setLoading] = useState(true);
    const [error, setError] = useState(null);
    
    useEffect(() => {
        const fetchStatus = async () => {
            try {
                const response = await fetch('http://192.168.1.66:5000/api/vpn/status');
                const data = await response.json();
                setStatus(data);
                setError(null);
            } catch (err) {
                setError(err.message);
            } finally {
                setLoading(false);
            }
        };
        
        // 立即执行一次
        fetchStatus();
        
        // 定时刷新
        const timer = setInterval(fetchStatus, interval);
        
        // 清理
        return () => clearInterval(timer);
    }, [interval]);
    
    return { status, loading, error };
}

// 使用Hook的组件
function VPNStatusCard() {
    const { status, loading, error } = useVPNStatus(5000);
    
    if (loading) return <div>加载中...</div>;
    if (error) return <div>错误: {error}</div>;
    
    return (
        <div className="status-card">
            <h3>VPN状态</h3>
            <p>状态: {status.status}</p>
            <p>连接数: {status.peers}</p>
        </div>
    );
}

20.4 完整应用示例

20.4.1 命令行监控工具

#!/usr/bin/env python3
"""
WireGuard VPN监控工具
"""
import requests
import time
import sys
from datetime import datetime

BASE_URL = 'http://192.168.1.66:5000'

def get_status():
    """获取VPN状态"""
    try:
        response = requests.get(f'{BASE_URL}/api/vpn/status', timeout=5)
        return response.json()
    except:
        return None

def get_latency():
    """获取延迟"""
    try:
        response = requests.get(f'{BASE_URL}/api/vpn/test/latency', timeout=10)
        return response.json()
    except:
        return None

def print_status():
    """打印状态信息"""
    # 清屏
    print('\033[2J\033[H', end='')
    
    # 打印标题
    print('=' * 60)
    print('WireGuard VPN 实时监控')
    print('=' * 60)
    print(f'时间: {datetime.now().strftime("%Y-%m-%d %H:%M:%S")}')
    print('-' * 60)
    
    # 获取状态
    status = get_status()
    if status:
        if status['status'] == 'running':
            print(f'状态: ✓ 运行中')
            print(f'连接数: {status["peers"]}')
            
            # 获取延迟
            latency = get_latency()
            if latency and latency.get('success'):
                print(f'延迟: {latency["rtt_avg"]}ms')
                print(f'丢包: {latency["packet_loss"]}%')
            else:
                print('延迟: 测试失败')
        else:
            print(f'状态: ✗ 已停止')
    else:
        print('状态: ✗ API无响应')
    
    print('-' * 60)
    print('按Ctrl+C退出')

def main():
    """主函数"""
    try:
        while True:
            print_status()
            time.sleep(5)
    except KeyboardInterrupt:
        print('\n\n监控已停止')
        sys.exit(0)

if __name__ == '__main__':
    main()

使用方法：

chmod +x vpn-monitor.py
./vpn-monitor.py