mirror of
https://github.com/MeshCore-Beacon/beacon-server.git
synced 2026-10-22 12:22:25 +00:00
feat(api): rate limit REST requests by trusted client IP (#140)
This commit is contained in:
1 parent
0067e51357
commit
595ef4442b
15 files changed
+343
-9
No files matched your search
@@ -12,6 +12,8 @@ projects that make this possible:
|
||||
- [sqlc](https://sqlc.dev) — Type-safe SQL code generation
|
||||
- [chi](https://github.com/go-chi/chi) — HTTP router
|
||||
- [go-chi/cors](https://github.com/go-chi/cors) — CORS middleware for chi
|
||||
- [go-chi/httprate](https://github.com/go-chi/httprate) — Per-client HTTP
|
||||
sliding-window rate limits and retry headers
|
||||
- [go-redis](https://github.com/redis/go-redis) — Redis client
|
||||
- [coder/websocket](https://github.com/coder/websocket) — WebSocket
|
||||
implementation
|
||||
|
||||
+2
-1
@@ -41,6 +41,7 @@ var version = "dev"
|
||||
// @title MeshCore Beacon API
|
||||
// @version 1.6.0
|
||||
// @description MeshCore network observation backend. Ingests LoRa packets from MQTT brokers, stores in PostgreSQL, and streams live events via WebSocket.
|
||||
// @description REST requests share a configurable per-client rate limit (default 300/minute and a 300-request one-second burst cap). Exceeded limits return HTTP 429 with error.code=rate_limited and a Retry-After header in seconds. CORS preflights and WebSocket upgrades do not consume this API budget.
|
||||
// @termsOfService https://github.com/MeshCore-Beacon/beacon-server
|
||||
|
||||
// @contact.name MeshCore Beacon
|
||||
@@ -275,7 +276,7 @@ func main() {
|
||||
go scheduler.Start(ctx)
|
||||
|
||||
// ── HTTP server ──────────────────────────────────────────────────────────
|
||||
r := router.New(h, reader, []*ingest.Worker{broker1, broker2}, resolved.MaxConnsPerIP, cfg.CORS, cfg.Server, cfg.Auth)
|
||||
r := router.New(h, reader, []*ingest.Worker{broker1, broker2}, resolved.MaxConnsPerIP, cfg.CORS, cfg.Server, cfg.Auth, resolved.RateLimit)
|
||||
|
||||
srv := &http.Server{
|
||||
Addr: addr,
|
||||
|
||||
@@ -25,6 +25,17 @@ server:
|
||||
# and pass its validated client address rather than a client-supplied header.
|
||||
trusted_proxies: []
|
||||
|
||||
# Applies to /api/v1/* only. Both counters are local to this Beacon instance.
|
||||
# Set trusted_proxies correctly before using a reverse proxy, or clients share
|
||||
# the proxy's budget. IPv6 clients share a /64 budget to prevent address rotation.
|
||||
ratelimit:
|
||||
enabled: true
|
||||
requests_per_minute: 300
|
||||
# Maximum requests in a one-second sliding window, not token-bucket capacity.
|
||||
# Omitted/zero defaults to requests_per_minute. Negative limits prevent startup.
|
||||
# A 429 returns error.code=rate_limited and Retry-After (1 or 60 seconds).
|
||||
burst: 300
|
||||
|
||||
iatas:
|
||||
YVR:
|
||||
name: Vancouver International
|
||||
|
||||
+1
-1
@@ -4149,7 +4149,7 @@ var SwaggerInfo = &swag.Spec{
|
||||
BasePath: "/api/v1",
|
||||
Schemes: []string{"http", "https"},
|
||||
Title: "MeshCore Beacon API",
|
||||
Description: "MeshCore network observation backend. Ingests LoRa packets from MQTT brokers, stores in PostgreSQL, and streams live events via WebSocket.",
|
||||
Description: "MeshCore network observation backend. Ingests LoRa packets from MQTT brokers, stores in PostgreSQL, and streams live events via WebSocket.\nREST requests share a configurable per-client rate limit (default 300/minute and a 300-request one-second burst cap). Exceeded limits return HTTP 429 with error.code=rate_limited and a Retry-After header in seconds. CORS preflights and WebSocket upgrades do not consume this API budget.",
|
||||
InfoInstanceName: "swagger",
|
||||
SwaggerTemplate: docTemplate,
|
||||
LeftDelim: "{{",
|
||||
|
||||
+1
-1
@@ -5,7 +5,7 @@
|
||||
],
|
||||
"swagger": "2.0",
|
||||
"info": {
|
||||
"description": "MeshCore network observation backend. Ingests LoRa packets from MQTT brokers, stores in PostgreSQL, and streams live events via WebSocket.",
|
||||
"description": "MeshCore network observation backend. Ingests LoRa packets from MQTT brokers, stores in PostgreSQL, and streams live events via WebSocket.\nREST requests share a configurable per-client rate limit (default 300/minute and a 300-request one-second burst cap). Exceeded limits return HTTP 429 with error.code=rate_limited and a Retry-After header in seconds. CORS preflights and WebSocket upgrades do not consume this API budget.",
|
||||
"title": "MeshCore Beacon API",
|
||||
"termsOfService": "https://github.com/MeshCore-Beacon/beacon-server",
|
||||
"contact": {
|
||||
|
||||
+3
-2
@@ -1182,8 +1182,9 @@ info:
|
||||
contact:
|
||||
name: MeshCore Beacon
|
||||
url: https://github.com/MeshCore-Beacon/beacon-server
|
||||
description: MeshCore network observation backend. Ingests LoRa packets from MQTT
|
||||
brokers, stores in PostgreSQL, and streams live events via WebSocket.
|
||||
description: |-
|
||||
MeshCore network observation backend. Ingests LoRa packets from MQTT brokers, stores in PostgreSQL, and streams live events via WebSocket.
|
||||
REST requests share a configurable per-client rate limit (default 300/minute and a 300-request one-second burst cap). Exceeded limits return HTTP 429 with error.code=rate_limited and a Retry-After header in seconds. CORS preflights and WebSocket upgrades do not consume this API budget.
|
||||
license:
|
||||
name: AGPL-3-or-later
|
||||
termsOfService: https://github.com/MeshCore-Beacon/beacon-server
|
||||
|
||||
@@ -8,6 +8,7 @@ require (
|
||||
github.com/eclipse/paho.mqtt.golang v1.5.1
|
||||
github.com/go-chi/chi/v5 v5.3.0
|
||||
github.com/go-chi/cors v1.2.2
|
||||
github.com/go-chi/httprate v0.16.0
|
||||
github.com/google/uuid v1.6.0
|
||||
github.com/jackc/pgx/v5 v5.9.2
|
||||
github.com/joho/godotenv v1.5.1
|
||||
@@ -33,16 +34,19 @@ require (
|
||||
github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 // indirect
|
||||
github.com/jackc/puddle/v2 v2.2.2 // indirect
|
||||
github.com/josharian/intern v1.0.0 // indirect
|
||||
github.com/klauspost/cpuid/v2 v2.2.10 // indirect
|
||||
github.com/mailru/easyjson v0.7.6 // indirect
|
||||
github.com/rogpeppe/go-internal v1.14.1 // indirect
|
||||
github.com/swaggo/files v0.0.0-20220610200504-28940afbdbfe // indirect
|
||||
github.com/yuin/gopher-lua v1.1.1 // indirect
|
||||
github.com/zeebo/xxh3 v1.1.0 // indirect
|
||||
go.mongodb.org/mongo-driver/v2 v2.5.0 // indirect
|
||||
go.uber.org/atomic v1.11.0 // indirect
|
||||
golang.org/x/crypto v0.53.0 // indirect
|
||||
golang.org/x/mod v0.37.0 // indirect
|
||||
golang.org/x/net v0.56.0 // indirect
|
||||
golang.org/x/sync v0.21.0 // indirect
|
||||
golang.org/x/sys v0.46.0 // indirect
|
||||
golang.org/x/text v0.39.0 // indirect
|
||||
golang.org/x/tools v0.47.0 // indirect
|
||||
gopkg.in/yaml.v2 v2.4.0 // indirect
|
||||
|
||||
@@ -22,6 +22,8 @@ github.com/go-chi/chi/v5 v5.3.0 h1:halUjDxhshgXHMrao5bB8eNBXo/rnzwr8m5m36glehM=
|
||||
github.com/go-chi/chi/v5 v5.3.0/go.mod h1:R+tYY2hNuVUUjxoPtqUdgBqevM9s9njzkTLutVsOCto=
|
||||
github.com/go-chi/cors v1.2.2 h1:Jmey33TE+b+rB7fT8MUy1u0I4L+NARQlK6LhzKPSyQE=
|
||||
github.com/go-chi/cors v1.2.2/go.mod h1:sSbTewc+6wYHBBCW7ytsFSn836hqM7JxpglAy2Vzc58=
|
||||
github.com/go-chi/httprate v0.16.0 h1:8V5DH9j6pSK6UQoBsTpvMyFxycqaKEIToyPKzHJjUa8=
|
||||
github.com/go-chi/httprate v0.16.0/go.mod h1:A8lo+qRhk+s9LiuP5saS7XCGDXRXMcrueq0NfIuCa/I=
|
||||
github.com/go-openapi/jsonpointer v0.19.3/go.mod h1:Pl9vOtqEWErmShwVjC8pYs9cog34VGT37dQOVbmoatg=
|
||||
github.com/go-openapi/jsonpointer v0.19.5 h1:gZr+CIYByUqjcgeLXnQu2gHYQC9o73G2XUeOFYEICuY=
|
||||
github.com/go-openapi/jsonpointer v0.19.5/go.mod h1:Pl9vOtqEWErmShwVjC8pYs9cog34VGT37dQOVbmoatg=
|
||||
@@ -88,6 +90,8 @@ github.com/swaggo/swag v1.16.6 h1:qBNcx53ZaX+M5dxVyTrgQ0PJ/ACK+NzhwcbieTt+9yI=
|
||||
github.com/swaggo/swag v1.16.6/go.mod h1:ngP2etMK5a0P3QBizic5MEwpRmluJZPHjXcMoj4Xesg=
|
||||
github.com/yuin/gopher-lua v1.1.1 h1:kYKnWBjvbNP4XLT3+bPEwAXJx262OhaHDWDVOPjL46M=
|
||||
github.com/yuin/gopher-lua v1.1.1/go.mod h1:GBR0iDaNXjAgGg9zfCvksxSRnQx76gclCIb7kdAd1Pw=
|
||||
github.com/zeebo/assert v1.3.0 h1:g7C04CbJuIDKNPFHmsk4hwZDO5O+kntRxzaUoNXj+IQ=
|
||||
github.com/zeebo/assert v1.3.0/go.mod h1:Pq9JiuJQpG8JLJdtkwrJESF0Foym2/D9XMU5ciN/wJ0=
|
||||
github.com/zeebo/xxh3 v1.1.0 h1:s7DLGDK45Dyfg7++yxI0khrfwq9661w9EN78eP/UZVs=
|
||||
github.com/zeebo/xxh3 v1.1.0/go.mod h1:IisAie1LELR4xhVinxWS5+zf1lA4p0MW4T+w+W07F5s=
|
||||
go.mongodb.org/mongo-driver/v2 v2.5.0 h1:yXUhImUjjAInNcpTcAlPHiT7bIXhshCTL3jVBkF3xaE=
|
||||
|
||||
@@ -0,0 +1,36 @@
|
||||
// Copyright 2026 Beacon Contributors
|
||||
// SPDX-License-Identifier: AGPL-3.0-or-later
|
||||
|
||||
package middleware
|
||||
|
||||
import (
|
||||
"net/http"
|
||||
"time"
|
||||
|
||||
"github.com/go-chi/httprate"
|
||||
|
||||
"github.com/MeshCore-Beacon/beacon-server/internal/config"
|
||||
)
|
||||
|
||||
// RateLimit applies the minute budget and one-second burst cap to the API.
|
||||
// TrustedProxyIP must run first; the access logger must wrap this middleware.
|
||||
func RateLimit(cfg config.ResolvedRateLimitConfig) func(http.Handler) http.Handler {
|
||||
if !cfg.Enabled {
|
||||
return func(next http.Handler) http.Handler { return next }
|
||||
}
|
||||
options := []httprate.Option{
|
||||
httprate.WithLimitHandler(func(w http.ResponseWriter, r *http.Request) {
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
w.WriteHeader(http.StatusTooManyRequests)
|
||||
_, _ = w.Write([]byte(`{"error":{"code":"rate_limited","message":"Request rate limit exceeded. Retry later."}}`))
|
||||
}),
|
||||
// Only expose the backoff header, since the two windows have different budgets.
|
||||
httprate.WithResponseHeaders(httprate.ResponseHeaders{RetryAfter: "Retry-After"}),
|
||||
}
|
||||
// RemoteAddr has already been resolved and forwarding headers removed.
|
||||
minute := httprate.LimitBy(cfg.RequestsPerMinute, time.Minute, httprate.KeyByIP, options...)
|
||||
burst := httprate.LimitBy(cfg.Burst, time.Second, httprate.KeyByIP, options...)
|
||||
return func(next http.Handler) http.Handler {
|
||||
return minute(burst(next))
|
||||
}
|
||||
}
|
||||
@@ -13,7 +13,7 @@ import (
|
||||
|
||||
func TestAdminAuthBoundary(t *testing.T) {
|
||||
for _, key := range []string{"", "synthetic-test-key"} {
|
||||
handler := New(nil, nil, nil, 5, config.CORSConfig{}, config.ServerConfig{}, config.AuthConfig{APIKey: key})
|
||||
handler := New(nil, nil, nil, 5, config.CORSConfig{}, config.ServerConfig{}, config.AuthConfig{APIKey: key}, config.ResolvedRateLimitConfig{})
|
||||
for _, path := range []string{"/api/v1/admin", "/api/v1/admin/", "/api/v1/admin/config", "/api/v1/admin//config"} {
|
||||
for _, method := range []string{"GET", "POST", "PUT", "DELETE", "OPTIONS"} {
|
||||
for _, token := range []string{"", "wrong", "synthetic-test-key"} {
|
||||
|
||||
@@ -0,0 +1,181 @@
|
||||
// Copyright 2026 Beacon Contributors
|
||||
// SPDX-License-Identifier: AGPL-3.0-or-later
|
||||
|
||||
package router
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"encoding/json"
|
||||
"log"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"net/netip"
|
||||
"strings"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
"testing/synctest"
|
||||
"time"
|
||||
|
||||
"github.com/go-chi/chi/v5/middleware"
|
||||
|
||||
"github.com/MeshCore-Beacon/beacon-server/internal/config"
|
||||
)
|
||||
|
||||
func TestDefaultAPIRateLimit(t *testing.T) {
|
||||
handler := New(nil, nil, nil, 5, config.CORSConfig{}, config.ServerConfig{}, config.AuthConfig{}, config.Resolve(&config.Config{}).RateLimit)
|
||||
for i := 0; i <= 300; i++ {
|
||||
request := httptest.NewRequest(http.MethodGet, "/api/v1/brokers", nil)
|
||||
request.RemoteAddr = "198.51.100.1:1234"
|
||||
response := httptest.NewRecorder()
|
||||
handler.ServeHTTP(response, request)
|
||||
want := http.StatusOK
|
||||
if i == 300 {
|
||||
want = http.StatusTooManyRequests
|
||||
}
|
||||
if response.Code != want {
|
||||
t.Fatalf("request %d: got %d, want %d", i+1, response.Code, want)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func rateRequest(handler http.Handler, path, peer, headerIP string) *httptest.ResponseRecorder {
|
||||
request := httptest.NewRequest(http.MethodGet, path, nil)
|
||||
request.RemoteAddr = peer
|
||||
request.Header.Set("Origin", "https://example.test")
|
||||
request.Header.Set("X-Real-IP", headerIP)
|
||||
request.Header.Set("X-Forwarded-For", headerIP)
|
||||
request.Header.Set("True-Client-IP", headerIP)
|
||||
response := httptest.NewRecorder()
|
||||
handler.ServeHTTP(response, request)
|
||||
return response
|
||||
}
|
||||
|
||||
func TestAPIRateLimitContractAndLogging(t *testing.T) {
|
||||
var logs bytes.Buffer
|
||||
previous := middleware.DefaultLogger
|
||||
middleware.DefaultLogger = middleware.RequestLogger(&middleware.DefaultLogFormatter{Logger: log.New(&logs, "", 0), NoColor: true})
|
||||
t.Cleanup(func() { middleware.DefaultLogger = previous })
|
||||
proxy := config.ServerConfig{TrustedProxies: []netip.Prefix{netip.MustParsePrefix("127.0.0.1/32")}}
|
||||
handler := New(nil, nil, nil, 5, config.CORSConfig{}, proxy, config.AuthConfig{}, config.ResolvedRateLimitConfig{Enabled: true, RequestsPerMinute: 2, Burst: 10})
|
||||
for _, path := range []string{"/api/v1/brokers", "/api/v1/packets?limit=0"} {
|
||||
response := rateRequest(handler, path, "127.0.0.1:1234", "198.51.100.25")
|
||||
if response.Code != http.StatusOK && response.Code != http.StatusBadRequest {
|
||||
t.Fatalf("initial request: %d", response.Code)
|
||||
}
|
||||
}
|
||||
response := rateRequest(handler, "/api/v1/brokers", "127.0.0.1:1234", "198.51.100.25")
|
||||
var body struct {
|
||||
Error struct{ Code, Message string }
|
||||
}
|
||||
if err := json.Unmarshal(response.Body.Bytes(), &body); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if response.Code != http.StatusTooManyRequests || body.Error.Code != "rate_limited" || body.Error.Message == "" {
|
||||
t.Fatalf("wrong rate-limit response: %d %s", response.Code, response.Body.String())
|
||||
}
|
||||
if response.Header().Get("Content-Type") != "application/json" || response.Header().Get("Retry-After") != "60" || !strings.EqualFold(response.Header().Get("Access-Control-Expose-Headers"), "Retry-After") || response.Header().Get("Access-Control-Allow-Origin") == "" {
|
||||
t.Fatalf("missing JSON/backoff/CORS headers: %v", response.Header())
|
||||
}
|
||||
if strings.Count(logs.String(), " - 429 ") != 1 || !strings.Contains(logs.String(), "from 198.51.100.25 - 429 ") || !strings.Contains(logs.String(), "/api/v1/brokers") {
|
||||
t.Fatalf("expected one rejection log with resolved client and path: %s", logs.String())
|
||||
}
|
||||
}
|
||||
|
||||
func TestAPIRateLimitClientIdentity(t *testing.T) {
|
||||
for _, tc := range []struct {
|
||||
name, firstPeer, secondPeer, firstHeader, secondHeader string
|
||||
trusted, shared bool
|
||||
}{
|
||||
{"untrusted header rotation", "198.51.100.1:1", "198.51.100.1:2", "192.0.2.1", "192.0.2.2", false, true},
|
||||
{"independent peers", "198.51.100.1:1", "198.51.100.2:1", "192.0.2.1", "192.0.2.1", false, false},
|
||||
{"trusted independent clients", "127.0.0.1:1", "127.0.0.1:2", "198.51.100.1", "198.51.100.2", true, false},
|
||||
{"trusted shared client", "127.0.0.1:1", "127.0.0.1:2", "198.51.100.1", "198.51.100.1", true, true},
|
||||
{"IPv6 shared prefix", "[2001:db8:1::1]:1", "[2001:db8:1::2]:2", "", "", false, true},
|
||||
{"IPv6 independent prefixes", "[2001:db8:1::1]:1", "[2001:db8:2::1]:2", "", "", false, false},
|
||||
} {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
proxy := config.ServerConfig{}
|
||||
if tc.trusted {
|
||||
proxy.TrustedProxies = []netip.Prefix{netip.MustParsePrefix("127.0.0.1/32")}
|
||||
}
|
||||
handler := New(nil, nil, nil, 5, config.CORSConfig{}, proxy, config.AuthConfig{}, config.ResolvedRateLimitConfig{Enabled: true, RequestsPerMinute: 1, Burst: 1})
|
||||
if response := rateRequest(handler, "/api/v1/brokers", tc.firstPeer, tc.firstHeader); response.Code != http.StatusOK {
|
||||
t.Fatalf("first client: %d", response.Code)
|
||||
}
|
||||
want := http.StatusOK
|
||||
if tc.shared {
|
||||
want = http.StatusTooManyRequests
|
||||
}
|
||||
if response := rateRequest(handler, "/api/v1/brokers", tc.secondPeer, tc.secondHeader); response.Code != want {
|
||||
t.Fatalf("second client: %d, want %d", response.Code, want)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestAPIRateLimitWindowsAndExclusions(t *testing.T) {
|
||||
synctest.Test(t, func(t *testing.T) {
|
||||
handler := New(nil, nil, nil, 5, config.CORSConfig{}, config.ServerConfig{}, config.AuthConfig{}, config.ResolvedRateLimitConfig{Enabled: true, RequestsPerMinute: 4, Burst: 2})
|
||||
request := func() *httptest.ResponseRecorder {
|
||||
return rateRequest(handler, "/api/v1/brokers", "198.51.100.1:1", "")
|
||||
}
|
||||
preflight := httptest.NewRequest(http.MethodOptions, "/api/v1/brokers", nil)
|
||||
preflight.RemoteAddr = "198.51.100.1:1"
|
||||
preflight.Header.Set("Origin", "https://example.test")
|
||||
preflight.Header.Set("Access-Control-Request-Method", "GET")
|
||||
response := httptest.NewRecorder()
|
||||
handler.ServeHTTP(response, preflight)
|
||||
if response.Code != http.StatusOK || request().Code != http.StatusOK || request().Code != http.StatusOK {
|
||||
t.Fatal("preflight consumed the API budget")
|
||||
}
|
||||
if response := request(); response.Code != http.StatusTooManyRequests || response.Header().Get("Retry-After") != "1" {
|
||||
t.Fatalf("burst limit: %d %v", response.Code, response.Header())
|
||||
}
|
||||
time.Sleep(2 * time.Second)
|
||||
if request().Code != http.StatusOK {
|
||||
t.Fatal("burst did not recover")
|
||||
}
|
||||
if response := request(); response.Code != http.StatusTooManyRequests || response.Header().Get("Retry-After") != "60" {
|
||||
t.Fatalf("minute limit: %d %v", response.Code, response.Header())
|
||||
}
|
||||
response = httptest.NewRecorder()
|
||||
handler.ServeHTTP(response, preflight)
|
||||
if response.Code != http.StatusOK {
|
||||
t.Fatal("exhausted API budget blocked preflight")
|
||||
}
|
||||
if response := rateRequest(handler, "/swagger", "198.51.100.1:1", ""); response.Code != http.StatusMovedPermanently {
|
||||
t.Fatalf("Swagger shares the API budget: %d", response.Code)
|
||||
}
|
||||
time.Sleep(2 * time.Minute)
|
||||
if request().Code != http.StatusOK {
|
||||
t.Fatal("minute budget did not recover")
|
||||
}
|
||||
})
|
||||
handler := New(nil, nil, nil, 5, config.CORSConfig{}, config.ServerConfig{}, config.AuthConfig{}, config.ResolvedRateLimitConfig{Enabled: false, RequestsPerMinute: 1, Burst: 1})
|
||||
for range 5 {
|
||||
if response := rateRequest(handler, "/api/v1/brokers", "198.51.100.1:1", ""); response.Code != http.StatusOK || response.Header().Get("Retry-After") != "" {
|
||||
t.Fatal("disabled limiter still applied")
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestAPIRateLimitConcurrentRequests(t *testing.T) {
|
||||
handler := New(nil, nil, nil, 5, config.CORSConfig{}, config.ServerConfig{}, config.AuthConfig{}, config.ResolvedRateLimitConfig{Enabled: true, RequestsPerMinute: 5, Burst: 5})
|
||||
var accepted, rejected atomic.Int32
|
||||
var group sync.WaitGroup
|
||||
for range 20 {
|
||||
group.Go(func() {
|
||||
switch rateRequest(handler, "/api/v1/brokers", "198.51.100.1:1", "").Code {
|
||||
case http.StatusOK:
|
||||
accepted.Add(1)
|
||||
case http.StatusTooManyRequests:
|
||||
rejected.Add(1)
|
||||
}
|
||||
})
|
||||
}
|
||||
group.Wait()
|
||||
if accepted.Load() != 5 || rejected.Load() != 15 {
|
||||
t.Fatalf("concurrent budget: accepted=%d rejected=%d", accepted.Load(), rejected.Load())
|
||||
}
|
||||
}
|
||||
@@ -39,7 +39,7 @@ import (
|
||||
// /stats → stats subrouter
|
||||
//
|
||||
// Admin handlers are added separately; the reserved subtree is protected now.
|
||||
func New(h *hub.Hub, reader api.Reader, workers []*ingest.Worker, maxConnsPerIP int, corsCfg config.CORSConfig, serverCfg config.ServerConfig, authCfg config.AuthConfig) http.Handler {
|
||||
func New(h *hub.Hub, reader api.Reader, workers []*ingest.Worker, maxConnsPerIP int, corsCfg config.CORSConfig, serverCfg config.ServerConfig, authCfg config.AuthConfig, rateLimitCfg config.ResolvedRateLimitConfig) http.Handler {
|
||||
r := chi.NewRouter()
|
||||
|
||||
// ── CORS ─────────────────────────────────────────────────────────────────
|
||||
@@ -63,6 +63,7 @@ func New(h *hub.Hub, reader api.Reader, workers []*ingest.Worker, maxConnsPerIP
|
||||
AllowedOrigins: allowedOrigins,
|
||||
AllowedMethods: allowedMethods,
|
||||
AllowedHeaders: allowedHeaders,
|
||||
ExposedHeaders: []string{"Retry-After"},
|
||||
AllowCredentials: corsCfg.AllowCredentials,
|
||||
MaxAge: maxAge,
|
||||
}))
|
||||
@@ -89,6 +90,7 @@ func New(h *hub.Hub, reader api.Reader, workers []*ingest.Worker, maxConnsPerIP
|
||||
|
||||
// ── Public REST API (v1) ─────────────────────────────────────────────────
|
||||
r.Route("/api/v1", func(r chi.Router) {
|
||||
r.Use(mw.RateLimit(rateLimitCfg))
|
||||
// Public group — no authentication required.
|
||||
r.Group(func(r chi.Router) {
|
||||
r.Mount("/packets", handlers.PacketsRouter(reader))
|
||||
|
||||
@@ -36,8 +36,24 @@ func TestWebSocketLimitUsesTrustedClientIP(t *testing.T) {
|
||||
|
||||
func checkWebSocketLimit(t *testing.T, cfg config.ServerConfig, secondIP string, wantStatus int) {
|
||||
t.Helper()
|
||||
server := httptest.NewServer(New(hub.New(), nil, nil, 1, config.CORSConfig{}, cfg, config.AuthConfig{}))
|
||||
server := httptest.NewServer(New(hub.New(), nil, nil, 1, config.CORSConfig{}, cfg, config.AuthConfig{}, config.ResolvedRateLimitConfig{Enabled: true, RequestsPerMinute: 1, Burst: 1}))
|
||||
defer server.Close()
|
||||
// Exhaust this client's REST budget before checking its independent WS cap.
|
||||
for _, want := range []int{http.StatusOK, http.StatusTooManyRequests} {
|
||||
request, err := http.NewRequest(http.MethodGet, server.URL+"/api/v1/brokers", nil)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
request.Header.Set("X-Real-IP", "198.51.100.1")
|
||||
response, err := server.Client().Do(request)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
response.Body.Close()
|
||||
if response.StatusCode != want {
|
||||
t.Fatalf("REST budget setup: %d, want %d", response.StatusCode, want)
|
||||
}
|
||||
}
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
||||
defer cancel()
|
||||
url := "ws" + strings.TrimPrefix(server.URL, "http") + "/ws"
|
||||
|
||||
@@ -30,6 +30,7 @@ type Config struct {
|
||||
Scopes []ScopeConfig `yaml:"scopes"`
|
||||
Cache CacheConfig `yaml:"cache"`
|
||||
CORS CORSConfig `yaml:"cors"`
|
||||
RateLimit RateLimitConfig `yaml:"ratelimit"`
|
||||
Background BackgroundConfig `yaml:"background"`
|
||||
Presence PresenceConfig `yaml:"presence"`
|
||||
Nodes NodesConfig `yaml:"nodes"`
|
||||
@@ -69,6 +70,7 @@ func (c *ServerConfig) UnmarshalYAML(node *yaml.Node) error {
|
||||
|
||||
// ResolvedConfig holds all runtime configuration with defaults applied.
|
||||
type ResolvedConfig struct {
|
||||
RateLimit ResolvedRateLimitConfig
|
||||
TelemetryResolution time.Duration
|
||||
TelemetryRetention time.Duration
|
||||
PacketRetention time.Duration
|
||||
@@ -95,6 +97,20 @@ type ResolvedConfig struct {
|
||||
ObserverDeleteAfter time.Duration
|
||||
}
|
||||
|
||||
// RateLimitConfig controls the per-client REST API request budget.
|
||||
type RateLimitConfig struct {
|
||||
Enabled *bool `yaml:"enabled"` // Defaults to true; false disables both windows.
|
||||
RequestsPerMinute int `yaml:"requests_per_minute"`
|
||||
Burst int `yaml:"burst"` // One-second window cap, not token-bucket capacity.
|
||||
}
|
||||
|
||||
// ResolvedRateLimitConfig has defaults applied and no optional values.
|
||||
type ResolvedRateLimitConfig struct {
|
||||
Enabled bool
|
||||
RequestsPerMinute int
|
||||
Burst int
|
||||
}
|
||||
|
||||
// PresenceConfig controls coalescing of presence bookkeeping writes
|
||||
// (observer last_seen, observer_brokers, packet last_heard_at bumps).
|
||||
type PresenceConfig struct {
|
||||
@@ -356,6 +372,9 @@ func Load(path string) (*Config, error) {
|
||||
return nil, fmt.Errorf("server.trusted_proxies[%d] must be a valid CIDR", i)
|
||||
}
|
||||
}
|
||||
if cfg.RateLimit.RequestsPerMinute < 0 || cfg.RateLimit.Burst < 0 {
|
||||
return nil, fmt.Errorf("ratelimit.requests_per_minute and ratelimit.burst must be positive or zero for defaults")
|
||||
}
|
||||
configDir := filepath.Dir(path)
|
||||
for iata, details := range cfg.IATAs {
|
||||
if details.BorderFile != "" && !filepath.IsAbs(details.BorderFile) {
|
||||
@@ -369,6 +388,11 @@ func Load(path string) (*Config, error) {
|
||||
// Resolve returns a ResolvedConfig with defaults applied for any zero values.
|
||||
func Resolve(cfg *Config) ResolvedConfig {
|
||||
r := ResolvedConfig{
|
||||
RateLimit: ResolvedRateLimitConfig{
|
||||
Enabled: cfg.RateLimit.Enabled == nil || *cfg.RateLimit.Enabled,
|
||||
RequestsPerMinute: cfg.RateLimit.RequestsPerMinute,
|
||||
Burst: cfg.RateLimit.Burst,
|
||||
},
|
||||
TelemetryResolution: cfg.Telemetry.Resolution.Duration,
|
||||
TelemetryRetention: cfg.Telemetry.Retention.Duration,
|
||||
PacketRetention: cfg.Packets.Retention.Duration,
|
||||
@@ -388,6 +412,12 @@ func Resolve(cfg *Config) ResolvedConfig {
|
||||
NodeDeleteAfter: cfg.Nodes.DeleteAfter.Duration,
|
||||
ObserverDeleteAfter: cfg.Observers.DeleteAfter.Duration,
|
||||
}
|
||||
if r.RateLimit.RequestsPerMinute == 0 {
|
||||
r.RateLimit.RequestsPerMinute = 300
|
||||
}
|
||||
if r.RateLimit.Burst == 0 {
|
||||
r.RateLimit.Burst = r.RateLimit.RequestsPerMinute
|
||||
}
|
||||
if r.TelemetryResolution == 0 {
|
||||
r.TelemetryResolution = time.Hour
|
||||
}
|
||||
@@ -440,10 +470,11 @@ func Resolve(cfg *Config) ResolvedConfig {
|
||||
|
||||
func (r ResolvedConfig) String() string {
|
||||
return fmt.Sprintf(
|
||||
"telemetryResolution=%s telemetryRetention=%s packetRetention=%s routeRetention=%s routeGrace=%s routeMinObs=%d maxConnsPerIP=%d viewRefresh=%s reconfirm=%s cleanup=%s presenceFlush=%s presencePacketTTL=%s clockDriftThreshold=%s nodeStaleThreshold=%s nodeDeleteAfter=%s observerDeleteAfter=%s",
|
||||
"telemetryResolution=%s telemetryRetention=%s packetRetention=%s routeRetention=%s routeGrace=%s routeMinObs=%d maxConnsPerIP=%d viewRefresh=%s reconfirm=%s cleanup=%s presenceFlush=%s presencePacketTTL=%s clockDriftThreshold=%s nodeStaleThreshold=%s nodeDeleteAfter=%s observerDeleteAfter=%s rateLimitEnabled=%t requestsPerMinute=%d burst=%d",
|
||||
r.TelemetryResolution, r.TelemetryRetention, r.PacketRetention, r.RouteRetention, r.RouteGrace, r.RouteMinObservations,
|
||||
r.MaxConnsPerIP, r.ViewRefreshInterval, r.ReconfirmInterval, r.CleanupInterval,
|
||||
r.PresenceFlushInterval, r.PresencePacketTTL, r.ClockDriftThreshold,
|
||||
r.NodeStaleThreshold, r.NodeDeleteAfter, r.ObserverDeleteAfter,
|
||||
r.RateLimit.Enabled, r.RateLimit.RequestsPerMinute, r.RateLimit.Burst,
|
||||
)
|
||||
}
|
||||
@@ -0,0 +1,45 @@
|
||||
// Copyright 2026 Beacon Contributors
|
||||
// SPDX-License-Identifier: AGPL-3.0-or-later
|
||||
|
||||
package config
|
||||
|
||||
import (
|
||||
"os"
|
||||
"path/filepath"
|
||||
"testing"
|
||||
)
|
||||
|
||||
func TestRateLimitConfig(t *testing.T) {
|
||||
for _, tc := range []struct {
|
||||
name, yaml string
|
||||
want ResolvedRateLimitConfig
|
||||
wantError bool
|
||||
}{
|
||||
{"omitted", "{}", ResolvedRateLimitConfig{true, 300, 300}, false},
|
||||
{"disabled", "ratelimit: {enabled: false}", ResolvedRateLimitConfig{false, 300, 300}, false},
|
||||
{"custom", "ratelimit: {enabled: true, requests_per_minute: 900, burst: 100}", ResolvedRateLimitConfig{true, 900, 100}, false},
|
||||
{"burst default follows budget", "ratelimit: {requests_per_minute: 600}", ResolvedRateLimitConfig{true, 600, 600}, false},
|
||||
{"zero defaults", "ratelimit: {requests_per_minute: 0, burst: 0}", ResolvedRateLimitConfig{true, 300, 300}, false},
|
||||
{"negative minute", "ratelimit: {requests_per_minute: -1}", ResolvedRateLimitConfig{}, true},
|
||||
{"negative burst", "ratelimit: {burst: -1}", ResolvedRateLimitConfig{}, true},
|
||||
{"invalid disabled config", "ratelimit: {enabled: false, burst: -1}", ResolvedRateLimitConfig{}, true},
|
||||
} {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
path := filepath.Join(t.TempDir(), "config.yaml")
|
||||
if err := os.WriteFile(path, []byte(tc.yaml), 0600); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
cfg, err := Load(path)
|
||||
if (err != nil) != tc.wantError {
|
||||
t.Fatalf("Load error = %v, want error = %v", err, tc.wantError)
|
||||
}
|
||||
if err == nil && Resolve(cfg).RateLimit != tc.want {
|
||||
t.Fatalf("resolved limits = %+v, want %+v", Resolve(cfg).RateLimit, tc.want)
|
||||
}
|
||||
})
|
||||
}
|
||||
cfg, err := Load(filepath.Join(t.TempDir(), "missing.yaml"))
|
||||
if err != nil || Resolve(cfg).RateLimit != (ResolvedRateLimitConfig{true, 300, 300}) {
|
||||
t.Fatalf("missing config did not enable defaults: %v", err)
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user