diff --git a/cmd/config.go b/cmd/config.go new file mode 100644 index 0000000..5ebd3cb --- /dev/null +++ b/cmd/config.go @@ -0,0 +1,65 @@ +package cmd + +import ( + "fmt" + "strconv" + + "github.com/neozmmv/blindspot/internal/utils" + "github.com/spf13/cobra" +) + +// ConfigCmd shows or changes persistent settings in ~/.blindspot/config.json. +// Settings apply to every future connect — CLI and tray alike — so nothing +// needs to be passed per session. Changes take effect on the next connect. +var ConfigCmd = &cobra.Command{ + Use: "config [setting] [value]", + Short: "Show or change persistent settings", + Long: `Show or change persistent settings (stored in ~/.blindspot/config.json). + +Settings: + up-mbit Force a fixed upload cap in Mbit/s. 0 (the default) means + automatic: the tunnel adapts its rate to the path on its + own, staying uncapped until real packet loss appears. + Only set this if automatic shaping misbehaves on your + link. Applies on the next connect. + +Examples: + blindspot config show current settings + blindspot config up-mbit 80 force a fixed 80 Mbit/s cap + blindspot config up-mbit 0 back to automatic`, + Args: cobra.MaximumNArgs(2), + Run: func(cmd *cobra.Command, args []string) { + cfg := utils.LoadConfig() + if len(args) == 0 { + if cfg.UpMbit > 0 { + fmt.Printf("up-mbit: %d (fixed cap)\n", cfg.UpMbit) + } else { + fmt.Println("up-mbit: 0 (automatic)") + } + return + } + if args[0] != "up-mbit" { + fmt.Printf("Unknown setting %q. Available: up-mbit\n", args[0]) + return + } + if len(args) == 1 { + fmt.Printf("up-mbit: %d\n", cfg.UpMbit) + return + } + v, err := strconv.Atoi(args[1]) + if err != nil || v < 0 { + fmt.Println("up-mbit must be a non-negative number of Mbit/s") + return + } + cfg.UpMbit = v + if err := utils.SaveConfig(cfg); err != nil { + fmt.Println("Error saving config:", err) + return + } + if v == 0 { + fmt.Println("Upload shaping set to automatic. Takes effect on the next connect.") + } else { + fmt.Printf("Upload capped at %d Mbit/s. Takes effect on the next connect.\n", v) + } + }, +} diff --git a/cmd/connect.go b/cmd/connect.go index d2e8f68..0a9c2d7 100644 --- a/cmd/connect.go +++ b/cmd/connect.go @@ -13,6 +13,7 @@ import ( "strconv" "strings" "sync" + "sync/atomic" "syscall" "time" @@ -75,6 +76,19 @@ var ConnectCmd = &cobra.Command{ daemon, _ := cmd.Flags().GetBool("daemon") statusFile, _ := cmd.Flags().GetString("status-file") insecure, _ := cmd.Flags().GetBool("insecure") + // Upload cap precedence: --up-mbit flag > BLINDSPOT_UP_MBIT env > + // persistent config ('blindspot config up-mbit N'). The config file is + // what the tray path uses — it shells out to connect without flags, and + // the UAC-elevated daemon shares the same ~/.blindspot. + upMbit, _ := cmd.Flags().GetInt("up-mbit") + if upMbit == 0 { + if v, err := strconv.Atoi(os.Getenv("BLINDSPOT_UP_MBIT")); err == nil && v > 0 { + upMbit = v + } + } + if upMbit == 0 { + upMbit = utils.LoadConfig().UpMbit + } if len(password) < 8 && isNew { fmt.Println("Password must be at least 8 characters long") @@ -192,6 +206,9 @@ var ConnectCmd = &cobra.Command{ publicAddr string ) peerConn := network.NewPeerConn(tr, privateKey, publicKey, psk, prologue) + if upMbit > 0 { + peerConn.SetFixedUploadRate(float64(upMbit) * 1e6 / 8) + } quit := make(chan struct{}) var quitOnce sync.Once closeQuit := func() { quitOnce.Do(func() { close(quit) }) } @@ -333,6 +350,62 @@ var ConnectCmd = &cobra.Command{ } }() + // TUN-side counters for the stats log: what enters from the OS, what is + // written back to the OS, and what the reverse-path filter rejects. + var tunInPkts, tunInBytes, tunOutPkts, tunOutBytes, rpfDrops, noPeerDrops atomic.Uint64 + + // Stats logger: once per second, append a delta line to stats.log while + // there is tunnel activity. Comparing this file across the two machines + // shows exactly where packets die: sender tx vs receiver rx is path + // loss; rx vs tun_out is local processing loss; dec_fail/replay/rekey + // indicate session-level trouble. + go func() { + logPath := filepath.Join(utils.GetBlindspotDir(), "stats.log") + f, err := os.OpenFile(logPath, os.O_CREATE|os.O_WRONLY|os.O_TRUNC, 0600) + if err != nil { + return + } + defer f.Close() + type snap struct { + txP, txB, txE, rxP, rxB, dec, rep, rek, tmo uint64 + tiP, tiB, toP, toB, rpf, nop uint64 + } + take := func() snap { + s := &network.Stats + return snap{ + s.TxPkts.Load(), s.TxBytes.Load(), s.TxErrs.Load(), + s.RxPkts.Load(), s.RxBytes.Load(), + s.RxDecryptFail.Load(), s.RxReplayDrop.Load(), + s.Rekeys.Load(), s.Timeouts.Load(), + tunInPkts.Load(), tunInBytes.Load(), + tunOutPkts.Load(), tunOutBytes.Load(), + rpfDrops.Load(), noPeerDrops.Load(), + } + } + prev := take() + ticker := time.NewTicker(time.Second) + defer ticker.Stop() + for { + select { + case <-quit: + return + case <-ticker.C: + cur := take() + if cur == prev { + continue + } + fmt.Fprintf(f, + "%s tun_in=%d/%dB tx=%d/%dB txerr=%d | rx=%d/%dB tun_out=%d/%dB dec_fail=%d replay=%d rpf=%d nopeer=%d | rekey=%d timeout=%d pace=%dbps\n", + time.Now().Format("15:04:05"), + cur.tiP-prev.tiP, cur.tiB-prev.tiB, cur.txP-prev.txP, cur.txB-prev.txB, cur.txE-prev.txE, + cur.rxP-prev.rxP, cur.rxB-prev.rxB, cur.toP-prev.toP, cur.toB-prev.toB, + cur.dec-prev.dec, cur.rep-prev.rep, cur.rpf-prev.rpf, cur.nop-prev.nop, + cur.rek-prev.rek, cur.tmo-prev.tmo, network.Stats.PaceBps.Load()*8) + prev = cur + } + } + }() + // tunBufPool recycles packet buffers across both pump directions so the // steady state allocates nothing per packet. tunBufPool := sync.Pool{New: func() any { @@ -370,9 +443,12 @@ var ConnectCmd = &cobra.Command{ // spoofing another peer's virtual IP inside the (authenticated) tunnel. expectedVIP, ok := addrToVIP.Load(senders[i]) if !ok || !bstun.SrcIPMatchesVirtualIP(bufs[i], expectedVIP.(string)) { + rpfDrops.Add(1) continue } wr = append(wr, bufs[i]) + tunOutPkts.Add(1) + tunOutBytes.Add(uint64(len(bufs[i]))) } if len(wr) == 0 { continue @@ -407,6 +483,8 @@ var ConnectCmd = &cobra.Command{ if len(packet) < 20 || packet[0]>>4 != 4 { continue // not an IPv4 packet } + tunInPkts.Add(1) + tunInBytes.Add(uint64(len(packet))) select { case outCh <- packet: bufs[i] = getTunBuf() // buffer ownership moved to the sender @@ -420,6 +498,9 @@ var ConnectCmd = &cobra.Command{ // Sender: aggregate whatever the reader has produced, group consecutive // packets by destination peer, and push each group through one batched // encrypt+send (one counter reservation, parallel AEAD, few syscalls). + // Upload shaping happens inside SendTunBatch: adaptive by default + // (engages only when CtrlAck feedback shows path loss), or fixed when + // --up-mbit / config override it. go func() { pending := make([][]byte, 0, 128) flush := func() { @@ -427,6 +508,7 @@ var ConnectCmd = &cobra.Command{ destIP := net.IP(pending[start][16:20]).String() addrVal, ok := virtualIPMap.Load(destIP) if !ok { + noPeerDrops.Add(1) putTunBuf(pending[start]) // no peer with that virtual IP start++ continue @@ -524,6 +606,7 @@ func init() { ConnectCmd.Flags().StringP("password", "p", "", "Session password") ConnectCmd.Flags().BoolP("new", "n", false, "Create new session with password") ConnectCmd.Flags().Bool("insecure", false, "Allow a plaintext http:// rendezvous (NOT recommended)") + ConnectCmd.Flags().Int("up-mbit", 0, "Force a fixed upload cap in Mbit/s (0 = automatic: adapts to the path)") ConnectCmd.MarkFlagRequired("session") ConnectCmd.Flags().Bool("daemon", false, "") ConnectCmd.Flags().String("status-file", "", "") diff --git a/cmd/root.go b/cmd/root.go index d96fafb..eda246a 100644 --- a/cmd/root.go +++ b/cmd/root.go @@ -27,6 +27,7 @@ func init() { rootCmd.AddCommand(SendCmd) rootCmd.AddCommand(ReceiveCmd) rootCmd.AddCommand(IdentityCmd) + rootCmd.AddCommand(ConfigCmd) rootCmd.AddCommand(VersionCmd) } diff --git a/internal/network/pace_test.go b/internal/network/pace_test.go new file mode 100644 index 0000000..e400601 --- /dev/null +++ b/internal/network/pace_test.go @@ -0,0 +1,118 @@ +package network + +import ( + "testing" + "time" +) + +// ackStep simulates one CtrlAck arriving after `elapsed`, with the session +// having sent sentB bytes / dMax packets of which dAcc were accepted. +func ackStep(p *PeerConn, s *peerSession, elapsed time.Duration, sentB uint64, dMax, dAcc uint64) { + s.paceMu.Lock() + s.lastAckAt = time.Now().Add(-elapsed) + s.paceSentB = sentB + base := s.lastAckMax + baseAcc := s.lastAckAcc + s.paceMu.Unlock() + p.handleAck(s, base+dMax, baseAcc+dAcc) +} + +// TestAdaptiveShaper drives handleAck through the engage → cut → probe cycle +// and checks the control law: uncapped until real loss, engaged near the +// observed send rate on a loss event, growing again on clean utilized +// intervals, and not double-cutting within the guard window. +func TestAdaptiveShaper(t *testing.T) { + p := &PeerConn{} + s := &peerSession{} + + // Baseline ack: seeds the snapshot, no rate yet. + p.handleAck(s, 100, 100) + if got := p.effectiveRate(s); got != 0 { + t.Fatalf("rate engaged with no loss signal: %v", got) + } + + // Clean traffic: stays uncapped. + ackStep(p, s, 200*time.Millisecond, 2_000_000, 1000, 1000) + if got := p.effectiveRate(s); got != 0 { + t.Fatalf("rate engaged on clean interval: %v", got) + } + + // 15% loss while sending 2 MB in 200ms (10 MB/s): engages at + // sendRate*cutBig = 10e6*0.6 = 6 MB/s. + ackStep(p, s, 200*time.Millisecond, 2_000_000, 1000, 850) + rate := p.effectiveRate(s) + if rate < 5e6 || rate > 7e6 { + t.Fatalf("expected ~6 MB/s after severe loss engage, got %.0f", rate) + } + + // Another severe-loss ack immediately after: inside the cut guard, no + // second cut. + ackStep(p, s, 200*time.Millisecond, 1_200_000, 800, 600) + if got := p.effectiveRate(s); got != rate { + t.Fatalf("double cut within guard window: %.0f -> %.0f", rate, got) + } + + // Clean and >50%% utilized: probes upward by paceGrow. + s.paceMu.Lock() + s.lastCut = time.Now().Add(-time.Second) // move past the guard + s.paceMu.Unlock() + ackStep(p, s, 200*time.Millisecond, uint64(rate*0.2), 800, 800) + if got := p.effectiveRate(s); got <= rate { + t.Fatalf("expected upward probe on clean utilized interval, got %.0f (was %.0f)", got, rate) + } + + // Clean but idle (2%% utilization): rate must hold, not balloon. + before := p.effectiveRate(s) + ackStep(p, s, 200*time.Millisecond, uint64(before*0.2*0.02), 60, 60) + if got := p.effectiveRate(s); got != before { + t.Fatalf("rate changed on idle interval: %.0f -> %.0f", before, got) + } + + // Tiny sample (below ackMinPkts): ignored entirely. + ackStep(p, s, 200*time.Millisecond, 10_000, 10, 2) + if got := p.effectiveRate(s); got != before { + t.Fatalf("rate reacted to sub-threshold sample: %.0f -> %.0f", before, got) + } + + // Manual override wins over the adaptive rate and freezes adaptation. + p.SetFixedUploadRate(1e6) + if got := p.effectiveRate(s); got != 1e6 { + t.Fatalf("fixed override not applied: %.0f", got) + } + ackStep(p, s, 200*time.Millisecond, 2_000_000, 1000, 500) + if got := p.effectiveRate(s); got != 1e6 { + t.Fatalf("adaptive logic ran under fixed override: %.0f", got) + } + p.SetFixedUploadRate(0) + if got := p.effectiveRate(s); got != before { + t.Fatalf("adaptive rate lost after clearing override: %.0f (want %.0f)", got, before) + } +} + +// TestPaceAdmitThrottles checks the virtual-clock pacer actually spaces sends: +// pushing ~3x a 10 MB/s budget's worth of bytes in a tight loop must take +// roughly the budgeted time, while an uncapped session must not sleep. +func TestPaceAdmitThrottles(t *testing.T) { + s := &peerSession{} + + start := time.Now() + for range 100 { + s.paceAdmit(1500, 0) + } + if d := time.Since(start); d > 50*time.Millisecond { + t.Fatalf("uncapped paceAdmit slept: %v", d) + } + + const rate = 10e6 // 10 MB/s + total := 0 + start = time.Now() + for range 100 { + s.paceAdmit(15_000, rate) // 1.5 MB total = 150ms of budget + total += 15_000 + } + elapsed := time.Since(start) + want := time.Duration(float64(total) / rate * float64(time.Second)) + if elapsed < want/2 { + t.Fatalf("paced sends finished in %v, want >= ~%v", elapsed, want) + } +} diff --git a/internal/network/peer.go b/internal/network/peer.go index b64cabc..0ef798e 100644 --- a/internal/network/peer.go +++ b/internal/network/peer.go @@ -8,6 +8,7 @@ import ( "encoding/hex" "errors" "fmt" + "math" "net" "runtime" "sync" @@ -40,6 +41,41 @@ const ( counterLen = 8 ) +// Adaptive upload shaping. A TCP flow inside the tunnel cannot sense the real +// bottleneck (the tunnel emits UDP at local line rate), so without shaping its +// growing window is fired as salvos that overflow the bottleneck queue in one +// go — burst loss TCP reads as a timeout, collapsing throughput. The shaper +// stays fully uncapped until CtrlAck feedback shows real path loss, then +// engages at the observed send rate and AIMD-tracks the path: multiplicative +// cut on loss, multiplicative probe upward while clean and utilized. +const ( + // ackInterval is how often the receive side reports delivery per session. + ackInterval = 200 * time.Millisecond + // ackMinPkts is the minimum packets per interval to treat the measured + // loss as signal rather than noise. + ackMinPkts = 50 + // paceChunk bounds packets per paced send so sleep slices stay in the + // low-millisecond range where OS timers are accurate. + paceChunk = 32 + // paceBurstCredit is how far the virtual clock may lag wall time: a small + // burst allowance after idle without unbounded credit. + paceBurstCredit = 5 * time.Millisecond + // paceMinRate floors the shaper (2 Mbit/s) so a loss storm cannot choke + // the tunnel to nothing; paceMaxRate is where shaping stops mattering and + // the session returns to uncapped. + paceMinRate = 250e3 + paceMaxRate = 1.5e9 + // Loss thresholds and AIMD gains per ack interval. + lossEngage = 0.03 // engage/cut mildly above this + lossSevere = 0.10 // cut hard above this + paceGrow = 1.25 + paceCut = 0.85 + paceCutBig = 0.6 + // paceCutGuard spaces rate cuts so one loss event (reported across + // consecutive acks) is not punished twice. + paceCutGuard = 300 * time.Millisecond +) + // peerSession holds the per-peer handshake and transport state. All mutable // fields are guarded by mu. type peerSession struct { @@ -81,6 +117,27 @@ type peerSession struct { // while this is fresh — pongs alone are unreliable under congestion. lastRx time.Time + // Receive-side delivery counters for CtrlAck generation, guarded by mu + // with the rest of the receive state. rxMaxCtr/rxAccepted are cumulative + // for the current keys; ackSentMax is the last rxMaxCtr already reported. + rxMaxCtr uint64 + rxAccepted uint64 + ackSentMax uint64 + + // Send-side adaptive shaper state, guarded by paceMu (its own mutex so the + // per-chunk pacing never contends with the receive path on mu). paceRate + // is the current cap in bytes/sec, 0 = uncapped; paceNext is the virtual + // clock; paceSentB accumulates bytes handed to the bind since the last + // ack, giving the utilization and engage-rate measurements. + paceMu sync.Mutex + paceRate float64 + paceNext time.Time + paceSentB uint64 + lastAckAt time.Time + lastAckMax uint64 + lastAckAcc uint64 + lastCut time.Time + // rekeyNotice is a prebuilt CtrlRekey packet sealed under the previous // session keys. driveHandshake retransmits it until the new handshake // establishes, so a peer that still holds the old session tears it down @@ -130,6 +187,13 @@ func (s *peerSession) installTransportKeysLocked(cs0, cs1 *noise.CipherState) er s.txAEAD, s.rxAEAD = txAEAD, rxAEAD s.txCtr = 0 s.rekeyNotice = nil + // Fresh keys restart the counter space: reset the delivery bookkeeping but + // keep paceRate — what we learned about the path survives a rekey. + s.rxMaxCtr, s.rxAccepted, s.ackSentMax = 0, 0, 0 + s.paceMu.Lock() + s.lastAckAt, s.lastAckMax, s.lastAckAcc = time.Time{}, 0, 0 + s.paceSentB = 0 + s.paceMu.Unlock() return nil } @@ -154,15 +218,21 @@ func (s *peerSession) decrypt(dst []byte, pkt []byte) ([]byte, error) { nonce := aesgcmNonce(counter) plaintext, err := aead.Open(dst, nonce[:], pkt[transportHeaderLen:], header) if err != nil { + Stats.RxDecryptFail.Add(1) return nil, err } s.mu.Lock() fresh := s.replay.Check(counter) if fresh { s.lastRx = time.Now() + s.rxAccepted++ + if counter > s.rxMaxCtr { + s.rxMaxCtr = counter + } } s.mu.Unlock() if !fresh { + Stats.RxReplayDrop.Add(1) return nil, fmt.Errorf("replayed or out-of-window packet from %s", s.addr) } return plaintext, nil @@ -213,6 +283,147 @@ type PeerConn struct { // tunPkts/tunOK are ReadTunBatch scratch space (single consumer). tunPkts []rxPacket tunOK []bool + + // fixedRateBits, when non-zero, is a math.Float64bits bytes/sec cap that + // overrides the adaptive shaper for every session (the --up-mbit knob). + fixedRateBits atomic.Uint64 +} + +// SetFixedUploadRate forces a fixed upload cap in bytes/sec across all +// sessions, disabling the adaptive shaper. 0 restores adaptive mode. +func (p *PeerConn) SetFixedUploadRate(bytesPerSec float64) { + p.fixedRateBits.Store(math.Float64bits(bytesPerSec)) +} + +// effectiveRate returns the pacing rate for a session: the manual override if +// set, else the session's adaptive rate. 0 means uncapped. +func (p *PeerConn) effectiveRate(s *peerSession) float64 { + if f := math.Float64frombits(p.fixedRateBits.Load()); f > 0 { + return f + } + s.paceMu.Lock() + r := s.paceRate + s.paceMu.Unlock() + return r +} + +// paceAdmit charges nbytes against the session's virtual send clock at the +// given rate, sleeping off any deficit. It also accumulates the sent-bytes +// meter the ack handler uses for utilization. rate 0 = uncapped, no sleep. +func (s *peerSession) paceAdmit(nbytes int, rate float64) { + s.paceMu.Lock() + s.paceSentB += uint64(nbytes) + if rate <= 0 { + s.paceMu.Unlock() + return + } + now := time.Now() + if s.paceNext.Before(now.Add(-paceBurstCredit)) { + s.paceNext = now.Add(-paceBurstCredit) + } + wait := s.paceNext.Sub(now) + s.paceNext = s.paceNext.Add(time.Duration(float64(nbytes) / rate * float64(time.Second))) + s.paceMu.Unlock() + if wait > 0 { + time.Sleep(wait) + } +} + +// handleAck folds one CtrlAck delivery report into the session's shaper. Loss +// is measured from counter-space deltas (Δaccepted vs Δmax-seen), which has no +// in-flight bias. The shaper engages only on real loss: until then the session +// sends uncapped, so clean paths (LAN) never pay for shaping. +func (p *PeerConn) handleAck(s *peerSession, maxCtr, accepted uint64) { + if math.Float64frombits(p.fixedRateBits.Load()) > 0 { + return // manual override active; adaptive state not maintained + } + now := time.Now() + s.paceMu.Lock() + defer s.paceMu.Unlock() + if s.lastAckAt.IsZero() || maxCtr < s.lastAckMax { + // First report of this key epoch (or a stale one after rekey): just + // snapshot a baseline. + s.lastAckAt, s.lastAckMax, s.lastAckAcc = now, maxCtr, accepted + s.paceSentB = 0 + return + } + dMax := maxCtr - s.lastAckMax + dAcc := accepted - s.lastAckAcc + elapsed := now.Sub(s.lastAckAt).Seconds() + sentB := s.paceSentB + s.lastAckAt, s.lastAckMax, s.lastAckAcc = now, maxCtr, accepted + s.paceSentB = 0 + if dMax < ackMinPkts || elapsed <= 0 { + return // too little traffic to read loss from + } + loss := 1 - float64(dAcc)/float64(dMax) + switch { + case loss > lossSevere && now.Sub(s.lastCut) > paceCutGuard: + if s.paceRate <= 0 { + s.paceRate = float64(sentB) / elapsed + } + s.paceRate *= paceCutBig + s.lastCut = now + case loss > lossEngage && now.Sub(s.lastCut) > paceCutGuard: + if s.paceRate <= 0 { + s.paceRate = float64(sentB) / elapsed + } + s.paceRate *= paceCut + s.lastCut = now + case loss < 0.01 && s.paceRate > 0: + // Clean interval: probe upward, but only when actually pushing near + // the cap — otherwise an idle session's rate would balloon. + if float64(sentB) > 0.5*s.paceRate*elapsed { + s.paceRate *= paceGrow + } + } + if s.paceRate > 0 { + if s.paceRate < paceMinRate { + s.paceRate = paceMinRate + } + if s.paceRate > paceMaxRate { + s.paceRate = 0 // outgrew relevance: back to uncapped + } + } + Stats.PaceBps.Store(uint64(s.paceRate)) +} + +// ackLoop periodically reports per-session delivery back to each peer, feeding +// the peer's shaper. ~5 tiny control packets per second per active peer. +func (p *PeerConn) ackLoop() { + ticker := time.NewTicker(ackInterval) + defer ticker.Stop() + for { + select { + case <-p.stop: + return + case <-ticker.C: + } + p.mu.Lock() + sessions := make([]*peerSession, 0, len(p.sessions)) + for _, s := range p.sessions { + sessions = append(sessions, s) + } + p.mu.Unlock() + for _, s := range sessions { + s.mu.Lock() + est := s.established + maxc, acc := s.rxMaxCtr, s.rxAccepted + changed := maxc != s.ackSentMax + if est && changed { + s.ackSentMax = maxc + } + s.mu.Unlock() + if !est || !changed { + continue + } + var payload [17]byte + payload[0] = CtrlAck + binary.BigEndian.PutUint64(payload[1:9], maxc) + binary.BigEndian.PutUint64(payload[9:17], acc) + p.sendBatchSession(s, [][]byte{payload[:]}, PacketControl) + } + } } // NewPeerConn creates a PeerConn on an opened Transport and starts its receive @@ -244,6 +455,7 @@ func NewPeerConn(t *Transport, privateKey, publicKey, psk, prologue []byte) *Pee p.recvWG.Wait() close(p.recvDone) }() + go p.ackLoop() return p } @@ -374,12 +586,20 @@ func (p *PeerConn) handlePacket(pkt []byte, ep wgconn.Endpoint) { if gen, ok := p.rearmSession(s, false); ok { go p.watchRearm(s, gen) } + case CtrlAck: + if len(plaintext) == 17 { + p.handleAck(s, + binary.BigEndian.Uint64(plaintext[1:9]), + binary.BigEndian.Uint64(plaintext[9:17])) + } } case PacketData, PacketTun: s := p.sessionForEndpoint(ep) if s == nil { return } + Stats.RxPkts.Add(1) + Stats.RxBytes.Add(uint64(len(pkt))) buf := getPacketBuf(len(pkt)) copy(buf, pkt) select { @@ -858,19 +1078,38 @@ func (p *PeerConn) sendBatchSession(s *peerSession, payloads [][]byte, pktType b wire[i] = sealPacket(aead, pktType, base+uint64(i), payloads[i]) }) + // When the shaper is engaged, hand the batch to the bind in small paced + // chunks; uncapped sessions keep full-width batches. + rate := p.effectiveRate(s) + step := p.batch + if rate > 0 && step > paceChunk { + step = paceChunk + } var err error - for off := 0; off < m; off += p.batch { - end := off + p.batch + for off := 0; off < m; off += step { + end := off + step if end > m { end = m } + nbytes := 0 + for _, w := range wire[off:end] { + nbytes += len(w) + 28 // + IPv4/UDP header overhead on the wire + } + s.paceAdmit(nbytes, rate) if e := p.bind.Send(wire[off:end], ep); e != nil && err == nil { err = e } } + var payloadBytes uint64 for _, w := range wire { + payloadBytes += uint64(len(w)) putPacketBuf(w) } + Stats.TxPkts.Add(uint64(m)) + Stats.TxBytes.Add(payloadBytes) + if err != nil { + Stats.TxErrs.Add(1) + } return err } @@ -955,6 +1194,7 @@ func (p *PeerConn) RemovePeer(addr *net.UDPAddr) { // so the session heals on its own if the outage was transient (the peer never // told the rendezvous it left, so nothing else would ever reconnect the two). func (p *PeerConn) TimeoutPeer(addr *net.UDPAddr) { + Stats.Timeouts.Add(1) key := canonAddrPort(addr.AddrPort()).String() p.mu.Lock() delete(p.missedPings, key) @@ -1009,6 +1249,7 @@ func (p *PeerConn) rearmSession(s *peerSession, notify bool) (uint64, bool) { return 0, false } s.rekeyNotice = notice + Stats.Rekeys.Add(1) go p.driveHandshake(s, gen, time.Now().Add(retryWindow)) return gen, true } diff --git a/internal/network/protocol.go b/internal/network/protocol.go index 526fe2a..95409f4 100644 --- a/internal/network/protocol.go +++ b/internal/network/protocol.go @@ -58,9 +58,17 @@ const ( // side drops its copy and re-handshakes immediately instead of answering the // fresh msg1 with a stale cached msg2. Peers that don't know the opcode simply // ignore it and heal via their own keepalive timeout, as before. +// CtrlAck is the delivery report driving the adaptive upload shaper: the +// receiver periodically reports the highest transport counter it has seen and +// the total packets it accepted, both cumulative for the session. From the +// deltas the sender measures true path loss (counter gaps have no in-flight +// bias — a gap means the packet was overtaken or lost) and shapes its send +// rate to the path. Payload: [opcode][maxCounter:8][accepted:8]. Peers that +// don't know the opcode ignore it, leaving the sender uncapped (old behavior). const ( CtrlPing byte = 0x01 CtrlPong byte = 0x02 CtrlDead byte = 0x03 CtrlRekey byte = 0x04 + CtrlAck byte = 0x05 ) diff --git a/internal/network/stats.go b/internal/network/stats.go new file mode 100644 index 0000000..1ec3067 --- /dev/null +++ b/internal/network/stats.go @@ -0,0 +1,31 @@ +package network + +import "sync/atomic" + +// TunnelStats are process-global counters across the tunnel hot paths, cheap +// enough (single atomic add per event) to leave always-on. The daemon samples +// them once per second and logs deltas so a misbehaving transfer shows exactly +// which stage packets die at: sent-but-not-received is path loss, received- +// but-failing-decrypt is a key/session problem, replay drops are reordering +// beyond the window, rekeys/timeouts are session churn. +type TunnelStats struct { + TxPkts atomic.Uint64 // transport packets handed to the bind + TxBytes atomic.Uint64 // wire bytes across those packets + TxErrs atomic.Uint64 // batches whose bind.Send returned an error + + RxPkts atomic.Uint64 // data/tun packets received and queued (pre-decrypt) + RxBytes atomic.Uint64 // wire bytes across those packets + + RxDecryptFail atomic.Uint64 // packets failing AEAD authentication + RxReplayDrop atomic.Uint64 // authenticated packets outside the replay window + + Rekeys atomic.Uint64 // sessions re-armed (keepalive timeout or CtrlRekey) + Timeouts atomic.Uint64 // peers declared dead by the keepalive + + // PaceBps is a gauge: the adaptive shaper's current rate in bytes/sec + // (last session updated wins; 0 = uncapped). + PaceBps atomic.Uint64 +} + +// Stats is the process-wide instance updated by PeerConn. +var Stats TunnelStats diff --git a/internal/utils/config.go b/internal/utils/config.go new file mode 100644 index 0000000..991140c --- /dev/null +++ b/internal/utils/config.go @@ -0,0 +1,44 @@ +package utils + +import ( + "encoding/json" + "os" + "path/filepath" +) + +// Config holds persistent user settings, stored as ~/.blindspot/config.json. +// Both the CLI and the tray-launched daemon read it (same user profile, so the +// UAC-elevated daemon sees the same file), which is what makes a setting like +// the upload cap stick without passing flags on every connect. +type Config struct { + // UpMbit paces the tunnel's upload at this many Mbit/s. 0 means uncapped. + // Set it to ~90% of the machine's real uplink so tunnelled TCP transfers + // don't overflow the bottleneck queue in line-rate bursts. + UpMbit int `json:"up_mbit,omitempty"` +} + +func configPath() string { return filepath.Join(GetBlindspotDir(), "config.json") } + +// LoadConfig reads the config file. A missing or unreadable file yields the +// zero Config — settings are always optional. +func LoadConfig() Config { + var c Config + data, err := os.ReadFile(configPath()) + if err != nil { + return c + } + json.Unmarshal(data, &c) + return c +} + +// SaveConfig writes the config file, creating ~/.blindspot if needed. +func SaveConfig(c Config) error { + if err := os.MkdirAll(GetBlindspotDir(), 0700); err != nil { + return err + } + data, err := json.MarshalIndent(c, "", " ") + if err != nil { + return err + } + return os.WriteFile(configPath(), data, 0600) +}