From 82b3988974a393000f6229015006c3ef887f7a73 Mon Sep 17 00:00:00 2001 From: wangjia <809946525@qq.com> Date: Tue, 28 Jul 2026 08:11:48 +0800 Subject: [PATCH] =?UTF-8?q?feat(routing):=20connect=20=E8=AF=BB=20per-user?= =?UTF-8?q?=20=E6=A1=A3=E6=A1=88=E4=BC=A0=E5=85=A5=E6=B8=B2=E6=9F=93(fail-?= =?UTF-8?q?safe=20=E5=9B=9E=E9=80=80=E9=BB=98=E8=AE=A4)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- server/cmd/server/main.go | 2 +- server/internal/httpapi/nodes.go | 28 ++- server/internal/httpapi/nodes_connect_test.go | 205 ++++++++++++++++++ .../internal/httpapi/nodes_disconnect_test.go | 2 +- 4 files changed, 230 insertions(+), 7 deletions(-) create mode 100644 server/internal/httpapi/nodes_connect_test.go diff --git a/server/cmd/server/main.go b/server/cmd/server/main.go index 36e45f2..0d12f8c 100644 --- a/server/cmd/server/main.go +++ b/server/cmd/server/main.go @@ -431,7 +431,7 @@ func mountV1(r chi.Router, sqlDB *sql.DB, rdb *redis.Client, nodeSvc *nodes.Serv privateSplitDomains = append(privateSplitDomains, d) } } - nodeAPI = httpapi.NewNodeAPI(nodeStore, nodeSvc.Hub(), nodeSvc.Load(), os.Getenv("NODE_DERIVE_KEY"), publicURL, privateSplitDomains) + nodeAPI = httpapi.NewNodeAPI(nodeStore, nodeSvc.Hub(), nodeSvc.Load(), os.Getenv("NODE_DERIVE_KEY"), publicURL, privateSplitDomains, routing.NewStore(sqlDB)) } // 国内分流(#5)的 rule-set 静态服务:GET /v1/rules/{name}.srs(自托管, diff --git a/server/internal/httpapi/nodes.go b/server/internal/httpapi/nodes.go index 8f0e232..ab38bcd 100644 --- a/server/internal/httpapi/nodes.go +++ b/server/internal/httpapi/nodes.go @@ -16,6 +16,7 @@ import ( "github.com/wangjia/pangolin/server/internal/auth" "github.com/wangjia/pangolin/server/internal/nodes" agentv1 "github.com/wangjia/pangolin/server/internal/pb/agentv1" + "github.com/wangjia/pangolin/server/internal/routing" ) const ( @@ -45,13 +46,17 @@ type NodeAPI struct { // privateSplitDomains 私有服务域名(PANGOLIN_PRIVATE_SPLIT_DOMAINS,逗号分隔), // 见 ClientConfigOpts.PrivateSplitDomains;空则不渲染相关规则。 privateSplitDomains []string + // routingStore 读取用户可配置分流档案(Task 1)。nil = 未注入(旧路径/未迁移 + // 调用点),ConnectNode 回退到 ?split_cn query 兜底,不产坏配置。 + routingStore *routing.Store } // NewNodeAPI creates a NodeAPI. load may be nil (then all nodes are treated as -// data-plane healthy — agent gRPC liveness still gates status). -func NewNodeAPI(store nodes.NodeStore, hub *nodes.Hub, load nodeLoadReader, deriveKey, rulesBaseURL string, privateSplitDomains []string) *NodeAPI { +// data-plane healthy — agent gRPC liveness still gates status). routingStore +// may be nil (per-user 路由档案不生效,ConnectNode 回退到 ?split_cn query)。 +func NewNodeAPI(store nodes.NodeStore, hub *nodes.Hub, load nodeLoadReader, deriveKey, rulesBaseURL string, privateSplitDomains []string, routingStore *routing.Store) *NodeAPI { return &NodeAPI{store: store, hub: hub, load: load, deriveKey: deriveKey, - rulesBaseURL: rulesBaseURL, privateSplitDomains: privateSplitDomains} + rulesBaseURL: rulesBaseURL, privateSplitDomains: privateSplitDomains, routingStore: routingStore} } // dataPlaneHealthy reports the node's last sing-box health (default true when @@ -316,10 +321,23 @@ func (a *NodeAPI) ConnectNode(w http.ResponseWriter, r *http.Request) { } // 7. Render and return the full sing-box CLIENT config JSON. - // split_cn=1/true → 国内 IP/域名直连(#5);客户端按 smartRoute 偏好传。 + // 可配置分流(Phase 1):读取用户路由档案,读取/解析失败一律回退 prof=nil + // (fail-safe,不产坏配置)。有档案时国内分流开关由档案决定;无档案(未迁移 + // 用户/store 未注入)保留 ?split_cn query 兜底旧行为。 + var prof *routing.Profile + if a.routingStore != nil { + if p, perr := a.routingStore.Get(r.Context(), uid); perr == nil { + prof = p + } else { + slog.Warn("connect: routing profile load failed, falling back to default", "user", uid, "err", perr) + } + } splitCN := r.URL.Query().Get("split_cn") == "1" || r.URL.Query().Get("split_cn") == "true" + if prof != nil { + splitCN = prof.Mode == "rule" && prof.Builtin.ChinaDirect + } cfgJSON, renderErr := BuildClientConfig(node, dpUUID, a.deriveKey, - ClientConfigOpts{SplitCN: splitCN, RulesBaseURL: a.rulesBaseURL, + ClientConfigOpts{Profile: prof, SplitCN: splitCN, RulesBaseURL: a.rulesBaseURL, PrivateSplitDomains: a.privateSplitDomains}) if renderErr != nil { slog.Error("connect: build client config failed", "node", nodeUUID, "err", renderErr) diff --git a/server/internal/httpapi/nodes_connect_test.go b/server/internal/httpapi/nodes_connect_test.go new file mode 100644 index 0000000..ead8e0c --- /dev/null +++ b/server/internal/httpapi/nodes_connect_test.go @@ -0,0 +1,205 @@ +package httpapi + +import ( + "context" + "database/sql" + "encoding/json" + "net/http/httptest" + "strings" + "testing" + "time" + + "github.com/alicebob/miniredis/v2" + "github.com/go-chi/chi/v5" + "github.com/redis/go-redis/v9" + + "github.com/wangjia/pangolin/server/internal/codes" + "github.com/wangjia/pangolin/server/internal/config" + "github.com/wangjia/pangolin/server/internal/nodes" + agentv1 "github.com/wangjia/pangolin/server/internal/pb/agentv1" + "github.com/wangjia/pangolin/server/internal/routing" + "github.com/wangjia/pangolin/server/internal/store" +) + +// fakeConnectStore implements just the NodeStore methods ConnectNode's happy +// path touches when the entitlement disables the optional gates (AdGate=false, +// MaxDevices=0, DailyMB invalid) — everything else panics via the embedded nil +// interface (未用到即安全，同 nodes_disconnect_test.go 的写法)。 +type fakeConnectStore struct { + nodes.NodeStore + + node *nodes.NodeRow + ent *nodes.Entitlement + devDp string +} + +func (f *fakeConnectStore) EntitlementForUser(context.Context, int64) (*nodes.Entitlement, error) { + return f.ent, nil +} + +func (f *fakeConnectStore) NodeByUUID(context.Context, string) (*nodes.NodeRow, error) { + return f.node, nil +} + +func (f *fakeConnectStore) EnsureDeviceDpUUID(context.Context, int64, string) (string, int64, error) { + return f.devDp, 1, nil +} + +func (f *fakeConnectStore) PersistCredential(context.Context, int64, *agentv1.Credential, time.Time) error { + return nil +} + +// openRoutingDB opens an in-memory SQLite DB migrated up, for a real +// routing.Store in tests (same pattern as internal/routing/store_sqlite_test.go). +func openRoutingDB(t *testing.T) *sql.DB { + t.Helper() + db, err := store.Open(&config.Config{Driver: "sqlite", DSN: ":memory:"}) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = db.Close() }) + if err := store.MigrateUp(db, "sqlite"); err != nil { + t.Fatal(err) + } + return db +} + +// newOnlineHub returns a real *nodes.Hub (backed by miniredis) with nodeUUID +// registered online, so ConnectNode's a.hub.IsOnline(...) gate passes and +// Push(...) succeeds without a real Redis deployment. +func newOnlineHub(t *testing.T, nodeUUID string) *nodes.Hub { + t.Helper() + mr := miniredis.RunT(t) + rdb := redis.NewClient(&redis.Options{Addr: mr.Addr()}) + t.Cleanup(func() { rdb.Close() }) + hub := nodes.NewHub(rdb) + _, done := hub.Register(nodeUUID) + t.Cleanup(done) + return hub +} + +func doConnect(t *testing.T, api *NodeAPI, uid int64) *httptest.ResponseRecorder { + t.Helper() + req := httptest.NewRequest("POST", "/v1/nodes/node-1/connect", strings.NewReader(`{"device_id":"dev-1"}`)) + rctx := chi.NewRouteContext() + rctx.URLParams.Add("id", "node-1") + ctx := context.WithValue(req.Context(), chi.RouteCtxKey, rctx) + ctx = context.WithValue(ctx, codes.CtxKeyUserID, uid) + rec := httptest.NewRecorder() + api.ConnectNode(rec, req.WithContext(ctx)) + return rec +} + +// TestConnectNode_ReadsRoutingProfile is Task 5's Step 1: a profile with a +// user rule is stored for the connecting user; connect must read it (via +// routingStore.Get) and forward it into BuildClientConfig — the returned +// sing-box config's route.rules must contain that user rule. +func TestConnectNode_ReadsRoutingProfile(t *testing.T) { + const uid = int64(42) + + db := openRoutingDB(t) + rst := routing.NewStore(db) + p := routing.Default() + p.Rules = []routing.Rule{ + {Type: "domain_suffix", Value: "github.com", Action: "proxy", Enabled: true}, + } + if err := rst.Upsert(context.Background(), uid, p); err != nil { + t.Fatal(err) + } + + hub := newOnlineHub(t, "node-1") + nodeStore := &fakeConnectStore{ + node: &nodes.NodeRow{ID: 1, UUID: "node-1", Status: "up", Endpoint: "1.2.3.4:443", + RealityPBK: "pbk", RealityShortID: "sid", RealitySNI: "www.apple.com"}, + ent: &nodes.Entitlement{DpUUID: "acct-dp"}, // AdGate=false, MaxDevices=0, DailyMB invalid → 跳过额外配额查询 + devDp: "device-dp", + } + api := NewNodeAPI(nodeStore, hub, nil, "k", "http://x", nil, rst) + + rec := doConnect(t, api, uid) + if rec.Code != 200 { + t.Fatalf("status = %d, body=%s", rec.Code, rec.Body.String()) + } + + var cfg map[string]any + if err := json.Unmarshal(rec.Body.Bytes(), &cfg); err != nil { + t.Fatalf("unmarshal config: %v (body=%s)", err, rec.Body.String()) + } + rules, _ := cfg["route"].(map[string]any)["rules"].([]any) + if ruleIndexByDomain(rules, "github.com") < 0 { + t.Fatalf("connect config missing user routing rule for github.com; rules=%v", rules) + } +} + +// TestConnectNode_NilRoutingStore_FailsSafe: routingStore not injected (nil, +// e.g. an unmigrated call site) must not error/panic — connect renders the +// old default (prof=nil), falling back to the ?split_cn query. +func TestConnectNode_NilRoutingStore_FailsSafe(t *testing.T) { + const uid = int64(43) + + hub := newOnlineHub(t, "node-1") + nodeStore := &fakeConnectStore{ + node: &nodes.NodeRow{ID: 1, UUID: "node-1", Status: "up", Endpoint: "1.2.3.4:443", + RealityPBK: "pbk", RealityShortID: "sid", RealitySNI: "www.apple.com"}, + ent: &nodes.Entitlement{DpUUID: "acct-dp"}, + devDp: "device-dp", + } + api := NewNodeAPI(nodeStore, hub, nil, "k", "http://x", nil, nil) // routingStore=nil + + rec := doConnect(t, api, uid) + if rec.Code != 200 { + t.Fatalf("status = %d, body=%s", rec.Code, rec.Body.String()) + } + var cfg map[string]any + if err := json.Unmarshal(rec.Body.Bytes(), &cfg); err != nil { + t.Fatalf("unmarshal config: %v (body=%s)", err, rec.Body.String()) + } + // 无档案、无 ?split_cn query → 与既有行为一致:不分流,route 里无 rule_set。 + if _, ok := cfg["route"].(map[string]any)["rule_set"]; ok { + t.Errorf("no profile + no split_cn query should have no rule_set, got %v", cfg["route"]) + } +} + +// TestConnectNode_RoutingStoreErr_FailsSafe: routingStore.Get erroring (e.g. +// corrupt profile_json) must not fail the connect — falls back to prof=nil, +// same as no profile at all, and still honors the ?split_cn query fallback. +func TestConnectNode_RoutingStoreErr_FailsSafe(t *testing.T) { + const uid = int64(44) + + db := openRoutingDB(t) + // 写入一条无法反序列化的 profile_json，模拟 Get 出错。 + if _, err := db.Exec(`INSERT INTO routing_profiles (user_id, profile_json, updated_at) VALUES (?,?,?)`, + uid, "{not-json", time.Now().UTC()); err != nil { + t.Fatal(err) + } + rst := routing.NewStore(db) + + hub := newOnlineHub(t, "node-1") + nodeStore := &fakeConnectStore{ + node: &nodes.NodeRow{ID: 1, UUID: "node-1", Status: "up", Endpoint: "1.2.3.4:443", + RealityPBK: "pbk", RealityShortID: "sid", RealitySNI: "www.apple.com"}, + ent: &nodes.Entitlement{DpUUID: "acct-dp"}, + devDp: "device-dp", + } + api := NewNodeAPI(nodeStore, hub, nil, "k", "http://x", nil, rst) + + req := httptest.NewRequest("POST", "/v1/nodes/node-1/connect?split_cn=1", strings.NewReader(`{"device_id":"dev-1"}`)) + rctx := chi.NewRouteContext() + rctx.URLParams.Add("id", "node-1") + ctx := context.WithValue(req.Context(), chi.RouteCtxKey, rctx) + ctx = context.WithValue(ctx, codes.CtxKeyUserID, uid) + rec := httptest.NewRecorder() + api.ConnectNode(rec, req.WithContext(ctx)) + + if rec.Code != 200 { + t.Fatalf("status = %d, body=%s", rec.Code, rec.Body.String()) + } + var cfg map[string]any + if err := json.Unmarshal(rec.Body.Bytes(), &cfg); err != nil { + t.Fatalf("unmarshal config: %v (body=%s)", err, rec.Body.String()) + } + // prof=nil(读档案出错回退)+ ?split_cn=1 query 兜底 → 仍走国内分流(旧行为)。 + if _, ok := cfg["route"].(map[string]any)["rule_set"]; !ok { + t.Errorf("routing store error should fall back to ?split_cn query behavior (rule_set expected), got %v", cfg["route"]) + } +} diff --git a/server/internal/httpapi/nodes_disconnect_test.go b/server/internal/httpapi/nodes_disconnect_test.go index d28c7cb..c2a9d7e 100644 --- a/server/internal/httpapi/nodes_disconnect_test.go +++ b/server/internal/httpapi/nodes_disconnect_test.go @@ -51,7 +51,7 @@ func (f *fakeDisconnectStore) PersistCredential(context.Context, int64, *agentv1 func doDisconnect(t *testing.T, store *fakeDisconnectStore, body string) int { t.Helper() - api := NewNodeAPI(store, nil, nil, "", "", nil) // nil hub:跳过 Push,只验证凭证删除 + api := NewNodeAPI(store, nil, nil, "", "", nil, nil) // nil hub:跳过 Push,只验证凭证删除 req := httptest.NewRequest("POST", "/v1/nodes/node-1/disconnect", strings.NewReader(body)) rctx := chi.NewRouteContext() rctx.URLParams.Add("id", "node-1")