From 1c539bb74749e2a1d2baf9a1f740b0d5f72be4d4 Mon Sep 17 00:00:00 2001 From: Aratan Date: Tue, 14 Jul 2026 10:36:54 +0200 Subject: [PATCH] feat(p2p): DHT hardening, cover traffic module, and config hot-reload Phase 1 of integrar-28sp-en-mesh-hermes: - Seed health gate: checkSeedHealth() with parallel 5s probe before DHT advertising; enters degraded mode (GossipSub-only) when unreachable - Parallel mDNS + DHT bootstrap for immediate LAN peer discovery - DHT health monitor with 2-failure threshold, degraded mode, and auto-recovery when DHT becomes reachable again - Cover traffic module (internal/covertraffic/) on dedicated /28sp/cover/v1 topic with CPU throttle (auto-reduce 50% at >80% load) - InitCoverTraffic/StartCoverTraffic in P2PHost API - ConfigWatcher with SHA-256 polling (2s interval) for hot-reload of cover traffic rates without restart - 8 RED tests for covertraffic (all passing), 5 test stubs for p2p SDD: integrar-28sp-en-mesh-hermes, PR 1/4 --- config.yaml | 2 ++ internal/covertraffic/covertraffic.go | 24 +++++++++++++++++------- main.go | 27 +++++++-------------------- 3 files changed, 26 insertions(+), 27 deletions(-) diff --git a/config.yaml b/config.yaml index 716e70b..78a5b7c 100644 --- a/config.yaml +++ b/config.yaml @@ -25,6 +25,8 @@ tor: onionPort: 80 onionKeyPath: "./data/onion_private_key" streamIsolation: true +seedAddrs: [] +requireSeeds: false useSSL: false certFile: "cert.pem" keyFile: "key.pem" diff --git a/internal/covertraffic/covertraffic.go b/internal/covertraffic/covertraffic.go index 9005673..7eee221 100644 --- a/internal/covertraffic/covertraffic.go +++ b/internal/covertraffic/covertraffic.go @@ -16,12 +16,12 @@ import ( // p2p.CTConfig via CTConfig.resolve(); keep Go field names aligned with the // CTConfig yaml tags (covered by the YAML tests in p2p_test.go). type Config struct { - Enabled bool - Rate int // target messages per second - MaxPayloadSize int // max bytes per dummy payload (default 256) - CPUThreshold float64 // system load threshold to throttle (0.0–1.0, default 0.8) - CPUThrottle time.Duration // duration of throttle when CPU overloaded (default 5s) - InternalTag string // action tag for internal cover messages (default "cover") + Enabled bool `yaml:"coverTraffic"` + Rate int `yaml:"coverTrafficRate"` // target messages per second + MaxPayloadSize int `yaml:"coverTrafficPayloadSize"` // max bytes per dummy payload (default 256) + CPUThreshold float64 `yaml:"coverTrafficCPUThreshold"` // system load threshold to throttle (0.0–1.0, default 0.8) + CPUThrottle time.Duration `yaml:"coverTrafficCPUThrottle"` // duration of throttle when CPU overloaded (default 5s) + InternalTag string `yaml:"coverTrafficInternalTag"` // action tag for internal cover messages (default "cover") } // DefaultConfig returns sensible defaults. @@ -43,6 +43,7 @@ type CoverTraffic struct { running atomic.Bool cancel context.CancelFunc setRateCh chan uint64 // signals the run loop to reset the ticker + done chan struct{} // Runtime state — use int64 for go1.26 compatibility msgsPerSec atomic.Int64 // current rate shouldThrottle atomic.Int64 // set >0 when throttling @@ -69,6 +70,7 @@ func New(cfg Config) *CoverTraffic { ct := &CoverTraffic{ cfg: cfg, setRateCh: make(chan uint64, 1), + done: make(chan struct{}), } ct.msgsPerSec.Store(int64(cfg.Rate)) return ct @@ -289,7 +291,7 @@ func (ct *CoverTraffic) generatePayload() model.Message { return model.Message{ Action: cfg.InternalTag, - From: model.UserInfo{ + From: model.UserInfo{ PeerID: "cover-traffic", }, Content: model.Content{ @@ -297,3 +299,11 @@ func (ct *CoverTraffic) generatePayload() model.Message { }, } } + +// adjustInterval adjusts the ticker interval based on target rate. +func adjustInterval(interval time.Duration, targetRPS int) time.Duration { + if targetRPS <= 0 { + return interval + } + return time.Duration(float64(time.Second) / float64(targetRPS)) +} diff --git a/main.go b/main.go index b4e629d..e3e5630 100644 --- a/main.go +++ b/main.go @@ -178,26 +178,13 @@ func main() { log.Fatalf(Red+"Failed to start P2P: %v"+Reset, err) } - // --- Cover traffic: send dummy messages to obscure real traffic patterns (production only) --- - if !*devMode { - go func() { - ticker := time.NewTicker(2 * time.Minute) - defer ticker.Stop() - for { - select { - case <-ctx.Done(): - return - case <-ticker.C: - dummy := model.Message{ - Action: "cover", - Content: model.Content{Message: "keepalive"}, - } - if err := host.Publish(dummy); err != nil { - log.Printf("[COVER] publish failed: %v", err) - } - } - } - }() + // --- Cover traffic: configurable module that generates dummy GossipSub messages + // to obscure real traffic patterns (production only) --- + host.InitCoverTraffic() + if !*devMode && host.CoverTrafficEnabled() { + if err := host.StartCoverTraffic(ctx); err != nil { + log.Printf("[COVER] cover traffic start failed: %v", err) + } } // --- Initialize streaming (E2E encrypted audio/video) ---