diff --git a/backend/wireguard/jobs.go b/backend/wireguard/jobs.go index 303a7b0a..9fc0543a 100644 --- a/backend/wireguard/jobs.go +++ b/backend/wireguard/jobs.go @@ -43,6 +43,19 @@ 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.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 @@ -73,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 new file mode 100644 index 00000000..1393f01b --- /dev/null +++ b/backend/wireguard/stats_counter_reset_test.go @@ -0,0 +1,264 @@ +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() +} + +// 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() + 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, and traffic moved since the last stats +// tick right before a ReplacePeers. +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() + // 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() + pull() + + // 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() + + // 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_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_full_sync.go b/backend/wireguard/user_full_sync.go index e0f7d03b..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,14 +36,19 @@ 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) } 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 +56,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/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)) + } +} diff --git a/backend/wireguard/wireguard.go b/backend/wireguard/wireguard.go index 241b4af0..b10b6ad0 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) // @@ -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") @@ -293,6 +296,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/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..b69f0dc3 100644 --- a/config/config.go +++ b/config/config.go @@ -4,9 +4,10 @@ import ( "errors" "fmt" "log" + "net/netip" "os" - "regexp" "strconv" + "strings" "github.com/google/uuid" "github.com/joho/godotenv" @@ -86,20 +87,27 @@ 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. +// 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 { + host := strings.TrimSpace(value) + if strings.EqualFold(host, "localhost") { + return "127.0.0.1" + } + 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" + } + 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..39717e84 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,81 @@ 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", + "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) { + 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", "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) + 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()) + } + }) + } +} 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()