From fd82d82fe35db42c1c0ab4eac8ad3d69261a1db5 Mon Sep 17 00:00:00 2001 From: wangjia <809946525@qq.com> Date: Thu, 25 Jun 2026 07:04:26 +0800 Subject: [PATCH] =?UTF-8?q?test(e2e):=20=E7=AC=AC5=E5=88=80b=20L4=20gRPC?= =?UTF-8?q?=20=E5=85=A8=E9=93=BE=E8=B7=AF(enroll=E2=86=92ReportUsage?= =?UTF-8?q?=E2=86=92=E7=BB=9F=E8=AE=A1=E7=9C=9F=E5=85=A5=E5=BA=93=E7=9C=9F?= =?UTF-8?q?=E8=AF=BB=E5=87=BA)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 在 HTTP 段基础上接 gRPC mTLS 全链路,真验证「统计准不准」: - 脚本:生成 Node CA + gRPC server 证书(镜像 deploy.sh)、起 server 带 gRPC、 nodectl 签 bootstrap token,传 E2E_GRPC_ADDR/CA/token/node 给 driver - driver:复用 agentd.EnsureEnrolled 真 TCP enroll → enroll 到的 client cert 建 mTLS → ReportUsage 注入 1GiB/100MiB(dp_uuid=用户) → 轮询 /v1/usage 断言 bytes_up=1GiB·bytes_down=100MiB 真入库真读回 - 本地全链路绿;go build ./... 不受影响(tag e2e) Co-Authored-By: Claude Opus 4.8 --- scripts/e2e-smoke.sh | 63 +++++++++++++------ server/test/e2e/smoke_test.go | 114 ++++++++++++++++++++++++++++++++++ 2 files changed, 158 insertions(+), 19 deletions(-) diff --git a/scripts/e2e-smoke.sh b/scripts/e2e-smoke.sh index 9accd85..9080fa0 100755 --- a/scripts/e2e-smoke.sh +++ b/scripts/e2e-smoke.sh @@ -1,16 +1,16 @@ #!/usr/bin/env bash # e2e-smoke.sh — L4 后端进程级端到端冒烟。 # -# 真起打包好的 server 二进制(sqlite 临时库 + Redis),跑一条真 HTTP/gRPC over TCP/TLS -# 的用户旅程,验证 in-process 测试给不了的信心:二进制能起、迁移能跑、路由/鉴权/DB -# 端到端通、用量真入库真从 API 读出。 +# 真起打包好的 server 二进制(sqlite 临时库 + Redis + gRPC mTLS),跑一条真 HTTP/gRPC +# over TCP/TLS 的全链路,验证 in-process 测试给不了的信心:二进制能起、迁移能跑、 +# 路由/鉴权/DB 通、用量经 gRPC enroll+ReportUsage 真入库、再经 HTTP /v1/usage 真读出。 # # 用法: bash scripts/e2e-smoke.sh -# 依赖: go · openssl · python3(取随机端口) · curl +# 依赖: go · openssl · python3(随机端口+UUID) · curl # Redis 三选一:E2E_REDIS_ADDR(已有) > docker(redis:7-alpine) > 内嵌 miniredis # -# 阶段: 5a = HTTP 段(登录→/v1/me→/v1/usage);5b 起 gRPC 全链路(enroll+ReportUsage)。 -# 全部断言在 Go driver: server/test/e2e/smoke_test.go(build tag e2e)。 +# 全部断言在 Go driver: server/test/e2e/smoke_test.go(build tag e2e): +# HTTP 段(登录→/v1/me→/v1/usage) + gRPC 全链路(enroll→ReportUsage→/v1/usage bytes 断言)。 set -euo pipefail ROOT="$(cd "$(dirname "$0")/.." && pwd)" @@ -34,19 +34,14 @@ rand_port() { python3 -c 'import socket;s=socket.socket();s.bind(("",0));print(s.getsockname()[1]);s.close()' } -# start_redis — 按优先级落一个真 TCP Redis,地址写入 REDIS_ADDR_RESOLVED。 start_redis() { if [ -n "${E2E_REDIS_ADDR:-}" ]; then - REDIS_ADDR_RESOLVED="$E2E_REDIS_ADDR" - echo "[e2e] 用已有 Redis $REDIS_ADDR_RESOLVED" - return + REDIS_ADDR_RESOLVED="$E2E_REDIS_ADDR"; echo "[e2e] 用已有 Redis $REDIS_ADDR_RESOLVED"; return fi if docker info >/dev/null 2>&1; then local p; p="$(rand_port)" REDIS_CID="$(docker run -d -p "$p:6379" redis:7-alpine)" - REDIS_ADDR_RESOLVED="127.0.0.1:$p" - echo "[e2e] docker Redis $REDIS_ADDR_RESOLVED" - return + REDIS_ADDR_RESOLVED="127.0.0.1:$p"; echo "[e2e] docker Redis $REDIS_ADDR_RESOLVED"; return fi echo "[e2e] 无 docker,用内嵌 miniredis 兜底 ..." ( cd server && go run -tags e2e ./test/e2e/miniredis ) > "$TMP/mr.addr" 2>"$TMP/mr.log" & @@ -58,24 +53,46 @@ start_redis() { } HTTP_PORT="$(rand_port)" +GRPC_PORT="$(rand_port)" +NODE_UUID="$(python3 -c 'import uuid;print(uuid.uuid4())')" start_redis -echo "[e2e] 生成 JWT 密钥 ..." +echo "[e2e] 生成 JWT 密钥 + Node CA + gRPC server 证书 ..." openssl genrsa -out "$TMP/jwt_private.pem" 2048 2>/dev/null openssl rsa -in "$TMP/jwt_private.pem" -pubout -out "$TMP/jwt_public.pem" 2>/dev/null +# Node CA(ECDSA P-256,mtls.NewCA 读它)+ gRPC server 证书(CA 签,SAN localhost+127.0.0.1) +# —— 镜像 deploy/single-node/deploy.sh L119-142。 +openssl ecparam -name prime256v1 -genkey -noout -out "$TMP/ca.key" 2>/dev/null +openssl req -x509 -new -key "$TMP/ca.key" -days 3650 -out "$TMP/ca.crt" \ + -subj "/O=Pangolin/CN=Pangolin Node CA" \ + -addext "basicConstraints=critical,CA:TRUE" \ + -addext "keyUsage=critical,keyCertSign,cRLSign" 2>/dev/null +openssl ecparam -name prime256v1 -genkey -noout -out "$TMP/grpc.key" 2>/dev/null +openssl req -new -key "$TMP/grpc.key" -out "$TMP/grpc.csr" -subj "/CN=pangolin-control-plane" 2>/dev/null +cat > "$TMP/grpc.ext" </dev/null -echo "[e2e] go build server + migrate ..." -( cd server && go build -o "$TMP/server" ./cmd/server && go build -o "$TMP/migrate" ./cmd/migrate ) +echo "[e2e] go build server + migrate + nodectl ..." +( cd server && go build -o "$TMP/server" ./cmd/server \ + && go build -o "$TMP/migrate" ./cmd/migrate \ + && go build -o "$TMP/nodectl" ./cmd/nodectl ) DB_FILE="$TMP/e2e.db" echo "[e2e] migrate(sqlite) ..." DB_DRIVER=sqlite DB_DSN="$DB_FILE" "$TMP/migrate" up -echo "[e2e] 起 server :$HTTP_PORT ..." +echo "[e2e] 起 server HTTP :$HTTP_PORT gRPC :$GRPC_PORT ..." DB_DRIVER=sqlite DB_DSN="$DB_FILE" \ REDIS_ADDR="$REDIS_ADDR_RESOLVED" \ ADDR=":$HTTP_PORT" \ + GRPC_ADDR=":$GRPC_PORT" \ + CA_KEY_PATH="$TMP/ca.key" CA_CERT_PATH="$TMP/ca.crt" \ + GRPC_CERT_PATH="$TMP/grpc.crt" GRPC_KEY_PATH="$TMP/grpc.key" \ JWT_PRIVATE_KEY_PATH="$TMP/jwt_private.pem" \ JWT_KEY_ID="e2e-key-1" \ JWT_PUBLIC_KEYS="e2e-key-1:$TMP/jwt_public.pem" \ @@ -96,14 +113,22 @@ done [ "$ok" = 1 ] || { echo "[e2e] ✗ /healthz 超时,日志:" >&2; cat "$TMP/server.log" >&2; exit 1; } echo "[e2e] 控制面健康。" -echo "[e2e] 跑 Go driver ..." +echo "[e2e] 签发 agent bootstrap token(node=$NODE_UUID)..." +TOKEN="$(REDIS_ADDR="$REDIS_ADDR_RESOLVED" "$TMP/nodectl" bootstrap-token -node="$NODE_UUID")" +[ -n "$TOKEN" ] || { echo "[e2e] ✗ bootstrap token 签发失败" >&2; cat "$TMP/server.log" >&2; exit 1; } + +echo "[e2e] 跑 Go driver(HTTP + gRPC 全链路)..." ( cd server E2E_HTTP="http://127.0.0.1:$HTTP_PORT" \ E2E_DB_DSN="$DB_FILE" \ E2E_SERVER_LOG="$TMP/server.log" \ + E2E_GRPC_ADDR="127.0.0.1:$GRPC_PORT" \ + E2E_CA_CERT="$TMP/ca.crt" \ + E2E_BOOTSTRAP_TOKEN="$TOKEN" \ + E2E_NODE_UUID="$NODE_UUID" \ DB_DRIVER=sqlite \ go test -tags e2e ./test/e2e -run TestE2ESmoke -count=1 -v ) -echo "[e2e] ✓ PASS — 进程级端到端冒烟通过" +echo "[e2e] ✓ PASS — 进程级端到端冒烟通过(含 gRPC 全链路)" diff --git a/server/test/e2e/smoke_test.go b/server/test/e2e/smoke_test.go index e531a7d..8b3df66 100644 --- a/server/test/e2e/smoke_test.go +++ b/server/test/e2e/smoke_test.go @@ -8,6 +8,9 @@ package e2e import ( "bytes" + "context" + "crypto/tls" + "crypto/x509" "encoding/json" "io" "net/http" @@ -15,8 +18,12 @@ import ( "testing" "time" + "github.com/wangjia/pangolin/server/internal/agentd" "github.com/wangjia/pangolin/server/internal/auth" "github.com/wangjia/pangolin/server/internal/db" + agentv1 "github.com/wangjia/pangolin/server/internal/pb/agentv1" + "google.golang.org/grpc" + "google.golang.org/grpc/credentials" ) const ( @@ -59,8 +66,115 @@ func TestE2ESmoke(t *testing.T) { assertUnauthorized(t, base, "/v1/me") t.Log("✓ HTTP 段通过:登录 → /v1/me(dp_uuid/plan/email) → /v1/usage(鉴权)") + + // ── gRPC 全链路:enroll → ReportUsage → /v1/usage bytes 断言 ────── + if os.Getenv("E2E_GRPC_ADDR") != "" { + runGRPCChain(t, base, token) + } else { + t.Log("E2E_GRPC_ADDR 未设,跳过 gRPC 全链路段") + } } +// runGRPCChain 镜像真 agent:enroll(mTLS,bootstrap token)→ 用 enroll 到的 client +// cert 调 ReportUsage 注入用量 → 轮询 HTTP /v1/usage 断言 bytes 真入库真读出。 +func runGRPCChain(t *testing.T, base, token string) { + t.Helper() + grpcAddr := os.Getenv("E2E_GRPC_ADDR") + caPEM, err := os.ReadFile(os.Getenv("E2E_CA_CERT")) + if err != nil { + t.Fatalf("读 CA cert: %v", err) + } + ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second) + defer cancel() + + // 1. enroll(复用生产 agent 的 EnsureEnrolled,真 TCP enroll dialer) + cfg := agentd.Config{ + BootstrapToken: os.Getenv("E2E_BOOTSTRAP_TOKEN"), + StateDir: t.TempDir(), + ServerName: "localhost", + AgentVersion: "e2e-smoke", + } + nodeUUID, err := agentd.EnsureEnrolled(ctx, cfg, enrollDialer(grpcAddr, caPEM)) + if err != nil { + t.Fatalf("enroll: %v", err) + } + if nodeUUID == "" { + t.Fatal("enroll 返回空 node uuid") + } + t.Logf("✓ enroll 成功 node=%s", nodeUUID) + + // 2. 用 enroll 到的 client cert 建 mTLS 连接 → ReportUsage + cert, err := tls.LoadX509KeyPair(cfg.CertPath(), cfg.KeyPath()) + if err != nil { + t.Fatalf("加载 node 证书: %v", err) + } + pool := x509.NewCertPool() + if !pool.AppendCertsFromPEM(caPEM) { + t.Fatal("追加 CA 失败") + } + conn, err := grpc.NewClient(grpcAddr, grpc.WithTransportCredentials( + credentials.NewTLS(&tls.Config{ + Certificates: []tls.Certificate{cert}, + RootCAs: pool, + ServerName: "localhost", + }), + )) + if err != nil { + t.Fatalf("mTLS 拨号: %v", err) + } + defer conn.Close() + + const wantUp, wantDown = int64(1) << 30, int64(100) << 20 // 1GiB / 100MiB + now := time.Now().Unix() + if _, err := agentv1.NewAgentServiceClient(conn).ReportUsage(ctx, &agentv1.UsageReport{ + NodeUUID: nodeUUID, + WindowStartUnix: now - 60, + WindowEndUnix: now, + Entries: []*agentv1.UsageEntry{{ + DpUUID: e2eDPUUID, BytesUp: wantUp, BytesDown: wantDown, SessionMinutes: 1, + }}, + }); err != nil { + t.Fatalf("ReportUsage: %v", err) + } + t.Log("✓ ReportUsage 上报 1GiB/100MiB") + + // 3. 轮询 /v1/usage 直到用量入库可读(聚合是同步写,留少量重试容错) + var up, down int64 + for i := 0; i < 30; i++ { + up, down = sumUsage(getJSON(t, base, "/v1/usage", token)) + if up == wantUp && down == wantDown { + t.Logf("✓ /v1/usage 读回 bytes_up=%d bytes_down=%d —— gRPC 全链路通", up, down) + return + } + time.Sleep(150 * time.Millisecond) + } + t.Fatalf("/v1/usage 最终 bytes_up=%d(want %d) bytes_down=%d(want %d)", up, wantUp, down, wantDown) +} + +// enrollDialer 返回未认证(无 client cert,bootstrap token 鉴权)的 enroll 连接, +// 仅信任 server CA。enroll 拿到 client cert 后,后续 RPC 才走 mTLS。 +func enrollDialer(addr string, caPEM []byte) agentd.EnrollDialer { + return func(_ context.Context) (*grpc.ClientConn, error) { + pool := x509.NewCertPool() + pool.AppendCertsFromPEM(caPEM) + return grpc.NewClient(addr, grpc.WithTransportCredentials( + credentials.NewTLS(&tls.Config{RootCAs: pool, ServerName: "localhost"}), + )) + } +} + +func sumUsage(m map[string]any) (up, down int64) { + pts, _ := m["points"].([]any) + for _, p := range pts { + pm, _ := p.(map[string]any) + up += toInt64(pm["bytes_up"]) + down += toInt64(pm["bytes_down"]) + } + return up, down +} + +func toInt64(v any) int64 { f, _ := v.(float64); return int64(f) } + // seedUser 直接写 sqlite 造一个 active 用户(复用 argon2id HashPassword), // 比走注册验证码稳。server 刚起基本空闲,一次性 INSERT 锁风险极低。 func seedUser(t *testing.T, dsn string) {