From 9437334bf35cf5f1cfc7481900ed34b49c78589a Mon Sep 17 00:00:00 2001 From: T3ST3ST3R0N Date: Wed, 21 Oct 2026 21:11:04 +0330 Subject: [PATCH 1/5] fix(stats): count WireGuard traffic after a peer counter reset When the node re-creates a kernel peer (full SyncUsers and Restart use ReplacePeers, or a key is removed and added back) its counters restart at zero. The tracker rebased so that only the old pending delta was reported and the first sample of the new counter counted as already reported, so every byte between the re-create and the next stats tick was lost. A new counter that had already passed the old value was not seen as a reset at all. RemoveStats and the new MarkCounterReset, called after every ReplacePeers, now keep the pending delta as a negative base and restart the current counter at zero, so the next sample is counted in full on top of it. A counter that goes down without a mark is handled the same way. Stats sampling now holds syncMu, so a sample read before a re-create is never applied after its mark and never counts the old counter twice. TestStatsTracker_ReaddAfterDeletePreservesPendingDelta expected the lost bytes; it now expects the pending delta plus the new traffic. --- backend/wireguard/jobs.go | 6 + backend/wireguard/stats_counter_reset_test.go | 239 ++++++++++++++++++ backend/wireguard/user_full_sync.go | 4 + backend/wireguard/wireguard.go | 9 +- pkg/stats/tracker.go | 64 ++--- pkg/stats/tracker_test.go | 138 ++++++++-- 6 files changed, 415 insertions(+), 45 deletions(-) create mode 100644 backend/wireguard/stats_counter_reset_test.go diff --git a/backend/wireguard/jobs.go b/backend/wireguard/jobs.go index 303a7b0a..6378f18b 100644 --- a/backend/wireguard/jobs.go +++ b/backend/wireguard/jobs.go @@ -43,6 +43,12 @@ func (wg *WireGuard) updateConnectedPeers(ctx context.Context) { default: } + // A sync that re-creates peers marks their counter reset in the tracker. Holding syncMu + // from the device read to the tracker update keeps a sample read before such a reset + // from being applied after the mark, which would count the old counter twice. + wg.syncMu.Lock() + defer wg.syncMu.Unlock() + wg.mu.RLock() mgr := wg.manager cfg := wg.config diff --git a/backend/wireguard/stats_counter_reset_test.go b/backend/wireguard/stats_counter_reset_test.go new file mode 100644 index 00000000..3dbb60d9 --- /dev/null +++ b/backend/wireguard/stats_counter_reset_test.go @@ -0,0 +1,239 @@ +package wireguard + +import ( + "context" + "fmt" + "sync" + "testing" + "time" + + "github.com/pasarguard/node/common" + "github.com/pasarguard/node/pkg/stats" + "golang.zx2c4.com/wireguard/wgctrl/wgtypes" +) + +// counterKernel models the kernel peer table with its traffic counters: a peer created by a +// config starts at zero, ReplacePeers re-creates every peer and Remove drops one. +type counterKernel struct { + mu sync.Mutex + peers map[wgtypes.Key]*wgtypes.Peer +} + +func newCounterKernel() *counterKernel { + return &counterKernel{peers: make(map[wgtypes.Key]*wgtypes.Peer)} +} + +func (k *counterKernel) ConfigureDevice(_ string, cfg wgtypes.Config) error { + k.mu.Lock() + defer k.mu.Unlock() + + if cfg.ReplacePeers { + k.peers = make(map[wgtypes.Key]*wgtypes.Peer) + } + for _, peer := range cfg.Peers { + if peer.Remove { + delete(k.peers, peer.PublicKey) + continue + } + if _, exists := k.peers[peer.PublicKey]; !exists { + k.peers[peer.PublicKey] = &wgtypes.Peer{PublicKey: peer.PublicKey} + } + } + return nil +} + +func (k *counterKernel) Device(string) (*wgtypes.Device, error) { + k.mu.Lock() + defer k.mu.Unlock() + + device := &wgtypes.Device{} + for _, peer := range k.peers { + device.Peers = append(device.Peers, *peer) + } + return device, nil +} + +func (k *counterKernel) Close() error { return nil } + +// move counts n bytes in each direction on a live peer, as a connected client would. +func (k *counterKernel) move(t *testing.T, publicKey string, n int64) { + t.Helper() + key, err := wgtypes.ParseKey(publicKey) + if err != nil { + t.Fatalf("parse key: %v", err) + } + + k.mu.Lock() + defer k.mu.Unlock() + peer, exists := k.peers[key] + if !exists { + t.Fatalf("peer %s is not in the kernel", publicKey) + } + peer.ReceiveBytes += n + peer.TransmitBytes += n + peer.LastHandshakeTime = time.Now() +} + +func newCounterTestWG(t *testing.T, client wgClient) *WireGuard { + t.Helper() + privateKey, _, err := GenerateKeyPair() + if err != nil { + t.Fatalf("failed to generate key pair: %v", err) + } + cfg, err := NewConfig(fmt.Sprintf(`{ + "interface_name":"wg-test", + "listen_port":51820, + "address":["10.74.0.1/24"], + "private_key":"%s" + }`, privateKey)) + if err != nil { + t.Fatalf("failed to create config: %v", err) + } + + return &WireGuard{ + config: cfg, + peerStore: NewPeerStore(), + statsTracker: stats.New(), + manager: &Manager{iFaceName: "wg-test", client: client}, + state: lifecycleRunning, + } +} + +func counterTestUser(email, publicKey, peerIP string) *common.User { + user := &common.User{ + Email: email, + Proxies: &common.Proxy{Wireguard: &common.Wireguard{PublicKey: publicKey}}, + } + if peerIP != "" { + user.Inbounds = []string{"wg-test"} + user.Proxies.Wireguard.PeerIps = []string{peerIP} + } + return user +} + +// Every way the node re-creates a kernel peer (full sync ReplacePeers, Restart, remove then +// re-add before or after a stats pull) restarts its counters at zero. The traffic counted on +// both sides of each reset must reach the panel exactly once, including a new counter that +// already passed the old one when it is first sampled. +func TestPeerCounterResetsReportEachByteOnce(t *testing.T) { + ctx := context.Background() + kernel := newCounterKernel() + wg := newCounterTestWG(t, kernel) + + keys := make([]string, 3) + for i := range keys { + _, publicKey, err := GenerateKeyPair() + if err != nil { + t.Fatalf("failed to generate key pair: %v", err) + } + keys[i] = publicKey + } + keep := counterTestUser("1", keys[0], "10.74.0.2/32") + other := counterTestUser("2", keys[1], "10.74.0.3/32") + otherWithoutAccess := counterTestUser("2", keys[1], "") + third := counterTestUser("3", keys[2], "10.74.0.4/32") + + var moved, reportedRx, reportedTx int64 + move := func(publicKey string, n int64) { + kernel.move(t, publicKey, n) + moved += n + } + sample := func() { wg.updateConnectedPeers(ctx) } + pull := func() { + resp, err := wg.GetStats(ctx, &common.StatRequest{Type: common.StatType_UsersStat, Reset_: true}) + if err != nil { + t.Fatalf("GetStats failed: %v", err) + } + for _, stat := range resp.GetStats() { + switch stat.GetType() { + case "downlink": + reportedRx += stat.GetValue() + case "uplink": + reportedTx += stat.GetValue() + } + } + } + mustSync := func(err error) { + t.Helper() + if err != nil { + t.Fatalf("sync failed: %v", err) + } + } + + mustSync(wg.SyncUsers(ctx, []*common.User{keep, other})) + move(keys[0], 1000) + sample() + pull() + + // Full sync with a change: ReplacePeers re-creates every peer. The new counter passes + // the old one (1500) before the next sample. + move(keys[0], 500) + sample() + mustSync(wg.SyncUsers(ctx, []*common.User{keep, other, third})) + move(keys[0], 5000) + sample() + pull() + + // Restart also re-applies the whole peer set with ReplacePeers. + move(keys[0], 100) + sample() + mustSync(wg.Restart()) + move(keys[0], 50) + sample() + + // Removed and re-added before the panel pulled the pending delta. + move(keys[1], 300) + sample() + mustSync(wg.UpdateUsers(ctx, []*common.User{otherWithoutAccess})) + mustSync(wg.UpdateUsers(ctx, []*common.User{other})) + move(keys[1], 40) + sample() + pull() + + // Removed, everything reported, then re-added. + mustSync(wg.UpdateUsers(ctx, []*common.User{otherWithoutAccess})) + pull() + mustSync(wg.UpdateUsers(ctx, []*common.User{other})) + move(keys[1], 7) + sample() + pull() + + if reportedRx != moved || reportedTx != moved { + t.Fatalf("reported rx %d / tx %d bytes, peers moved %d", reportedRx, reportedTx, moved) + } +} + +// A sample read from the device before a ReplacePeers and applied after the reset mark would +// count the old counter a second time, so sampling must wait for a running peer sync. +func TestUpdateConnectedPeersWaitsForRunningSync(t *testing.T) { + wg := newCounterTestWG(t, newCounterKernel()) + + deviceRead := make(chan struct{}, 1) + wg.manager.client = &fakeWGClient{ + deviceFn: func(string) (*wgtypes.Device, error) { + deviceRead <- struct{}{} + return &wgtypes.Device{}, nil + }, + } + + wg.syncMu.Lock() + done := make(chan struct{}) + go func() { + wg.updateConnectedPeers(context.Background()) + close(done) + }() + + select { + case <-deviceRead: + wg.syncMu.Unlock() + t.Fatal("stats sampling read the device while a peer sync held syncMu") + case <-time.After(100 * time.Millisecond): + } + + wg.syncMu.Unlock() + select { + case <-done: + case <-time.After(5 * time.Second): + t.Fatal("stats sampling did not finish after the sync released syncMu") + } +} diff --git a/backend/wireguard/user_full_sync.go b/backend/wireguard/user_full_sync.go index e0f7d03b..9024477a 100644 --- a/backend/wireguard/user_full_sync.go +++ b/backend/wireguard/user_full_sync.go @@ -40,9 +40,11 @@ func (wg *WireGuard) syncUsersFull(users []*common.User) error { } var validTargetPeers []*PeerInfo + resetKeys := make([]string, 0, len(appliedKeys)) for key, peer := range diff.TargetPeers { if _, ok := appliedKeys[key]; ok { validTargetPeers = append(validTargetPeers, peer) + resetKeys = append(resetKeys, peer.PublicKey.String()) } } @@ -50,6 +52,8 @@ func (wg *WireGuard) syncUsersFull(users []*common.User) error { for _, key := range removedKeys { wg.statsTracker.RemoveStats(key) } + // ReplacePeers re-created every applied peer, so their counters restarted at zero. + wg.statsTracker.MarkCounterReset(resetKeys) return nil } diff --git a/backend/wireguard/wireguard.go b/backend/wireguard/wireguard.go index 241b4af0..71cef39f 100644 --- a/backend/wireguard/wireguard.go +++ b/backend/wireguard/wireguard.go @@ -35,7 +35,7 @@ var ( // WireGuard locking hierarchy — must always be acquired in this order: // -// wg.syncMu serialises all peer sync/remove operations +// wg.syncMu serialises all peer sync/remove operations and stats sampling // wg.mu guards lifecycle state (manager, state, config, version) // m.mu guards Manager internals (client, configure, nl) // @@ -293,6 +293,13 @@ func (wg *WireGuard) restartLocked() error { return fmt.Errorf("failed to reconfigure interface during restart: %w", err) } + // ReplacePeers re-created every peer, so their counters restarted at zero. + resetKeys := make([]string, 0, len(allPeers)) + for _, peer := range allPeers { + resetKeys = append(resetKeys, peer.PublicKey.String()) + } + wg.statsTracker.MarkCounterReset(resetKeys) + log.Println("wireguard interface reconfigured successfully without downtime") wg.emitInfoLogf("wireguard interface reconfigured successfully without downtime") return nil diff --git a/pkg/stats/tracker.go b/pkg/stats/tracker.go index c8e26bca..b2c134d8 100644 --- a/pkg/stats/tracker.go +++ b/pkg/stats/tracker.go @@ -15,8 +15,6 @@ type Entry struct { CurrentTx int64 BaseRx int64 BaseTx int64 - LastDeltaRx int64 - LastDeltaTx int64 LastActiveTime time.Time EndpointIP string IsDeleted bool @@ -71,14 +69,11 @@ func (st *Tracker) applySampleLocked(sample Sample, activeAt time.Time) { entry.Email = sample.Email entry.IsDeleted = false - // Restore pending delta baseline when a deleted entry is re-added. - readded := false - if entry.LastDeltaRx != 0 || entry.LastDeltaTx != 0 { - entry.BaseRx = sample.Rx - entry.LastDeltaRx - entry.BaseTx = sample.Tx - entry.LastDeltaTx - entry.LastDeltaRx = 0 - entry.LastDeltaTx = 0 - readded = true + // Kernel counters only go down when the peer was re-created. The backend marks the + // resets it causes itself; this catches any other one (it cannot see a new counter + // that already passed the old value). + if sample.Rx < entry.CurrentRx || sample.Tx < entry.CurrentTx { + rebaseForCounterReset(entry) } // Skip update if values haven't changed @@ -87,20 +82,6 @@ func (st *Tracker) applySampleLocked(sample Sample, activeAt time.Time) { return } - // Detect kernel counter rollback/reset and preserve pending deltas - if !readded && (sample.Rx < entry.CurrentRx || sample.Tx < entry.CurrentTx) { - pendingRx := entry.CurrentRx - entry.BaseRx - pendingTx := entry.CurrentTx - entry.BaseTx - if pendingRx < 0 { - pendingRx = 0 - } - if pendingTx < 0 { - pendingTx = 0 - } - entry.BaseRx = sample.Rx - pendingRx - entry.BaseTx = sample.Tx - pendingTx - } - // Any traffic growth implies peer is active. if sample.Rx > entry.CurrentRx || sample.Tx > entry.CurrentTx { entry.LastActiveTime = activeAt @@ -201,8 +182,6 @@ func (st *Tracker) GetStatsEntries(keys []string) map[string]*Entry { CurrentTx: entry.CurrentTx, BaseRx: entry.BaseRx, BaseTx: entry.BaseTx, - LastDeltaRx: entry.LastDeltaRx, - LastDeltaTx: entry.LastDeltaTx, LastActiveTime: entry.LastActiveTime, EndpointIP: entry.EndpointIP, IsDeleted: entry.IsDeleted, @@ -257,6 +236,33 @@ func (st *Tracker) EndpointActivity(keys []string) map[string]int64 { return result } +// rebaseForCounterReset keeps the unreported delta across a kernel counter reset: the +// counter restarts at zero and the baseline becomes -pending, so the next sample is +// reported in full on top of what was still pending. Calling it again before a new +// sample changes nothing. +func rebaseForCounterReset(entry *Entry) { + entry.BaseRx = -max(entry.CurrentRx-entry.BaseRx, 0) + entry.BaseTx = -max(entry.CurrentTx-entry.BaseTx, 0) + entry.CurrentRx = 0 + entry.CurrentTx = 0 +} + +// MarkCounterReset records that the kernel re-created these peers, so their counters +// restarted at zero. Pending traffic is kept and the next sample counts in full, even +// when it is already above the old counter. Only mark peers that really were re-created, +// and never apply a sample read before the reset after this call: either would report +// the old counter a second time. +func (st *Tracker) MarkCounterReset(keys []string) { + st.mu.Lock() + defer st.mu.Unlock() + + for _, key := range keys { + if entry, exists := st.stats[key]; exists { + rebaseForCounterReset(entry) + } + } +} + // RemoveStats marks a peer as deleted but keeps counters until a reset reports them. func (st *Tracker) RemoveStats(publicKey string) { st.mu.Lock() @@ -267,9 +273,9 @@ func (st *Tracker) RemoveStats(publicKey string) { return } - // Persist pending deltas so re-add before stats pull won't lose data. - entry.LastDeltaRx = entry.CurrentRx - entry.BaseRx - entry.LastDeltaTx = entry.CurrentTx - entry.BaseTx + // The kernel peer is gone with its counters: a re-add of this key starts a new + // counter at zero, while the pending delta stays reportable. + rebaseForCounterReset(entry) // Mark deleted and clear active time so online checks return offline immediately. entry.IsDeleted = true diff --git a/pkg/stats/tracker_test.go b/pkg/stats/tracker_test.go index 76148364..1dc7697c 100644 --- a/pkg/stats/tracker_test.go +++ b/pkg/stats/tracker_test.go @@ -175,24 +175,19 @@ func TestStatsTracker_ZeroDeltaSkipping(t *testing.T) { } } -func TestStatsTracker_RemoveStatsStoresPendingLastDelta(t *testing.T) { +func TestStatsTracker_RemoveStatsKeepsPendingDelta(t *testing.T) { tracker := New() updateSample(tracker, "key1", "user1@example.com", 1000, 2000, "1.2.3.4") tracker.RemoveStats("key1") - entries := tracker.GetStatsEntries([]string{"key1"}) - entry, ok := entries["key1"] - if !ok { - t.Fatal("expected entry to exist") - } - if entry.LastDeltaRx != 1000 { - t.Fatalf("expected LastDeltaRx=1000, got %d", entry.LastDeltaRx) - } - if entry.LastDeltaTx != 2000 { - t.Fatalf("expected LastDeltaTx=2000, got %d", entry.LastDeltaTx) + if rx, tx := pullTotals(tracker); rx != 1000 || tx != 2000 { + t.Fatalf("expected pending 1000/2000 after RemoveStats, got %d/%d", rx, tx) } } +// A re-added key is a new kernel peer whose counters restarted at zero, so its first sample +// (100/200) is new traffic and is reported on top of the delta still pending from before the +// removal (1000/2000). This test used to expect only 1000/2000, which dropped those bytes. func TestStatsTracker_ReaddAfterDeletePreservesPendingDelta(t *testing.T) { tracker := New() ctx := context.Background() @@ -205,11 +200,11 @@ func TestStatsTracker_ReaddAfterDeletePreservesPendingDelta(t *testing.T) { resp := tracker.GetStats(ctx, []string{"key1"}, false) for _, s := range resp.Stats { - if s.Type == "downlink" && s.Value != 1000 { - t.Errorf("Expected preserved rx=1000, got %d", s.Value) + if s.Type == "downlink" && s.Value != 1100 { + t.Errorf("Expected pending plus new rx=1100, got %d", s.Value) } - if s.Type == "uplink" && s.Value != 2000 { - t.Errorf("Expected preserved tx=2000, got %d", s.Value) + if s.Type == "uplink" && s.Value != 2200 { + t.Errorf("Expected pending plus new tx=2200, got %d", s.Value) } } @@ -221,6 +216,119 @@ func TestStatsTracker_ReaddAfterDeletePreservesPendingDelta(t *testing.T) { } } +// pullTotals reads and resets every entry like the panel's usage job, returning rx/tx totals. +func pullTotals(tracker *Tracker) (rx, tx int64) { + for _, s := range tracker.GetUsersStats(context.Background(), true).Stats { + switch s.Type { + case "downlink": + rx += s.Value + case "uplink": + tx += s.Value + } + } + return rx, tx +} + +func TestStatsTracker_ReaddAfterReportedDeleteCountsNewTraffic(t *testing.T) { + tracker := New() + + updateSample(tracker, "key1", "user1@example.com", 1000, 1000, "1.2.3.4") + pullTotals(tracker) + tracker.RemoveStats("key1") // nothing pending, entry kept until cleanup + + updateSample(tracker, "key1", "user1@example.com", 300, 400, "1.2.3.4") + if rx, tx := pullTotals(tracker); rx != 300 || tx != 400 { + t.Fatalf("expected the re-added peer's 300/400 to be reported, got %d/%d", rx, tx) + } +} + +// A counter that goes down without a mark (peer re-created outside the node's own syncs) +// is still a reset: keep what was pending and count the new counter in full. +func TestStatsTracker_CounterRollbackKeepsPendingAndNewTraffic(t *testing.T) { + tracker := New() + + updateSample(tracker, "key1", "user1@example.com", 1000, 1000, "1.2.3.4") + pullTotals(tracker) + updateSample(tracker, "key1", "user1@example.com", 1500, 1600, "1.2.3.4") + updateSample(tracker, "key1", "user1@example.com", 200, 300, "1.2.3.4") + + if rx, tx := pullTotals(tracker); rx != 700 || tx != 900 { + t.Fatalf("expected pending 500/600 plus new 200/300, got %d/%d", rx, tx) + } +} + +// After a reset the new counter can pass the old one before the next sample, which looks like +// normal growth. Only an explicit mark from the backend can tell the two apart. +func TestStatsTracker_MarkCounterResetCountsNewCounterPastOldValue(t *testing.T) { + tracker := New() + + updateSample(tracker, "key1", "user1@example.com", 1000, 1000, "1.2.3.4") + pullTotals(tracker) + updateSample(tracker, "key1", "user1@example.com", 1500, 1500, "1.2.3.4") + + tracker.MarkCounterReset([]string{"key1", "unknown"}) + if entries := tracker.GetStatsEntries([]string{"unknown"}); len(entries) != 0 { + t.Fatal("MarkCounterReset must not create entries for unknown keys") + } + + updateSample(tracker, "key1", "user1@example.com", 5000, 5000, "1.2.3.4") + if rx, tx := pullTotals(tracker); rx != 5500 || tx != 5500 { + t.Fatalf("expected pending 500 plus new 5000, got %d/%d", rx, tx) + } +} + +// Every byte a peer moves is reported exactly once across pulls, marked resets, unmarked +// rollbacks, repeated marks and remove/re-add, whatever order they come in. +func TestStatsTracker_CounterResetsReportEachByteOnce(t *testing.T) { + tracker := New() + var counter, moved, reported int64 + + sample := func(add int64) { + counter += add + moved += add + updateSample(tracker, "key1", "user1@example.com", counter, counter, "1.2.3.4") + } + pull := func() { + rx, tx := pullTotals(tracker) + if rx != tx { + t.Fatalf("rx %d and tx %d diverged", rx, tx) + } + reported += rx + } + reset := func() { counter = 0 } + + sample(1000) + pull() + sample(500) + reset() + tracker.MarkCounterReset([]string{"key1"}) + pull() + sample(200) + sample(0) + pull() + pull() + sample(250) + reset() + tracker.MarkCounterReset([]string{"key1"}) + tracker.MarkCounterReset([]string{"key1"}) + sample(4000) + sample(10) + reset() + sample(30) // unmarked rollback + tracker.RemoveStats("key1") + reset() + sample(70) // re-added before the pull + pull() + tracker.RemoveStats("key1") + reset() + sample(5) // re-added after everything was reported + pull() + + if reported != moved { + t.Fatalf("reported %d bytes, peer moved %d", reported, moved) + } +} + func TestStatsTracker_AnyActiveSince(t *testing.T) { tracker := New() From db3ce45a991f923007fae180d442eeae479f3daa Mon Sep 17 00:00:00 2001 From: T3ST3ST3R0N Date: Wed, 21 Oct 2026 21:11:58 +0330 Subject: [PATCH 2/5] fix(wireguard): key desired peers by the canonical public key collectDesiredPeers keyed desired peers by the public key string the panel sent, while the peer store and the kernel use the canonical wgtypes.Key.String() form. ParseKey also accepts other base64 spellings of the same key (the unused low bits of the last character), so for such a key every later partial sync sent an add and a remove for the same kernel peer and cut the user off, and every full sync saw a change and re-sent a ReplacePeers snapshot that reset all peers. Desired peers and the duplicate key check now use the canonical form. Two users holding two spellings of one key are rejected like any other duplicate key. Stats are unchanged: they are reported under the user's email, and their link already used the canonical key read from the kernel. --- backend/wireguard/user_common.go | 11 ++- backend/wireguard/user_sync_test.go | 133 ++++++++++++++++++++++++++++ 2 files changed, 140 insertions(+), 4 deletions(-) diff --git a/backend/wireguard/user_common.go b/backend/wireguard/user_common.go index e2f8ee5c..807a6aa8 100644 --- a/backend/wireguard/user_common.go +++ b/backend/wireguard/user_common.go @@ -264,13 +264,16 @@ func (wg *WireGuard) collectDesiredPeers(users []*common.User) (map[string]*Desi continue } - if existing, exists := desiredPeers[publicKey]; exists && existing.Email != email { - return nil, fmt.Errorf("wireguard public key %s is assigned to multiple users: %s and %s", publicKey, existing.Email, email) + // Key by the canonical form, like the peer store and the kernel: ParseKey also accepts + // other base64 spellings of the same key, which would look like a different peer. + canonicalKey := parsedKey.String() + if existing, exists := desiredPeers[canonicalKey]; exists && existing.Email != email { + return nil, fmt.Errorf("wireguard public key %s is assigned to multiple users: %s and %s", canonicalKey, existing.Email, email) } - desiredPeers[publicKey] = &DesiredPeer{ + desiredPeers[canonicalKey] = &DesiredPeer{ Email: email, - PublicKey: publicKey, + PublicKey: canonicalKey, ParsedKey: parsedKey, AllowedIPNets: allowedIPNets, } diff --git a/backend/wireguard/user_sync_test.go b/backend/wireguard/user_sync_test.go index 4c8d9133..4ff86360 100644 --- a/backend/wireguard/user_sync_test.go +++ b/backend/wireguard/user_sync_test.go @@ -1022,3 +1022,136 @@ func TestUpdateUsersAllowsKeyHandoverFromUserWhoLostAccess(t *testing.T) { t.Fatal("expected the former owner to have no peer") } } + +// nonCanonicalKey returns another base64 spelling of the same 32-byte key. ParseKey accepts +// it because Go's StdEncoding ignores the two unused low bits of the last data character. +func nonCanonicalKey(t *testing.T, canonical string) string { + t.Helper() + const alphabet = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/" + spelled := []byte(canonical) + spelled[42] = alphabet[strings.IndexByte(alphabet, spelled[42])|1] + + key, err := wgtypes.ParseKey(string(spelled)) + if err != nil || key.String() != canonical || string(spelled) == canonical { + t.Fatalf("could not build a non-canonical spelling of %s: %v", canonical, err) + } + return string(spelled) +} + +func newKeySpellingTestWG(t *testing.T, applied *[]wgtypes.Config) *WireGuard { + t.Helper() + cfg, err := NewConfig(`{ + "interface_name":"wg-test", + "listen_port":51820, + "address":["10.75.0.1/24"] + }`) + if err != nil { + t.Fatalf("failed to create config: %v", err) + } + + return &WireGuard{ + config: cfg, + peerStore: NewPeerStore(), + statsTracker: stats.New(), + manager: &Manager{ + iFaceName: "wg-test", + client: &fakeWGClient{ + configureDeviceFn: func(_ string, cfg wgtypes.Config) error { + *applied = append(*applied, cfg) + return nil + }, + }, + }, + state: lifecycleRunning, + } +} + +func keySpellingTestUser(email, publicKey, peerIP string) *common.User { + return &common.User{ + Email: email, + Inbounds: []string{"wg-test"}, + Proxies: &common.Proxy{ + Wireguard: &common.Wireguard{PublicKey: publicKey, PeerIps: []string{peerIP}}, + }, + } +} + +// The store keys peers by the canonical key, so a desired peer sent with another valid spelling +// of the same key used to look like a new key: a later partial sync sent [add, remove] for the +// same kernel peer and the user was cut off. +func TestUpdateUsersKeepsPeerWithNonCanonicalKey(t *testing.T) { + _, canonical, err := GenerateKeyPair() + if err != nil { + t.Fatalf("failed to generate key: %v", err) + } + var applied []wgtypes.Config + wg := newKeySpellingTestWG(t, &applied) + user := keySpellingTestUser("7", nonCanonicalKey(t, canonical), "10.75.0.7/32") + + if err := wg.SyncUser(context.Background(), user); err != nil { + t.Fatalf("SyncUser failed: %v", err) + } + applied = nil + + if err := wg.UpdateUsers(context.Background(), []*common.User{user}); err != nil { + t.Fatalf("UpdateUsers failed: %v", err) + } + for _, cfg := range applied { + for _, peer := range cfg.Peers { + if peer.Remove { + t.Fatalf("unchanged user's peer %s was removed from the kernel", peer.PublicKey) + } + } + } + if len(applied) != 0 { + t.Fatalf("expected no kernel change for an unchanged user, got %d ConfigureDevice calls", len(applied)) + } + if peer := wg.peerStore.GetByEmail("7"); peer == nil || peer.PublicKey.String() != canonical { + t.Fatalf("expected the store to keep the peer under %s, got %+v", canonical, peer) + } +} + +// An identical full sync must stay a no-op: a ReplacePeers would reset every peer's counters +// and sessions. +func TestSyncUsersNonCanonicalKeyNoEffectiveChangeSkipsReplaceApply(t *testing.T) { + _, canonical, err := GenerateKeyPair() + if err != nil { + t.Fatalf("failed to generate key: %v", err) + } + var applied []wgtypes.Config + wg := newKeySpellingTestWG(t, &applied) + users := []*common.User{keySpellingTestUser("7", nonCanonicalKey(t, canonical), "10.75.0.7/32")} + + if err := wg.SyncUsers(context.Background(), users); err != nil { + t.Fatalf("SyncUsers failed: %v", err) + } + applied = nil + + if err := wg.SyncUsers(context.Background(), users); err != nil { + t.Fatalf("SyncUsers failed: %v", err) + } + if len(applied) != 0 { + t.Fatalf("expected no ConfigureDevice calls for a no-op full sync, got %d", len(applied)) + } +} + +func TestSyncUsersRejectsOneKeySpelledTwoWaysAcrossUsers(t *testing.T) { + _, canonical, err := GenerateKeyPair() + if err != nil { + t.Fatalf("failed to generate key: %v", err) + } + var applied []wgtypes.Config + wg := newKeySpellingTestWG(t, &applied) + users := []*common.User{ + keySpellingTestUser("7", canonical, "10.75.0.7/32"), + keySpellingTestUser("8", nonCanonicalKey(t, canonical), "10.75.0.8/32"), + } + + err = wg.SyncUsers(context.Background(), users) + if err == nil || !strings.Contains(err.Error(), "assigned to multiple users") { + t.Fatalf("expected a duplicate key error, got %v", err) + } + if len(applied) != 0 { + t.Fatalf("expected no kernel change for a rejected sync, got %d ConfigureDevice calls", len(applied)) + } +} From ffbd53f8927a30b5eb91ea25ba08afc552a3d2e1 Mon Sep 17 00:00:00 2001 From: T3ST3ST3R0N Date: Wed, 21 Oct 2026 21:13:16 +0330 Subject: [PATCH 3/5] fix(config): accept IPv6 NODE_HOST and log an invalid value as an error NODE_HOST was checked with a dotted-quad IPv4 regex. "::", any IPv6 address, or a value with surrounding spaces silently became 127.0.0.1 with one plain log line, so the panel could not reach the node and nothing clearly said why. main also built the listen address with "%s:%d", which is not valid for IPv6. NODE_HOST is now trimmed and parsed with net.ParseIP, and the listen address is built with net.JoinHostPort. "localhost" still means 127.0.0.1. Any other value keeps the old fallback to 127.0.0.1, so a node that started before an upgrade still starts after it, but the bad value is now logged at error level. --- cmd/node/main.go | 5 +-- config/config.go | 29 ++++++++++------ config/config_test.go | 80 +++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 101 insertions(+), 13 deletions(-) diff --git a/cmd/node/main.go b/cmd/node/main.go index 196465e6..7ddd2016 100644 --- a/cmd/node/main.go +++ b/cmd/node/main.go @@ -2,10 +2,11 @@ package main import ( "context" - "fmt" "log" + "net" "os" "os/signal" + "strconv" "syscall" "time" @@ -22,7 +23,7 @@ func main() { log.Fatal(err) } - addr := fmt.Sprintf("%s:%d", cfg.NodeHost, cfg.ServicePort) + addr := net.JoinHostPort(cfg.NodeHost, strconv.Itoa(cfg.ServicePort)) tlsConfig, err := tlsutil.LoadTLSCredentials(cfg.SslCertFile, cfg.SslKeyFile) if err != nil { diff --git a/config/config.go b/config/config.go index 0ec5b6bc..47cf2f2a 100644 --- a/config/config.go +++ b/config/config.go @@ -4,9 +4,10 @@ import ( "errors" "fmt" "log" + "net" "os" - "regexp" "strconv" + "strings" "github.com/google/uuid" "github.com/joho/godotenv" @@ -86,20 +87,26 @@ func Load() (*Config, error) { apiKeyErr = fmt.Errorf("invalid API_KEY: %w", apiKeyErr) } - nodeHostStr := GetEnv("NODE_HOST", "0.0.0.0") - ipPattern := `^(?:(?:25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)\.){3}(?:25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)$` - re := regexp.MustCompile(ipPattern) - - if re.MatchString(nodeHostStr) { - cfg.NodeHost = nodeHostStr - } else { - log.Println(nodeHostStr, " is not a valid IP address.\n NODE_HOST will be set to 127.0.0.1") - cfg.NodeHost = "127.0.0.1" - } + cfg.NodeHost = parseNodeHost(GetEnv("NODE_HOST", "0.0.0.0")) return cfg, apiKeyErr } +// parseNodeHost returns the IPv4 or IPv6 address to listen on; "localhost" means 127.0.0.1. +// Any other value keeps the old fallback to 127.0.0.1, so a node that started before still +// starts, but it is logged as an error: the panel usually cannot reach a loopback-only node. +func parseNodeHost(value string) string { + host := strings.TrimSpace(value) + if strings.EqualFold(host, "localhost") { + return "127.0.0.1" + } + if net.ParseIP(host) == nil { + log.Printf("[Error] NODE_HOST %q is not an IP address (for example 0.0.0.0 or ::). Falling back to 127.0.0.1: only clients on this host can reach the node.", value) + return "127.0.0.1" + } + return host +} + // NewTestConfig creates a config for testing func NewTestConfig(generatedConfigPath string, key uuid.UUID) *Config { cfg, _ := Load() diff --git a/config/config_test.go b/config/config_test.go index adb12bb1..2a553450 100644 --- a/config/config_test.go +++ b/config/config_test.go @@ -1,6 +1,9 @@ package config import ( + "bytes" + "fmt" + "log" "os" "strings" "testing" @@ -60,3 +63,80 @@ func TestLoadAcceptsValidAPIKey(t *testing.T) { t.Fatalf("expected ApiKey %s, got %s", key, cfg.ApiKey) } } + +func TestLoadAcceptsNodeHostIPsAndLocalhost(t *testing.T) { + cases := map[string]string{ + "0.0.0.0": "0.0.0.0", + "10.0.0.5": "10.0.0.5", + "::": "::", + "::1": "::1", + "2001:db8::1": "2001:db8::1", + " 0.0.0.0 ": "0.0.0.0", + "localhost": "127.0.0.1", + " LocalHost ": "127.0.0.1", + } + for value, want := range cases { + t.Run(value, func(t *testing.T) { + t.Setenv("API_KEY", "3fa85f64-5717-4562-b3fc-2c963f66afa6") + t.Setenv("NODE_HOST", value) + logs := captureLog(t) + + cfg, err := Load() + if err != nil { + t.Fatalf("Load failed for NODE_HOST=%q: %v", value, err) + } + if cfg.NodeHost != want { + t.Fatalf("expected NodeHost %q for NODE_HOST=%q, got %q", want, value, cfg.NodeHost) + } + if strings.Contains(logs.String(), "NODE_HOST") { + t.Fatalf("did not expect a NODE_HOST log for %q, got %q", value, logs.String()) + } + }) + } +} + +func TestLoadDefaultsNodeHostToAllIPv4(t *testing.T) { + t.Setenv("API_KEY", "3fa85f64-5717-4562-b3fc-2c963f66afa6") + unsetEnv(t, "NODE_HOST") + + cfg, err := Load() + if err != nil { + t.Fatalf("Load failed: %v", err) + } + if cfg.NodeHost != "0.0.0.0" { + t.Fatalf("expected default NodeHost 0.0.0.0, got %q", cfg.NodeHost) + } +} + +// captureLog collects what the standard logger writes during the test. +func captureLog(t *testing.T) *bytes.Buffer { + t.Helper() + var buf bytes.Buffer + previous := log.Writer() + log.SetOutput(&buf) + t.Cleanup(func() { log.SetOutput(previous) }) + return &buf +} + +// An invalid NODE_HOST keeps the old fallback to 127.0.0.1, so an upgrade never stops a node +// that used to start, but it is now logged as an error with the bad value. +func TestLoadFallsBackToLoopbackOnInvalidNodeHost(t *testing.T) { + for _, value := range []string{"", "node.example.com", "256.1.1.1", "[::1]", "0.0.0.0:62050"} { + t.Run(value, func(t *testing.T) { + t.Setenv("API_KEY", "3fa85f64-5717-4562-b3fc-2c963f66afa6") + t.Setenv("NODE_HOST", value) + logs := captureLog(t) + + cfg, err := Load() + if err != nil { + t.Fatalf("an invalid NODE_HOST must not fail Load, got %v", err) + } + if cfg.NodeHost != "127.0.0.1" { + t.Fatalf("expected the loopback fallback for NODE_HOST=%q, got %q", value, cfg.NodeHost) + } + if want := fmt.Sprintf("[Error] NODE_HOST %q", value); !strings.Contains(logs.String(), want) { + t.Fatalf("expected an error log containing %s, got %q", want, logs.String()) + } + }) + } +} From 84f8c4474d300f3b3a5f825436c1c4cd9acad8ff Mon Sep 17 00:00:00 2001 From: T3ST3ST3R0N Date: Wed, 21 Oct 2026 21:24:32 +0330 Subject: [PATCH 4/5] fix(wireguard): sample peer stats before replacing peers A full SyncUsers with a change and Restart apply the peer set with ReplacePeers, which destroys every kernel peer counter. Traffic moved since the last stats tick (10 s by default, longer for a peer whose last handshake is past the 45 s cutoff, since WireGuard rekeys only every ~2 minutes) was never reported. This closes most of known issue W1; the sessions are still reset. Both paths now read the peer counters into the tracker right before ReplacePeers, inside the same syncMu section, so the stats loop cannot interleave. That pre-reset sample includes every peer that completed a handshake, not only recently active ones. The sample is applied before the counter reset mark, so no byte is counted twice. It costs one device read per full sync that changes something. --- backend/wireguard/jobs.go | 9 ++++++- backend/wireguard/stats_counter_reset_test.go | 27 ++++++++++++++++++- backend/wireguard/user_full_sync.go | 4 +++ backend/wireguard/wireguard.go | 3 +++ 4 files changed, 41 insertions(+), 2 deletions(-) diff --git a/backend/wireguard/jobs.go b/backend/wireguard/jobs.go index 6378f18b..9fc0543a 100644 --- a/backend/wireguard/jobs.go +++ b/backend/wireguard/jobs.go @@ -49,6 +49,13 @@ func (wg *WireGuard) updateConnectedPeers(ctx context.Context) { wg.syncMu.Lock() defer wg.syncMu.Unlock() + wg.samplePeersLocked(ctx, false) +} + +// samplePeersLocked reads the peer counters into the tracker. The caller holds syncMu. +// Before a ReplacePeers it reads every peer that has completed a handshake, not only the +// recently active ones, because their counters are about to be destroyed. +func (wg *WireGuard) samplePeersLocked(ctx context.Context, beforeReset bool) { wg.mu.RLock() mgr := wg.manager cfg := wg.config @@ -79,7 +86,7 @@ func (wg *WireGuard) updateConnectedPeers(ctx context.Context) { if peer.LastHandshakeTime.IsZero() { continue // never connected } - if peer.LastHandshakeTime.Before(activeHandshakeCutoff) { + if !beforeReset && peer.LastHandshakeTime.Before(activeHandshakeCutoff) { continue // stale/offline peer } diff --git a/backend/wireguard/stats_counter_reset_test.go b/backend/wireguard/stats_counter_reset_test.go index 3dbb60d9..1393f01b 100644 --- a/backend/wireguard/stats_counter_reset_test.go +++ b/backend/wireguard/stats_counter_reset_test.go @@ -74,6 +74,23 @@ func (k *counterKernel) move(t *testing.T, publicKey string, n int64) { peer.LastHandshakeTime = time.Now() } +// ageHandshake moves a peer's last handshake into the past, as between WireGuard rekeys. +func (k *counterKernel) ageHandshake(t *testing.T, publicKey string, age time.Duration) { + t.Helper() + key, err := wgtypes.ParseKey(publicKey) + if err != nil { + t.Fatalf("parse key: %v", err) + } + + k.mu.Lock() + defer k.mu.Unlock() + peer, exists := k.peers[key] + if !exists { + t.Fatalf("peer %s is not in the kernel", publicKey) + } + peer.LastHandshakeTime = time.Now().Add(-age) +} + func newCounterTestWG(t *testing.T, client wgClient) *WireGuard { t.Helper() privateKey, _, err := GenerateKeyPair() @@ -114,7 +131,8 @@ func counterTestUser(email, publicKey, peerIP string) *common.User { // Every way the node re-creates a kernel peer (full sync ReplacePeers, Restart, remove then // re-add before or after a stats pull) restarts its counters at zero. The traffic counted on // both sides of each reset must reach the panel exactly once, including a new counter that -// already passed the old one when it is first sampled. +// already passed the old one when it is first sampled, and traffic moved since the last stats +// tick right before a ReplacePeers. func TestPeerCounterResetsReportEachByteOnce(t *testing.T) { ctx := context.Background() kernel := newCounterKernel() @@ -169,6 +187,12 @@ func TestPeerCounterResetsReportEachByteOnce(t *testing.T) { // the old one (1500) before the next sample. move(keys[0], 500) sample() + // Traffic since the last tick, on a recently active peer and on one whose last handshake is + // past the 45 s online cutoff (WireGuard rekeys only every ~2 minutes). The sync samples both + // right before ReplacePeers destroys their counters. + move(keys[0], 25) + move(keys[1], 60) + kernel.ageHandshake(t, keys[1], time.Minute) mustSync(wg.SyncUsers(ctx, []*common.User{keep, other, third})) move(keys[0], 5000) sample() @@ -177,6 +201,7 @@ func TestPeerCounterResetsReportEachByteOnce(t *testing.T) { // Restart also re-applies the whole peer set with ReplacePeers. move(keys[0], 100) sample() + move(keys[0], 15) // since the last tick mustSync(wg.Restart()) move(keys[0], 50) sample() diff --git a/backend/wireguard/user_full_sync.go b/backend/wireguard/user_full_sync.go index 9024477a..6602d881 100644 --- a/backend/wireguard/user_full_sync.go +++ b/backend/wireguard/user_full_sync.go @@ -1,6 +1,7 @@ package wireguard import ( + "context" "fmt" "github.com/pasarguard/node/common" @@ -35,6 +36,9 @@ func (wg *WireGuard) syncUsersFull(users []*common.User) error { manager := wg.manager wg.mu.RUnlock() + // ReplacePeers destroys every counter: count the traffic since the last stats tick first. + wg.samplePeersLocked(context.Background(), true) + if err := manager.ApplyPeersReplaceAll(peerConfigs); err != nil { return fmt.Errorf("failed to apply full peer snapshot: %w", err) } diff --git a/backend/wireguard/wireguard.go b/backend/wireguard/wireguard.go index 71cef39f..b10b6ad0 100644 --- a/backend/wireguard/wireguard.go +++ b/backend/wireguard/wireguard.go @@ -279,6 +279,9 @@ func (wg *WireGuard) restartLocked() error { manager := wg.manager wg.mu.Unlock() + // ReplacePeers destroys every counter: count the traffic since the last stats tick first. + wg.samplePeersLocked(context.Background(), true) + log.Println("dynamically reconfiguring wireguard interface") wg.emitInfoLogf("dynamically reconfiguring wireguard interface") From abe8932c7cbdc7b317f639330796ef945ee28a08 Mon Sep 17 00:00:00 2001 From: T3ST3ST3R0N Date: Thu, 22 Oct 2026 10:49:13 +0330 Subject: [PATCH 5/5] fix(config): accept a scoped IPv6 NODE_HOST net.ParseIP does not accept a zone, so a link-local NODE_HOST such as fe80::1%eth0 fell back to 127.0.0.1 with an error log, although net.Listen can bind it. NODE_HOST is now parsed with netip.ParseAddr, which accepts the zone and otherwise takes the same values. An IPv4 address with a zone or an empty zone is still invalid and keeps the loopback fallback. --- config/config.go | 5 +++-- config/config_test.go | 19 ++++++++++--------- 2 files changed, 13 insertions(+), 11 deletions(-) diff --git a/config/config.go b/config/config.go index 47cf2f2a..b69f0dc3 100644 --- a/config/config.go +++ b/config/config.go @@ -4,7 +4,7 @@ import ( "errors" "fmt" "log" - "net" + "net/netip" "os" "strconv" "strings" @@ -93,6 +93,7 @@ func Load() (*Config, error) { } // parseNodeHost returns the IPv4 or IPv6 address to listen on; "localhost" means 127.0.0.1. +// A scoped IPv6 address (fe80::1%eth0) is accepted too, since net.Listen takes the zone. // Any other value keeps the old fallback to 127.0.0.1, so a node that started before still // starts, but it is logged as an error: the panel usually cannot reach a loopback-only node. func parseNodeHost(value string) string { @@ -100,7 +101,7 @@ func parseNodeHost(value string) string { if strings.EqualFold(host, "localhost") { return "127.0.0.1" } - if net.ParseIP(host) == nil { + if _, err := netip.ParseAddr(host); err != nil { log.Printf("[Error] NODE_HOST %q is not an IP address (for example 0.0.0.0 or ::). Falling back to 127.0.0.1: only clients on this host can reach the node.", value) return "127.0.0.1" } diff --git a/config/config_test.go b/config/config_test.go index 2a553450..39717e84 100644 --- a/config/config_test.go +++ b/config/config_test.go @@ -66,14 +66,15 @@ func TestLoadAcceptsValidAPIKey(t *testing.T) { func TestLoadAcceptsNodeHostIPsAndLocalhost(t *testing.T) { cases := map[string]string{ - "0.0.0.0": "0.0.0.0", - "10.0.0.5": "10.0.0.5", - "::": "::", - "::1": "::1", - "2001:db8::1": "2001:db8::1", - " 0.0.0.0 ": "0.0.0.0", - "localhost": "127.0.0.1", - " LocalHost ": "127.0.0.1", + "0.0.0.0": "0.0.0.0", + "10.0.0.5": "10.0.0.5", + "::": "::", + "::1": "::1", + "2001:db8::1": "2001:db8::1", + "fe80::1%eth0": "fe80::1%eth0", + " 0.0.0.0 ": "0.0.0.0", + "localhost": "127.0.0.1", + " LocalHost ": "127.0.0.1", } for value, want := range cases { t.Run(value, func(t *testing.T) { @@ -121,7 +122,7 @@ func captureLog(t *testing.T) *bytes.Buffer { // An invalid NODE_HOST keeps the old fallback to 127.0.0.1, so an upgrade never stops a node // that used to start, but it is now logged as an error with the bad value. func TestLoadFallsBackToLoopbackOnInvalidNodeHost(t *testing.T) { - for _, value := range []string{"", "node.example.com", "256.1.1.1", "[::1]", "0.0.0.0:62050"} { + for _, value := range []string{"", "node.example.com", "256.1.1.1", "[::1]", "0.0.0.0:62050", "10.0.0.5%eth0", "fe80::1%"} { t.Run(value, func(t *testing.T) { t.Setenv("API_KEY", "3fa85f64-5717-4562-b3fc-2c963f66afa6") t.Setenv("NODE_HOST", value)