diff --git a/server/go/cmd/main.go b/server/go/cmd/main.go index 531ac66..9f541a1 100644 --- a/server/go/cmd/main.go +++ b/server/go/cmd/main.go @@ -26,11 +26,12 @@ import ( // MyHandler implements the server.Handler interface type MyHandler struct { - log *logger.Logger - auth *auth.Authenticator - srv *server.Server - hub *hub.Hub - signer licensing.Signer // CMD_MASTERSETTING signer (local / remote / noop) + log *logger.Logger + auth *auth.Authenticator + srv *server.Server + hub *hub.Hub + signer licensing.Signer // CMD_MASTERSETTING signer (local / remote / noop) + loginLimit *wsauth.RateLimiter // per-IP TOKEN_AUTH throttle } // OnConnect is called when a client connects @@ -497,6 +498,20 @@ func (h *MyHandler) sendMasterSetting(ctx *connection.Context, startTime, client // handleAuth handles authorization request (TOKEN_AUTH = 100) func (h *MyHandler) handleAuth(ctx *connection.Context, data []byte) { + // Per-IP throttle on TOKEN_AUTH: rate-limit online passcode-guessing + // attempts against the device authorization protocol. On exceed we answer + // with a normal "failed" response (resp[0]=0) so the client gets a + // definitive answer instead of hanging, while a brute-forcer is capped. + ip := ctx.GetPeerIP() + if !h.loginLimit.Allow(ip) { + h.log.Warn("TOKEN_AUTH rate-limited: ip=%s conn=%d", ip, ctx.ID) + resp := h.auth.BuildResponse(&auth.AuthResult{Valid: false, Message: "尝试过于频繁,请稍后再试"}) + if err := h.srv.Send(ctx, resp); err != nil { + h.log.Error("Failed to send rate-limit auth response to client %d: %v", ctx.ID, err) + } + return + } + result := h.auth.Authenticate(data) info := ctx.GetInfo() @@ -674,6 +689,19 @@ func normalizeBasePath(s string) string { return "/" + s } +// envInt parses an integer env var, returning def when unset or malformed. +func envInt(name string, def int) int { + v := strings.TrimSpace(os.Getenv(name)) + if v == "" { + return def + } + n, err := strconv.Atoi(v) + if err != nil { + return def + } + return n +} + // runIssueToken handles the "issue-token" subcommand. It mints a customer JWT // signed with the operator's RSA private key and prints it to stdout. // @@ -875,23 +903,43 @@ func main() { log.Warn("Failed to load %s: %v (continuing with admin only)", usersFile, err) } + // TCP authorization (TOKEN_AUTH) brute-force throttle. Keyed by source IP + // (the real TCP peer — there is no X-Forwarded-For on the device port). + // Generous enough that a NAT office whose devices all re-auth at once is + // unaffected, tight enough to make online passcode guessing impractical. + authLoginLimit := wsauth.NewRateLimiter(120, time.Minute) + + // Bounded-key hygiene: prune aged-out entries so the per-IP map cannot grow + // without bound. Runs for the process lifetime (a daemon goroutine; the + // process exits via the SIGINT/SIGTERM handler at the bottom of main). + go func() { + t := time.NewTicker(time.Minute) + defer t.Stop() + for range t.C { + authLoginLimit.Sweep() + } + }() + // Create servers for each port var servers []*server.Server for _, port := range ports { config := server.DefaultConfig() config.Port = port config.MaxConnections = 9999 + config.PerIPMaxConnections = envInt("YAMA_PERIP_MAX_CONNS", config.PerIPMaxConnections) + config.PerIPConnRateLimit = envInt("YAMA_PERIP_CONN_RATE", config.PerIPConnRateLimit) srv := server.New(config) srv.SetLogger(log.WithPrefix(fmt.Sprintf("Server:%d", port))) // Create handler for this server handler := &MyHandler{ - log: log.WithPrefix(fmt.Sprintf("Handler:%d", port)), - auth: authenticator, - srv: srv, - hub: deviceHub, - signer: signer, + log: log.WithPrefix(fmt.Sprintf("Handler:%d", port)), + auth: authenticator, + srv: srv, + hub: deviceHub, + signer: signer, + loginLimit: authLoginLimit, } srv.SetHandler(handler) diff --git a/server/go/server/server.go b/server/go/server/server.go index a5bf118..80c6ab1 100644 --- a/server/go/server/server.go +++ b/server/go/server/server.go @@ -22,18 +22,30 @@ type Config struct { KeepAliveTime time.Duration ReadTimeout time.Duration WriteTimeout time.Duration + + // Per-IP connection limits. The TCP listener (port 6543) is exposed + // directly to the public internet, so a single host must not be able to + // exhaust the global connection budget or flood new-connection attempts. + // Defaults are generous enough for a NAT office with dozens of devices + // sharing one source IP; raise them if you run a larger NAT deployment. + PerIPMaxConnections int // max concurrent connections per source IP (0 = unlimited) + PerIPConnRateLimit int // max new connections per PerIPConnRateWindow per IP (0 = unlimited) + PerIPConnRateWindow time.Duration // window for PerIPConnRateLimit } // DefaultConfig returns default configuration func DefaultConfig() Config { return Config{ - Port: 6543, - MaxConnections: 9999, - ReadBufferSize: 8192, - WriteBufferSize: 8192, - KeepAliveTime: time.Minute * 5, - ReadTimeout: time.Minute * 2, - WriteTimeout: time.Second * 30, + Port: 6543, + MaxConnections: 9999, + ReadBufferSize: 8192, + WriteBufferSize: 8192, + KeepAliveTime: time.Minute * 5, + ReadTimeout: time.Minute * 2, + WriteTimeout: time.Second * 30, + PerIPMaxConnections: 256, + PerIPConnRateLimit: 120, + PerIPConnRateWindow: time.Minute, } } @@ -51,6 +63,7 @@ type Server struct { manager *connection.Manager handler Handler parser *protocol.Parser + ipLimit *ipLimiter running atomic.Bool wg sync.WaitGroup @@ -69,6 +82,7 @@ func New(config Config) *Server { config: config, manager: connection.NewManager(config.MaxConnections), parser: protocol.NewParser(), + ipLimit: newIPLimiter(config.PerIPMaxConnections, config.PerIPConnRateLimit, config.PerIPConnRateWindow), ctx: ctx, cancel: cancel, logger: logger.New(logger.DefaultConfig()).WithPrefix("Server"), @@ -77,6 +91,110 @@ func New(config Config) *Server { return s } +// ipLimiter caps concurrent connections and the rate of new connections per +// source IP. Both limits are per-host, not global, so a single attacker (or a +// single misbehaving host) cannot starve legitimate devices of the global +// connection budget. +type ipLimiter struct { + mu sync.Mutex + maxConcurrent int + window time.Duration + maxPerWindow int + concurrent map[string]int + recent map[string][]time.Time +} + +func newIPLimiter(maxConcurrent, maxPerWindow int, window time.Duration) *ipLimiter { + return &ipLimiter{ + maxConcurrent: maxConcurrent, + maxPerWindow: maxPerWindow, + window: window, + concurrent: make(map[string]int), + recent: make(map[string][]time.Time), + } +} + +// allow records a new connection from ip and reports whether it is permitted. +// It enforces both a concurrent cap and a rolling accept-rate cap. An empty ip +// (which should not occur for a real TCP peer) is not throttled. +func (l *ipLimiter) allow(ip string) bool { + if ip == "" { + return true + } + now := time.Now() + l.mu.Lock() + defer l.mu.Unlock() + + if l.maxConcurrent > 0 && l.concurrent[ip] >= l.maxConcurrent { + return false + } + + if l.maxPerWindow > 0 && l.window > 0 { + cutoff := now.Add(-l.window) + times := l.recent[ip] + keep := times[:0] + for _, t := range times { + if t.After(cutoff) { + keep = append(keep, t) + } + } + if len(keep) >= l.maxPerWindow { + l.recent[ip] = keep + return false + } + l.recent[ip] = append(keep, now) + } + + l.concurrent[ip]++ + return true +} + +// release decrements the concurrent count for ip. Safe to call on an ip that +// was never allowed. +func (l *ipLimiter) release(ip string) { + if ip == "" { + return + } + l.mu.Lock() + if n := l.concurrent[ip]; n > 1 { + l.concurrent[ip] = n - 1 + } else { + delete(l.concurrent, ip) + } + l.mu.Unlock() +} + +// sweep drops per-IP rate-limit entries whose timestamps have all aged out, +// keeping the map bounded over a long-running process. +func (l *ipLimiter) sweep() { + l.mu.Lock() + defer l.mu.Unlock() + cutoff := time.Now().Add(-l.window) + for ip, times := range l.recent { + keep := times[:0] + for _, t := range times { + if t.After(cutoff) { + keep = append(keep, t) + } + } + if len(keep) == 0 { + delete(l.recent, ip) + } else { + l.recent[ip] = keep + } + } +} + +// peerIP returns the host part of a net.Addr's string form, mirroring +// connection.Context.GetPeerIP so accept-time and release-time keys agree. +func peerIP(addr net.Addr) string { + host, _, err := net.SplitHostPort(addr.String()) + if err != nil { + return addr.String() + } + return host +} + // SetHandler sets the event handler func (s *Server) SetHandler(h Handler) { s.handler = h @@ -120,9 +238,32 @@ func (s *Server) Start() error { s.wg.Add(1) go s.acceptLoop() + // Periodically prune aged-out per-IP rate-limit entries so the map stays + // bounded over a long-running process. Stopped by s.ctx.Done() in Stop(). + s.wg.Add(1) + go s.ipSweepLoop() + return nil } +func (s *Server) ipSweepLoop() { + defer s.wg.Done() + interval := s.config.PerIPConnRateWindow + if interval <= 0 { + interval = time.Minute + } + t := time.NewTicker(interval) + defer t.Stop() + for { + select { + case <-s.ctx.Done(): + return + case <-t.C: + s.ipLimit.sweep() + } + } +} + func itoa(i int) string { if i == 0 { return "0" @@ -194,6 +335,15 @@ func (s *Server) acceptLoop() { continue } + // Per-IP connection cap + accept rate. Rejecting here (before any + // buffer allocation or goroutine) is cheap, and a rejected device + // simply retries on its reconnect loop. + if !s.ipLimit.allow(peerIP(conn.RemoteAddr())) { + s.logger.Warn("Per-IP connection limit reached, rejecting %s", conn.RemoteAddr()) + _ = conn.Close() + continue + } + // Handle each connection in its own goroutine go s.handleConnection(conn) } @@ -204,6 +354,10 @@ func (s *Server) handleConnection(conn net.Conn) { // Create context ctx := connection.NewContext(conn, nil) + // Always release the per-IP slot on every exit path, including the + // manager.Add failure below (a rare accept/add race). + defer s.ipLimit.release(ctx.GetPeerIP()) + // Add to manager if err := s.manager.Add(ctx); err != nil { s.logger.Warn("Failed to add connection: %v", err) diff --git a/server/go/server/server_test.go b/server/go/server/server_test.go new file mode 100644 index 0000000..e7b8e0a --- /dev/null +++ b/server/go/server/server_test.go @@ -0,0 +1,93 @@ +package server + +import ( + "net" + "testing" + "time" +) + +func TestIPLimiterConcurrentCap(t *testing.T) { + l := newIPLimiter(2, 0, 0) // 2 concurrent, no rate limit + + if !l.allow("10.0.0.1") { + t.Fatal("first connection from an IP should be allowed") + } + if !l.allow("10.0.0.1") { + t.Fatal("second connection from an IP should be allowed") + } + if l.allow("10.0.0.1") { + t.Fatal("third concurrent connection from the same IP should be rejected") + } + + // A different IP is unaffected. + if !l.allow("10.0.0.2") { + t.Fatal("a different IP should not share the first IP's cap") + } + + // Releasing one slot frees capacity again. + l.release("10.0.0.1") + if !l.allow("10.0.0.1") { + t.Fatal("after release, the IP should have a free slot") + } +} + +func TestIPLimiterRateLimit(t *testing.T) { + l := newIPLimiter(0, 3, time.Minute) // no concurrent cap, 3/min per IP + + for range 3 { + if !l.allow("10.0.0.1") { + t.Fatal("an attempt within budget should be allowed") + } + } + if l.allow("10.0.0.1") { + t.Fatal("4th connection within the window should be rejected") + } + // A different IP has its own budget. + if !l.allow("10.0.0.2") { + t.Fatal("a different IP should have its own rate budget") + } +} + +func TestIPLimiterEmptyIPBypasses(t *testing.T) { + l := newIPLimiter(1, 1, time.Minute) + // An empty peer IP (shouldn't happen for real TCP) must not be throttled, + // so a caller that loses the IP falls back to accepting. + if !l.allow("") { + t.Fatal("empty IP should bypass the limiter") + } +} + +func TestIPLimiterSweep(t *testing.T) { + l := newIPLimiter(0, 1, time.Minute) + + l.mu.Lock() + l.recent["stale"] = []time.Time{time.Now().Add(-time.Hour)} + l.recent["fresh"] = []time.Time{time.Now()} + l.mu.Unlock() + + l.sweep() + + l.mu.Lock() + defer l.mu.Unlock() + if _, ok := l.recent["stale"]; ok { + t.Fatal("stale rate-limit entry should be swept") + } + if _, ok := l.recent["fresh"]; !ok { + t.Fatal("fresh rate-limit entry should survive sweep") + } +} + +func TestPeerIP(t *testing.T) { + tests := []struct { + addr net.Addr + want string + }{ + {&net.TCPAddr{IP: net.ParseIP("1.2.3.4"), Port: 1234}, "1.2.3.4"}, + {&net.TCPAddr{IP: net.ParseIP("::1"), Port: 6543}, "::1"}, + } + for _, tt := range tests { + if got := peerIP(tt.addr); got != tt.want { + t.Fatalf("peerIP(%v) = %q, want %q", tt.addr, got, tt.want) + } + } +} diff --git a/server/go/web/server.go b/server/go/web/server.go index 526bbba..2c41217 100644 --- a/server/go/web/server.go +++ b/server/go/web/server.go @@ -10,6 +10,7 @@ import ( "net/http" "strconv" "strings" + "sync" "time" "github.com/yuanyuanxiang/SimpleRemoter/server/go/hub" @@ -33,6 +34,8 @@ type Server struct { trustForwardedFor bool // honor X-Forwarded-For (behind trusted proxy only) basePath string // web UI mount prefix (e.g. "/remoter"); "" = domain root sso SSODetector // nil = SSO disabled + sweepDone chan struct{} // closes to stop the login-limiter sweeper + sweepStopOnce sync.Once } // Config tunes the server's exposed-on-public-HTTPS hardening knobs. @@ -130,11 +133,40 @@ func (s *Server) Start() error { s.log.Error("HTTP server stopped: %v", err) } }() + + // Prune aged-out login rate-limit entries so the key maps (especially the + // username-keyed one, which is attacker-controlled) stay bounded. + s.sweepDone = make(chan struct{}) + go s.sweepLoginLimiters(s.sweepDone) + return nil } +// sweepLoginLimiters periodically sweeps both web login rate limiters. Runs +// until `stop` is closed (from Stop). +func (s *Server) sweepLoginLimiters(stop <-chan struct{}) { + t := time.NewTicker(time.Minute) + defer t.Stop() + for { + select { + case <-stop: + return + case <-t.C: + if s.loginIPLimit != nil { + s.loginIPLimit.Sweep() + } + if s.loginUserLimit != nil { + s.loginUserLimit.Sweep() + } + } + } +} + // Stop gracefully shuts the server down. func (s *Server) Stop() { + if s.sweepDone != nil { + s.sweepStopOnce.Do(func() { close(s.sweepDone) }) + } if s.ws != nil { s.ws.stop() }