PasarGuard
节点

官方库

用于 PasarGuard Node 集成的官方 Go 和 Python 库

PasarGuard 提供官方库以简化与 Node 的集成。这些库处理身份验证、序列化，并为所有 API 操作提供类型安全的接口。

Go 库

node-bridge

用于无缝 PasarGuard Node 集成的官方 Go 库。

仓库： PasarGuard/node_bridge

安装

go get github.com/PasarGuard/node_bridge

快速开始

package main

import (
    "context"
    "log"

    bridge "github.com/PasarGuard/node_bridge"
)

示例

启动后端

package main

import (
    "context"
    "log"
    
    "github.com/PasarGuard/node_bridge"
    pb "github.com/PasarGuard/node_bridge/proto"
)

func main() {
    client, _ := nodebridge.NewClient("https://node:443", "session-id")
    defer client.Close()

    backend := &pb.Backend{
        Type: pb.BackendType_XRAY,
        Config: `{
            "log": {"loglevel": "warning"},
            "inbounds": [...],
            "outbounds": [...]
        }`,
        Users: []*pb.User{
            {
                Email: "user1@example.com",
                Proxies: &pb.Proxy{
                    Vmess: &pb.Vmess{
                        Id: "uuid-here",
                    },
                },
                Inbounds: []string{"inbound-tag"},
            },
        },
        KeepAlive: 60,
    }

    info, err := client.Start(context.Background(), backend)
    if err != nil {
        log.Fatal(err)
    }
    
    log.Printf("后端已启动: %v", info)
}

同步用户

// 同步单个用户
user := &pb.User{
    Email: "newuser@example.com",
    Proxies: &pb.Proxy{
        Vmess: &pb.Vmess{Id: "uuid"},
    },
    Inbounds: []string{"inbound-tag"},
}

err := client.SyncUser(context.Background(), user)
if err != nil {
    log.Fatal(err)
}

// 删除用户（空入站）
user.Inbounds = []string{}
err = client.SyncUser(context.Background(), user)

获取统计

// 获取所有用户的统计
statsReq := &pb.StatRequest{
    Type: pb.StatType_UsersStat,
    Reset: false,
}

stats, err := client.GetStats(context.Background(), statsReq)
if err != nil {
    log.Fatal(err)
}

for _, stat := range stats.Stats {
    log.Printf("用户: %s, 流量: %d 字节", stat.Name, stat.Value)
}

// 获取特定用户的统计并重置
userStatsReq := &pb.StatRequest{
    Name: "user@example.com",
    Type: pb.StatType_UserStat,
    Reset: true,
}

userStats, err := client.GetStats(context.Background(), userStatsReq)

流式传输日志

ctx := context.Background()
logStream, err := client.GetLogs(ctx)
if err != nil {
    log.Fatal(err)
}

for {
    logMsg, err := logStream.Recv()
    if err != nil {
        log.Fatal(err)
    }

    log.Printf("后端日志: %s", logMsg.Detail)
}

获取系统统计

sysStats, err := client.GetSystemStats(context.Background())
if err != nil {
    log.Fatal(err)
}

log.Printf("CPU 使用率: %.2f%%", sysStats.CpuUsage)
log.Printf("内存: %d/%d MB",
    sysStats.MemUsed/1024/1024,
    sysStats.MemTotal/1024/1024)

高级用法

使用 gRPC

client, err := nodebridge.NewGRPCClient(
    "node:50051",
    "session-id",
    grpc.WithTransportCredentials(...),
)

带超时的上下文

ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()

info, err := client.GetBaseInfo(ctx)

错误处理

import "google.golang.org/grpc/codes"
import "google.golang.org/grpc/status"

info, err := client.GetBaseInfo(ctx)
if err != nil {
    st, ok := status.FromError(err)
    if ok {
        switch st.Code() {
        case codes.Unauthenticated:
            log.Println("无效的 session ID")
        case codes.Unavailable:
            log.Println("节点不可用")
        default:
            log.Printf("错误: %v", st.Message())
        }
    }
}

Python 库

node-bridge-py

用于 PasarGuard Node 集成的官方 Python 库。

仓库： PasarGuard/node-bridge-py

安装

pip install pasarguard-node-bridge

快速开始

from pasarguard_node_bridge import NodeClient

# 创建客户端
client = NodeClient(
    url="https://your-node:443",
    session_id="your-session-id"
)

# 获取节点信息
info = client.get_base_info()
print(f"节点版本: {info.version}, 运行中: {info.running}")

示例

启动后端

from pasarguard_node_bridge import NodeClient, Backend, User, Proxy, Vmess

client = NodeClient("https://node:443", "session-id")

backend = Backend(
    type=Backend.Type.XRAY,
    config="""{
        "log": {"loglevel": "warning"},
        "inbounds": [...],
        "outbounds": [...]
    }""",
    users=[
        User(
            email="user1@example.com",
            proxies=Proxy(
                vmess=Vmess(id="uuid-here")
            ),
            inbounds=["inbound-tag"]
        )
    ],
    keep_alive=60
)

info = client.start(backend)
print(f"后端已启动: {info}")

同步用户

from pasarguard_node_bridge import User, Proxy, Vmess

# 添加/更新用户
user = User(
    email="newuser@example.com",
    proxies=Proxy(
        vmess=Vmess(id="uuid")
    ),
    inbounds=["inbound-tag"]
)

client.sync_user(user)

# 删除用户（空入站）
user.inbounds = []
client.sync_user(user)

获取统计

from pasarguard_node_bridge import StatRequest, StatType

# 获取所有用户的统计
stats_req = StatRequest(
    type=StatType.USERS_STAT,
    reset=False
)

stats = client.get_stats(stats_req)
for stat in stats.stats:
    print(f"用户: {stat.name}, 流量: {stat.value} 字节")

# 获取特定用户的统计
user_stats_req = StatRequest(
    name="user@example.com",
    type=StatType.USER_STAT,
    reset=True
)

user_stats = client.get_stats(user_stats_req)

流式传输日志

# 实时流式传输日志
for log in client.get_logs():
    print(f"后端日志: {log.detail}")

获取系统统计

sys_stats = client.get_system_stats()

print(f"CPU 使用率: {sys_stats.cpu_usage:.2f}%")
print(f"内存: {sys_stats.mem_used/1024/1024}/{sys_stats.mem_total/1024/1024} MB")
print(f"下载: {sys_stats.incoming_bandwidth_speed/1024/1024:.2f} MB/s")
print(f"上传: {sys_stats.outgoing_bandwidth_speed/1024/1024:.2f} MB/s")

获取在线用户

online_req = StatRequest(
    name="user@example.com",
    type=StatType.USER_STAT
)

online_stats = client.get_user_online_stats(online_req)
print(f"用户有 {online_stats.value} 个活动连接")

# 获取 IP 列表
ip_list = client.get_user_online_ip_list(online_req)
for ip, count in ip_list.value.items():
    print(f"IP: {ip}, 连接数: {count}")

高级用法

异步支持

from pasarguard_node_bridge import AsyncNodeClient

async def main():
    client = AsyncNodeClient("https://node:443", "session-id")

    info = await client.get_base_info()
    print(info)

    await client.close()

# 使用 asyncio 运行
import asyncio
asyncio.run(main())

错误处理

from pasarguard_node_bridge import NodeError, AuthenticationError

try:
    info = client.get_base_info()
except AuthenticationError:
    print("无效的 session ID")
except NodeError as e:
    print(f"节点错误: {e}")
except Exception as e:
    print(f"意外错误: {e}")

使用 gRPC

client = NodeClient(
    url="node:50051",
    session_id="session-id",
    use_grpc=True
)

最佳实践

连接管理

Go：

// 始终关闭客户端
defer client.Close()

Python：

# 使用上下文管理器
with NodeClient(url, session_id) as client:
    # 使用客户端
    pass

# 或手动关闭
client.close()

错误处理

始终适当处理错误：

  • 身份验证错误 → 检查 SESSION_ID
  • 连接错误 → 验证节点正在运行
  • 超时错误 → 增加上下文超时

性能提示

  1. 重用客户端 - 不要为每个请求创建新客户端
  2. 批量操作 - 使用 SyncUsers 进行多个用户更新
  3. 缓存统计 - 不要过于频繁地轮询统计信息
  4. 使用流式传输 - 对于日志，使用流式传输而不是轮询

示例和教程

在库仓库中查看完整示例：

API 参考

PasarGuard Node 的完整 REST API 和 gRPC 参考

迁移指南

迁移指南集合。

On this page

Go 库node-bridge安装快速开始示例启动后端同步用户获取统计流式传输日志获取系统统计高级用法使用 gRPC带超时的上下文错误处理Python 库node-bridge-py安装快速开始示例启动后端同步用户获取统计流式传输日志获取系统统计获取在线用户高级用法异步支持错误处理使用 gRPC最佳实践连接管理错误处理性能提示示例和教程